gabriel / musehub public
test_push_object_integrity.py python
310 lines 10.9 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """TDD — push must verify objects exist in storage, not just in the DB.
2
3 Root cause of the staging missing-objects incident (2026-04-28):
4 The referential integrity check at end of push queried musehub_objects DB
5 rows for externally-referenced objects, but did NOT verify those objects
6 exist in R2/S3 storage. Result: DB row present, R2 bytes absent → push
7 accepted → raw/{file} returns 404 (object missing from storage).
8
9 These tests codify the three failure modes and the correct behaviour:
10
11 I1 When backend.put() silently returns but stores nothing, subsequent
12 GET must fail — and the push handler must detect this and emit ERROR.
13
14 I2 When an externally-referenced object is in the DB but absent from
15 actual storage, the referential integrity check must emit ERROR, not
16 RESULT ok=True.
17
18 I3 After a successful push, every object referenced by any snapshot
19 manifest must be retrievable from storage (exists() == True).
20
21 I4 Static check: the integrity check in musehub_wire.py must call
22 backend.exists() for externally-referenced objects, not just query DB.
23 """
24 from __future__ import annotations
25
26 import inspect
27 import msgpack
28 import pytest
29 from httpx import AsyncClient
30 from sqlalchemy.ext.asyncio import AsyncSession
31 from unittest.mock import AsyncMock, MagicMock, patch
32
33 from muse.core.types import blob_id
34 from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter
35 from musehub.models.wire import (
36 SFRAME_COMMIT_PACK,
37 SFRAME_END,
38 SFRAME_HEADER,
39 SFRAME_RESULT,
40 SFRAME_OBJECT,
41 )
42 from musehub.types.json_types import JSONObject, JSONValue, StrDict
43 from tests.factories import create_repo
44
45 _fw = MuseWireFrameWriter()
46
47
48 # ---------------------------------------------------------------------------
49 # Helpers
50 # ---------------------------------------------------------------------------
51
52 def _pack(obj: JSONValue) -> bytes:
53 return msgpack.packb(obj, use_bin_type=True)
54
55
56 def _wrap(ft: str, data: JSONValue) -> bytes:
57 return _fw.wrap(frame_type=ft, payload=_pack(data))
58
59
60 def _oid(data: bytes) -> str:
61 return blob_id(data)
62
63
64 def _header_frame(n_objects: int = 0, n_commits: int = 1) -> bytes:
65 return _wrap(SFRAME_HEADER, {
66 "t": SFRAME_HEADER,
67 "branch": "main",
68 "force": False,
69 "have": [],
70 "head": _oid(b"head"),
71 "n_objects": n_objects,
72 "n_commits": n_commits,
73 })
74
75
76 def _object_frame(raw: bytes, path: str = "file.py") -> tuple[str, bytes]:
77 oid = _oid(raw)
78 frame = _wrap(SFRAME_OBJECT, {
79 "t": SFRAME_OBJECT,
80 "id": oid,
81 "path": path,
82 "enc": "raw",
83 "content": raw,
84 })
85 return oid, frame
86
87
88 def _commit_pack_frame(
89 commits: list[JSONObject],
90 snapshots: list[JSONObject],
91 ) -> bytes:
92 return _wrap(SFRAME_COMMIT_PACK, {
93 "t": SFRAME_COMMIT_PACK,
94 "commits": commits,
95 "snapshots": snapshots,
96 })
97
98
99 def _end_frame(n_objects: int = 0, n_commits: int = 1) -> bytes:
100 return _wrap(SFRAME_END, {
101 "t": SFRAME_END,
102 "n_objects": n_objects,
103 "n_commits": n_commits,
104 })
105
106
107 def _make_commit(snapshot_id: str, branch: str = "main") -> JSONObject:
108 return {
109 "commit_id": _oid(f"commit-{snapshot_id}".encode()),
110 "parent_ids": [],
111 "snapshot_id": snapshot_id,
112 "branch": branch,
113 "message": "integrity test commit",
114 "author": "gabriel",
115 "committed_at": "2026-04-28T00:00:00+00:00",
116 "signature": "",
117 "signer_key_id": "",
118 "agent_id": "claude-code",
119 "model_id": "claude-sonnet-4-6",
120 "metadata": {},
121 }
122
123
124 def _make_snapshot(snap_id: str, manifest: StrDict) -> JSONObject:
125 return {"snapshot_id": snap_id, "manifest": manifest}
126
127
128 # ---------------------------------------------------------------------------
129 # I1 — when put() silently fails, storage must not report the object present
130 # ---------------------------------------------------------------------------
131
132 def test_i1_local_backend_put_is_durable(tmp_path: "Path") -> None: # type: ignore[name-defined] # noqa: F821
133 """LocalBackend.put() must persist bytes such that exists() returns True.
134
135 This test pins the contract: if put() returns a URI, exists() must be True
136 afterwards. A backend that violates this contract allows ghost DB rows.
137 """
138 import asyncio
139 from musehub.storage.backends import LocalBackend
140
141 backend = LocalBackend()
142 repo_root = tmp_path / "repos" / "gabriel" / "test"
143 raw = b"hello world"
144 oid = _oid(raw)
145
146 asyncio.run(backend.put(oid, raw, repo_root=repo_root))
147
148 assert asyncio.run(backend.exists(oid, repo_root=repo_root)) is True
149
150
151 # ---------------------------------------------------------------------------
152 # I2 — externally-referenced object in DB but absent from storage → ERROR
153 # ---------------------------------------------------------------------------
154
155 @pytest.mark.asyncio
156 async def test_i2_external_ref_in_db_but_missing_from_storage_is_rejected(
157 client: AsyncClient,
158 db_session: AsyncSession,
159 wire_headers: StrDict,
160 ) -> None:
161 """Push must be rejected when a snapshot references an object in the DB
162 but absent from actual storage.
163
164 Scenario:
165 - A previous push stored object X in DB (musehub_objects row exists).
166 - The R2/S3 bytes for X are gone — storage.exists() returns False.
167 - A new push sends a snapshot referencing X as an external object
168 (not included in this push's object frames).
169 Expected: push emits ERROR, not RESULT ok=True.
170
171 Note: request sessions are independent from the test session.
172 Test data must be committed (not just flushed) to be visible to the
173 push handler's session.
174 """
175 import musehub.db.musehub_models as db_models
176
177 repo = await create_repo(db_session, owner="test-user-wire", name="i2-external-missing")
178
179 # Plant a ghost DB row: object in musehub_objects, but NOT in storage.
180 # The tmp-path backend from conftest._tmp_objects_dir starts empty, so
181 # any object in DB but not pushed through the backend is a ghost.
182 ghost_raw = b"ghost object bytes - i2"
183 ghost_oid = _oid(ghost_raw)
184 db_session.add(db_models.MusehubObject(
185 object_id=ghost_oid,
186 path="ghost.py",
187 size_bytes=len(ghost_raw),
188 disk_path=f"objects/{ghost_oid[7:9]}/{ghost_oid[9:]}",
189 storage_uri=f"local://objects/{ghost_oid[7:9]}/{ghost_oid[9:]}",
190 ))
191 # Commit so the push handler's independent session can see this row.
192 await db_session.commit()
193
194 # Build a snapshot that references the ghost object externally.
195 snap_id = _oid(b"snap-i2")
196 manifest = {"ghost.py": ghost_oid}
197 commit = _make_commit(snapshot_id=snap_id)
198 snap = _make_snapshot(snap_id, manifest)
199
200 # Push with NO object frames — ghost_oid is external (not in this push).
201 body = (
202 _header_frame(n_objects=0, n_commits=1)
203 + _commit_pack_frame([commit], [snap])
204 + _end_frame(n_objects=0, n_commits=1)
205 )
206
207 resp = await client.post(
208 f"/{repo.owner}/{repo.slug}/push/stream",
209 content=body,
210 headers={**wire_headers, "Content-Type": WIRE_CONTENT_TYPE},
211 )
212
213 # The response may be 200 with embedded error frames or a 4xx.
214 assert resp.status_code in (200, 422), (
215 f"Unexpected status {resp.status_code}: {resp.text}"
216 )
217
218 if resp.status_code == 200:
219 unpacker = msgpack.Unpacker(raw=False)
220 unpacker.feed(resp.content)
221 frames = list(unpacker)
222 ok_true = [f for f in frames if f.get("ok") is True]
223 assert not ok_true, (
224 f"Push must not emit ok=True when a referenced object is in DB "
225 f"but missing from storage. Got frames: {frames}"
226 )
227 # 422 from the server is also a correct rejection.
228
229
230 # ---------------------------------------------------------------------------
231 # I3 — after successful push, all snapshot-referenced objects are in storage
232 # ---------------------------------------------------------------------------
233
234 @pytest.mark.asyncio
235 async def test_i3_successful_push_all_objects_retrievable(
236 client: AsyncClient,
237 db_session: AsyncSession,
238 wire_headers: StrDict,
239 ) -> None:
240 """Every object referenced by a snapshot must be in storage after a push.
241
242 This is the golden-path integration test: push a commit with a snapshot
243 whose manifest references one object, then verify backend.exists() returns
244 True for that object. Regression guard for the staging gap.
245 """
246 import musehub.services.musehub_wire as wire_svc
247
248 repo = await create_repo(db_session, owner="test-user-wire", name="i3-all-objects")
249
250 raw = b"real file bytes"
251 oid, obj_frame = _object_frame(raw, path="real.py")
252
253 snap_id = _oid(b"snap-i3")
254 manifest = {"real.py": oid}
255 commit = _make_commit(snapshot_id=snap_id)
256 snap = _make_snapshot(snap_id, manifest)
257
258 body = (
259 _header_frame(n_objects=1, n_commits=1)
260 + obj_frame
261 + _commit_pack_frame([commit], [snap])
262 + _end_frame(n_objects=1, n_commits=1)
263 )
264
265 resp = await client.post(
266 f"/{repo.owner}/{repo.slug}/push/stream",
267 content=body,
268 headers={**wire_headers, "Content-Type": WIRE_CONTENT_TYPE},
269 )
270
271 assert resp.status_code == 200
272 unpacker = msgpack.Unpacker(raw=False)
273 unpacker.feed(resp.content)
274 frames = list(unpacker)
275 ok_frames = [f for f in frames if f.get("ok") is True]
276 assert ok_frames, f"Expected ok=True, got: {frames}"
277
278 # All snapshot-referenced objects must now be in storage.
279 from musehub.storage.backends import repo_root_for
280 backend = wire_svc.get_backend()
281 _repo_root = repo_root_for(repo.owner, repo.slug)
282 for path, ref_oid in manifest.items():
283 assert await backend.exists(ref_oid, repo_root=_repo_root), (
284 f"Object {ref_oid} (path={path}) is referenced by snapshot manifest "
285 f"but missing from storage after successful push."
286 )
287
288
289 # ---------------------------------------------------------------------------
290 # I4 — static check: integrity check must use backend.exists(), not just DB
291 # ---------------------------------------------------------------------------
292
293 def test_i4_integrity_check_uses_backend_exists() -> None:
294 """The referential integrity check in wire_push_stream must call
295 backend.exists() for externally-referenced objects, not just query the DB.
296
297 Querying DB rows alone cannot detect the ghost-object scenario where a row
298 exists in musehub_objects but the storage bytes are absent. This is a
299 static assertion on the source of musehub_wire.py.
300 """
301 from musehub.services import musehub_wire
302
303 source = inspect.getsource(musehub_wire)
304
305 assert "backend.exists" in source, (
306 "musehub_wire.py referential integrity check must call backend.exists() "
307 "to verify objects are in actual storage, not just in the DB. "
308 "DB rows can exist without corresponding R2/S3 bytes (ghost objects). "
309 "Add: `if not await backend.exists(oid): ...` for externally-referenced objects."
310 )
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago