gabriel / musehub public
test_mpack_index_job.py python
382 lines 14.0 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 49 days ago
1 """TDD — mpack.index job: object writes, object refs, idempotency.
2
3 The one principle: Content-addressing is a proof, not a label.
4 sha256(wire_bytes) == mpack_key already authenticated every byte in the
5 mpack — including every object's content and its declared ID.
6
7 Invariants:
8 1. After job runs: musehub_objects has one row per object with mpack:// URI.
9 2. After job runs: musehub_object_refs has rows for every (repo_id, object_id).
10 3. No per-object sha256 re-verification — that is checking the work of the hash.
11 4. Job is idempotent — running it twice produces identical DB state.
12 5. Return dict includes mpack_index_written, elapsed_ms (and other counts).
13 """
14 from __future__ import annotations
15
16 import datetime
17 import hashlib
18 import pathlib
19
20 import msgpack
21 import pytest
22 import pytest_asyncio
23
24 from httpx import AsyncClient, ASGITransport
25 from sqlalchemy import func, select
26 from sqlalchemy.ext.asyncio import AsyncSession
27
28 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
29 from musehub.db.musehub_repo_models import MusehubObject, MusehubObjectRef, MusehubRepo
30 from musehub.db.database import get_db
31 from musehub.main import app
32
33 from muse.core.object_store import write_object
34 from muse.core.mpack import build_mpack
35 from muse.core.paths import muse_dir
36 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
37 from muse.core.commits import CommitRecord, write_commit
38 from muse.core.refs import write_branch_ref
39 from muse.core.snapshots import SnapshotRecord, write_snapshot
40 from muse.core.types import blob_id
41 from musehub.types.json_types import JSONObject
42
43
44 _AUTH_CTX = MSignContext(
45 handle="gabriel",
46 identity_id="sha256:" + "0" * 64,
47 is_agent=False,
48 is_admin=True,
49 )
50
51 # Small enough to be fast, large enough to exercise batching
52 _N_FILES = 30
53 _N_COMMITS = 15
54 _FILES_CHANGED = 3
55 _BLOB_SIZE = 256
56
57
58 # ── fixtures ───────────────────────────────────────────────────────────────
59
60 @pytest_asyncio.fixture()
61 async def client(db_session: AsyncSession) -> None:
62 async def _override_get_db() -> None:
63 yield db_session
64
65 app.dependency_overrides[get_db] = _override_get_db
66 app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX
67 app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX
68
69 async with AsyncClient(
70 transport=ASGITransport(app=app),
71 base_url="https://localhost:1337",
72 ) as c:
73 yield c
74
75 app.dependency_overrides.clear()
76
77
78 @pytest_asyncio.fixture()
79 async def repo(client: AsyncClient) -> None:
80 resp = await client.post(
81 "/api/repos",
82 json={"owner": "gabriel", "name": "mpack-index-test", "visibility": "public", "initialize": False},
83 )
84 assert resp.status_code in (200, 201), resp.text
85 data = resp.json()
86 yield data["slug"]
87 await client.delete(f"/api/repos/{data['repoId']}")
88
89
90 def _make_repo(tmp: pathlib.Path) -> tuple[pathlib.Path, str, dict]:
91 """Return (repo_path, head_commit_id, mpack_dict)."""
92 tmp.mkdir(parents=True, exist_ok=True)
93 dot = muse_dir(tmp)
94 dot.mkdir()
95 (dot / "repo.json").write_text('{"repo_id":"idx-test","owner":"gabriel"}')
96 for d in ("commits", "snapshots", "objects"):
97 (dot / d).mkdir()
98 (dot / "refs" / "heads").mkdir(parents=True)
99 (dot / "HEAD").write_text("ref: refs/heads/main\n")
100 (dot / "config.toml").write_text("")
101
102 blob_ids: list[str] = []
103 for i in range(_N_FILES):
104 data = f"base-{i:04d}".encode() + b"x" * _BLOB_SIZE
105 oid = blob_id(data)
106 write_object(tmp, oid, data)
107 blob_ids.append(oid)
108
109 base_manifest = {f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)}
110 parent = None
111 tip = ""
112 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
113
114 for i in range(_N_COMMITS):
115 manifest = dict(base_manifest)
116 for j in range(_FILES_CHANGED):
117 idx = (i * _FILES_CHANGED + j) % _N_FILES
118 raw = f"c{i:04d}-f{j}".encode() + b"y" * _BLOB_SIZE
119 oid = blob_id(raw)
120 write_object(tmp, oid, raw)
121 manifest[f"src/file_{idx:04d}.py"] = oid
122
123 sid = compute_snapshot_id(manifest)
124 write_snapshot(tmp, SnapshotRecord(snapshot_id=sid, manifest=manifest))
125 msg = f"commit-{i:05d}"
126 cid = compute_commit_id(
127 parent_ids=[parent] if parent else [],
128 snapshot_id=sid,
129 message=msg,
130 committed_at_iso=ts.isoformat(),
131 author="gabriel",
132 )
133 write_commit(tmp, CommitRecord(
134 commit_id=cid, branch="main",
135 snapshot_id=sid, message=msg, committed_at=ts,
136 parent_commit_id=parent, parent2_commit_id=None,
137 author="gabriel", metadata={}, structured_delta=None,
138 sem_ver_bump="none", breaking_changes=[],
139 agent_id="", model_id="", toolchain_id="",
140 prompt_hash="", signature="", signer_key_id="",
141 ))
142 parent = cid
143 tip = cid
144 ts += datetime.timedelta(seconds=60)
145
146 write_branch_ref(tmp, "main", tip)
147 mpack = build_mpack(tmp, [tip], have=[])
148 return tmp, tip, mpack
149
150
151 async def _push_mpack(
152 repo_slug: str,
153 mpack: JSONObject,
154 head: str,
155 db_session: AsyncSession,
156 ) -> str:
157 """Store mpack in MemoryBackend and create a mpack.index job row. Returns job_id.
158
159 Does NOT call push/unpack-mpack (that route enqueues background jobs and is
160 tested separately). Creates the job row directly so tests can call
161 process_mpack_index_job in isolation.
162 """
163 import musehub.storage.backends as _backends_mod
164 from datetime import datetime, timezone
165 from musehub.core.genesis import compute_job_id as _compute_job_id
166 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
167
168 repo_row = (await db_session.execute(
169 select(MusehubRepo).where(MusehubRepo.slug == repo_slug)
170 )).scalar_one()
171 repo_id = repo_row.repo_id
172
173 wire_bytes = msgpack.packb(mpack, use_bin_type=True)
174 mpack_key = "sha256:" + hashlib.sha256(wire_bytes).hexdigest()
175 n_objects = len(mpack.get("blobs") or [])
176
177 # Store directly in the MemoryBackend patched by the conftest autouse fixture.
178 backend = _backends_mod.get_backend()
179 await backend.put_mpack(mpack_key, wire_bytes)
180
181 now = datetime.now(tz=timezone.utc)
182 job_id = _compute_job_id(repo_id, "mpack.index", now.isoformat())
183 db_session.add(MusehubBackgroundJob(
184 job_id=job_id,
185 repo_id=repo_id,
186 job_type="mpack.index",
187 payload={
188 "mpack_key": mpack_key,
189 "pusher_id": "sha256:" + "0" * 64,
190 "branch": "main",
191 "head": head,
192 "force": False,
193 "declared_objects_count": n_objects,
194 },
195 status="pending",
196 created_at=now,
197 attempt=0,
198 ))
199 await db_session.flush()
200 return job_id
201
202
203 # ── Phase 1 tests ──────────────────────────────────────────────────────────
204
205 @pytest.mark.asyncio
206 async def test_objects_written_to_minio_after_job(
207 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
208 ) -> None:
209 """After mpack.index runs: every object blob is retrievable from MinIO by object_id.
210
211 This is the load-bearing test. Fetch/clone/pull are all blocked until
212 objects are at their sha256 addresses in storage.
213 """
214 _, head, mpack = _make_repo(tmp_path / "repo")
215 job_id = await _push_mpack(repo, mpack, head, db_session)
216
217 from musehub.services.musehub_wire import process_mpack_index_job
218 result = await process_mpack_index_job(db_session, job_id)
219 await db_session.commit()
220
221 raw_objects = mpack.get("blobs") or []
222 assert raw_objects, "mpack must contain objects"
223
224 # Phase 2: objects are stored via mpack URI (no individual MinIO writes).
225 # Verify musehub_objects rows exist and storage_uri points to the mpack.
226 all_oids = [obj["object_id"] for obj in raw_objects]
227 rows = (await db_session.execute(
228 select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
229 )).scalars().all()
230 row_map = {r.object_id: r for r in rows}
231
232 missing = [oid for oid in all_oids if oid not in row_map]
233 assert not missing, (
234 f"{len(missing)}/{len(raw_objects)} objects have no musehub_objects row after job: "
235 f"{missing[:3]}"
236 )
237 bad_uri = [oid for oid in all_oids if not (row_map[oid].storage_uri or "").startswith("mpack://")]
238 assert not bad_uri, (
239 f"{len(bad_uri)} objects have unexpected storage_uri (expected mpack://...): "
240 + str({oid: row_map[oid].storage_uri for oid in bad_uri[:3]})
241 )
242
243
244 @pytest.mark.asyncio
245 async def test_musehub_objects_rows_after_job(
246 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
247 ) -> None:
248 """After mpack.index runs: musehub_objects has one row per object."""
249 _, head, mpack = _make_repo(tmp_path / "repo")
250 job_id = await _push_mpack(repo, mpack, head, db_session)
251
252 from musehub.services.musehub_wire import process_mpack_index_job
253 result = await process_mpack_index_job(db_session, job_id)
254 await db_session.commit()
255
256 raw_objects = mpack.get("blobs") or []
257 all_oids = [obj["object_id"] for obj in raw_objects]
258
259 rows = (await db_session.execute(
260 select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
261 )).scalars().all()
262
263 assert len(rows) == len(all_oids), (
264 f"expected {len(all_oids)} musehub_objects rows, got {len(rows)}"
265 )
266 # mpack_index_written = number of MusehubMPackIndex rows upserted (one per object).
267 assert result["mpack_index_written"] == len(all_oids), result
268
269
270 @pytest.mark.asyncio
271 async def test_object_refs_after_job(
272 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
273 ) -> None:
274 """After mpack.index runs: musehub_object_refs links every object to this repo."""
275 _, head, mpack = _make_repo(tmp_path / "repo")
276 job_id = await _push_mpack(repo, mpack, head, db_session)
277
278 from musehub.services.musehub_wire import process_mpack_index_job
279 result = await process_mpack_index_job(db_session, job_id)
280 await db_session.commit()
281
282 repo_row = (await db_session.execute(
283 select(MusehubRepo).where(MusehubRepo.slug == repo)
284 )).scalar_one()
285
286 ref_count = (await db_session.execute(
287 select(func.count()).select_from(MusehubObjectRef).where(
288 MusehubObjectRef.repo_id == repo_row.repo_id
289 )
290 )).scalar_one()
291
292 raw_objects = mpack.get("blobs") or []
293 assert ref_count == len(raw_objects), (
294 f"expected {len(raw_objects)} object_refs rows, got {ref_count}"
295 )
296
297
298 @pytest.mark.asyncio
299 async def test_no_per_object_minio_writes(
300 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
301 ) -> None:
302 """mpack.index must not write per-object MinIO keys.
303
304 sha256(wire_bytes) == mpack_key already authenticated every byte in the
305 mpack. Objects are served from the covering mpack via mpack:// URIs;
306 individual MinIO keys are never written.
307 """
308 _, head, mpack = _make_repo(tmp_path / "repo")
309 job_id = await _push_mpack(repo, mpack, head, db_session)
310
311 from musehub.services.musehub_wire import process_mpack_index_job
312 await process_mpack_index_job(db_session, job_id)
313 await db_session.commit()
314
315 raw_objects = mpack.get("blobs") or []
316 all_oids = [obj["object_id"] for obj in raw_objects]
317 rows = (await db_session.execute(
318 select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
319 )).scalars().all()
320
321 bad_uri = [r for r in rows if not (r.storage_uri or "").startswith("mpack://")]
322 assert not bad_uri, (
323 f"{len(bad_uri)} objects have unexpected storage_uri (expected mpack://...): "
324 + str({r.object_id: r.storage_uri for r in bad_uri[:3]})
325 )
326
327
328 @pytest.mark.asyncio
329 async def test_job_is_idempotent(
330 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
331 ) -> None:
332 """Running mpack.index twice produces identical DB state to running it once.
333
334 Worker crashes, retries, and duplicate enqueues must all be safe.
335 """
336 _, head, mpack = _make_repo(tmp_path / "repo")
337 job_id = await _push_mpack(repo, mpack, head, db_session)
338
339 from musehub.services.musehub_wire import process_mpack_index_job
340
341 r1 = await process_mpack_index_job(db_session, job_id)
342 await db_session.commit()
343
344 r2 = await process_mpack_index_job(db_session, job_id)
345 await db_session.commit()
346
347 repo_row = (await db_session.execute(
348 select(MusehubRepo).where(MusehubRepo.slug == repo)
349 )).scalar_one()
350
351 obj_count = (await db_session.execute(
352 select(func.count()).select_from(MusehubObjectRef).where(
353 MusehubObjectRef.repo_id == repo_row.repo_id
354 )
355 )).scalar_one()
356
357 raw_objects = mpack.get("blobs") or []
358 assert obj_count == len(raw_objects), (
359 f"after two runs: expected {len(raw_objects)} object_refs, got {obj_count} "
360 "(idempotency failure — rows were doubled)"
361 )
362
363
364 @pytest.mark.asyncio
365 async def test_return_dict_includes_counts_and_timing(
366 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
367 ) -> None:
368 """Return dict contains objects_written, object_refs_written, elapsed_ms."""
369 _, head, mpack = _make_repo(tmp_path / "repo")
370 job_id = await _push_mpack(repo, mpack, head, db_session)
371
372 from musehub.services.musehub_wire import process_mpack_index_job
373 result = await process_mpack_index_job(db_session, job_id)
374 await db_session.commit()
375
376 assert "mpack_index_written" in result, result
377 assert "commit_graph_written" in result, result
378 assert "mpack_size_bytes" in result, result
379 assert "elapsed_ms" in result, result
380 assert result["mpack_index_written"] >= 0, result
381 assert result["mpack_size_bytes"] > 0, result
382 assert result["elapsed_ms"] > 0, result
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 49 days ago