"""TDD — Phase 3: wire_fetch_mpack e2e with indexed mpack state (issue #63). PF-1 wire_fetch_mpack with force_build=True returns a presigned mpack_url and correct commit/blob counts after a real HTTP push + index job. PF-2 Delta fetch (have=[tip_a], want=[tip_b]) returns only new commits — commit_count is less than the total commits in the repo. PF-3 Without force_build, a cold cache raises MPackNotReadyError — the production behaviour requiring the prebuild job to warm the cache first. """ from __future__ import annotations import datetime import pathlib import msgpack import pytest import pytest_asyncio from httpx import AsyncClient, ASGITransport from sqlalchemy.ext.asyncio import AsyncSession from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request from musehub.db.database import get_db from musehub.main import app from muse.core.object_store import write_object from muse.core.mpack import build_mpack, build_wire_mpack from muse.core.paths import muse_dir from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.commits import CommitRecord, write_commit from muse.core.refs import write_branch_ref from muse.core.snapshots import SnapshotRecord, write_snapshot from muse.core.types import blob_id pytestmark = pytest.mark.wire _AUTH_CTX = MSignContext( handle="gabriel", identity_id="sha256:" + "0" * 64, is_agent=False, is_admin=True, ) _N_FILES = 6 _BLOB_SIZE = 64 # ── fixtures ────────────────────────────────────────────────────────────────── @pytest_asyncio.fixture() async def client(db_session: AsyncSession) -> None: async def _override_get_db() -> None: yield db_session app.dependency_overrides[get_db] = _override_get_db app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX async with AsyncClient( transport=ASGITransport(app=app), base_url="https://localhost:1337", ) as c: yield c app.dependency_overrides.clear() @pytest_asyncio.fixture() async def repo(client: AsyncClient) -> None: resp = await client.post( "/api/repos", json={"owner": "gabriel", "name": "fetch-phase3-test", "visibility": "public", "initialize": False}, ) assert resp.status_code in (200, 201), resp.text data = resp.json() yield data await client.delete(f"/api/repos/{data['repoId']}") # ── helpers ─────────────────────────────────────────────────────────────────── def _make_local_repo(tmp: pathlib.Path) -> pathlib.Path: tmp.mkdir(parents=True, exist_ok=True) dot = muse_dir(tmp) dot.mkdir() (dot / "repo.json").write_text('{"repo_id":"fetch-phase3","owner":"gabriel"}') for d in ("commits", "snapshots", "objects"): (dot / d).mkdir() (dot / "refs" / "heads").mkdir(parents=True) (dot / "HEAD").write_text("ref: refs/heads/main\n") (dot / "config.toml").write_text("") return tmp def _populate(repo: pathlib.Path, n_commits: int, seed: str = "pf") -> list[str]: """Populate local repo with n_commits and return all commit IDs in order.""" blob_ids: list[str] = [] for i in range(_N_FILES): data = f"{seed}-base-{i:04d}".encode() + b"x" * _BLOB_SIZE oid = blob_id(data) write_object(repo, oid, data) blob_ids.append(oid) base_manifest = {f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)} parent = None ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) commit_ids: list[str] = [] for i in range(n_commits): manifest = dict(base_manifest) raw = f"{seed}-c{i:04d}".encode() + b"y" * _BLOB_SIZE oid = blob_id(raw) write_object(repo, oid, raw) manifest[f"src/file_{i % _N_FILES:04d}.py"] = oid sid = compute_snapshot_id(manifest) write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest)) msg = f"commit-{i:05d}" cid = compute_commit_id( parent_ids=[parent] if parent else [], snapshot_id=sid, message=msg, committed_at_iso=ts.isoformat(), author="gabriel", ) write_commit(repo, CommitRecord( commit_id=cid, branch="main", snapshot_id=sid, message=msg, committed_at=ts, parent_commit_id=parent, parent2_commit_id=None, author="gabriel", metadata={}, structured_delta=None, sem_ver_bump="none", breaking_changes=[], agent_id="", model_id="", toolchain_id="", prompt_hash="", signature="", signer_key_id="", )) parent = cid commit_ids.append(cid) ts += datetime.timedelta(seconds=60) tip = commit_ids[-1] write_branch_ref(repo, "main", tip) return commit_ids async def _http_push( client: AsyncClient, repo_slug: str, local_repo: pathlib.Path, tip: str, have: list[str] | None = None, ) -> tuple[str, dict]: """Push via HTTP unpack-mpack. Returns (mpack_key, mpack_dict).""" import musehub.storage.backends as _backends_mod from muse.core.types import blob_id as _blob_id mpack_dict = build_mpack(local_repo, [tip], have=have or []) wire_bytes = build_wire_mpack(mpack_dict) mpack_key = _blob_id(wire_bytes) backend = _backends_mod.get_backend() await backend.put_mpack(mpack_key, wire_bytes) n_commits = len(mpack_dict.get("commits") or []) n_blobs = len(mpack_dict.get("blobs") or []) resp = await client.post( f"/gabriel/{repo_slug}/push/unpack-mpack", content=msgpack.packb({ "mpack_key": mpack_key, "branch": "main", "head": tip, "commits_count": n_commits, "blobs_count": n_blobs, }, use_bin_type=True), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code == 200, f"push failed: {resp.status_code} {resp.text}" return mpack_key, mpack_dict async def _run_index_job( db_session: AsyncSession, repo_id: str, mpack_key: str, tip: str, n_blobs: int, ) -> None: """Create a mpack.index job row and run process_mpack_index_job.""" from datetime import datetime, timezone from musehub.core.genesis import compute_job_id from musehub.db.musehub_jobs_models import MusehubBackgroundJob from musehub.services.musehub_wire import process_mpack_index_job now = datetime.now(tz=timezone.utc) job_id = compute_job_id(repo_id, "mpack.index", now.isoformat()) db_session.add(MusehubBackgroundJob( job_id=job_id, repo_id=repo_id, job_type="mpack.index", payload={ "mpack_key": mpack_key, "pusher_id": "sha256:" + "0" * 64, "branch": "main", "head": tip, "force": False, "declared_objects_count": n_blobs, }, status="pending", )) await db_session.flush() await process_mpack_index_job(db_session, job_id) await db_session.commit() # ── PF-1 ────────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_pf1_fetch_returns_mpack_url( client: AsyncClient, repo: dict, tmp_path: pathlib.Path, db_session: AsyncSession, ) -> None: """wire_fetch_mpack returns a presigned mpack_url after push + index. force_build=True bypasses the prebuild cache so the fetch path is exercised end-to-end without needing the prebuild job. """ from musehub.services.musehub_wire import wire_fetch_mpack local_repo = _make_local_repo(tmp_path / "repo") commit_ids = _populate(local_repo, n_commits=3) tip = commit_ids[-1] mpack_key, mpack_dict = await _http_push(client, repo["slug"], local_repo, tip) n_blobs = len(mpack_dict.get("blobs") or []) await _run_index_job(db_session, repo["repoId"], mpack_key, tip, n_blobs) result = await wire_fetch_mpack( db_session, repo["repoId"], want=[tip], have=[], ttl_seconds=3600, force_build=True, ) assert result["mpack_url"] is not None, "mpack_url must be set after a successful fetch" assert result["mpack_id"] is not None, "mpack_id must be set" assert result["commit_count"] > 0, ( f"expected commit_count > 0, got {result['commit_count']}" ) # ── PF-2 ────────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_pf2_delta_fetch_returns_only_new_commits( client: AsyncClient, repo: dict, tmp_path: pathlib.Path, db_session: AsyncSession, ) -> None: """Delta fetch with have=[tip_a] returns only the commits not yet on the client. Pushes 3 commits in one mpack, runs the index job, then fetches with have=[first_commit] — the response should cover only commits 2 and 3, not commit 1. """ from musehub.services.musehub_wire import wire_fetch_mpack local_repo = _make_local_repo(tmp_path / "repo") commit_ids = _populate(local_repo, n_commits=3) tip_a = commit_ids[0] # client already has this tip_b = commit_ids[-1] # client wants this mpack_key, mpack_dict = await _http_push(client, repo["slug"], local_repo, tip_b) n_blobs = len(mpack_dict.get("blobs") or []) await _run_index_job(db_session, repo["repoId"], mpack_key, tip_b, n_blobs) result = await wire_fetch_mpack( db_session, repo["repoId"], want=[tip_b], have=[tip_a], ttl_seconds=3600, force_build=True, ) assert result["mpack_url"] is not None, "delta fetch must return a mpack_url" total_commits = len(commit_ids) assert result["commit_count"] < total_commits, ( f"delta fetch must return fewer than {total_commits} commits, " f"got {result['commit_count']}" ) # ── PF-3 ────────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_pf3_cold_cache_raises_mpack_not_ready( client: AsyncClient, repo: dict, tmp_path: pathlib.Path, db_session: AsyncSession, ) -> None: """wire_fetch_mpack with have=[] and force_build=False raises MPackNotReadyError. In production, the prebuild job warms the cache before the first clone. Without force_build, a cache miss is a signal that the mpack is not yet ready and the client should retry. """ from musehub.services.musehub_wire_shared import MPackNotReadyError from musehub.services.musehub_wire import wire_fetch_mpack local_repo = _make_local_repo(tmp_path / "repo") commit_ids = _populate(local_repo, n_commits=2) tip = commit_ids[-1] # Push so commits exist in the DB, but do NOT warm the prebuild cache. mpack_key, mpack_dict = await _http_push(client, repo["slug"], local_repo, tip) with pytest.raises(MPackNotReadyError): await wire_fetch_mpack( db_session, repo["repoId"], want=[tip], have=[], ttl_seconds=3600, force_build=False, )