gabriel / musehub public
test_proposal_reimagination_phase1.py python
682 lines 29.4 KB Cold
Raw
sha256:50b52eda7afb2f122863aef47d684d1a9e4684b48f5f95367fc956e28ceb7d42 refactor: rename merge strategy aliases to canonical names Sonnet 4.6 minor ⚠ breaking 140 days ago
1 """Phase 1 — Proposal reimagination model and ORM tests (issue #37).
2
3 Tier 1 — Unit (no DB)
4 - ProposalType, ProposalState, MergeStrategy enum values and str behaviour
5 - MergeConditions defaults and validation bounds
6 - ProposalCommentTarget target_type defaults and domain-specific fields
7 - DimensionalRiskVector type alias usage
8 - ProposalCreate with all new fields; defaults; round-trip JSON
9 - ProposalResponse carries new fields; defaults survive serialization
10
11 Tier 5 — Integration (DB required)
12 - MusehubProposal stores and round-trips all 15 new columns
13 - MusehubProposalReview stores and round-trips 3 new columns
14 - MusehubProposalDependency creates valid DAG edge; unique constraint fires on duplicate
15 - MusehubProposalSimulation caches result; unique constraint fires on (proposal, type)
16 - Dependency cascade: deleting a proposal cascades to its dependency edges
17 """
18
19 from __future__ import annotations
20
21 import uuid
22 from datetime import datetime, timezone
23
24 import pytest
25 from sqlalchemy.ext.asyncio import AsyncSession
26
27 from musehub.types.json_types import JSONObject
28 from musehub.models.musehub import (
29 DimensionalRiskVector,
30 MergeConditions,
31 MergeStrategy,
32 ProposalCommentTarget,
33 ProposalCreate,
34 ProposalResponse,
35 ProposalState,
36 ProposalType,
37 )
38
39
40 # ─────────────────────────────────────────────────────────────────────────────
41 # Helpers
42 # ─────────────────────────────────────────────────────────────────────────────
43
44
45 def _now() -> datetime:
46 return datetime.now(tz=timezone.utc)
47
48
49 def _uid() -> str:
50 return uuid.uuid4().hex[:12]
51
52
53 async def _make_repo(session: AsyncSession) -> str:
54 from musehub.core.genesis import compute_identity_id, compute_repo_id
55 from musehub.db.musehub_repo_models import MusehubRepo
56 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
57
58 owner = "p1tester"
59 slug = f"repo-{_uid()}"
60 owner_id = compute_identity_id(owner.encode())
61 created_at = _now()
62 repo = MusehubRepo(
63 repo_id=compute_repo_id(owner_id, slug, "code", created_at.isoformat()),
64 name=slug,
65 owner=owner,
66 slug=slug,
67 visibility="public",
68 owner_user_id=owner_id,
69 description="",
70 tags=[],
71 created_at=created_at,
72 )
73 session.add(repo)
74 await session.flush()
75 return repo.repo_id
76
77
78 async def _make_proposal(session: AsyncSession, repo_id: str, *, proposal_number: int = 1) -> "MusehubProposal":
79 from musehub.core.genesis import compute_identity_id, compute_proposal_id
80 from musehub.db.musehub_repo_models import MusehubRepo
81 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
82
83 author_id = compute_identity_id(b"p1tester")
84 created_at = _now()
85 proposal = MusehubProposal(
86 proposal_id=compute_proposal_id(repo_id, author_id, f"feat/{_uid()}", "dev", created_at.isoformat()),
87 repo_id=repo_id,
88 proposal_number=proposal_number,
89 title="feat: phase 1 test proposal",
90 from_branch=f"feat/{_uid()}",
91 to_branch="dev",
92 created_at=created_at,
93 )
94 session.add(proposal)
95 await session.flush()
96 return proposal
97
98
99 # ─────────────────────────────────────────────────────────────────────────────
100 # Tier 1 — Unit: enums
101 # ─────────────────────────────────────────────────────────────────────────────
102
103
104 class TestProposalTypeEnum:
105 def test_all_seven_values_exist(self) -> None:
106 expected = {
107 "merge", "stem_integration", "midi_evolution",
108 "payment_settlement", "agent_delegation", "identity_transition",
109 "canonical_release",
110 }
111 assert {m.value for m in ProposalType} == expected
112
113 def test_is_str_subclass(self) -> None:
114 assert isinstance(ProposalType.MERGE, str)
115 assert ProposalType.MERGE == "merge"
116
117 def test_json_serialises_as_plain_string(self) -> None:
118 import json
119 payload = json.dumps({"t": ProposalType.MIDI_EVOLUTION})
120 assert json.loads(payload)["t"] == "midi_evolution"
121
122 def test_no_legacy_prefixed_values(self) -> None:
123 """musehub#229 — migration 0070 canonicalised merge_strategy's
124 values but missed proposal_type, leaving STATE_MERGE = "state_merge"
125 as the one non-canonical member. Guards against this drift
126 recurring: no ProposalType value may use the old state_*/domain_*
127 naming MergeStrategy also used to have before its own canonicalisation.
128 """
129 for member in ProposalType:
130 assert not member.value.startswith("state_"), (
131 f"{member!r} uses legacy state_* naming — rename to the "
132 "plain-noun canonical form (see musehub#229)"
133 )
134 assert not member.value.startswith("domain_"), (
135 f"{member!r} uses legacy domain_* naming — rename to the "
136 "plain-noun canonical form (see musehub#229)"
137 )
138
139
140 class TestProposalStateEnum:
141 def test_all_seven_values_exist(self) -> None:
142 expected = {"drafting", "open", "in_review", "approved", "settling", "merged", "abandoned"}
143 assert {m.value for m in ProposalState} == expected
144
145 def test_is_str_subclass(self) -> None:
146 assert ProposalState.OPEN == "open"
147 assert ProposalState.MERGED != ProposalState.ABANDONED
148
149
150 class TestMergeStrategyEnum:
151 def test_all_values_exist(self) -> None:
152 # recursive + snapshot were added with the unified merge-engine strategy vocabulary.
153 expected = {"recursive", "overlay", "snapshot", "replay", "weave", "selective", "phased", "cherry_pick"}
154 assert {m.value for m in MergeStrategy} == expected
155
156 def test_is_str_subclass(self) -> None:
157 assert isinstance(MergeStrategy.PHASED, str)
158 assert MergeStrategy.SELECTIVE == "selective"
159
160
161 # ─────────────────────────────────────────────────────────────────────────────
162 # Tier 1 — Unit: MergeConditions
163 # ─────────────────────────────────────────────────────────────────────────────
164
165
166 class TestMergeConditions:
167 def test_defaults(self) -> None:
168 mc = MergeConditions()
169 assert mc.require_approvals == 2
170 assert mc.require_domains_approved == []
171 assert mc.max_risk_score == 1.0
172 assert mc.require_signed_commits is False
173 assert mc.require_no_breakage is False
174 assert mc.require_test_coverage is False
175 assert mc.require_payment_settled is False
176 assert mc.require_dependency_merged is True
177 assert mc.max_agent_commit_ratio == 1.0
178
179 def test_require_approvals_non_negative(self) -> None:
180 mc = MergeConditions(require_approvals=0)
181 assert mc.require_approvals == 0
182 with pytest.raises(Exception):
183 MergeConditions(require_approvals=-1)
184
185 def test_max_risk_score_clamped(self) -> None:
186 MergeConditions(max_risk_score=0.0)
187 MergeConditions(max_risk_score=1.0)
188 with pytest.raises(Exception):
189 MergeConditions(max_risk_score=1.1)
190 with pytest.raises(Exception):
191 MergeConditions(max_risk_score=-0.1)
192
193 def test_max_agent_commit_ratio_clamped(self) -> None:
194 MergeConditions(max_agent_commit_ratio=0.0)
195 MergeConditions(max_agent_commit_ratio=1.0)
196 with pytest.raises(Exception):
197 MergeConditions(max_agent_commit_ratio=1.5)
198
199 def test_require_domains_approved_list(self) -> None:
200 mc = MergeConditions(require_domains_approved=["code", "midi"])
201 assert mc.require_domains_approved == ["code", "midi"]
202
203 def test_round_trip_json(self) -> None:
204 mc = MergeConditions(require_approvals=3, require_no_breakage=True)
205 restored = MergeConditions.model_validate(mc.model_dump())
206 assert restored == mc
207
208
209 # ─────────────────────────────────────────────────────────────────────────────
210 # Tier 1 — Unit: ProposalCommentTarget
211 # ─────────────────────────────────────────────────────────────────────────────
212
213
214 class TestProposalCommentTarget:
215 def test_default_target_type_is_general(self) -> None:
216 t = ProposalCommentTarget()
217 assert t.target_type == "general"
218 assert t.symbol_address is None
219
220 def test_code_domain_fields(self) -> None:
221 t = ProposalCommentTarget(
222 target_type="code",
223 symbol_address="auth.py::AuthService.login",
224 line_start=42,
225 line_end=58,
226 )
227 assert t.symbol_address == "auth.py::AuthService.login"
228 assert t.line_start == 42
229 assert t.line_end == 58
230
231 def test_midi_domain_fields(self) -> None:
232 t = ProposalCommentTarget(
233 target_type="midi",
234 track_name="piano",
235 beat_start=1.0,
236 beat_end=5.0,
237 note_pitch=60,
238 )
239 assert t.track_name == "piano"
240 assert t.note_pitch == 60
241
242 def test_note_pitch_bounds(self) -> None:
243 ProposalCommentTarget(note_pitch=0)
244 ProposalCommentTarget(note_pitch=127)
245 with pytest.raises(Exception):
246 ProposalCommentTarget(note_pitch=128)
247 with pytest.raises(Exception):
248 ProposalCommentTarget(note_pitch=-1)
249
250 def test_payment_domain_field(self) -> None:
251 t = ProposalCommentTarget(target_type="payment", nonce_hex="deadbeef")
252 assert t.nonce_hex == "deadbeef"
253
254 def test_identity_domain_field(self) -> None:
255 t = ProposalCommentTarget(target_type="identity", identity_handle="gabriel")
256 assert t.identity_handle == "gabriel"
257
258 def test_stem_domain_fields(self) -> None:
259 t = ProposalCommentTarget(
260 target_type="stem",
261 stem_id="stem_abc123",
262 timestamp_start=0.5,
263 timestamp_end=4.0,
264 )
265 assert t.stem_id == "stem_abc123"
266
267
268 # ─────────────────────────────────────────────────────────────────────────────
269 # Tier 1 — Unit: DimensionalRiskVector
270 # ─────────────────────────────────────────────────────────────────────────────
271
272
273 class TestDimensionalRiskVector:
274 def test_is_dict_alias(self) -> None:
275 v: DimensionalRiskVector = {"code": 0.8, "midi": 0.3}
276 assert v["code"] == 0.8
277 assert isinstance(v, dict)
278
279 def test_empty_vector_is_valid(self) -> None:
280 v: DimensionalRiskVector = {}
281 assert len(v) == 0
282
283
284 # ─────────────────────────────────────────────────────────────────────────────
285 # Tier 1 — Unit: ProposalCreate with new fields
286 # ─────────────────────────────────────────────────────────────────────────────
287
288
289 class TestProposalCreateNewFields:
290 def _minimal(self) -> JSONObject:
291 return {"title": "feat: x", "from_branch": "feat/x", "to_branch": "dev"}
292
293 def test_defaults_for_new_fields(self) -> None:
294 pc = ProposalCreate(**self._minimal())
295 assert pc.proposal_type == ProposalType.MERGE
296 assert pc.is_draft is False
297 assert pc.merge_conditions is None
298 assert pc.merge_strategy == MergeStrategy.OVERLAY
299 assert pc.selective_domains is None
300 assert pc.depends_on == []
301
302 def test_proposal_type_set(self) -> None:
303 pc = ProposalCreate(**self._minimal(), proposal_type=ProposalType.MIDI_EVOLUTION)
304 assert pc.proposal_type == ProposalType.MIDI_EVOLUTION
305
306 def test_is_draft_set(self) -> None:
307 pc = ProposalCreate(**self._minimal(), is_draft=True)
308 assert pc.is_draft is True
309
310 def test_merge_conditions_set(self) -> None:
311 mc = MergeConditions(require_approvals=1, require_no_breakage=True)
312 pc = ProposalCreate(**self._minimal(), merge_conditions=mc)
313 assert pc.merge_conditions is not None
314 assert pc.merge_conditions.require_approvals == 1
315
316 def test_selective_strategy(self) -> None:
317 pc = ProposalCreate(
318 **self._minimal(),
319 merge_strategy=MergeStrategy.SELECTIVE,
320 selective_domains=["code"],
321 )
322 assert pc.merge_strategy == MergeStrategy.SELECTIVE
323 assert pc.selective_domains == ["code"]
324
325 def test_depends_on_list(self) -> None:
326 dep_id = "sha256:" + "a" * 64
327 pc = ProposalCreate(**self._minimal(), depends_on=[dep_id])
328 assert dep_id in pc.depends_on
329
330 def test_camel_json_round_trip(self) -> None:
331 mc = MergeConditions(require_approvals=3)
332 pc = ProposalCreate(
333 **self._minimal(),
334 proposal_type=ProposalType.PAYMENT_SETTLEMENT,
335 is_draft=True,
336 merge_conditions=mc,
337 merge_strategy=MergeStrategy.PHASED,
338 )
339 wire = pc.model_dump(by_alias=True)
340 assert wire["proposalType"] == "payment_settlement"
341 assert wire["isDraft"] is True
342 assert wire["mergeConditions"]["requireApprovals"] == 3
343
344 def test_existing_fields_unchanged(self) -> None:
345 pc = ProposalCreate(title="feat: x", from_branch="feat/x", to_branch="main", body="desc")
346 assert pc.title == "feat: x"
347 assert pc.body == "desc"
348
349
350 # ─────────────────────────────────────────────────────────────────────────────
351 # Tier 1 — Unit: ProposalResponse carries new fields
352 # ─────────────────────────────────────────────────────────────────────────────
353
354
355 class TestProposalResponseNewFields:
356 def _base(self) -> JSONObject:
357 return {
358 "proposal_id": "sha256:" + "b" * 64,
359 "title": "feat: x",
360 "body": "",
361 "state": "open",
362 "from_branch": "feat/x",
363 "to_branch": "dev",
364 "created_at": datetime(2026, 1, 1, tzinfo=timezone.utc),
365 }
366
367 def test_defaults_present(self) -> None:
368 pr = ProposalResponse(**self._base())
369 assert pr.proposal_type == ProposalType.MERGE
370 assert pr.is_draft is False
371 assert pr.merge_conditions is None
372 assert pr.merge_strategy == MergeStrategy.OVERLAY
373 assert pr.selective_domains is None
374 assert pr.depends_on == []
375 assert pr.risk_score is None
376 assert pr.dimensional_risk == {}
377
378 def test_dimensional_risk_round_trip(self) -> None:
379 pr = ProposalResponse(**self._base(), dimensional_risk={"code": 0.7, "midi": 0.2})
380 assert pr.dimensional_risk["code"] == 0.7
381
382 def test_risk_score_bounds(self) -> None:
383 ProposalResponse(**self._base(), risk_score=0.0)
384 ProposalResponse(**self._base(), risk_score=1.0)
385 with pytest.raises(Exception):
386 ProposalResponse(**self._base(), risk_score=1.1)
387
388
389 # ─────────────────────────────────────────────────────────────────────────────
390 # Tier 5 — Integration: ORM round-trips (require db_session fixture)
391 # ─────────────────────────────────────────────────────────────────────────────
392
393
394 class TestMusehubProposalORM:
395 @pytest.mark.asyncio
396 async def test_new_columns_default_values(self, db_session: AsyncSession) -> None:
397 from musehub.db.musehub_repo_models import MusehubRepo
398 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
399 repo_id = await _make_repo(db_session)
400 proposal = await _make_proposal(db_session, repo_id)
401 await db_session.refresh(proposal)
402
403 assert proposal.proposal_type == "merge"
404 assert proposal.is_draft is False
405 assert proposal.merge_conditions is None
406 assert proposal.merge_strategy == "overlay"
407 assert proposal.selective_domains is None
408 assert proposal.dimensional_risk == {}
409 assert proposal.midi_tracks_changed == 0
410 assert proposal.midi_notes_delta == 0
411 assert proposal.harmonic_tension_delta is None
412 assert proposal.payment_claim_count == 0
413 assert proposal.payment_ledger_delta_nano == 0
414 assert proposal.payment_avax_address is None
415 assert proposal.agent_model is None
416 assert proposal.agent_spawned_by is None
417
418 @pytest.mark.asyncio
419 async def test_new_columns_explicit_values(self, db_session: AsyncSession) -> None:
420 from musehub.core.genesis import compute_identity_id, compute_proposal_id
421 from musehub.db.musehub_repo_models import MusehubRepo
422 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
423
424 repo_id = await _make_repo(db_session)
425 author_id = compute_identity_id(b"p1tester")
426 created_at = _now()
427
428 proposal = MusehubProposal(
429 proposal_id=compute_proposal_id(repo_id, author_id, "feat/rich", "dev", created_at.isoformat()),
430 repo_id=repo_id,
431 proposal_number=2,
432 title="feat: rich proposal",
433 from_branch="feat/rich",
434 to_branch="dev",
435 created_at=created_at,
436 proposal_type="midi_evolution",
437 is_draft=True,
438 merge_conditions={"require_approvals": 1, "require_no_breakage": True},
439 merge_strategy="selective",
440 selective_domains=["code", "midi"],
441 dimensional_risk={"code": 0.8, "midi": 0.3},
442 midi_tracks_changed=4,
443 midi_notes_delta=-12,
444 harmonic_tension_delta=0.15,
445 payment_claim_count=3,
446 payment_ledger_delta_nano=500_000_000,
447 payment_avax_address="0xdeadbeef",
448 agent_model="claude-sonnet-4-6",
449 agent_spawned_by="gabriel",
450 )
451 db_session.add(proposal)
452 await db_session.flush()
453 await db_session.refresh(proposal)
454
455 assert proposal.proposal_type == "midi_evolution"
456 assert proposal.is_draft is True
457 assert proposal.merge_conditions["require_approvals"] == 1
458 assert proposal.merge_strategy == "selective"
459 assert proposal.selective_domains == ["code", "midi"]
460 assert proposal.dimensional_risk == {"code": 0.8, "midi": 0.3}
461 assert proposal.midi_tracks_changed == 4
462 assert proposal.midi_notes_delta == -12
463 assert abs(proposal.harmonic_tension_delta - 0.15) < 1e-6
464 assert proposal.payment_claim_count == 3
465 assert proposal.payment_ledger_delta_nano == 500_000_000
466 assert proposal.payment_avax_address == "0xdeadbeef"
467 assert proposal.agent_model == "claude-sonnet-4-6"
468 assert proposal.agent_spawned_by == "gabriel"
469
470
471 class TestMusehubProposalReviewORM:
472 @pytest.mark.asyncio
473 async def test_new_review_columns_defaults(self, db_session: AsyncSession) -> None:
474 from musehub.core.genesis import compute_identity_id, compute_review_id
475 from musehub.db.musehub_repo_models import MusehubRepo
476 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
477
478 repo_id = await _make_repo(db_session)
479 proposal = await _make_proposal(db_session, repo_id)
480
481 reviewer_id = compute_identity_id(b"reviewer1")
482 created_at = _now()
483 review = MusehubProposalReview(
484 review_id=compute_review_id(proposal.proposal_id, reviewer_id, created_at.isoformat()),
485 proposal_id=proposal.proposal_id,
486 reviewer_username="reviewer1",
487 )
488 db_session.add(review)
489 await db_session.flush()
490 await db_session.refresh(review)
491
492 assert review.reviewed_domains == []
493 assert review.domain_risk_acknowledged == {}
494 assert review.suggested_merge_strategy is None
495
496 @pytest.mark.asyncio
497 async def test_new_review_columns_explicit(self, db_session: AsyncSession) -> None:
498 from musehub.core.genesis import compute_identity_id, compute_review_id
499 from musehub.db.musehub_repo_models import MusehubRepo
500 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
501
502 repo_id = await _make_repo(db_session)
503 proposal = await _make_proposal(db_session, repo_id, proposal_number=2)
504
505 reviewer_id = compute_identity_id(b"reviewer2")
506 created_at = _now()
507 review = MusehubProposalReview(
508 review_id=compute_review_id(proposal.proposal_id, reviewer_id, created_at.isoformat()),
509 proposal_id=proposal.proposal_id,
510 reviewer_username="reviewer2",
511 reviewed_domains=["code", "midi"],
512 domain_risk_acknowledged={"code": True, "midi": False},
513 suggested_merge_strategy="weave",
514 )
515 db_session.add(review)
516 await db_session.flush()
517 await db_session.refresh(review)
518
519 assert review.reviewed_domains == ["code", "midi"]
520 assert review.domain_risk_acknowledged["code"] is True
521 assert review.suggested_merge_strategy == "weave"
522
523
524 class TestMusehubProposalDependencyORM:
525 @pytest.mark.asyncio
526 async def test_dependency_edge_created(self, db_session: AsyncSession) -> None:
527 from musehub.db.musehub_repo_models import MusehubRepo
528 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
529
530 repo_id = await _make_repo(db_session)
531 p1 = await _make_proposal(db_session, repo_id, proposal_number=1)
532 p2 = await _make_proposal(db_session, repo_id, proposal_number=2)
533
534 dep = MusehubProposalDependency(
535 dep_id=f"dep-{_uid()}",
536 dependent_proposal_id=p2.proposal_id,
537 dependency_proposal_id=p1.proposal_id,
538 )
539 db_session.add(dep)
540 await db_session.flush()
541 await db_session.refresh(dep)
542
543 assert dep.dependent_proposal_id == p2.proposal_id
544 assert dep.dependency_proposal_id == p1.proposal_id
545 assert dep.created_at is not None
546
547 @pytest.mark.asyncio
548 async def test_duplicate_edge_raises(self, db_session: AsyncSession) -> None:
549 from sqlalchemy.exc import IntegrityError
550 from musehub.db.musehub_repo_models import MusehubRepo
551 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
552
553 repo_id = await _make_repo(db_session)
554 p1 = await _make_proposal(db_session, repo_id, proposal_number=1)
555 p2 = await _make_proposal(db_session, repo_id, proposal_number=2)
556
557 dep_a = MusehubProposalDependency(
558 dep_id=f"dep-{_uid()}",
559 dependent_proposal_id=p2.proposal_id,
560 dependency_proposal_id=p1.proposal_id,
561 )
562 dep_b = MusehubProposalDependency(
563 dep_id=f"dep-{_uid()}",
564 dependent_proposal_id=p2.proposal_id,
565 dependency_proposal_id=p1.proposal_id,
566 )
567 db_session.add(dep_a)
568 db_session.add(dep_b)
569 with pytest.raises(IntegrityError):
570 await db_session.flush()
571
572 @pytest.mark.asyncio
573 async def test_cascade_delete_on_proposal_delete(self, db_session: AsyncSession) -> None:
574 from sqlalchemy import select
575 from musehub.db.musehub_repo_models import MusehubRepo
576 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
577
578 repo_id = await _make_repo(db_session)
579 p1 = await _make_proposal(db_session, repo_id, proposal_number=1)
580 p2 = await _make_proposal(db_session, repo_id, proposal_number=2)
581
582 dep_id = f"dep-{_uid()}"
583 dep = MusehubProposalDependency(
584 dep_id=dep_id,
585 dependent_proposal_id=p2.proposal_id,
586 dependency_proposal_id=p1.proposal_id,
587 )
588 db_session.add(dep)
589 await db_session.flush()
590
591 await db_session.delete(p1)
592 await db_session.flush()
593
594 result = await db_session.execute(
595 select(MusehubProposalDependency).where(
596 MusehubProposalDependency.dep_id == dep_id
597 )
598 )
599 assert result.scalar_one_or_none() is None
600
601
602 class TestMusehubProposalSimulationORM:
603 @pytest.mark.asyncio
604 async def test_simulation_stored_and_retrieved(self, db_session: AsyncSession) -> None:
605 from musehub.db.musehub_repo_models import MusehubRepo
606 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
607
608 repo_id = await _make_repo(db_session)
609 proposal = await _make_proposal(db_session, repo_id)
610
611 sim = MusehubProposalSimulation(
612 simulation_id=f"sim-{_uid()}",
613 proposal_id=proposal.proposal_id,
614 simulation_type="conflict_scan",
615 from_branch_commit_id="sha256:" + "c" * 64,
616 result={"conflicts": [], "safe_merge": True},
617 duration_ms=142,
618 )
619 db_session.add(sim)
620 await db_session.flush()
621 await db_session.refresh(sim)
622
623 assert sim.simulation_type == "conflict_scan"
624 assert sim.result["safe_merge"] is True
625 assert sim.duration_ms == 142
626 assert sim.expires_at is None
627
628 @pytest.mark.asyncio
629 async def test_duplicate_simulation_type_raises(self, db_session: AsyncSession) -> None:
630 from sqlalchemy.exc import IntegrityError
631 from musehub.db.musehub_repo_models import MusehubRepo
632 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
633
634 repo_id = await _make_repo(db_session)
635 proposal = await _make_proposal(db_session, repo_id, proposal_number=2)
636
637 commit_id = "sha256:" + "d" * 64
638 sim_a = MusehubProposalSimulation(
639 simulation_id=f"sim-{_uid()}",
640 proposal_id=proposal.proposal_id,
641 simulation_type="risk_projection",
642 from_branch_commit_id=commit_id,
643 result={},
644 )
645 sim_b = MusehubProposalSimulation(
646 simulation_id=f"sim-{_uid()}",
647 proposal_id=proposal.proposal_id,
648 simulation_type="risk_projection",
649 from_branch_commit_id=commit_id,
650 result={},
651 )
652 db_session.add(sim_a)
653 db_session.add(sim_b)
654 with pytest.raises(IntegrityError):
655 await db_session.flush()
656
657 @pytest.mark.asyncio
658 async def test_three_simulation_types_coexist(self, db_session: AsyncSession) -> None:
659 from musehub.db.musehub_repo_models import MusehubRepo
660 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
661
662 repo_id = await _make_repo(db_session)
663 proposal = await _make_proposal(db_session, repo_id, proposal_number=3)
664 commit_id = "sha256:" + "e" * 64
665
666 for sim_type in ("conflict_scan", "risk_projection", "dependency_order"):
667 db_session.add(MusehubProposalSimulation(
668 simulation_id=f"sim-{_uid()}",
669 proposal_id=proposal.proposal_id,
670 simulation_type=sim_type,
671 from_branch_commit_id=commit_id,
672 result={"type": sim_type},
673 ))
674 await db_session.flush()
675
676 from sqlalchemy import select, func
677 result = await db_session.execute(
678 select(func.count()).select_from(MusehubProposalSimulation).where(
679 MusehubProposalSimulation.proposal_id == proposal.proposal_id
680 )
681 )
682 assert result.scalar_one() == 3
File History 1 commit
sha256:50b52eda7afb2f122863aef47d684d1a9e4684b48f5f95367fc956e28ceb7d42 refactor: rename merge strategy aliases to canonical names Sonnet 4.6 minor ⚠ 140 days ago