gabriel / musehub public
test_proposal_reimagination_phase1.py python
665 lines 28.6 KB
Raw
sha256:50b52eda7afb2f122863aef47d684d1a9e4684b48f5f95367fc956e28ceb7d42 refactor: rename merge strategy aliases to canonical names Sonnet 4.6 minor ⚠ breaking 10 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 "state_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.STATE_MERGE, str)
115 assert ProposalType.STATE_MERGE == "state_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
123 class TestProposalStateEnum:
124 def test_all_seven_values_exist(self) -> None:
125 expected = {"drafting", "open", "in_review", "approved", "settling", "merged", "abandoned"}
126 assert {m.value for m in ProposalState} == expected
127
128 def test_is_str_subclass(self) -> None:
129 assert ProposalState.OPEN == "open"
130 assert ProposalState.MERGED != ProposalState.ABANDONED
131
132
133 class TestMergeStrategyEnum:
134 def test_all_values_exist(self) -> None:
135 # recursive + snapshot were added with the unified merge-engine strategy vocabulary.
136 expected = {"recursive", "overlay", "snapshot", "replay", "weave", "selective", "phased", "cherry_pick"}
137 assert {m.value for m in MergeStrategy} == expected
138
139 def test_is_str_subclass(self) -> None:
140 assert isinstance(MergeStrategy.PHASED, str)
141 assert MergeStrategy.SELECTIVE == "selective"
142
143
144 # ─────────────────────────────────────────────────────────────────────────────
145 # Tier 1 — Unit: MergeConditions
146 # ─────────────────────────────────────────────────────────────────────────────
147
148
149 class TestMergeConditions:
150 def test_defaults(self) -> None:
151 mc = MergeConditions()
152 assert mc.require_approvals == 2
153 assert mc.require_domains_approved == []
154 assert mc.max_risk_score == 1.0
155 assert mc.require_signed_commits is False
156 assert mc.require_no_breakage is False
157 assert mc.require_test_coverage is False
158 assert mc.require_payment_settled is False
159 assert mc.require_dependency_merged is True
160 assert mc.max_agent_commit_ratio == 1.0
161
162 def test_require_approvals_non_negative(self) -> None:
163 mc = MergeConditions(require_approvals=0)
164 assert mc.require_approvals == 0
165 with pytest.raises(Exception):
166 MergeConditions(require_approvals=-1)
167
168 def test_max_risk_score_clamped(self) -> None:
169 MergeConditions(max_risk_score=0.0)
170 MergeConditions(max_risk_score=1.0)
171 with pytest.raises(Exception):
172 MergeConditions(max_risk_score=1.1)
173 with pytest.raises(Exception):
174 MergeConditions(max_risk_score=-0.1)
175
176 def test_max_agent_commit_ratio_clamped(self) -> None:
177 MergeConditions(max_agent_commit_ratio=0.0)
178 MergeConditions(max_agent_commit_ratio=1.0)
179 with pytest.raises(Exception):
180 MergeConditions(max_agent_commit_ratio=1.5)
181
182 def test_require_domains_approved_list(self) -> None:
183 mc = MergeConditions(require_domains_approved=["code", "midi"])
184 assert mc.require_domains_approved == ["code", "midi"]
185
186 def test_round_trip_json(self) -> None:
187 mc = MergeConditions(require_approvals=3, require_no_breakage=True)
188 restored = MergeConditions.model_validate(mc.model_dump())
189 assert restored == mc
190
191
192 # ─────────────────────────────────────────────────────────────────────────────
193 # Tier 1 — Unit: ProposalCommentTarget
194 # ─────────────────────────────────────────────────────────────────────────────
195
196
197 class TestProposalCommentTarget:
198 def test_default_target_type_is_general(self) -> None:
199 t = ProposalCommentTarget()
200 assert t.target_type == "general"
201 assert t.symbol_address is None
202
203 def test_code_domain_fields(self) -> None:
204 t = ProposalCommentTarget(
205 target_type="code",
206 symbol_address="auth.py::AuthService.login",
207 line_start=42,
208 line_end=58,
209 )
210 assert t.symbol_address == "auth.py::AuthService.login"
211 assert t.line_start == 42
212 assert t.line_end == 58
213
214 def test_midi_domain_fields(self) -> None:
215 t = ProposalCommentTarget(
216 target_type="midi",
217 track_name="piano",
218 beat_start=1.0,
219 beat_end=5.0,
220 note_pitch=60,
221 )
222 assert t.track_name == "piano"
223 assert t.note_pitch == 60
224
225 def test_note_pitch_bounds(self) -> None:
226 ProposalCommentTarget(note_pitch=0)
227 ProposalCommentTarget(note_pitch=127)
228 with pytest.raises(Exception):
229 ProposalCommentTarget(note_pitch=128)
230 with pytest.raises(Exception):
231 ProposalCommentTarget(note_pitch=-1)
232
233 def test_payment_domain_field(self) -> None:
234 t = ProposalCommentTarget(target_type="payment", nonce_hex="deadbeef")
235 assert t.nonce_hex == "deadbeef"
236
237 def test_identity_domain_field(self) -> None:
238 t = ProposalCommentTarget(target_type="identity", identity_handle="gabriel")
239 assert t.identity_handle == "gabriel"
240
241 def test_stem_domain_fields(self) -> None:
242 t = ProposalCommentTarget(
243 target_type="stem",
244 stem_id="stem_abc123",
245 timestamp_start=0.5,
246 timestamp_end=4.0,
247 )
248 assert t.stem_id == "stem_abc123"
249
250
251 # ─────────────────────────────────────────────────────────────────────────────
252 # Tier 1 — Unit: DimensionalRiskVector
253 # ─────────────────────────────────────────────────────────────────────────────
254
255
256 class TestDimensionalRiskVector:
257 def test_is_dict_alias(self) -> None:
258 v: DimensionalRiskVector = {"code": 0.8, "midi": 0.3}
259 assert v["code"] == 0.8
260 assert isinstance(v, dict)
261
262 def test_empty_vector_is_valid(self) -> None:
263 v: DimensionalRiskVector = {}
264 assert len(v) == 0
265
266
267 # ─────────────────────────────────────────────────────────────────────────────
268 # Tier 1 — Unit: ProposalCreate with new fields
269 # ─────────────────────────────────────────────────────────────────────────────
270
271
272 class TestProposalCreateNewFields:
273 def _minimal(self) -> JSONObject:
274 return {"title": "feat: x", "from_branch": "feat/x", "to_branch": "dev"}
275
276 def test_defaults_for_new_fields(self) -> None:
277 pc = ProposalCreate(**self._minimal())
278 assert pc.proposal_type == ProposalType.STATE_MERGE
279 assert pc.is_draft is False
280 assert pc.merge_conditions is None
281 assert pc.merge_strategy == MergeStrategy.OVERLAY
282 assert pc.selective_domains is None
283 assert pc.depends_on == []
284
285 def test_proposal_type_set(self) -> None:
286 pc = ProposalCreate(**self._minimal(), proposal_type=ProposalType.MIDI_EVOLUTION)
287 assert pc.proposal_type == ProposalType.MIDI_EVOLUTION
288
289 def test_is_draft_set(self) -> None:
290 pc = ProposalCreate(**self._minimal(), is_draft=True)
291 assert pc.is_draft is True
292
293 def test_merge_conditions_set(self) -> None:
294 mc = MergeConditions(require_approvals=1, require_no_breakage=True)
295 pc = ProposalCreate(**self._minimal(), merge_conditions=mc)
296 assert pc.merge_conditions is not None
297 assert pc.merge_conditions.require_approvals == 1
298
299 def test_selective_strategy(self) -> None:
300 pc = ProposalCreate(
301 **self._minimal(),
302 merge_strategy=MergeStrategy.SELECTIVE,
303 selective_domains=["code"],
304 )
305 assert pc.merge_strategy == MergeStrategy.SELECTIVE
306 assert pc.selective_domains == ["code"]
307
308 def test_depends_on_list(self) -> None:
309 dep_id = "sha256:" + "a" * 64
310 pc = ProposalCreate(**self._minimal(), depends_on=[dep_id])
311 assert dep_id in pc.depends_on
312
313 def test_camel_json_round_trip(self) -> None:
314 mc = MergeConditions(require_approvals=3)
315 pc = ProposalCreate(
316 **self._minimal(),
317 proposal_type=ProposalType.PAYMENT_SETTLEMENT,
318 is_draft=True,
319 merge_conditions=mc,
320 merge_strategy=MergeStrategy.PHASED,
321 )
322 wire = pc.model_dump(by_alias=True)
323 assert wire["proposalType"] == "payment_settlement"
324 assert wire["isDraft"] is True
325 assert wire["mergeConditions"]["requireApprovals"] == 3
326
327 def test_existing_fields_unchanged(self) -> None:
328 pc = ProposalCreate(title="feat: x", from_branch="feat/x", to_branch="main", body="desc")
329 assert pc.title == "feat: x"
330 assert pc.body == "desc"
331
332
333 # ─────────────────────────────────────────────────────────────────────────────
334 # Tier 1 — Unit: ProposalResponse carries new fields
335 # ─────────────────────────────────────────────────────────────────────────────
336
337
338 class TestProposalResponseNewFields:
339 def _base(self) -> JSONObject:
340 return {
341 "proposal_id": "sha256:" + "b" * 64,
342 "title": "feat: x",
343 "body": "",
344 "state": "open",
345 "from_branch": "feat/x",
346 "to_branch": "dev",
347 "created_at": datetime(2026, 1, 1, tzinfo=timezone.utc),
348 }
349
350 def test_defaults_present(self) -> None:
351 pr = ProposalResponse(**self._base())
352 assert pr.proposal_type == ProposalType.STATE_MERGE
353 assert pr.is_draft is False
354 assert pr.merge_conditions is None
355 assert pr.merge_strategy == MergeStrategy.OVERLAY
356 assert pr.selective_domains is None
357 assert pr.depends_on == []
358 assert pr.risk_score is None
359 assert pr.dimensional_risk == {}
360
361 def test_dimensional_risk_round_trip(self) -> None:
362 pr = ProposalResponse(**self._base(), dimensional_risk={"code": 0.7, "midi": 0.2})
363 assert pr.dimensional_risk["code"] == 0.7
364
365 def test_risk_score_bounds(self) -> None:
366 ProposalResponse(**self._base(), risk_score=0.0)
367 ProposalResponse(**self._base(), risk_score=1.0)
368 with pytest.raises(Exception):
369 ProposalResponse(**self._base(), risk_score=1.1)
370
371
372 # ─────────────────────────────────────────────────────────────────────────────
373 # Tier 5 — Integration: ORM round-trips (require db_session fixture)
374 # ─────────────────────────────────────────────────────────────────────────────
375
376
377 class TestMusehubProposalORM:
378 @pytest.mark.asyncio
379 async def test_new_columns_default_values(self, db_session: AsyncSession) -> None:
380 from musehub.db.musehub_repo_models import MusehubRepo
381 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
382 repo_id = await _make_repo(db_session)
383 proposal = await _make_proposal(db_session, repo_id)
384 await db_session.refresh(proposal)
385
386 assert proposal.proposal_type == "state_merge"
387 assert proposal.is_draft is False
388 assert proposal.merge_conditions is None
389 assert proposal.merge_strategy == "overlay"
390 assert proposal.selective_domains is None
391 assert proposal.dimensional_risk == {}
392 assert proposal.midi_tracks_changed == 0
393 assert proposal.midi_notes_delta == 0
394 assert proposal.harmonic_tension_delta is None
395 assert proposal.payment_claim_count == 0
396 assert proposal.payment_ledger_delta_nano == 0
397 assert proposal.payment_avax_address is None
398 assert proposal.agent_model is None
399 assert proposal.agent_spawned_by is None
400
401 @pytest.mark.asyncio
402 async def test_new_columns_explicit_values(self, db_session: AsyncSession) -> None:
403 from musehub.core.genesis import compute_identity_id, compute_proposal_id
404 from musehub.db.musehub_repo_models import MusehubRepo
405 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
406
407 repo_id = await _make_repo(db_session)
408 author_id = compute_identity_id(b"p1tester")
409 created_at = _now()
410
411 proposal = MusehubProposal(
412 proposal_id=compute_proposal_id(repo_id, author_id, "feat/rich", "dev", created_at.isoformat()),
413 repo_id=repo_id,
414 proposal_number=2,
415 title="feat: rich proposal",
416 from_branch="feat/rich",
417 to_branch="dev",
418 created_at=created_at,
419 proposal_type="midi_evolution",
420 is_draft=True,
421 merge_conditions={"require_approvals": 1, "require_no_breakage": True},
422 merge_strategy="selective",
423 selective_domains=["code", "midi"],
424 dimensional_risk={"code": 0.8, "midi": 0.3},
425 midi_tracks_changed=4,
426 midi_notes_delta=-12,
427 harmonic_tension_delta=0.15,
428 payment_claim_count=3,
429 payment_ledger_delta_nano=500_000_000,
430 payment_avax_address="0xdeadbeef",
431 agent_model="claude-sonnet-4-6",
432 agent_spawned_by="gabriel",
433 )
434 db_session.add(proposal)
435 await db_session.flush()
436 await db_session.refresh(proposal)
437
438 assert proposal.proposal_type == "midi_evolution"
439 assert proposal.is_draft is True
440 assert proposal.merge_conditions["require_approvals"] == 1
441 assert proposal.merge_strategy == "selective"
442 assert proposal.selective_domains == ["code", "midi"]
443 assert proposal.dimensional_risk == {"code": 0.8, "midi": 0.3}
444 assert proposal.midi_tracks_changed == 4
445 assert proposal.midi_notes_delta == -12
446 assert abs(proposal.harmonic_tension_delta - 0.15) < 1e-6
447 assert proposal.payment_claim_count == 3
448 assert proposal.payment_ledger_delta_nano == 500_000_000
449 assert proposal.payment_avax_address == "0xdeadbeef"
450 assert proposal.agent_model == "claude-sonnet-4-6"
451 assert proposal.agent_spawned_by == "gabriel"
452
453
454 class TestMusehubProposalReviewORM:
455 @pytest.mark.asyncio
456 async def test_new_review_columns_defaults(self, db_session: AsyncSession) -> None:
457 from musehub.core.genesis import compute_identity_id, compute_review_id
458 from musehub.db.musehub_repo_models import MusehubRepo
459 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
460
461 repo_id = await _make_repo(db_session)
462 proposal = await _make_proposal(db_session, repo_id)
463
464 reviewer_id = compute_identity_id(b"reviewer1")
465 created_at = _now()
466 review = MusehubProposalReview(
467 review_id=compute_review_id(proposal.proposal_id, reviewer_id, created_at.isoformat()),
468 proposal_id=proposal.proposal_id,
469 reviewer_username="reviewer1",
470 )
471 db_session.add(review)
472 await db_session.flush()
473 await db_session.refresh(review)
474
475 assert review.reviewed_domains == []
476 assert review.domain_risk_acknowledged == {}
477 assert review.suggested_merge_strategy is None
478
479 @pytest.mark.asyncio
480 async def test_new_review_columns_explicit(self, db_session: AsyncSession) -> None:
481 from musehub.core.genesis import compute_identity_id, compute_review_id
482 from musehub.db.musehub_repo_models import MusehubRepo
483 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
484
485 repo_id = await _make_repo(db_session)
486 proposal = await _make_proposal(db_session, repo_id, proposal_number=2)
487
488 reviewer_id = compute_identity_id(b"reviewer2")
489 created_at = _now()
490 review = MusehubProposalReview(
491 review_id=compute_review_id(proposal.proposal_id, reviewer_id, created_at.isoformat()),
492 proposal_id=proposal.proposal_id,
493 reviewer_username="reviewer2",
494 reviewed_domains=["code", "midi"],
495 domain_risk_acknowledged={"code": True, "midi": False},
496 suggested_merge_strategy="weave",
497 )
498 db_session.add(review)
499 await db_session.flush()
500 await db_session.refresh(review)
501
502 assert review.reviewed_domains == ["code", "midi"]
503 assert review.domain_risk_acknowledged["code"] is True
504 assert review.suggested_merge_strategy == "weave"
505
506
507 class TestMusehubProposalDependencyORM:
508 @pytest.mark.asyncio
509 async def test_dependency_edge_created(self, db_session: AsyncSession) -> None:
510 from musehub.db.musehub_repo_models import MusehubRepo
511 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
512
513 repo_id = await _make_repo(db_session)
514 p1 = await _make_proposal(db_session, repo_id, proposal_number=1)
515 p2 = await _make_proposal(db_session, repo_id, proposal_number=2)
516
517 dep = MusehubProposalDependency(
518 dep_id=f"dep-{_uid()}",
519 dependent_proposal_id=p2.proposal_id,
520 dependency_proposal_id=p1.proposal_id,
521 )
522 db_session.add(dep)
523 await db_session.flush()
524 await db_session.refresh(dep)
525
526 assert dep.dependent_proposal_id == p2.proposal_id
527 assert dep.dependency_proposal_id == p1.proposal_id
528 assert dep.created_at is not None
529
530 @pytest.mark.asyncio
531 async def test_duplicate_edge_raises(self, db_session: AsyncSession) -> None:
532 from sqlalchemy.exc import IntegrityError
533 from musehub.db.musehub_repo_models import MusehubRepo
534 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
535
536 repo_id = await _make_repo(db_session)
537 p1 = await _make_proposal(db_session, repo_id, proposal_number=1)
538 p2 = await _make_proposal(db_session, repo_id, proposal_number=2)
539
540 dep_a = MusehubProposalDependency(
541 dep_id=f"dep-{_uid()}",
542 dependent_proposal_id=p2.proposal_id,
543 dependency_proposal_id=p1.proposal_id,
544 )
545 dep_b = MusehubProposalDependency(
546 dep_id=f"dep-{_uid()}",
547 dependent_proposal_id=p2.proposal_id,
548 dependency_proposal_id=p1.proposal_id,
549 )
550 db_session.add(dep_a)
551 db_session.add(dep_b)
552 with pytest.raises(IntegrityError):
553 await db_session.flush()
554
555 @pytest.mark.asyncio
556 async def test_cascade_delete_on_proposal_delete(self, db_session: AsyncSession) -> None:
557 from sqlalchemy import select
558 from musehub.db.musehub_repo_models import MusehubRepo
559 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
560
561 repo_id = await _make_repo(db_session)
562 p1 = await _make_proposal(db_session, repo_id, proposal_number=1)
563 p2 = await _make_proposal(db_session, repo_id, proposal_number=2)
564
565 dep_id = f"dep-{_uid()}"
566 dep = MusehubProposalDependency(
567 dep_id=dep_id,
568 dependent_proposal_id=p2.proposal_id,
569 dependency_proposal_id=p1.proposal_id,
570 )
571 db_session.add(dep)
572 await db_session.flush()
573
574 await db_session.delete(p1)
575 await db_session.flush()
576
577 result = await db_session.execute(
578 select(MusehubProposalDependency).where(
579 MusehubProposalDependency.dep_id == dep_id
580 )
581 )
582 assert result.scalar_one_or_none() is None
583
584
585 class TestMusehubProposalSimulationORM:
586 @pytest.mark.asyncio
587 async def test_simulation_stored_and_retrieved(self, db_session: AsyncSession) -> None:
588 from musehub.db.musehub_repo_models import MusehubRepo
589 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
590
591 repo_id = await _make_repo(db_session)
592 proposal = await _make_proposal(db_session, repo_id)
593
594 sim = MusehubProposalSimulation(
595 simulation_id=f"sim-{_uid()}",
596 proposal_id=proposal.proposal_id,
597 simulation_type="conflict_scan",
598 from_branch_commit_id="sha256:" + "c" * 64,
599 result={"conflicts": [], "safe_merge": True},
600 duration_ms=142,
601 )
602 db_session.add(sim)
603 await db_session.flush()
604 await db_session.refresh(sim)
605
606 assert sim.simulation_type == "conflict_scan"
607 assert sim.result["safe_merge"] is True
608 assert sim.duration_ms == 142
609 assert sim.expires_at is None
610
611 @pytest.mark.asyncio
612 async def test_duplicate_simulation_type_raises(self, db_session: AsyncSession) -> None:
613 from sqlalchemy.exc import IntegrityError
614 from musehub.db.musehub_repo_models import MusehubRepo
615 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
616
617 repo_id = await _make_repo(db_session)
618 proposal = await _make_proposal(db_session, repo_id, proposal_number=2)
619
620 commit_id = "sha256:" + "d" * 64
621 sim_a = MusehubProposalSimulation(
622 simulation_id=f"sim-{_uid()}",
623 proposal_id=proposal.proposal_id,
624 simulation_type="risk_projection",
625 from_branch_commit_id=commit_id,
626 result={},
627 )
628 sim_b = MusehubProposalSimulation(
629 simulation_id=f"sim-{_uid()}",
630 proposal_id=proposal.proposal_id,
631 simulation_type="risk_projection",
632 from_branch_commit_id=commit_id,
633 result={},
634 )
635 db_session.add(sim_a)
636 db_session.add(sim_b)
637 with pytest.raises(IntegrityError):
638 await db_session.flush()
639
640 @pytest.mark.asyncio
641 async def test_three_simulation_types_coexist(self, db_session: AsyncSession) -> None:
642 from musehub.db.musehub_repo_models import MusehubRepo
643 from musehub.db.musehub_social_models import MusehubProposal, MusehubProposalDependency, MusehubProposalReview, MusehubProposalSimulation
644
645 repo_id = await _make_repo(db_session)
646 proposal = await _make_proposal(db_session, repo_id, proposal_number=3)
647 commit_id = "sha256:" + "e" * 64
648
649 for sim_type in ("conflict_scan", "risk_projection", "dependency_order"):
650 db_session.add(MusehubProposalSimulation(
651 simulation_id=f"sim-{_uid()}",
652 proposal_id=proposal.proposal_id,
653 simulation_type=sim_type,
654 from_branch_commit_id=commit_id,
655 result={"type": sim_type},
656 ))
657 await db_session.flush()
658
659 from sqlalchemy import select, func
660 result = await db_session.execute(
661 select(func.count()).select_from(MusehubProposalSimulation).where(
662 MusehubProposalSimulation.proposal_id == proposal.proposal_id
663 )
664 )
665 assert result.scalar_one() == 3
File History 1 commit
sha256:50b52eda7afb2f122863aef47d684d1a9e4684b48f5f95367fc956e28ceb7d42 refactor: rename merge strategy aliases to canonical names Sonnet 4.6 minor ⚠ 10 days ago