"""TDD: object encoding integrity — push → storage → blob read roundtrip. Failure mode being hunted: Objects pushed via wire_push_object_pack arrive zlib-encoded (encoding="zlib"). If the server fails to decode before storing, the blob view serves compressed binary gibberish instead of the original source text. All object IDs use the canonical sha256: prefix — bare hex is rejected at the Pydantic boundary. Coverage matrix (each test logs exactly what was stored and what was served): 1. zlib + sha256: ID → storage → assert plain bytes 2. zlib + sha256: ID → blob view endpoint → assert readable text 3. delta+zlib + sha256: ID → storage → assert reconstructed plain bytes 4. encoding=None + sha256: ID → storage → assert NOT compressed (regression guard) All tests print a structured log block so a CI failure shows the real bytes, not just an assertion message. """ from __future__ import annotations import hashlib import logging import struct import zlib import msgpack import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from musehub.db import musehub_models as db from musehub.types.compression import decompress_if_needed from musehub.types.json_types import JSONValue, StrDict from tests.factories import create_repo as factory_create_repo logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Helpers — mirror exact client behaviour # --------------------------------------------------------------------------- def _sha256_oid(raw: bytes) -> str: """Canonical object ID: sha256:.""" return "sha256:" + hashlib.sha256(raw).hexdigest() def _mp(data: JSONValue) -> bytes: return msgpack.packb(data, use_bin_type=True) def _zlib_compress(raw: bytes) -> bytes: """Tier-1 encoding: plain zlib, level 1 (fast), matching muse CLI compress_zlib.""" return zlib.compress(raw, level=1) def _compute_delta(base: bytes, target: bytes) -> bytes: """Build zlib-compressed delta instruction stream (mirrors muse/core/compression.py).""" # Simple delta: chunk_size = 32 bytes CHUNK = 32 table: dict[bytes, int] = {} for i in range(0, len(base) - CHUNK + 1, CHUNK): table[base[i: i + CHUNK]] = i result = bytearray() pos = 0 while pos < len(target): chunk = target[pos: pos + CHUNK] if chunk in table: offset = table[chunk] length = CHUNK # Extend match while (pos + length < len(target) and offset + length < len(base) and target[pos + length] == base[offset + length]): length += 1 result += b"\x00" + struct.pack(">II", offset, length) pos += length else: literal = target[pos: pos + 1] result += b"\x01" + struct.pack(">I", 1) + literal pos += 1 return zlib.compress(bytes(result), level=1) def _log_bytes(label: str, data: bytes | None, n: int = 120) -> None: """Emit a structured log block for test forensics.""" if data is None: logger.warning("[ENCODING-TEST] %s: None", label) print(f"\n [ENCODING-TEST] {label}: None") return snippet = data[:n] is_text = all(0x09 <= b <= 0x7E or b in (0x0A, 0x0D) for b in snippet) magic = data[:2].hex() if len(data) >= 2 else "??" print( f"\n [ENCODING-TEST] {label}:" f"\n len={len(data)} magic_bytes=0x{magic} readable={is_text}" f"\n first {n}B: {snippet!r}" ) logger.info( "[ENCODING-TEST] %s: len=%d magic=0x%s readable=%s first=%r", label, len(data), magic, is_text, snippet, ) # --------------------------------------------------------------------------- # Test 1: zlib-encoded object with bare hex ID — storage must hold plain bytes # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_zlib_stored_as_plain_bytes( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Push a zlib-encoded object; storage must contain plain bytes. This is the most critical invariant: the server must decode before writing. If this fails, every blob in the repo is gibberish. """ import musehub.services.musehub_wire as _wire_svc repo = await factory_create_repo( db_session, slug="encoding-rt-zlib-storage", owner="test-user-wire" ) raw = b"# muse-zsh\n\nOh My ZSH plugin for Muse version control.\n" * 20 oid = _sha256_oid(raw) compressed = _zlib_compress(raw) print(f"\n [ENCODING-TEST] input: len={len(raw)} oid={oid[:23]}…") print(f" [ENCODING-TEST] compressed: len={len(compressed)} magic=0x{compressed[:2].hex()}") r = await client.post( f"/{repo.owner}/{repo.slug}/push/object-pack", content=_mp({"objects": [{ "object_id": oid, "content": compressed, "path": "README.md", "encoding": "zlib", }]}), headers=wire_headers, ) print(f" [ENCODING-TEST] push response: status={r.status_code} body={r.content[:200]!r}") assert r.status_code == 200, r.text resp_data = msgpack.unpackb(r.content, raw=False) print(f" [ENCODING-TEST] push result: stored={resp_data.get('stored')} skipped={resp_data.get('skipped')}") backend = _wire_svc.get_backend() stored = await backend.get(oid) _log_bytes("stored bytes from backend", stored) # Also check content_cache in DB obj_row = (await db_session.get(db.MusehubObject, oid)) if obj_row: cache = obj_row.content_cache _log_bytes("content_cache in DB", cache) print(f" [ENCODING-TEST] storage_uri={obj_row.storage_uri!r}") assert stored is not None or (obj_row and obj_row.content_cache is not None), \ "object must be findable in storage or content_cache" actual = stored if stored is not None else obj_row.content_cache # type: ignore[union-attr] _log_bytes("bytes that will be served", actual) assert actual == raw, ( f"\nSTORED BYTES DO NOT MATCH ORIGINAL — encoding bug confirmed.\n" f" original first 80B: {raw[:80]!r}\n" f" stored first 80B: {actual[:80]!r}\n" f" stored is zlib: {actual[:2].hex() in ('7801','789c','78da','785e')}" ) # --------------------------------------------------------------------------- # Test 2: zlib-encoded object — blob view must serve readable text # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_zlib_blob_view_serves_plain_text( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """After pushing with zlib encoding, fetch/objects must return decompressed plain text.""" import musehub.services.musehub_wire as _wire_svc repo = await factory_create_repo( db_session, slug="encoding-rt-zlib-blob", owner="test-user-wire" ) raw = b"# muse.plugin.zsh\n\ntypeset -gA MUSE_DOMAIN_ICONS\n" * 15 oid = _sha256_oid(raw) compressed = _zlib_compress(raw) await client.post( f"/{repo.owner}/{repo.slug}/push/object-pack", content=_mp({"objects": [{ "object_id": oid, "content": compressed, "path": "muse.plugin.zsh", "encoding": "zlib", }]}), headers=wire_headers, ) # Fetch the raw object bytes via the fetch-objects endpoint (same path the # blob view and pull use). fetch_r = await client.post( f"/{repo.owner}/{repo.slug}/fetch/objects", content=_mp({"object_ids": [oid]}), headers=wire_headers, ) print(f"\n [ENCODING-TEST] fetch-objects status={fetch_r.status_code}") assert fetch_r.status_code == 200, fetch_r.text # Streaming endpoint: one msgpack frame per object unpacker = msgpack.Unpacker(raw=False) unpacker.feed(fetch_r.content) objects = [obj for obj in unpacker if isinstance(obj, dict) and "object_id" in obj] print(f" [ENCODING-TEST] fetch-objects returned {len(objects)} object(s)") assert objects, "fetch-objects must return the pushed object" served = objects[0]["content"] if isinstance(served, (bytearray, memoryview)): served = bytes(served) _log_bytes("bytes served by fetch-objects", served) # decompress_if_needed is what the UI applies — simulate it decoded = decompress_if_needed(served) _log_bytes("after decompress_if_needed", decoded) assert decoded == raw, ( f"\nBLOB VIEW WOULD SHOW GIBBERISH — encoding bug confirmed.\n" f" original first 80B: {raw[:80]!r}\n" f" served first 80B: {served[:80]!r}\n" f" decoded first 80B: {decoded[:80]!r}\n" f" served is zlib: {served[:2].hex() in ('7801','789c','78da','785e')}" ) # --------------------------------------------------------------------------- # Test 3: delta+zlib — storage must hold reconstructed plain bytes # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_delta_zlib_stored_as_plain_bytes( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """delta+zlib push must produce correct plain bytes in storage.""" import musehub.services.musehub_wire as _wire_svc repo = await factory_create_repo( db_session, slug="encoding-rt-delta-storage", owner="test-user-wire" ) path = "muse.plugin.zsh" base_raw = b"# muse.plugin.zsh\n\nMUSE_DOMAIN_ICONS=(midi 'NOTE' code 'OPT')\n" * 30 base_oid = _sha256_oid(base_raw) base_compressed = _zlib_compress(base_raw) # Store base as zlib-compressed (simulates legacy state — the bug scenario) backend = _wire_svc.get_backend() await backend.put(base_oid, base_compressed) await db_session.execute( db.MusehubObject.__table__.insert().values( object_id=base_oid, path=path, size_bytes=len(base_raw), disk_path="", storage_uri=backend.uri_for(base_oid), content_cache=None, ) ) db_session.add(db.MusehubObjectRef(repo_id=repo.repo_id, object_id=base_oid)) await db_session.commit() target_raw = b"# muse.plugin.zsh\n\nMUSE_DOMAIN_ICONS=(midi 'NOTE' code 'OPT' scaffold 'HEX')\n" * 30 target_oid = _sha256_oid(target_raw) delta = _compute_delta(base_raw, target_raw) print(f"\n [ENCODING-TEST] base: oid={base_oid[:23]}… len={len(base_raw)} stored_as=zlib") print(f" [ENCODING-TEST] target: oid={target_oid[:23]}… len={len(target_raw)}") print(f" [ENCODING-TEST] delta: len={len(delta)} magic=0x{delta[:2].hex()}") r = await client.post( f"/{repo.owner}/{repo.slug}/push/object-pack", content=_mp({"objects": [{ "object_id": target_oid, "content": delta, "path": path, "encoding": "delta+zlib", "base_id": base_oid, }]}), headers=wire_headers, ) print(f" [ENCODING-TEST] push status={r.status_code}") assert r.status_code == 200, r.text stored = await backend.get(target_oid) obj_row = await db_session.get(db.MusehubObject, target_oid) cache = obj_row.content_cache if obj_row else None _log_bytes("stored in backend", stored) _log_bytes("content_cache in DB", cache) actual = stored if stored is not None else cache assert actual is not None, "reconstructed object must be findable" _log_bytes("bytes that will be served", actual) assert actual == target_raw, ( f"\nDELTA RECONSTRUCTION CORRUPT — encoding bug confirmed.\n" f" target first 80B: {target_raw[:80]!r}\n" f" stored first 80B: {actual[:80]!r}" ) # --------------------------------------------------------------------------- # Test 4: encoding=None — must NOT store compressed bytes (regression guard) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_encoding_none_stored_correctly( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """encoding=None (raw) — server must store bytes as-is. The real muse CLI always sets encoding="zlib". This test guards against a regression where raw bytes are accidentally re-compressed on the server. """ import musehub.services.musehub_wire as _wire_svc repo = await factory_create_repo( db_session, slug="encoding-rt-none", owner="test-user-wire" ) raw = b"# plain file\n\nNo encoding.\n" * 20 oid = _sha256_oid(raw) print(f"\n [ENCODING-TEST] raw len={len(raw)} oid={oid[:23]}…") r = await client.post( f"/{repo.owner}/{repo.slug}/push/object-pack", content=_mp({"objects": [{ "object_id": oid, "content": raw, "path": "plain.txt", # no encoding field — server receives encoding=None }]}), headers=wire_headers, ) print(f" [ENCODING-TEST] push status={r.status_code}") assert r.status_code == 200, r.text backend = _wire_svc.get_backend() stored = await backend.get(oid) obj_row = await db_session.get(db.MusehubObject, oid) cache = obj_row.content_cache if obj_row else None actual = stored if stored is not None else cache _log_bytes("bytes stored for encoding=None", actual) assert actual is not None, "object must be stored" assert actual == raw, ( f"\nEncoding=None object stored incorrectly.\n" f" original first 40B: {raw[:40]!r}\n" f" stored first 40B: {actual[:40]!r}" )