tests/test_database_migration_contracts.py

from __future__ import annotations

import re
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_latest_migration_adds_approval_request_workspace_integrity(self) -> None:
        latest = MIGRATION_FILES[-1].read_text()

        self.assertIn("alter table approval_requests", latest)
        self.assertIn("add column if not exists workspace_id text", latest)
        self.assertIn("alter column workspace_id set not null", latest)
        self.assertIn("create index if not exists ix_approval_requests_workspace", latest)

    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()