test_mpack_fetch_phase3.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """TDD — Phase 3: wire_fetch_mpack e2e with indexed mpack state (issue #63). |
| 2 | |
| 3 | PF-1 wire_fetch_mpack with force_build=True returns a presigned mpack_url and |
| 4 | correct commit/blob counts after a real HTTP push + index job. |
| 5 | |
| 6 | PF-2 Delta fetch (have=[tip_a], want=[tip_b]) returns only new commits — |
| 7 | commit_count is less than the total commits in the repo. |
| 8 | |
| 9 | PF-3 Without force_build, a cold cache raises MPackNotReadyError — the |
| 10 | production behaviour requiring the prebuild job to warm the cache first. |
| 11 | """ |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import datetime |
| 15 | import pathlib |
| 16 | |
| 17 | import msgpack |
| 18 | import pytest |
| 19 | import pytest_asyncio |
| 20 | from httpx import AsyncClient, ASGITransport |
| 21 | from sqlalchemy.ext.asyncio import AsyncSession |
| 22 | |
| 23 | from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request |
| 24 | from musehub.db.database import get_db |
| 25 | from musehub.main import app |
| 26 | |
| 27 | from muse.core.object_store import write_object |
| 28 | from muse.core.mpack import build_mpack, build_wire_mpack |
| 29 | from muse.core.paths import muse_dir |
| 30 | from muse.core.snapshot import compute_commit_id, compute_snapshot_id |
| 31 | from muse.core.commits import CommitRecord, write_commit |
| 32 | from muse.core.refs import write_branch_ref |
| 33 | from muse.core.snapshots import SnapshotRecord, write_snapshot |
| 34 | from muse.core.types import blob_id |
| 35 | |
| 36 | pytestmark = pytest.mark.wire |
| 37 | |
| 38 | |
| 39 | _AUTH_CTX = MSignContext( |
| 40 | handle="gabriel", |
| 41 | identity_id="sha256:" + "0" * 64, |
| 42 | is_agent=False, |
| 43 | is_admin=True, |
| 44 | ) |
| 45 | |
| 46 | _N_FILES = 6 |
| 47 | _BLOB_SIZE = 64 |
| 48 | |
| 49 | |
| 50 | # ── fixtures ────────────────────────────────────────────────────────────────── |
| 51 | |
| 52 | @pytest_asyncio.fixture() |
| 53 | async def client(db_session: AsyncSession) -> None: |
| 54 | async def _override_get_db() -> None: |
| 55 | yield db_session |
| 56 | |
| 57 | app.dependency_overrides[get_db] = _override_get_db |
| 58 | app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX |
| 59 | app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX |
| 60 | |
| 61 | async with AsyncClient( |
| 62 | transport=ASGITransport(app=app), |
| 63 | base_url="https://localhost:1337", |
| 64 | ) as c: |
| 65 | yield c |
| 66 | |
| 67 | app.dependency_overrides.clear() |
| 68 | |
| 69 | |
| 70 | @pytest_asyncio.fixture() |
| 71 | async def repo(client: AsyncClient) -> None: |
| 72 | resp = await client.post( |
| 73 | "/api/repos", |
| 74 | json={"owner": "gabriel", "name": "fetch-phase3-test", "visibility": "public", "initialize": False}, |
| 75 | ) |
| 76 | assert resp.status_code in (200, 201), resp.text |
| 77 | data = resp.json() |
| 78 | yield data |
| 79 | await client.delete(f"/api/repos/{data['repoId']}") |
| 80 | |
| 81 | |
| 82 | # ── helpers ─────────────────────────────────────────────────────────────────── |
| 83 | |
| 84 | def _make_local_repo(tmp: pathlib.Path) -> pathlib.Path: |
| 85 | tmp.mkdir(parents=True, exist_ok=True) |
| 86 | dot = muse_dir(tmp) |
| 87 | dot.mkdir() |
| 88 | (dot / "repo.json").write_text('{"repo_id":"fetch-phase3","owner":"gabriel"}') |
| 89 | for d in ("commits", "snapshots", "objects"): |
| 90 | (dot / d).mkdir() |
| 91 | (dot / "refs" / "heads").mkdir(parents=True) |
| 92 | (dot / "HEAD").write_text("ref: refs/heads/main\n") |
| 93 | (dot / "config.toml").write_text("") |
| 94 | return tmp |
| 95 | |
| 96 | |
| 97 | def _populate(repo: pathlib.Path, n_commits: int, seed: str = "pf") -> list[str]: |
| 98 | """Populate local repo with n_commits and return all commit IDs in order.""" |
| 99 | blob_ids: list[str] = [] |
| 100 | for i in range(_N_FILES): |
| 101 | data = f"{seed}-base-{i:04d}".encode() + b"x" * _BLOB_SIZE |
| 102 | oid = blob_id(data) |
| 103 | write_object(repo, 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 | ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) |
| 109 | commit_ids: list[str] = [] |
| 110 | |
| 111 | for i in range(n_commits): |
| 112 | manifest = dict(base_manifest) |
| 113 | raw = f"{seed}-c{i:04d}".encode() + b"y" * _BLOB_SIZE |
| 114 | oid = blob_id(raw) |
| 115 | write_object(repo, oid, raw) |
| 116 | manifest[f"src/file_{i % _N_FILES:04d}.py"] = oid |
| 117 | |
| 118 | sid = compute_snapshot_id(manifest) |
| 119 | write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest)) |
| 120 | msg = f"commit-{i:05d}" |
| 121 | cid = compute_commit_id( |
| 122 | parent_ids=[parent] if parent else [], |
| 123 | snapshot_id=sid, |
| 124 | message=msg, |
| 125 | committed_at_iso=ts.isoformat(), |
| 126 | author="gabriel", |
| 127 | ) |
| 128 | write_commit(repo, CommitRecord( |
| 129 | commit_id=cid, branch="main", |
| 130 | snapshot_id=sid, message=msg, committed_at=ts, |
| 131 | parent_commit_id=parent, parent2_commit_id=None, |
| 132 | author="gabriel", metadata={}, structured_delta=None, |
| 133 | sem_ver_bump="none", breaking_changes=[], |
| 134 | agent_id="", model_id="", toolchain_id="", |
| 135 | prompt_hash="", signature="", signer_key_id="", |
| 136 | )) |
| 137 | parent = cid |
| 138 | commit_ids.append(cid) |
| 139 | ts += datetime.timedelta(seconds=60) |
| 140 | |
| 141 | tip = commit_ids[-1] |
| 142 | write_branch_ref(repo, "main", tip) |
| 143 | return commit_ids |
| 144 | |
| 145 | |
| 146 | async def _http_push( |
| 147 | client: AsyncClient, |
| 148 | repo_slug: str, |
| 149 | local_repo: pathlib.Path, |
| 150 | tip: str, |
| 151 | have: list[str] | None = None, |
| 152 | ) -> tuple[str, dict]: |
| 153 | """Push via HTTP unpack-mpack. Returns (mpack_key, mpack_dict).""" |
| 154 | import musehub.storage.backends as _backends_mod |
| 155 | from muse.core.types import blob_id as _blob_id |
| 156 | |
| 157 | mpack_dict = build_mpack(local_repo, [tip], have=have or []) |
| 158 | wire_bytes = build_wire_mpack(mpack_dict) |
| 159 | mpack_key = _blob_id(wire_bytes) |
| 160 | |
| 161 | backend = _backends_mod.get_backend() |
| 162 | await backend.put_mpack(mpack_key, wire_bytes) |
| 163 | |
| 164 | n_commits = len(mpack_dict.get("commits") or []) |
| 165 | n_blobs = len(mpack_dict.get("blobs") or []) |
| 166 | |
| 167 | resp = await client.post( |
| 168 | f"/gabriel/{repo_slug}/push/unpack-mpack", |
| 169 | content=msgpack.packb({ |
| 170 | "mpack_key": mpack_key, |
| 171 | "branch": "main", |
| 172 | "head": tip, |
| 173 | "commits_count": n_commits, |
| 174 | "blobs_count": n_blobs, |
| 175 | }, use_bin_type=True), |
| 176 | headers={"Content-Type": "application/x-msgpack"}, |
| 177 | ) |
| 178 | assert resp.status_code == 200, f"push failed: {resp.status_code} {resp.text}" |
| 179 | return mpack_key, mpack_dict |
| 180 | |
| 181 | |
| 182 | async def _run_index_job( |
| 183 | db_session: AsyncSession, |
| 184 | repo_id: str, |
| 185 | mpack_key: str, |
| 186 | tip: str, |
| 187 | n_blobs: int, |
| 188 | ) -> None: |
| 189 | """Create a mpack.index job row and run process_mpack_index_job.""" |
| 190 | from datetime import datetime, timezone |
| 191 | from musehub.core.genesis import compute_job_id |
| 192 | from musehub.db.musehub_jobs_models import MusehubBackgroundJob |
| 193 | from musehub.services.musehub_wire import process_mpack_index_job |
| 194 | |
| 195 | now = datetime.now(tz=timezone.utc) |
| 196 | job_id = compute_job_id(repo_id, "mpack.index", now.isoformat()) |
| 197 | db_session.add(MusehubBackgroundJob( |
| 198 | job_id=job_id, |
| 199 | repo_id=repo_id, |
| 200 | job_type="mpack.index", |
| 201 | payload={ |
| 202 | "mpack_key": mpack_key, |
| 203 | "pusher_id": "sha256:" + "0" * 64, |
| 204 | "branch": "main", |
| 205 | "head": tip, |
| 206 | "force": False, |
| 207 | "declared_objects_count": n_blobs, |
| 208 | }, |
| 209 | status="pending", |
| 210 | )) |
| 211 | await db_session.flush() |
| 212 | await process_mpack_index_job(db_session, job_id) |
| 213 | await db_session.commit() |
| 214 | |
| 215 | |
| 216 | # ── PF-1 ────────────────────────────────────────────────────────────────────── |
| 217 | |
| 218 | @pytest.mark.asyncio |
| 219 | async def test_pf1_fetch_returns_mpack_url( |
| 220 | client: AsyncClient, repo: dict, tmp_path: pathlib.Path, db_session: AsyncSession, |
| 221 | ) -> None: |
| 222 | """wire_fetch_mpack returns a presigned mpack_url after push + index. |
| 223 | |
| 224 | force_build=True bypasses the prebuild cache so the fetch path is |
| 225 | exercised end-to-end without needing the prebuild job. |
| 226 | """ |
| 227 | from musehub.services.musehub_wire import wire_fetch_mpack |
| 228 | |
| 229 | local_repo = _make_local_repo(tmp_path / "repo") |
| 230 | commit_ids = _populate(local_repo, n_commits=3) |
| 231 | tip = commit_ids[-1] |
| 232 | |
| 233 | mpack_key, mpack_dict = await _http_push(client, repo["slug"], local_repo, tip) |
| 234 | n_blobs = len(mpack_dict.get("blobs") or []) |
| 235 | await _run_index_job(db_session, repo["repoId"], mpack_key, tip, n_blobs) |
| 236 | |
| 237 | result = await wire_fetch_mpack( |
| 238 | db_session, repo["repoId"], |
| 239 | want=[tip], have=[], |
| 240 | ttl_seconds=3600, |
| 241 | force_build=True, |
| 242 | ) |
| 243 | |
| 244 | assert result["mpack_url"] is not None, "mpack_url must be set after a successful fetch" |
| 245 | assert result["mpack_id"] is not None, "mpack_id must be set" |
| 246 | assert result["commit_count"] > 0, ( |
| 247 | f"expected commit_count > 0, got {result['commit_count']}" |
| 248 | ) |
| 249 | |
| 250 | |
| 251 | # ── PF-2 ────────────────────────────────────────────────────────────────────── |
| 252 | |
| 253 | @pytest.mark.asyncio |
| 254 | async def test_pf2_delta_fetch_returns_only_new_commits( |
| 255 | client: AsyncClient, repo: dict, tmp_path: pathlib.Path, db_session: AsyncSession, |
| 256 | ) -> None: |
| 257 | """Delta fetch with have=[tip_a] returns only the commits not yet on the client. |
| 258 | |
| 259 | Pushes 3 commits in one mpack, runs the index job, then fetches with |
| 260 | have=[first_commit] — the response should cover only commits 2 and 3, |
| 261 | not commit 1. |
| 262 | """ |
| 263 | from musehub.services.musehub_wire import wire_fetch_mpack |
| 264 | |
| 265 | local_repo = _make_local_repo(tmp_path / "repo") |
| 266 | commit_ids = _populate(local_repo, n_commits=3) |
| 267 | tip_a = commit_ids[0] # client already has this |
| 268 | tip_b = commit_ids[-1] # client wants this |
| 269 | |
| 270 | mpack_key, mpack_dict = await _http_push(client, repo["slug"], local_repo, tip_b) |
| 271 | n_blobs = len(mpack_dict.get("blobs") or []) |
| 272 | await _run_index_job(db_session, repo["repoId"], mpack_key, tip_b, n_blobs) |
| 273 | |
| 274 | result = await wire_fetch_mpack( |
| 275 | db_session, repo["repoId"], |
| 276 | want=[tip_b], have=[tip_a], |
| 277 | ttl_seconds=3600, |
| 278 | force_build=True, |
| 279 | ) |
| 280 | |
| 281 | assert result["mpack_url"] is not None, "delta fetch must return a mpack_url" |
| 282 | total_commits = len(commit_ids) |
| 283 | assert result["commit_count"] < total_commits, ( |
| 284 | f"delta fetch must return fewer than {total_commits} commits, " |
| 285 | f"got {result['commit_count']}" |
| 286 | ) |
| 287 | |
| 288 | |
| 289 | # ── PF-3 ────────────────────────────────────────────────────────────────────── |
| 290 | |
| 291 | @pytest.mark.asyncio |
| 292 | async def test_pf3_cold_cache_raises_mpack_not_ready( |
| 293 | client: AsyncClient, repo: dict, tmp_path: pathlib.Path, db_session: AsyncSession, |
| 294 | ) -> None: |
| 295 | """wire_fetch_mpack with have=[] and force_build=False raises MPackNotReadyError. |
| 296 | |
| 297 | In production, the prebuild job warms the cache before the first clone. |
| 298 | Without force_build, a cache miss is a signal that the mpack is not yet |
| 299 | ready and the client should retry. |
| 300 | """ |
| 301 | from musehub.services.musehub_wire_shared import MPackNotReadyError |
| 302 | from musehub.services.musehub_wire import wire_fetch_mpack |
| 303 | |
| 304 | local_repo = _make_local_repo(tmp_path / "repo") |
| 305 | commit_ids = _populate(local_repo, n_commits=2) |
| 306 | tip = commit_ids[-1] |
| 307 | |
| 308 | # Push so commits exist in the DB, but do NOT warm the prebuild cache. |
| 309 | mpack_key, mpack_dict = await _http_push(client, repo["slug"], local_repo, tip) |
| 310 | |
| 311 | with pytest.raises(MPackNotReadyError): |
| 312 | await wire_fetch_mpack( |
| 313 | db_session, repo["repoId"], |
| 314 | want=[tip], have=[], |
| 315 | ttl_seconds=3600, |
| 316 | force_build=False, |
| 317 | ) |