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