"""Wire protocol endpoint tests. Covers the three Muse CLI transport endpoints (Git-style URLs): GET /{owner}/{slug}/refs POST /{owner}/{slug}/push POST /{owner}/{slug}/fetch And the content-addressed CDN endpoint: GET /o/{object_id} Remote URL format (same pattern as Git): muse remote add origin https://musehub.ai/gabriel/muse """ from __future__ import annotations import time import uuid from datetime import datetime, timezone import msgpack import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from musehub.db import musehub_models as db from musehub.db.musehub_collaborator_models import MusehubCollaborator from musehub.main import app from tests.factories import create_repo as factory_create_repo from musehub.muse_contracts.json_types import JSONObject, StrDict # ── helpers ──────────────────────────────────────────────────────────────────── def _utc_now() -> datetime: return datetime.now(tz=timezone.utc) def _make_commit(repo_id: str, commit_id: str | None = None, parent: str | None = None) -> JSONObject: return { "commit_id": commit_id or str(uuid.uuid4()), "repo_id": repo_id, "branch": "main", "snapshot_id": f"snap_{uuid.uuid4().hex[:8]}", "message": "chore: add test commit", "committed_at": _utc_now().isoformat(), "parent_commit_id": parent, "author": "Test User ", "sem_ver_bump": "patch", } def _make_object(content: bytes = b"hello world") -> JSONObject: oid = uuid.uuid4().hex return { "object_id": oid, "content": content, "path": "README.md", } def _make_snapshot(snap_id: str, object_id: str) -> JSONObject: return { "snapshot_id": snap_id, "manifest": {"README.md": object_id}, "created_at": _utc_now().isoformat(), } def _mp(data: JSONObject) -> bytes: """Encode data as msgpack for test request bodies.""" return msgpack.packb(data, use_bin_type=True) # ── refs endpoint ────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_refs_returns_404_for_unknown_owner_slug(client: AsyncClient) -> None: resp = await client.get("/no-such-owner/no-such-slug/refs") assert resp.status_code == 404 @pytest.mark.asyncio async def test_refs_returns_branch_heads( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="muse-test", domain_meta={"domain": "code"}) branch = db.MusehubBranch( repo_id=repo.repo_id, name="main", head_commit_id="abc123", ) db_session.add(branch) await db_session.commit() owner = repo.owner slug = repo.slug resp = await client.get(f"/{owner}/{slug}/refs") assert resp.status_code == 200 data = resp.json() assert data["repo_id"] == repo.repo_id assert data["default_branch"] == "main" assert data["domain"] == "code" assert data["branch_heads"]["main"] == "abc123" @pytest.mark.asyncio async def test_refs_url_is_owner_slash_slug( client: AsyncClient, db_session: AsyncSession, ) -> None: """Confirm the remote URL pattern matches Git: /{owner}/{slug}/refs — no /wire/ prefix.""" repo = await factory_create_repo(db_session, slug="git-style-test") owner, slug = repo.owner, repo.slug resp = await client.get(f"/{owner}/{slug}/refs") assert resp.status_code == 200 # Should NOT need /wire/ in the path resp_wire = await client.get(f"/wire/repos/{repo.repo_id}/refs") assert resp_wire.status_code == 404 @pytest.mark.asyncio async def test_refs_empty_repo_has_empty_branch_heads( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="empty-test") resp = await client.get(f"/{repo.owner}/{repo.slug}/refs") assert resp.status_code == 200 data = resp.json() assert data["branch_heads"] == {} # ── push endpoint ────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_push_requires_auth(client: AsyncClient, db_session: AsyncSession) -> None: repo = await factory_create_repo(db_session, slug="push-auth-test", owner_user_id="test-user-wire") resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code in (401, 403) @pytest.mark.asyncio async def test_push_404_for_unknown_repo( client: AsyncClient, wire_headers: StrDict, ) -> None: resp = await client.post( "/nobody/no-such-repo/push", content=_mp({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}), headers=wire_headers, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_push_rejected_for_non_owner( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Authenticated user who is NOT the repo owner must be rejected.""" repo = await factory_create_repo( db_session, slug="push-nonowner-test", owner_user_id="someone-else", # different from test-user-wire ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}), headers=wire_headers, ) assert resp.status_code == 409 assert "not authorized" in resp.json()["detail"] @pytest.mark.asyncio async def test_push_ingests_commit_and_branch( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="push-ingest-test", owner="test-user-wire") commit_id = uuid.uuid4().hex obj = _make_object() snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit = _make_commit(repo.repo_id, commit_id=commit_id) commit["snapshot_id"] = snap_id payload = { "bundle": { "commits": [commit], "snapshots": [snap], "objects": [obj], }, "branch": "main", "force": False, } resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers, ) assert resp.status_code == 200, resp.text data = msgpack.unpackb(resp.content, raw=False) assert data["ok"] is True assert "main" in data["branch_heads"] assert data["remote_head"] == commit_id @pytest.mark.asyncio async def test_push_is_idempotent( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Pushing the same commit twice must succeed both times.""" repo = await factory_create_repo(db_session, slug="push-idempotent-test", owner="test-user-wire") commit = _make_commit(repo.repo_id) payload = { "bundle": {"commits": [commit], "snapshots": [], "objects": []}, "branch": "main", } url = f"/{repo.owner}/{repo.slug}/push" resp1 = await client.post(url, content=_mp(payload), headers=wire_headers) assert resp1.status_code == 200 resp2 = await client.post(url, content=_mp(payload), headers=wire_headers) assert resp2.status_code == 200 @pytest.mark.asyncio async def test_push_non_fast_forward_rejected( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="push-nff-test", owner="test-user-wire") existing_commit_id = uuid.uuid4().hex branch = db.MusehubBranch( repo_id=repo.repo_id, name="main", head_commit_id=existing_commit_id, ) db_session.add(branch) await db_session.commit() # Push a commit without existing_commit_id as parent new_commit = _make_commit(repo.repo_id, parent=None) payload = { "bundle": {"commits": [new_commit], "snapshots": [], "objects": []}, "branch": "main", "force": False, } resp = await client.post(f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers) assert resp.status_code == 409 # 409 Conflict for non-fast-forward assert "non-fast-forward" in resp.json()["detail"] @pytest.mark.asyncio async def test_push_force_overwrites_branch( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="push-force-test", owner="test-user-wire") old_head = uuid.uuid4().hex branch = db.MusehubBranch( repo_id=repo.repo_id, name="main", head_commit_id=old_head, ) db_session.add(branch) await db_session.commit() new_commit = _make_commit(repo.repo_id, parent=None) payload = { "bundle": {"commits": [new_commit], "snapshots": [], "objects": []}, "branch": "main", "force": True, } resp = await client.post(f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["ok"] is True assert data["branch_heads"]["main"] != old_head # ── fetch endpoint ───────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_fetch_404_for_unknown_repo(client: AsyncClient) -> None: resp = await client.post( "/nobody/no-such-repo/fetch", content=_mp({"want": [], "have": []}), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_fetch_empty_want_returns_empty_bundle( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="fetch-empty-test") resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch", content=_mp({"want": [], "have": []}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["commits"] == [] assert data["snapshots"] == [] # fetch returns only VCS metadata — no object bytes (use /fetch/objects for that) @pytest.mark.asyncio async def test_fetch_returns_missing_commits( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="fetch-commits-test") commit_id = uuid.uuid4().hex commit_row = db.MusehubCommit( commit_id=commit_id, repo_id=repo.repo_id, branch="main", parent_ids=[], message="initial commit", author="Test", timestamp=_utc_now(), snapshot_id=None, commit_meta={}, ) branch_row = db.MusehubBranch( repo_id=repo.repo_id, name="main", head_commit_id=commit_id, ) db_session.add(commit_row) db_session.add(branch_row) await db_session.commit() resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch", content=_mp({"want": [commit_id], "have": []}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert len(data["commits"]) == 1 assert data["commits"][0]["commit_id"] == commit_id assert data["branch_heads"]["main"] == commit_id # ── content-addressed CDN ────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_object_cdn_returns_404_for_missing(client: AsyncClient) -> None: resp = await client.get("/o/nonexistent-sha-12345") assert resp.status_code == 404 # ── unit tests ───────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_wire_models_parse_correctly() -> None: """WireBundle Pydantic parsing mirrors Muse CLI format.""" from musehub.models.wire import WireBundle, WireCommit, WirePushRequest commit_dict = { "commit_id": "abc123", "message": "feat: add track", "committed_at": "2026-03-19T10:00:00+00:00", "author": "Gabriel ", "sem_ver_bump": "minor", "breaking_changes": [], "agent_id": "", "format_version": 5, } req = WirePushRequest( bundle=WireBundle(commits=[WireCommit.model_validate(commit_dict)], snapshots=[], objects=[]), branch="main", force=False, ) assert req.bundle.commits[0].commit_id == "abc123" assert req.bundle.commits[0].sem_ver_bump == "minor" assert req.force is False @pytest.mark.asyncio async def test_topological_sort_orders_parents_first() -> None: from musehub.models.wire import WireCommit from musehub.services.musehub_wire import _topological_sort c1 = WireCommit(commit_id="parent", message="parent") c2 = WireCommit(commit_id="child", message="child", parent_commit_id="parent") sorted_ = _topological_sort([c2, c1]) ids = [c.commit_id for c in sorted_] assert ids.index("parent") < ids.index("child") @pytest.mark.asyncio async def test_remote_url_format_matches_git_pattern( client: AsyncClient, db_session: AsyncSession, ) -> None: """The remote URL is /{owner}/{slug} — no /wire/ prefix, no UUID. This mirrors Git: git remote add origin https://github.com/owner/repo versus UUID-based alternatives like: muse remote add origin https://musehub.ai/wire/repos/550e8400-.../ """ repo = await factory_create_repo(db_session, slug="url-format-test") # /{owner}/{slug}/refs must work resp = await client.get(f"/{repo.owner}/{repo.slug}/refs") assert resp.status_code == 200 # The response confirms which repo was resolved — no UUID needed in the URL data = resp.json() assert data["repo_id"] == repo.repo_id # ── filter-objects endpoint (MWP Phase 1) ─────────────────────────────────── @pytest.mark.asyncio async def test_filter_objects_requires_auth( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="filter-auth-test") resp = await client.post( f"/{repo.owner}/{repo.slug}/filter-objects", content=_mp({"object_ids": ["abc123"]}), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code in (401, 403) @pytest.mark.asyncio async def test_filter_objects_empty_list_returns_empty( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="filter-empty-test", owner="test-user-wire") resp = await client.post( f"/{repo.owner}/{repo.slug}/filter-objects", content=_mp({"object_ids": []}), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["missing"] == [] @pytest.mark.asyncio async def test_filter_objects_all_missing( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="filter-all-missing-test", owner="test-user-wire") oids = [uuid.uuid4().hex for _ in range(5)] resp = await client.post( f"/{repo.owner}/{repo.slug}/filter-objects", content=_mp({"object_ids": oids}), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert set(data["missing"]) == set(oids) @pytest.mark.asyncio async def test_filter_objects_returns_only_missing( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Push one object, then filter-objects — it should NOT appear in missing.""" repo = await factory_create_repo(db_session, slug="filter-delta-test", owner="test-user-wire") obj = _make_object(b"stored content") commit = _make_commit(repo.repo_id) snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit["snapshot_id"] = snap_id push_payload = {"bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main"} r = await client.post(f"/{repo.owner}/{repo.slug}/push", content=_mp(push_payload), headers=wire_headers) assert r.status_code == 200 new_oid = uuid.uuid4().hex resp = await client.post( f"/{repo.owner}/{repo.slug}/filter-objects", content=_mp({"object_ids": [obj["object_id"], new_oid]}), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert obj["object_id"] not in data["missing"] assert new_oid in data["missing"] @pytest.mark.asyncio async def test_filter_objects_accepts_json_body( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """filter-objects must accept application/json as well as msgpack.""" repo = await factory_create_repo(db_session, slug="filter-json-test", owner="test-user-wire") import json as _json headers = {**wire_headers, "Content-Type": "application/json"} resp = await client.post( f"/{repo.owner}/{repo.slug}/filter-objects", content=_json.dumps({"object_ids": ["abc"]}).encode(), headers=headers, ) assert resp.status_code == 200 # ── presign endpoint (MWP Phase 3) ────────────────────────────────────────── @pytest.mark.asyncio async def test_presign_local_backend_returns_all_inline( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Local backend has no presign_put/get; all object IDs must be returned in inline.""" repo = await factory_create_repo(db_session, slug="presign-inline-test", owner="test-user-wire") oids = [uuid.uuid4().hex for _ in range(3)] resp = await client.post( f"/{repo.owner}/{repo.slug}/presign", content=_mp({"object_ids": oids, "direction": "put", "ttl_seconds": 3600}), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert set(data["inline"]) == set(oids) assert data["presigned"] == {} @pytest.mark.asyncio async def test_presign_requires_auth( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="presign-auth-test") resp = await client.post( f"/{repo.owner}/{repo.slug}/presign", content=_mp({"object_ids": ["abc"], "direction": "put", "ttl_seconds": 300}), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code in (401, 403) @pytest.mark.asyncio async def test_presign_get_local_backend_returns_all_inline( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """direction=get also falls back to inline on local backend.""" repo = await factory_create_repo(db_session, slug="presign-get-test", owner="test-user-wire") oids = [uuid.uuid4().hex] resp = await client.post( f"/{repo.owner}/{repo.slug}/presign", content=_mp({"object_ids": oids, "direction": "get", "ttl_seconds": 300}), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert oids[0] in data["inline"] # ── negotiate endpoint (MWP Phase 5) ──────────────────────────────────────── @pytest.mark.asyncio async def test_negotiate_full_clone_ready_immediately( client: AsyncClient, db_session: AsyncSession, ) -> None: """When client has no have-IDs (full clone), ready must be True immediately.""" repo = await factory_create_repo(db_session, slug="negotiate-clone-test") commit_id = uuid.uuid4().hex commit_row = db.MusehubCommit( commit_id=commit_id, repo_id=repo.repo_id, branch="main", parent_ids=[], message="initial", author="Test", timestamp=_utc_now(), snapshot_id=None, commit_meta={}, ) db_session.add(commit_row) await db_session.commit() resp = await client.post( f"/{repo.owner}/{repo.slug}/negotiate", content=_mp({"have": [], "want": [commit_id]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["ready"] is True assert data["ack"] == [] @pytest.mark.asyncio async def test_negotiate_acks_known_have_ids( client: AsyncClient, db_session: AsyncSession, ) -> None: """Server acks have-IDs it recognises, reports common_base, and sets ready=True.""" repo = await factory_create_repo(db_session, slug="negotiate-ack-test") parent_id = uuid.uuid4().hex child_id = uuid.uuid4().hex parent_row = db.MusehubCommit( commit_id=parent_id, repo_id=repo.repo_id, branch="main", parent_ids=[], message="parent", author="T", timestamp=_utc_now(), snapshot_id=None, commit_meta={}, ) child_row = db.MusehubCommit( commit_id=child_id, repo_id=repo.repo_id, branch="main", parent_ids=[parent_id], message="child", author="T", timestamp=_utc_now(), snapshot_id=None, commit_meta={}, ) db_session.add(parent_row) db_session.add(child_row) await db_session.commit() resp = await client.post( f"/{repo.owner}/{repo.slug}/negotiate", content=_mp({"have": [parent_id], "want": [child_id]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert parent_id in data["ack"] assert data["common_base"] == parent_id assert data["ready"] is True @pytest.mark.asyncio async def test_negotiate_404_for_unknown_repo(client: AsyncClient) -> None: resp = await client.post( "/nobody/no-such/negotiate", content=_mp({"have": [], "want": []}), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code == 404 # ── push/objects endpoint (chunked pre-upload) ──────────────────────────────── @pytest.mark.asyncio async def test_push_objects_stores_objects( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="push-objects-store-test", owner="test-user-wire") obj = _make_object(b"chunked content") resp = await client.post( f"/{repo.owner}/{repo.slug}/push/objects", content=_mp({"objects": [obj]}), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["stored"] == 1 assert data["skipped"] == 0 @pytest.mark.asyncio async def test_push_objects_skips_duplicates( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Uploading the same object twice: second upload counts as skipped.""" repo = await factory_create_repo(db_session, slug="push-objects-dedup-test", owner="test-user-wire") obj = _make_object(b"dedup me") url = f"/{repo.owner}/{repo.slug}/push/objects" r1 = await client.post(url, content=_mp({"objects": [obj]}), headers=wire_headers) assert r1.status_code == 200 d1 = msgpack.unpackb(r1.content, raw=False) assert d1["stored"] == 1 r2 = await client.post(url, content=_mp({"objects": [obj]}), headers=wire_headers) assert r2.status_code == 200 d2 = msgpack.unpackb(r2.content, raw=False) assert d2["skipped"] == 1 @pytest.mark.asyncio async def test_push_objects_requires_auth( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="push-objects-auth-test") resp = await client.post( f"/{repo.owner}/{repo.slug}/push/objects", content=_mp({"objects": []}), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code in (401, 403) # ── CDN endpoint ────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_object_cdn_serves_pushed_object( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Object pushed via push/objects must be retrievable via the CDN endpoint.""" repo = await factory_create_repo(db_session, slug="cdn-happy-path-test", owner="test-user-wire") content = b"cdn test content" obj = _make_object(content) upload = await client.post( f"/{repo.owner}/{repo.slug}/push/objects", content=_mp({"objects": [obj]}), headers=wire_headers, ) assert upload.status_code == 200 cdn_resp = await client.get(f"/o/{obj['object_id']}?repo_id={repo.repo_id}") assert cdn_resp.status_code == 200 assert cdn_resp.content == content assert cdn_resp.headers["cache-control"].startswith("public") @pytest.mark.asyncio async def test_object_cdn_has_immutable_cache_headers( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="cdn-cache-test", owner="test-user-wire") obj = _make_object(b"immutable blob") await client.post( f"/{repo.owner}/{repo.slug}/push/objects", content=_mp({"objects": [obj]}), headers=wire_headers, ) cdn = await client.get(f"/o/{obj['object_id']}?repo_id={repo.repo_id}") assert "immutable" in cdn.headers.get("cache-control", "") assert cdn.headers.get("etag", "").strip('"') == obj["object_id"] # ── private repo visibility ──────────────────────────────────────────────────── @pytest.mark.asyncio async def test_refs_private_repo_returns_404_for_unauthenticated( client: AsyncClient, db_session: AsyncSession, ) -> None: """Private repos must be invisible to unauthenticated callers.""" repo = await factory_create_repo(db_session, slug="private-refs-test", visibility="private") resp = await client.get(f"/{repo.owner}/{repo.slug}/refs") assert resp.status_code == 404 @pytest.mark.asyncio async def test_refs_private_repo_visible_to_owner( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo( db_session, slug="private-owner-test", owner="test-user-wire", visibility="private", ) resp = await client.get(f"/{repo.owner}/{repo.slug}/refs", headers=wire_headers) assert resp.status_code == 200 @pytest.mark.asyncio async def test_fetch_private_repo_returns_404_for_stranger( client: AsyncClient, db_session: AsyncSession, ) -> None: """Private repo fetch must 404 for unauthenticated stranger.""" repo = await factory_create_repo(db_session, slug="private-fetch-test", visibility="private") resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch", content=_mp({"want": [], "have": []}), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code == 404 # ── collaborator auth ────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_push_write_collaborator_is_allowed( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """A write collaborator must be permitted to push.""" repo = await factory_create_repo(db_session, slug="collab-write-push-test", owner="repo-owner-different") collab = MusehubCollaborator( repo_id=repo.repo_id, identity_handle="test-user-wire", permission="write", accepted_at=_utc_now(), ) db_session.add(collab) await db_session.commit() commit = _make_commit(repo.repo_id) payload = {"bundle": {"commits": [commit], "snapshots": [], "objects": []}, "branch": "main"} resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["ok"] is True @pytest.mark.asyncio async def test_push_read_collaborator_is_rejected( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """A read-only collaborator must NOT be able to push.""" repo = await factory_create_repo(db_session, slug="collab-read-push-test", owner="repo-owner-different-2") collab = MusehubCollaborator( repo_id=repo.repo_id, identity_handle="test-user-wire", permission="read", accepted_at=_utc_now(), ) db_session.add(collab) await db_session.commit() commit = _make_commit(repo.repo_id) payload = {"bundle": {"commits": [commit], "snapshots": [], "objects": []}, "branch": "main"} resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers, ) assert resp.status_code == 409 assert "not authorized" in resp.json()["detail"] @pytest.mark.asyncio async def test_push_unaccepted_invite_is_rejected( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """A collaborator with a pending (unaccepted) invite must NOT be able to push.""" repo = await factory_create_repo(db_session, slug="collab-pending-push-test", owner="repo-owner-different-3") collab = MusehubCollaborator( repo_id=repo.repo_id, identity_handle="test-user-wire", permission="write", accepted_at=None, # not yet accepted ) db_session.add(collab) await db_session.commit() commit = _make_commit(repo.repo_id) payload = {"bundle": {"commits": [commit], "snapshots": [], "objects": []}, "branch": "main"} resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers, ) assert resp.status_code == 409 assert "not authorized" in resp.json()["detail"] # ── delete branch endpoint ──────────────────────────────────────────────────── @pytest.mark.asyncio async def test_delete_branch_owner_can_delete( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="delete-branch-test", owner="test-user-wire") branch = db.MusehubBranch( repo_id=repo.repo_id, name="feat/old", head_commit_id=uuid.uuid4().hex, ) db_session.add(branch) await db_session.commit() resp = await client.delete( f"/{repo.owner}/{repo.slug}/branches/feat/old", headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["deleted"] == "feat/old" @pytest.mark.asyncio async def test_delete_branch_cannot_delete_default( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="delete-default-branch-test", owner="test-user-wire") branch = db.MusehubBranch( repo_id=repo.repo_id, name="main", head_commit_id=uuid.uuid4().hex, ) db_session.add(branch) await db_session.commit() resp = await client.delete( f"/{repo.owner}/{repo.slug}/branches/main", headers=wire_headers, ) assert resp.status_code == 409 assert "default branch" in resp.json()["detail"] @pytest.mark.asyncio async def test_delete_branch_non_owner_gets_403( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="delete-branch-403-test", owner="someone-else") branch = db.MusehubBranch( repo_id=repo.repo_id, name="feat/x", head_commit_id=uuid.uuid4().hex, ) db_session.add(branch) await db_session.commit() resp = await client.delete( f"/{repo.owner}/{repo.slug}/branches/feat/x", headers=wire_headers, ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_delete_branch_missing_branch_404( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="delete-branch-404-test", owner="test-user-wire") resp = await client.delete( f"/{repo.owner}/{repo.slug}/branches/does-not-exist", headers=wire_headers, ) assert resp.status_code == 404 # ── delete release endpoint ─────────────────────────────────────────────────── @pytest.mark.asyncio async def test_delete_release_non_owner_gets_403( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="delete-release-403-test", owner="someone-else") resp = await client.delete( f"/{repo.owner}/{repo.slug}/releases/1.0.0", headers=wire_headers, ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_delete_release_missing_tag_404( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="delete-release-404-test", owner="test-user-wire") resp = await client.delete( f"/{repo.owner}/{repo.slug}/releases/99.0.0", headers=wire_headers, ) assert resp.status_code == 404 # ── push tags endpoint ──────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_push_tags_stores_tags( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="push-tags-test", owner="test-user-wire") tag = { "tag_id": uuid.uuid4().hex, "commit_id": uuid.uuid4().hex, "tag": "status:reviewed", "created_at": _utc_now().isoformat(), } resp = await client.post( f"/{repo.owner}/{repo.slug}/tags", content=_mp({"tags": [tag]}), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["stored"] == 1 @pytest.mark.asyncio async def test_push_tags_invalid_body_422( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: repo = await factory_create_repo(db_session, slug="push-tags-422-test", owner="test-user-wire") resp = await client.post( f"/{repo.owner}/{repo.slug}/tags", content=_mp({"tags": "not-a-list"}), headers=wire_headers, ) assert resp.status_code == 422 @pytest.mark.asyncio async def test_push_tags_requires_auth( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="push-tags-auth-test") resp = await client.post( f"/{repo.owner}/{repo.slug}/tags", content=_mp({"tags": []}), headers={"Content-Type": "application/x-msgpack"}, ) assert resp.status_code in (401, 403) # ── DoS limits (Pydantic model validation) ──────────────────────────────────── def test_wire_object_rejects_oversized_content() -> None: """WireObject must reject content larger than MAX_OBJECT_BYTES.""" from pydantic import ValidationError from musehub.models.wire import WireObject, MAX_OBJECT_BYTES oversized = b"x" * (MAX_OBJECT_BYTES + 1) with pytest.raises(ValidationError, match="content"): WireObject(object_id="abc", content=oversized, path="big.bin") def test_wire_bundle_rejects_too_many_objects() -> None: """WireBundle.objects list must be capped at MAX_OBJECTS_PER_PUSH.""" from pydantic import ValidationError from musehub.models.wire import WireBundle, MAX_OBJECTS_PER_PUSH objs = [{"object_id": uuid.uuid4().hex, "content": b"x", "path": "f"} for _ in range(MAX_OBJECTS_PER_PUSH + 1)] with pytest.raises(ValidationError): WireBundle(commits=[], snapshots=[], objects=objs) def test_wire_bundle_rejects_too_many_commits() -> None: """WireBundle.commits list must be capped at MAX_COMMITS_PER_PUSH.""" from pydantic import ValidationError from musehub.models.wire import WireBundle, WireCommit, MAX_COMMITS_PER_PUSH commits = [WireCommit(commit_id=uuid.uuid4().hex, message="c") for _ in range(MAX_COMMITS_PER_PUSH + 1)] with pytest.raises(ValidationError): WireBundle(commits=commits, snapshots=[], objects=[]) def test_wire_fetch_request_rejects_too_many_wants() -> None: """WireFetchRequest.want must be capped at MAX_WANT_PER_FETCH.""" from pydantic import ValidationError from musehub.models.wire import WireFetchRequest, MAX_WANT_PER_FETCH wants = [uuid.uuid4().hex for _ in range(MAX_WANT_PER_FETCH + 1)] with pytest.raises(ValidationError): WireFetchRequest(want=wants, have=[]) # ── integration: push → fetch round-trip ────────────────────────────────────── @pytest.mark.asyncio async def test_push_then_fetch_round_trip( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Full round-trip: push a commit+snapshot+object, then fetch it back and verify content.""" repo = await factory_create_repo(db_session, slug="round-trip-test", owner="test-user-wire") content = b"round trip content" obj = _make_object(content) snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit_id = uuid.uuid4().hex commit = _make_commit(repo.repo_id, commit_id=commit_id) commit["snapshot_id"] = snap_id push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({"bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main"}), headers=wire_headers, ) assert push_resp.status_code == 200 # Phase 1: fetch VCS metadata (no object bytes) fetch_resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch", content=_mp({"want": [commit_id], "have": []}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert fetch_resp.status_code == 200 data = msgpack.unpackb(fetch_resp.content, raw=False) assert any(c["commit_id"] == commit_id for c in data["commits"]) assert any(s["snapshot_id"] == snap_id for s in data["snapshots"]) # Phase 2: fetch object bytes objects_resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch/objects", content=_mp({"object_ids": [obj["object_id"]]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert objects_resp.status_code == 200 objects_data = msgpack.unpackb(objects_resp.content, raw=False) assert any(bytes(o["content"]) == content for o in objects_data.get("objects", [])) @pytest.mark.asyncio async def test_fetch_have_excludes_known_commits( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Commits in have must not appear in the fetch response.""" repo = await factory_create_repo(db_session, slug="fetch-have-test", owner="test-user-wire") parent_id = uuid.uuid4().hex child_id = uuid.uuid4().hex parent = _make_commit(repo.repo_id, commit_id=parent_id) child = _make_commit(repo.repo_id, commit_id=child_id, parent=parent_id) for c in [parent, child]: push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({"bundle": {"commits": [c], "snapshots": [], "objects": []}, "branch": "main"}), headers=wire_headers, ) assert push_resp.status_code == 200 # Client already has parent — should only receive child fetch_resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch", content=_mp({"want": [child_id], "have": [parent_id]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert fetch_resp.status_code == 200 data = msgpack.unpackb(fetch_resp.content, raw=False) commit_ids = [c["commit_id"] for c in data["commits"]] assert child_id in commit_ids assert parent_id not in commit_ids # ── content negotiation ──────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_refs_returns_json_when_no_msgpack_accept( client: AsyncClient, db_session: AsyncSession, ) -> None: """When Accept header is not msgpack, response must be application/json.""" repo = await factory_create_repo(db_session, slug="content-neg-test") resp = await client.get(f"/{repo.owner}/{repo.slug}/refs", headers={"Accept": "application/json"}) assert resp.status_code == 200 assert "application/json" in resp.headers.get("content-type", "") data = resp.json() assert "repo_id" in data @pytest.mark.asyncio async def test_push_accepts_json_content_type( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """push endpoint must accept application/json bodies, not only msgpack.""" import json as _json repo = await factory_create_repo(db_session, slug="push-json-ct-test", owner="test-user-wire") commit = _make_commit(repo.repo_id) payload = {"bundle": {"commits": [commit], "snapshots": [], "objects": []}, "branch": "main"} json_headers = {**wire_headers, "Content-Type": "application/json"} resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_json.dumps(payload).encode(), headers=json_headers, ) assert resp.status_code == 200 # ── unit: _is_ancestor_in_bundle ────────────────────────────────────────────── def test_is_ancestor_in_bundle_direct_parent() -> None: from musehub.models.wire import WireCommit from musehub.services.musehub_wire import _is_ancestor_in_bundle parent = WireCommit(commit_id="p", message="parent") child = WireCommit(commit_id="c", message="child", parent_commit_id="p") assert _is_ancestor_in_bundle("p", [parent, child]) is True def test_is_ancestor_in_bundle_not_in_chain() -> None: from musehub.models.wire import WireCommit from musehub.services.musehub_wire import _is_ancestor_in_bundle c1 = WireCommit(commit_id="a", message="a") c2 = WireCommit(commit_id="b", message="b", parent_commit_id="a") assert _is_ancestor_in_bundle("x", [c1, c2]) is False def test_is_ancestor_in_bundle_merge_commit() -> None: """Merge commits have two parents — both must be checked.""" from musehub.models.wire import WireCommit from musehub.services.musehub_wire import _is_ancestor_in_bundle p1 = WireCommit(commit_id="p1", message="first parent") p2 = WireCommit(commit_id="p2", message="second parent") merge = WireCommit( commit_id="m", message="merge", parent_commit_id="p1", parent2_commit_id="p2", ) assert _is_ancestor_in_bundle("p2", [p1, p2, merge]) is True def test_topological_sort_handles_empty_list() -> None: from musehub.services.musehub_wire import _topological_sort assert _topological_sort([]) == [] def test_topological_sort_handles_orphan_commits() -> None: """Commits with no parent relationship must all appear in the result.""" from musehub.models.wire import WireCommit from musehub.services.musehub_wire import _topological_sort commits = [WireCommit(commit_id=str(i), message=str(i)) for i in range(5)] sorted_ = _topological_sort(commits) assert len(sorted_) == 5 # ── regression: empty-content objects (SHA-256("") = e3b0c44…) ─────────────── # # Python's `not b""` is True, so a naive `if not content: continue` guard # silently drops empty files on push. These tests lock in the fix: # - push/objects: empty-content object is stored, not skipped # - push bundle: empty-content object in bundle is stored, not skipped # - fetch: empty-content object is returned inline (even if never # stored, the server synthesizes it from the known SHA) _EMPTY_OID = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" @pytest.mark.asyncio async def test_push_objects_stores_empty_content_object( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """push/objects must store an object whose content is b'' (empty file). Before the fix, `if not wire_obj.content: continue` treated b'' as falsy and silently dropped the object. The correct guard is `wire_obj.content is None`. """ repo = await factory_create_repo(db_session, slug="empty-obj-push-test", owner="test-user-wire") empty_obj = {"object_id": _EMPTY_OID, "content": b"", "path": ".museattributes"} resp = await client.post( f"/{repo.owner}/{repo.slug}/push/objects", content=_mp({"objects": [empty_obj]}), headers=wire_headers, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data["stored"] == 1, f"empty object was not stored: {data}" assert data["skipped"] == 0 @pytest.mark.asyncio async def test_push_bundle_stores_empty_content_object( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """push bundle must persist an object with content=b'' in the bundle.objects list.""" repo = await factory_create_repo(db_session, slug="empty-obj-bundle-test", owner="test-user-wire") snap_id = f"snap_{uuid.uuid4().hex[:8]}" # Snapshot references both a normal file and an empty file snap = { "snapshot_id": snap_id, "manifest": { "README.md": uuid.uuid4().hex, ".museattributes": _EMPTY_OID, }, "created_at": _utc_now().isoformat(), } normal_obj = _make_object(b"readme content") empty_obj = {"object_id": _EMPTY_OID, "content": b"", "path": ".museattributes"} commit_id = uuid.uuid4().hex commit = _make_commit(repo.repo_id, commit_id=commit_id) commit["snapshot_id"] = snap_id resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({ "bundle": { "commits": [commit], "snapshots": [snap], "objects": [normal_obj, empty_obj], }, "branch": "main", "force": False, }), headers=wire_headers, ) assert resp.status_code == 200, resp.text data = msgpack.unpackb(resp.content, raw=False) assert data["ok"] is True # Verify empty object survives a filter-objects check (i.e. it was stored) filter_resp = await client.post( f"/{repo.owner}/{repo.slug}/filter-objects", content=_mp({"object_ids": [_EMPTY_OID]}), headers=wire_headers, ) assert filter_resp.status_code == 200 filter_data = msgpack.unpackb(filter_resp.content, raw=False) assert _EMPTY_OID not in filter_data["missing"], ( "empty object was dropped on push — 'not content' falsiness bug not fixed" ) @pytest.mark.asyncio async def test_fetch_objects_returns_empty_object_inline( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """fetch/objects must synthesize the empty-content object inline even if never stored in DB. The server synthesizes the EMPTY_OID on the fly when requested — no DB row or disk read needed. This covers repos pushed before the empty-object fix. """ repo = await factory_create_repo(db_session, slug="empty-obj-fetch-test", owner="test-user-wire") resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch/objects", content=_mp({"object_ids": [_EMPTY_OID]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) objects = data.get("objects", []) assert len(objects) == 1 assert objects[0]["object_id"] == _EMPTY_OID assert bytes(objects[0]["content"]) == b"" # ── push self-healing: re-push recovers missing bytes ───────────────────────── @pytest.mark.asyncio async def test_push_heals_missing_bytes( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Re-pushing an object whose bytes are missing from disk must restore them. Sequence: 1. Push normally — DB row created, bytes written to disk. 2. Delete the disk file to simulate data loss. 3. Re-push the same object — wire_push must detect missing bytes and re-write. 4. fetch/objects must return the original bytes. """ from musehub.storage.backends import get_backend repo = await factory_create_repo(db_session, slug="heal-missing-test", owner="test-user-wire") content = b"self-healing object content" obj = _make_object(content) snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit = _make_commit(repo.repo_id) commit["snapshot_id"] = snap_id # 1. Push normally push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({"bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main"}), headers=wire_headers, ) assert push_resp.status_code == 200, push_resp.text # 2. Delete the disk file to simulate data loss backend = get_backend() disk_path = backend._path(repo.repo_id, obj["object_id"]) assert disk_path.exists(), "bytes must be on disk after push" import stat as _stat disk_path.chmod(_stat.S_IRUSR | _stat.S_IWUSR) # make writable before unlink (put sets 0o444) disk_path.unlink() assert not disk_path.exists(), "sanity: file deleted" # 3. Re-push — self-healing must restore the bytes push_resp2 = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({"bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main"}), headers=wire_headers, ) assert push_resp2.status_code == 200, push_resp2.text # 4. fetch/objects must now return the correct bytes fetch_resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch/objects", content=_mp({"object_ids": [obj["object_id"]]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert fetch_resp.status_code == 200, fetch_resp.text data = msgpack.unpackb(fetch_resp.content, raw=False) objects = data.get("objects", []) assert len(objects) == 1 assert objects[0]["object_id"] == obj["object_id"] assert bytes(objects[0]["content"]) == content # ── fetch/objects endpoint (Phase 2 of two-phase fetch) ─────────────────────── @pytest.mark.asyncio async def test_fetch_objects_returns_bytes( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Phase 2: /fetch/objects streams the bytes for requested object IDs.""" repo = await factory_create_repo( db_session, slug="fetch-objects-test", owner="test-user-wire" ) content = b"phase two object content" obj = _make_object(content) snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit = _make_commit(repo.repo_id) commit["snapshot_id"] = snap_id push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({"bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main"}), headers=wire_headers, ) assert push_resp.status_code == 200, push_resp.text fetch_resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch/objects", content=_mp({"object_ids": [obj["object_id"]]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert fetch_resp.status_code == 200, fetch_resp.text data = msgpack.unpackb(fetch_resp.content, raw=False) objects = data.get("objects", []) assert len(objects) == 1 assert objects[0]["object_id"] == obj["object_id"] assert bytes(objects[0]["content"]) == content @pytest.mark.asyncio async def test_fetch_objects_unknown_ids_returns_empty( client: AsyncClient, db_session: AsyncSession, ) -> None: """Unknown object IDs are silently omitted — no 404.""" repo = await factory_create_repo(db_session, slug="fetch-objects-missing-test") resp = await client.post( f"/{repo.owner}/{repo.slug}/fetch/objects", content=_mp({"object_ids": ["nonexistent-id-abc123"]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert resp.status_code == 200 data = msgpack.unpackb(resp.content, raw=False) assert data.get("objects", []) == [] @pytest.mark.asyncio async def test_fetch_objects_cross_repo_dedup( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """fetch/objects must return bytes for objects whose DB row belongs to a different repo. Content-addressed objects are globally unique by hash. When repo_a pushes content X first, ``musehub_objects`` gets ``repo_id = A``. When repo_b pushes the same content, ``wire_push`` correctly skips the second byte-write (idempotent by hash), but no new DB row is created for repo_b. A subsequent ``POST /fetch/objects`` for repo_b must still return the bytes — not silently omit the object — because knowing the hash is a sufficient content commitment (same hash → same bytes, always). """ shared_oid = f"shared-dedup-oid-{uuid.uuid4().hex[:8]}" content = b"identical content pushed from two separate repos" # ── Push to repo_a: creates the DB row with repo_id = A ────────────────── repo_a = await factory_create_repo( db_session, slug="cross-repo-dedup-a", owner="test-user-wire" ) obj = {"object_id": shared_oid, "content": content, "path": "shared.txt"} snap_id_a = f"snap_{uuid.uuid4().hex[:8]}" snap_a = { "snapshot_id": snap_id_a, "manifest": {"shared.txt": shared_oid}, "created_at": _utc_now().isoformat(), } commit_a = _make_commit(repo_a.repo_id) commit_a["snapshot_id"] = snap_id_a push_a = await client.post( f"/{repo_a.owner}/{repo_a.slug}/push", content=_mp({ "bundle": {"commits": [commit_a], "snapshots": [snap_a], "objects": [obj]}, "branch": "main", }), headers=wire_headers, ) assert push_a.status_code == 200, push_a.text # ── Push same content to repo_b: server skips byte-write (idempotent) ──── # No new DB row is created; the existing row still has repo_id = A. repo_b = await factory_create_repo( db_session, slug="cross-repo-dedup-b", owner="test-user-wire" ) snap_id_b = f"snap_{uuid.uuid4().hex[:8]}" snap_b = { "snapshot_id": snap_id_b, "manifest": {"shared.txt": shared_oid}, "created_at": _utc_now().isoformat(), } commit_b = _make_commit(repo_b.repo_id) commit_b["snapshot_id"] = snap_id_b push_b = await client.post( f"/{repo_b.owner}/{repo_b.slug}/push", content=_mp({ "bundle": {"commits": [commit_b], "snapshots": [snap_b], "objects": [obj]}, "branch": "main", }), headers=wire_headers, ) assert push_b.status_code == 200, push_b.text # ── Phase 2 fetch from repo_b: bytes must be returned ──────────────────── # Bug: WHERE repo_id = B AND object_id = X returns nothing (row has repo_id = A). resp = await client.post( f"/{repo_b.owner}/{repo_b.slug}/fetch/objects", content=_mp({"object_ids": [shared_oid]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert resp.status_code == 200, resp.text data = msgpack.unpackb(resp.content, raw=False) objects = data.get("objects", []) assert len(objects) == 1, ( f"Expected 1 object from repo_b, got {len(objects)}. " "Cross-repo dedup bug: DB row has repo_id=A so WHERE repo_id=B misses it." ) assert objects[0]["object_id"] == shared_oid assert bytes(objects[0]["content"]) == content @pytest.mark.asyncio async def test_fetch_two_phase_separates_metadata_from_bytes( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Two-phase protocol: Phase 1 returns only metadata; Phase 2 returns bytes. This is the core architectural invariant: - POST /fetch → commits + snapshots + branch_heads (no objects key) - POST /fetch/objects → {"objects": [...]} with bytes Separation ensures fetch latency is proportional to commit count (always small), not to repo object size (can be gigabytes). """ repo = await factory_create_repo( db_session, slug="two-phase-fetch-test", owner="test-user-wire" ) content = b"some file content for two-phase test" obj = _make_object(content) snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit_id = uuid.uuid4().hex commit = _make_commit(repo.repo_id, commit_id=commit_id) commit["snapshot_id"] = snap_id push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp({"bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main"}), headers=wire_headers, ) assert push_resp.status_code == 200 # Phase 1: metadata only — no objects in response phase1 = await client.post( f"/{repo.owner}/{repo.slug}/fetch", content=_mp({"want": [commit_id], "have": []}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert phase1.status_code == 200 p1_data = msgpack.unpackb(phase1.content, raw=False) assert len(p1_data["commits"]) == 1 assert p1_data["commits"][0]["commit_id"] == commit_id assert "objects" not in p1_data or p1_data.get("objects") is None # Phase 2: request the specific object ID — get bytes back phase2 = await client.post( f"/{repo.owner}/{repo.slug}/fetch/objects", content=_mp({"object_ids": [obj["object_id"]]}), headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}, ) assert phase2.status_code == 200 p2_data = msgpack.unpackb(phase2.content, raw=False) objects = p2_data.get("objects", []) assert len(objects) == 1 assert objects[0]["object_id"] == obj["object_id"] assert bytes(objects[0]["content"]) == content # ── performance regression ───────────────────────────────────────────────────── @pytest.mark.asyncio async def test_push_large_bundle_completes_under_budget( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Large push bundle (100 commits × 1 snapshot each) must complete in < 5s. This is the regression guard for the N+1 query bug fixed in wire_push: before the fix, 856 commits × per-row session.get() calls took 4m43s. After the fix (2 bulk SELECT INs + zero-DB-call loop), the same push is effectively O(1) round-trips regardless of bundle size. 100 commits is enough to catch a regression without inflating test time. """ _BUDGET_SECONDS = 5.0 _COMMIT_COUNT = 100 repo = await factory_create_repo( db_session, slug="perf-large-bundle", owner="test-user-wire" ) # Build a linear chain: commit_0 ← commit_1 ← … ← commit_99 commits: list[dict] = [] snapshots: list[dict] = [] objects: list[dict] = [] prev_id: str | None = None for i in range(_COMMIT_COUNT): obj = _make_object(f"file content {i}".encode()) snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit_id = uuid.uuid4().hex commit = _make_commit(repo.repo_id, commit_id=commit_id, parent=prev_id) commit["snapshot_id"] = snap_id objects.append(obj) snapshots.append(snap) commits.append(commit) prev_id = commit_id body = _mp({ "bundle": { "commits": commits, "snapshots": snapshots, "objects": objects, }, "branch": "main", }) t0 = time.monotonic() resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=body, headers=wire_headers, ) elapsed = time.monotonic() - t0 assert resp.status_code == 200, f"push failed: {resp.text}" assert elapsed < _BUDGET_SECONDS, ( f"push of {_COMMIT_COUNT} commits took {elapsed:.2f}s — " f"exceeds {_BUDGET_SECONDS}s budget (N+1 regression?)" ) # ── structured_delta serialization regression tests ─────────────────────────── @pytest.mark.asyncio async def test_push_commit_with_null_structured_delta( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """A commit with structured_delta=null must persist without 500.""" repo = await factory_create_repo(db_session, slug="delta-null-test", owner="test-user-wire") commit_id = uuid.uuid4().hex obj = _make_object() snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit = _make_commit(repo.repo_id, commit_id=commit_id) commit["snapshot_id"] = snap_id commit["structured_delta"] = None # explicit null payload = { "bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main", "force": False, } resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers, ) assert resp.status_code == 200, resp.text data = msgpack.unpackb(resp.content, raw=False) assert data["ok"] is True @pytest.mark.asyncio async def test_push_commit_with_nested_structured_delta( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """A commit with a nested-dict structured_delta must persist without 500. Regression: wire_push was calling .root on PydanticJson, which only unwraps one level — nested dicts/lists still contained PydanticJson instances that SQLAlchemy's json_serializer could not encode. Fix: use unwrap() for full recursive unwrapping. """ repo = await factory_create_repo(db_session, slug="delta-nested-test", owner="test-user-wire") commit_id = uuid.uuid4().hex obj = _make_object() snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit = _make_commit(repo.repo_id, commit_id=commit_id) commit["snapshot_id"] = snap_id commit["structured_delta"] = { "kind": "code", "symbols": ["MyClass", "my_func"], "stats": {"added": 10, "removed": 2}, "nested": {"deep": {"value": True}}, } payload = { "bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main", "force": False, } resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers, ) assert resp.status_code == 200, resp.text data = msgpack.unpackb(resp.content, raw=False) assert data["ok"] is True @pytest.mark.asyncio async def test_push_commit_with_list_structured_delta( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """A commit with a list-typed structured_delta must persist without 500.""" repo = await factory_create_repo(db_session, slug="delta-list-test", owner="test-user-wire") commit_id = uuid.uuid4().hex obj = _make_object() snap_id = f"snap_{uuid.uuid4().hex[:8]}" snap = _make_snapshot(snap_id, obj["object_id"]) commit = _make_commit(repo.repo_id, commit_id=commit_id) commit["snapshot_id"] = snap_id commit["structured_delta"] = [{"op": "add", "path": "/foo"}, {"op": "remove", "path": "/bar"}] payload = { "bundle": {"commits": [commit], "snapshots": [snap], "objects": [obj]}, "branch": "main", "force": False, } resp = await client.post( f"/{repo.owner}/{repo.slug}/push", content=_mp(payload), headers=wire_headers, ) assert resp.status_code == 200, resp.text data = msgpack.unpackb(resp.content, raw=False) assert data["ok"] is True