gabriel / musehub public
test_repo_card_integrity.py python
248 lines 10.3 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """
2 Tier 6 — Integrity tests for enrich_repo_cards().
3
4 Integrity tests verify structural invariants that must hold for every
5 enrichment result regardless of input shape — bucket count, field ranges,
6 type contracts, and cross-field consistency.
7
8 Test IDs
9 --------
10 T600 — pulse_buckets is always exactly _PULSE_DAYS entries
11 T601 — pulse bucket dates are strictly ascending with no gaps or duplicates
12 T602 — pulse bucket counts are always non-negative integers
13 T603 — pulse bucket h values are in range [0, _SPARKLINE_HEIGHT]
14 T604 — autonomy_pct is always in range [0, 100]
15 T605 — dead_count, error_count, warning_count are always non-negative
16 T606 — health_status is always one of {'clean', 'warn', 'risk'}
17 T607 — hottest_symbol.churn_30d is always > 0 (never a zero-churn symbol)
18 T608 — blast_leader.blast is always > 0 (never a zero-blast symbol)
19 T609 — result always contains exactly the requested repo_ids as keys
20 T610 — pulse bucket colors are all valid hex strings
21 """
22 from __future__ import annotations
23
24 import re
25 from datetime import datetime, timedelta, timezone
26
27 import pytest
28 from sqlalchemy.ext.asyncio import AsyncSession
29
30 from musehub.db.musehub_models import (
31 MusehubIntelBreakageMeta,
32 MusehubIntelDead,
33 MusehubSymbolIntel,
34 )
35 from musehub.services.repo_card_enrichment import (
36 _PULSE_DAYS,
37 _SPARKLINE_HEIGHT,
38 enrich_repo_cards,
39 )
40 from tests.factories import create_commit, create_repo
41
42
43 def _utc_now() -> datetime:
44 return datetime.now(tz=timezone.utc)
45
46
47 _HEX_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
48
49
50 # ---------------------------------------------------------------------------
51 # Shared fixture: one richly-populated repo and one empty repo
52 # ---------------------------------------------------------------------------
53
54 async def _seed_two_repos(db: AsyncSession):
55 """Return (rich_repo_id, empty_repo_id) after seeding data for rich."""
56 rich = await create_repo(db, visibility="public")
57 empty = await create_repo(db, visibility="public")
58
59 # Populate rich repo with every signal type
60 for i in range(5):
61 commit = await create_commit(db, rich.repo_id, timestamp=_utc_now() - timedelta(days=i))
62 db.add(MusehubSymbolIntel(
63 repo_id=rich.repo_id, address="src/a.py::fast_fn", churn_30d=10, blast=50
64 ))
65 db.add(MusehubSymbolIntel(
66 repo_id=rich.repo_id, address="src/b.py::slow_fn", churn_30d=2, blast=200
67 ))
68 db.add(MusehubIntelDead(
69 repo_id=rich.repo_id, address="src/old.py::dead_fn",
70 kind="function", confidence="high", ref="main"
71 ))
72 db.add(MusehubIntelBreakageMeta(
73 repo_id=rich.repo_id, total_issues=1,
74 error_count=0, warning_count=1, file_count=1, ref="main"
75 ))
76 await db.commit()
77 return rich.repo_id, empty.repo_id
78
79
80 # ---------------------------------------------------------------------------
81 # T600 — exactly _PULSE_DAYS buckets
82 # ---------------------------------------------------------------------------
83
84 @pytest.mark.asyncio
85 async def test_t600_pulse_bucket_count_invariant(db_session: AsyncSession) -> None:
86 """T600: pulse_buckets always has exactly _PULSE_DAYS entries."""
87 rich_id, empty_id = await _seed_two_repos(db_session)
88 result = await enrich_repo_cards(db_session, [rich_id, empty_id])
89 for repo_id, enc in result.items():
90 assert len(enc.pulse_buckets) == _PULSE_DAYS, (
91 f"repo {repo_id}: expected {_PULSE_DAYS} buckets, got {len(enc.pulse_buckets)}"
92 )
93
94
95 # ---------------------------------------------------------------------------
96 # T601 — strictly ascending dates, no gaps or duplicates
97 # ---------------------------------------------------------------------------
98
99 @pytest.mark.asyncio
100 async def test_t601_pulse_dates_strictly_ascending(db_session: AsyncSession) -> None:
101 """T601: bucket dates are strictly ascending ISO strings with no gaps."""
102 from datetime import date, timedelta
103 rich_id, empty_id = await _seed_two_repos(db_session)
104 result = await enrich_repo_cards(db_session, [rich_id, empty_id])
105
106 for repo_id, enc in result.items():
107 dates = [date.fromisoformat(b.date) for b in enc.pulse_buckets]
108 # Strictly ascending
109 assert dates == sorted(dates), f"repo {repo_id}: dates not sorted"
110 assert len(dates) == len(set(dates)), f"repo {repo_id}: duplicate dates"
111 # No gaps: each consecutive pair differs by exactly 1 day
112 for a, b in zip(dates, dates[1:]):
113 assert (b - a).days == 1, f"repo {repo_id}: gap between {a} and {b}"
114
115
116 # ---------------------------------------------------------------------------
117 # T602 — non-negative counts
118 # ---------------------------------------------------------------------------
119
120 @pytest.mark.asyncio
121 async def test_t602_pulse_counts_non_negative(db_session: AsyncSession) -> None:
122 """T602: every bucket.count is >= 0."""
123 rich_id, empty_id = await _seed_two_repos(db_session)
124 result = await enrich_repo_cards(db_session, [rich_id, empty_id])
125 for repo_id, enc in result.items():
126 for b in enc.pulse_buckets:
127 assert b.count >= 0, f"repo {repo_id}: negative count {b.count} on {b.date}"
128
129
130 # ---------------------------------------------------------------------------
131 # T603 — h in [0, _SPARKLINE_HEIGHT]
132 # ---------------------------------------------------------------------------
133
134 @pytest.mark.asyncio
135 async def test_t603_pulse_h_within_range(db_session: AsyncSession) -> None:
136 """T603: every bucket.h is in [0, _SPARKLINE_HEIGHT]."""
137 rich_id, empty_id = await _seed_two_repos(db_session)
138 result = await enrich_repo_cards(db_session, [rich_id, empty_id])
139 for repo_id, enc in result.items():
140 for b in enc.pulse_buckets:
141 assert 0 <= b.h <= _SPARKLINE_HEIGHT, (
142 f"repo {repo_id}: h={b.h} out of [0,{_SPARKLINE_HEIGHT}] on {b.date}"
143 )
144
145
146 # ---------------------------------------------------------------------------
147 # T604 — autonomy_pct in [0, 100]
148 # ---------------------------------------------------------------------------
149
150 @pytest.mark.asyncio
151 async def test_t604_autonomy_pct_bounded(db_session: AsyncSession) -> None:
152 """T604: autonomy_pct is always in [0, 100]."""
153 rich_id, empty_id = await _seed_two_repos(db_session)
154 result = await enrich_repo_cards(db_session, [rich_id, empty_id])
155 for repo_id, enc in result.items():
156 assert 0 <= enc.autonomy_pct <= 100, (
157 f"repo {repo_id}: autonomy_pct={enc.autonomy_pct} out of bounds"
158 )
159
160
161 # ---------------------------------------------------------------------------
162 # T605 — dead/error/warning counts non-negative
163 # ---------------------------------------------------------------------------
164
165 @pytest.mark.asyncio
166 async def test_t605_intel_counts_non_negative(db_session: AsyncSession) -> None:
167 """T605: dead_count, error_count, and warning_count are always >= 0."""
168 rich_id, empty_id = await _seed_two_repos(db_session)
169 result = await enrich_repo_cards(db_session, [rich_id, empty_id])
170 for repo_id, enc in result.items():
171 assert enc.dead_count >= 0
172 assert enc.error_count >= 0
173 assert enc.warning_count >= 0
174
175
176 # ---------------------------------------------------------------------------
177 # T606 — health_status is a known literal
178 # ---------------------------------------------------------------------------
179
180 @pytest.mark.asyncio
181 async def test_t606_health_status_is_valid_literal(db_session: AsyncSession) -> None:
182 """T606: health_status is always one of {'clean', 'warn', 'risk'}."""
183 rich_id, empty_id = await _seed_two_repos(db_session)
184 result = await enrich_repo_cards(db_session, [rich_id, empty_id])
185 valid = {"clean", "warn", "risk"}
186 for repo_id, enc in result.items():
187 assert enc.health_status in valid, (
188 f"repo {repo_id}: unexpected health_status={enc.health_status!r}"
189 )
190
191
192 # ---------------------------------------------------------------------------
193 # T607 — hottest_symbol.churn_30d > 0
194 # ---------------------------------------------------------------------------
195
196 @pytest.mark.asyncio
197 async def test_t607_hottest_symbol_has_positive_churn(db_session: AsyncSession) -> None:
198 """T607: when hottest_symbol is not None its churn_30d is > 0."""
199 rich_id, _ = await _seed_two_repos(db_session)
200 result = await enrich_repo_cards(db_session, [rich_id])
201 enc = result[rich_id]
202 if enc.hottest_symbol is not None:
203 assert enc.hottest_symbol.churn_30d > 0, (
204 f"hottest_symbol has churn_30d=0: {enc.hottest_symbol.address}"
205 )
206
207
208 # ---------------------------------------------------------------------------
209 # T608 — blast_leader.blast > 0
210 # ---------------------------------------------------------------------------
211
212 @pytest.mark.asyncio
213 async def test_t608_blast_leader_has_positive_blast(db_session: AsyncSession) -> None:
214 """T608: when blast_leader is not None its blast score is > 0."""
215 rich_id, _ = await _seed_two_repos(db_session)
216 result = await enrich_repo_cards(db_session, [rich_id])
217 enc = result[rich_id]
218 if enc.blast_leader is not None:
219 assert enc.blast_leader.blast > 0, (
220 f"blast_leader has blast=0: {enc.blast_leader.address}"
221 )
222
223
224 # ---------------------------------------------------------------------------
225 # T609 — result keys match requested repo_ids exactly
226 # ---------------------------------------------------------------------------
227
228 @pytest.mark.asyncio
229 async def test_t609_result_keys_match_requested_ids(db_session: AsyncSession) -> None:
230 """T609: the returned dict has exactly the requested repo_ids as keys."""
231 repos = [await create_repo(db_session, visibility="public") for _ in range(10)]
232 repo_ids = [r.repo_id for r in repos]
233
234 result = await enrich_repo_cards(db_session, repo_ids)
235 assert set(result.keys()) == set(repo_ids)
236
237
238 # ---------------------------------------------------------------------------
239 # T610 — bucket colors are valid hex strings
240 # ---------------------------------------------------------------------------
241
242 @pytest.mark.asyncio
243 async def test_t610_pulse_bucket_colors_are_valid_hex(db_session: AsyncSession) -> None:
244 """T610: every bucket.color is a valid 6-digit lowercase hex color string."""
245 repo = await create_repo(db_session, visibility="public")
246 result = await enrich_repo_cards(db_session, [repo.repo_id])
247 for b in result[repo.repo_id].pulse_buckets:
248 assert _HEX_RE.match(b.color), f"invalid color {b.color!r} on {b.date}"
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago