"""TDD — MWP v2 streaming push: eight-tier test coverage. Tier map -------- T1 Unit — frame codec helpers (_sp, _prog, _err, _result) T2 Unit — server protocol state machine (pure frame dispatch, no DB/R2) T3 Component — object validation (hash check, size limits, enc modes) T4 Component — commit-pack validation (schema, limits, signature gate) T5 Service — wire_push_stream() async generator against in-memory stubs T6 Integration — service layer against real test DB, stub R2 backend T7 Route — POST /push/stream via ASGI test client (StreamingResponse) T8 E2E — complete round-trip: push objects + commits → GET /refs confirms head All frame construction uses the canonical SFRAME_* constants from ``musehub.models.wire`` so the tests act as a contract for the wire format itself — any change to the frame shape will break these tests first. """ from __future__ import annotations import asyncio import struct import zlib from collections.abc import AsyncGenerator, AsyncIterator from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch import msgpack import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import blob_id, fake_id, now_utc_iso from musehub.db.musehub_models import MusehubRepo from musehub.types.json_types import JSONObject, JSONValue, StrDict from musehub.models.wire import ( SFRAME_COMMIT_PACK, SFRAME_END, SFRAME_ERROR, SFRAME_HEADER, SFRAME_OBJECT, SFRAME_PROGRESS, SFRAME_RESULT, STREAM_MAX_COMMITS, STREAM_MAX_OBJECT_WIRE_BYTES, STREAM_MAX_OBJECTS, ) from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter from tests.factories import create_repo _fw = MuseWireFrameWriter() # --------------------------------------------------------------------------- # Shared codec helpers # --------------------------------------------------------------------------- def _pack(data: JSONValue) -> bytes: """Encode one msgpack frame payload (without transport envelope).""" return msgpack.packb(data, use_bin_type=True) def _wrap(ft: str, data: JSONValue) -> bytes: """Encode and wrap in a raw MWP envelope — no gRPC prefix.""" return _fw.wrap(frame_type=ft, payload=_pack(data)) def _unpack_all(raw: bytes) -> list[dict]: """Decode concatenated raw MWP frames into a list of payload dicts. Each MWP frame: magic(4=b"muse") | version(1) | header_len(4) | header(N) | payload_len(8) | payload(M) """ import struct results = [] offset = 0 while offset + 17 <= len(raw): if raw[offset:offset + 4] != b"muse": break header_len = struct.unpack(">I", raw[offset + 5:offset + 9])[0] pl_start = offset + 9 + header_len if pl_start + 8 > len(raw): break payload_len = struct.unpack(">Q", raw[pl_start:pl_start + 8])[0] payload = raw[pl_start + 8:pl_start + 8 + payload_len] results.append(msgpack.unpackb(payload, raw=False)) offset = pl_start + 8 + payload_len return results def _last_frame(raw: bytes) -> JSONObject: """Return the last msgpack object from a push-stream response body.""" unpacker = msgpack.Unpacker(raw=False) unpacker.feed(raw) last: JSONObject = {} for frame in unpacker: last = frame return last def _make_obj_bytes(content: bytes = b"hello world") -> tuple[str, bytes]: """Return (sha256_oid, raw_content) for a test object.""" oid = blob_id(content) return oid, content def _header_frame( branch: str = "main", force: bool = False, have: list[str] | None = None, head: str = "sha256:abc", n_objects: int = 0, n_commits: int = 1, ) -> bytes: return _wrap(SFRAME_HEADER, { "t": SFRAME_HEADER, "branch": branch, "force": force, "have": have or [], "head": head, "n_objects": n_objects, "n_commits": n_commits, }) def _object_frame( oid: str, content: bytes, path: str = "track.wav", enc: str = "raw", base: str = "", ) -> bytes: payload: dict = { "t": SFRAME_OBJECT, "id": oid, "content": content, "path": path, "enc": enc, } if base: payload["base"] = base return _wrap(SFRAME_OBJECT, payload) def _commit_pack_frame(commits: list[dict], snapshots: list[dict] | None = None) -> bytes: return _wrap(SFRAME_COMMIT_PACK, { "t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots or [], }) def _end_frame(n_objects: int = 0, n_commits: int = 0) -> bytes: return _wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits}) def _make_commit( commit_id: str | None = None, parent_ids: list[str] | None = None, snapshot_id: str | None = None, branch: str = "main", author: str = "gabriel", ) -> JSONObject: cid = commit_id or blob_id(f"commit-{now_utc_iso()}".encode()) pids = parent_ids or [] return { "commit_id": cid, "parent_ids": pids, "parent_commit_id": pids[0] if len(pids) > 0 else None, "parent2_commit_id": pids[1] if len(pids) > 1 else None, "snapshot_id": snapshot_id or blob_id(b"default-snap"), "branch": branch, "message": "test commit", "author": author, "committed_at": now_utc_iso(), "signature": "", "signer_key_id": "", "agent_id": "", "model_id": "", "metadata": {}, } def _make_snapshot(snapshot_id: str, manifest: JSONObject | None = None) -> JSONObject: return { "snapshot_id": snapshot_id, "manifest": manifest or {}, "committed_at": now_utc_iso(), } async def _collect_frames(gen: AsyncGenerator[bytes, None]) -> list[dict]: """Drain an async generator of plain msgpack frame bytes into a list of dicts.""" unpacker = msgpack.Unpacker(raw=False) async for chunk in gen: unpacker.feed(chunk) return list(unpacker) async def _body_iter(*frames: bytes) -> AsyncIterator[bytes]: """Wrap pre-built frames as an async iterator for wire_push_stream().""" for f in frames: yield f # --------------------------------------------------------------------------- # T1 — Unit: frame codec helpers # --------------------------------------------------------------------------- class TestT1FrameCodec: """Tier 1: verify frame construction and MIME constant.""" def test_stream_mime_type(self) -> None: assert WIRE_CONTENT_TYPE == "application/x-muse-wire" def test_sframe_constants_are_single_chars(self) -> None: for const in ( SFRAME_HEADER, SFRAME_OBJECT, SFRAME_COMMIT_PACK, SFRAME_END, SFRAME_PROGRESS, SFRAME_ERROR, SFRAME_RESULT, ): assert len(const) == 1, f"{const!r} should be a single character" def test_client_server_frame_tags_are_disjoint(self) -> None: client_tags = {SFRAME_HEADER, SFRAME_OBJECT, SFRAME_COMMIT_PACK, SFRAME_END} server_tags = {SFRAME_PROGRESS, SFRAME_ERROR, SFRAME_RESULT} assert client_tags.isdisjoint(server_tags), ( "client and server frame tags must not overlap" ) def test_header_frame_round_trips(self) -> None: raw = _header_frame(branch="dev", n_objects=3, n_commits=2) frames = _unpack_all(raw) assert len(frames) == 1 f = frames[0] assert f["t"] == SFRAME_HEADER assert f["branch"] == "dev" assert f["n_objects"] == 3 assert f["n_commits"] == 2 def test_object_frame_round_trips(self) -> None: oid, content = _make_obj_bytes(b"muse track data") raw = _object_frame(oid, content, path="beat.mid") frames = _unpack_all(raw) f = frames[0] assert f["t"] == SFRAME_OBJECT assert f["id"] == oid assert bytes(f["content"]) == content def test_commit_pack_frame_round_trips(self) -> None: commit = _make_commit() raw = _commit_pack_frame([commit]) frames = _unpack_all(raw) f = frames[0] assert f["t"] == SFRAME_COMMIT_PACK assert len(f["commits"]) == 1 def test_end_frame_round_trips(self) -> None: frames = _unpack_all(_end_frame()) assert frames[0]["t"] == SFRAME_END def test_multiple_frames_concatenated_parse_correctly(self) -> None: body = ( _header_frame() + _object_frame(*_make_obj_bytes()) + _end_frame() ) frames = _unpack_all(body) assert [f["t"] for f in frames] == [SFRAME_HEADER, SFRAME_OBJECT, SFRAME_END] def test_stream_limits_are_positive(self) -> None: assert STREAM_MAX_OBJECTS > 0 assert STREAM_MAX_COMMITS > 0 assert STREAM_MAX_OBJECT_WIRE_BYTES > 0 # --------------------------------------------------------------------------- # T2 — Unit: server helper functions # --------------------------------------------------------------------------- class TestT2ServerHelpers: """Tier 2: test _sp, _prog, _err, _result frame builders.""" def _import_helpers(self) -> None: from musehub.services.musehub_wire import _sp, _prog, _err, _result return _sp, _prog, _err, _result def test_prog_encodes_progress_frame(self) -> None: _, _prog, _, _ = self._import_helpers() f = msgpack.unpackb(_prog("uploading objects"), raw=False) assert f["t"] == SFRAME_PROGRESS assert f["msg"] == "uploading objects" def test_err_encodes_error_frame_with_code(self) -> None: _, _, _err, _ = self._import_helpers() f = msgpack.unpackb(_err("repo not found", 404), raw=False) assert f["t"] == SFRAME_ERROR assert f["code"] == 404 assert "repo not found" in f["msg"] def test_err_default_code_is_400(self) -> None: _, _, _err, _ = self._import_helpers() f = msgpack.unpackb(_err("bad request"), raw=False) assert f["code"] == 400 def test_result_ok_encodes_correctly(self) -> None: _, _, _, _result = self._import_helpers() heads = {"main": "sha256:abc"} f = msgpack.unpackb(_result(True, "pushed", heads, "sha256:abc"), raw=False) assert f["t"] == SFRAME_RESULT assert f["ok"] is True assert f["heads"] == heads def test_result_failure_encodes_correctly(self) -> None: _, _, _, _result = self._import_helpers() f = msgpack.unpackb(_result(False, "rejected", {}, ""), raw=False) assert f["ok"] is False # --------------------------------------------------------------------------- # T3 — Component: object validation edge cases # --------------------------------------------------------------------------- class TestT3ObjectValidation: """Tier 3: hash mismatch, size limits, compression encoding.""" def _oid_for(self, raw: bytes) -> str: return blob_id(raw) def test_sha256_oid_format(self) -> None: raw = b"guitar riff" oid = self._oid_for(raw) assert oid.startswith("sha256:") assert len(oid) == len("sha256:") + 64 def test_zlib_object_frame_decompresses_correctly(self) -> None: raw = b"MIDI note data " * 50 compressed = zlib.compress(raw) oid, _ = _make_obj_bytes(raw) # sha256 of raw (before compression) frame_bytes = _object_frame(oid, compressed, enc="zlib") frames = _unpack_all(frame_bytes) f = frames[0] assert f["enc"] == "zlib" assert zlib.decompress(bytes(f["content"])) == raw def test_wire_size_limit_constant_is_reasonable(self) -> None: # Must be at least 1 MB and at most 512 MB per object wire payload. assert 1 * 1024 * 1024 <= STREAM_MAX_OBJECT_WIRE_BYTES <= 512 * 1024 * 1024 def test_object_frame_with_empty_content_is_encodable(self) -> None: raw = b"" oid = blob_id(raw) frame_bytes = _object_frame(oid, raw) frames = _unpack_all(frame_bytes) assert bytes(frames[0]["content"]) == b"" def test_large_object_frame_exceeds_limit_is_detectable(self) -> None: """Frame content larger than STREAM_MAX_OBJECT_WIRE_BYTES should be flagged.""" oversized = b"x" * (STREAM_MAX_OBJECT_WIRE_BYTES + 1) oid = blob_id(oversized) frame = _unpack_all(_object_frame(oid, oversized))[0] assert len(bytes(frame["content"])) > STREAM_MAX_OBJECT_WIRE_BYTES def test_raw_encoding_preserves_exact_bytes(self) -> None: raw = b"\x00\x01\x02\x03" * 100 oid = blob_id(raw) frame = _unpack_all(_object_frame(oid, raw, enc="raw"))[0] assert bytes(frame["content"]) == raw def test_delta_object_frame_carries_base_field(self) -> None: """O-frame with delta encoding must carry 'base', not 'base_id'.""" base_raw = b"old content for delta base" target_raw = b"new content for delta target -- modified" base_oid = blob_id(base_raw) target_oid = blob_id(target_raw) delta = _compute_delta(base_raw, target_raw) frame = _unpack_all(_object_frame(target_oid, delta, enc="delta+zlib", base=base_oid))[0] assert frame["enc"] == "delta+zlib" assert "base" in frame assert "base_id" not in frame assert frame["base"] == base_oid assert frame["id"] == target_oid # --------------------------------------------------------------------------- # Delta helpers # --------------------------------------------------------------------------- def _compute_delta(base: bytes, target: bytes) -> bytes: """Minimal delta encoder matching muse.core.compression.compute_delta format.""" raw = b"\x01" + struct.pack(">I", len(target)) + target return zlib.compress(raw, level=1) # --------------------------------------------------------------------------- # T4 — Component: commit-pack schema validation # --------------------------------------------------------------------------- class TestT4CommitPackValidation: """Tier 4: WireCommit/WireSnapshot schema, commit-count limit.""" def test_minimal_commit_passes_model_validate(self) -> None: from musehub.models.wire import WireCommit commit = _make_commit() obj = WireCommit.model_validate(commit) assert obj.commit_id == commit["commit_id"] def test_commit_without_required_fields_raises(self) -> None: from musehub.models.wire import WireCommit import pydantic with pytest.raises((pydantic.ValidationError, Exception)): WireCommit.model_validate({"message": "incomplete"}) def test_snapshot_passes_model_validate(self) -> None: from musehub.models.wire import WireSnapshot sid = blob_id(b"snap") snap = _make_snapshot(sid, {"file.wav": sid}) obj = WireSnapshot.model_validate(snap) assert obj.snapshot_id == sid def test_commit_pack_limit_constant(self) -> None: assert STREAM_MAX_COMMITS >= 1_000 def test_commit_pack_frame_encodes_many_commits(self) -> None: commits = [_make_commit() for _ in range(10)] frame = _unpack_all(_commit_pack_frame(commits))[0] assert len(frame["commits"]) == 10 def test_signed_commit_has_signature_fields(self) -> None: from musehub.models.wire import WireCommit commit = _make_commit() commit["signature"] = "sig_base64" commit["signer_key_id"] = "key123" commit["agent_id"] = "agent-1" commit["model_id"] = "claude-opus-4-6" obj = WireCommit.model_validate(commit) assert obj.signature == "sig_base64" def test_wire_commit_has_branch_field(self) -> None: """WireCommit.branch is the canonical field name — mirrors CommitRecord.branch.""" from musehub.models.wire import WireCommit commit = _make_commit() obj = WireCommit.model_validate(commit) assert hasattr(obj, "branch"), "WireCommit must have a 'branch' field" def test_wire_commit_has_no_created_on_branch_field(self) -> None: """WireCommit must not expose 'created_on_branch' — that name is retired.""" from musehub.models.wire import WireCommit commit = _make_commit() obj = WireCommit.model_validate(commit) assert not hasattr(obj, "created_on_branch"), ( "WireCommit must not have a 'created_on_branch' attribute" ) def test_wire_commit_branch_populated_from_branch_key(self) -> None: """Wire payload with 'branch' key populates WireCommit.branch correctly.""" from musehub.models.wire import WireCommit commit = _make_commit() commit["branch"] = "task/my-feature" obj = WireCommit.model_validate(commit) assert obj.branch == "task/my-feature" def test_wire_commit_branch_empty_by_default(self) -> None: """WireCommit.branch defaults to empty string when omitted.""" from musehub.models.wire import WireCommit commit = _make_commit() commit.pop("branch", None) commit.pop("created_on_branch", None) obj = WireCommit.model_validate(commit) assert obj.branch == "" # --------------------------------------------------------------------------- # T5 — Service: wire_push_stream() with stubbed DB and R2 backend # --------------------------------------------------------------------------- class TestT5ServiceStream: """Tier 5: wire_push_stream() async generator against minimal stubs. We patch the storage backend and DB session so tests run without infrastructure — only the frame-parsing and protocol state machine are exercised here. """ @pytest.fixture() def stub_backend(self, monkeypatch: pytest.MonkeyPatch) -> MagicMock: backend = AsyncMock() backend.exists = AsyncMock(return_value=False) backend.put = AsyncMock(return_value="https://r2.example.com/obj") backend.get = AsyncMock(return_value=b"raw bytes") monkeypatch.setattr( "musehub.services.musehub_wire.get_backend", lambda: backend, ) return backend @pytest.fixture() def stub_session(self) -> AsyncMock: session = AsyncMock(spec=AsyncSession) session.execute = AsyncMock(return_value=MagicMock(scalar=lambda: None, fetchall=lambda: [])) session.commit = AsyncMock() session.add = MagicMock() return session @pytest.mark.asyncio async def test_missing_header_yields_error( self, stub_backend: MagicMock, stub_session: AsyncMock ) -> None: from musehub.services.musehub_wire import wire_push_stream async def body() -> None: yield _object_frame(*_make_obj_bytes()) + _end_frame() frames = await _collect_frames( wire_push_stream(stub_session, "repo-id", body(), "gabriel") ) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert error_frames, "expected an ERROR frame when OBJECT sent before HEADER" assert "OBJECT frame before HEADER" in error_frames[0]["msg"] @pytest.mark.asyncio async def test_missing_end_frame_yields_error( self, stub_backend: MagicMock, stub_session: AsyncMock ) -> None: from musehub.services.musehub_wire import wire_push_stream commit = _make_commit() snap_id = blob_id(b"snap") snap = _make_snapshot(snap_id) commit["snapshot_id"] = snap_id async def body() -> None: yield _header_frame() + _commit_pack_frame([commit], [snap]) # no END frame frames = await _collect_frames( wire_push_stream(stub_session, "repo-id", body(), "gabriel") ) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert error_frames assert "without END frame" in error_frames[0]["msg"] @pytest.mark.asyncio async def test_missing_commit_pack_yields_error( self, stub_backend: MagicMock, stub_session: AsyncMock ) -> None: from musehub.services.musehub_wire import wire_push_stream async def body() -> None: yield _header_frame() + _end_frame() frames = await _collect_frames( wire_push_stream(stub_session, "repo-id", body(), "gabriel") ) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert error_frames assert "COMMIT_PACK" in error_frames[0]["msg"] @pytest.mark.asyncio async def test_object_hash_mismatch_yields_error( self, stub_backend: MagicMock, stub_session: AsyncMock ) -> None: from musehub.services.musehub_wire import wire_push_stream oid = fake_id("wrong-hash-object") # wrong hash content = b"this content does not match the oid" async def body() -> None: yield _header_frame(n_objects=1) + _object_frame(oid, content) frames = await _collect_frames( wire_push_stream(stub_session, "repo-id", body(), "gabriel") ) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert error_frames @pytest.mark.asyncio async def test_progress_frames_emitted_on_header( self, stub_backend: MagicMock, stub_session: AsyncMock ) -> None: from musehub.services.musehub_wire import wire_push_stream commit = _make_commit() snap_id = blob_id(b"snap-data") snap = _make_snapshot(snap_id) commit["snapshot_id"] = snap_id async def body() -> None: yield _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame() frames = await _collect_frames( wire_push_stream(stub_session, "repo-id", body(), "gabriel") ) progress_frames = [f for f in frames if f.get("t") == SFRAME_PROGRESS] assert progress_frames, "expected at least one PROGRESS frame" @pytest.mark.asyncio async def test_delta_zlib_object_reconstructed_correctly( self, stub_backend: MagicMock, stub_session: AsyncMock ) -> None: """T5: server correctly reconstructs a delta+zlib object and verifies its hash. This is the exact scenario that caused the push hash mismatch bug: a delta-encoded object must be decompressed and reconstructed before sha256 verification, not hashed as raw delta bytes. """ from musehub.services.musehub_wire import wire_push_stream base_raw = b"base content: the old version of the file\n" * 20 target_raw = b"target content: the new version of the file\n" * 20 base_oid = blob_id(base_raw) target_oid = blob_id(target_raw) delta = _compute_delta(base_raw, target_raw) # Server must return the base object when asked. stub_backend.get = AsyncMock(return_value=base_raw) stub_backend.exists = AsyncMock(return_value=False) commit = _make_commit() snap_id = blob_id(b"delta-test-snap") snap = _make_snapshot(snap_id, {"src/main.py": target_oid}) commit["snapshot_id"] = snap_id async def body() -> None: yield ( _header_frame(n_objects=1, n_commits=1) + _object_frame(target_oid, delta, enc="delta+zlib", base=base_oid) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=1, n_commits=1) ) frames = await _collect_frames( wire_push_stream(stub_session, "repo-id", body(), "gabriel") ) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert not error_frames, ( f"unexpected ERROR frame(s): {[f.get('msg') for f in error_frames]}" ) @pytest.mark.asyncio async def test_delta_zlib_wrong_base_yields_error( self, stub_backend: MagicMock, stub_session: AsyncMock ) -> None: """T5: missing base object yields a 422-style error frame, not hash mismatch.""" from musehub.services.musehub_wire import wire_push_stream base_oid = blob_id(b"base that does not exist on server") target_raw = b"target content" target_oid = blob_id(target_raw) delta = _compute_delta(b"", target_raw) stub_backend.get = AsyncMock(return_value=None) # base not found stub_backend.exists = AsyncMock(return_value=False) async def body() -> None: yield ( _header_frame(n_objects=1) + _object_frame(target_oid, delta, enc="delta+zlib", base=base_oid) + _end_frame(n_objects=1) ) frames = await _collect_frames( wire_push_stream(stub_session, "repo-id", body(), "gabriel") ) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert error_frames, "expected an ERROR frame when delta base is missing" # --------------------------------------------------------------------------- # T6 — Integration: service against real DB, stub R2 # --------------------------------------------------------------------------- def _stub_r2_backend(monkeypatch: pytest.MonkeyPatch) -> None: """Patch the R2 backend with an in-memory dict store.""" _store: dict[str, bytes] = {} async def _exists(oid: str, **_: JSONValue) -> bool: return oid in _store async def _put(oid: str, data: bytes, **kwargs: JSONValue) -> str: _store[oid] = data return f"https://r2.fake/{oid}" async def _get(oid: str) -> bytes | None: return _store.get(oid) backend = AsyncMock() backend.exists = _exists backend.put = _put backend.get = _get monkeypatch.setattr( "musehub.services.musehub_wire.get_backend", lambda: backend, ) async def _make_repo(db_session: AsyncSession, name: str, owner: str = "gabriel") -> MusehubRepo: """Create a repo row + main branch, committed and visible to any session.""" from datetime import datetime, timezone from musehub.db.musehub_models import MusehubRepo, MusehubBranch from musehub.core.genesis import compute_identity_id, compute_repo_id, compute_branch_id owner_user_id = compute_identity_id(owner.encode()) slug = name.lower().replace(" ", "-") created_at = datetime.now(tz=timezone.utc) repo_id = compute_repo_id(owner_user_id, slug, "code", created_at.isoformat()) repo = MusehubRepo( repo_id=repo_id, name=name, owner=owner, slug=slug, visibility="public", owner_user_id=owner_user_id, description="", tags=[], created_at=created_at, ) db_session.add(repo) await db_session.commit() branch = MusehubBranch( branch_id=compute_branch_id(repo_id, "main"), repo_id=repo_id, name="main", ) db_session.add(branch) await db_session.commit() await db_session.refresh(repo) return repo @pytest.mark.asyncio async def test_t6_push_single_commit_no_objects( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """T6: push one commit with no objects against the real test DB.""" from musehub.services.musehub_wire import wire_push_stream _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T6 Single Commit") snap_id = blob_id(b"t6-snap-1") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id) async def body() -> None: yield _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) frames = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel") ) result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT] assert result_frames, f"no RESULT frame; frames: {[f.get('t') for f in frames]}" assert result_frames[0]["ok"] is True @pytest.mark.asyncio async def test_t6_push_with_object( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """T6: push one object + commit; confirm RESULT.ok.""" from musehub.services.musehub_wire import wire_push_stream _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T6 With Object") raw = b"audio data for test track" oid = blob_id(raw) snap_id = blob_id(b"t6-snap-obj") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"track.wav": oid}) async def body() -> None: yield ( _header_frame(n_objects=1, n_commits=1) + _object_frame(oid, raw) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=1, n_commits=1) ) frames = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel") ) result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT] assert result_frames and result_frames[0]["ok"] is True @pytest.mark.asyncio async def test_t6_push_zlib_compressed_object( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """T6: push a zlib-compressed object; server decompresses and confirms.""" from musehub.services.musehub_wire import wire_push_stream _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T6 Zlib") raw = b"raw MIDI data " * 100 compressed = zlib.compress(raw) oid = blob_id(raw) snap_id = blob_id(b"t6-snap-zlib") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"beat.mid": oid}) async def body() -> None: yield ( _header_frame(n_objects=1, n_commits=1) + _object_frame(oid, compressed, enc="zlib") + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=1, n_commits=1) ) frames = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel") ) result = [f for f in frames if f.get("t") == SFRAME_RESULT] assert result and result[0]["ok"] is True @pytest.mark.asyncio async def test_t6_force_push_advances_branch( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """T6: two sequential pushes — second uses force=True to advance divergent branch.""" from musehub.services.musehub_wire import wire_push_stream _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T6 Force Push") snap_id = blob_id(b"t6-snap-force-1") commit1 = _make_commit(snapshot_id=snap_id) snap1 = _make_snapshot(snap_id) async def body1() -> None: yield _header_frame() + _commit_pack_frame([commit1], [snap1]) + _end_frame(n_commits=1) frames1 = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body1(), "gabriel") ) assert any(f.get("t") == SFRAME_RESULT and f["ok"] for f in frames1) snap_id2 = blob_id(b"t6-snap-force-2") commit2 = _make_commit(snapshot_id=snap_id2) snap2 = _make_snapshot(snap_id2) async def body2() -> None: yield _header_frame(force=True) + _commit_pack_frame([commit2], [snap2]) + _end_frame(n_commits=1) frames2 = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body2(), "gabriel") ) result2 = [f for f in frames2 if f.get("t") == SFRAME_RESULT] assert result2 and result2[0]["ok"] is True # --------------------------------------------------------------------------- # T6 — Provenance columns: agent_id, model_id, commit_branch as DB columns # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_t6_push_stores_agent_id_and_model_id_as_columns( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """T6/provenance: wire push writes agent_id and model_id as first-class columns.""" from sqlalchemy import select as _select from musehub.db.musehub_models import MusehubCommit from musehub.services.musehub_wire import wire_push_stream _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T6 Provenance Columns") snap_id = blob_id(b"t6-prov-snap") commit = _make_commit(snapshot_id=snap_id) commit["agent_id"] = "claude-code" commit["model_id"] = "claude-sonnet-4-6" snap = _make_snapshot(snap_id) async def body() -> None: yield _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) frames = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel") ) assert any(f.get("t") == SFRAME_RESULT and f["ok"] for f in frames) row = (await db_session.execute( _select(MusehubCommit).where(MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.agent_id == "claude-code", f"agent_id column should be 'claude-code', got {row.agent_id!r}" assert row.model_id == "claude-sonnet-4-6", f"model_id column should be 'claude-sonnet-4-6', got {row.model_id!r}" @pytest.mark.asyncio async def test_t6_push_stores_commit_branch_from_wire_commit( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """T6/provenance: commit_branch stores author branch, not push-target branch. The push header says branch='main' (push target), but the commit record carries branch='task/my-feature' (where the author worked). The DB column commit_branch must reflect WireCommit.branch, not the push target. """ from sqlalchemy import select as _select from musehub.db.musehub_models import MusehubCommit from musehub.services.musehub_wire import wire_push_stream _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T6 Commit Branch") snap_id = blob_id(b"t6-cbranch-snap") commit = _make_commit(snapshot_id=snap_id, branch="task/my-feature") snap = _make_snapshot(snap_id) # Push header targets 'main', but the commit itself was authored on 'task/my-feature' async def body() -> None: yield ( _header_frame(branch="main", n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) ) frames = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel") ) assert any(f.get("t") == SFRAME_RESULT and f["ok"] for f in frames) row = (await db_session.execute( _select(MusehubCommit).where(MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.commit_branch == "task/my-feature", ( f"commit_branch should be WireCommit.branch 'task/my-feature', got {row.commit_branch!r}" ) @pytest.mark.asyncio async def test_t6_push_commit_branch_empty_when_wire_commit_has_no_branch( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """T6/provenance: commit_branch is None when WireCommit.branch is empty.""" from sqlalchemy import select as _select from musehub.db.musehub_models import MusehubCommit from musehub.services.musehub_wire import wire_push_stream _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T6 No Commit Branch") snap_id = blob_id(b"t6-nobranch-snap") commit = _make_commit(snapshot_id=snap_id, branch="") snap = _make_snapshot(snap_id) async def body() -> None: yield _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) frames = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel") ) assert any(f.get("t") == SFRAME_RESULT and f["ok"] for f in frames) row = (await db_session.execute( _select(MusehubCommit).where(MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.commit_branch is None or row.commit_branch == "", ( f"commit_branch should be None or '' when WireCommit.branch is empty, got {row.commit_branch!r}" ) @pytest.mark.asyncio async def test_t6_push_delta_zlib_object( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """T6: push a delta+zlib object — server must reconstruct and verify hash. Regression test for the push hash mismatch bug: when a client sends enc='delta+zlib', the server must apply the delta against the stored base to reconstruct the target, then verify sha256(reconstructed) == declared oid. Previously untested, which is why the bug lived undetected. """ from musehub.services.musehub_wire import wire_push_stream _store: dict[str, bytes] = {} base_raw = b"original file content: line one\nline two\nline three\n" * 10 target_raw = b"modified file content: line one\nline two CHANGED\nline three\n" * 10 base_oid = blob_id(base_raw) target_oid = blob_id(target_raw) delta = _compute_delta(base_raw, target_raw) # Pre-populate the base object in storage (as if it was pushed in a prior push). _store[base_oid] = base_raw async def _exists(oid: str, **_: object) -> bool: return oid in _store async def _put(oid: str, data: bytes, **_: object) -> str: _store[oid] = data return f"https://r2.fake/{oid}" async def _get(oid: str, **_: object) -> bytes | None: return _store.get(oid) backend = AsyncMock() backend.exists = _exists backend.put = _put backend.get = _get monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) repo = await _make_repo(db_session, "T6 Delta Zlib Push") snap_id = blob_id(b"t6-delta-snap") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"src/main.py": target_oid}) async def body() -> None: yield ( _header_frame(n_objects=1, n_commits=1) + _object_frame(target_oid, delta, enc="delta+zlib", base=base_oid) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=1, n_commits=1) ) frames = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body(), "gabriel") ) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert not error_frames, ( f"unexpected ERROR: {[f.get('msg') for f in error_frames]}" ) result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT] assert result_frames and result_frames[0]["ok"] is True # Verify server stored the reconstructed (raw) bytes, not the delta bytes. assert _store.get(target_oid) == target_raw, ( "server must store reconstructed raw content, not delta bytes" ) # --------------------------------------------------------------------------- # T7 — Route: ASGI test client hitting POST /{owner}/{slug}/push/stream # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_t7_push_stream_returns_200_with_packstream_content_type( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T7: route returns 200 with application/x-muse-packstream content-type.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T7 Route Test 1", owner="testuser") snap_id = blob_id(b"t7-snap-ct") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id) body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code == 200 assert resp.headers.get("content-type", "").startswith(WIRE_CONTENT_TYPE) @pytest.mark.asyncio async def test_t7_push_stream_response_contains_result_frame( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T7: response body is a plain msgpack dict with t=RESULT.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T7 Route Test 2", owner="testuser") snap_id = blob_id(b"t7-snap-result") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id) body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _last_frame(resp.content) assert result.get("ok") is True, f"expected ok=True result, got: {result}" @pytest.mark.asyncio async def test_t7_push_stream_requires_auth( client: AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """T7: unauthenticated push yields error.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T7 Route Auth", owner="testuser") body = _header_frame() + _end_frame() resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={"Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code in (200, 401, 403) if resp.status_code == 200: result = _last_frame(resp.content) assert result.get("t") == SFRAME_ERROR, "unauthenticated push should yield error frame" @pytest.mark.asyncio async def test_t7_push_stream_404_for_missing_repo( client: AsyncClient, auth_headers: StrDict, ) -> None: """T7: push to a repo that doesn't exist yields 404 or error frame.""" body = _header_frame() + _end_frame() resp = await client.post( "/gabriel/nonexistent-repo-xyz-t7/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) if resp.status_code == 200: result = _last_frame(resp.content) assert result.get("t") == SFRAME_ERROR and result.get("code") == 404 else: assert resp.status_code == 404 @pytest.mark.asyncio async def test_t7_old_push_endpoints_deleted( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, ) -> None: """T7: all MWP v1 push endpoints return 404 — they were deleted in MWP v2.""" repo = await _make_repo(db_session, "T7 Route v1 Deleted", owner="testuser") deleted_paths = [ f"/{repo.owner}/{repo.slug}/filter-objects", f"/{repo.owner}/{repo.slug}/presign-objects", f"/{repo.owner}/{repo.slug}/presign", f"/{repo.owner}/{repo.slug}/push/objects", f"/{repo.owner}/{repo.slug}/push/objects/confirm", f"/{repo.owner}/{repo.slug}/push", ] for path in deleted_paths: resp = await client.post(path, headers=auth_headers, content=b"{}") assert resp.status_code == 404, ( f"Expected 404 for deleted endpoint {path}, got {resp.status_code}" ) # --------------------------------------------------------------------------- # T8 — E2E: full push → GET /refs confirms branch head updated # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_t8_push_then_refs_show_commit( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, ) -> None: """T8: push one commit; GET /refs confirms branch head updated.""" repo = await _make_repo(db_session, "T8 E2E Refs", owner="testuser") snap_id = blob_id(b"t8-e2e-snap-1") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id) body = ( _header_frame(head=commit["commit_id"]) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) ) push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert push_resp.status_code == 200 result = _last_frame(push_resp.content) assert result.get("ok") is True, f"push failed: {result}" refs_resp = await client.get( f"/{repo.owner}/{repo.slug}/refs", headers=auth_headers, ) assert refs_resp.status_code == 200 branch_heads = refs_resp.json().get("branch_heads", {}) assert branch_heads.get("main") == commit["commit_id"], ( f"Expected main head={commit['commit_id']!r}, got {branch_heads}" ) @pytest.mark.asyncio async def test_t8_push_with_objects_then_fetch_objects( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, ) -> None: """T8: push an object; fetch it back and verify content integrity.""" repo = await _make_repo(db_session, "T8 E2E Fetch", owner="testuser") raw = b"audio track bytes for e2e test" oid = blob_id(raw) snap_id = blob_id(b"t8-e2e-obj-snap") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"track.wav": oid}) body = ( _header_frame(n_objects=1) + _object_frame(oid, raw) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=1, n_commits=1) ) push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert push_resp.status_code == 200 result = _last_frame(push_resp.content) assert result.get("ok") is True, f"push failed: {result}" fetch_resp = await client.get( f"/o/{oid}", headers=auth_headers, ) assert fetch_resp.status_code == 200 assert fetch_resp.content == raw @pytest.mark.asyncio async def test_t8_push_chain_of_commits_parent_linking( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, ) -> None: """T8: push two chained commits; branch head advances to the child.""" repo = await _make_repo(db_session, "T8 E2E Chain", owner="testuser") snap1_id = blob_id(b"t8-e2e-snap-chain-1") commit1 = _make_commit(snapshot_id=snap1_id) snap1 = _make_snapshot(snap1_id) snap2_id = blob_id(b"t8-e2e-snap-chain-2") commit2 = _make_commit(snapshot_id=snap2_id, parent_ids=[commit1["commit_id"]]) snap2 = _make_snapshot(snap2_id) body = ( _header_frame(n_commits=2, head=commit2["commit_id"]) + _commit_pack_frame([commit1, commit2], [snap1, snap2]) + _end_frame(n_commits=2) ) push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert push_resp.status_code == 200 result = _last_frame(push_resp.content) assert result.get("ok") is True, f"push failed: {result}" refs_resp = await client.get( f"/{repo.owner}/{repo.slug}/refs", headers=auth_headers, ) refs = refs_resp.json().get("branch_heads", {}) assert refs.get("main") == commit2["commit_id"] # --------------------------------------------------------------------------- # T9 — Regression: MPackStreamWriter frames with compressed binary content # must not raise UnicodeDecodeError on the server. # # Root cause of staging bug: # 'utf-8' codec can't decode byte 0xad in position 2: invalid start byte # # The client sends O frames where the "content" field is zlib-compressed # binary bytes packed with use_bin_type=True (msgpack bin type 0xc4/c5/c6). # If the server Unpacker is misconfigured (raw=True, or content field encoded # as str/fixstr), raw=False raises UnicodeDecodeError on non-UTF-8 bytes. # # These tests use MPackStreamWriter (the actual client encoder) to build the # exact bytes the client sends, then POST them through the ASGI app. # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_t9_mpackstreamwriter_compressed_frames_decode_clean( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, ) -> None: """T9: server must decode MPackStreamWriter O frames with zlib-compressed binary content. Regression for: 'utf-8' codec can't decode byte 0xad in position 2. Client uses MPackStreamWriter (use_bin_type=True); server Unpacker uses raw=False. Binary content including byte 0xad must arrive as bytes, not trigger UnicodeDecodeError. """ from muse.core.mpack import MPackStreamWriter from muse.core.types import blob_id repo = await _make_repo(db_session, "T9 Compressed Binary Frames", owner="testuser") w = MPackStreamWriter() # Content that includes 0xad and other non-UTF-8 bytes — the exact failing case content = bytes(range(256)) * 10 oid = blob_id(content) snap_id = blob_id(b"t9-snap-compressed") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"file.bin": oid}) body = ( _fw.wrap(frame_type="H", payload=w.write_header(op="push", branch="main", n_objects=1, n_commits=1)) + _fw.wrap(frame_type="O", payload=w.write_object_raw(object_id=oid, raw_bytes=content, compress="zlib")) + _fw.wrap(frame_type="C", payload=w.write_commit_pack(commits=[commit], snapshots=[snap])) + _fw.wrap(frame_type="E", payload=w.write_end(n_objects=1, n_commits=1)) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code == 200 result = _last_frame(resp.content) assert result.get("t") != SFRAME_ERROR, f"Server returned error: {result.get('msg')}" assert result.get("ok") is True, f"No ok=True in result: {result}" @pytest.mark.asyncio async def test_t9_920_objects_with_binary_content_no_unicode_error( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, ) -> None: """T9: 920 objects with full byte range (0x00-0xff) — none may produce UnicodeDecodeError. Reproduces the staging scenario: ~900 small objects each containing binary data. The server must process all O frames without any 'utf-8 codec' error. """ from muse.core.mpack import MPackStreamWriter from muse.core.types import blob_id n = 920 repo = await _make_repo(db_session, "T9 920 Binary Objects", owner="testuser") w = MPackStreamWriter() snap_id = blob_id(b"t9-snap-920") commit = _make_commit(snapshot_id=snap_id) manifest = {} parts = [_fw.wrap(frame_type="H", payload=w.write_header(op="push", branch="main", n_objects=n, n_commits=1))] for i in range(n): raw_content = (bytes(range(256)) * 2)[i % 256: i % 256 + 256] + i.to_bytes(4, "big") oid = blob_id(raw_content) manifest[f"file_{i}.bin"] = oid parts.append(_fw.wrap(frame_type="O", payload=w.write_object_raw(object_id=oid, raw_bytes=raw_content, compress="zlib"))) snap = _make_snapshot(snap_id, manifest) parts.append(_fw.wrap(frame_type="C", payload=w.write_commit_pack(commits=[commit], snapshots=[snap]))) parts.append(_fw.wrap(frame_type="E", payload=w.write_end(n_objects=n, n_commits=1))) body = b"".join(parts) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code == 200 result = _last_frame(resp.content) assert result.get("t") != SFRAME_ERROR, f"Server returned error: {result.get('msg')}" assert result.get("ok") is True, f"No ok=True in result: {result}" # --------------------------------------------------------------------------- # T10 — Regression: server response frames must use only string map keys. # # Root cause of staging bug: # stream read error: int is not allowed for map key when strict_map_key=True # # MPackStreamReader (muse/core/mpack.py) uses msgpack.Unpacker with the # default strict_map_key=True. If the server sends any frame where a map # key is an integer, the client raises and the push fails. # # This test decodes the server response with strict_map_key=True — the exact # setting the client uses — and asserts every key in every frame is a str. # --------------------------------------------------------------------------- def _unpack_all_strict(raw: bytes) -> list[JSONValue]: """Decode all msgpack frames with strict_map_key=True (same as MPackStreamReader).""" unpacker = msgpack.Unpacker(raw=False, strict_map_key=True) unpacker.feed(raw) return list(unpacker) @pytest.mark.asyncio async def test_t10_response_frames_use_only_string_map_keys( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, ) -> None: """T10: all server response frame map keys must be strings. MPackStreamReader uses strict_map_key=True (msgpack default). Any integer key in a server frame raises 'int is not allowed for map key' on the client and aborts the push. This test uses the same strict decoder to catch the mismatch at the server level before it reaches production. """ repo = await _make_repo(db_session, "T10 String Map Keys", owner="testuser") snap_id = blob_id(b"t10-snap-1") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id) body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code == 200 # Use the same strict decoder the client uses — must not raise on any frame. try: frames = _unpack_all_strict(resp.content) except Exception as exc: raise AssertionError( f"Server response failed strict msgpack decode: {exc}\n" f"Raw response (first 512 bytes): {resp.content[:512]!r}" ) from exc assert frames, "Expected at least one frame in response" for result in frames: assert isinstance(result, dict), f"Expected dict frame, got {type(result)}" int_keys = [k for k in result if not isinstance(k, str)] assert not int_keys, ( f"Response frame has integer map keys: {int_keys!r}\nFull frame: {result!r}" ) # --------------------------------------------------------------------------- # T11 — Phase 5: E frame count verification # # Server must reject a push where the E frame's n_objects or n_commits # field does not match the actual number of frames received. # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_t11_e_frame_wrong_n_objects_rejected( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T11: E frame claiming wrong n_objects is rejected with an error.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T11 E Objects Mismatch", owner="testuser") raw = b"audio bytes t11" oid = blob_id(raw) snap_id = blob_id(b"t11-snap-objs") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"a.wav": oid}) # Send 1 object frame but E frame claims 0 body = ( _header_frame(n_objects=1, n_commits=1) + _object_frame(oid, raw) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=0, n_commits=1) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _last_frame(resp.content) assert resp.status_code == 400 or result.get("t") == SFRAME_ERROR, ( f"Expected 400 or error frame, got status={resp.status_code} result={result}" ) assert "count mismatch" in result.get("msg", "").lower() or result.get("code") == 400 @pytest.mark.asyncio async def test_t11_e_frame_wrong_n_commits_rejected( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T11: E frame claiming wrong n_commits is rejected with an error.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T11 E Commits Mismatch", owner="testuser") snap_id = blob_id(b"t11-snap-commits") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id) # Send 1 commit but E frame claims 2 body = ( _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=0, n_commits=2) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _last_frame(resp.content) assert resp.status_code == 400 or result.get("t") == SFRAME_ERROR, ( f"Expected 400 or error frame, got status={resp.status_code} result={result}" ) assert "count mismatch" in result.get("msg", "").lower() or result.get("code") == 400 @pytest.mark.asyncio async def test_t11_e_frame_overstated_objects_rejected( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T11: E frame claiming more objects than received is rejected.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T11 E Overstated Objects", owner="testuser") snap_id = blob_id(b"t11-snap-over") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id) # Send 0 objects but E frame claims 5 body = ( _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=5, n_commits=1) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _last_frame(resp.content) assert resp.status_code == 400 or result.get("t") == SFRAME_ERROR, ( f"Expected 400 or error frame, got status={resp.status_code} result={result}" ) assert "count mismatch" in result.get("msg", "").lower() or result.get("code") == 400 # --------------------------------------------------------------------------- # T12 — Phase 6: Ingest transaction model # # P6A: snapshot referential integrity — reject if snapshot references an # object not in the push bundle and not in storage. # P6B: snapshot references an object from a PRIOR push — accepted (already # in storage path). # P6C: atomicity — branch ref remains unchanged after a rejected push. # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_t12_p6a_snapshot_missing_object_rejected( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T12/P6A: snapshot manifest references an object not in bundle or storage → 422.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T12 P6A Missing Object", owner="testuser") ghost_oid = blob_id(b"ghost-object-never-pushed") snap_id = blob_id(b"t12-p6a-snap") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"missing.wav": ghost_oid}) body = ( _header_frame(n_objects=0, n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=0, n_commits=1) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _last_frame(resp.content) assert resp.status_code in (200, 422), f"Unexpected status: {resp.status_code}" assert resp.status_code == 422 or result.get("t") == SFRAME_ERROR, ( f"Expected 422 or error frame for missing snapshot object, got: {result}" ) msg = result.get("msg", "") assert "missing" in msg.lower() or result.get("code") in (422, 400), ( f"Error message should mention missing object: {msg!r}" ) @pytest.mark.asyncio async def test_t12_p6b_snapshot_object_from_prior_push_accepted( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T12/P6B: snapshot references an object stored in a previous push → accepted.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T12 P6B Prior Object", owner="testuser") raw = b"audio content from push 1" oid = blob_id(raw) snap1_id = blob_id(b"t12-p6b-snap1") commit1 = _make_commit(snapshot_id=snap1_id) snap1 = _make_snapshot(snap1_id, {"track.wav": oid}) body1 = ( _header_frame(n_objects=1, n_commits=1) + _object_frame(oid, raw) + _commit_pack_frame([commit1], [snap1]) + _end_frame(n_objects=1, n_commits=1) ) resp1 = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body1, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp1.status_code == 200 result1 = _last_frame(resp1.content) assert result1.get("ok") is True, f"Push 1 failed: {result1}" # Push 2: new commit referencing the SAME object — not re-sent in bundle snap2_id = blob_id(b"t12-p6b-snap2") commit2 = _make_commit(snapshot_id=snap2_id, parent_ids=[commit1["commit_id"]]) snap2 = _make_snapshot(snap2_id, {"track.wav": oid}) body2 = ( _header_frame(n_objects=0, n_commits=1) + _commit_pack_frame([commit2], [snap2]) + _end_frame(n_objects=0, n_commits=1) ) resp2 = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body2, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp2.status_code == 200 result2 = _last_frame(resp2.content) assert result2.get("ok") is True, ( f"Push 2 should succeed — object already in storage. Got: {result2}" ) @pytest.mark.asyncio async def test_t12_p6c_failed_push_leaves_branch_unchanged( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T12/P6C: a rejected push must not advance the branch head (atomicity).""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T12 P6C Atomicity", owner="testuser") snap1_id = blob_id(b"t12-p6c-snap1") commit1 = _make_commit(snapshot_id=snap1_id) snap1 = _make_snapshot(snap1_id) body1 = ( _header_frame(n_commits=1, head=commit1["commit_id"]) + _commit_pack_frame([commit1], [snap1]) + _end_frame(n_commits=1) ) resp1 = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body1, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp1.status_code == 200 result1 = _last_frame(resp1.content) assert result1.get("ok") is True, f"Push 1 failed: {result1}" refs1 = (await client.get(f"/{repo.owner}/{repo.slug}/refs", headers=auth_headers)).json() head_after_push1 = refs1.get("branch_heads", {}).get("main") assert head_after_push1 == commit1["commit_id"] # Push 2: snapshot references a ghost object — must be rejected ghost_oid = blob_id(b"t12-p6c-ghost-never-exists") snap2_id = blob_id(b"t12-p6c-snap2") commit2 = _make_commit(snapshot_id=snap2_id, parent_ids=[commit1["commit_id"]]) snap2 = _make_snapshot(snap2_id, {"ghost.wav": ghost_oid}) body2 = ( _header_frame(n_objects=0, n_commits=1) + _commit_pack_frame([commit2], [snap2]) + _end_frame(n_objects=0, n_commits=1) ) resp2 = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body2, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result2 = _last_frame(resp2.content) assert resp2.status_code in (200, 422) and ( resp2.status_code == 422 or result2.get("t") == SFRAME_ERROR ), f"Push 2 should be rejected, got: status={resp2.status_code} result={result2}" # Branch head must still be commit1 refs2 = (await client.get(f"/{repo.owner}/{repo.slug}/refs", headers=auth_headers)).json() head_after_push2 = refs2.get("branch_heads", {}).get("main") assert head_after_push2 == commit1["commit_id"], ( f"Branch head must remain commit1 after rejected push. " f"Got: {head_after_push2!r}" ) # --------------------------------------------------------------------------- # Phase 4 — Server-side branch ref CAS (SELECT FOR UPDATE) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_phase4_sequential_pushes_both_advance_branch( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """Phase 4 / Data: two sequential pushes must each advance the branch. The SELECT FOR UPDATE on the branch row ensures that concurrent pushes are serialized at the DB level. This test uses sequential pushes with separate sessions to verify the core invariant: the second push sees the branch head left by the first push and advances it correctly. """ from musehub.services.musehub_wire import wire_push_stream _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "P4 Sequential CAS") snap1_id = blob_id(b"p4-snap-1") commit1 = _make_commit(snapshot_id=snap1_id) snap1 = _make_snapshot(snap1_id) # Push 1 async def body1() -> None: yield ( _header_frame(n_commits=1, head=commit1["commit_id"]) + _commit_pack_frame([commit1], [snap1]) + _end_frame(n_commits=1) ) frames1 = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body1(), "gabriel") ) result1 = [f for f in frames1 if f.get("t") == SFRAME_RESULT] assert result1 and result1[0]["ok"] is True, f"Push 1 failed: {frames1}" # Push 2 — parent is commit1 snap2_id = blob_id(b"p4-snap-2") commit2 = _make_commit( snapshot_id=snap2_id, parent_ids=[commit1["commit_id"]], ) snap2 = _make_snapshot(snap2_id) async def body2() -> None: yield ( _header_frame(n_commits=1, head=commit2["commit_id"]) + _commit_pack_frame([commit2], [snap2]) + _end_frame(n_commits=1) ) frames2 = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body2(), "gabriel") ) result2 = [f for f in frames2 if f.get("t") == SFRAME_RESULT] assert result2 and result2[0]["ok"] is True, f"Push 2 failed: {frames2}" # Branch must point to commit2 (the latest) from musehub.db.musehub_models import MusehubBranch from sqlalchemy import select as _select branch = (await db_session.execute( _select(MusehubBranch).where( MusehubBranch.repo_id == repo.repo_id, MusehubBranch.name == "main", ) )).scalar_one() assert branch.head_commit_id == commit2["commit_id"], ( f"Branch should point to commit2 after second push. " f"Got: {branch.head_commit_id!r}" ) @pytest.mark.asyncio async def test_phase4_non_ff_push_rejected_after_concurrent_advance( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """Phase 4 / Unit: a push that tries to set a non-FF head is rejected. This is the key invariant the SELECT FOR UPDATE protects: if another push advances the branch between our read and our write, our push fails the fast-forward check on the now-current head rather than silently overwriting. """ from musehub.services.musehub_wire import wire_push_stream from musehub.db.musehub_models import MusehubBranch from sqlalchemy import select as _select _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "P4 Non-FF After Advance") # Establish an initial commit on the branch snap0_id = blob_id(b"p4-nff-snap-0") commit0 = _make_commit(snapshot_id=snap0_id) snap0 = _make_snapshot(snap0_id) async def body0() -> None: yield ( _header_frame(n_commits=1) + _commit_pack_frame([commit0], [snap0]) + _end_frame(n_commits=1) ) frames0 = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body0(), "gabriel") ) assert [f for f in frames0 if f.get("t") == SFRAME_RESULT and f["ok"]], ( f"Initial push failed: {frames0}" ) # Manually advance the branch to simulate a concurrent push winning branch_row = (await db_session.execute( _select(MusehubBranch).where( MusehubBranch.repo_id == repo.repo_id, MusehubBranch.name == "main", ) )).scalar_one() concurrent_commit_id = blob_id(b"p4-concurrent-winner") branch_row.head_commit_id = concurrent_commit_id await db_session.commit() # Now push a commit that is a child of commit0 — diverges from the branch snap1_id = blob_id(b"p4-nff-snap-1") commit1 = _make_commit(snapshot_id=snap1_id, parent_ids=[commit0["commit_id"]]) snap1 = _make_snapshot(snap1_id) async def body1() -> None: yield ( _header_frame(n_commits=1, force=False) + _commit_pack_frame([commit1], [snap1]) + _end_frame(n_commits=1) ) frames1 = await _collect_frames( wire_push_stream(db_session, str(repo.repo_id), body1(), "gabriel") ) error_frames = [f for f in frames1 if f.get("t") == SFRAME_ERROR] assert error_frames, ( "Expected non-FF push to be rejected after branch was concurrently advanced. " f"Got frames: {[f.get('t') for f in frames1]}" ) assert "non-fast-forward" in error_frames[0]["msg"].lower(), ( f"Expected non-fast-forward error, got: {error_frames[0]['msg']!r}" )