"""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 hashlib import zlib from collections.abc import AsyncGenerator, AsyncIterator from datetime import datetime, timezone from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import msgpack import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession 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 GRPC_CONTENT_TYPE, MuseWireFrameWriter, grpc_frame from tests.factories import create_repo _fw = MuseWireFrameWriter() # --------------------------------------------------------------------------- # Shared codec helpers # --------------------------------------------------------------------------- def _pack(data: object) -> bytes: """Encode one msgpack frame payload (without transport envelope).""" return msgpack.packb(data, use_bin_type=True) def _wrap(ft: str, data: object) -> bytes: """Encode, wrap in MWP envelope, and add gRPC length-prefix. wire_push_stream calls iter_wire_frames_grpc, which expects each MWP frame prefixed with a 5-byte gRPC header (compress_flag + uint32 length). """ return grpc_frame(_fw.wrap(frame_type=ft, payload=_pack(data))) def _unpack_all(raw: bytes) -> list[dict]: """Decode all concatenated gRPC-framed bytes into a list of dicts. Handles two formats: - gRPC-wrapped MWP: 0x00 + length(4) + b"muse" envelope → parse inner MWP - gRPC-wrapped msgpack: 0x00 + length(4) + msgpack payload → parse as dict """ import struct results = [] offset = 0 while offset < len(raw): if offset + 5 > len(raw): break if raw[offset] != 0x00: break grpc_length = struct.unpack(">I", raw[offset + 1:offset + 5])[0] if offset + 5 + grpc_length > len(raw): break inner = raw[offset + 5: offset + 5 + grpc_length] offset += 5 + grpc_length if inner[:4] == b"muse": # gRPC-wrapped MWP envelope — parse inner MWP frame header_len = struct.unpack(">I", inner[5:9])[0] payload_start = 9 + header_len + 8 payload_len = struct.unpack(">Q", inner[9 + header_len:payload_start])[0] payload = inner[payload_start:payload_start + payload_len] results.append(msgpack.unpackb(payload, raw=False)) else: # gRPC-wrapped raw msgpack — server response frames (P/X/R) decoded = msgpack.unpackb(inner, raw=False) if isinstance(decoded, dict): results.append(decoded) return results def _sha256_oid(raw: bytes) -> str: return "sha256:" + hashlib.sha256(raw).hexdigest() def _utc() -> str: return datetime.now(tz=timezone.utc).isoformat() def _make_obj_bytes(content: bytes = b"hello world") -> tuple[str, bytes]: """Return (sha256_oid, raw_content) for a test object.""" oid = _sha256_oid(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") -> bytes: return _wrap(SFRAME_OBJECT, { "t": SFRAME_OBJECT, "id": oid, "content": content, "path": path, "enc": enc, }) 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() -> bytes: return _wrap(SFRAME_END, {"t": SFRAME_END}) 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", ) -> dict: cid = commit_id or _sha256_oid(f"commit-{_utc()}".encode()) return { "commit_id": cid, "parent_ids": parent_ids or [], "snapshot_id": snapshot_id or _sha256_oid(b"default-snap"), "branch": branch, "message": "test commit", "author": author, "committed_at": _utc(), "signature": "", "signer_key_id": "", "agent_id": "", "model_id": "", "metadata": {}, } def _make_snapshot(snapshot_id: str, manifest: dict | None = None) -> dict: return { "snapshot_id": snapshot_id, "manifest": manifest or {}, "committed_at": _utc(), } async def _collect_frames(gen: AsyncGenerator[bytes, None]) -> list[dict]: """Drain an async generator of frame bytes into a list of decoded dicts.""" raw = b"" async for chunk in gen: raw += chunk return _unpack_all(raw) 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 GRPC_CONTENT_TYPE == "application/grpc+muse" 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): 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() frames = _unpack_all(_prog("uploading objects")) assert frames[0]["t"] == SFRAME_PROGRESS assert frames[0]["msg"] == "uploading objects" def test_err_encodes_error_frame_with_code(self) -> None: _, _, _err, _ = self._import_helpers() frames = _unpack_all(_err("repo not found", 404)) f = frames[0] 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() frames = _unpack_all(_err("bad request")) assert frames[0]["code"] == 400 def test_result_ok_encodes_correctly(self) -> None: _, _, _, _result = self._import_helpers() heads = {"main": "sha256:abc"} frames = _unpack_all(_result(True, "pushed", heads, "sha256:abc")) f = frames[0] 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() frames = _unpack_all(_result(False, "rejected", {}, "")) assert frames[0]["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 _sha256_oid(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 = _sha256_oid(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 = _sha256_oid(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 = _sha256_oid(raw) frame = _unpack_all(_object_frame(oid, raw, enc="raw"))[0] assert bytes(frame["content"]) == raw # --------------------------------------------------------------------------- # 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 = _sha256_oid(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" # --------------------------------------------------------------------------- # 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(): 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 = _sha256_oid(b"snap") snap = _make_snapshot(snap_id) commit["snapshot_id"] = snap_id async def body(): 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(): 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 = "sha256:" + "a" * 64 # wrong hash content = b"this content does not match the oid" async def body(): 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 = _sha256_oid(b"snap-data") snap = _make_snapshot(snap_id) commit["snapshot_id"] = snap_id async def body(): 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" # --------------------------------------------------------------------------- # 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) -> bool: return oid in _store async def _put(oid: str, data: bytes) -> 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") -> Any: """Create a repo row + main branch, committed and visible to any session.""" import uuid as _uuid_mod from datetime import datetime, timezone from musehub.db.musehub_models import MusehubRepo, MusehubBranch from musehub.core.genesis import compute_repo_id, compute_branch_id owner_user_id = str(_uuid_mod.uuid4()) slug = name.lower().replace(" ", "-") created_at = datetime.now(tz=timezone.utc) repo_id = compute_repo_id(owner_user_id, slug, "", 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 = _sha256_oid(b"t6-snap-1") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id) async def body(): yield _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame() 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 = _sha256_oid(raw) snap_id = _sha256_oid(b"t6-snap-obj") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"track.wav": oid}) async def body(): yield ( _header_frame(n_objects=1, n_commits=1) + _object_frame(oid, raw) + _commit_pack_frame([commit], [snap]) + _end_frame() ) 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 = _sha256_oid(raw) snap_id = _sha256_oid(b"t6-snap-zlib") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"beat.mid": oid}) async def body(): yield ( _header_frame(n_objects=1, n_commits=1) + _object_frame(oid, compressed, enc="zlib") + _commit_pack_frame([commit], [snap]) + _end_frame() ) 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 = _sha256_oid(b"t6-snap-force-1") commit1 = _make_commit(snapshot_id=snap_id) snap1 = _make_snapshot(snap_id) async def body1(): yield _header_frame() + _commit_pack_frame([commit1], [snap1]) + _end_frame() 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 = _sha256_oid(b"t6-snap-force-2") commit2 = _make_commit(snapshot_id=snap_id2) snap2 = _make_snapshot(snap_id2) async def body2(): yield _header_frame(force=True) + _commit_pack_frame([commit2], [snap2]) + _end_frame() 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 # --------------------------------------------------------------------------- # 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: dict, 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 = _sha256_oid(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() resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE}, ) assert resp.status_code == 200 assert GRPC_CONTENT_TYPE in resp.headers.get("content-type", "") @pytest.mark.asyncio async def test_t7_push_stream_response_contains_result_frame( client: AsyncClient, db_session: AsyncSession, auth_headers: dict, monkeypatch: pytest.MonkeyPatch, ) -> None: """T7: response body is a frame stream that ends with a RESULT frame.""" _stub_r2_backend(monkeypatch) repo = await _make_repo(db_session, "T7 Route Test 2", owner="testuser") snap_id = _sha256_oid(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() resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE}, ) frames = _unpack_all(resp.content) 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]}" @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": GRPC_CONTENT_TYPE}, ) assert resp.status_code in (200, 401, 403) if resp.status_code == 200: frames = _unpack_all(resp.content) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert error_frames, "unauthenticated push should yield error frame" @pytest.mark.asyncio async def test_t7_push_stream_404_for_missing_repo( client: AsyncClient, auth_headers: dict, ) -> 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": GRPC_CONTENT_TYPE}, ) if resp.status_code == 200: frames = _unpack_all(resp.content) err = [f for f in frames if f.get("t") == SFRAME_ERROR] assert err and err[0].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: dict, ) -> 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}/push/objects", f"/{repo.owner}/{repo.slug}/push/object-pack", 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: dict, ) -> None: """T8: push one commit; GET /refs confirms branch head updated.""" repo = await _make_repo(db_session, "T8 E2E Refs", owner="testuser") snap_id = _sha256_oid(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() ) push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE}, ) assert push_resp.status_code == 200 frames = _unpack_all(push_resp.content) result = next((f for f in frames if f.get("t") == SFRAME_RESULT), None) assert result is not None, f"no RESULT frame; got: {[f.get('t') for f in frames]}" assert result["ok"] is True 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: dict, ) -> 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 = _sha256_oid(raw) snap_id = _sha256_oid(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() ) push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE}, ) assert push_resp.status_code == 200 result = next( (f for f in _unpack_all(push_resp.content) if f.get("t") == SFRAME_RESULT), None ) assert result and result["ok"] is True 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: dict, ) -> 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 = _sha256_oid(b"t8-e2e-snap-chain-1") commit1 = _make_commit(snapshot_id=snap1_id) snap1 = _make_snapshot(snap1_id) snap2_id = _sha256_oid(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() ) push_resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE}, ) assert push_resp.status_code == 200 result = next( (f for f in _unpack_all(push_resp.content) if f.get("t") == SFRAME_RESULT), None ) assert result and result["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: dict, ) -> 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 = _sha256_oid(b"t9-snap-compressed") commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, {"file.bin": oid}) body = ( grpc_frame(_fw.wrap(frame_type="H", payload=w.write_header(op="push", branch="main", n_objects=1, n_commits=1))) + grpc_frame(_fw.wrap(frame_type="O", payload=w.write_object_raw(object_id=oid, raw_bytes=content, compress="zlib"))) + grpc_frame(_fw.wrap(frame_type="C", payload=w.write_commit_pack(commits=[commit], snapshots=[snap]))) + grpc_frame(_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": GRPC_CONTENT_TYPE}, ) assert resp.status_code == 200 frames = _unpack_all(resp.content) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert not error_frames, ( f"Server returned error frame(s): {[f.get('msg') for f in error_frames]}" ) result = next((f for f in frames if f.get("t") == SFRAME_RESULT), None) assert result is not None, f"No RESULT frame; got: {[f.get('t') for f in frames]}" assert result["ok"] is True @pytest.mark.asyncio async def test_t9_920_objects_with_binary_content_no_unicode_error( client: AsyncClient, db_session: AsyncSession, auth_headers: dict, ) -> 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 = _sha256_oid(b"t9-snap-920") commit = _make_commit(snapshot_id=snap_id) manifest = {} parts = [grpc_frame(_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(grpc_frame(_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(grpc_frame(_fw.wrap(frame_type="C", payload=w.write_commit_pack(commits=[commit], snapshots=[snap])))) parts.append(grpc_frame(_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": GRPC_CONTENT_TYPE}, ) assert resp.status_code == 200 frames = _unpack_all(resp.content) error_frames = [f for f in frames if f.get("t") == SFRAME_ERROR] assert not error_frames, ( f"Server returned error frame(s): {[f.get('msg') for f in error_frames]}" ) result = next((f for f in frames if f.get("t") == SFRAME_RESULT), None) assert result is not None, f"No RESULT frame; got: {[f.get('t') for f in frames]}" assert result["ok"] is True # --------------------------------------------------------------------------- # 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: """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: dict, ) -> 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 = _sha256_oid(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() resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE}, ) assert resp.status_code == 200 # Use the same strict decoder the client uses — must not raise. try: frames = _unpack_all_strict(resp.content) except Exception as exc: raise AssertionError( f"Server response failed strict msgpack decode (same as MPackStreamReader): {exc}\n" f"Raw response (first 512 bytes): {resp.content[:512]!r}" ) from exc # Belt-and-suspenders: assert every key in every frame dict is a str. for i, frame in enumerate(frames): if not isinstance(frame, dict): continue int_keys = [k for k in frame if not isinstance(k, str)] assert not int_keys, ( f"Frame {i} (t={frame.get('t')!r}) has integer map keys: {int_keys!r}\n" f"Full frame: {frame!r}" )