test_push_streaming_client.py
python
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d
feat(pack): delta-encode snapshots in MPackBundle wire format
Sonnet 4.6
minor
⚠ breaking
126 days ago
| 1 | """TDD — push client must parse response as a frame stream, not a single dict. |
| 2 | |
| 3 | The push/stream response is documented as a sequence of msgpack objects: |
| 4 | zero or more PROGRESS frames (t='P') followed by one RESULT frame (t='R'). |
| 5 | |
| 6 | The old client implementation did: |
| 7 | result = _msgpack.unpackb(resp.content, raw=False) |
| 8 | |
| 9 | This buffers the full response body and parses ONE msgpack object. |
| 10 | When the server starts streaming P frames before the final result, this breaks: |
| 11 | - `unpackb` raises an exception (multiple objects look like trailing garbage) |
| 12 | - The client never sees P frames (cannot print progress to stderr) |
| 13 | |
| 14 | Tests: |
| 15 | |
| 16 | C1 Static: push_stream_async must NOT call `_msgpack.unpackb(resp.content)`. |
| 17 | |
| 18 | C2 Static: push_stream_async must use an Unpacker or MPackStreamReader to |
| 19 | iterate through all response frames until it finds the RESULT frame. |
| 20 | |
| 21 | C3 Integration: The client's response-parsing logic correctly handles a |
| 22 | response that contains P frames followed by an R frame. |
| 23 | |
| 24 | C4 Static: push_stream_coro must use `client.stream()` (streaming HTTP), |
| 25 | not `client.post()` (buffered HTTP), so P frames are printed to stderr |
| 26 | as each chunk arrives — the same way git prints "remote: ..." in real |
| 27 | time during a push, not after the server closes the connection. |
| 28 | """ |
| 29 | from __future__ import annotations |
| 30 | |
| 31 | import inspect |
| 32 | |
| 33 | import msgpack |
| 34 | import pytest |
| 35 | |
| 36 | from muse.core.pack import PushResult |
| 37 | from muse.core.types import MsgpackDict |
| 38 | |
| 39 | |
| 40 | # --------------------------------------------------------------------------- |
| 41 | # C1 — client must NOT call unpackb(resp.content) |
| 42 | # --------------------------------------------------------------------------- |
| 43 | |
| 44 | def test_c1_push_client_does_not_unpackb_full_content() -> None: |
| 45 | """push_stream_async must not call `_msgpack.unpackb(resp.content)`. |
| 46 | |
| 47 | When the server starts streaming P frames before the R frame, resp.content |
| 48 | contains multiple concatenated msgpack objects. `unpackb` only parses the |
| 49 | first one and raises ExtraData (or silently ignores the rest), so the |
| 50 | client would never see the RESULT frame. |
| 51 | |
| 52 | The correct approach: iterate with msgpack.Unpacker (or MPackStreamReader) |
| 53 | over resp.content until the R or X frame is found. |
| 54 | """ |
| 55 | from muse.core import transport |
| 56 | |
| 57 | source = inspect.getsource(transport.HttpTransport.push_stream_coro) |
| 58 | |
| 59 | assert "_msgpack.unpackb(resp.content" not in source and "unpackb(resp.content" not in source, ( |
| 60 | "push_stream_async calls `_msgpack.unpackb(resp.content)` which cannot handle " |
| 61 | "a streaming response with multiple msgpack objects (P frames + R frame). " |
| 62 | "Replace with:\n" |
| 63 | " unpacker = _msgpack.Unpacker(raw=False)\n" |
| 64 | " unpacker.feed(resp.content)\n" |
| 65 | " for frame in unpacker:\n" |
| 66 | " if frame.get('t') == 'R': result = frame; break\n" |
| 67 | " elif frame.get('t') == 'X': raise TransportError(...)\n" |
| 68 | ) |
| 69 | |
| 70 | |
| 71 | # --------------------------------------------------------------------------- |
| 72 | # C2 — client must use Unpacker or stream iteration |
| 73 | # --------------------------------------------------------------------------- |
| 74 | |
| 75 | def test_c2_push_client_uses_unpacker_for_response() -> None: |
| 76 | """push_stream_async must iterate response frames with Unpacker or equivalent. |
| 77 | |
| 78 | The response is a sequence of msgpack objects: P frames then R frame. |
| 79 | The client must parse them one by one, not assume a single object. |
| 80 | """ |
| 81 | from muse.core import transport |
| 82 | |
| 83 | source = inspect.getsource(transport.HttpTransport.push_stream_coro) |
| 84 | |
| 85 | uses_unpacker = ( |
| 86 | "Unpacker" in source |
| 87 | or "MPackStreamReader" in source |
| 88 | or "unpackb" not in source # if unpackb gone entirely, something else parses it |
| 89 | ) |
| 90 | assert uses_unpacker, ( |
| 91 | "push_stream_async must use msgpack.Unpacker (or MPackStreamReader) to parse " |
| 92 | "the response as a sequence of frames, not msgpack.unpackb for a single object." |
| 93 | ) |
| 94 | |
| 95 | |
| 96 | # --------------------------------------------------------------------------- |
| 97 | # C3 — client correctly parses P frames + R frame |
| 98 | # --------------------------------------------------------------------------- |
| 99 | |
| 100 | def test_c3_push_client_parses_progress_then_result_frame() -> None: |
| 101 | """Client response-parsing logic handles P* then R frame sequence. |
| 102 | |
| 103 | Simulate the new server response format: two P frames then an R frame. |
| 104 | Verify the client extracts the correct PushResult from the R frame. |
| 105 | """ |
| 106 | import io |
| 107 | |
| 108 | # Build a mock response body: P, P, R |
| 109 | p1 = msgpack.packb({"t": "P", "msg": "committing 100/827…"}, use_bin_type=True) |
| 110 | p2 = msgpack.packb({"t": "P", "msg": "committing 200/827…"}, use_bin_type=True) |
| 111 | r = msgpack.packb({ |
| 112 | "t": "R", |
| 113 | "ok": True, |
| 114 | "msg": "pushed 827 commits, 6860 objects", |
| 115 | "head": "sha256:abc123", |
| 116 | "heads": {"main": "sha256:abc123"}, |
| 117 | "stored_commits": 827, |
| 118 | "stored_objects": 6860, |
| 119 | "already_present_objects": 0, |
| 120 | "code": 200, |
| 121 | }, use_bin_type=True) |
| 122 | |
| 123 | response_body = p1 + p2 + r |
| 124 | |
| 125 | # Parse using the same logic the client should use. |
| 126 | unpacker = msgpack.Unpacker(raw=False) |
| 127 | unpacker.feed(response_body) |
| 128 | |
| 129 | result: MsgpackDict | None = None |
| 130 | progress_msgs: list[str] = [] |
| 131 | for frame in unpacker: |
| 132 | t = frame.get("t") |
| 133 | if t == "P": |
| 134 | progress_msgs.append(frame.get("msg", "")) |
| 135 | elif t == "R": |
| 136 | result = frame |
| 137 | break |
| 138 | elif t == "X": |
| 139 | raise AssertionError(f"Unexpected error frame: {frame}") |
| 140 | |
| 141 | assert result is not None, "Client must find R frame in response" |
| 142 | assert result.get("ok") is True |
| 143 | assert result.get("stored_commits") == 827 |
| 144 | assert result.get("head") == "sha256:abc123" |
| 145 | assert len(progress_msgs) == 2, f"Expected 2 P frames, got: {progress_msgs}" |
| 146 | assert "committing 100/827" in progress_msgs[0] |
| 147 | |
| 148 | |
| 149 | # --------------------------------------------------------------------------- |
| 150 | # C4 — client must use client.stream(), not client.post() |
| 151 | # --------------------------------------------------------------------------- |
| 152 | |
| 153 | def test_c4_push_client_uses_streaming_http_not_buffered() -> None: |
| 154 | """push_stream_coro must use `client.stream()` (async context manager), |
| 155 | not `await client.post()` (buffered). |
| 156 | |
| 157 | With `await client.post()`, httpx downloads the entire response body before |
| 158 | returning — P frames only print AFTER the server closes the connection. |
| 159 | For a push of 800+ commits (8+ seconds), the terminal goes silent during |
| 160 | the entire processing phase, exactly like watching a progress bar that only |
| 161 | updates when the task is done. |
| 162 | |
| 163 | With `async with client.stream()` + `resp.aiter_bytes()`, each chunk is |
| 164 | available as soon as the server flushes it. P frames are fed to the |
| 165 | Unpacker immediately and printed to stderr in real time — same as git's |
| 166 | "remote: Resolving deltas: 100% (N/N), done." sideband stream. |
| 167 | |
| 168 | This test verifies the source uses the streaming pattern. |
| 169 | """ |
| 170 | from muse.core import transport |
| 171 | |
| 172 | source = inspect.getsource(transport.HttpTransport.push_stream_coro) |
| 173 | |
| 174 | assert "client.stream(" in source, ( |
| 175 | "push_stream_coro must use `async with client.stream(...)` for the push POST. " |
| 176 | "Using `await client.post()` buffers the full response before returning — " |
| 177 | "P frames are printed only after all processing is done, giving the user " |
| 178 | "no feedback during the longest phase. " |
| 179 | "Fix: replace `await client.post(...)` with `async with client.stream(...) as resp:` " |
| 180 | "and iterate `async for chunk in resp.aiter_bytes():` to feed the Unpacker." |
| 181 | ) |
| 182 | |
| 183 | assert "aiter_bytes" in source, ( |
| 184 | "push_stream_coro must iterate response chunks with `resp.aiter_bytes()` " |
| 185 | "to process P frames as they arrive. " |
| 186 | "Found `client.stream()` but not `aiter_bytes` — the streaming context is " |
| 187 | "opened but the response is still being read in one shot." |
| 188 | ) |
| 189 | |
| 190 | assert "await client.post(" not in source, ( |
| 191 | "push_stream_coro still calls `await client.post(...)`. " |
| 192 | "This must be replaced with `async with client.stream(...) as resp:`. " |
| 193 | "`client.post()` buffers the full response body before returning, " |
| 194 | "defeating the streaming heartbeat mechanism entirely." |
| 195 | ) |
File History
1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d
feat(pack): delta-encode snapshots in MPackBundle wire format
Sonnet 4.6
minor
⚠
126 days ago