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