gabriel / musehub public
test_proposal_list_phase1.py python
568 lines 23.8 KB
Raw
sha256:94ef169c149a452bff7c604ded8b280b19bd477c2dabcb56972780b0b784c7aa Merge 'fix/assignee-sigil-inline' into 'dev' — proposal: As… Human 1 day ago
1 """Tests for Phase 1 of issue #35 — proposal list enrichment models and service.
2
3 Tier 1 — Unit
4 Pure logic tests: _score_to_band, _enrich_one (via minimal stubs), model
5 field defaults, and ProposalListFilters validation. No DB, no async.
6
7 Tier 5 — Data Integrity
8 Invariant assertions over the enrichment layer: zero DB I/O per row after
9 prefetch, deterministic output for identical input, field-range constraints,
10 and correct behaviour under null / edge-case DB values.
11 """
12
13 from __future__ import annotations
14
15 import uuid
16 from datetime import datetime, timezone
17 from typing import Any
18 from unittest.mock import AsyncMock, MagicMock, patch
19
20 import pytest
21
22 from musehub.models.musehub import (
23 DomainHeatEntry,
24 DomainHeatResponse,
25 MergeReadinessResponse,
26 ProposalListEntry,
27 ProposalListFilters,
28 )
29 from musehub.services.musehub_proposals import (
30 _DEFAULT_REQUIRED_APPROVALS,
31 _DOMAIN_WEIGHTS,
32 _ProposalPrefetch,
33 _enrich_one,
34 _score_to_band,
35 )
36
37
38 # ─────────────────────────────────────────────────────────────────────────────
39 # Helpers
40 # ─────────────────────────────────────────────────────────────────────────────
41
42 def _make_proposal(
43 *,
44 proposal_id: str | None = None,
45 proposal_number: int = 1,
46 title: str = "feat: add thing",
47 state: str = "open",
48 from_branch: str = "feat/add-thing",
49 to_branch: str = "dev",
50 author: str = "gabriel",
51 risk_score: float = 0.0,
52 breakage_count: int = 0,
53 test_gap_count: int = 0,
54 symbols_changed: int = 0,
55 touched_symbols: list[str] | None = None,
56 merged_at: datetime | None = None,
57 proposal_type: str = "state_merge",
58 merge_strategy: str = "overlay",
59 agent_model: str | None = None,
60 agent_spawned_by: str | None = None,
61 payment_avax_address: str | None = None,
62 payment_claim_count: int = 0,
63 payment_ledger_delta_nano: int = 0,
64 midi_tracks_changed: int = 0,
65 midi_notes_delta: int = 0,
66 harmonic_tension_delta: float | None = None,
67 ) -> MagicMock:
68 """Return a minimal ORM-like stub for a MusehubProposal."""
69 p = MagicMock()
70 p.proposal_id = proposal_id or str(uuid.uuid4())
71 p.proposal_number = proposal_number
72 p.title = title
73 p.state = state
74 p.from_branch = from_branch
75 p.to_branch = to_branch
76 p.author = author
77 p.risk_score = risk_score
78 p.breakage_count = breakage_count
79 p.test_gap_count = test_gap_count
80 p.symbols_changed = symbols_changed
81 p.touched_symbols = touched_symbols or []
82 p.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
83 p.merged_at = merged_at
84 p.proposal_type = proposal_type
85 p.merge_strategy = merge_strategy
86 p.agent_model = agent_model
87 p.agent_spawned_by = agent_spawned_by
88 p.payment_avax_address = payment_avax_address
89 p.payment_claim_count = payment_claim_count
90 p.payment_ledger_delta_nano = payment_ledger_delta_nano
91 p.midi_tracks_changed = midi_tracks_changed
92 p.midi_notes_delta = midi_notes_delta
93 p.harmonic_tension_delta = harmonic_tension_delta
94 p.is_draft = state == "drafting"
95 return p
96
97
98 def _make_review(*, proposal_id: str, state: str = "approved") -> MagicMock:
99 r = MagicMock()
100 r.proposal_id = proposal_id
101 r.state = state
102 return r
103
104
105 def _empty_prefetch(proposal_id: str | None = None) -> _ProposalPrefetch:
106 pid = proposal_id or str(uuid.uuid4())
107 return _ProposalPrefetch(
108 reviews_by_proposal={pid: []},
109 author_types={},
110 )
111
112
113 def _prefetch_with_reviews(
114 proposal_id: str, reviews: list[MagicMock], *, author_type: str = "human"
115 ) -> _ProposalPrefetch:
116 return _ProposalPrefetch(
117 reviews_by_proposal={proposal_id: reviews},
118 author_types={"gabriel": author_type},
119 )
120
121
122 # ─────────────────────────────────────────────────────────────────────────────
123 # Tier 1 — Unit: _score_to_band
124 # ─────────────────────────────────────────────────────────────────────────────
125
126 class TestUnitScoreToBand:
127 """_score_to_band maps [0.0, 1.0] floats to band labels.
128
129 Tier 1 (Unit): pure function, no I/O.
130 """
131
132 def test_zero_returns_none(self) -> None:
133 assert _score_to_band(0.0) == "none"
134
135 def test_below_low_threshold_returns_none(self) -> None:
136 # anything under 0.01 is none
137 assert _score_to_band(0.005) == "none"
138
139 def test_exactly_low_threshold_returns_low(self) -> None:
140 assert _score_to_band(0.01) == "low"
141
142 def test_below_medium_threshold_returns_low(self) -> None:
143 assert _score_to_band(0.24) == "low"
144
145 def test_exactly_medium_threshold_returns_medium(self) -> None:
146 assert _score_to_band(0.25) == "medium"
147
148 def test_below_high_threshold_returns_medium(self) -> None:
149 assert _score_to_band(0.49) == "medium"
150
151 def test_exactly_high_threshold_returns_high(self) -> None:
152 assert _score_to_band(0.50) == "high"
153
154 def test_below_critical_threshold_returns_high(self) -> None:
155 assert _score_to_band(0.74) == "high"
156
157 def test_exactly_critical_threshold_returns_critical(self) -> None:
158 assert _score_to_band(0.75) == "critical"
159
160 def test_max_score_returns_critical(self) -> None:
161 assert _score_to_band(1.0) == "critical"
162
163 def test_mid_critical_range(self) -> None:
164 assert _score_to_band(0.9) == "critical"
165
166
167 # ─────────────────────────────────────────────────────────────────────────────
168 # Tier 1 — Unit: _enrich_one risk computation
169 # ─────────────────────────────────────────────────────────────────────────────
170
171 class TestUnitEnrichOneRisk:
172 """_enrich_one populates risk fields correctly from proposal.risk_score.
173
174 Tier 1 (Unit): synchronous, no DB.
175 """
176
177 def test_zero_risk_score_yields_no_active_domains(self) -> None:
178 p = _make_proposal(risk_score=0.0)
179 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
180 assert entry.active_domains == []
181 assert entry.domain_risk == {}
182 assert entry.aggregate_risk_score == 0.0
183 assert entry.aggregate_risk_band == "none"
184
185 def test_nonzero_risk_activates_code_domain(self) -> None:
186 p = _make_proposal(risk_score=0.6)
187 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
188 assert "code" in entry.active_domains
189 assert "code" in entry.domain_risk
190 assert entry.domain_risk["code"] == pytest.approx(0.6)
191
192 def test_aggregate_score_equals_code_when_only_domain(self) -> None:
193 p = _make_proposal(risk_score=0.8)
194 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
195 assert entry.aggregate_risk_score == pytest.approx(0.8)
196 assert entry.aggregate_risk_band == "critical"
197
198 def test_domain_risk_band_follows_score(self) -> None:
199 p = _make_proposal(risk_score=0.3)
200 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
201 assert entry.domain_risk_band["code"] == "medium"
202
203 def test_invalid_risk_score_raises_value_error(self) -> None:
204 p = _make_proposal(risk_score=1.5)
205 with pytest.raises(ValueError, match="risk_score"):
206 _enrich_one(p, _empty_prefetch(p.proposal_id))
207
208 def test_negative_risk_score_raises_value_error(self) -> None:
209 p = _make_proposal(risk_score=-0.1)
210 with pytest.raises(ValueError, match="risk_score"):
211 _enrich_one(p, _empty_prefetch(p.proposal_id))
212
213 def test_aggregate_score_rounded_to_4_decimal_places(self) -> None:
214 p = _make_proposal(risk_score=1 / 3)
215 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
216 # Must be rounded to 4dp; raw 1/3 has many more
217 assert entry.aggregate_risk_score == round(1 / 3, 4)
218
219
220 # ─────────────────────────────────────────────────────────────────────────────
221 # Tier 1 — Unit: _enrich_one review/approval logic
222 # ─────────────────────────────────────────────────────────────────────────────
223
224 class TestUnitEnrichOneApprovals:
225 """_enrich_one computes approval status from pre-fetched reviews.
226
227 Tier 1 (Unit): synchronous, no DB.
228 """
229
230 def test_no_reviews_zero_approvals(self) -> None:
231 p = _make_proposal()
232 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
233 assert entry.approval_count == 0
234 assert entry.all_merge_conditions_met is False
235
236 def test_one_approved_review_increments_count(self) -> None:
237 p = _make_proposal(risk_score=0.5)
238 review = _make_review(proposal_id=p.proposal_id)
239 pre = _prefetch_with_reviews(p.proposal_id, [review])
240 entry = _enrich_one(p, pre)
241 assert entry.approval_count == 1
242
243 def test_required_approvals_default_is_two(self) -> None:
244 p = _make_proposal()
245 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
246 assert entry.required_approvals == _DEFAULT_REQUIRED_APPROVALS
247
248 def test_all_merge_conditions_met_requires_approvals_and_no_breakage(self) -> None:
249 p = _make_proposal(breakage_count=0)
250 reviews = [
251 _make_review(proposal_id=p.proposal_id),
252 _make_review(proposal_id=p.proposal_id),
253 ]
254 pre = _prefetch_with_reviews(p.proposal_id, reviews)
255 entry = _enrich_one(p, pre)
256 assert entry.all_merge_conditions_met is True
257
258 def test_breakage_blocks_merge_even_with_enough_approvals(self) -> None:
259 p = _make_proposal(breakage_count=3)
260 reviews = [
261 _make_review(proposal_id=p.proposal_id),
262 _make_review(proposal_id=p.proposal_id),
263 ]
264 pre = _prefetch_with_reviews(p.proposal_id, reviews)
265 entry = _enrich_one(p, pre)
266 assert entry.all_merge_conditions_met is False
267
268 def test_non_approved_reviews_not_counted(self) -> None:
269 p = _make_proposal()
270 reviews = [
271 _make_review(proposal_id=p.proposal_id, state="changes_requested"),
272 _make_review(proposal_id=p.proposal_id, state="pending"),
273 ]
274 pre = _prefetch_with_reviews(p.proposal_id, reviews)
275 entry = _enrich_one(p, pre)
276 assert entry.approval_count == 0
277 assert entry.all_merge_conditions_met is False
278
279 def test_domains_approved_populated_when_code_active_and_approved(self) -> None:
280 p = _make_proposal(risk_score=0.4)
281 reviews = [_make_review(proposal_id=p.proposal_id)]
282 pre = _prefetch_with_reviews(p.proposal_id, reviews)
283 entry = _enrich_one(p, pre)
284 assert "code" in entry.domains_approved
285
286 def test_domains_pending_review_when_not_approved(self) -> None:
287 p = _make_proposal(risk_score=0.4)
288 pre = _prefetch_with_reviews(p.proposal_id, [])
289 entry = _enrich_one(p, pre)
290 assert "code" in entry.domains_pending_review
291 assert "code" not in entry.domains_approved
292
293
294 # ─────────────────────────────────────────────────────────────────────────────
295 # Tier 1 — Unit: _enrich_one misc fields
296 # ─────────────────────────────────────────────────────────────────────────────
297
298 class TestUnitEnrichOneMiscFields:
299 """_enrich_one populates metadata fields correctly.
300
301 Tier 1 (Unit): synchronous, no DB.
302 """
303
304 def test_title_passthrough_under_limit(self) -> None:
305 p = _make_proposal(title="short title")
306 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
307 assert entry.title == "short title"
308
309 def test_title_truncated_at_80_chars(self) -> None:
310 long_title = "x" * 100
311 p = _make_proposal(title=long_title)
312 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
313 assert len(entry.title) <= 82 # 80 chars + "…"
314 assert entry.title.endswith("…")
315
316 def test_is_draft_true_for_drafting_state(self) -> None:
317 p = _make_proposal(state="drafting")
318 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
319 assert entry.is_draft is True
320
321 def test_is_draft_false_for_open_state(self) -> None:
322 p = _make_proposal(state="open")
323 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
324 assert entry.is_draft is False
325
326 def test_author_type_resolved_from_prefetch(self) -> None:
327 p = _make_proposal(author="gabriel")
328 pre = _ProposalPrefetch(
329 reviews_by_proposal={p.proposal_id: []},
330 author_types={"gabriel": "agent"},
331 )
332 entry = _enrich_one(p, pre)
333 assert entry.author_type == "agent"
334
335 def test_author_type_defaults_to_human_if_missing(self) -> None:
336 p = _make_proposal(author="unknown-handle")
337 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
338 assert entry.author_type == "human"
339
340 def test_touched_symbols_preview_capped_at_three(self) -> None:
341 symbols = ["a::Fn", "b::Fn", "c::Fn", "d::Fn", "e::Fn"]
342 p = _make_proposal(touched_symbols=symbols)
343 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
344 assert entry.touched_symbols_preview == symbols[:3]
345
346 def test_touched_symbols_preview_empty_list(self) -> None:
347 p = _make_proposal(touched_symbols=[])
348 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
349 assert entry.touched_symbols_preview == []
350
351 def test_is_blocked_false_by_default(self) -> None:
352 p = _make_proposal()
353 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
354 assert entry.is_blocked is False
355 assert entry.blocked_by == []
356 assert entry.blocks == []
357
358
359 # ─────────────────────────────────────────────────────────────────────────────
360 # Tier 1 — Unit: ProposalListFilters validation
361 # ─────────────────────────────────────────────────────────────────────────────
362
363 class TestUnitProposalListFilters:
364 """ProposalListFilters enforces field constraints via Pydantic.
365
366 Tier 1 (Unit): pure Pydantic validation, no DB.
367 """
368
369 def test_default_state_is_open(self) -> None:
370 f = ProposalListFilters()
371 assert f.state == "open"
372
373 def test_valid_states_accepted(self) -> None:
374 for state in ("open", "in_review", "approved", "drafting", "settling", "merged", "abandoned", "all"):
375 f = ProposalListFilters(state=state)
376 assert f.state == state
377
378 def test_invalid_state_raises(self) -> None:
379 from pydantic import ValidationError
380 with pytest.raises(ValidationError):
381 ProposalListFilters(state="nonsense")
382
383 def test_default_limit_is_20(self) -> None:
384 f = ProposalListFilters()
385 assert f.limit == 20
386
387 def test_limit_min_is_1(self) -> None:
388 from pydantic import ValidationError
389 with pytest.raises(ValidationError):
390 ProposalListFilters(limit=0)
391
392 def test_limit_max_is_100(self) -> None:
393 from pydantic import ValidationError
394 with pytest.raises(ValidationError):
395 ProposalListFilters(limit=101)
396
397 def test_default_sort_is_newest(self) -> None:
398 f = ProposalListFilters()
399 assert f.sort == "newest"
400
401 def test_valid_sorts_accepted(self) -> None:
402 for sort in ("newest", "oldest", "risk_desc", "risk_asc", "merge_ready_first"):
403 f = ProposalListFilters(sort=sort)
404 assert f.sort == sort
405
406 def test_invalid_sort_raises(self) -> None:
407 from pydantic import ValidationError
408 with pytest.raises(ValidationError):
409 ProposalListFilters(sort="random")
410
411 def test_default_author_type_is_all(self) -> None:
412 f = ProposalListFilters()
413 assert f.author_type == "all"
414
415 def test_cursor_is_none_by_default(self) -> None:
416 f = ProposalListFilters()
417 assert f.cursor is None
418
419 def test_assigned_reviewer_pattern_rejects_spaces(self) -> None:
420 from pydantic import ValidationError
421 with pytest.raises(ValidationError):
422 ProposalListFilters(assigned_reviewer="bad handle")
423
424
425 # ─────────────────────────────────────────────────────────────────────────────
426 # Tier 1 — Unit: DomainHeatResponse and MergeReadinessResponse defaults
427 # ─────────────────────────────────────────────────────────────────────────────
428
429 class TestUnitResponseDefaults:
430 """Response model default values.
431
432 Tier 1 (Unit): pure Pydantic instantiation.
433 """
434
435 def test_domain_heat_response_defaults(self) -> None:
436 r = DomainHeatResponse()
437 assert r.domains == {}
438 assert r.total_open == 0
439
440 def test_domain_heat_entry_defaults(self) -> None:
441 e = DomainHeatEntry()
442 assert e.count == 0
443 assert e.avg_risk == 0.0
444
445 def test_merge_readiness_response_defaults(self) -> None:
446 r = MergeReadinessResponse()
447 assert r.ready == []
448 assert r.blocked == []
449 assert r.settling == []
450 assert r.needs_review == []
451
452 def test_domain_heat_entry_rejects_negative_count(self) -> None:
453 from pydantic import ValidationError
454 with pytest.raises(ValidationError):
455 DomainHeatEntry(count=-1)
456
457 def test_domain_heat_entry_rejects_risk_above_1(self) -> None:
458 from pydantic import ValidationError
459 with pytest.raises(ValidationError):
460 DomainHeatEntry(avg_risk=1.1)
461
462
463 # ─────────────────────────────────────────────────────────────────────────────
464 # Tier 5 — Data Integrity: invariant assertions
465 # ─────────────────────────────────────────────────────────────────────────────
466
467 class TestDataIntegrityEnrichOne:
468 """_enrich_one output satisfies domain-level invariants for all inputs.
469
470 Tier 5 (Data Integrity): no DB; asserts invariants hold over ranges of
471 inputs rather than specific expected values.
472 """
473
474 def test_aggregate_risk_score_always_in_0_1(self) -> None:
475 for score in (0.0, 0.01, 0.25, 0.5, 0.75, 1.0):
476 p = _make_proposal(risk_score=score)
477 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
478 assert 0.0 <= entry.aggregate_risk_score <= 1.0, (
479 f"out of range for risk_score={score}"
480 )
481
482 def test_domain_risk_values_always_in_0_1(self) -> None:
483 for score in (0.0, 0.5, 1.0):
484 p = _make_proposal(risk_score=score)
485 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
486 for v in entry.domain_risk.values():
487 assert 0.0 <= v <= 1.0
488
489 def test_active_domains_subset_of_domain_risk_keys(self) -> None:
490 for score in (0.0, 0.3, 0.8):
491 p = _make_proposal(risk_score=score)
492 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
493 assert set(entry.active_domains) == set(entry.domain_risk.keys())
494
495 def test_domains_approved_subset_of_active_domains(self) -> None:
496 for score in (0.0, 0.5, 1.0):
497 p = _make_proposal(risk_score=score)
498 reviews = [_make_review(proposal_id=p.proposal_id)]
499 pre = _prefetch_with_reviews(p.proposal_id, reviews)
500 entry = _enrich_one(p, pre)
501 assert set(entry.domains_approved).issubset(set(entry.active_domains))
502
503 def test_domains_pending_review_complement_of_approved_within_active(self) -> None:
504 for score in (0.0, 0.5):
505 p = _make_proposal(risk_score=score)
506 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
507 active = set(entry.active_domains)
508 approved = set(entry.domains_approved)
509 pending = set(entry.domains_pending_review)
510 assert approved | pending == active
511 assert approved & pending == set()
512
513 def test_deterministic_output_same_input(self) -> None:
514 p = _make_proposal(risk_score=0.6, breakage_count=1)
515 pre = _empty_prefetch(p.proposal_id)
516 entry1 = _enrich_one(p, pre)
517 entry2 = _enrich_one(p, pre)
518 assert entry1.model_dump() == entry2.model_dump()
519
520 def test_proposal_id_preserved_exactly(self) -> None:
521 pid = str(uuid.uuid4())
522 p = _make_proposal(proposal_id=pid)
523 entry = _enrich_one(p, _empty_prefetch(pid))
524 assert entry.proposal_id == pid
525
526 def test_touched_symbols_preview_never_longer_than_3(self) -> None:
527 for n in range(6):
528 symbols = [f"file.py::Fn{i}" for i in range(n)]
529 p = _make_proposal(touched_symbols=symbols)
530 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
531 assert len(entry.touched_symbols_preview) <= 3
532
533 def test_aggregate_band_consistent_with_score(self) -> None:
534 for score in (0.0, 0.1, 0.3, 0.55, 0.8, 1.0):
535 p = _make_proposal(risk_score=score)
536 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
537 expected = _score_to_band(entry.aggregate_risk_score)
538 assert entry.aggregate_risk_band == expected, (
539 f"band mismatch at score={score}"
540 )
541
542 def test_merge_condition_met_only_when_no_breakage_and_sufficient_approvals(self) -> None:
543 cases: list[tuple[int, int, bool]] = [
544 (0, 2, True),
545 (0, 1, False),
546 (1, 2, False),
547 (3, 3, False),
548 ]
549 for breakage, n_approvals, expected in cases:
550 p = _make_proposal(breakage_count=breakage)
551 reviews = [_make_review(proposal_id=p.proposal_id) for _ in range(n_approvals)]
552 pre = _prefetch_with_reviews(p.proposal_id, reviews)
553 entry = _enrich_one(p, pre)
554 assert entry.all_merge_conditions_met is expected, (
555 f"breakage={breakage} approvals={n_approvals} expected={expected}"
556 )
557
558 def test_null_touched_symbols_treated_as_empty(self) -> None:
559 p = _make_proposal(touched_symbols=None)
560 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
561 assert entry.touched_symbols_preview == []
562
563 def test_null_risk_score_treated_as_zero(self) -> None:
564 p = _make_proposal(risk_score=0.0)
565 p.risk_score = None
566 entry = _enrich_one(p, _empty_prefetch(p.proposal_id))
567 assert entry.aggregate_risk_score == 0.0
568 assert entry.aggregate_risk_band == "none"
File History 3 commits
sha256:94ef169c149a452bff7c604ded8b280b19bd477c2dabcb56972780b0b784c7aa Merge 'fix/assignee-sigil-inline' into 'dev' — proposal: As… Human 1 day ago
sha256:6b1949fc2797ca4c1936a637a4cbfec828ef56cf52398a2e74ca3c4f494e728f fix: use wire_bytes not mpack_bytes_raw in compute_object_b… Sonnet 4.6 patch 9 days ago
sha256:4aed3d8601c8dd3ed37074de35f11f4a9699a0a4b99d43727048fd3f8e6fd13d chore: doc sweep, ignore wrangler build state, misc fixes Sonnet 4.6 minor ⚠ 12 days ago