gabriel / musehub public
test_data_integrity.py python
111 lines 4.1 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 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 )
48 for model in (MusehubIssue, MusehubIssueComment):
49 cols = {c.name for c in model.__table__.columns}
50 assert "updated_at" in cols, f"{model.__name__} is missing updated_at"
51
52
53 async def test_repo_updated_at_is_populated_on_create(
54 db_session: AsyncSession,
55 ) -> None:
56 """A freshly created repo must have a non-null updated_at."""
57 repo = await create_repo(db_session, slug="updated-at-test", owner="testuser")
58 assert repo.updated_at is not None, "updated_at must be set on repo creation"
59
60
61 # ── Orphan object scan ─────────────────────────────────────────────────────────
62
63 async def test_orphan_scan_returns_empty_when_no_orphans(
64 db_session: AsyncSession,
65 ) -> None:
66 """scan_orphan_objects must return empty when all objects have at least one ref."""
67 from musehub.maintenance.orphan_scan import scan_orphan_objects
68 from musehub.db import musehub_models as db_models
69
70 repo = await create_repo(db_session, slug="orphan-scan-clean", owner="testuser")
71
72 oid = "sha256:" + "a" * 64
73 obj = db_models.MusehubObject(
74 object_id=oid,
75 path="test.bin",
76 size_bytes=4,
77 disk_path="test.bin",
78 storage_uri="local://test.bin",
79 )
80 db_session.add(obj)
81 db_session.add(db_models.MusehubObjectRef(repo_id=repo.repo_id, object_id=oid))
82 await db_session.commit()
83
84 result = await scan_orphan_objects(db_session)
85 assert result.ok
86 assert result.count == 0
87
88
89 async def test_orphan_scan_detects_objects_with_no_refs(
90 db_session: AsyncSession,
91 ) -> None:
92 """scan_orphan_objects must find objects that have no row in musehub_object_refs."""
93 from musehub.maintenance.orphan_scan import scan_orphan_objects
94 from musehub.db import musehub_models as db_models
95
96 orphan_obj_id = "sha256:" + "c" * 64
97
98 # Insert an object with no corresponding ref row.
99 obj = db_models.MusehubObject(
100 object_id=orphan_obj_id,
101 path="orphan.bin",
102 size_bytes=4,
103 disk_path="orphan.bin",
104 storage_uri="local://orphan.bin",
105 )
106 db_session.add(obj)
107 await db_session.commit()
108
109 result = await scan_orphan_objects(db_session)
110 assert not result.ok, "Orphan scan should detect the unreferenced object"
111 assert orphan_obj_id in result.orphaned_object_ids
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago