from __future__ import annotations import unittest from pathlib import Path MIGRATION_FILES = sorted(Path("app/projects/migrations").glob("*.sql")) ALLOWED_TABLES = { "teams", "team_members", "invitations", "project_collaborators", "approval_workflows", "approval_requests", "review_comments", "collaboration_activity", "notification_preferences", } class MigrationContractTests(unittest.TestCase): def test_approval_workspace_integrity_migration_remains_present(self) -> None: migration = Path( "app/projects/migrations/0014_approval_workspace_integrity.sql" ).read_text() self.assertIn("alter table approval_requests", migration) self.assertIn("add column if not exists workspace_id text", migration) self.assertIn("alter column workspace_id set not null", migration) self.assertIn("create index if not exists ix_approval_requests_workspace", migration) def test_latest_migration_repairs_project_brand_kit_reference(self) -> None: latest = MIGRATION_FILES[-1] self.assertEqual(latest.name, "0015_repair_project_brand_kit_column.sql") body = latest.read_text() self.assertIn("alter table projects", body) self.assertIn("add column if not exists brand_kit_id text", body) self.assertIn("fk_projects_brand_kit", body) self.assertIn("ix_projects_workspace_brand_kit", body) def test_migrations_do_not_recreate_shared_tables(self) -> None: counts: dict[str, int] = {} for path in MIGRATION_FILES: body = path.read_text() for table in ALLOWED_TABLES: counts[table] = counts.get(table, 0) + body.count( f"create table if not exists {table}" ) for table, count in counts.items(): self.assertEqual( count, 1, f"Duplicate table creation detected for {table}: {count} migrations recreate it", ) if __name__ == "__main__": unittest.main()