gabriel / musehub public
test_object_store_canonical.py python
482 lines 19.1 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Object store canonical contract — TDD spec.
2
3 THE SINGLE RULE:
4 object_id = "sha256:" + hashlib.sha256(content).hexdigest()
5
6 This format is the law everywhere:
7 - musehub_objects.object_id column
8 - snapshot manifests (path → object_id)
9 - all wire protocol payloads (push, fetch, fetch/objects)
10 - filesystem key: sha256_<hex> (colon → underscore, safe for all FSes)
11 - S3/R2 key: objects/sha256_<hex>
12 - LocalBackend path: <musehub_objects_dir>/sha256_<hex>
13
14 No raw hex. No stripping. No conditionals. No "bare_id".
15
16 Tiers:
17 1 – Unit pure logic, no network, no DB
18 2 – Schema server HTTP contract
19 3 – Integration push → fetch/objects → sha256 round-trip
20 4 – Stress 100 objects, all must round-trip
21 5 – Persistence object_id in DB is sha256: prefixed
22 6 – Performance round-trip under 200ms
23 7 – Security wrong prefix / malformed id rejected
24 """
25 from __future__ import annotations
26
27 import hashlib
28 import uuid
29 from datetime import datetime, timezone
30
31 import msgpack
32 import pytest
33 from httpx import AsyncClient
34 from sqlalchemy import select, text
35 from sqlalchemy.ext.asyncio import AsyncSession
36
37 from musehub.db import musehub_models as db
38 from tests.factories import create_repo as factory_create_repo
39
40 # ── constants ─────────────────────────────────────────────────────────────────
41
42 _OWNER = "test-user-wire" # matches _WIRE_CONTEXT.handle in conftest
43
44
45 def _sha256_oid(data: bytes) -> str:
46 """The one true object ID function."""
47 return "sha256:" + hashlib.sha256(data).hexdigest()
48
49
50 def _mp(obj: object) -> bytes:
51 return msgpack.packb(obj, use_bin_type=True)
52
53
54 def _now() -> str:
55 return datetime.now(tz=timezone.utc).isoformat()
56
57
58 def _parse_stream(raw: bytes) -> list[dict]:
59 """Parse concatenated self-delimiting msgpack frames."""
60 unpacker = msgpack.Unpacker(raw=False)
61 unpacker.feed(raw)
62 return list(unpacker)
63
64
65 def _stream_headers(wire_headers: dict) -> dict:
66 return {**wire_headers, "Accept": "application/x-msgpack-stream"}
67
68
69 async def _push(
70 client: AsyncClient,
71 owner: str,
72 slug: str,
73 objects: list[tuple[str, bytes]], # [(path, content), ...]
74 wire_headers: dict,
75 *,
76 force: bool = True,
77 ) -> str:
78 """Push objects to a repo; return commit_id."""
79 commit_id = "sha256:" + hashlib.sha256(b"commit-" + uuid.uuid4().bytes).hexdigest()
80 snap_id = "sha256:" + hashlib.sha256(b"snap-" + uuid.uuid4().bytes).hexdigest()
81 oids = {path: _sha256_oid(content) for path, content in objects}
82 resp = await client.post(
83 f"/{owner}/{slug}/push",
84 content=_mp({
85 "bundle": {
86 "commits": [{
87 "commit_id": commit_id,
88 "repo_id": "",
89 "branch": "main",
90 "snapshot_id": snap_id,
91 "message": "test push",
92 "committed_at": _now(),
93 "parent_commit_id": None,
94 "author": "Test <[email protected]>",
95 "sem_ver_bump": "patch",
96 }],
97 "snapshots": [{
98 "snapshot_id": snap_id,
99 "manifest": oids,
100 "created_at": _now(),
101 }],
102 "objects": [
103 {"object_id": oid, "content": content, "path": path}
104 for path, content in objects
105 for oid in [oids[path]]
106 ],
107 },
108 "branch": "main",
109 "force": force,
110 "local_head": commit_id,
111 }),
112 headers=wire_headers,
113 )
114 assert resp.status_code in (200, 201), f"push failed {resp.status_code}: {resp.text}"
115 return commit_id
116
117
118 async def _fetch_objects(
119 client: AsyncClient,
120 owner: str,
121 slug: str,
122 oids: list[str],
123 wire_headers: dict,
124 ) -> list[dict]:
125 resp = await client.post(
126 f"/{owner}/{slug}/fetch/objects",
127 content=_mp({"object_ids": oids}),
128 headers=_stream_headers(wire_headers),
129 )
130 assert resp.status_code == 200, f"fetch/objects failed: {resp.text}"
131 return _parse_stream(resp.content)
132
133
134 # ── Tier 1 — Unit ─────────────────────────────────────────────────────────────
135
136
137 class TestUnit:
138 """Pure logic — no network, no DB."""
139
140 def test_sha256_oid_has_prefix(self) -> None:
141 oid = _sha256_oid(b"hello")
142 assert oid.startswith("sha256:")
143
144 def test_sha256_oid_hex_is_64_chars(self) -> None:
145 oid = _sha256_oid(b"hello")
146 assert len(oid.removeprefix("sha256:")) == 64
147
148 def test_sha256_oid_known_value(self) -> None:
149 # echo -n "hello" | sha256sum
150 assert _sha256_oid(b"hello") == (
151 "sha256:2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
152 )
153
154 def test_empty_content_oid(self) -> None:
155 oid = _sha256_oid(b"")
156 assert oid == "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
157
158 def test_local_backend_path_uses_prefix_form(self) -> None:
159 """LocalBackend must derive filesystem key from object_id directly — no extra dirs."""
160 from musehub.storage.backends import LocalBackend
161 import tempfile, os
162 with tempfile.TemporaryDirectory() as tmp:
163 backend = LocalBackend(objects_dir=tmp)
164 oid = _sha256_oid(b"test data")
165 path = backend._path(oid)
166 # Must be exactly <root>/<safe_oid> — no subdirectories
167 assert path.parent == backend._root.resolve()
168 # Colon sanitised to underscore, no other transforms
169 assert path.name == oid.replace(":", "_").replace("/", "_")
170
171 def test_safe_id_sanitises_colon(self) -> None:
172 from musehub.storage.backends import LocalBackend
173 import tempfile
174 with tempfile.TemporaryDirectory() as tmp:
175 b = LocalBackend(objects_dir=tmp)
176 assert b._safe_id("sha256:abc") == "sha256_abc"
177
178 def test_s3_key_uses_prefix_form(self) -> None:
179 """S3Backend key must be objects/sha256_<hex>."""
180 from musehub.storage.backends import S3Backend
181 b = S3Backend.__new__(S3Backend)
182 oid = _sha256_oid(b"test data")
183 key = b._key(oid)
184 assert key.startswith("objects/sha256_")
185 assert len(key) == len("objects/sha256_") + 64
186
187
188 # ── Tier 2 — Schema ───────────────────────────────────────────────────────────
189
190
191 class TestSchema:
192 """HTTP contract — response shapes and content types."""
193
194 async def test_fetch_objects_returns_stream_content_type(
195 self,
196 client: AsyncClient,
197 db_session: AsyncSession,
198 wire_headers: dict,
199 ) -> None:
200 repo = await factory_create_repo(db_session, owner=_OWNER)
201 content = b"schema test"
202 oid = _sha256_oid(content)
203 await _push(client, _OWNER, repo.slug, [("f.py", content)], wire_headers)
204 resp = await client.post(
205 f"/{_OWNER}/{repo.slug}/fetch/objects",
206 content=_mp({"object_ids": [oid]}),
207 headers=_stream_headers(wire_headers),
208 )
209 assert resp.status_code == 200
210 assert "application/x-msgpack-stream" in resp.headers["content-type"]
211
212 async def test_fetched_object_id_has_sha256_prefix(
213 self,
214 client: AsyncClient,
215 db_session: AsyncSession,
216 wire_headers: dict,
217 ) -> None:
218 """object_id in the stream response must carry the sha256: prefix."""
219 repo = await factory_create_repo(db_session, owner=_OWNER)
220 content = b"prefix check"
221 oid = _sha256_oid(content)
222 await _push(client, _OWNER, repo.slug, [("a.py", content)], wire_headers)
223 objs = await _fetch_objects(client, _OWNER, repo.slug, [oid], wire_headers)
224 assert len(objs) == 1
225 assert objs[0]["object_id"] == oid
226 assert objs[0]["object_id"].startswith("sha256:")
227
228 async def test_push_manifest_uses_sha256_prefix(
229 self,
230 client: AsyncClient,
231 db_session: AsyncSession,
232 wire_headers: dict,
233 ) -> None:
234 """Snapshot manifest returned by /fetch must use sha256: prefixed ids."""
235 repo = await factory_create_repo(db_session, owner=_OWNER)
236 content = b"manifest test"
237 oid = _sha256_oid(content)
238 await _push(client, _OWNER, repo.slug, [("m.py", content)], wire_headers)
239 resp = await client.post(
240 f"/{_OWNER}/{repo.slug}/fetch",
241 content=_mp({"want": [], "have": []}),
242 headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"},
243 )
244 assert resp.status_code == 200
245 bundle = msgpack.unpackb(resp.content, raw=False)
246 for snap in bundle.get("snapshots", []):
247 for path, manifest_oid in snap.get("manifest", {}).items():
248 assert manifest_oid.startswith("sha256:"), (
249 f"manifest entry {path!r} has object_id {manifest_oid!r} — "
250 "expected sha256: prefix"
251 )
252
253
254 # ── Tier 3 — Integration ──────────────────────────────────────────────────────
255
256
257 class TestIntegration:
258 """Push → fetch/objects → sha256 integrity round-trip."""
259
260 async def test_single_object_roundtrip(
261 self,
262 client: AsyncClient,
263 db_session: AsyncSession,
264 wire_headers: dict,
265 ) -> None:
266 """sha256(received_bytes) must equal the object_id."""
267 repo = await factory_create_repo(db_session, owner=_OWNER)
268 content = b"round trip content"
269 oid = _sha256_oid(content)
270 await _push(client, _OWNER, repo.slug, [("r.py", content)], wire_headers)
271 objs = await _fetch_objects(client, _OWNER, repo.slug, [oid], wire_headers)
272 assert len(objs) == 1
273 received = objs[0]["content"]
274 assert isinstance(received, bytes)
275 assert _sha256_oid(received) == oid
276
277 async def test_multiple_objects_all_roundtrip(
278 self,
279 client: AsyncClient,
280 db_session: AsyncSession,
281 wire_headers: dict,
282 ) -> None:
283 repo = await factory_create_repo(db_session, owner=_OWNER)
284 files = [(f"file{i}.py", f"content {i} {uuid.uuid4().hex}".encode()) for i in range(10)]
285 oids = [_sha256_oid(c) for _, c in files]
286 await _push(client, _OWNER, repo.slug, files, wire_headers)
287 objs = await _fetch_objects(client, _OWNER, repo.slug, oids, wire_headers)
288 assert len(objs) == 10
289 received = {o["object_id"]: o["content"] for o in objs}
290 for _, content in files:
291 oid = _sha256_oid(content)
292 assert oid in received
293 assert _sha256_oid(received[oid]) == oid
294
295 async def test_unknown_oid_silently_omitted(
296 self,
297 client: AsyncClient,
298 db_session: AsyncSession,
299 wire_headers: dict,
300 ) -> None:
301 repo = await factory_create_repo(db_session, owner=_OWNER)
302 ghost = "sha256:" + "a" * 64
303 objs = await _fetch_objects(client, _OWNER, repo.slug, [ghost], wire_headers)
304 assert objs == []
305
306 async def test_empty_object_roundtrip(
307 self,
308 client: AsyncClient,
309 db_session: AsyncSession,
310 wire_headers: dict,
311 ) -> None:
312 """The empty object (sha256 of b'') must round-trip correctly."""
313 repo = await factory_create_repo(db_session, owner=_OWNER)
314 content = b""
315 oid = _sha256_oid(content)
316 await _push(client, _OWNER, repo.slug, [("empty.py", content)], wire_headers)
317 objs = await _fetch_objects(client, _OWNER, repo.slug, [oid], wire_headers)
318 assert len(objs) == 1
319 assert objs[0]["content"] == b""
320 assert _sha256_oid(objs[0]["content"]) == oid
321
322
323 # ── Tier 4 — Stress ───────────────────────────────────────────────────────────
324
325
326 class TestStress:
327 """100 objects — none dropped, all integrity checks pass."""
328
329 async def test_100_objects_all_roundtrip(
330 self,
331 client: AsyncClient,
332 db_session: AsyncSession,
333 wire_headers: dict,
334 ) -> None:
335 repo = await factory_create_repo(db_session, owner=_OWNER)
336 files = [(f"f{i}.bin", f"stress {i} {uuid.uuid4().hex}".encode()) for i in range(100)]
337 oids = [_sha256_oid(c) for _, c in files]
338 await _push(client, _OWNER, repo.slug, files, wire_headers)
339 objs = await _fetch_objects(client, _OWNER, repo.slug, oids, wire_headers)
340 assert len(objs) == 100
341 for obj in objs:
342 assert obj["object_id"].startswith("sha256:")
343 assert _sha256_oid(obj["content"]) == obj["object_id"]
344
345
346 # ── Tier 5 — Persistence ─────────────────────────────────────────────────────
347
348
349 class TestPersistence:
350 """DB rows must store sha256: prefixed object_ids."""
351
352 async def test_db_stores_sha256_prefixed_object_id(
353 self,
354 client: AsyncClient,
355 db_session: AsyncSession,
356 wire_headers: dict,
357 ) -> None:
358 repo = await factory_create_repo(db_session, owner=_OWNER)
359 content = b"db persistence check"
360 oid = _sha256_oid(content)
361 await _push(client, _OWNER, repo.slug, [("p.py", content)], wire_headers)
362 # Query directly — must find the row with sha256: prefix
363 row = await db_session.execute(
364 select(db.MusehubObject).where(db.MusehubObject.object_id == oid)
365 )
366 obj_row = row.scalar_one_or_none()
367 assert obj_row is not None, f"No DB row found for object_id={oid!r}"
368 assert obj_row.object_id == oid
369 assert obj_row.object_id.startswith("sha256:")
370
371 async def test_db_has_no_raw_hex_object_ids(
372 self,
373 client: AsyncClient,
374 db_session: AsyncSession,
375 wire_headers: dict,
376 ) -> None:
377 """After any push, no musehub_objects row may have a bare hex object_id."""
378 repo = await factory_create_repo(db_session, owner=_OWNER)
379 await _push(client, _OWNER, repo.slug, [("x.py", b"check raw hex")], wire_headers)
380 result = await db_session.execute(
381 text("SELECT COUNT(*) FROM musehub_objects WHERE object_id NOT LIKE 'sha256:%'")
382 )
383 count = result.scalar()
384 assert count == 0, f"{count} object(s) stored without sha256: prefix"
385
386
387 # ── Tier 6 — Performance ─────────────────────────────────────────────────────
388
389
390 class TestPerformance:
391 """Latency budgets."""
392
393 async def test_10_objects_under_100ms(
394 self,
395 client: AsyncClient,
396 db_session: AsyncSession,
397 wire_headers: dict,
398 ) -> None:
399 import time
400 repo = await factory_create_repo(db_session, owner=_OWNER)
401 files = [(f"f{i}.py", f"perf {i}".encode()) for i in range(10)]
402 oids = [_sha256_oid(c) for _, c in files]
403 await _push(client, _OWNER, repo.slug, files, wire_headers)
404 t0 = time.perf_counter()
405 objs = await _fetch_objects(client, _OWNER, repo.slug, oids, wire_headers)
406 elapsed_ms = (time.perf_counter() - t0) * 1000
407 assert len(objs) == 10
408 assert elapsed_ms < 100, f"fetch/objects took {elapsed_ms:.0f}ms (budget: 100ms)"
409
410
411 # ── Tier 7 — Security ────────────────────────────────────────────────────────
412
413
414 class TestSecurity:
415 """Malformed ids rejected; wrong content rejected."""
416
417 async def test_raw_hex_oid_in_push_rejected(
418 self,
419 client: AsyncClient,
420 db_session: AsyncSession,
421 wire_headers: dict,
422 ) -> None:
423 """Push with a bare hex object_id (no sha256: prefix) must be rejected."""
424 repo = await factory_create_repo(db_session, owner=_OWNER)
425 content = b"raw hex push"
426 bare_hex = hashlib.sha256(content).hexdigest() # no prefix
427 commit_id = "sha256:" + hashlib.sha256(b"commit-" + uuid.uuid4().bytes).hexdigest()
428 snap_id = "sha256:" + hashlib.sha256(b"snap-" + uuid.uuid4().bytes).hexdigest()
429 resp = await client.post(
430 f"/{_OWNER}/{repo.slug}/push",
431 content=_mp({
432 "bundle": {
433 "commits": [{
434 "commit_id": commit_id, "repo_id": "", "branch": "main",
435 "snapshot_id": snap_id, "message": "bad push",
436 "committed_at": _now(), "parent_commit_id": None,
437 "author": "T <[email protected]>", "sem_ver_bump": "patch",
438 }],
439 "snapshots": [{"snapshot_id": snap_id, "manifest": {"f.py": bare_hex}, "created_at": _now()}],
440 "objects": [{"object_id": bare_hex, "content": content, "path": "f.py"}],
441 },
442 "branch": "main", "force": True, "local_head": commit_id,
443 }),
444 headers=wire_headers,
445 )
446 assert resp.status_code in (400, 422), (
447 f"Expected 400/422 for bare hex object_id, got {resp.status_code}"
448 )
449
450 async def test_content_hash_mismatch_rejected_on_push(
451 self,
452 client: AsyncClient,
453 db_session: AsyncSession,
454 wire_headers: dict,
455 ) -> None:
456 """Push where sha256(content) != object_id must be rejected."""
457 repo = await factory_create_repo(db_session, owner=_OWNER)
458 real_content = b"real content"
459 wrong_content = b"wrong content"
460 oid = _sha256_oid(real_content) # correct oid for real_content
461 commit_id = "sha256:" + hashlib.sha256(b"commit-" + uuid.uuid4().bytes).hexdigest()
462 snap_id = "sha256:" + hashlib.sha256(b"snap-" + uuid.uuid4().bytes).hexdigest()
463 resp = await client.post(
464 f"/{_OWNER}/{repo.slug}/push",
465 content=_mp({
466 "bundle": {
467 "commits": [{
468 "commit_id": commit_id, "repo_id": "", "branch": "main",
469 "snapshot_id": snap_id, "message": "hash mismatch",
470 "committed_at": _now(), "parent_commit_id": None,
471 "author": "T <[email protected]>", "sem_ver_bump": "patch",
472 }],
473 "snapshots": [{"snapshot_id": snap_id, "manifest": {"f.py": oid}, "created_at": _now()}],
474 "objects": [{"object_id": oid, "content": wrong_content, "path": "f.py"}],
475 },
476 "branch": "main", "force": True, "local_head": commit_id,
477 }),
478 headers=wire_headers,
479 )
480 assert resp.status_code in (400, 422), (
481 f"Expected 400/422 for hash mismatch, got {resp.status_code}"
482 )
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago