test_encoding_roundtrip.py
python
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65
refactor: enforce gRPC framing on all MWP wire traffic
Sonnet 4.6
minor
⚠ breaking
155 days ago
| 1 | """TDD: object encoding integrity — push → storage → blob read roundtrip. |
| 2 | |
| 3 | Failure mode being hunted: |
| 4 | Objects pushed via wire_push_object_pack arrive zlib-encoded (encoding="zlib"). |
| 5 | If the server fails to decode before storing, the blob view serves compressed |
| 6 | binary gibberish instead of the original source text. |
| 7 | |
| 8 | All object IDs use the canonical sha256: prefix — bare hex is rejected at the |
| 9 | Pydantic boundary. |
| 10 | |
| 11 | Coverage matrix (each test logs exactly what was stored and what was served): |
| 12 | 1. zlib + sha256: ID → storage → assert plain bytes |
| 13 | 2. zlib + sha256: ID → blob view endpoint → assert readable text |
| 14 | 3. delta+zlib + sha256: ID → storage → assert reconstructed plain bytes |
| 15 | 4. encoding=None + sha256: ID → storage → assert NOT compressed (regression guard) |
| 16 | |
| 17 | All tests print a structured log block so a CI failure shows the real bytes, not |
| 18 | just an assertion message. |
| 19 | """ |
| 20 | from __future__ import annotations |
| 21 | |
| 22 | import hashlib |
| 23 | import logging |
| 24 | import struct |
| 25 | import zlib |
| 26 | |
| 27 | import msgpack |
| 28 | import pytest |
| 29 | from httpx import AsyncClient |
| 30 | from sqlalchemy.ext.asyncio import AsyncSession |
| 31 | |
| 32 | from musehub.db import musehub_models as db |
| 33 | from musehub.types.compression import decompress_if_needed |
| 34 | from musehub.types.json_types import JSONValue, StrDict |
| 35 | from tests.factories import create_repo as factory_create_repo |
| 36 | |
| 37 | logger = logging.getLogger(__name__) |
| 38 | |
| 39 | |
| 40 | # --------------------------------------------------------------------------- |
| 41 | # Helpers — mirror exact client behaviour |
| 42 | # --------------------------------------------------------------------------- |
| 43 | |
| 44 | def _sha256_oid(raw: bytes) -> str: |
| 45 | """Canonical object ID: sha256:<hex>.""" |
| 46 | return "sha256:" + hashlib.sha256(raw).hexdigest() |
| 47 | |
| 48 | |
| 49 | def _mp(data: JSONValue) -> bytes: |
| 50 | return msgpack.packb(data, use_bin_type=True) |
| 51 | |
| 52 | |
| 53 | def _zlib_compress(raw: bytes) -> bytes: |
| 54 | """Tier-1 encoding: plain zlib, level 1 (fast), matching muse CLI compress_zlib.""" |
| 55 | return zlib.compress(raw, level=1) |
| 56 | |
| 57 | |
| 58 | def _compute_delta(base: bytes, target: bytes) -> bytes: |
| 59 | """Build zlib-compressed delta instruction stream (mirrors muse/core/compression.py).""" |
| 60 | # Simple delta: chunk_size = 32 bytes |
| 61 | CHUNK = 32 |
| 62 | table: dict[bytes, int] = {} |
| 63 | for i in range(0, len(base) - CHUNK + 1, CHUNK): |
| 64 | table[base[i: i + CHUNK]] = i |
| 65 | |
| 66 | result = bytearray() |
| 67 | pos = 0 |
| 68 | while pos < len(target): |
| 69 | chunk = target[pos: pos + CHUNK] |
| 70 | if chunk in table: |
| 71 | offset = table[chunk] |
| 72 | length = CHUNK |
| 73 | # Extend match |
| 74 | while (pos + length < len(target) |
| 75 | and offset + length < len(base) |
| 76 | and target[pos + length] == base[offset + length]): |
| 77 | length += 1 |
| 78 | result += b"\x00" + struct.pack(">II", offset, length) |
| 79 | pos += length |
| 80 | else: |
| 81 | literal = target[pos: pos + 1] |
| 82 | result += b"\x01" + struct.pack(">I", 1) + literal |
| 83 | pos += 1 |
| 84 | |
| 85 | return zlib.compress(bytes(result), level=1) |
| 86 | |
| 87 | |
| 88 | def _log_bytes(label: str, data: bytes | None, n: int = 120) -> None: |
| 89 | """Emit a structured log block for test forensics.""" |
| 90 | if data is None: |
| 91 | logger.warning("[ENCODING-TEST] %s: None", label) |
| 92 | print(f"\n [ENCODING-TEST] {label}: None") |
| 93 | return |
| 94 | snippet = data[:n] |
| 95 | is_text = all(0x09 <= b <= 0x7E or b in (0x0A, 0x0D) for b in snippet) |
| 96 | magic = data[:2].hex() if len(data) >= 2 else "??" |
| 97 | print( |
| 98 | f"\n [ENCODING-TEST] {label}:" |
| 99 | f"\n len={len(data)} magic_bytes=0x{magic} readable={is_text}" |
| 100 | f"\n first {n}B: {snippet!r}" |
| 101 | ) |
| 102 | logger.info( |
| 103 | "[ENCODING-TEST] %s: len=%d magic=0x%s readable=%s first=%r", |
| 104 | label, len(data), magic, is_text, snippet, |
| 105 | ) |
| 106 | |
| 107 | |
| 108 | # --------------------------------------------------------------------------- |
| 109 | # Test 1: zlib-encoded object with bare hex ID — storage must hold plain bytes |
| 110 | # --------------------------------------------------------------------------- |
| 111 | |
| 112 | @pytest.mark.asyncio |
| 113 | async def test_zlib_stored_as_plain_bytes( |
| 114 | client: AsyncClient, |
| 115 | db_session: AsyncSession, |
| 116 | wire_headers: StrDict, |
| 117 | ) -> None: |
| 118 | """Push a zlib-encoded object; storage must contain plain bytes. |
| 119 | |
| 120 | This is the most critical invariant: the server must decode before writing. |
| 121 | If this fails, every blob in the repo is gibberish. |
| 122 | """ |
| 123 | import musehub.services.musehub_wire as _wire_svc |
| 124 | |
| 125 | repo = await factory_create_repo( |
| 126 | db_session, slug="encoding-rt-zlib-storage", owner="test-user-wire" |
| 127 | ) |
| 128 | raw = b"# muse-zsh\n\nOh My ZSH plugin for Muse version control.\n" * 20 |
| 129 | oid = _sha256_oid(raw) |
| 130 | compressed = _zlib_compress(raw) |
| 131 | |
| 132 | print(f"\n [ENCODING-TEST] input: len={len(raw)} oid={oid[:23]}…") |
| 133 | print(f" [ENCODING-TEST] compressed: len={len(compressed)} magic=0x{compressed[:2].hex()}") |
| 134 | |
| 135 | r = await client.post( |
| 136 | f"/{repo.owner}/{repo.slug}/push/object-pack", |
| 137 | content=_mp({"objects": [{ |
| 138 | "object_id": oid, |
| 139 | "content": compressed, |
| 140 | "path": "README.md", |
| 141 | "encoding": "zlib", |
| 142 | }]}), |
| 143 | headers=wire_headers, |
| 144 | ) |
| 145 | print(f" [ENCODING-TEST] push response: status={r.status_code} body={r.content[:200]!r}") |
| 146 | assert r.status_code == 200, r.text |
| 147 | |
| 148 | resp_data = msgpack.unpackb(r.content, raw=False) |
| 149 | print(f" [ENCODING-TEST] push result: stored={resp_data.get('stored')} skipped={resp_data.get('skipped')}") |
| 150 | |
| 151 | backend = _wire_svc.get_backend() |
| 152 | stored = await backend.get(oid) |
| 153 | _log_bytes("stored bytes from backend", stored) |
| 154 | |
| 155 | # Also check content_cache in DB |
| 156 | obj_row = (await db_session.get(db.MusehubObject, oid)) |
| 157 | if obj_row: |
| 158 | cache = obj_row.content_cache |
| 159 | _log_bytes("content_cache in DB", cache) |
| 160 | print(f" [ENCODING-TEST] storage_uri={obj_row.storage_uri!r}") |
| 161 | |
| 162 | assert stored is not None or (obj_row and obj_row.content_cache is not None), \ |
| 163 | "object must be findable in storage or content_cache" |
| 164 | |
| 165 | actual = stored if stored is not None else obj_row.content_cache # type: ignore[union-attr] |
| 166 | _log_bytes("bytes that will be served", actual) |
| 167 | |
| 168 | assert actual == raw, ( |
| 169 | f"\nSTORED BYTES DO NOT MATCH ORIGINAL — encoding bug confirmed.\n" |
| 170 | f" original first 80B: {raw[:80]!r}\n" |
| 171 | f" stored first 80B: {actual[:80]!r}\n" |
| 172 | f" stored is zlib: {actual[:2].hex() in ('7801','789c','78da','785e')}" |
| 173 | ) |
| 174 | |
| 175 | |
| 176 | # --------------------------------------------------------------------------- |
| 177 | # Test 2: zlib-encoded object — blob view must serve readable text |
| 178 | # --------------------------------------------------------------------------- |
| 179 | |
| 180 | @pytest.mark.asyncio |
| 181 | async def test_zlib_blob_view_serves_plain_text( |
| 182 | client: AsyncClient, |
| 183 | db_session: AsyncSession, |
| 184 | wire_headers: StrDict, |
| 185 | ) -> None: |
| 186 | """After pushing with zlib encoding, fetch/objects must return decompressed plain text.""" |
| 187 | import musehub.services.musehub_wire as _wire_svc |
| 188 | |
| 189 | repo = await factory_create_repo( |
| 190 | db_session, slug="encoding-rt-zlib-blob", owner="test-user-wire" |
| 191 | ) |
| 192 | raw = b"# muse.plugin.zsh\n\ntypeset -gA MUSE_DOMAIN_ICONS\n" * 15 |
| 193 | oid = _sha256_oid(raw) |
| 194 | compressed = _zlib_compress(raw) |
| 195 | |
| 196 | await client.post( |
| 197 | f"/{repo.owner}/{repo.slug}/push/object-pack", |
| 198 | content=_mp({"objects": [{ |
| 199 | "object_id": oid, |
| 200 | "content": compressed, |
| 201 | "path": "muse.plugin.zsh", |
| 202 | "encoding": "zlib", |
| 203 | }]}), |
| 204 | headers=wire_headers, |
| 205 | ) |
| 206 | |
| 207 | # Fetch the raw object bytes via the fetch-objects endpoint (same path the |
| 208 | # blob view and pull use). |
| 209 | fetch_r = await client.post( |
| 210 | f"/{repo.owner}/{repo.slug}/fetch/objects", |
| 211 | content=_mp({"object_ids": [oid]}), |
| 212 | headers=wire_headers, |
| 213 | ) |
| 214 | print(f"\n [ENCODING-TEST] fetch-objects status={fetch_r.status_code}") |
| 215 | |
| 216 | assert fetch_r.status_code == 200, fetch_r.text |
| 217 | # Streaming endpoint: one msgpack frame per object |
| 218 | unpacker = msgpack.Unpacker(raw=False) |
| 219 | unpacker.feed(fetch_r.content) |
| 220 | objects = [obj for obj in unpacker if isinstance(obj, dict) and "object_id" in obj] |
| 221 | print(f" [ENCODING-TEST] fetch-objects returned {len(objects)} object(s)") |
| 222 | |
| 223 | assert objects, "fetch-objects must return the pushed object" |
| 224 | served = objects[0]["content"] |
| 225 | if isinstance(served, (bytearray, memoryview)): |
| 226 | served = bytes(served) |
| 227 | |
| 228 | _log_bytes("bytes served by fetch-objects", served) |
| 229 | |
| 230 | # decompress_if_needed is what the UI applies — simulate it |
| 231 | decoded = decompress_if_needed(served) |
| 232 | _log_bytes("after decompress_if_needed", decoded) |
| 233 | |
| 234 | assert decoded == raw, ( |
| 235 | f"\nBLOB VIEW WOULD SHOW GIBBERISH — encoding bug confirmed.\n" |
| 236 | f" original first 80B: {raw[:80]!r}\n" |
| 237 | f" served first 80B: {served[:80]!r}\n" |
| 238 | f" decoded first 80B: {decoded[:80]!r}\n" |
| 239 | f" served is zlib: {served[:2].hex() in ('7801','789c','78da','785e')}" |
| 240 | ) |
| 241 | |
| 242 | |
| 243 | # --------------------------------------------------------------------------- |
| 244 | # Test 3: delta+zlib — storage must hold reconstructed plain bytes |
| 245 | # --------------------------------------------------------------------------- |
| 246 | |
| 247 | @pytest.mark.asyncio |
| 248 | async def test_delta_zlib_stored_as_plain_bytes( |
| 249 | client: AsyncClient, |
| 250 | db_session: AsyncSession, |
| 251 | wire_headers: StrDict, |
| 252 | ) -> None: |
| 253 | """delta+zlib push must produce correct plain bytes in storage.""" |
| 254 | import musehub.services.musehub_wire as _wire_svc |
| 255 | |
| 256 | repo = await factory_create_repo( |
| 257 | db_session, slug="encoding-rt-delta-storage", owner="test-user-wire" |
| 258 | ) |
| 259 | path = "muse.plugin.zsh" |
| 260 | |
| 261 | base_raw = b"# muse.plugin.zsh\n\nMUSE_DOMAIN_ICONS=(midi 'NOTE' code 'OPT')\n" * 30 |
| 262 | base_oid = _sha256_oid(base_raw) |
| 263 | base_compressed = _zlib_compress(base_raw) |
| 264 | |
| 265 | # Store base as zlib-compressed (simulates legacy state — the bug scenario) |
| 266 | backend = _wire_svc.get_backend() |
| 267 | await backend.put(base_oid, base_compressed) |
| 268 | await db_session.execute( |
| 269 | db.MusehubObject.__table__.insert().values( |
| 270 | object_id=base_oid, |
| 271 | path=path, |
| 272 | size_bytes=len(base_raw), |
| 273 | disk_path="", |
| 274 | storage_uri=backend.uri_for(base_oid), |
| 275 | content_cache=None, |
| 276 | ) |
| 277 | ) |
| 278 | db_session.add(db.MusehubObjectRef(repo_id=repo.repo_id, object_id=base_oid)) |
| 279 | await db_session.commit() |
| 280 | |
| 281 | target_raw = b"# muse.plugin.zsh\n\nMUSE_DOMAIN_ICONS=(midi 'NOTE' code 'OPT' scaffold 'HEX')\n" * 30 |
| 282 | target_oid = _sha256_oid(target_raw) |
| 283 | delta = _compute_delta(base_raw, target_raw) |
| 284 | |
| 285 | print(f"\n [ENCODING-TEST] base: oid={base_oid[:23]}… len={len(base_raw)} stored_as=zlib") |
| 286 | print(f" [ENCODING-TEST] target: oid={target_oid[:23]}… len={len(target_raw)}") |
| 287 | print(f" [ENCODING-TEST] delta: len={len(delta)} magic=0x{delta[:2].hex()}") |
| 288 | |
| 289 | r = await client.post( |
| 290 | f"/{repo.owner}/{repo.slug}/push/object-pack", |
| 291 | content=_mp({"objects": [{ |
| 292 | "object_id": target_oid, |
| 293 | "content": delta, |
| 294 | "path": path, |
| 295 | "encoding": "delta+zlib", |
| 296 | "base_id": base_oid, |
| 297 | }]}), |
| 298 | headers=wire_headers, |
| 299 | ) |
| 300 | print(f" [ENCODING-TEST] push status={r.status_code}") |
| 301 | assert r.status_code == 200, r.text |
| 302 | |
| 303 | stored = await backend.get(target_oid) |
| 304 | obj_row = await db_session.get(db.MusehubObject, target_oid) |
| 305 | cache = obj_row.content_cache if obj_row else None |
| 306 | |
| 307 | _log_bytes("stored in backend", stored) |
| 308 | _log_bytes("content_cache in DB", cache) |
| 309 | |
| 310 | actual = stored if stored is not None else cache |
| 311 | assert actual is not None, "reconstructed object must be findable" |
| 312 | |
| 313 | _log_bytes("bytes that will be served", actual) |
| 314 | |
| 315 | assert actual == target_raw, ( |
| 316 | f"\nDELTA RECONSTRUCTION CORRUPT — encoding bug confirmed.\n" |
| 317 | f" target first 80B: {target_raw[:80]!r}\n" |
| 318 | f" stored first 80B: {actual[:80]!r}" |
| 319 | ) |
| 320 | |
| 321 | |
| 322 | # --------------------------------------------------------------------------- |
| 323 | # Test 4: encoding=None — must NOT store compressed bytes (regression guard) |
| 324 | # --------------------------------------------------------------------------- |
| 325 | |
| 326 | @pytest.mark.asyncio |
| 327 | async def test_encoding_none_stored_correctly( |
| 328 | client: AsyncClient, |
| 329 | db_session: AsyncSession, |
| 330 | wire_headers: StrDict, |
| 331 | ) -> None: |
| 332 | """encoding=None (raw) — server must store bytes as-is. |
| 333 | |
| 334 | The real muse CLI always sets encoding="zlib". This test guards against a |
| 335 | regression where raw bytes are accidentally re-compressed on the server. |
| 336 | """ |
| 337 | import musehub.services.musehub_wire as _wire_svc |
| 338 | |
| 339 | repo = await factory_create_repo( |
| 340 | db_session, slug="encoding-rt-none", owner="test-user-wire" |
| 341 | ) |
| 342 | raw = b"# plain file\n\nNo encoding.\n" * 20 |
| 343 | oid = _sha256_oid(raw) |
| 344 | |
| 345 | print(f"\n [ENCODING-TEST] raw len={len(raw)} oid={oid[:23]}…") |
| 346 | |
| 347 | r = await client.post( |
| 348 | f"/{repo.owner}/{repo.slug}/push/object-pack", |
| 349 | content=_mp({"objects": [{ |
| 350 | "object_id": oid, |
| 351 | "content": raw, |
| 352 | "path": "plain.txt", |
| 353 | # no encoding field — server receives encoding=None |
| 354 | }]}), |
| 355 | headers=wire_headers, |
| 356 | ) |
| 357 | print(f" [ENCODING-TEST] push status={r.status_code}") |
| 358 | assert r.status_code == 200, r.text |
| 359 | |
| 360 | backend = _wire_svc.get_backend() |
| 361 | stored = await backend.get(oid) |
| 362 | obj_row = await db_session.get(db.MusehubObject, oid) |
| 363 | cache = obj_row.content_cache if obj_row else None |
| 364 | |
| 365 | actual = stored if stored is not None else cache |
| 366 | _log_bytes("bytes stored for encoding=None", actual) |
| 367 | |
| 368 | assert actual is not None, "object must be stored" |
| 369 | assert actual == raw, ( |
| 370 | f"\nEncoding=None object stored incorrectly.\n" |
| 371 | f" original first 40B: {raw[:40]!r}\n" |
| 372 | f" stored first 40B: {actual[:40]!r}" |
| 373 | ) |
File History
1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65
refactor: enforce gRPC framing on all MWP wire traffic
Sonnet 4.6
minor
⚠
155 days ago