"""Tests for muse.core.attestation — Phase 7.1 framework interface. Covers Rule #0 tiers that apply to a pure framework module: * unit — Protocol conformance, registry thread-safety, error hierarchy. * integration — Stub provider compute → anchor → verify round-trip. * security — Misuse-resistance: partial verification raises, anchor() is write-once, registry isolation between domains, error type hierarchy is the only path for failure signalling. The "stress" / "performance" tiers are intentionally omitted at this layer — the framework has no I/O, so latency/throughput tests would only measure Python interpreter overhead and add noise without value. Stress at the provider implementations (Phase 7.2 / 7.4) is where it belongs. """ from __future__ import annotations import threading from typing import cast import pytest from muse.core.attestation import ( AnchorReceipt, AttestationError, AttestationRecord, AttestationRequiredError, AttestationVerifyError, ImmutableReceiptError, MuseAttestationProvider, VerifyResult, _registered_domains, get_attestation_provider, register_attestation_provider, unregister_attestation_provider, ) # --------------------------------------------------------------------------- # Stub providers — minimal, deterministic, used across every tier. # --------------------------------------------------------------------------- class _StubProvider: """Deterministic in-memory stub that satisfies MuseAttestationProvider.""" def __init__(self, provider_id: str = "stub") -> None: self.provider_id = provider_id self._anchored: dict[str, AnchorReceipt] = {} self._lock = threading.Lock() def compute_attestation( self, snapshot_id: str, commit_meta: dict[str, object] ) -> AttestationRecord: path = str(commit_meta.get("path", "")) return { "id": f"{snapshot_id}:{path}", "action": str(commit_meta.get("action", "write")), "path": path, "timestamp": "2026-05-13T00:00:00Z", "content_hash": str(commit_meta.get("content_hash", "")), "sig": "stub-sig", "algorithm": "stub", "provider_id": self.provider_id, } def anchor(self, record: AttestationRecord) -> AnchorReceipt: rid = record.get("id", "") with self._lock: if rid in self._anchored: raise ImmutableReceiptError( f"record {rid!r} already anchored (write-once)" ) receipt: AnchorReceipt = { "tx_id": f"tx-{len(self._anchored) + 1}", "anchor_url": f"https://stub.example/attest/{rid}", "anchored_at": "2026-05-13T00:00:01Z", "provider_id": self.provider_id, } self._anchored[rid] = receipt return receipt def verify( self, record: AttestationRecord, receipt: AnchorReceipt ) -> VerifyResult: if record.get("provider_id") != receipt.get("provider_id"): raise AttestationVerifyError( "provider_id mismatch between record and receipt" ) with self._lock: anchored = self._anchored.get(record.get("id", "")) if anchored is None: raise AttestationVerifyError( "no receipt on file — cannot complete verification", code="NO_RECEIPT", ) if anchored != receipt: return { "valid": False, "reason": "receipt does not match anchored copy", "depth": 0, "provider_id": self.provider_id, } return { "valid": True, "reason": "", "depth": 0, "provider_id": self.provider_id, } class _NotAProvider: """Object that does NOT implement the protocol (missing methods).""" def compute_attestation(self, snapshot_id: str, commit_meta: dict[str, object]) -> AttestationRecord: # noqa: ARG002 return cast(AttestationRecord, {}) # Intentionally missing anchor() and verify(). # --------------------------------------------------------------------------- # Auto-cleanup fixture — keep the registry pristine between tests. # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _clean_registry() -> None: """Snapshot and restore registered providers around every test.""" before = _registered_domains() yield for d in _registered_domains(): if d not in before: unregister_attestation_provider(d) # =========================================================================== # UNIT TIER # =========================================================================== class TestProtocolConformance: """Verify Protocol structural typing works as documented.""" def test_stub_provider_satisfies_protocol(self) -> None: """Stub provider with all three methods is recognised as the Protocol.""" provider = _StubProvider() assert isinstance(provider, MuseAttestationProvider) def test_incomplete_object_does_not_satisfy_protocol(self) -> None: """Objects missing required methods fail isinstance — registration rejects.""" with pytest.raises(TypeError, match="MuseAttestationProvider"): register_attestation_provider("nope", cast(MuseAttestationProvider, _NotAProvider())) class TestErrorHierarchy: """Every typed error derives from AttestationError; codes are stable.""" def test_required_error_subclasses_base(self) -> None: """AttestationRequiredError is catchable via the base class.""" with pytest.raises(AttestationError): raise AttestationRequiredError("required but missing") def test_verify_error_subclasses_base(self) -> None: """AttestationVerifyError is catchable via the base class.""" with pytest.raises(AttestationError): raise AttestationVerifyError("cannot verify") def test_immutable_receipt_subclasses_base(self) -> None: """ImmutableReceiptError is catchable via the base class.""" with pytest.raises(AttestationError): raise ImmutableReceiptError("write-once violation") def test_required_error_code_constant(self) -> None: """code attribute is the published constant — used by CLI/MCP layers.""" try: raise AttestationRequiredError("x") except AttestationRequiredError as exc: assert exc.code == "ATTESTATION_REQUIRED" def test_verify_error_default_and_override_codes(self) -> None: """code defaults to VERIFY_INCOMPLETE; overrides are honoured.""" try: raise AttestationVerifyError("default") except AttestationVerifyError as exc: assert exc.code == "VERIFY_INCOMPLETE" try: raise AttestationVerifyError("custom", code="ICP_UNREACHABLE") except AttestationVerifyError as exc: assert exc.code == "ICP_UNREACHABLE" def test_immutable_receipt_code_constant(self) -> None: """ImmutableReceiptError carries its stable code.""" try: raise ImmutableReceiptError("dup") except ImmutableReceiptError as exc: assert exc.code == "IMMUTABLE_RECEIPT" class TestRegistryBasics: """register/get/unregister behave per documented contract.""" def test_register_then_get_returns_same_instance(self) -> None: """get() returns the exact instance passed to register().""" provider = _StubProvider() register_attestation_provider("stubdom", provider) assert get_attestation_provider("stubdom") is provider def test_get_unregistered_domain_returns_none(self) -> None: """Looking up a domain with no provider returns None, not a fallback.""" assert get_attestation_provider("never-registered") is None def test_register_replaces_previous(self) -> None: """Re-registering the same domain replaces the prior provider.""" first = _StubProvider("first") second = _StubProvider("second") register_attestation_provider("stubdom", first) register_attestation_provider("stubdom", second) assert get_attestation_provider("stubdom") is second def test_unregister_returns_true_when_present(self) -> None: """unregister returns True when a provider was removed.""" register_attestation_provider("stubdom", _StubProvider()) assert unregister_attestation_provider("stubdom") is True def test_unregister_returns_false_when_absent(self) -> None: """unregister returns False on a domain that was never registered.""" assert unregister_attestation_provider("never-here") is False def test_register_rejects_empty_domain(self) -> None: """Empty domain names are rejected at registration time.""" with pytest.raises(ValueError, match="non-empty string"): register_attestation_provider("", _StubProvider()) def test_register_rejects_non_string_domain(self) -> None: """Non-string domain identifiers are rejected at registration time.""" with pytest.raises(ValueError, match="non-empty string"): register_attestation_provider(cast(str, 42), _StubProvider()) class TestRegistryThreadSafety: """Concurrent registers/gets must not lose entries.""" def test_concurrent_registrations_visible_to_all_getters(self) -> None: """Spin 16 threads registering distinct domains; every entry survives.""" N = 16 barrier = threading.Barrier(N) def _worker(idx: int) -> None: barrier.wait() register_attestation_provider(f"dom-{idx}", _StubProvider(f"p{idx}")) threads = [ threading.Thread(target=_worker, args=(i,)) for i in range(N) ] for t in threads: t.start() for t in threads: t.join() for i in range(N): provider = get_attestation_provider(f"dom-{i}") assert provider is not None assert provider.provider_id == f"p{i}" # type: ignore[attr-defined] def test_concurrent_get_during_register_does_not_crash(self) -> None: """Hammer get() while register() is running — must never raise.""" stop = threading.Event() errors: list[BaseException] = [] def _registrar() -> None: for i in range(200): try: register_attestation_provider("hot", _StubProvider(f"r{i}")) except Exception as e: # pragma: no cover — only fires on bug errors.append(e) def _reader() -> None: while not stop.is_set(): try: get_attestation_provider("hot") except Exception as e: # pragma: no cover — only fires on bug errors.append(e) readers = [threading.Thread(target=_reader) for _ in range(4)] for r in readers: r.start() registrar = threading.Thread(target=_registrar) registrar.start() registrar.join() stop.set() for r in readers: r.join() assert not errors # =========================================================================== # INTEGRATION TIER # =========================================================================== class TestRoundTrip: """Stub provider → register → compute → anchor → verify.""" def test_full_round_trip_returns_valid(self) -> None: """Happy path: registered provider produces a valid result end to end.""" provider = _StubProvider("rt") register_attestation_provider("rtdom", provider) looked_up = get_attestation_provider("rtdom") assert looked_up is provider record = looked_up.compute_attestation( "snapshot-abc", {"path": "notes/a.md", "action": "write", "content_hash": "deadbeef"}, ) assert record["id"] == "snapshot-abc:notes/a.md" assert record["provider_id"] == "rt" receipt = looked_up.anchor(record) assert receipt["tx_id"] assert receipt["anchor_url"].startswith("https://") result = looked_up.verify(record, receipt) assert result["valid"] is True assert result["reason"] == "" assert result["depth"] == 0 def test_compute_is_deterministic(self) -> None: """Same (snapshot_id, commit_meta) → same record id every time.""" provider = _StubProvider() meta = {"path": "x.md", "action": "write", "content_hash": "abc"} a = provider.compute_attestation("s1", meta) b = provider.compute_attestation("s1", meta) assert a == b # =========================================================================== # SECURITY TIER (mandatory per Phase 7 spec) # =========================================================================== class TestSecurity: """Hardening contracts that must hold against misuse and adversarial input.""" def test_partial_verification_is_a_hard_error(self) -> None: """verify() with no anchored record raises, never returns valid=True.""" provider = _StubProvider() record = provider.compute_attestation( "s", {"path": "p", "content_hash": "c"} ) bogus_receipt: AnchorReceipt = { "tx_id": "fake", "anchor_url": "https://fake.example/x", "anchored_at": "2026-05-13T00:00:00Z", "provider_id": "stub", } with pytest.raises(AttestationVerifyError) as exc_info: provider.verify(record, bogus_receipt) assert exc_info.value.code == "NO_RECEIPT" def test_known_bad_record_returns_valid_false_not_raises(self) -> None: """A *provably-bad* record returns valid=False — distinct from incomplete.""" provider = _StubProvider() record = provider.compute_attestation( "s", {"path": "p", "content_hash": "c"} ) receipt = provider.anchor(record) tampered_receipt: AnchorReceipt = dict(receipt) # type: ignore[assignment] tampered_receipt["tx_id"] = "tampered-tx" result = provider.verify(record, tampered_receipt) assert result["valid"] is False assert result["reason"] def test_anchor_is_write_once(self) -> None: """Re-anchoring the same record raises ImmutableReceiptError.""" provider = _StubProvider() record = provider.compute_attestation( "s", {"path": "p", "content_hash": "c"} ) provider.anchor(record) with pytest.raises(ImmutableReceiptError) as exc_info: provider.anchor(record) assert exc_info.value.code == "IMMUTABLE_RECEIPT" def test_registry_isolation_between_domains(self) -> None: """Provider for domain A is never returned for domain B.""" provider_a = _StubProvider("A") register_attestation_provider("dom-a", provider_a) assert get_attestation_provider("dom-b") is None assert get_attestation_provider("dom-A") is None # case-sensitive def test_verify_provider_id_mismatch_is_hard_error(self) -> None: """Cross-provider receipts trigger AttestationVerifyError, not False.""" provider = _StubProvider("legit") record = provider.compute_attestation( "s", {"path": "p", "content_hash": "c"} ) spoofed_receipt: AnchorReceipt = { "tx_id": "x", "anchor_url": "https://x.example/x", "anchored_at": "2026-05-13T00:00:00Z", "provider_id": "evil", } with pytest.raises(AttestationVerifyError): provider.verify(record, spoofed_receipt) def test_verify_error_codes_are_stable_strings(self) -> None: """Codes are stable enough for CLI/MCP layers to map without regex.""" try: raise AttestationVerifyError("x", code="ICP_UNREACHABLE") except AttestationVerifyError as e: assert isinstance(e.code, str) assert e.code.isupper() and e.code.replace("_", "").isalnum()