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