"""Canonical ID contract tests — pins the sha256: prefix format for all content-addressed IDs. Every content-addressed ID in the Muse ecosystem uses: ``sha256:<64 lowercase hex chars>`` This covers object_ids, snapshot_ids, and commit_ids. Random/opaque IDs (repo_id, collaborator IDs, etc.) are excluded from this contract. Tiers: 1. Hash functions return canonical form (synchronous) 2. WireCommit validation (synchronous) 3. WireSnapshot validation (synchronous) 4. Wire push endpoint rejects non-canonical IDs (async, requires DB) """ from __future__ import annotations import re import sys from datetime import datetime, timezone from pathlib import Path import msgpack import pytest from httpx import AsyncClient from pydantic import ValidationError from sqlalchemy.ext.asyncio import AsyncSession from musehub.muse_cli.snapshot import compute_snapshot_id, compute_commit_id from musehub.types.json_types import JSONObject, JSONValue from musehub.models.wire import WireCommit, WireSnapshot, WireSnapshotDelta # Cross-verify against the Muse CLI implementation. sys.path.insert(0, str(Path.home() / "ecosystem" / "muse")) from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter from muse.core.snapshot import compute_snapshot_id as cli_compute_snapshot_id from muse.core.snapshot import compute_commit_id as cli_compute_commit_id from muse.core.types import long_id, now_utc_iso from tests.factories import create_repo # ── Module-level canonical ID regex ─────────────────────────────────────────── _CANONICAL_ID_RE = re.compile(r"^sha256:[0-9a-f]{64}$") # ── Shared test inputs (deterministic) ──────────────────────────────────────── _MANIFEST: dict[str, str] = { "README.md": long_id("a" * 64), "src/main.py": long_id("b" * 64), } _DIRS: list[str] = ["src"] _MESSAGE = "feat: add canonical ID tests" _TIMESTAMP = "2026-01-01T00:00:00+00:00" _PARENT_IDS: list[str] = [] # Pre-computed IDs used across tiers. _SNAP_ID_HUB = compute_snapshot_id(_MANIFEST, _DIRS) _COMMIT_ID_HUB = compute_commit_id(_PARENT_IDS, _SNAP_ID_HUB, _MESSAGE, _TIMESTAMP) # A valid canonical ID — a real sha256 hex wrapped in the prefix. from muse.core.types import fake_id, blob_id _VALID_SNAP_ID = fake_id("snapshot-fixture") _VALID_COMMIT_ID = fake_id("commit-fixture") _VALID_PARENT_ID = fake_id("parent-fixture") _VALID_OBJECT_ID = fake_id("object-fixture") # Bare hex (no prefix) — always invalid in the canonical contract. # Strip the sha256: prefix to produce deliberately unprefixed hex. from muse.core.types import long_id _BARE_HEX_64 = long_id(fake_id("bare"), strip=True) # 64-char hex, no prefix assert len(_BARE_HEX_64) == 64 # Plain string — not a content-addressed ID, also invalid. _PLAIN_ID = "not-a-content-addressed-id" # ── Helpers ─────────────────────────────────────────────────────────────────── _fw = MuseWireFrameWriter() _SFRAME_HEADER = "H" _SFRAME_COMMIT_PACK = "C" _SFRAME_END = "E" _SFRAME_ERROR = "X" _SFRAME_RESULT = "R" def _wrap(ft: str, data: JSONValue) -> bytes: return _fw.wrap(frame_type=ft, payload=msgpack.packb(data, use_bin_type=True)) def _make_mwp_push( *, commit_id: str = _VALID_COMMIT_ID, snapshot_id: str = _VALID_SNAP_ID, ) -> bytes: """Build a minimal MWP push body with controllable commit_id and snapshot_id.""" header = _wrap(_SFRAME_HEADER, { "t": _SFRAME_HEADER, "branch": "main", "force": False, "have": [], "head": commit_id, "n_objects": 0, "n_commits": 1, }) commit_pack = _wrap(_SFRAME_COMMIT_PACK, { "t": _SFRAME_COMMIT_PACK, "commits": [{ "commit_id": commit_id, "parent_commit_id": None, "parent2_commit_id": None, "snapshot_id": snapshot_id, "branch": "main", "message": "test: canonical ID push", "author": "Test User ", "committed_at": now_utc_iso(), "signature": "", "signer_key_id": "", "agent_id": "", "model_id": "", "metadata": {}, }], "snapshots": [{"snapshot_id": snapshot_id, "manifest": {}, "committed_at": now_utc_iso()}], }) end = _wrap(_SFRAME_END, {"t": _SFRAME_END, "n_objects": 0, "n_commits": 1}) return header + commit_pack + end def _parse_mwp_result(raw: bytes) -> JSONObject: """Return the last msgpack frame from an MWP response stream.""" unpacker = msgpack.Unpacker(raw=False) unpacker.feed(raw) last: JSONObject = {} for frame in unpacker: last = frame return last # ── Tier 1 — Hash functions return canonical form ───────────────────────────── class TestHashFunctionsReturnCanonicalForm: """compute_snapshot_id and compute_commit_id must return sha256:<64-hex> strings.""" def test_compute_snapshot_id_hub_returns_canonical(self) -> None: result = compute_snapshot_id(_MANIFEST, _DIRS) assert _CANONICAL_ID_RE.match(result), ( f"compute_snapshot_id (hub) returned {result!r}, " f"expected sha256:<64 lowercase hex chars>" ) def test_compute_commit_id_hub_returns_canonical(self) -> None: snap_id = compute_snapshot_id(_MANIFEST, _DIRS) result = compute_commit_id(_PARENT_IDS, snap_id, _MESSAGE, _TIMESTAMP) assert _CANONICAL_ID_RE.match(result), ( f"compute_commit_id (hub) returned {result!r}, " f"expected sha256:<64 lowercase hex chars>" ) def test_compute_snapshot_id_cli_returns_canonical(self) -> None: result = cli_compute_snapshot_id(_MANIFEST, _DIRS) assert _CANONICAL_ID_RE.match(result), ( f"compute_snapshot_id (cli) returned {result!r}, " f"expected sha256:<64 lowercase hex chars>" ) def test_compute_commit_id_cli_returns_canonical(self) -> None: snap_id = cli_compute_snapshot_id(_MANIFEST, _DIRS) result = cli_compute_commit_id(_PARENT_IDS, snap_id, _MESSAGE, _TIMESTAMP) assert _CANONICAL_ID_RE.match(result), ( f"compute_commit_id (cli) returned {result!r}, " f"expected sha256:<64 lowercase hex chars>" ) def test_hub_and_cli_snapshot_id_agree(self) -> None: """Both implementations must produce identical output for the same manifest.""" hub_result = compute_snapshot_id(_MANIFEST, _DIRS) cli_result = cli_compute_snapshot_id(_MANIFEST, _DIRS) assert hub_result == cli_result, ( f"snapshot_id mismatch: hub={hub_result!r}, cli={cli_result!r}" ) def test_hub_and_cli_commit_id_agree(self) -> None: """Both implementations must produce identical output for the same inputs.""" hub_snap = compute_snapshot_id(_MANIFEST, _DIRS) cli_snap = cli_compute_snapshot_id(_MANIFEST, _DIRS) # snapshot_ids must match first (tested separately), use hub value as input. hub_commit = compute_commit_id(_PARENT_IDS, hub_snap, _MESSAGE, _TIMESTAMP) cli_commit = cli_compute_commit_id(_PARENT_IDS, cli_snap, _MESSAGE, _TIMESTAMP) assert hub_commit == cli_commit, ( f"commit_id mismatch: hub={hub_commit!r}, cli={cli_commit!r}" ) def test_snapshot_id_is_deterministic(self) -> None: """Same inputs always produce the same snapshot_id.""" a = compute_snapshot_id(_MANIFEST, _DIRS) b = compute_snapshot_id(_MANIFEST, _DIRS) assert a == b def test_commit_id_is_deterministic(self) -> None: """Same inputs always produce the same commit_id.""" snap = compute_snapshot_id(_MANIFEST, _DIRS) a = compute_commit_id(_PARENT_IDS, snap, _MESSAGE, _TIMESTAMP) b = compute_commit_id(_PARENT_IDS, snap, _MESSAGE, _TIMESTAMP) assert a == b def test_snapshot_id_without_dirs_returns_canonical(self) -> None: """Omitting the directories argument still returns canonical form.""" result = compute_snapshot_id(_MANIFEST) assert _CANONICAL_ID_RE.match(result), ( f"compute_snapshot_id (no dirs) returned {result!r}" ) def test_snapshot_id_empty_manifest_returns_canonical(self) -> None: """An empty manifest produces a canonical ID (not an error).""" result = compute_snapshot_id({}) assert _CANONICAL_ID_RE.match(result), ( f"compute_snapshot_id (empty manifest) returned {result!r}" ) # ── Tier 2 — WireCommit validation ──────────────────────────────────────────── class TestWireCommitValidation: """WireCommit must enforce sha256: prefix on commit_id, snapshot_id, and parent_commit_id.""" def test_canonical_commit_id_accepted(self) -> None: commit = WireCommit( commit_id=_VALID_COMMIT_ID, snapshot_id=_VALID_SNAP_ID, ) assert commit.commit_id == _VALID_COMMIT_ID def test_bare_hex_commit_id_rejected(self) -> None: """A 64-char hex commit_id without the sha256: prefix must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireCommit( commit_id=_BARE_HEX_64, snapshot_id=_VALID_SNAP_ID, ) def test_plain_id_commit_id_rejected(self) -> None: """A plain string commit_id is not a content-addressed ID and must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireCommit( commit_id=_PLAIN_ID, snapshot_id=_VALID_SNAP_ID, ) def test_bare_hex_snapshot_id_rejected(self) -> None: """WireCommit.snapshot_id must also carry the sha256: prefix.""" with pytest.raises((ValueError, ValidationError)): WireCommit( commit_id=_VALID_COMMIT_ID, snapshot_id=_BARE_HEX_64, ) def test_canonical_parent_commit_id_accepted(self) -> None: """A canonical parent_commit_id is valid.""" commit = WireCommit( commit_id=_VALID_COMMIT_ID, snapshot_id=_VALID_SNAP_ID, parent_commit_id=_VALID_PARENT_ID, ) assert commit.parent_commit_id == _VALID_PARENT_ID def test_none_parent_commit_id_accepted(self) -> None: """parent_commit_id=None is valid (root commit).""" commit = WireCommit( commit_id=_VALID_COMMIT_ID, snapshot_id=_VALID_SNAP_ID, parent_commit_id=None, ) assert commit.parent_commit_id is None def test_bare_hex_parent_commit_id_rejected(self) -> None: """parent_commit_id when non-None must carry the sha256: prefix.""" with pytest.raises((ValueError, ValidationError)): WireCommit( commit_id=_VALID_COMMIT_ID, snapshot_id=_VALID_SNAP_ID, parent_commit_id=_BARE_HEX_64, ) # ── Tier 3 — WireSnapshot validation ────────────────────────────────────────── class TestWireSnapshotValidation: """WireSnapshot must enforce sha256: prefix on snapshot_id.""" def test_canonical_snapshot_id_accepted(self) -> None: snap = WireSnapshot(snapshot_id=_VALID_SNAP_ID) assert snap.snapshot_id == _VALID_SNAP_ID def test_bare_hex_snapshot_id_rejected(self) -> None: """A 64-char hex snapshot_id without the sha256: prefix must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireSnapshot(snapshot_id=_BARE_HEX_64) def test_plain_id_snapshot_id_rejected(self) -> None: """A plain string snapshot_id is not a content-addressed ID and must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireSnapshot(snapshot_id=_PLAIN_ID) def test_canonical_snapshot_with_manifest_accepted(self) -> None: snap = WireSnapshot( snapshot_id=_VALID_SNAP_ID, manifest={"README.md": _VALID_OBJECT_ID}, ) assert snap.snapshot_id == _VALID_SNAP_ID assert "README.md" in snap.manifest def test_manifest_bare_hex_object_id_rejected(self) -> None: """Manifest values that are bare hex (no sha256: prefix) must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireSnapshot( snapshot_id=_VALID_SNAP_ID, manifest={"src/main.py": _BARE_HEX_64}, ) def test_manifest_plain_id_object_id_rejected(self) -> None: """Manifest values that are plain strings (not sha256:) must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireSnapshot( snapshot_id=_VALID_SNAP_ID, manifest={"src/main.py": _PLAIN_ID}, ) def test_manifest_multiple_entries_all_canonical(self) -> None: """All manifest values must be validated — not just the first.""" with pytest.raises((ValueError, ValidationError)): WireSnapshot( snapshot_id=_VALID_SNAP_ID, manifest={ "README.md": _VALID_OBJECT_ID, # good "src/main.py": _BARE_HEX_64, # bad — second entry }, ) class TestWireSnapshotDeltaValidation: """WireSnapshotDelta.added values must carry the sha256: prefix.""" def test_canonical_delta_accepted(self) -> None: delta = WireSnapshotDelta( snapshot_id=_VALID_SNAP_ID, base_id=_VALID_SNAP_ID, added={"src/new.py": _VALID_OBJECT_ID}, ) assert delta.added["src/new.py"] == _VALID_OBJECT_ID def test_added_bare_hex_object_id_rejected(self) -> None: """added values that are bare hex (no sha256: prefix) must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireSnapshotDelta( snapshot_id=_VALID_SNAP_ID, base_id=_VALID_SNAP_ID, added={"src/new.py": _BARE_HEX_64}, ) def test_added_plain_id_object_id_rejected(self) -> None: """added values that are plain strings (not sha256:) must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireSnapshotDelta( snapshot_id=_VALID_SNAP_ID, base_id=_VALID_SNAP_ID, added={"src/new.py": _PLAIN_ID}, ) def test_added_multiple_entries_all_validated(self) -> None: """All added values must be validated — not just the first.""" with pytest.raises((ValueError, ValidationError)): WireSnapshotDelta( snapshot_id=_VALID_SNAP_ID, base_id=_VALID_SNAP_ID, added={ "README.md": _VALID_OBJECT_ID, # good "src/bad.py": _BARE_HEX_64, # bad }, ) def test_empty_added_accepted(self) -> None: """An empty added dict (no new files) is valid.""" delta = WireSnapshotDelta( snapshot_id=_VALID_SNAP_ID, base_id=_VALID_SNAP_ID, ) assert delta.added == {} class TestWireCommitPromptHashValidation: """WireCommit.prompt_hash must be empty or sha256:<64-hex>.""" def test_empty_prompt_hash_accepted(self) -> None: commit = WireCommit(commit_id=_VALID_COMMIT_ID, prompt_hash="") assert commit.prompt_hash == "" def test_canonical_prompt_hash_accepted(self) -> None: commit = WireCommit( commit_id=_VALID_COMMIT_ID, prompt_hash=_VALID_OBJECT_ID, ) assert commit.prompt_hash == _VALID_OBJECT_ID def test_bare_hex_prompt_hash_rejected(self) -> None: """A bare 64-char hex prompt_hash without sha256: prefix must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireCommit( commit_id=_VALID_COMMIT_ID, prompt_hash=_BARE_HEX_64, ) def test_arbitrary_string_prompt_hash_rejected(self) -> None: """An arbitrary string prompt_hash must be rejected.""" with pytest.raises((ValueError, ValidationError)): WireCommit( commit_id=_VALID_COMMIT_ID, prompt_hash="abc123", ) # ── Tier 4 — Wire push/stream endpoint rejects non-canonical IDs ────────────── @pytest.mark.asyncio async def test_push_with_canonical_ids_returns_ok( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], monkeypatch: pytest.MonkeyPatch, ) -> None: """A push with fully canonical sha256: IDs throughout must succeed (ok=True result frame).""" from tests.test_wire_push_stream import _stub_r2_backend _stub_r2_backend(monkeypatch) repo = await create_repo(db_session, slug="canonical-ids-ok", owner="testuser") body = _make_mwp_push(commit_id=_VALID_COMMIT_ID, snapshot_id=_VALID_SNAP_ID) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code == 200, f"Expected 200 but got {resp.status_code}: {resp.text}" result = _parse_mwp_result(resp.content) assert result.get("ok") is True, f"Expected ok=True result frame, got: {result}" @pytest.mark.asyncio async def test_push_with_bare_hex_commit_id_returns_error( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], monkeypatch: pytest.MonkeyPatch, ) -> None: """A push where commit_id is bare hex (no sha256: prefix) must be rejected.""" from tests.test_wire_push_stream import _stub_r2_backend _stub_r2_backend(monkeypatch) repo = await create_repo(db_session, slug="bare-hex-commit-err", owner="testuser") body = _make_mwp_push(commit_id=_BARE_HEX_64, snapshot_id=_VALID_SNAP_ID) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code in (200, 422), f"Expected rejection but got {resp.status_code}" if resp.status_code == 200: result = _parse_mwp_result(resp.content) assert result.get("ok") is not True, f"Expected rejection, got ok=True: {result}" @pytest.mark.asyncio async def test_push_with_bare_hex_snapshot_id_returns_error( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], monkeypatch: pytest.MonkeyPatch, ) -> None: """A push where snapshot_id is bare hex (no sha256: prefix) must be rejected.""" from tests.test_wire_push_stream import _stub_r2_backend _stub_r2_backend(monkeypatch) repo = await create_repo(db_session, slug="bare-hex-snap-err", owner="testuser") body = _make_mwp_push(commit_id=_VALID_COMMIT_ID, snapshot_id=_BARE_HEX_64) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code in (200, 422), f"Expected rejection but got {resp.status_code}" if resp.status_code == 200: result = _parse_mwp_result(resp.content) assert result.get("ok") is not True, f"Expected rejection, got ok=True: {result}"