Spaces:
Running
Running
| 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() | |