File size: 2,081 Bytes
7cc81cb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
26591c4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7cc81cb
 
 
 
 
 
26591c4
 
 
7cc81cb
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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()