"""Tests for iter_wire_frames — server-side async exact-byte frame reader. Test plan: A. Roundtrip — write with MuseWireFrameWriter, read with iter_wire_frames B. Truncated header — clean WireFrameError, not msgpack garbage C. Truncated payload — clean WireFrameError D. Hash mismatch — deterministic WireFrameError E. Size mismatch — envelope sz vs binary length prefix F. Envelope/logical mismatch — verified by caller, not reader (reader yields both; dispatcher checks ft == frame["t"]) G. Multiple frames — reader yields them all in order H. Empty stream — reader returns without error """ from __future__ import annotations import struct from collections.abc import AsyncIterator import msgpack import pytest from muse.core.types import blob_id from musehub.types.json_types import JSONObject, JSONValue # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _pack(obj: JSONValue) -> bytes: return msgpack.packb(obj, use_bin_type=True) def _make_frame(ft: str, payload: bytes) -> bytes: """Build one wire frame using MuseWireFrameWriter.""" from muse.core.mpack import MuseWireFrameWriter return MuseWireFrameWriter().wrap(frame_type=ft, payload=payload) async def _body(*chunks: bytes) -> None: for c in chunks: yield c async def _collect(frames_iter: AsyncIterator[tuple[JSONObject, bytes]]) -> list[tuple[JSONObject, bytes]]: result = [] async for header, payload in frames_iter: result.append((header, payload)) return result # --------------------------------------------------------------------------- # A — Roundtrip # --------------------------------------------------------------------------- class TestRoundtrip: """A. Frames produced by MuseWireFrameWriter are read correctly.""" @pytest.mark.asyncio async def test_import(self) -> None: from musehub.services.musehub_wire import iter_wire_frames # noqa: F401 @pytest.mark.asyncio async def test_single_header_frame(self) -> None: from musehub.services.musehub_wire import iter_wire_frames payload = _pack({"t": "H", "branch": "main", "n_objects": 0}) data = _make_frame("H", payload) frames = await _collect(iter_wire_frames(_body(data))) assert len(frames) == 1 header, decoded_payload = frames[0] assert header["ft"] == "H" assert decoded_payload == payload @pytest.mark.asyncio async def test_multiple_frames_in_sequence(self) -> None: from musehub.services.musehub_wire import iter_wire_frames h_payload = _pack({"t": "H", "branch": "main", "n_objects": 0}) c_payload = _pack({"t": "C", "commits": [], "snapshots": []}) e_payload = _pack({"t": "E", "n_objects": 0, "n_commits": 0}) data = ( _make_frame("H", h_payload) + _make_frame("C", c_payload) + _make_frame("E", e_payload) ) frames = await _collect(iter_wire_frames(_body(data))) assert len(frames) == 3 assert frames[0][0]["ft"] == "H" assert frames[1][0]["ft"] == "C" assert frames[2][0]["ft"] == "E" @pytest.mark.asyncio async def test_frames_split_across_chunks(self) -> None: """Reader must handle chunk boundaries that fall mid-frame.""" from musehub.services.musehub_wire import iter_wire_frames payload = _pack({"t": "H", "branch": "main"}) data = _make_frame("H", payload) # Split into 3-byte chunks chunks = [data[i:i + 3] for i in range(0, len(data), 3)] frames = await _collect(iter_wire_frames(_body(*chunks))) assert len(frames) == 1 assert frames[0][0]["ft"] == "H" assert frames[0][1] == payload @pytest.mark.asyncio async def test_payload_parses_as_msgpack(self) -> None: from musehub.services.musehub_wire import iter_wire_frames payload = _pack({"t": "O", "id": blob_id(b"x"), "content": b"x", "enc": "raw"}) data = _make_frame("O", payload) frames = await _collect(iter_wire_frames(_body(data))) assert len(frames) == 1 decoded = msgpack.unpackb(frames[0][1], raw=False) assert decoded["t"] == "O" @pytest.mark.asyncio async def test_header_id_verified(self) -> None: """Reader verifies blob_id(payload) == header['id'].""" from musehub.services.musehub_wire import iter_wire_frames from muse.core.types import blob_id payload = _pack({"t": "E"}) data = _make_frame("E", payload) frames = await _collect(iter_wire_frames(_body(data))) assert blob_id(frames[0][1]) == frames[0][0]["id"] # --------------------------------------------------------------------------- # H — Empty stream # --------------------------------------------------------------------------- class TestEmptyStream: """H. Reader returns immediately on empty body without error.""" @pytest.mark.asyncio async def test_empty_body_yields_nothing(self) -> None: from musehub.services.musehub_wire import iter_wire_frames frames = await _collect(iter_wire_frames(_body())) assert frames == [] @pytest.mark.asyncio async def test_empty_chunk_yields_nothing(self) -> None: from musehub.services.musehub_wire import iter_wire_frames frames = await _collect(iter_wire_frames(_body(b""))) assert frames == [] # --------------------------------------------------------------------------- # B — Truncated header # --------------------------------------------------------------------------- class TestTruncatedHeader: """B. Truncation within the envelope header raises WireFrameError.""" @pytest.mark.asyncio async def test_truncated_at_version_byte(self) -> None: from musehub.services.musehub_wire import iter_wire_frames from muse.core.mpack import WireFrameError payload = _pack({"t": "H"}) data = _make_frame("H", payload) truncated = data[:5] # magic + version only, no header_len with pytest.raises(WireFrameError): await _collect(iter_wire_frames(_body(truncated))) @pytest.mark.asyncio async def test_truncated_mid_header_bytes(self) -> None: from musehub.services.musehub_wire import iter_wire_frames from muse.core.mpack import WireFrameError payload = _pack({"t": "H"}) data = _make_frame("H", payload) # Cut mid-header: after magic(4) + version(1) + header_len(4) + a few header bytes truncated = data[:12] with pytest.raises(WireFrameError): await _collect(iter_wire_frames(_body(truncated))) @pytest.mark.asyncio async def test_invalid_magic_raises(self) -> None: from musehub.services.musehub_wire import iter_wire_frames from muse.core.mpack import WireFrameError payload = _pack({"t": "H"}) data = _make_frame("H", payload) # Corrupt the magic bad = b"XXXX" + data[4:] with pytest.raises(WireFrameError, match="magic"): await _collect(iter_wire_frames(_body(bad))) @pytest.mark.asyncio async def test_wrong_version_raises(self) -> None: from musehub.services.musehub_wire import iter_wire_frames from muse.core.mpack import WireFrameError payload = _pack({"t": "H"}) data = _make_frame("H", payload) # Replace version byte (index 4) with 0x02 (unsupported) bad = data[:4] + bytes([2]) + data[5:] with pytest.raises(WireFrameError, match="version"): await _collect(iter_wire_frames(_body(bad))) # --------------------------------------------------------------------------- # C — Truncated payload # --------------------------------------------------------------------------- class TestTruncatedPayload: """C. Truncation mid-payload raises WireFrameError, not msgpack garbage.""" @pytest.mark.asyncio async def test_truncated_payload_raises(self) -> None: from musehub.services.musehub_wire import iter_wire_frames from muse.core.mpack import WireFrameError payload = _pack({"t": "C", "commits": list(range(50)), "snapshots": []}) data = _make_frame("C", payload) # Trim the last 100 bytes of payload truncated = data[:-100] with pytest.raises(WireFrameError): await _collect(iter_wire_frames(_body(truncated))) @pytest.mark.asyncio async def test_truncation_raises_wire_error_not_msgpack_error(self) -> None: """Truncation must raise WireFrameError, not BufferError or msgpack exceptions.""" from musehub.services.musehub_wire import iter_wire_frames from muse.core.mpack import WireFrameError payload = _pack({"t": "C", "commits": list(range(20)), "snapshots": []}) data = _make_frame("C", payload) for cut in [len(data) - 1, len(data) - 50, len(data) - 200]: if cut <= 0: continue with pytest.raises(WireFrameError): await _collect(iter_wire_frames(_body(data[:cut]))) # --------------------------------------------------------------------------- # D — Hash mismatch # --------------------------------------------------------------------------- class TestHashMismatch: """D. Tampered payload raises WireFrameError with hash mismatch message.""" @pytest.mark.asyncio async def test_tampered_payload_raises(self) -> None: from musehub.services.musehub_wire import iter_wire_frames from muse.core.mpack import WireFrameError payload = _pack({"t": "C", "commits": [], "snapshots": []}) data = _make_frame("C", payload) # Flip the last byte of the payload tampered = data[:-1] + bytes([data[-1] ^ 0xFF]) with pytest.raises(WireFrameError, match="hash"): await _collect(iter_wire_frames(_body(tampered))) # --------------------------------------------------------------------------- # E — Size mismatch # --------------------------------------------------------------------------- class TestSizeMismatch: """E. Mismatched envelope sz vs binary payload_len raises WireFrameError.""" @pytest.mark.asyncio async def test_size_mismatch_raises(self) -> None: from musehub.services.musehub_wire import iter_wire_frames from muse.core.mpack import WireFrameError from muse.core.types import blob_id payload = _pack({"t": "H"}) # Build frame with envelope sz = len(payload) + 50 but actual payload is correct bad_header = {"ft": "H", "id": blob_id(payload), "sz": len(payload) + 50} bad_header_bytes = msgpack.packb(bad_header, use_bin_type=True) tampered = b"".join([ b"muse", bytes([1]), struct.pack(">I", len(bad_header_bytes)), bad_header_bytes, struct.pack(">Q", len(payload)), payload, ]) with pytest.raises(WireFrameError, match="size"): await _collect(iter_wire_frames(_body(tampered)))