gabriel / musehub public
test_data_integrity_section51.py python
150 lines 5.6 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Tests for checklist section 5.1 — Database integrity.
2
3 Covers:
4 - updated_at present on critical tables
5 - FK constraints enforced (PostgreSQL)
6 - Orphan object scan (scan and delete)
7 """
8 from __future__ import annotations
9
10 import pytest
11 from sqlalchemy.ext.asyncio import AsyncSession
12
13 from tests.factories import create_repo
14
15
16 # ── updated_at on critical tables ─────────────────────────────────────────────
17
18 def test_musehub_repo_has_updated_at() -> None:
19 from musehub.db.musehub_models import MusehubRepo
20 cols = {c.name for c in MusehubRepo.__table__.columns}
21 assert "updated_at" in cols, "MusehubRepo is missing updated_at column"
22
23
24 def test_musehub_proposal_has_updated_at() -> None:
25 from musehub.db.musehub_models import MusehubProposal
26 cols = {c.name for c in MusehubProposal.__table__.columns}
27 assert "updated_at" in cols, "MusehubProposal is missing updated_at column"
28
29
30 def test_musehub_webhook_has_updated_at() -> None:
31 from musehub.db.musehub_models import MusehubWebhook
32 cols = {c.name for c in MusehubWebhook.__table__.columns}
33 assert "updated_at" in cols, "MusehubWebhook is missing updated_at column"
34
35
36 def test_musehub_release_has_updated_at() -> None:
37 from musehub.db.musehub_models import MusehubRelease
38 cols = {c.name for c in MusehubRelease.__table__.columns}
39 assert "updated_at" in cols, "MusehubRelease is missing updated_at column"
40
41
42 def test_existing_critical_tables_have_updated_at() -> None:
43 """Spot-check that pre-existing updated_at columns are still present."""
44 from musehub.db.musehub_models import (
45 MusehubIssue,
46 MusehubIssueComment,
47 MusehubMilestone,
48 )
49 for model in (MusehubIssue, MusehubIssueComment, MusehubMilestone):
50 cols = {c.name for c in model.__table__.columns}
51 assert "updated_at" in cols, f"{model.__name__} is missing updated_at"
52
53
54 @pytest.mark.anyio
55 async def test_repo_updated_at_is_populated_on_create(
56 db_session: AsyncSession,
57 ) -> None:
58 """A freshly created repo must have a non-null updated_at."""
59 repo = await create_repo(db_session, slug="updated-at-test", owner="testuser")
60 assert repo.updated_at is not None, "updated_at must be set on repo creation"
61
62
63 # ── Foreign key constraints ────────────────────────────────────────────────────
64
65
66 def test_fk_constraints_defined_on_object_table() -> None:
67 """MusehubObject must declare a FK on repo_id pointing to musehub_repos."""
68 from musehub.db.musehub_models import MusehubObject
69 fk_targets = {
70 fk.column.table.name
71 for col in MusehubObject.__table__.columns
72 for fk in col.foreign_keys
73 }
74 assert "musehub_repos" in fk_targets, (
75 "MusehubObject.repo_id must have a FK to musehub_repos"
76 )
77
78
79 def test_fk_ondelete_cascade_on_object_table() -> None:
80 """MusehubObject FK on repo_id must use ondelete=CASCADE."""
81 from musehub.db.musehub_models import MusehubObject
82 repo_col = MusehubObject.__table__.c["repo_id"]
83 for fk in repo_col.foreign_keys:
84 assert fk.ondelete == "CASCADE", (
85 f"MusehubObject.repo_id FK must have ondelete=CASCADE, got {fk.ondelete!r}"
86 )
87
88
89 # ── Orphan object scan ─────────────────────────────────────────────────────────
90
91 @pytest.mark.anyio
92 async def test_orphan_scan_returns_empty_when_no_orphans(
93 db_session: AsyncSession,
94 ) -> None:
95 """scan_orphan_objects must return an empty result when all objects have valid repos."""
96 from musehub.maintenance.orphan_scan import scan_orphan_objects
97 from musehub.db import musehub_models as db_models
98
99 repo = await create_repo(db_session, slug="orphan-scan-clean", owner="testuser")
100
101 obj = db_models.MusehubObject(
102 object_id="sha256:" + "a" * 64,
103 repo_id=repo.repo_id,
104 path="test.bin",
105 size_bytes=4,
106 disk_path="test.bin",
107 storage_uri="local://test.bin",
108 )
109 db_session.add(obj)
110 await db_session.commit()
111
112 result = await scan_orphan_objects(db_session)
113 assert result.ok
114 assert result.count == 0
115
116
117 @pytest.mark.anyio
118 async def test_orphan_scan_detects_objects_with_deleted_repo(
119 db_session: AsyncSession,
120 ) -> None:
121 """scan_orphan_objects must find objects whose repo_id no longer exists.
122
123 We disable FK enforcement temporarily so we can insert a row with a
124 dangling repo_id — simulating what a direct DB edit or failed migration
125 could leave behind.
126 """
127 from musehub.maintenance.orphan_scan import scan_orphan_objects
128 from musehub.db import musehub_models as db_models
129 from sqlalchemy import text
130
131 orphan_obj_id = "sha256:" + "c" * 64
132
133 # Temporarily disable FK triggers to insert the orphan row (PostgreSQL equivalent).
134 await db_session.execute(text("SET session_replication_role = 'replica'"))
135 await db_session.execute(
136 db_models.MusehubObject.__table__.insert().values(
137 object_id=orphan_obj_id,
138 repo_id="nonexistent-repo-id-orphan-test",
139 path="orphan.bin",
140 size_bytes=4,
141 disk_path="orphan.bin",
142 storage_uri="local://orphan.bin",
143 )
144 )
145 await db_session.commit()
146 await db_session.execute(text("SET session_replication_role = DEFAULT"))
147
148 result = await scan_orphan_objects(db_session)
149 assert not result.ok, "Orphan scan should detect the dangling row"
150 assert orphan_obj_id in result.orphaned_object_ids
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago