"""Tests for MPack frame TypedDicts and protocol constants. Coverage tiers -------------- - Constants: MPACK_VERSION, MPACK_CONTENT_TYPE, all frame tag values - Frame TypedDicts: required fields, type annotations, total/partial flags - Frame tag registry: every tag maps to a TypedDict - Canonical signature payload: deterministic bytes, field coverage - Forward-compatibility: unknown frame tags are single ASCII characters """ from __future__ import annotations import msgpack from typing import get_type_hints # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- class TestConstants: def test_mpack_version(self) -> None: from muse.core.mpack import MPACK_VERSION assert MPACK_VERSION == "mpack/1.0" def test_mpack_content_type(self) -> None: from muse.core.mpack import MPACK_CONTENT_TYPE assert MPACK_CONTENT_TYPE == "application/x-muse-mpack" def test_frame_tags_are_single_uppercase_ascii(self) -> None: from muse.core.mpack import ( FRAME_HEADER, FRAME_OBJECT, FRAME_COMMIT_PACK, FRAME_END, FRAME_PROGRESS, FRAME_ERROR, FRAME_RESULT, FRAME_ACK, ) for tag in (FRAME_HEADER, FRAME_OBJECT, FRAME_COMMIT_PACK, FRAME_END, FRAME_PROGRESS, FRAME_ERROR, FRAME_RESULT, FRAME_ACK): assert isinstance(tag, str) and len(tag) == 1 and tag.isupper(), ( f"Frame tag {tag!r} must be a single uppercase ASCII character" ) def test_frame_tags_are_distinct(self) -> None: from muse.core.mpack import ( FRAME_HEADER, FRAME_OBJECT, FRAME_COMMIT_PACK, FRAME_END, FRAME_PROGRESS, FRAME_ERROR, FRAME_RESULT, FRAME_ACK, ) tags = [FRAME_HEADER, FRAME_OBJECT, FRAME_COMMIT_PACK, FRAME_END, FRAME_PROGRESS, FRAME_ERROR, FRAME_RESULT, FRAME_ACK] assert len(set(tags)) == len(tags), "All frame tags must be unique" def test_frame_header_tag_is_H(self) -> None: from muse.core.mpack import FRAME_HEADER assert FRAME_HEADER == "H" def test_frame_object_tag_is_O(self) -> None: from muse.core.mpack import FRAME_OBJECT assert FRAME_OBJECT == "O" def test_frame_commit_pack_tag_is_C(self) -> None: from muse.core.mpack import FRAME_COMMIT_PACK assert FRAME_COMMIT_PACK == "C" def test_frame_end_tag_is_E(self) -> None: from muse.core.mpack import FRAME_END assert FRAME_END == "E" def test_frame_progress_tag_is_P(self) -> None: from muse.core.mpack import FRAME_PROGRESS assert FRAME_PROGRESS == "P" def test_frame_error_tag_is_X(self) -> None: from muse.core.mpack import FRAME_ERROR assert FRAME_ERROR == "X" def test_frame_result_tag_is_R(self) -> None: from muse.core.mpack import FRAME_RESULT assert FRAME_RESULT == "R" def test_frame_ack_tag_is_A(self) -> None: from muse.core.mpack import FRAME_ACK assert FRAME_ACK == "A" # --------------------------------------------------------------------------- # Frame TypedDicts — field annotations # --------------------------------------------------------------------------- class TestMPackHeaderFrame: def test_exists(self) -> None: from muse.core.mpack import MPackHeaderFrame assert MPackHeaderFrame is not None def test_has_t_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "t" in hints def test_has_version_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "v" in hints def test_has_op_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "op" in hints def test_has_branch_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "branch" in hints def test_has_n_objects_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "n_objects" in hints def test_has_n_commits_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "n_commits" in hints def test_has_objects_bytes_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "objects_bytes" in hints def test_has_agent_id_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "agent_id" in hints def test_has_model_id_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "model_id" in hints def test_has_signer_key_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "signer_key" in hints def test_has_signature_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "signature" in hints def test_has_domains_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "domains" in hints def test_has_have_field(self) -> None: from muse.core.mpack import MPackHeaderFrame hints = get_type_hints(MPackHeaderFrame) assert "have" in hints class TestMPackObjectFrame: def test_exists(self) -> None: from muse.core.mpack import MPackObjectFrame assert MPackObjectFrame is not None def test_has_t_field(self) -> None: from muse.core.mpack import MPackObjectFrame hints = get_type_hints(MPackObjectFrame) assert "t" in hints def test_has_id_field(self) -> None: from muse.core.mpack import MPackObjectFrame hints = get_type_hints(MPackObjectFrame) assert "id" in hints def test_has_content_field(self) -> None: from muse.core.mpack import MPackObjectFrame hints = get_type_hints(MPackObjectFrame) assert "content" in hints def test_has_enc_field(self) -> None: from muse.core.mpack import MPackObjectFrame hints = get_type_hints(MPackObjectFrame) assert "enc" in hints def test_has_sz_field(self) -> None: """sz = uncompressed byte size — receiver budgets before decode.""" from muse.core.mpack import MPackObjectFrame hints = get_type_hints(MPackObjectFrame) assert "sz" in hints def test_has_dom_field(self) -> None: """dom = domain: code | midi | audio | model-weights | ...""" from muse.core.mpack import MPackObjectFrame hints = get_type_hints(MPackObjectFrame) assert "dom" in hints def test_has_path_field(self) -> None: from muse.core.mpack import MPackObjectFrame hints = get_type_hints(MPackObjectFrame) assert "path" in hints class TestMPackCommitPackFrame: def test_exists(self) -> None: from muse.core.mpack import MPackCommitPackFrame assert MPackCommitPackFrame is not None def test_has_required_fields(self) -> None: from muse.core.mpack import MPackCommitPackFrame hints = get_type_hints(MPackCommitPackFrame) for f in ("t", "commits", "snapshots"): assert f in hints, f"MPackCommitPackFrame missing field {f!r}" class TestMPackEndFrame: def test_exists(self) -> None: from muse.core.mpack import MPackEndFrame assert MPackEndFrame is not None def test_has_n_objects_for_integrity_check(self) -> None: from muse.core.mpack import MPackEndFrame hints = get_type_hints(MPackEndFrame) assert "n_objects" in hints def test_has_n_commits_for_integrity_check(self) -> None: from muse.core.mpack import MPackEndFrame hints = get_type_hints(MPackEndFrame) assert "n_commits" in hints class TestMPackProgressFrame: def test_has_pct_field(self) -> None: """pct enables real progress bars (0.0–100.0), not just text.""" from muse.core.mpack import MPackProgressFrame hints = get_type_hints(MPackProgressFrame) assert "pct" in hints def test_has_msg_field(self) -> None: from muse.core.mpack import MPackProgressFrame hints = get_type_hints(MPackProgressFrame) assert "msg" in hints class TestMPackErrorFrame: def test_has_msg_and_code(self) -> None: from muse.core.mpack import MPackErrorFrame hints = get_type_hints(MPackErrorFrame) assert "msg" in hints assert "code" in hints class TestMPackResultFrame: def test_has_ok_msg_heads_head(self) -> None: from muse.core.mpack import MPackResultFrame hints = get_type_hints(MPackResultFrame) for f in ("ok", "msg", "heads", "head"): assert f in hints, f"MPackResultFrame missing field {f!r}" class TestMPackAckFrame: def test_has_ack_common_ready(self) -> None: from muse.core.mpack import MPackAckFrame hints = get_type_hints(MPackAckFrame) for f in ("ack", "common", "ready"): assert f in hints, f"MPackAckFrame missing field {f!r}" # --------------------------------------------------------------------------- # Canonical signature payload # --------------------------------------------------------------------------- class TestCanonicalHeaderPayload: def test_returns_bytes(self) -> None: from muse.core.mpack import _canonical_header_payload result = _canonical_header_payload( op="push", branch="main", n_objects=10, n_commits=5, agent_id="claude-code", model_id="claude-sonnet-4-6", ) assert isinstance(result, bytes) def test_is_valid_msgpack(self) -> None: from muse.core.mpack import _canonical_header_payload payload = _canonical_header_payload( op="push", branch="main", n_objects=10, n_commits=5, agent_id="claude-code", model_id="claude-sonnet-4-6", ) decoded = msgpack.unpackb(payload, raw=False) assert isinstance(decoded, dict) def test_contains_version(self) -> None: from muse.core.mpack import _canonical_header_payload, MPACK_VERSION payload = _canonical_header_payload( op="push", branch="main", n_objects=10, n_commits=5, agent_id="", model_id="", ) decoded = msgpack.unpackb(payload, raw=False) assert decoded["v"] == MPACK_VERSION def test_covers_all_signing_fields(self) -> None: from muse.core.mpack import _canonical_header_payload payload = _canonical_header_payload( op="push", branch="dev", n_objects=3, n_commits=1, agent_id="claude-code", model_id="claude-sonnet-4-6", ) decoded = msgpack.unpackb(payload, raw=False) for field in ("v", "op", "branch", "n_objects", "n_commits", "agent_id", "model_id"): assert field in decoded, f"Canonical payload missing field {field!r}" def test_deterministic_across_calls(self) -> None: from muse.core.mpack import _canonical_header_payload kwargs = dict(op="push", branch="main", n_objects=5, n_commits=2, agent_id="claude-code", model_id="claude-sonnet-4-6") a = _canonical_header_payload(**kwargs) b = _canonical_header_payload(**kwargs) assert a == b, "Canonical payload must be deterministic" def test_different_inputs_produce_different_payloads(self) -> None: from muse.core.mpack import _canonical_header_payload a = _canonical_header_payload( op="push", branch="main", n_objects=5, n_commits=2, agent_id="claude-code", model_id="claude-sonnet-4-6", ) b = _canonical_header_payload( op="fetch", branch="dev", n_objects=10, n_commits=3, agent_id="agentception", model_id="claude-opus-4-6", ) assert a != b # --------------------------------------------------------------------------- # Docstring # --------------------------------------------------------------------------- class TestDocstring: def _doc(self) -> str: import muse.core.mpack as mod return mod.__doc__ or "" def test_documents_frame_sequence(self) -> None: assert "H" in self._doc() and "O" in self._doc() and "C" in self._doc() def test_documents_mpack_version(self) -> None: assert "mpack/1.0" in self._doc() def test_documents_content_type(self) -> None: assert "application/x-muse-mpack" in self._doc() def test_documents_provenance(self) -> None: assert "agent_id" in self._doc() or "provenance" in self._doc().lower() def test_documents_compression(self) -> None: assert "zstd" in self._doc() or "zlib" in self._doc()