"""Object store canonical contract — TDD spec. THE SINGLE RULE: object_id = blob_id(content) # → "sha256:<64-hex>" This format is the law everywhere: - musehub_objects.object_id column - snapshot manifests (path → object_id) - all wire protocol payloads (push, fetch, fetch/objects) - filesystem key: sha256_ (colon → underscore, safe for all FSes) - S3/R2 key: objects/sha256_ - LocalBackend path: /sha256_ No raw hex. No stripping. No conditionals. No "bare_id". Tiers: 1 – Unit pure logic, no network, no DB 2 – Schema server HTTP contract 3 – Integration push → fetch/objects → sha256 round-trip 4 – Stress 100 objects, all must round-trip 5 – Persistence object_id in DB is sha256: prefixed 6 – Performance round-trip under 200ms 7 – Security wrong prefix / malformed id rejected """ from __future__ import annotations import secrets import struct from datetime import datetime, timezone import msgpack import pytest from httpx import AsyncClient from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import blob_id, fake_id, now_utc_iso from musehub.db import musehub_models as db from musehub.types.json_types import JSONObject, JSONValue, StrDict from tests.factories import create_repo as factory_create_repo # ── constants ───────────────────────────────────────────────────────────────── _OWNER = "test-user-wire" # matches _WIRE_CONTEXT.handle in conftest def _mp(obj: JSONValue) -> bytes: return msgpack.packb(obj, use_bin_type=True) def _parse_stream(raw: bytes) -> list[dict]: """Parse concatenated self-delimiting msgpack frames.""" unpacker = msgpack.Unpacker(raw=False) unpacker.feed(raw) return list(unpacker) def _stream_headers(wire_headers: StrDict) -> StrDict: return {**wire_headers, "Accept": "application/x-msgpack-stream"} def _mwp_frame(ft: str, data: JSONObject) -> bytes: payload = msgpack.packb(data, use_bin_type=True) envelope = msgpack.packb( {"ft": ft, "sz": len(payload), "id": blob_id(payload)}, use_bin_type=True, ) return ( b"muse" + b"\x01" + struct.pack(">I", len(envelope)) + envelope + struct.pack(">Q", len(payload)) + payload ) def _mwp_stream( commits: list[dict], snapshots: list[dict], objects: list[dict], *, branch: str = "main", force: bool = True, ) -> bytes: frames: list[bytes] = [ _mwp_frame("H", { "t": "H", "branch": branch, "force": force, "head": None, "have": [], "n_objects": len(objects), "n_commits": len(commits), }) ] for obj in objects: frames.append(_mwp_frame("O", { "t": "O", "id": obj["object_id"], "path": obj.get("path", ""), "content": obj["content"], "enc": "raw", })) frames.append(_mwp_frame("C", { "t": "C", "commits": commits, "snapshots": snapshots, })) frames.append(_mwp_frame("E", { "t": "E", "n_objects": len(objects), "n_commits": len(commits), })) return b"".join(frames) async def _push( client: AsyncClient, owner: str, slug: str, objects: list[tuple[str, bytes]], # [(path, content), ...] wire_headers: StrDict, *, force: bool = True, ) -> str: """Push objects to a repo via push/stream; return commit_id.""" commit_id = blob_id(b"commit-" + secrets.token_bytes(16)) snap_id = blob_id(b"snap-" + secrets.token_bytes(16)) oids = {path: blob_id(content) for path, content in objects} wire_objects = [ {"object_id": oids[path], "content": content, "path": path} for path, content in objects ] commits = [{ "commit_id": commit_id, "repo_id": "", "branch": "main", "snapshot_id": snap_id, "message": "test push", "committed_at": now_utc_iso(), "parent_commit_id": None, "author": "Test ", "sem_ver_bump": "patch", }] snapshots = [{ "snapshot_id": snap_id, "manifest": oids, "directories": [], "created_at": now_utc_iso(), }] resp = await client.post( f"/{owner}/{slug}/push/stream", content=_mwp_stream(commits, snapshots, wire_objects, force=force), headers={**wire_headers, "Content-Type": "application/x-muse-wire"}, ) assert resp.status_code in (200, 201), f"push failed {resp.status_code}: {resp.text}" return commit_id async def _fetch_objects( client: AsyncClient, owner: str, slug: str, oids: list[str], wire_headers: StrDict, ) -> list[dict]: resp = await client.post( f"/{owner}/{slug}/fetch/objects", content=_mp({"object_ids": oids}), headers=_stream_headers(wire_headers), ) assert resp.status_code == 200, f"fetch/objects failed: {resp.text}" return _parse_stream(resp.content) # ── Tier 1 — Unit ───────────────────────────────────────────────────────────── class TestUnit: """Pure logic — no network, no DB.""" def test_sha256_oid_has_prefix(self) -> None: oid = blob_id(b"hello") assert oid.startswith("sha256:") def test_sha256_oid_hex_is_64_chars(self) -> None: oid = blob_id(b"hello") assert len(oid.removeprefix("sha256:")) == 64 def test_sha256_oid_known_value(self) -> None: # echo -n "hello" | sha256sum assert blob_id(b"hello") == ( "sha256:2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" ) def test_empty_content_oid(self) -> None: oid = blob_id(b"") assert oid == "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" def test_local_backend_path_uses_sharded_layout(self) -> None: """LocalBackend uses algo-namespaced sharded layout: objects/sha256/<2-hex>/<62-hex>.""" from musehub.storage.backends import LocalBackend from pathlib import Path import tempfile with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) backend = LocalBackend(repo_root=root) oid = blob_id(b"test data") path = backend._path(oid, repo_root=root) _, hex_part = oid.split(":", 1) assert path.name == hex_part[2:] # remaining 62-hex chars assert path.parent.name == hex_part[:2] # 2-hex shard prefix assert path.parent.parent.name == "sha256" # algo namespace def test_local_backend_path_is_under_objects_dir(self) -> None: """The path must sit inside /objects/.""" from musehub.storage.backends import LocalBackend from pathlib import Path import tempfile with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) backend = LocalBackend(repo_root=root) oid = blob_id(b"test data") path = backend._path(oid, repo_root=root) assert path.is_relative_to(root) def test_s3_key_uses_prefix_form(self) -> None: """S3Backend key must be objects/sha256:.""" from musehub.storage.backends import S3Backend b = S3Backend.__new__(S3Backend) oid = blob_id(b"test data") key = b._key(oid) assert key.startswith("objects/sha256:") assert len(key) == len("objects/sha256:") + 64 # ── Tier 2 — Schema ─────────────────────────────────────────────────────────── class TestSchema: """HTTP contract — response shapes and content types.""" async def test_fetch_objects_returns_stream_content_type( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, owner=_OWNER) content = b"schema test" oid = blob_id(content) await _push(client, _OWNER, repo.slug, [("f.py", content)], wire_headers) resp = await client.post( f"/{_OWNER}/{repo.slug}/fetch/objects", content=_mp({"object_ids": [oid]}), headers=_stream_headers(wire_headers), ) assert resp.status_code == 200 assert "application/x-msgpack" in resp.headers["content-type"] async def test_fetched_object_id_has_sha256_prefix( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """object_id in the stream response must carry the sha256: prefix.""" repo = await factory_create_repo(db_session, owner=_OWNER) content = b"prefix check" oid = blob_id(content) await _push(client, _OWNER, repo.slug, [("a.py", content)], wire_headers) objs = await _fetch_objects(client, _OWNER, repo.slug, [oid], wire_headers) assert len(objs) == 1 assert objs[0]["object_id"] == oid assert objs[0]["object_id"].startswith("sha256:") async def test_push_manifest_uses_sha256_prefix( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Stored snapshot manifests must use sha256: prefixed object IDs.""" repo = await factory_create_repo(db_session, owner=_OWNER) content = b"manifest test" oid = blob_id(content) await _push(client, _OWNER, repo.slug, [("m.py", content)], wire_headers) # Decode manifest_blob from MusehubSnapshot — entries are stored there. rows = (await db_session.execute( select(db.MusehubSnapshot).where( db.MusehubSnapshot.repo_id == repo.repo_id ) )).scalars().all() assert rows, "no snapshot row found after push" for row in rows: manifest: JSONObject = msgpack.unpackb(row.manifest_blob, raw=False) for path, manifest_oid in manifest.items(): assert manifest_oid.startswith("sha256:"), ( f"snapshot path={path!r} has object_id={manifest_oid!r} — " "expected sha256: prefix" ) # ── Tier 3 — Integration ────────────────────────────────────────────────────── class TestIntegration: """Push → fetch/objects → sha256 integrity round-trip.""" async def test_single_object_roundtrip( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """sha256(received_bytes) must equal the object_id.""" repo = await factory_create_repo(db_session, owner=_OWNER) content = b"round trip content" oid = blob_id(content) await _push(client, _OWNER, repo.slug, [("r.py", content)], wire_headers) objs = await _fetch_objects(client, _OWNER, repo.slug, [oid], wire_headers) assert len(objs) == 1 received = objs[0]["content"] assert isinstance(received, bytes) assert blob_id(received) == oid async def test_multiple_objects_all_roundtrip( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, owner=_OWNER) files = [(f"file{i}.py", f"content {i} {secrets.token_hex(16)}".encode()) for i in range(10)] oids = [blob_id(c) for _, c in files] await _push(client, _OWNER, repo.slug, files, wire_headers) objs = await _fetch_objects(client, _OWNER, repo.slug, oids, wire_headers) assert len(objs) == 10 received = {o["object_id"]: o["content"] for o in objs} for _, content in files: oid = blob_id(content) assert oid in received assert blob_id(received[oid]) == oid async def test_unknown_oid_silently_omitted( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, owner=_OWNER) ghost = fake_id("ghost-object") objs = await _fetch_objects(client, _OWNER, repo.slug, [ghost], wire_headers) assert objs == [] async def test_empty_object_roundtrip( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """The empty object (sha256 of b'') must round-trip correctly.""" repo = await factory_create_repo(db_session, owner=_OWNER) content = b"" oid = blob_id(content) await _push(client, _OWNER, repo.slug, [("empty.py", content)], wire_headers) objs = await _fetch_objects(client, _OWNER, repo.slug, [oid], wire_headers) assert len(objs) == 1 assert objs[0]["content"] == b"" assert blob_id(objs[0]["content"]) == oid # ── Tier 4 — Stress ─────────────────────────────────────────────────────────── class TestStress: """100 objects — none dropped, all integrity checks pass.""" async def test_100_objects_all_roundtrip( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, owner=_OWNER) files = [(f"f{i}.bin", f"stress {i} {secrets.token_hex(16)}".encode()) for i in range(100)] oids = [blob_id(c) for _, c in files] await _push(client, _OWNER, repo.slug, files, wire_headers) objs = await _fetch_objects(client, _OWNER, repo.slug, oids, wire_headers) assert len(objs) == 100 for obj in objs: assert obj["object_id"].startswith("sha256:") assert blob_id(obj["content"]) == obj["object_id"] # ── Tier 5 — Persistence ───────────────────────────────────────────────────── class TestPersistence: """DB rows must store sha256: prefixed object_ids.""" async def test_db_stores_sha256_prefixed_object_id( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, owner=_OWNER) content = b"db persistence check" oid = blob_id(content) await _push(client, _OWNER, repo.slug, [("p.py", content)], wire_headers) # Query directly — must find the row with sha256: prefix row = await db_session.execute( select(db.MusehubObject).where(db.MusehubObject.object_id == oid) ) obj_row = row.scalar_one_or_none() assert obj_row is not None, f"No DB row found for object_id={oid!r}" assert obj_row.object_id == oid assert obj_row.object_id.startswith("sha256:") async def test_db_has_no_raw_hex_object_ids( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """After any push, no musehub_objects row may have a bare hex object_id.""" repo = await factory_create_repo(db_session, owner=_OWNER) await _push(client, _OWNER, repo.slug, [("x.py", b"check raw hex")], wire_headers) result = await db_session.execute( text("SELECT COUNT(*) FROM musehub_objects WHERE object_id NOT LIKE 'sha256:%'") ) count = result.scalar() assert count == 0, f"{count} object(s) stored without sha256: prefix" # ── Tier 6 — Performance ───────────────────────────────────────────────────── class TestPerformance: """Latency budgets.""" async def test_10_objects_under_100ms( self, client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: import time repo = await factory_create_repo(db_session, owner=_OWNER) files = [(f"f{i}.py", f"perf {i}".encode()) for i in range(10)] oids = [blob_id(c) for _, c in files] await _push(client, _OWNER, repo.slug, files, wire_headers) t0 = time.perf_counter() objs = await _fetch_objects(client, _OWNER, repo.slug, oids, wire_headers) elapsed_ms = (time.perf_counter() - t0) * 1000 assert len(objs) == 10 assert elapsed_ms < 100, f"fetch/objects took {elapsed_ms:.0f}ms (budget: 100ms)"