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