"""TDD — Phase 14B: OC (Object Chunk) frame protocol. Large objects (> OC_CHUNK_SIZE) are split into OC frames so they can be streamed through MWP without hitting proxy body limits. Rules encoded here: P14B-0 OC_CHUNK_SIZE exported from muse.core.transport, equals 290 * 1024. P14B-1 Object of exactly OC_CHUNK_SIZE bytes → 1 plain O-frame (no chunking). P14B-2 Object of OC_CHUNK_SIZE + 1 bytes → 2 OC-frames; each ≤ OC_CHUNK_SIZE. P14B-3 Object of 3 × OC_CHUNK_SIZE bytes → 3 OC-frames; bytes sum to original. P14B-4 chunk_index=0 has `encoding`, `path`, `sz`; subsequent chunks do not. P14B-5 `split_object_into_oc_frames` is exported from muse.core.transport and returns a list of frame dicts (not ObjectPayload — raw dicts for wire). P14B-6 All returned frames have t="OC", object_id, chunk_index, total_chunks, content. Last frame has chunk_index == total_chunks - 1. """ from __future__ import annotations import math from typing import Any import pytest # --------------------------------------------------------------------------- # P14B-0 — OC_CHUNK_SIZE exported from muse.core.transport # --------------------------------------------------------------------------- class TestP14B0ChunkSizeExported: def test_oc_chunk_size_is_positive_int(self) -> None: from muse.core.transport import OC_CHUNK_SIZE # type: ignore[attr-defined] assert isinstance(OC_CHUNK_SIZE, int) assert OC_CHUNK_SIZE > 0 def test_oc_chunk_size_equals_290kb(self) -> None: from muse.core.transport import OC_CHUNK_SIZE # type: ignore[attr-defined] assert OC_CHUNK_SIZE == 290 * 1024 # --------------------------------------------------------------------------- # P14B-1 — Object exactly OC_CHUNK_SIZE → 1 plain O-frame (no chunking) # --------------------------------------------------------------------------- class TestP14B1ExactSizeNoChunking: def test_object_at_chunk_size_is_single_o_frame(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] obj = { "object_id": "sha256:" + "b" * 64, "content": bytes(OC_CHUNK_SIZE), "encoding": "raw", "path": "big.bin", } frames = split_object_into_oc_frames(obj) # Exactly one frame, and it's a regular O-frame assert len(frames) == 1 assert frames[0]["t"] == "O" assert frames[0]["content"] == obj["content"] def test_object_below_chunk_size_is_single_o_frame(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] obj = { "object_id": "sha256:" + "c" * 64, "content": bytes(OC_CHUNK_SIZE - 1), "encoding": "raw", "path": "medium.bin", } frames = split_object_into_oc_frames(obj) assert len(frames) == 1 assert frames[0]["t"] == "O" # --------------------------------------------------------------------------- # P14B-2 — Object of OC_CHUNK_SIZE + 1 bytes → 2 OC-frames # --------------------------------------------------------------------------- class TestP14B2TwoChunks: def _get_frames(self, size: int) -> list[dict]: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] pattern = bytes(range(256)) content = (pattern * ((size // 256) + 1))[:size] obj = { "object_id": "sha256:" + "d" * 64, "content": content, "encoding": "raw", "path": "large.bin", } return split_object_into_oc_frames(obj) def test_one_byte_over_chunk_size_produces_two_frames(self) -> None: from muse.core.transport import OC_CHUNK_SIZE # type: ignore[attr-defined] frames = self._get_frames(OC_CHUNK_SIZE + 1) assert len(frames) == 2 def test_all_frames_are_oc_type(self) -> None: from muse.core.transport import OC_CHUNK_SIZE # type: ignore[attr-defined] frames = self._get_frames(OC_CHUNK_SIZE + 1) for f in frames: assert f["t"] == "OC" def test_each_frame_content_le_chunk_size(self) -> None: from muse.core.transport import OC_CHUNK_SIZE # type: ignore[attr-defined] frames = self._get_frames(OC_CHUNK_SIZE + 1) for f in frames: assert len(f["content"]) <= OC_CHUNK_SIZE def test_frame_content_sums_to_original(self) -> None: from muse.core.transport import OC_CHUNK_SIZE # type: ignore[attr-defined] size = OC_CHUNK_SIZE + 1 frames = self._get_frames(size) reassembled = b"".join(f["content"] for f in frames) assert len(reassembled) == size # --------------------------------------------------------------------------- # P14B-3 — Object of 3 × OC_CHUNK_SIZE bytes → 3 OC-frames # --------------------------------------------------------------------------- class TestP14B3ThreeChunks: def test_three_chunk_size_object_produces_three_frames(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] size = OC_CHUNK_SIZE * 3 content = bytes([i % 256 for i in range(size)]) obj = { "object_id": "sha256:" + "e" * 64, "content": content, "encoding": "raw", "path": "triple.bin", } frames = split_object_into_oc_frames(obj) assert len(frames) == 3 def test_three_chunks_reassemble_correctly(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] size = OC_CHUNK_SIZE * 3 content = bytes([i % 256 for i in range(size)]) obj = { "object_id": "sha256:" + "f" * 64, "content": content, "encoding": "raw", "path": "triple.bin", } frames = split_object_into_oc_frames(obj) reassembled = b"".join(f["content"] for f in frames) assert reassembled == content def test_total_chunks_field_is_correct(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] size = OC_CHUNK_SIZE * 3 content = bytes(size) obj = { "object_id": "sha256:" + "e" * 64, "content": content, "encoding": "raw", "path": "triple.bin", } frames = split_object_into_oc_frames(obj) for f in frames: assert f["total_chunks"] == 3 def test_chunk_indices_are_sequential(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] size = OC_CHUNK_SIZE * 3 content = bytes(size) obj = { "object_id": "sha256:" + "e" * 64, "content": content, "encoding": "raw", "path": "triple.bin", } frames = split_object_into_oc_frames(obj) indices = [f["chunk_index"] for f in frames] assert indices == [0, 1, 2] # --------------------------------------------------------------------------- # P14B-4 — Only chunk_index=0 carries encoding, path, sz # --------------------------------------------------------------------------- class TestP14B4MetadataOnFirstChunkOnly: def _frames_for_large_object(self) -> list[dict]: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] size = OC_CHUNK_SIZE * 2 + 1 content = bytes(size) obj = { "object_id": "sha256:" + "a" * 64, "content": content, "encoding": "raw", "path": "large.bin", } return split_object_into_oc_frames(obj) def test_first_chunk_has_encoding(self) -> None: frames = self._frames_for_large_object() assert "encoding" in frames[0] def test_first_chunk_has_path(self) -> None: frames = self._frames_for_large_object() assert "path" in frames[0] def test_first_chunk_has_sz(self) -> None: frames = self._frames_for_large_object() assert "sz" in frames[0] from muse.core.transport import OC_CHUNK_SIZE # type: ignore[attr-defined] assert frames[0]["sz"] == OC_CHUNK_SIZE * 2 + 1 def test_subsequent_chunks_have_no_encoding(self) -> None: frames = self._frames_for_large_object() for f in frames[1:]: assert "encoding" not in f def test_subsequent_chunks_have_no_path(self) -> None: frames = self._frames_for_large_object() for f in frames[1:]: assert "path" not in f def test_subsequent_chunks_have_no_sz(self) -> None: frames = self._frames_for_large_object() for f in frames[1:]: assert "sz" not in f # --------------------------------------------------------------------------- # P14B-5 — split_object_into_oc_frames exported, returns list of dicts # --------------------------------------------------------------------------- class TestP14B5FunctionExported: def test_function_is_callable(self) -> None: from muse.core.transport import split_object_into_oc_frames # type: ignore[attr-defined] assert callable(split_object_into_oc_frames) def test_returns_list_of_dicts(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] obj = { "object_id": "sha256:" + "a" * 64, "content": bytes(10), "encoding": "raw", "path": "tiny.bin", } result = split_object_into_oc_frames(obj) assert isinstance(result, list) for item in result: assert isinstance(item, dict) # --------------------------------------------------------------------------- # P14B-6 — All OC frames have required fields; last frame has correct index # --------------------------------------------------------------------------- class TestP14B6FrameFields: def test_all_oc_frames_have_required_fields(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] size = OC_CHUNK_SIZE * 2 + 512 content = bytes(size) obj = { "object_id": "sha256:" + "b" * 64, "content": content, "encoding": "raw", "path": "multi.bin", } frames = split_object_into_oc_frames(obj) for f in frames: assert "t" in f assert "object_id" in f assert "chunk_index" in f assert "total_chunks" in f assert "content" in f assert f["object_id"] == obj["object_id"] def test_last_frame_chunk_index_equals_total_minus_one(self) -> None: from muse.core.transport import OC_CHUNK_SIZE, split_object_into_oc_frames # type: ignore[attr-defined] size = OC_CHUNK_SIZE * 4 content = bytes(size) obj = { "object_id": "sha256:" + "c" * 64, "content": content, "encoding": "raw", "path": "quad.bin", } frames = split_object_into_oc_frames(obj) last = frames[-1] assert last["chunk_index"] == last["total_chunks"] - 1