"""TDD — push client must parse response as a frame stream, not a single dict. The push/stream response is documented as a sequence of msgpack objects: zero or more PROGRESS frames (t='P') followed by one RESULT frame (t='R'). The old client implementation did: result = _msgpack.unpackb(resp.content, raw=False) This buffers the full response body and parses ONE msgpack object. When the server starts streaming P frames before the final result, this breaks: - `unpackb` raises an exception (multiple objects look like trailing garbage) - The client never sees P frames (cannot print progress to stderr) Tests: C1 Static: push_stream_async must NOT call `_msgpack.unpackb(resp.content)`. C2 Static: push_stream_async must use an Unpacker or MPackStreamReader to iterate through all response frames until it finds the RESULT frame. C3 Integration: The client's response-parsing logic correctly handles a response that contains P frames followed by an R frame. C4 Static: push_stream_coro must use `client.stream()` (streaming HTTP), not `client.post()` (buffered HTTP), so P frames are printed to stderr as each chunk arrives — the same way git prints "remote: ..." in real time during a push, not after the server closes the connection. """ from __future__ import annotations import inspect import msgpack import pytest from muse.core.pack import PushResult from muse.core.types import MsgpackDict # --------------------------------------------------------------------------- # C1 — client must NOT call unpackb(resp.content) # --------------------------------------------------------------------------- def test_c1_push_client_does_not_unpackb_full_content() -> None: """push_stream_async must not call `_msgpack.unpackb(resp.content)`. When the server starts streaming P frames before the R frame, resp.content contains multiple concatenated msgpack objects. `unpackb` only parses the first one and raises ExtraData (or silently ignores the rest), so the client would never see the RESULT frame. The correct approach: iterate with msgpack.Unpacker (or MPackStreamReader) over resp.content until the R or X frame is found. """ from muse.core import transport source = inspect.getsource(transport.HttpTransport.push_stream_coro) assert "_msgpack.unpackb(resp.content" not in source and "unpackb(resp.content" not in source, ( "push_stream_async calls `_msgpack.unpackb(resp.content)` which cannot handle " "a streaming response with multiple msgpack objects (P frames + R frame). " "Replace with:\n" " unpacker = _msgpack.Unpacker(raw=False)\n" " unpacker.feed(resp.content)\n" " for frame in unpacker:\n" " if frame.get('t') == 'R': result = frame; break\n" " elif frame.get('t') == 'X': raise TransportError(...)\n" ) # --------------------------------------------------------------------------- # C2 — client must use Unpacker or stream iteration # --------------------------------------------------------------------------- def test_c2_push_client_uses_unpacker_for_response() -> None: """push_stream_async must iterate response frames with Unpacker or equivalent. The response is a sequence of msgpack objects: P frames then R frame. The client must parse them one by one, not assume a single object. """ from muse.core import transport source = inspect.getsource(transport.HttpTransport.push_stream_coro) uses_unpacker = ( "Unpacker" in source or "MPackStreamReader" in source or "unpackb" not in source # if unpackb gone entirely, something else parses it ) assert uses_unpacker, ( "push_stream_async must use msgpack.Unpacker (or MPackStreamReader) to parse " "the response as a sequence of frames, not msgpack.unpackb for a single object." ) # --------------------------------------------------------------------------- # C3 — client correctly parses P frames + R frame # --------------------------------------------------------------------------- def test_c3_push_client_parses_progress_then_result_frame() -> None: """Client response-parsing logic handles P* then R frame sequence. Simulate the new server response format: two P frames then an R frame. Verify the client extracts the correct PushResult from the R frame. """ import io # Build a mock response body: P, P, R p1 = msgpack.packb({"t": "P", "msg": "committing 100/827…"}, use_bin_type=True) p2 = msgpack.packb({"t": "P", "msg": "committing 200/827…"}, use_bin_type=True) r = msgpack.packb({ "t": "R", "ok": True, "msg": "pushed 827 commits, 6860 objects", "head": "sha256:abc123", "heads": {"main": "sha256:abc123"}, "stored_commits": 827, "stored_objects": 6860, "already_present_objects": 0, "code": 200, }, use_bin_type=True) response_body = p1 + p2 + r # Parse using the same logic the client should use. unpacker = msgpack.Unpacker(raw=False) unpacker.feed(response_body) result: MsgpackDict | None = None progress_msgs: list[str] = [] for frame in unpacker: t = frame.get("t") if t == "P": progress_msgs.append(frame.get("msg", "")) elif t == "R": result = frame break elif t == "X": raise AssertionError(f"Unexpected error frame: {frame}") assert result is not None, "Client must find R frame in response" assert result.get("ok") is True assert result.get("stored_commits") == 827 assert result.get("head") == "sha256:abc123" assert len(progress_msgs) == 2, f"Expected 2 P frames, got: {progress_msgs}" assert "committing 100/827" in progress_msgs[0] # --------------------------------------------------------------------------- # C4 — client must use client.stream(), not client.post() # --------------------------------------------------------------------------- def test_c4_push_client_uses_streaming_http_not_buffered() -> None: """push_stream_coro must use `client.stream()` (async context manager), not `await client.post()` (buffered). With `await client.post()`, httpx downloads the entire response body before returning — P frames only print AFTER the server closes the connection. For a push of 800+ commits (8+ seconds), the terminal goes silent during the entire processing phase, exactly like watching a progress bar that only updates when the task is done. With `async with client.stream()` + `resp.aiter_bytes()`, each chunk is available as soon as the server flushes it. P frames are fed to the Unpacker immediately and printed to stderr in real time — same as git's "remote: Resolving deltas: 100% (N/N), done." sideband stream. This test verifies the source uses the streaming pattern. """ from muse.core import transport source = inspect.getsource(transport.HttpTransport.push_stream_coro) assert "client.stream(" in source, ( "push_stream_coro must use `async with client.stream(...)` for the push POST. " "Using `await client.post()` buffers the full response before returning — " "P frames are printed only after all processing is done, giving the user " "no feedback during the longest phase. " "Fix: replace `await client.post(...)` with `async with client.stream(...) as resp:` " "and iterate `async for chunk in resp.aiter_bytes():` to feed the Unpacker." ) assert "aiter_bytes" in source, ( "push_stream_coro must iterate response chunks with `resp.aiter_bytes()` " "to process P frames as they arrive. " "Found `client.stream()` but not `aiter_bytes` — the streaming context is " "opened but the response is still being read in one shot." ) assert "await client.post(" not in source, ( "push_stream_coro still calls `await client.post(...)`. " "This must be replaced with `async with client.stream(...) as resp:`. " "`client.post()` buffers the full response body before returning, " "defeating the streaming heartbeat mechanism entirely." )