gabriel / musehub public
test_mpack_index_job_phase3.py python
288 lines 11.1 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 49 days ago
1 """Phase 3 mpack unpack invariants — idempotency and cold-start safety.
2
3 wire_push_unpack_mpack is synchronous and inline (no background job for the
4 sync path itself; commit-graph indexing is a separate background job).
5 These tests verify the invariants that matter for the current design:
6
7 1. Idempotency — calling unpack-mpack twice with the same mpack produces
8 no duplicate rows and leaves the DB in the correct final state.
9 2. Correctness — a single unpack-mpack call writes all commits, snapshots,
10 and blobs with the right storage_uri and counts.
11 3. Cold-start — mpack already in storage (e.g. after a crash before DB
12 writes) is correctly handled by a fresh unpack-mpack call.
13 """
14 from __future__ import annotations
15
16 import datetime
17 import pathlib
18
19 import pytest
20 import pytest_asyncio
21
22 from httpx import AsyncClient, ASGITransport
23 from sqlalchemy import select
24 from sqlalchemy.ext.asyncio import AsyncSession
25
26 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
27 from musehub.db.musehub_repo_models import MusehubObject, MusehubCommit, MusehubSnapshot
28 from musehub.db.database import get_db
29 from musehub.main import app
30
31 from muse.core.object_store import write_object
32 from muse.core.mpack import build_mpack, build_wire_mpack
33 from muse.core.paths import muse_dir
34 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
35 from muse.core.commits import CommitRecord, write_commit
36 from muse.core.refs import write_branch_ref
37 from muse.core.snapshots import SnapshotRecord, write_snapshot
38 from muse.core.types import blob_id
39 from musehub.types.json_types import JSONObject
40
41
42 _AUTH_CTX = MSignContext(
43 handle="gabriel",
44 identity_id="sha256:" + "0" * 64,
45 is_agent=False,
46 is_admin=True,
47 )
48
49 _N_FILES = 8
50 _N_COMMITS = 4
51 _FILES_CHANGED = 2
52 _BLOB_SIZE = 128
53
54
55 # ── fixtures ────────────────────────────────────────────────────────────────
56
57 @pytest_asyncio.fixture()
58 async def client(db_session: AsyncSession) -> None:
59 async def _override_get_db() -> None:
60 yield db_session
61
62 app.dependency_overrides[get_db] = _override_get_db
63 app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX
64 app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX
65
66 async with AsyncClient(
67 transport=ASGITransport(app=app),
68 base_url="https://localhost:1337",
69 ) as c:
70 yield c
71
72 app.dependency_overrides.clear()
73
74
75 @pytest_asyncio.fixture()
76 async def repo(client: AsyncClient) -> None:
77 resp = await client.post(
78 "/api/repos",
79 json={"owner": "gabriel", "name": "phase3-retry-test", "visibility": "public", "initialize": False},
80 )
81 assert resp.status_code in (200, 201), resp.text
82 data = resp.json()
83 yield data["slug"]
84 await client.delete(f"/api/repos/{data['repoId']}")
85
86
87 def _make_repo(tmp: pathlib.Path) -> tuple[pathlib.Path, str, bytes, dict]:
88 """Build a local repo and return (path, head_commit_id, wire_bytes, raw_mpack)."""
89 tmp.mkdir(parents=True, exist_ok=True)
90 dot = muse_dir(tmp)
91 dot.mkdir()
92 (dot / "repo.json").write_text('{"repo_id":"phase3-test","owner":"gabriel"}')
93 for d in ("commits", "snapshots", "objects"):
94 (dot / d).mkdir()
95 (dot / "refs" / "heads").mkdir(parents=True)
96 (dot / "HEAD").write_text("ref: refs/heads/main\n")
97 (dot / "config.toml").write_text("")
98
99 blob_ids: list[str] = []
100 for i in range(_N_FILES):
101 data = f"base-{i:04d}".encode() + b"x" * _BLOB_SIZE
102 oid = blob_id(data)
103 write_object(tmp, oid, data)
104 blob_ids.append(oid)
105
106 base_manifest = {f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)}
107 parent = None
108 tip = ""
109 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
110
111 for i in range(_N_COMMITS):
112 manifest = dict(base_manifest)
113 for j in range(_FILES_CHANGED):
114 idx = (i * _FILES_CHANGED + j) % _N_FILES
115 raw = f"c{i:04d}-f{j}".encode() + b"y" * _BLOB_SIZE
116 oid = blob_id(raw)
117 write_object(tmp, oid, raw)
118 manifest[f"src/file_{idx:04d}.py"] = oid
119
120 sid = compute_snapshot_id(manifest)
121 write_snapshot(tmp, SnapshotRecord(snapshot_id=sid, manifest=manifest))
122 msg = f"commit-{i:05d}"
123 cid = compute_commit_id(
124 parent_ids=[parent] if parent else [],
125 snapshot_id=sid,
126 message=msg,
127 committed_at_iso=ts.isoformat(),
128 author="gabriel",
129 )
130 write_commit(tmp, CommitRecord(
131 commit_id=cid, branch="main",
132 snapshot_id=sid, message=msg, committed_at=ts,
133 parent_commit_id=parent, parent2_commit_id=None,
134 author="gabriel", metadata={}, structured_delta=None,
135 sem_ver_bump="none", breaking_changes=[],
136 agent_id="", model_id="", toolchain_id="",
137 prompt_hash="", signature="", signer_key_id="",
138 ))
139 parent = cid
140 tip = cid
141 ts += datetime.timedelta(seconds=60)
142
143 write_branch_ref(tmp, "main", tip)
144 raw_mpack = build_mpack(tmp, [tip], have=[])
145 wire_bytes = build_wire_mpack(raw_mpack)
146 return tmp, tip, wire_bytes, raw_mpack
147
148
149 async def _push_and_unpack(client: AsyncClient, repo_slug: str, wire_bytes: bytes, head: str) -> JSONObject:
150 """Store mpack in MemoryBackend and call unpack-mpack. Returns the response dict.
151
152 Uses backend.put_mpack() directly instead of the presign+httpx PUT pattern —
153 the MemoryBackend patched by conftest has no real presigned URL infrastructure.
154 wire_bytes must be MUSE binary format (starts with b"MUSE").
155 mpack_key uses blob_id (sha256("blob <size>\\0<data>")) — same formula the
156 server's integrity check uses at step 2.
157 """
158 import musehub.storage.backends as _backends_mod
159 import msgpack as _msgpack
160 from muse.core.types import blob_id as _blob_id
161
162 mpack_key = _blob_id(wire_bytes)
163 backend = _backends_mod.get_backend()
164 await backend.put_mpack(mpack_key, wire_bytes)
165
166 ur = await client.post(
167 f"/gabriel/{repo_slug}/push/unpack-mpack",
168 content=_msgpack.packb(
169 {"mpack_key": mpack_key, "branch": "main", "head": head},
170 use_bin_type=True,
171 ),
172 headers={"Content-Type": "application/x-msgpack"},
173 )
174 assert ur.status_code == 200, ur.text
175 return ur.json()
176
177
178 # ── tests ────────────────────────────────────────────────────────────────────
179
180 @pytest.mark.asyncio
181 async def test_unpack_mpack_idempotent(
182 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
183 ) -> None:
184 """Calling unpack-mpack twice with the same mpack produces no duplicate rows.
185
186 wire_push_unpack_mpack uses ON CONFLICT DO NOTHING / ON CONFLICT DO UPDATE
187 throughout, so a second identical push must leave the DB in the same state
188 as a single push — no extra rows, no constraint violations.
189 """
190 _, head, wire_bytes, raw_mpack = _make_repo(tmp_path / "repo")
191
192 result1 = await _push_and_unpack(client, repo, wire_bytes, head)
193 assert result1.get("commits_written", 0) == _N_COMMITS, result1
194
195 # Second push — must not raise and must not duplicate rows.
196 result2 = await _push_and_unpack(client, repo, wire_bytes, head)
197 assert result2.get("commits_written", 0) == 0, (
198 "second push of same mpack must write 0 new commits (all already exist)"
199 )
200
201 all_oids = [obj["object_id"] for obj in (raw_mpack.get("blobs") or [])]
202 rows = (await db_session.execute(
203 select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
204 )).scalars().all()
205 assert len(rows) == len(all_oids), (
206 f"expected exactly {len(all_oids)} object rows after double push, got {len(rows)}"
207 )
208
209
210 @pytest.mark.asyncio
211 async def test_unpack_mpack_writes_all_entities(
212 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
213 ) -> None:
214 """A single unpack-mpack call correctly writes all commits, snapshots, and blobs.
215
216 Verifies counts in the response and presence in the DB with correct storage_uri.
217 """
218 _, head, wire_bytes, raw_mpack = _make_repo(tmp_path / "repo")
219 result = await _push_and_unpack(client, repo, wire_bytes, head)
220
221 assert result.get("commits_written") == _N_COMMITS, (
222 f"expected {_N_COMMITS} commits_written, got {result}"
223 )
224 assert result.get("snapshots_written") == _N_COMMITS, (
225 f"expected {_N_COMMITS} snapshots_written (one per commit), got {result}"
226 )
227 n_blobs = len(raw_mpack.get("blobs") or [])
228 assert result.get("blobs_written") == n_blobs, (
229 f"expected {n_blobs} blobs_written, got {result}"
230 )
231
232 all_oids = [obj["object_id"] for obj in (raw_mpack.get("blobs") or [])]
233 rows = (await db_session.execute(
234 select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
235 )).scalars().all()
236 assert len(rows) == n_blobs, (
237 f"expected {n_blobs} musehub_objects rows, got {len(rows)}"
238 )
239 # All objects must have a real storage_uri — nothing pending.
240 bad = [r for r in rows if r.storage_uri in (None, "pending")]
241 assert not bad, f"{len(bad)} objects still have storage_uri='pending'"
242
243
244 @pytest.mark.asyncio
245 async def test_unpack_mpack_cold_start(
246 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
247 ) -> None:
248 """unpack-mpack succeeds when mpack is already in storage with no prior DB record.
249
250 Simulates the crash-then-retry path: mpack bytes landed in storage but the
251 DB transaction was rolled back before unpack-mpack ran. A cold call to
252 unpack-mpack must write all entities correctly.
253 """
254 import musehub.storage.backends as _backends_mod
255 import msgpack as _msgpack
256
257 _, head, wire_bytes, raw_mpack = _make_repo(tmp_path / "repo")
258
259 # Store mpack directly — simulates completed PUT with no subsequent DB writes.
260 from muse.core.types import blob_id as _blob_id
261 mpack_key = _blob_id(wire_bytes)
262 backend = _backends_mod.get_backend()
263 await backend.put_mpack(mpack_key, wire_bytes)
264
265 # Cold unpack-mpack — no prior presign row, no prior unpack attempt in DB.
266 ur = await client.post(
267 f"/gabriel/{repo}/push/unpack-mpack",
268 content=_msgpack.packb(
269 {"mpack_key": mpack_key, "branch": "main", "head": head},
270 use_bin_type=True,
271 ),
272 headers={"Content-Type": "application/x-msgpack"},
273 )
274 assert ur.status_code == 200, ur.text
275 result = ur.json()
276
277 assert result.get("commits_written") == _N_COMMITS, (
278 f"expected {_N_COMMITS} commits after cold start, got {result}"
279 )
280
281 n_blobs = len(raw_mpack.get("blobs") or [])
282 all_oids = [obj["object_id"] for obj in (raw_mpack.get("blobs") or [])]
283 rows = (await db_session.execute(
284 select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
285 )).scalars().all()
286 assert len(rows) == n_blobs, (
287 f"expected {n_blobs} objects after cold start, got {len(rows)}"
288 )
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 49 days ago