"""TDD — push_stream route must stream the request body (not buffer it). Rules (from musewire-performance.md): S1 push_stream route source must NOT contain `await request.body()`. S2 push_stream route source must call `request.stream()`. S3 A valid push still yields RESULT ok=True after the streaming change. S4 Auth is checked before any stream bytes are consumed. Why this matters: `await request.body()` causes Cloudflare to buffer the entire request body before forwarding it to the origin, which triggers TLS bad_record_mac errors on large payloads. request.stream() lets Cloudflare forward chunks as they arrive — matching GitHub's push behavior. """ from __future__ import annotations import inspect from collections.abc import 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, 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 # --------------------------------------------------------------------------- def _pack(obj: JSONValue) -> bytes: return msgpack.packb(obj, use_bin_type=True) def _wrap(ft: str, data: JSONValue) -> bytes: return _fw.wrap(frame_type=ft, payload=_pack(data)) def _header_frame(n_objects: int = 0, n_commits: int = 1) -> bytes: return _wrap(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(SFRAME_COMMIT_PACK, {"t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots}) def _end_frame(n_objects: int = 0, n_commits: int = 1) -> bytes: return _wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits}) def _make_commit(snapshot_id: str | None = None, branch: str = "main") -> 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": branch, "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) # --------------------------------------------------------------------------- # S1 — route source must NOT contain `await request.body()` # --------------------------------------------------------------------------- def test_s1_push_stream_route_does_not_call_request_body() -> None: """push_stream must use request.stream(), not await request.body(). `await request.body()` forces Cloudflare to buffer the entire upload before forwarding it to the origin server — causing bad_record_mac on large payloads. This is a static check on the route source code. """ from musehub.api.routes import wire source = inspect.getsource(wire.push_stream) assert "await request.body()" not in source, ( "push_stream route calls `await request.body()` which buffers the full body " "before processing. Replace with `request.stream()` so Cloudflare streams " "chunks as they arrive. See docs/protocol/musewire-performance.md." ) # --------------------------------------------------------------------------- # S2 — route source must call `request.stream()` # --------------------------------------------------------------------------- def test_s2_push_stream_route_uses_request_stream() -> None: """push_stream must read the request body as a stream, not buffer it. Accepts either request.stream() or the ASGI body_iter pattern — both stream chunks as they arrive rather than buffering the full upload. """ from musehub.api.routes import wire source = inspect.getsource(wire.push_stream) streams = "request.stream()" in source or "body_iter" in source assert streams, ( "push_stream route must stream the request body (not buffer via request.body()). " "Use request.stream() or an ASGI body_iter that reads from receive()." ) # --------------------------------------------------------------------------- # S3 — valid push still succeeds after streaming fix # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_s3_valid_push_still_succeeds_with_streaming( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """After switching to request.stream(), a valid push must still yield RESULT ok=True.""" _stub_r2_backend(monkeypatch) repo = await create_repo(db_session, owner="testuser", name="s3-success") snap_id = blob_id(b"s3-snap") 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": WIRE_CONTENT_TYPE}, ) assert resp.status_code == 200 result = _last_frame(resp.content) assert result.get("ok") is True, f"Expected ok=True, got: {result}" # --------------------------------------------------------------------------- # S4 — auth is checked before stream consumption # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_s4_auth_checked_before_stream_consumed( client: AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """Unauthenticated push must be rejected without consuming the request stream.""" _stub_r2_backend(monkeypatch) repo = await create_repo(db_session, owner="testuser", name="s4-auth") 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) yield _pack({"t": SFRAME_RESULT, "ok": True, "msg": "ok", "heads": {}, "head": ""}) 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}, ) if resp.status_code in (401, 403): assert not wire_was_called, "wire_push_stream called despite auth failure" else: unpacker = msgpack.Unpacker(raw=False) unpacker.feed(resp.content) frames = list(unpacker) error_frames = [f for f in frames if f.get("t") == "X"] assert error_frames or not wire_was_called