gabriel / musehub public
test_encoding_roundtrip.py python
373 lines 13.6 KB
Raw
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