gabriel / musehub public
test_genesis_ids.py python
1,425 lines 59.5 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 143 days ago
1 """TDD: genesis-addressed IDs for every first-class semantic entity.
2
3 CONTRACT (issue #10):
4
5 Every first-class semantic entity in the Muse ecosystem has an identity
6 derived from the minimal immutable facts about the moment it was declared
7 to exist — its genesis context. The formula is universal:
8
9 entity_id = "<algo>:" + hash(NUL.join(genesis_fields)).hexdigest()
10
11 The algorithm prefix is intentionally not hardcoded to "sha256". Any
12 canonical <algo>:<hex> form is valid. This future-proofs the system for
13 hash algorithm upgrades without breaking the validator contract.
14
15 Tier 1 — canonical form
16 All compute_* functions return strings matching ^[a-z][a-z0-9]*:[0-9a-f]{32,}$.
17
18 Tier 2 — determinism
19 Same inputs always produce the same ID. Different inputs always produce
20 different IDs (collision resistance tested with minimal single-field diffs).
21
22 Tier 3 — separator injection safety
23 Field values containing NUL bytes, pipe chars, colons, or path separators
24 do not break the hash or allow crafted collisions.
25
26 Tier 3b — Pydantic boundary enforcement
27 Wire-protocol and response models reject non-canonical IDs at the
28 Pydantic validation boundary. Both request models (untrusted input) and
29 response models (service layer bug detection) are covered. Optional ID
30 fields accept None and valid canonical IDs, reject malformed strings.
31
32 Tier 4 — cross-verification (CLI ↔ hub)
33 Functions that exist in both muse.core.genesis and musehub.core.genesis
34 produce identical output for the same inputs.
35
36 Tier 5 — derivation chain
37 Entity IDs that take other entity IDs as genesis fields (e.g. issue_id
38 takes repo_id) form a verifiable chain: changing the parent ID changes
39 all descendant IDs.
40 """
41
42 from __future__ import annotations
43
44 import re
45 import sys
46 from datetime import datetime, timezone
47 from pathlib import Path
48
49 import pytest
50 from muse.core.types import fake_id
51 from pydantic import ValidationError
52
53 from musehub.models.musehub import (
54 CreateRepoRequest,
55 IssueCommentCreate,
56 ProposalCommentCreate,
57 ProposalReviewResponse,
58 ReleaseAssetResponse,
59 UserForkedRepoEntry,
60 WebhookResponse,
61 WireTagInput,
62 )
63 from musehub.api.routes.musehub.collaborators import CollaboratorResponse
64 from musehub.api.routes.musehub.labels import LabelResponse
65 from musehub.models.wire import (
66 WireCommit,
67 WireFetchRequest,
68 WireNegotiateRequest,
69 WireObject,
70 WireSnapshot,
71 )
72
73 # Hub-side genesis functions
74 from musehub.core.genesis import (
75 compute_asset_id,
76 compute_collaborator_id,
77 compute_comment_id,
78 compute_domain_id,
79 compute_fork_id,
80 compute_identity_id,
81 compute_issue_id,
82 compute_key_id,
83 compute_label_id,
84 compute_mist_id,
85 compute_proposal_id,
86 compute_release_id,
87 compute_repo_id,
88 compute_review_id,
89 compute_session_id,
90 compute_tag_id,
91 compute_webhook_id,
92 mist_short_id,
93 )
94
95 # CLI-side genesis functions (cross-verification) — skip gracefully if not yet present
96 sys.path.insert(0, str(Path.home() / "ecosystem" / "muse"))
97 try:
98 from muse.core.genesis import (
99 compute_release_id as cli_compute_release_id,
100 compute_tag_id as cli_compute_tag_id,
101 )
102 _CLI_GENESIS_AVAILABLE = True
103 except ModuleNotFoundError:
104 _CLI_GENESIS_AVAILABLE = False
105 cli_compute_release_id = None # type: ignore[assignment]
106 cli_compute_tag_id = None # type: ignore[assignment]
107
108 # Algo-agnostic canonical pattern: <lowercase-algo>:<lowercase-hex, ≥32 chars>
109 # Do NOT tighten to "sha256" only — this pattern must survive hash algorithm upgrades.
110 _CANONICAL_RE = re.compile(r"^[a-z][a-z0-9]*:[0-9a-f]{32,}$")
111
112 # ---------------------------------------------------------------------------
113 # Shared deterministic inputs
114 # ---------------------------------------------------------------------------
115
116 _PUBKEY = bytes(range(32)) # 32 deterministic bytes
117 _IDENTITY_ID = compute_identity_id(_PUBKEY)
118
119 _REPO_ID = compute_repo_id(_IDENTITY_ID, "my-repo", "code", "2026-01-01T00:00:00Z")
120 _ISSUE_ID = compute_issue_id(_REPO_ID, _IDENTITY_ID, "2026-01-02T00:00:00Z")
121 _PROPOSAL_ID = compute_proposal_id(_REPO_ID, _IDENTITY_ID, "feat/x", "main", "2026-01-03T00:00:00Z")
122 _RELEASE_ID = compute_release_id(_REPO_ID, "v1.0.0", "2026-01-04T00:00:00Z")
123 _COMMIT_ID = fake_id("commit-stub") # canonical stub; real commits use compute_commit_id
124 _TAG_ID = compute_tag_id(_REPO_ID, _COMMIT_ID, "emotion:joyful", "2026-01-05T00:00:00Z")
125 _SESSION_ID = compute_session_id(_REPO_ID, _IDENTITY_ID, "2026-01-06T00:00:00Z")
126 _MIST_ID = compute_mist_id(b"hello muse")
127 _COMMENT_ID = compute_comment_id(_ISSUE_ID, _IDENTITY_ID, "2026-01-07T00:00:00Z")
128 _REVIEW_ID = compute_review_id(_PROPOSAL_ID, _IDENTITY_ID, "2026-01-08T00:00:00Z")
129
130 # Phase 2 genesis IDs
131 _LABEL_ID = compute_label_id(_REPO_ID, "bug", "2026-01-09T00:00:00Z")
132 _ASSET_ID = compute_asset_id(_RELEASE_ID, "v1.0.0-linux-amd64.tar.gz", "2026-01-10T00:00:00Z")
133 _WEBHOOK_ID = compute_webhook_id(_REPO_ID, "https://ci.example.com/hook", "2026-01-11T00:00:00Z")
134 _FORK_REPO_ID = compute_repo_id(_IDENTITY_ID, "my-fork", "code", "2026-01-12T00:00:00Z")
135 _FORK_ID = compute_fork_id(_REPO_ID, _FORK_REPO_ID, "2026-01-13T00:00:00Z")
136 _COLLAB_IDENTITY_ID = compute_identity_id(bytes(range(1, 33)))
137 _COLLABORATOR_ID = compute_collaborator_id(_REPO_ID, _COLLAB_IDENTITY_ID, "2026-01-14T00:00:00Z")
138 _KEY_ID = compute_key_id(_IDENTITY_ID, "ed25519:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")
139 _DOMAIN_ID = compute_domain_id("gabriel", "code", "2026-01-15T00:00:00Z")
140
141
142 # ===========================================================================
143 # Tier 1 — canonical form
144 # ===========================================================================
145
146 class TestCanonicalForm:
147
148 def test_identity_id_canonical(self) -> None:
149 assert _CANONICAL_RE.match(_IDENTITY_ID)
150
151 def test_repo_id_canonical(self) -> None:
152 assert _CANONICAL_RE.match(_REPO_ID)
153
154 def test_issue_id_canonical(self) -> None:
155 assert _CANONICAL_RE.match(_ISSUE_ID)
156
157 def test_proposal_id_canonical(self) -> None:
158 assert _CANONICAL_RE.match(_PROPOSAL_ID)
159
160 def test_release_id_canonical(self) -> None:
161 assert _CANONICAL_RE.match(_RELEASE_ID)
162
163 def test_tag_id_canonical(self) -> None:
164 assert _CANONICAL_RE.match(_TAG_ID)
165
166 def test_session_id_canonical(self) -> None:
167 assert _CANONICAL_RE.match(_SESSION_ID)
168
169 def test_mist_id_canonical(self) -> None:
170 assert _CANONICAL_RE.match(_MIST_ID)
171
172 def test_comment_id_canonical(self) -> None:
173 assert _CANONICAL_RE.match(_COMMENT_ID)
174
175 def test_review_id_canonical(self) -> None:
176 assert _CANONICAL_RE.match(_REVIEW_ID)
177
178 def test_mist_short_id_is_12_chars(self) -> None:
179 short = mist_short_id(_MIST_ID)
180 assert len(short) == 12
181 assert re.match(r"^[0-9a-f]{12}$", short)
182
183 def test_mist_short_id_is_prefix_of_digest(self) -> None:
184 digest = _MIST_ID.removeprefix("sha256:")
185 assert mist_short_id(_MIST_ID) == digest[:12]
186
187
188 # ===========================================================================
189 # Tier 2 — determinism and collision resistance
190 # ===========================================================================
191
192 class TestDeterminism:
193
194 def test_identity_id_deterministic(self) -> None:
195 assert compute_identity_id(_PUBKEY) == compute_identity_id(_PUBKEY)
196
197 def test_repo_id_deterministic(self) -> None:
198 args = (_IDENTITY_ID, "repo", "code", "2026-01-01T00:00:00Z")
199 assert compute_repo_id(*args) == compute_repo_id(*args)
200
201 def test_issue_id_deterministic(self) -> None:
202 args = (_REPO_ID, _IDENTITY_ID, "2026-01-02T00:00:00Z")
203 assert compute_issue_id(*args) == compute_issue_id(*args)
204
205 def test_different_pubkeys_yield_different_identity_ids(self) -> None:
206 pk_a = bytes(range(32))
207 pk_b = bytes(range(1, 33))
208 assert compute_identity_id(pk_a) != compute_identity_id(pk_b)
209
210 def test_different_slugs_yield_different_repo_ids(self) -> None:
211 a = compute_repo_id(_IDENTITY_ID, "repo-a", "code", "2026-01-01T00:00:00Z")
212 b = compute_repo_id(_IDENTITY_ID, "repo-b", "code", "2026-01-01T00:00:00Z")
213 assert a != b
214
215 def test_different_domains_yield_different_repo_ids(self) -> None:
216 a = compute_repo_id(_IDENTITY_ID, "repo", "code", "2026-01-01T00:00:00Z")
217 b = compute_repo_id(_IDENTITY_ID, "repo", "music", "2026-01-01T00:00:00Z")
218 assert a != b
219
220 def test_different_timestamps_yield_different_issue_ids(self) -> None:
221 a = compute_issue_id(_REPO_ID, _IDENTITY_ID, "2026-01-01T00:00:00Z")
222 b = compute_issue_id(_REPO_ID, _IDENTITY_ID, "2026-01-02T00:00:00Z")
223 assert a != b
224
225 def test_different_branches_yield_different_proposal_ids(self) -> None:
226 a = compute_proposal_id(_REPO_ID, _IDENTITY_ID, "feat/a", "main", "2026-01-01T00:00:00Z")
227 b = compute_proposal_id(_REPO_ID, _IDENTITY_ID, "feat/b", "main", "2026-01-01T00:00:00Z")
228 assert a != b
229
230 def test_different_tags_yield_different_release_ids(self) -> None:
231 a = compute_release_id(_REPO_ID, "v1.0.0", "2026-01-01T00:00:00Z")
232 b = compute_release_id(_REPO_ID, "v2.0.0", "2026-01-01T00:00:00Z")
233 assert a != b
234
235 def test_different_labels_yield_different_tag_ids(self) -> None:
236 a = compute_tag_id(_REPO_ID, _COMMIT_ID, "emotion:joyful", "2026-01-01T00:00:00Z")
237 b = compute_tag_id(_REPO_ID, _COMMIT_ID, "emotion:melancholic", "2026-01-01T00:00:00Z")
238 assert a != b
239
240 def test_different_content_yields_different_mist_ids(self) -> None:
241 assert compute_mist_id(b"hello") != compute_mist_id(b"world")
242
243 def test_all_entity_ids_are_distinct(self) -> None:
244 """All entity IDs computed from their respective genesis contexts are unique."""
245 ids = [
246 _IDENTITY_ID, _REPO_ID, _ISSUE_ID, _PROPOSAL_ID,
247 _RELEASE_ID, _TAG_ID, _SESSION_ID, _MIST_ID,
248 _COMMENT_ID, _REVIEW_ID,
249 ]
250 assert len(ids) == len(set(ids)), "two entity IDs collided"
251
252
253 # ===========================================================================
254 # Tier 3 — separator injection safety
255 # ===========================================================================
256
257 class TestSeparatorInjection:
258
259 def test_nul_byte_in_slug_does_not_collide(self) -> None:
260 """A slug containing NUL + domain cannot be crafted to match a different (slug, domain) pair."""
261 # If the separator were not NUL, "a|b" + "|" + "c" could equal "a" + "|" + "b|c".
262 # With NUL separator, NUL inside a field value is structurally impossible in normal usage,
263 # but we verify the function still returns a valid ID for unusual inputs.
264 exotic = compute_repo_id(_IDENTITY_ID, "repo\x00extra", "code", "2026-01-01T00:00:00Z")
265 normal = compute_repo_id(_IDENTITY_ID, "repo", "code\x00extra", "2026-01-01T00:00:00Z")
266 assert _CANONICAL_RE.match(exotic)
267 assert exotic != normal, "NUL in field value must not produce collisions across fields"
268
269 def test_pipe_in_label_is_safe(self) -> None:
270 a = compute_tag_id(_REPO_ID, _COMMIT_ID, "section|verse", "2026-01-01T00:00:00Z")
271 assert _CANONICAL_RE.match(a)
272
273 def test_colon_in_label_is_safe(self) -> None:
274 a = compute_tag_id(_REPO_ID, _COMMIT_ID, "emotion:joyful:extra", "2026-01-01T00:00:00Z")
275 assert _CANONICAL_RE.match(a)
276
277 def test_path_separator_in_branch_is_safe(self) -> None:
278 a = compute_proposal_id(_REPO_ID, _IDENTITY_ID, "feat/nested/branch", "main", "2026-01-01T00:00:00Z")
279 assert _CANONICAL_RE.match(a)
280
281 def test_sha256_prefix_in_field_is_safe(self) -> None:
282 """Entity IDs used as genesis fields (which start with sha256:) are handled correctly."""
283 # repo_id and identity_id both start with "sha256:" — verify no double-prefix or truncation.
284 repo = compute_repo_id(_IDENTITY_ID, "test", "code", "2026-01-01T00:00:00Z")
285 issue = compute_issue_id(repo, _IDENTITY_ID, "2026-01-01T00:00:00Z")
286 assert _CANONICAL_RE.match(issue)
287
288
289 # ===========================================================================
290 # Tier 3b — Pydantic boundary enforcement
291 # ===========================================================================
292
293 _VALID_ID = fake_id("valid-id-stub")
294 _FUTURE_ID = "blake3:" + "b" * 64 # algo-agnostic: must also be accepted
295 _BAD_IDS = [
296 "not-a-sha",
297 "a1b2c3d4-e5f6-7890-abcd-ef1234567890", # UUID
298 f"SHA256:{'a' * 64}", # uppercase algo
299 f"sha256:{'A' * 64}", # uppercase hex
300 "",
301 "sha256:tooshort",
302 ]
303 _DT = datetime(2026, 1, 1, tzinfo=timezone.utc)
304
305
306 class TestPydanticBoundaryWireModels:
307 """Wire models reject non-canonical IDs; accept valid canonical forms."""
308
309 def test_wire_commit_rejects_bad_commit_id(self) -> None:
310 for bad in _BAD_IDS:
311 with pytest.raises(ValidationError):
312 WireCommit(commit_id=bad)
313
314 def test_wire_commit_accepts_future_algo(self) -> None:
315 c = WireCommit(commit_id=_FUTURE_ID)
316 assert c.commit_id == _FUTURE_ID
317
318 def test_wire_snapshot_rejects_bad_id(self) -> None:
319 with pytest.raises(ValidationError):
320 WireSnapshot(snapshot_id="bad-id")
321
322 def test_wire_snapshot_accepts_future_algo(self) -> None:
323 s = WireSnapshot(snapshot_id=_FUTURE_ID)
324 assert s.snapshot_id == _FUTURE_ID
325
326 def test_wire_object_rejects_bad_object_id(self) -> None:
327 with pytest.raises(ValidationError):
328 WireObject(object_id="bad", content=b"x")
329
330 def test_wire_object_accepts_future_algo(self) -> None:
331 o = WireObject(object_id=_FUTURE_ID, content=b"x")
332 assert o.object_id == _FUTURE_ID
333
334 def test_fetch_request_rejects_bad_want(self) -> None:
335 with pytest.raises(ValidationError):
336 WireFetchRequest(want=["not-valid"], have=[])
337
338 def test_fetch_request_rejects_bad_have(self) -> None:
339 with pytest.raises(ValidationError):
340 WireFetchRequest(want=[], have=["uuid-style"])
341
342 def test_fetch_request_accepts_valid_and_future_ids(self) -> None:
343 r = WireFetchRequest(want=[_VALID_ID, _FUTURE_ID], have=[_VALID_ID])
344 assert len(r.want) == 2
345
346 def test_negotiate_request_rejects_bad_have(self) -> None:
347 with pytest.raises(ValidationError):
348 WireNegotiateRequest(have=["bad"], want=[])
349
350 def test_negotiate_request_rejects_bad_want(self) -> None:
351 with pytest.raises(ValidationError):
352 WireNegotiateRequest(have=[], want=["bad"])
353
354 def test_negotiate_request_accepts_future_algo(self) -> None:
355 r = WireNegotiateRequest(have=[_FUTURE_ID], want=[_FUTURE_ID])
356 assert r.have == [_FUTURE_ID]
357
358
359 class TestPydanticBoundaryResponseModels:
360 """Response models reject non-canonical IDs (catches service layer bugs)."""
361
362 def test_proposal_review_rejects_bad_id(self) -> None:
363 with pytest.raises(ValidationError):
364 ProposalReviewResponse(
365 id="bad-id", proposal_id=_VALID_ID,
366 reviewer_username="gabriel", state="approved", created_at=_DT,
367 )
368
369 def test_proposal_review_rejects_bad_proposal_id(self) -> None:
370 with pytest.raises(ValidationError):
371 ProposalReviewResponse(
372 id=_VALID_ID, proposal_id="not-canonical",
373 reviewer_username="gabriel", state="approved", created_at=_DT,
374 )
375
376 def test_proposal_review_accepts_future_algo(self) -> None:
377 r = ProposalReviewResponse(
378 id=_FUTURE_ID, proposal_id=_FUTURE_ID,
379 reviewer_username="gabriel", state="approved", created_at=_DT,
380 )
381 assert r.id == _FUTURE_ID
382
383 def test_release_asset_rejects_bad_asset_id(self) -> None:
384 with pytest.raises(ValidationError):
385 ReleaseAssetResponse(
386 asset_id="bad", release_id=_VALID_ID,
387 name="f.tar.gz", download_url="https://x.com/f", created_at=_DT,
388 )
389
390 def test_release_asset_rejects_bad_release_id(self) -> None:
391 with pytest.raises(ValidationError):
392 ReleaseAssetResponse(
393 asset_id=_VALID_ID, release_id="uuid",
394 name="f.tar.gz", download_url="https://x.com/f", created_at=_DT,
395 )
396
397 def test_release_asset_accepts_future_algo(self) -> None:
398 a = ReleaseAssetResponse(
399 asset_id=_FUTURE_ID, release_id=_FUTURE_ID,
400 name="f.tar.gz", download_url="https://x.com/f", created_at=_DT,
401 )
402 assert a.release_id == _FUTURE_ID
403
404 def test_wire_tag_rejects_bad_tag_id(self) -> None:
405 with pytest.raises(ValidationError):
406 WireTagInput(tag_id="bad", commit_id=_VALID_ID, tag="section:verse")
407
408 def test_wire_tag_rejects_bad_commit_id(self) -> None:
409 with pytest.raises(ValidationError):
410 WireTagInput(tag_id=_VALID_ID, commit_id="bad", tag="section:verse")
411
412 def test_wire_tag_accepts_future_algo(self) -> None:
413 t = WireTagInput(tag_id=_FUTURE_ID, commit_id=_FUTURE_ID, tag="section:verse")
414 assert t.tag_id == _FUTURE_ID
415
416
417 class TestPydanticBoundaryOptionalFields:
418 """Optional genesis ID fields accept None, valid IDs, reject bad strings."""
419
420 def test_create_repo_template_none(self) -> None:
421 r = CreateRepoRequest(name="muse", owner="gabriel")
422 assert r.template_repo_id is None
423
424 def test_create_repo_template_valid(self) -> None:
425 r = CreateRepoRequest(name="muse", owner="gabriel", template_repo_id=_VALID_ID)
426 assert r.template_repo_id == _VALID_ID
427
428 def test_create_repo_template_future_algo(self) -> None:
429 r = CreateRepoRequest(name="muse", owner="gabriel", template_repo_id=_FUTURE_ID)
430 assert r.template_repo_id == _FUTURE_ID
431
432 def test_create_repo_template_bad(self) -> None:
433 with pytest.raises(ValidationError):
434 CreateRepoRequest(name="muse", owner="gabriel", template_repo_id="bad-uuid")
435
436 def test_issue_comment_parent_none(self) -> None:
437 assert IssueCommentCreate(body="hi").parent_id is None
438
439 def test_issue_comment_parent_valid(self) -> None:
440 c = IssueCommentCreate(body="reply", parent_id=_VALID_ID)
441 assert c.parent_id == _VALID_ID
442
443 def test_issue_comment_parent_future_algo(self) -> None:
444 c = IssueCommentCreate(body="reply", parent_id=_FUTURE_ID)
445 assert c.parent_id == _FUTURE_ID
446
447 def test_issue_comment_parent_bad(self) -> None:
448 with pytest.raises(ValidationError):
449 IssueCommentCreate(body="reply", parent_id="a1b2c3d4-uuid")
450
451 def test_proposal_comment_parent_none(self) -> None:
452 assert ProposalCommentCreate(body="hi").parent_comment_id is None
453
454 def test_proposal_comment_parent_valid(self) -> None:
455 c = ProposalCommentCreate(body="reply", parent_comment_id=_VALID_ID)
456 assert c.parent_comment_id == _VALID_ID
457
458 def test_proposal_comment_parent_future_algo(self) -> None:
459 c = ProposalCommentCreate(body="reply", parent_comment_id=_FUTURE_ID)
460 assert c.parent_comment_id == _FUTURE_ID
461
462 def test_proposal_comment_parent_bad_uuid(self) -> None:
463 with pytest.raises(ValidationError):
464 ProposalCommentCreate(
465 body="reply",
466 parent_comment_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
467 )
468
469
470 # ===========================================================================
471 # Tier 4 — cross-verification (CLI ↔ hub)
472 # ===========================================================================
473
474 @pytest.mark.skipif(not _CLI_GENESIS_AVAILABLE, reason="muse.core.genesis not yet implemented in CLI")
475 class TestCrossVerification:
476
477 def test_tag_id_cli_and_hub_agree(self) -> None:
478 args = (_REPO_ID, _COMMIT_ID, "emotion:joyful", "2026-01-01T00:00:00Z")
479 assert compute_tag_id(*args) == cli_compute_tag_id(*args)
480
481 def test_release_id_cli_and_hub_agree(self) -> None:
482 args = (_REPO_ID, "v1.0.0", "2026-01-01T00:00:00Z")
483 assert compute_release_id(*args) == cli_compute_release_id(*args)
484
485 def test_tag_id_cli_returns_canonical(self) -> None:
486 result = cli_compute_tag_id(_REPO_ID, _COMMIT_ID, "v1.0-wip", "2026-01-01T00:00:00Z")
487 assert _CANONICAL_RE.match(result)
488
489 def test_release_id_cli_returns_canonical(self) -> None:
490 result = cli_compute_release_id(_REPO_ID, "v2.0.0", "2026-01-01T00:00:00Z")
491 assert _CANONICAL_RE.match(result)
492
493
494 # ===========================================================================
495 # Tier 5 — derivation chain
496 # ===========================================================================
497
498 class TestDerivationChain:
499
500 def test_changing_owner_changes_repo_id(self) -> None:
501 id_a = compute_identity_id(bytes(range(32)))
502 id_b = compute_identity_id(bytes(range(1, 33)))
503 repo_a = compute_repo_id(id_a, "repo", "code", "2026-01-01T00:00:00Z")
504 repo_b = compute_repo_id(id_b, "repo", "code", "2026-01-01T00:00:00Z")
505 assert repo_a != repo_b
506
507 def test_changing_repo_changes_issue_id(self) -> None:
508 repo_a = compute_repo_id(_IDENTITY_ID, "repo-a", "code", "2026-01-01T00:00:00Z")
509 repo_b = compute_repo_id(_IDENTITY_ID, "repo-b", "code", "2026-01-01T00:00:00Z")
510 issue_a = compute_issue_id(repo_a, _IDENTITY_ID, "2026-01-02T00:00:00Z")
511 issue_b = compute_issue_id(repo_b, _IDENTITY_ID, "2026-01-02T00:00:00Z")
512 assert issue_a != issue_b
513
514 def test_changing_issue_changes_comment_id(self) -> None:
515 issue_a = compute_issue_id(_REPO_ID, _IDENTITY_ID, "2026-01-01T00:00:00Z")
516 issue_b = compute_issue_id(_REPO_ID, _IDENTITY_ID, "2026-01-02T00:00:00Z")
517 comment_a = compute_comment_id(issue_a, _IDENTITY_ID, "2026-01-03T00:00:00Z")
518 comment_b = compute_comment_id(issue_b, _IDENTITY_ID, "2026-01-03T00:00:00Z")
519 assert comment_a != comment_b
520
521 def test_changing_proposal_changes_review_id(self) -> None:
522 prop_a = compute_proposal_id(_REPO_ID, _IDENTITY_ID, "feat/a", "main", "2026-01-01T00:00:00Z")
523 prop_b = compute_proposal_id(_REPO_ID, _IDENTITY_ID, "feat/b", "main", "2026-01-01T00:00:00Z")
524 review_a = compute_review_id(prop_a, _IDENTITY_ID, "2026-01-02T00:00:00Z")
525 review_b = compute_review_id(prop_b, _IDENTITY_ID, "2026-01-02T00:00:00Z")
526 assert review_a != review_b
527
528 def test_identity_repo_issue_comment_chain_is_fully_verifiable(self) -> None:
529 """Full four-level chain: identity → repo → issue → comment."""
530 pk = bytes(range(32))
531 identity_id = compute_identity_id(pk)
532 repo_id = compute_repo_id(identity_id, "chain-test", "code", "2026-01-01T00:00:00Z")
533 issue_id = compute_issue_id(repo_id, identity_id, "2026-01-02T00:00:00Z")
534 comment_id = compute_comment_id(issue_id, identity_id, "2026-01-03T00:00:00Z")
535
536 # Every level is canonical
537 for entity_id in (identity_id, repo_id, issue_id, comment_id):
538 assert _CANONICAL_RE.match(entity_id), f"non-canonical: {entity_id}"
539
540 # Mutating the pubkey propagates through the entire chain
541 pk2 = bytes(range(1, 33))
542 identity_id2 = compute_identity_id(pk2)
543 repo_id2 = compute_repo_id(identity_id2, "chain-test", "code", "2026-01-01T00:00:00Z")
544 issue_id2 = compute_issue_id(repo_id2, identity_id2, "2026-01-02T00:00:00Z")
545 comment_id2 = compute_comment_id(issue_id2, identity_id2, "2026-01-03T00:00:00Z")
546
547 assert identity_id != identity_id2
548 assert repo_id != repo_id2
549 assert issue_id != issue_id2
550 assert comment_id != comment_id2
551
552
553 # ===========================================================================
554 # Tier 6 — service-layer contract (unit, mocked DB)
555 #
556 # Every service function that creates a first-class entity must compute its ID
557 # from genesis context — not from uuid4(). Tests here are RED until Phase 4
558 # is implemented and will stay GREEN thereafter.
559 # ===========================================================================
560
561
562 class TestServiceLayerGenesisIds:
563 """Service creation functions assign genesis-addressed IDs, never uuid4."""
564
565 # ------------------------------------------------------------------
566 # Helpers shared across tests
567 # ------------------------------------------------------------------
568
569 def _async_session(self, *, execute_returns: MagicMock | None = None) -> "AsyncMock":
570 """Return a minimal AsyncMock DB session."""
571 from unittest.mock import AsyncMock, MagicMock
572
573 session = AsyncMock()
574 scalar = MagicMock()
575 scalar.scalar_one_or_none.return_value = execute_returns
576 scalar.scalar_one.return_value = None
577 session.execute.return_value = scalar
578 session.flush = AsyncMock()
579 session.commit = AsyncMock()
580 session.delete = AsyncMock()
581
582 async def _refresh(obj: MagicMock) -> None:
583 from datetime import datetime, timezone
584 for attr in ("created_at", "updated_at"):
585 if not getattr(obj, attr, None):
586 try:
587 setattr(obj, attr, datetime.now(timezone.utc))
588 except Exception:
589 pass
590 for attr in ("last_used_at",):
591 if not hasattr(obj, attr):
592 try:
593 setattr(obj, attr, None)
594 except Exception:
595 pass
596
597 session.refresh = _refresh
598 return session
599
600 def _keypair(self) -> None:
601 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
602 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
603 priv = Ed25519PrivateKey.generate()
604 pub = priv.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
605 return priv, pub
606
607 # ------------------------------------------------------------------
608 # 6.1 Identity — register_agent_identity
609 # ------------------------------------------------------------------
610
611 @pytest.mark.asyncio
612 async def test_register_agent_identity_uses_compute_identity_id(self) -> None:
613 """register_agent_identity assigns identity_id = compute_identity_id(pub_bytes)."""
614 from muse.core.types import encode_pubkey, public_key_fingerprint
615 from musehub.services.musehub_auth import register_agent_identity
616
617 _, pub = self._keypair()
618 pub_b64 = encode_pubkey("ed25519", pub)
619 fp = public_key_fingerprint(pub)
620 expected = compute_identity_id(pub)
621
622 session = self._async_session()
623 await register_agent_identity(
624 session=session,
625 handle="test-agent",
626 public_key_b64=pub_b64,
627 fingerprint=fp,
628 algorithm="ed25519",
629 spawned_by="gabriel",
630 )
631
632 # First add() is the MusehubIdentity row.
633 identity = session.add.call_args_list[0][0][0]
634 assert identity.identity_id == expected, (
635 f"expected genesis ID {expected!r}, got {identity.identity_id!r}"
636 )
637 assert _CANONICAL_RE.match(identity.identity_id)
638
639 # ------------------------------------------------------------------
640 # 6.2 Session — upsert_session
641 # ------------------------------------------------------------------
642
643 @pytest.mark.asyncio
644 async def test_upsert_session_uses_compute_session_id(self) -> None:
645 """upsert_session assigns session_id = compute_session_id(repo_id, author_identity_id, started_at)."""
646 from unittest.mock import MagicMock, patch
647 from datetime import datetime, timezone
648 from musehub.models.musehub import SessionCreate
649 from musehub.services.musehub_sessions import upsert_session
650
651 repo_id = _REPO_ID
652 author_identity_id = _IDENTITY_ID
653 started_at = datetime(2026, 1, 6, 0, 0, 0, tzinfo=timezone.utc)
654 expected = compute_session_id(repo_id, author_identity_id, started_at.isoformat())
655
656 data = SessionCreate(started_at=started_at, participants=[], intent="", location="")
657 session = self._async_session()
658
659 with patch("musehub.services.musehub_sessions._to_response", return_value=MagicMock()):
660 await upsert_session(
661 session,
662 repo_id=repo_id,
663 author_identity_id=author_identity_id,
664 data=data,
665 )
666
667 added = session.add.call_args_list[0][0][0]
668 assert added.session_id == expected
669 assert _CANONICAL_RE.match(added.session_id)
670
671 # ------------------------------------------------------------------
672 # 6.3 Issue — create_issue
673 # ------------------------------------------------------------------
674
675 @pytest.mark.asyncio
676 async def test_create_issue_uses_compute_issue_id(self) -> None:
677 """create_issue assigns issue_id = compute_issue_id(repo_id, author_identity_id, created_at)."""
678 from unittest.mock import AsyncMock, MagicMock, patch
679 from datetime import datetime, timezone
680 from musehub.services import musehub_issues
681
682 repo_id = _REPO_ID
683 author_identity_id = _IDENTITY_ID
684 fixed_now = datetime(2026, 1, 2, 0, 0, 0, tzinfo=timezone.utc)
685 expected = compute_issue_id(repo_id, author_identity_id, fixed_now.isoformat())
686
687 session = self._async_session()
688 # _next_issue_number calls session.execute(...).scalar_one_or_none()
689 # Return None so next number = 1.
690 session.execute.return_value.scalar_one_or_none.return_value = None
691
692 with patch("musehub.services.musehub_issues.datetime") as mock_dt:
693 mock_dt.now.return_value = fixed_now
694 mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
695
696 await musehub_issues.create_issue(
697 session,
698 repo_id=repo_id,
699 title="Test issue",
700 body="body",
701 labels=[],
702 author="gabriel",
703 author_identity_id=author_identity_id,
704 )
705
706 added = session.add.call_args_list[0][0][0]
707 assert added.issue_id == expected
708 assert _CANONICAL_RE.match(added.issue_id)
709
710 # ------------------------------------------------------------------
711 # 6.4 Proposal — create_proposal
712 # ------------------------------------------------------------------
713
714 @pytest.mark.asyncio
715 async def test_create_proposal_uses_compute_proposal_id(self) -> None:
716 """create_proposal assigns proposal_id = compute_proposal_id(...)."""
717 from unittest.mock import AsyncMock, MagicMock, patch
718 from datetime import datetime, timezone
719 from musehub.services import musehub_proposals
720
721 repo_id = _REPO_ID
722 author_identity_id = _IDENTITY_ID
723 from_branch = "feat/x"
724 to_branch = "main"
725 fixed_now = datetime(2026, 1, 3, 0, 0, 0, tzinfo=timezone.utc)
726 expected = compute_proposal_id(repo_id, author_identity_id, from_branch, to_branch, fixed_now.isoformat())
727
728 # _get_branch makes a DB call — return a fake branch row.
729 from unittest.mock import MagicMock
730 fake_branch = MagicMock()
731 fake_branch.head_commit_id = fake_id("branch-head-stub")
732
733 session = self._async_session()
734 # First execute: _get_branch → returns branch
735 # Second execute: max proposal_number → returns None
736 # Third execute: _touched_symbols → returns []
737 from unittest.mock import AsyncMock
738 results = [
739 MagicMock(**{"scalar_one_or_none.return_value": fake_branch}),
740 MagicMock(**{"scalar_one_or_none.return_value": None}),
741 MagicMock(**{"scalars.return_value.all.return_value": []}),
742 ]
743 session.execute.side_effect = results
744
745 with patch("musehub.services.musehub_proposals._utc_now", return_value=fixed_now):
746 await musehub_proposals.create_proposal(
747 session,
748 repo_id=repo_id,
749 title="Test proposal",
750 from_branch=from_branch,
751 to_branch=to_branch,
752 body="",
753 author="gabriel",
754 author_identity_id=author_identity_id,
755 )
756
757 added = session.add.call_args_list[0][0][0]
758 assert added.proposal_id == expected
759 assert _CANONICAL_RE.match(added.proposal_id)
760
761 # ------------------------------------------------------------------
762 # 6.5 Repo — create_repo
763 # ------------------------------------------------------------------
764
765 @pytest.mark.asyncio
766 async def test_create_repo_uses_compute_repo_id(self) -> None:
767 """create_repo assigns repo_id = compute_repo_id(owner_user_id, slug, domain, created_at)."""
768 from unittest.mock import patch
769 from datetime import datetime, timezone
770 from musehub.services import musehub_repository
771
772 owner_identity_id = _IDENTITY_ID
773 slug = "my-repo"
774 domain = "code"
775 fixed_now = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
776 expected = compute_repo_id(owner_identity_id, slug, domain, fixed_now.isoformat())
777
778 from unittest.mock import AsyncMock
779 session = self._async_session()
780 # template lookup returns None; no other DB reads needed.
781 session.get = AsyncMock(return_value=None)
782
783 from unittest.mock import MagicMock
784
785 with patch("musehub.services.musehub_repository.datetime") as mock_dt, \
786 patch("musehub.services.musehub_repository._to_repo_response", return_value=MagicMock()):
787 mock_dt.now.return_value = fixed_now
788 mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
789
790 await musehub_repository.create_repo(
791 session,
792 name="My Repo",
793 owner="gabriel",
794 visibility="public",
795 owner_user_id=compute_identity_id(b"gabriel"),
796 owner_identity_id=owner_identity_id,
797 domain=domain,
798 )
799
800 added = session.add.call_args_list[0][0][0]
801 assert added.repo_id == expected
802 assert _CANONICAL_RE.match(added.repo_id)
803
804
805 # ===========================================================================
806 # Phase 2 — Tier 1: canonical form for new genesis functions
807 # ===========================================================================
808
809
810 class TestPhase2CanonicalForm:
811
812 def test_label_id_canonical(self) -> None:
813 assert _CANONICAL_RE.match(_LABEL_ID)
814
815 def test_asset_id_canonical(self) -> None:
816 assert _CANONICAL_RE.match(_ASSET_ID)
817
818 def test_webhook_id_canonical(self) -> None:
819 assert _CANONICAL_RE.match(_WEBHOOK_ID)
820
821 def test_fork_id_canonical(self) -> None:
822 assert _CANONICAL_RE.match(_FORK_ID)
823
824 def test_collaborator_id_canonical(self) -> None:
825 assert _CANONICAL_RE.match(_COLLABORATOR_ID)
826
827 def test_key_id_canonical(self) -> None:
828 assert _CANONICAL_RE.match(_KEY_ID)
829
830 def test_domain_id_canonical(self) -> None:
831 assert _CANONICAL_RE.match(_DOMAIN_ID)
832
833
834 # ===========================================================================
835 # Phase 2 — Tier 2: determinism and collision resistance
836 # ===========================================================================
837
838
839 class TestPhase2Determinism:
840
841 def test_label_id_deterministic(self) -> None:
842 args = (_REPO_ID, "bug", "2026-01-09T00:00:00Z")
843 assert compute_label_id(*args) == compute_label_id(*args)
844
845 def test_label_different_names_differ(self) -> None:
846 a = compute_label_id(_REPO_ID, "bug", "2026-01-09T00:00:00Z")
847 b = compute_label_id(_REPO_ID, "enhancement", "2026-01-09T00:00:00Z")
848 assert a != b
849
850 def test_label_different_repos_differ(self) -> None:
851 repo2 = compute_repo_id(_IDENTITY_ID, "other-repo", "code", "2026-01-01T00:00:00Z")
852 a = compute_label_id(_REPO_ID, "bug", "2026-01-09T00:00:00Z")
853 b = compute_label_id(repo2, "bug", "2026-01-09T00:00:00Z")
854 assert a != b
855
856 def test_asset_id_deterministic(self) -> None:
857 args = (_RELEASE_ID, "v1.0.0-linux-amd64.tar.gz", "2026-01-10T00:00:00Z")
858 assert compute_asset_id(*args) == compute_asset_id(*args)
859
860 def test_asset_different_filenames_differ(self) -> None:
861 a = compute_asset_id(_RELEASE_ID, "linux.tar.gz", "2026-01-10T00:00:00Z")
862 b = compute_asset_id(_RELEASE_ID, "darwin.tar.gz", "2026-01-10T00:00:00Z")
863 assert a != b
864
865 def test_webhook_id_deterministic(self) -> None:
866 args = (_REPO_ID, "https://ci.example.com/hook", "2026-01-11T00:00:00Z")
867 assert compute_webhook_id(*args) == compute_webhook_id(*args)
868
869 def test_webhook_different_urls_differ(self) -> None:
870 a = compute_webhook_id(_REPO_ID, "https://ci.example.com/a", "2026-01-11T00:00:00Z")
871 b = compute_webhook_id(_REPO_ID, "https://ci.example.com/b", "2026-01-11T00:00:00Z")
872 assert a != b
873
874 def test_fork_id_deterministic(self) -> None:
875 args = (_REPO_ID, _FORK_REPO_ID, "2026-01-13T00:00:00Z")
876 assert compute_fork_id(*args) == compute_fork_id(*args)
877
878 def test_fork_different_source_repos_differ(self) -> None:
879 repo2 = compute_repo_id(_IDENTITY_ID, "other-repo", "code", "2026-01-01T00:00:00Z")
880 a = compute_fork_id(_REPO_ID, _FORK_REPO_ID, "2026-01-13T00:00:00Z")
881 b = compute_fork_id(repo2, _FORK_REPO_ID, "2026-01-13T00:00:00Z")
882 assert a != b
883
884 def test_collaborator_id_deterministic(self) -> None:
885 args = (_REPO_ID, _COLLAB_IDENTITY_ID, "2026-01-14T00:00:00Z")
886 assert compute_collaborator_id(*args) == compute_collaborator_id(*args)
887
888 def test_collaborator_different_identities_differ(self) -> None:
889 id2 = compute_identity_id(bytes(range(2, 34)))
890 a = compute_collaborator_id(_REPO_ID, _COLLAB_IDENTITY_ID, "2026-01-14T00:00:00Z")
891 b = compute_collaborator_id(_REPO_ID, id2, "2026-01-14T00:00:00Z")
892 assert a != b
893
894 def test_key_id_deterministic(self) -> None:
895 args = (_IDENTITY_ID, "ed25519:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")
896 assert compute_key_id(*args) == compute_key_id(*args)
897
898 def test_key_different_pubkeys_differ(self) -> None:
899 a = compute_key_id(_IDENTITY_ID, "ed25519:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")
900 b = compute_key_id(_IDENTITY_ID, "ed25519:BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB")
901 assert a != b
902
903 def test_key_no_timestamp_by_design(self) -> None:
904 """compute_key_id takes no timestamp — a pubkey can only be registered once per identity."""
905 # Two calls with identical args must produce identical IDs (idempotent registration).
906 id1 = compute_key_id(_IDENTITY_ID, "ed25519:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")
907 id2 = compute_key_id(_IDENTITY_ID, "ed25519:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")
908 assert id1 == id2
909
910 def test_domain_id_deterministic(self) -> None:
911 args = ("gabriel", "code", "2026-01-15T00:00:00Z")
912 assert compute_domain_id(*args) == compute_domain_id(*args)
913
914 def test_domain_different_slugs_differ(self) -> None:
915 a = compute_domain_id("gabriel", "code", "2026-01-15T00:00:00Z")
916 b = compute_domain_id("gabriel", "music", "2026-01-15T00:00:00Z")
917 assert a != b
918
919 def test_all_phase2_ids_are_distinct(self) -> None:
920 ids = [_LABEL_ID, _ASSET_ID, _WEBHOOK_ID, _FORK_ID, _COLLABORATOR_ID, _KEY_ID, _DOMAIN_ID]
921 assert len(ids) == len(set(ids)), "two Phase 2 entity IDs collided"
922
923 def test_phase2_ids_distinct_from_phase1_ids(self) -> None:
924 phase1 = {_IDENTITY_ID, _REPO_ID, _ISSUE_ID, _PROPOSAL_ID, _RELEASE_ID, _TAG_ID, _SESSION_ID, _MIST_ID, _COMMENT_ID, _REVIEW_ID}
925 phase2 = {_LABEL_ID, _ASSET_ID, _WEBHOOK_ID, _FORK_ID, _COLLABORATOR_ID, _KEY_ID, _DOMAIN_ID}
926 assert phase1.isdisjoint(phase2), "a Phase 2 ID collided with a Phase 1 ID"
927
928
929 # ===========================================================================
930 # Phase 2 — Tier 3: separator injection safety
931 # ===========================================================================
932
933
934 class TestPhase2SeparatorInjection:
935
936 def test_label_nul_in_name_is_safe_canonical(self) -> None:
937 # NUL within a field value is structurally unusual; the function must
938 # still return a valid canonical ID even for exotic inputs.
939 a = compute_label_id(_REPO_ID, "bug\x00feature", "2026-01-09T00:00:00Z")
940 assert _CANONICAL_RE.match(a)
941
942 def test_webhook_url_with_colons_is_safe(self) -> None:
943 url = "https://user:[email protected]:8080/hook"
944 result = compute_webhook_id(_REPO_ID, url, "2026-01-11T00:00:00Z")
945 assert _CANONICAL_RE.match(result)
946
947 def test_domain_slug_with_path_separator_is_safe(self) -> None:
948 result = compute_domain_id("gabriel", "audio/midi", "2026-01-15T00:00:00Z")
949 assert _CANONICAL_RE.match(result)
950
951 def test_key_id_with_base64_padding_chars_is_safe(self) -> None:
952 pubkey = "ed25519:ABC+/DEF==padded=="
953 result = compute_key_id(_IDENTITY_ID, pubkey)
954 assert _CANONICAL_RE.match(result)
955
956 def test_fork_nul_in_repo_id_does_not_collide(self) -> None:
957 # Synthesize two repo IDs that differ only in NUL placement
958 a = compute_fork_id(_REPO_ID + "\x00x", _FORK_REPO_ID, "2026-01-13T00:00:00Z")
959 b = compute_fork_id(_REPO_ID, "\x00x" + _FORK_REPO_ID, "2026-01-13T00:00:00Z")
960 assert _CANONICAL_RE.match(a)
961 assert a != b
962
963
964 # ===========================================================================
965 # Phase 2 — Tier 3b: Pydantic boundary enforcement for Phase 2 models
966 # ===========================================================================
967
968
969 class TestPydanticBoundaryPhase2:
970
971 # LabelResponse ──────────────────────────────────────────────────────────
972
973 def test_label_response_rejects_bad_label_id(self) -> None:
974 with pytest.raises(ValidationError):
975 LabelResponse(
976 label_id="not-canonical",
977 repo_id=_VALID_ID,
978 name="bug",
979 color="#d73a4a",
980 description=None,
981 created_at=_DT,
982 )
983
984 def test_label_response_rejects_bad_repo_id(self) -> None:
985 with pytest.raises(ValidationError):
986 LabelResponse(
987 label_id=_VALID_ID,
988 repo_id="uuid-style",
989 name="bug",
990 color="#d73a4a",
991 description=None,
992 created_at=_DT,
993 )
994
995 def test_label_response_accepts_future_algo(self) -> None:
996 r = LabelResponse(
997 label_id=_FUTURE_ID,
998 repo_id=_FUTURE_ID,
999 name="bug",
1000 color="#d73a4a",
1001 description=None,
1002 created_at=_DT,
1003 )
1004 assert r.label_id == _FUTURE_ID
1005
1006 # CollaboratorResponse ───────────────────────────────────────────────────
1007
1008 def test_collaborator_response_rejects_bad_collaborator_id(self) -> None:
1009 with pytest.raises(ValidationError):
1010 CollaboratorResponse(
1011 collaborator_id="bad-id",
1012 repo_id=_VALID_ID,
1013 handle="alice",
1014 permission="write",
1015 invited_by=None,
1016 )
1017
1018 def test_collaborator_response_rejects_bad_repo_id(self) -> None:
1019 with pytest.raises(ValidationError):
1020 CollaboratorResponse(
1021 collaborator_id=_VALID_ID,
1022 repo_id="not-genesis",
1023 handle="alice",
1024 permission="write",
1025 invited_by=None,
1026 )
1027
1028 def test_collaborator_response_accepts_future_algo(self) -> None:
1029 r = CollaboratorResponse(
1030 collaborator_id=_FUTURE_ID,
1031 repo_id=_FUTURE_ID,
1032 handle="alice",
1033 permission="write",
1034 invited_by=None,
1035 )
1036 assert r.collaborator_id == _FUTURE_ID
1037
1038 # WebhookResponse ────────────────────────────────────────────────────────
1039
1040 def test_webhook_response_rejects_bad_webhook_id(self) -> None:
1041 with pytest.raises(ValidationError):
1042 WebhookResponse(
1043 webhook_id="uuid",
1044 repo_id=_VALID_ID,
1045 url="https://ci.example.com/hook",
1046 events=["push"],
1047 active=True,
1048 created_at=_DT,
1049 updated_at=_DT,
1050 )
1051
1052 def test_webhook_response_rejects_bad_repo_id(self) -> None:
1053 with pytest.raises(ValidationError):
1054 WebhookResponse(
1055 webhook_id=_VALID_ID,
1056 repo_id="bad",
1057 url="https://ci.example.com/hook",
1058 events=["push"],
1059 active=True,
1060 created_at=_DT,
1061 updated_at=_DT,
1062 )
1063
1064 def test_webhook_response_accepts_future_algo(self) -> None:
1065 r = WebhookResponse(
1066 webhook_id=_FUTURE_ID,
1067 repo_id=_FUTURE_ID,
1068 url="https://ci.example.com/hook",
1069 events=["push"],
1070 active=True,
1071 created_at=_DT,
1072 updated_at=_DT,
1073 )
1074 assert r.webhook_id == _FUTURE_ID
1075
1076 # UserForkedRepoEntry ────────────────────────────────────────────────────
1077
1078 def _make_fork_repo_response(self, repo_id: str = _VALID_ID) -> None:
1079 from musehub.models.musehub import RepoResponse
1080 return RepoResponse(
1081 repo_id=repo_id,
1082 name="my-fork",
1083 owner="alice",
1084 slug="my-fork",
1085 visibility="public",
1086 owner_user_id=_VALID_ID,
1087 clone_url="https://musehub.ai/api/repos/x",
1088 created_at=_DT,
1089 updated_at=_DT,
1090 )
1091
1092 def test_forked_repo_entry_rejects_bad_fork_id(self) -> None:
1093 with pytest.raises(ValidationError):
1094 UserForkedRepoEntry(
1095 fork_id="uuid-style",
1096 fork_repo=self._make_fork_repo_response(),
1097 source_owner="gabriel",
1098 source_slug="original-repo",
1099 forked_at=_DT,
1100 )
1101
1102 def test_forked_repo_entry_rejects_bad_source_repo_id(self) -> None:
1103 # UserForkedRepoEntry doesn't hold source_repo_id directly — fork_id is the only genesis field
1104 # Validate that a bad fork_id fails and a good one passes even if source info is minimal
1105 with pytest.raises(ValidationError):
1106 UserForkedRepoEntry(
1107 fork_id="bad-id",
1108 fork_repo=self._make_fork_repo_response(),
1109 source_owner="gabriel",
1110 source_slug="original-repo",
1111 forked_at=_DT,
1112 )
1113
1114 def test_forked_repo_entry_rejects_bad_fork_repo_id(self) -> None:
1115 # The nested RepoResponse.repo_id is also genesis-validated
1116 with pytest.raises(ValidationError):
1117 UserForkedRepoEntry(
1118 fork_id=_VALID_ID,
1119 fork_repo=self._make_fork_repo_response(repo_id="bad-id"),
1120 source_owner="gabriel",
1121 source_slug="original-repo",
1122 forked_at=_DT,
1123 )
1124
1125 def test_forked_repo_entry_accepts_future_algo(self) -> None:
1126 from musehub.models.musehub import RepoResponse
1127 fake_repo = RepoResponse(
1128 repo_id=_FUTURE_ID,
1129 name="my-fork",
1130 owner="alice",
1131 slug="my-fork",
1132 visibility="public",
1133 owner_user_id=_FUTURE_ID,
1134 clone_url="https://musehub.ai/api/repos/x",
1135 created_at=_DT,
1136 updated_at=_DT,
1137 )
1138 r = UserForkedRepoEntry(
1139 fork_id=_FUTURE_ID,
1140 fork_repo=fake_repo,
1141 source_owner="gabriel",
1142 source_slug="original-repo",
1143 forked_at=_DT,
1144 )
1145 assert r.fork_id == _FUTURE_ID
1146
1147
1148 # ===========================================================================
1149 # Phase 2 — Tier 6: service-layer genesis contract tests
1150 # ===========================================================================
1151
1152
1153 class TestServiceLayerPhase2GenesisIds:
1154 """Phase 2 service creation functions assign genesis-addressed IDs, never uuid4."""
1155
1156 def _async_session(self, *, execute_returns: MagicMock | None = None) -> "AsyncMock":
1157 from unittest.mock import AsyncMock, MagicMock
1158
1159 session = AsyncMock()
1160 scalar = MagicMock()
1161 scalar.scalar_one_or_none.return_value = execute_returns
1162 session.execute.return_value = scalar
1163 session.commit = AsyncMock()
1164 session.flush = AsyncMock()
1165
1166 async def _refresh(obj: MagicMock) -> None:
1167 from datetime import datetime, timezone
1168 for attr in ("created_at", "updated_at"):
1169 if not getattr(obj, attr, None):
1170 try:
1171 setattr(obj, attr, datetime.now(timezone.utc))
1172 except Exception:
1173 pass
1174
1175 session.refresh = _refresh
1176 return session
1177
1178 # 6.6 Label — create_label ─────────────────────────────────────────────
1179
1180 @pytest.mark.asyncio
1181 async def test_create_label_uses_compute_label_id(self) -> None:
1182 """create_label calls compute_label_id(repo_id, name, iso_ts) — not uuid4."""
1183 from unittest.mock import patch, AsyncMock, MagicMock
1184 import musehub.api.routes.musehub.labels as labels_module
1185
1186 repo_id = _REPO_ID
1187 name = "bug"
1188
1189 captured: list[tuple] = []
1190 _real = compute_label_id
1191 def _spy(r: str, n: str, t: str) -> None:
1192 result = _real(r, n, t)
1193 captured.append((r, n, t, result))
1194 return result
1195
1196 db = self._async_session()
1197 # _guard_repo_owner → get_repo + check_write_access; uniqueness check → no duplicate
1198 db.execute.return_value.scalar_one_or_none.return_value = None
1199
1200 with patch("musehub.api.routes.musehub.labels.musehub_repository") as mock_svc, \
1201 patch("musehub.api.routes.musehub.labels.compute_label_id", side_effect=_spy):
1202 fake_repo = MagicMock()
1203 mock_svc.get_repo = AsyncMock(return_value=fake_repo)
1204 mock_svc.check_write_access = AsyncMock(return_value=True)
1205
1206 try:
1207 await labels_module.create_label(
1208 repo_id=repo_id,
1209 body=labels_module.LabelCreate(name=name, color="#d73a4a"),
1210 db=db,
1211 token=MagicMock(handle="gabriel"),
1212 )
1213 except Exception:
1214 pass
1215
1216 assert captured, "compute_label_id was never called — label creation is broken"
1217 call_repo_id, call_name, call_ts, call_result = captured[0]
1218 assert call_repo_id == repo_id
1219 assert call_name == name
1220 assert _CANONICAL_RE.match(call_result)
1221
1222 # 6.7 Webhook — create_webhook ─────────────────────────────────────────
1223
1224 @pytest.mark.asyncio
1225 async def test_create_webhook_uses_compute_webhook_id(self) -> None:
1226 """Webhook rows are assigned compute_webhook_id(repo_id, url, created_at_iso)."""
1227 from unittest.mock import patch, AsyncMock, MagicMock
1228 from musehub.services.musehub_webhook_dispatcher import create_webhook
1229
1230 repo_id = _REPO_ID
1231 url = "https://ci.example.com/hook"
1232
1233 captured: list[tuple] = []
1234 _real = compute_webhook_id
1235 def _spy(r: str, u: str, t: str) -> None:
1236 result = _real(r, u, t)
1237 captured.append((r, u, t, result))
1238 return result
1239
1240 db = self._async_session()
1241
1242 with patch("musehub.services.musehub_webhook_dispatcher.compute_webhook_id", side_effect=_spy):
1243 try:
1244 await create_webhook(db, repo_id=repo_id, url=url, events=["push"], secret="")
1245 except Exception:
1246 pass
1247
1248 assert captured, "compute_webhook_id was never called — webhook creation is broken"
1249 call_repo_id, call_url, call_ts, call_result = captured[0]
1250 assert call_repo_id == repo_id
1251 assert call_url == url
1252 assert _CANONICAL_RE.match(call_result)
1253 # Verify the row passed to session.add has that ID
1254 if db.add.called:
1255 row = db.add.call_args_list[0][0][0]
1256 assert row.webhook_id == call_result
1257
1258 # 6.8 Fork — compute_fork_id called at fork creation ───────────────────
1259
1260 @pytest.mark.asyncio
1261 async def test_fork_repo_uses_compute_fork_id(self) -> None:
1262 """fork_repo calls compute_fork_id(source_repo_id, fork_repo_id, created_at_iso)."""
1263 from unittest.mock import patch, AsyncMock, MagicMock
1264 from musehub.services.musehub_repository import fork_repo
1265 from musehub.models.musehub import ForkRepoRequest
1266 from datetime import datetime, timezone
1267
1268 source_repo_id = _REPO_ID
1269 captured: list[tuple] = []
1270 _real = compute_fork_id
1271 def _spy(src: str, frk: str, t: str) -> None:
1272 result = _real(src, frk, t)
1273 captured.append((src, frk, t, result))
1274 return result
1275
1276 fake_source_repo = MagicMock()
1277 fake_source_repo.repo_id = source_repo_id
1278 fake_source_repo.name = "my-repo"
1279 fake_source_repo.slug = "my-repo"
1280 fake_source_repo.visibility = "public"
1281 fake_source_repo.description = "A test repo"
1282 fake_source_repo.domain = "code"
1283 fake_source_repo.owner = "gabriel"
1284 fake_source_repo.tags = []
1285
1286 db = self._async_session()
1287 call_count = 0
1288 def _execute(*a: MagicMock, **kw: MagicMock) -> None:
1289 nonlocal call_count
1290 call_count += 1
1291 m = MagicMock()
1292 if call_count == 1:
1293 m.scalar_one_or_none.return_value = fake_source_repo # source repo
1294 else:
1295 m.scalar_one_or_none.return_value = None # no duplicate / no slug collision
1296 return m
1297 db.execute.side_effect = _execute
1298
1299 # After flush()+refresh(), the fork_repo_row needs a repo_id so compute_fork_id can use it.
1300 # The DB would normally auto-assign it; we supply a canonical stub.
1301 _fork_repo_id_stub = fake_id("fork-repo-stub")
1302 refresh_count = 0
1303 async def _refresh_with_repo_id(obj: MagicMock) -> None:
1304 nonlocal refresh_count
1305 refresh_count += 1
1306 now = datetime.now(timezone.utc)
1307 for attr in ("created_at", "updated_at"):
1308 if not getattr(obj, attr, None):
1309 try:
1310 setattr(obj, attr, now)
1311 except Exception:
1312 pass
1313 # First refresh is for the fork repo row — give it a genesis-style repo_id.
1314 if refresh_count == 1 and hasattr(obj, "repo_id") and not getattr(obj, "repo_id", None):
1315 try:
1316 obj.repo_id = _fork_repo_id_stub
1317 except Exception:
1318 pass
1319 db.refresh = _refresh_with_repo_id
1320
1321 with patch("musehub.services.musehub_repository.compute_fork_id", side_effect=_spy):
1322 try:
1323 await fork_repo(
1324 db,
1325 source_repo_id=source_repo_id,
1326 forked_by_handle="alice",
1327 request=ForkRepoRequest(name=None),
1328 )
1329 except Exception:
1330 pass
1331
1332 assert captured, "compute_fork_id was never called — fork creation is broken"
1333 call_src, call_frk, call_ts, call_result = captured[0]
1334 assert call_src == source_repo_id
1335 assert _CANONICAL_RE.match(call_result)
1336
1337 # 6.9 Auth key — register_identity key_id ──────────────────────────────
1338
1339 @pytest.mark.asyncio
1340 async def test_register_key_uses_compute_key_id(self) -> None:
1341 """MusehubAuthKey rows use compute_key_id(identity_id, public_key_b64) — no uuid4."""
1342 from muse.core.types import encode_pubkey, public_key_fingerprint
1343 from musehub.services.musehub_auth import register_agent_identity
1344
1345 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
1346 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
1347 priv = Ed25519PrivateKey.generate()
1348 pub = priv.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
1349 pub_b64 = encode_pubkey("ed25519", pub)
1350 fp = public_key_fingerprint(pub)
1351 expected_identity_id = compute_identity_id(pub)
1352 expected_key_id = compute_key_id(expected_identity_id, pub_b64)
1353
1354 db = self._async_session()
1355 await register_agent_identity(
1356 session=db,
1357 handle="key-test-agent",
1358 public_key_b64=pub_b64,
1359 fingerprint=fp,
1360 algorithm="ed25519",
1361 spawned_by="gabriel",
1362 )
1363
1364 # add() call list: [0] = identity, [1] = key
1365 assert len(db.add.call_args_list) >= 2
1366 key_row = db.add.call_args_list[1][0][0]
1367 assert key_row.key_id == expected_key_id
1368 assert _CANONICAL_RE.match(key_row.key_id)
1369
1370 # 6.10 Collaborator invite — invite_collaborator ────────────────────────
1371
1372 @pytest.mark.asyncio
1373 async def test_invite_collaborator_uses_compute_collaborator_id(self) -> None:
1374 """Collaborator rows use compute_collaborator_id(repo_id, identity_id, invited_at_iso)."""
1375 from unittest.mock import patch, AsyncMock, MagicMock
1376 from datetime import datetime, timezone
1377 import musehub.api.routes.musehub.collaborators as collabs_module
1378
1379 repo_id = _REPO_ID
1380 fixed_now = datetime(2026, 1, 14, 0, 0, 0, tzinfo=timezone.utc)
1381 invitee_identity_id = _COLLAB_IDENTITY_ID
1382 expected = compute_collaborator_id(repo_id, invitee_identity_id, fixed_now.isoformat())
1383
1384 # Repo owner must match the actor so the 403 guard passes.
1385 fake_repo = MagicMock()
1386 fake_repo.owner = "gabriel"
1387
1388 fake_invitee_identity = MagicMock()
1389 fake_invitee_identity.identity_id = invitee_identity_id
1390
1391 db = self._async_session()
1392 call_count = 0
1393 def _execute(*a: MagicMock, **kw: MagicMock) -> None:
1394 nonlocal call_count
1395 call_count += 1
1396 m = MagicMock()
1397 if call_count == 1:
1398 m.scalar_one_or_none.return_value = MagicMock(permission="owner") # actor perm
1399 elif call_count == 2:
1400 m.scalar_one_or_none.return_value = None # no duplicate
1401 else:
1402 m.scalar_one_or_none.return_value = fake_invitee_identity # invitee lookup
1403 return m
1404 db.execute.side_effect = _execute
1405
1406 with patch("musehub.api.routes.musehub.collaborators.musehub_repository") as mock_svc, \
1407 patch("musehub.api.routes.musehub.collaborators.datetime") as mock_dt:
1408 mock_svc.get_repo = AsyncMock(return_value=fake_repo)
1409 mock_dt.now.return_value = fixed_now
1410 mock_dt.timezone = timezone
1411
1412 try:
1413 await collabs_module.invite_collaborator(
1414 repo_id=repo_id,
1415 body=collabs_module.CollaboratorInviteRequest(handle="alice"),
1416 db=db,
1417 token=MagicMock(handle="gabriel"),
1418 )
1419 except Exception:
1420 pass
1421
1422 assert db.add.called, "db.add was never called — collaborator row was not created"
1423 row = db.add.call_args_list[0][0][0]
1424 assert row.id == expected, f"expected {expected!r}, got {row.id!r}"
1425 assert _CANONICAL_RE.match(row.id)
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 143 days ago