"""TDD — server reads raw MWP frames and returns a plain final response. Rules: S7 push/stream returns Content-Type: application/x-muse-wire. S8 A valid push with raw MWP frames yields ok=True. Full round-trip: client sends raw MWP, server parses, returns result. S9 Auth is checked before any frames are consumed. """ from __future__ import annotations import inspect from collections.abc import AsyncIterator from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock import msgpack import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import blob_id, now_utc_iso from musehub.types.json_types import JSONObject, JSONValue, StrDict from musehub.models.wire import ( SFRAME_COMMIT_PACK, SFRAME_END, SFRAME_HEADER, SFRAME_RESULT, ) from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter from tests.factories import create_repo _fw = MuseWireFrameWriter() def _last_frame(raw: bytes) -> JSONObject: unpacker = msgpack.Unpacker(raw=False) unpacker.feed(raw) last: JSONObject = {} for frame in unpacker: last = frame return last # --------------------------------------------------------------------------- # Helpers — raw MWP (no gRPC prefix) # --------------------------------------------------------------------------- def _pack(obj: JSONValue) -> bytes: return msgpack.packb(obj, use_bin_type=True) def _wrap_raw(ft: str, data: JSONValue) -> bytes: """Wrap a logical frame in a raw MWP envelope — NO gRPC prefix.""" return _fw.wrap(frame_type=ft, payload=_pack(data)) def _header_frame(n_objects: int = 0, n_commits: int = 1) -> bytes: return _wrap_raw(SFRAME_HEADER, { "t": SFRAME_HEADER, "branch": "main", "force": False, "have": [], "head": blob_id(b"head"), "n_objects": n_objects, "n_commits": n_commits, }) def _commit_pack_frame(commits: list[dict], snapshots: list[dict]) -> bytes: return _wrap_raw(SFRAME_COMMIT_PACK, { "t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots, }) def _end_frame(n_objects: int = 0, n_commits: int = 0) -> bytes: return _wrap_raw(SFRAME_END, {"t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits}) def _make_commit(snapshot_id: str | None = None) -> JSONObject: snap = snapshot_id or blob_id(b"default-snap") return { "commit_id": blob_id(f"commit-{now_utc_iso()}".encode()), "parent_ids": [], "snapshot_id": snap, "branch": "main", "message": "test commit", "author": "gabriel", "committed_at": now_utc_iso(), "signature": "", "signer_key_id": "", "agent_id": "", "model_id": "", "metadata": {}, } def _make_snapshot(snap_id: str) -> JSONObject: return {"snapshot_id": snap_id, "manifest": {}} def _stub_r2_backend(monkeypatch: pytest.MonkeyPatch) -> None: backend = MagicMock() backend.store_object = AsyncMock(return_value=None) backend.object_exists = AsyncMock(return_value=False) backend.get_object = AsyncMock(return_value=None) monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) # --------------------------------------------------------------------------- # S7 — push/stream response Content-Type is application/x-muse-wire # --------------------------------------------------------------------------- def test_s7_push_route_response_content_type() -> None: """push/stream response Content-Type must be application/x-muse-wire.""" from musehub.api.routes import wire source = inspect.getsource(wire.push_stream) assert "WIRE_CONTENT_TYPE" in source or "application/x-muse-wire" in source, ( "push/stream route must set Content-Type: application/x-muse-wire. " "Replace GRPC_CONTENT_TYPE with WIRE_CONTENT_TYPE." ) assert "GRPC_CONTENT_TYPE" not in source, ( "push/stream route still uses GRPC_CONTENT_TYPE. " "Replace with WIRE_CONTENT_TYPE." ) # --------------------------------------------------------------------------- # S8 — valid push with raw MWP frames yields ok=True # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_s8_raw_mwp_push_succeeds( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """A push body with raw MWP frames (no gRPC prefix) must yield ok=True. This is the core end-to-end test: client builds raw MWP frames, sends them as a plain HTTPS POST, server reads and stores them, returns a plain msgpack result body. """ _stub_r2_backend(monkeypatch) repo = await create_repo(db_session, owner="testuser", name="s8-plain-push") snap_id = blob_id(b"s8-snap") 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) # Confirm body does NOT start with gRPC prefix assert body[:1] != b"\x00", "Test body should be raw MWP, not gRPC-prefixed" assert body[:4] == b"muse", "Test body should start with MWP magic" 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, f"Expected 200, got {resp.status_code}: {resp.text[:200]}" assert resp.headers.get("content-type", "").startswith(WIRE_CONTENT_TYPE), ( f"Expected Content-Type {WIRE_CONTENT_TYPE}, got {resp.headers.get('content-type')}" ) # Response is a plain msgpack dict, not a stream of gRPC frames result = _last_frame(resp.content) assert isinstance(result, dict), f"Expected dict response, got {type(result)}" assert result.get("ok") is True, f"Expected ok=True, got: {result}" assert "stored_commits" in result, f"Missing stored_commits in result: {result}" # --------------------------------------------------------------------------- # S9 — auth is still checked before stream consumption (unchanged) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_s9_auth_still_checked_before_stream( client: AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """Unauthenticated push must be rejected without consuming the request body.""" _stub_r2_backend(monkeypatch) repo = await create_repo(db_session, owner="testuser", name="s9-auth-plain") wire_was_called = [] async def _spy(session: AsyncSession, repo_id: str, body_iter: AsyncIterator[bytes], pusher_id: str | None) -> None: wire_was_called.append(True) monkeypatch.setattr("musehub.api.routes.wire.wire_push_stream", _spy) 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 (401, 403), ( f"Unauthenticated push should return 401/403, got {resp.status_code}" ) assert not wire_was_called, "wire_push_stream called despite missing auth"