Files
govoplan-idm/tests/test_migrations.py
T
zemion 21e8f0bc39
Module Package Release / publish-packages (push) Successful in 11s
feat(idm): govern delegation chains and timed escalation
2026-08-22 03:12:30 +02:00

73 lines
2.7 KiB
Python

from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine, inspect
from govoplan_core.db.migrations import migrate_database
from govoplan_identity.backend.manifest import get_manifest as identity_manifest
from govoplan_idm.backend.manifest import get_manifest as idm_manifest
from govoplan_organizations.backend.manifest import (
get_manifest as organizations_manifest,
)
class IdmMigrationTests(unittest.TestCase):
def test_migrations_create_typed_relationship_tables_and_head(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-idm-migration-") as directory:
url = f"sqlite:///{Path(directory) / 'idm.db'}"
migrate_database(
database_url=url,
enabled_modules=("identity", "organizations", "idm"),
manifest_factories=(
identity_manifest,
organizations_manifest,
idm_manifest,
),
)
engine = create_engine(url)
try:
with engine.connect() as connection:
self.assertIn(
"c2d3e4f5a6b7",
set(MigrationContext.configure(connection).get_current_heads()),
)
self.assertEqual(
{
"idm_function_assignment_change_events",
"idm_function_assignment_changes",
"idm_identity_relationships",
"idm_organization_function_assignments",
"idm_tenant_settings",
"idm_typed_groups",
},
{
name
for name in inspect(connection).get_table_names()
if name.startswith("idm_")
},
)
change_columns = {
item["name"]
for item in inspect(connection).get_columns(
"idm_function_assignment_changes"
)
}
self.assertTrue(
{
"review_deadline_at",
"escalated_at",
"escalation_from_state",
"escalation_target_function_id",
}.issubset(change_columns)
)
finally:
engine.dispose()
if __name__ == "__main__":
unittest.main()