add-ideas/govoplan-core
1
0
Fork 0
Code Issues 45 Pull Requests Actions Packages Projects Releases Wiki Activity

Release v0.1.8

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-11 17:00:37 +02:00
parent a00ef54821
commit 9a0c467d55
102 changed files with 2150 additions and 9272 deletions
Download Patch File Download Diff File Expand all files Collapse all files

550
tests/test_database_migrations.py
View File

@@ -9,12 +9,8 @@ from alembic.runtime.migration import MigrationContext
from alembic.script import ScriptDirectory
from sqlalchemy import create_engine, inspect, text
from govoplan_core.db.base import Base
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.db.migrations import (
REVISION_AUTH_RBAC,
REVISION_CORE_CHANGE_SEQUENCE,
REVISION_FILE_FOLDERS,
REVISION_HIERARCHICAL_SETTINGS,
alembic_config,
migrate_database,
reconcile_change_sequence_retention_floor_drift,
@@ -22,8 +18,20 @@ from govoplan_core.db.migrations import (
)
def configured_migration_heads(database_url: str) -> set[str]:
return set(ScriptDirectory.from_config(alembic_config(database_url=database_url)).get_heads())
def configured_migration_heads(
database_url: str,
*,
enabled_modules: tuple[str, ...] | list[str] | None = None,
migration_track: str = "release",
) -> set[str]:
script = ScriptDirectory.from_config(
alembic_config(
database_url=database_url,
enabled_modules=enabled_modules,
migration_track=migration_track,
),
)
return {revision.revision for revision in script.get_revisions("heads")}
def database_migration_heads(connection) -> set[str]:
@@ -31,16 +39,26 @@ def database_migration_heads(connection) -> set[str]:
class DatabaseMigrationTests(unittest.TestCase):
def test_migration_tracks_use_separate_version_locations(self) -> None:
release_locations = alembic_config(database_url="sqlite:////tmp/govoplan-release.db").get_main_option("version_locations")
dev_locations = alembic_config(
database_url="sqlite:////tmp/govoplan-dev.db",
migration_track="dev",
).get_main_option("version_locations")
self.assertIn("alembic/versions", release_locations)
self.assertNotIn("alembic/dev_versions", release_locations)
self.assertIn("alembic/dev_versions", dev_locations)
self.assertIn("migrations/dev_versions", dev_locations)
def test_repairs_missing_change_sequence_retention_floor(self) -> None:
with tempfile.TemporaryDirectory(prefix="msm-change-sequence-drift-test-") as directory:
with tempfile.TemporaryDirectory(prefix="govoplan-change-sequence-drift-test-") as directory:
database = Path(directory) / "change-sequence-drift.db"
url = f"sqlite:///{database}"
command.upgrade(alembic_config(database_url=url, enabled_modules=()), REVISION_CORE_CHANGE_SEQUENCE)
engine = create_engine(url)
try:
with engine.begin() as connection:
connection.execute(text("DROP TABLE core_change_sequence_retention_floor"))
ChangeSequenceEntry.__table__.create(bind=connection)
with engine.connect() as connection:
tables = set(inspect(connection).get_table_names())
self.assertIn("core_change_sequence", tables)
@@ -63,7 +81,7 @@ class DatabaseMigrationTests(unittest.TestCase):
engine.dispose()
def test_repairs_namespace_tables_when_database_was_stamped_ahead(self) -> None:
with tempfile.TemporaryDirectory(prefix="msm-namespace-drift-test-") as directory:
with tempfile.TemporaryDirectory(prefix="govoplan-namespace-drift-test-") as directory:
database = Path(directory) / "namespace-drift.db"
url = f"sqlite:///{database}"
engine = create_engine(url)
@@ -92,437 +110,185 @@ class DatabaseMigrationTests(unittest.TestCase):
finally:
engine.dispose()
def test_repairs_create_all_schema_drift_and_upgrades_to_head(self) -> None:
with tempfile.TemporaryDirectory(prefix="msm-migration-test-") as directory:
database = Path(directory) / "legacy.db"
def test_core_and_access_baselines_apply_to_fresh_database(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-core-baseline-test-") as directory:
database = Path(directory) / "core.db"
url = f"sqlite:///{database}"
# Reproduce the historical development database: Alembic was run
# through auth/RBAC, then create_all() created later file tables
# without advancing alembic_version and without altering the
# already-existing campaign_versions table.
command.upgrade(alembic_config(database_url=url), REVISION_AUTH_RBAC)
engine = create_engine(url)
try:
Base.metadata.create_all(bind=engine)
with engine.connect() as connection:
self.assertEqual(
MigrationContext.configure(connection).get_current_revision(),
REVISION_AUTH_RBAC,
)
self.assertIn("file_blobs", inspect(connection).get_table_names())
self.assertNotIn(
"user_lock_state",
{column["name"] for column in inspect(connection).get_columns("campaign_versions")},
)
finally:
engine.dispose()
result = migrate_database(database_url=url)
self.assertEqual(result.previous_revision, REVISION_AUTH_RBAC)
self.assertEqual(result.reconciled_revision, REVISION_FILE_FOLDERS)
engine = create_engine(url)
try:
with engine.connect() as connection:
current = database_migration_heads(connection)
inspector = inspect(connection)
columns = {
column["name"]
for column in inspector.get_columns("campaign_versions")
}
folder_indexes = {index["name"] for index in inspector.get_indexes("file_folders")}
job_columns = {column["name"] for column in inspector.get_columns("campaign_jobs")}
job_indexes = {index["name"] for index in inspector.get_indexes("campaign_jobs")}
attempt_columns = {column["name"] for column in inspector.get_columns("send_attempts")}
tables = set(inspector.get_table_names())
tenant_columns = {column["name"] for column in inspector.get_columns("core_scopes")}
user_columns = {column["name"] for column in inspector.get_columns("access_users")}
session_columns = {column["name"] for column in inspector.get_columns("access_auth_sessions")}
group_columns = {column["name"] for column in inspector.get_columns("access_groups")}
role_columns = {column["name"] for column in inspector.get_columns("access_roles")}
role_indexes = {index["name"] for index in inspector.get_indexes("access_roles")}
campaign_columns = {column["name"] for column in inspector.get_columns("campaigns")}
audit_columns = {column["name"] for column in inspector.get_columns("audit_log")}
audit_indexes = {index["name"] for index in inspector.get_indexes("audit_log")}
system_role_flags = {
row["slug"]: bool(row["is_builtin"])
for row in connection.execute(text(
"SELECT slug, is_builtin FROM access_roles WHERE tenant_id IS NULL AND slug IN ('system_owner', 'system_admin', 'system_auditor')"
)).mappings().all()
}
system_settings_columns = {column["name"] for column in inspector.get_columns("core_system_settings")}
governance_assignment_columns = {column["name"] for column in inspector.get_columns("admin_governance_template_assignments")}
import_profile_columns = {column["name"] for column in inspector.get_columns("campaign_recipient_import_mapping_profiles")}
import_profile_indexes = {index["name"] for index in inspector.get_indexes("campaign_recipient_import_mapping_profiles")}
account_count = connection.execute(text("SELECT COUNT(*) FROM access_accounts")).scalar_one()
user_count = connection.execute(text("SELECT COUNT(*) FROM access_users")).scalar_one()
system_owner_count = connection.execute(text(
"""
SELECT COUNT(*)
FROM access_system_role_assignments assignment
JOIN access_roles role ON role.id = assignment.role_id
JOIN access_accounts account ON account.id = assignment.account_id
WHERE role.slug = 'system_owner' AND account.is_active = 1
"""
)).scalar_one()
self.assertEqual(current, configured_migration_heads(url))
self.assertEqual(result.current_revision, ",".join(sorted(current)))
self.assertIn("user_lock_state", columns)
self.assertIn("user_locked_at", columns)
self.assertIn("user_locked_by_user_id", columns)
self.assertIn("uq_file_folders_active_user_path", folder_indexes)
self.assertIn("uq_file_folders_active_group_path", folder_indexes)
self.assertIn("execution_snapshot", columns)
self.assertIn("execution_snapshot_hash", columns)
self.assertIn("claimed_at", job_columns)
self.assertIn("claim_token", job_columns)
self.assertIn("smtp_started_at", job_columns)
self.assertIn("outcome_unknown_at", job_columns)
self.assertIn("eml_sha256", job_columns)
self.assertIn("ix_campaign_jobs_claim_token", job_indexes)
self.assertIn("ix_campaign_jobs_eml_sha256", job_indexes)
self.assertIn("status", attempt_columns)
self.assertIn("claim_token", attempt_columns)
self.assertIn("access_accounts", tables)
self.assertIn("access_system_role_assignments", tables)
self.assertIn("core_scopes", tables)
self.assertNotIn("tenancy_tenants", tables)
self.assertIn("core_system_settings", tables)
self.assertIn("admin_governance_templates", tables)
self.assertIn("admin_governance_template_assignments", tables)
self.assertIn("campaign_shares", tables)
self.assertIn("campaign_recipient_import_mapping_profiles", tables)
self.assertIn("owner_user_id", campaign_columns)
self.assertIn("owner_group_id", campaign_columns)
self.assertIn("ordered_header_fingerprint", import_profile_columns)
self.assertIn("unordered_header_fingerprint", import_profile_columns)
self.assertIn("mappings", import_profile_columns)
self.assertIn("ix_recipient_import_profiles_ordered_fp", import_profile_indexes)
self.assertIn("scope", audit_columns)
self.assertIn("ix_audit_log_scope", audit_indexes)
self.assertIn("ix_audit_log_scope_created_at", audit_indexes)
self.assertIn("ix_audit_log_tenant_scope_created_at", audit_indexes)
self.assertTrue(system_role_flags["system_owner"])
self.assertFalse(system_role_flags["system_admin"])
self.assertFalse(system_role_flags["system_auditor"])
self.assertIn("allow_custom_groups", tenant_columns)
self.assertIn("allow_custom_roles", tenant_columns)
self.assertIn("allow_api_keys", tenant_columns)
self.assertIn("description", tenant_columns)
self.assertIn("default_locale", tenant_columns)
self.assertIn("settings", tenant_columns)
self.assertIn("settings", user_columns)
self.assertIn("settings", group_columns)
self.assertIn("settings", campaign_columns)
self.assertIn("account_id", user_columns)
self.assertIn("account_id", session_columns)
self.assertIn("description", group_columns)
self.assertIn("is_active", group_columns)
self.assertIn("system_template_id", group_columns)
self.assertIn("system_required", group_columns)
self.assertIn("description", role_columns)
self.assertIn("is_builtin", role_columns)
self.assertIn("is_assignable", role_columns)
self.assertIn("system_template_id", role_columns)
self.assertIn("system_required", role_columns)
self.assertIn("allow_tenant_custom_groups", system_settings_columns)
self.assertIn("allow_tenant_custom_roles", system_settings_columns)
self.assertIn("allow_tenant_api_keys", system_settings_columns)
self.assertIn("mode", governance_assignment_columns)
self.assertIn("uq_roles_system_slug", role_indexes)
self.assertEqual(account_count, user_count)
self.assertEqual(system_owner_count, 1 if user_count else 0)
finally:
engine.dispose()
def test_module_migrations_apply_after_core_scope_table_rename(self) -> None:
with tempfile.TemporaryDirectory(prefix="msm-late-module-migration-test-") as directory:
database = Path(directory) / "late-modules.db"
url = f"sqlite:///{database}"
command.upgrade(alembic_config(database_url=url, enabled_modules=()), "heads")
engine = create_engine(url)
try:
with engine.connect() as connection:
tables = set(inspect(connection).get_table_names())
self.assertIn("core_scopes", tables)
self.assertNotIn("tenancy_tenants", tables)
inspector = inspect(connection)
tables = set(inspector.get_table_names())
role_flags = {
row["slug"]: bool(row["is_builtin"])
for row in connection.execute(
text(
"""
SELECT slug, is_builtin
FROM access_roles
WHERE tenant_id IS NULL
AND slug IN ('system_owner', 'system_admin', 'system_auditor')
"""
),
).mappings()
}
system_settings_count = connection.execute(
text("SELECT COUNT(*) FROM core_system_settings WHERE id = 'global'"),
).scalar_one()
current = database_migration_heads(connection)
self.assertEqual(current, configured_migration_heads(url, enabled_modules=()))
self.assertIn("core_scopes", tables)
self.assertNotIn("tenancy_tenants", tables)
self.assertIn("access_accounts", tables)
self.assertIn("access_users", tables)
self.assertIn("access_identities", tables)
self.assertIn("access_function_assignments", tables)
self.assertIn("audit_log", tables)
self.assertIn("audit_outbox_events", tables)
self.assertIn("file_assets", tables)
self.assertIn("mail_server_profiles", tables)
self.assertEqual(system_settings_count, 1)
self.assertEqual(
role_flags,
{"system_owner": True, "system_admin": False, "system_auditor": False},
)
finally:
engine.dispose()
migrate_database(database_url=url)
def test_default_module_baselines_apply_to_fresh_database(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-default-baseline-test-") as directory:
database = Path(directory) / "default.db"
url = f"sqlite:///{database}"
result = migrate_database(database_url=url)
engine = create_engine(url)
try:
with engine.connect() as connection:
inspector = inspect(connection)
mail_policy_fks = inspector.get_foreign_keys("mail_profile_policies")
referred_tables = {fk["referred_table"] for fk in mail_policy_fks}
self.assertIn("core_scopes", referred_tables)
self.assertNotIn("tenancy_tenants", referred_tables)
finally:
engine.dispose()
def test_repairs_current_schema_stamped_at_file_folders(self) -> None:
with tempfile.TemporaryDirectory(prefix="msm-current-schema-marker-test-") as directory:
database = Path(directory) / "current-schema-marker.db"
url = f"sqlite:///{database}"
command.upgrade(alembic_config(database_url=url), REVISION_HIERARCHICAL_SETTINGS)
engine = create_engine(url)
try:
with engine.begin() as connection:
connection.execute(
text(
"""
INSERT INTO mail_server_profiles
(id, tenant_id, scope_type, scope_id, name, slug, description, is_active,
smtp_config, smtp_password_encrypted, imap_config, imap_password_encrypted,
created_by_user_id, updated_by_user_id, created_at, updated_at)
VALUES
('profile-1', NULL, 'system', 'system', 'System Mail', 'system-mail', NULL, 1,
:smtp_config, NULL, :imap_config, NULL,
NULL, NULL, '2026-06-25 15:00:00', '2026-06-25 15:00:00')
"""
),
{
"smtp_config": '{"host":"smtp.example.test","port":587,"security":"starttls","username":"smtp-user","enabled":true}',
"imap_config": '{"host":"imap.example.test","port":993,"security":"ssl","username":"imap-user","enabled":true}',
},
)
connection.execute(
text("UPDATE alembic_version SET version_num = :revision"),
{"revision": REVISION_FILE_FOLDERS},
)
with engine.connect() as connection:
self.assertEqual(
MigrationContext.configure(connection).get_current_revision(),
REVISION_FILE_FOLDERS,
)
profile_columns = {
column["name"]
for column in inspect(connection).get_columns("mail_server_profiles")
}
self.assertNotIn("smtp_username", profile_columns)
self.assertNotIn("imap_username", profile_columns)
finally:
engine.dispose()
result = migrate_database(database_url=url)
self.assertEqual(result.previous_revision, REVISION_FILE_FOLDERS)
self.assertEqual(result.reconciled_revision, REVISION_HIERARCHICAL_SETTINGS)
engine = create_engine(url)
try:
with engine.connect() as connection:
tables = set(inspector.get_table_names())
current = database_migration_heads(connection)
self.assertEqual(current, configured_migration_heads(url))
self.assertEqual(result.current_revision, ",".join(sorted(current)))
profile_columns = {
column["name"]
for column in inspect(connection).get_columns("mail_server_profiles")
mail_policy_fks = inspector.get_foreign_keys("mail_profile_policies")
mail_referred_tables = {fk["referred_table"] for fk in mail_policy_fks}
folder_indexes = {index["name"] for index in inspector.get_indexes("file_folders")}
import_profile_indexes = {
index["name"]
for index in inspector.get_indexes("campaign_recipient_import_mapping_profiles")
}
profile = connection.execute(text(
"""
SELECT smtp_username, smtp_config, imap_username, imap_config
FROM mail_server_profiles
WHERE id = 'profile-1'
"""
)).mappings().one()
self.assertIn("smtp_username", profile_columns)
self.assertIn("imap_username", profile_columns)
self.assertEqual(profile["smtp_username"], "smtp-user")
self.assertEqual(profile["imap_username"], "imap-user")
self.assertNotIn("username", profile["smtp_config"])
self.assertNotIn("enabled", profile["smtp_config"])
self.assertNotIn("username", profile["imap_config"])
self.assertNotIn("enabled", profile["imap_config"])
self.assertIsNone(result.previous_revision)
self.assertIsNone(result.reconciled_revision)
self.assertEqual(current, configured_migration_heads(url))
self.assertEqual(result.current_revision, ",".join(sorted(current)))
self.assertIn("calendar_sync_credentials", tables)
self.assertIn("campaign_recipient_import_mapping_profiles", tables)
self.assertIn("file_connector_credentials", tables)
self.assertIn("file_connector_policies", tables)
self.assertIn("identity_identities", tables)
self.assertIn("identity_account_links", tables)
self.assertIn("mail_profile_policies", tables)
self.assertIn("organizations_functions", tables)
self.assertIn("core_scopes", mail_referred_tables)
self.assertIn("uq_file_folders_active_user_path", folder_indexes)
self.assertIn("uq_file_folders_active_group_path", folder_indexes)
self.assertIn("ix_recipient_import_profiles_ordered_fp", import_profile_indexes)
finally:
engine.dispose()
def test_migrates_legacy_login_identity_and_bootstraps_system_owner(self) -> None:
with tempfile.TemporaryDirectory(prefix="msm-account-migration-test-") as directory:
database = Path(directory) / "accounts.db"
def test_dev_migration_track_applies_detailed_chain_to_fresh_database(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-dev-track-migration-test-") as directory:
database = Path(directory) / "dev-track.db"
url = f"sqlite:///{database}"
command.upgrade(alembic_config(database_url=url), "7b8c9d0e1f2a")
engine = create_engine(url)
try:
with engine.begin() as connection:
timestamps = {"created_at": "2026-06-14 12:00:00", "updated_at": "2026-06-14 12:00:00"}
connection.execute(text(
"""
INSERT INTO tenancy_tenants (id, slug, name, is_active, created_at, updated_at)
VALUES ('tenant-1', 'legacy', 'Legacy', 1, :created_at, :updated_at)
"""
), timestamps)
connection.execute(text(
"""
INSERT INTO access_users
(id, tenant_id, email, display_name, is_active, is_tenant_admin,
auth_provider, password_hash, last_login_at, created_at, updated_at)
VALUES
('user-1', 'tenant-1', 'Owner@Example.Local', 'Legacy Owner', 1, 1,
'local', 'legacy-password-hash', NULL, :created_at, :updated_at)
"""
), timestamps)
connection.execute(text(
"""
INSERT INTO access_auth_sessions
(id, tenant_id, user_id, token_hash, expires_at, last_seen_at, revoked_at,
user_agent, ip_address, created_at, updated_at)
VALUES
('session-1', 'tenant-1', 'user-1', 'token-hash', '2026-06-15 12:00:00',
NULL, NULL, NULL, NULL, :created_at, :updated_at)
"""
), timestamps)
finally:
engine.dispose()
migrate_database(database_url=url, reconcile_legacy_schema=False)
result = migrate_database(database_url=url, migration_track="dev")
engine = create_engine(url)
try:
with engine.connect() as connection:
tables = set(inspect(connection).get_table_names())
account = connection.execute(text(
"SELECT id, normalized_email, password_hash FROM access_accounts"
)).mappings().one()
membership = connection.execute(text(
"SELECT account_id FROM access_users WHERE id = 'user-1'"
)).mappings().one()
migrated_session = connection.execute(text(
"SELECT account_id FROM access_auth_sessions WHERE id = 'session-1'"
)).mappings().one()
owner_assignment = connection.execute(text(
"""
SELECT role.slug
FROM access_user_role_assignments assignment
JOIN access_roles role ON role.id = assignment.role_id
WHERE assignment.user_id = 'user-1'
"""
)).scalar_one()
owner_permissions = connection.execute(text(
"SELECT permissions FROM access_roles WHERE tenant_id = 'tenant-1' AND slug = 'owner'"
)).scalar_one()
system_assignment = connection.execute(text(
"""
SELECT role.slug
FROM access_system_role_assignments assignment
JOIN access_roles role ON role.id = assignment.role_id
WHERE assignment.account_id = :account_id
"""
), {"account_id": account["id"]}).scalar_one()
system_role_flags = {
row["slug"]: bool(row["is_builtin"])
for row in connection.execute(text(
"SELECT slug, is_builtin FROM access_roles WHERE tenant_id IS NULL AND slug IN ('system_owner', 'system_admin', 'system_auditor')"
)).mappings().all()
}
scope_slug = connection.execute(text(
"SELECT slug FROM core_scopes WHERE id = 'tenant-1'"
)).scalar_one()
current = database_migration_heads(connection)
self.assertEqual(current, configured_migration_heads(url, migration_track="dev"))
self.assertEqual(result.current_revision, ",".join(sorted(current)))
self.assertIn("0f1e2d3c4b5a", current)
self.assertIn("audit_outbox_events", tables)
self.assertIn("file_connector_profiles", tables)
self.assertIn("core_scopes", tables)
self.assertNotIn("tenancy_tenants", tables)
self.assertEqual(account["normalized_email"], "owner@example.local")
self.assertEqual(account["password_hash"], "legacy-password-hash")
self.assertEqual(scope_slug, "legacy")
self.assertEqual(membership["account_id"], account["id"])
self.assertEqual(migrated_session["account_id"], account["id"])
self.assertEqual(owner_assignment, "owner")
self.assertIn("tenant:*", owner_permissions)
self.assertEqual(system_assignment, "system_owner")
self.assertTrue(system_role_flags["system_owner"])
self.assertFalse(system_role_flags["system_admin"])
self.assertFalse(system_role_flags["system_auditor"])
finally:
engine.dispose()
def test_backfills_explicit_system_and_tenant_audit_scopes(self) -> None:
with tempfile.TemporaryDirectory(prefix="msm-audit-scope-migration-test-") as directory:
database = Path(directory) / "audit-scope.db"
def test_migrate_database_removes_stale_dependency_parent_version_rows(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-stale-head-baseline-test-") as directory:
database = Path(directory) / "stale-heads.db"
url = f"sqlite:///{database}"
command.upgrade(alembic_config(database_url=url), "a0b1c2d3e4f5")
migrate_database(database_url=url)
engine = create_engine(url)
try:
with engine.begin() as connection:
timestamps = {
"created_at": "2026-06-15 12:00:00",
"updated_at": "2026-06-15 12:00:00",
}
connection.execute(text(
"""
INSERT INTO audit_log
(id, tenant_id, user_id, api_key_id, action, object_type, object_id, details, created_at, updated_at)
VALUES
('audit-system', NULL, NULL, NULL, 'system_settings.updated', 'core_system_settings', 'system', '{}', :created_at, :updated_at),
('audit-tenant', NULL, NULL, NULL, 'user.updated', 'user', 'user-1', '{}', :created_at, :updated_at)
"""
), timestamps)
connection.execute(
text("INSERT INTO alembic_version (version_num) VALUES (:revision)"),
{"revision": "4f2a9c8e7b6d"},
)
with engine.connect() as connection:
self.assertIn("4f2a9c8e7b6d", database_migration_heads(connection))
finally:
engine.dispose()
migrate_database(database_url=url, reconcile_legacy_schema=False)
result = migrate_database(database_url=url)
engine = create_engine(url)
try:
with engine.connect() as connection:
scopes = dict(connection.execute(text(
"SELECT id, scope FROM audit_log WHERE id IN ('audit-system', 'audit-tenant')"
)).all())
self.assertEqual(scopes["audit-system"], "system")
self.assertEqual(scopes["audit-tenant"], "tenant")
current = database_migration_heads(connection)
self.assertEqual(current, configured_migration_heads(url))
self.assertNotIn("4f2a9c8e7b6d", current)
self.assertEqual(result.current_revision, ",".join(sorted(current)))
finally:
engine.dispose()
def test_deduplicates_active_folder_paths_before_unique_indexes(self) -> None:
with tempfile.TemporaryDirectory(prefix="msm-folder-migration-test-") as directory:
database = Path(directory) / "duplicates.db"
def test_full_product_baselines_apply_with_idm(self) -> None:
enabled_modules = [
"tenancy",
"organizations",
"identity",
"access",
"admin",
"dashboard",
"policy",
"audit",
"files",
"mail",
"campaigns",
"calendar",
"docs",
"ops",
"idm",
]
with tempfile.TemporaryDirectory(prefix="govoplan-full-baseline-test-") as directory:
database = Path(directory) / "full.db"
url = f"sqlite:///{database}"
command.upgrade(alembic_config(database_url=url), "5f6a7b8c9d0e")
engine = create_engine(url)
try:
with engine.begin() as connection:
parameters = {
"tenant_id": "tenant-1",
"owner_user_id": "user-1",
"path": "archive",
"created_at": "2026-06-13 20:00:00",
"updated_at": "2026-06-13 20:00:00",
}
connection.execute(text(
"""
INSERT INTO file_folders
(id, tenant_id, owner_type, owner_user_id, owner_group_id, path,
created_by_user_id, deleted_at, metadata, created_at, updated_at)
VALUES
('folder-1', :tenant_id, 'user', :owner_user_id, NULL, :path,
NULL, NULL, '{}', :created_at, :updated_at),
('folder-2', :tenant_id, 'user', :owner_user_id, NULL, :path,
NULL, NULL, '{}', :created_at, :updated_at)
"""
), parameters)
finally:
engine.dispose()
migrate_database(database_url=url, reconcile_legacy_schema=False)
result = migrate_database(database_url=url, enabled_modules=enabled_modules)
engine = create_engine(url)
try:
with engine.connect() as connection:
active_count = connection.execute(text(
"SELECT COUNT(*) FROM file_folders WHERE path = 'archive' AND deleted_at IS NULL"
)).scalar_one()
deleted_count = connection.execute(text(
"SELECT COUNT(*) FROM file_folders WHERE path = 'archive' AND deleted_at IS NOT NULL"
)).scalar_one()
folder_indexes = {index["name"] for index in inspect(connection).get_indexes("file_folders")}
self.assertEqual(active_count, 1)
self.assertEqual(deleted_count, 1)
self.assertIn("uq_file_folders_active_user_path", folder_indexes)
self.assertIn("uq_file_folders_active_group_path", folder_indexes)
tables = set(inspect(connection).get_table_names())
current = database_migration_heads(connection)
self.assertEqual(current, configured_migration_heads(url, enabled_modules=enabled_modules))
self.assertEqual(result.current_revision, ",".join(sorted(current)))
self.assertIn("idm_tenant_settings", tables)
self.assertIn("idm_organization_function_assignments", tables)
finally:
engine.dispose()
if __name__ == "__main__":
unittest.main()