gabriel / muse public
test_mpack_chunked_push.py python
670 lines 25.7 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
1 """TDD tests for true chunked HTTP/1.1 streaming in HttpTransport.push_stream.
2
3 Phase 8 of the MPack protocol rollout: eliminate the ``b"".join(frames)``
4 memory bottleneck by streaming each MPack frame as a distinct HTTP chunk via
5 ``http.client.HTTPConnection`` with ``Transfer-Encoding: chunked``.
6
7 Security upgrade: MSign Authorization is computed over H frame bytes only
8 (not the full body). Security properties are preserved because:
9
10 1. The H frame carries its own embedded Ed25519 signature over the canonical
11 push intent (op, branch, n_objects, n_commits, agent_id, model_id).
12 2. Every O frame is content-addressed — ``sha256:`` prefix means a tampered
13 object is immediately detected on receipt.
14 3. The E frame integrity check cross-validates actual counts against H frame
15 advisory counts, catching truncated or injected streams.
16
17 Signing over H frame bytes is strictly better than signing the full body for
18 chunked streaming because:
19 - Full-body signing requires buffering all frames in memory first.
20 - H-frame signing lets the server validate auth after reading ~500 bytes,
21 while O frames are still in flight.
22
23 Red phase guarantees
24 --------------------
25 All tests in this file are intentionally written against the *desired*
26 behaviour. They are RED against the old implementation (which:
27 - uses ``_open_url`` / urllib (no chunked encoding)
28 - signs over full body bytes
29 - calls ``b"".join(frame_bytes)`` before sending
30 ) and GREEN only after the new implementation is in place.
31 """
32
33 from __future__ import annotations
34
35 import hashlib
36 import unittest.mock
37 from io import BytesIO
38
39 import msgpack
40 import pytest
41
42 from muse.core.mpack import (
43 MPACK_CONTENT_TYPE,
44 MPACK_VERSION,
45 MPackStreamWriter,
46 )
47 from muse.core.transport import HttpTransport, SigningIdentity, TransportError
48 from muse.core._types import long_id
49
50
51 # ---------------------------------------------------------------------------
52 # Helpers
53 # ---------------------------------------------------------------------------
54
55
56 def _make_signing() -> SigningIdentity:
57 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
58 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
59
60
61 def _parse_chunk(raw: bytes) -> bytes:
62 """Parse one HTTP/1.1 chunk: ``{size_hex}\\r\\n{data}\\r\\n`` → data."""
63 nl = raw.index(b"\r\n")
64 return raw[nl + 2 : -2] # strip size line and trailing \r\n
65
66
67 def _make_r_frame(ok: bool = True) -> bytes:
68 w = MPackStreamWriter()
69 return w.write_result(ok=ok, msg="ok", head=long_id("a" * 64))
70
71
72 class MockHTTPResponse:
73 """Minimal http.client.HTTPResponse stand-in."""
74
75 def __init__(self, body: bytes, status: int = 200) -> None:
76 self.status = status
77 self._buf = body
78 self._pos = 0
79
80 def read(self, n: int = -1) -> bytes:
81 if n == -1:
82 chunk = self._buf[self._pos :]
83 self._pos = len(self._buf)
84 return chunk
85 chunk = self._buf[self._pos : self._pos + n]
86 self._pos += n
87 return chunk
88
89 def getheader(self, name: str, default: str | None = None) -> str | None:
90 return default
91
92
93 class MockHTTPConnection:
94 """Captures all http.client interaction for assertion."""
95
96 def __init__(self, host: str = "", port: int = 80, *, timeout: int = 300) -> None:
97 self.host = host
98 self.port = port
99 self.timeout = timeout
100 self.request_method: str = ""
101 self.request_path: str = ""
102 self.headers: dict[str, str] = {} # lower-cased keys
103 self.sends: list[bytes] = [] # raw bytes passed to send()
104 self._response: MockHTTPResponse = MockHTTPResponse(_make_r_frame())
105 self.closed = False
106
107 def putrequest(
108 self, method: str, path: str, skip_accept_encoding: bool = True
109 ) -> None:
110 self.request_method = method
111 self.request_path = path
112
113 def putheader(self, key: str, value: str) -> None:
114 self.headers[key.lower()] = value
115
116 def endheaders(self, body: bytes | None = None) -> None:
117 pass
118
119 def send(self, data: bytes) -> None:
120 self.sends.append(bytes(data))
121
122 def getresponse(self) -> MockHTTPResponse:
123 return self._response
124
125 def close(self) -> None:
126 self.closed = True
127
128
129 def _run_push(
130 transport: HttpTransport,
131 *,
132 objects: list | None = None,
133 commits: list | None = None,
134 snapshots: list | None = None,
135 branch: str = "main",
136 force: bool = False,
137 have: list | None = None,
138 local_head: str | None = None,
139 signing: SigningIdentity | None = None,
140 conn: MockHTTPConnection | None = None,
141 response_body: bytes | None = None,
142 ) -> tuple[MockHTTPConnection, dict]:
143 """Invoke push_stream, intercept ``_open_chunked_connection``, return
144 (captured_conn, push_result_dict)."""
145 mock_conn = conn or MockHTTPConnection()
146 if response_body is not None:
147 mock_conn._response = MockHTTPResponse(response_body)
148
149 def factory(host, port, *, use_ssl, timeout, context=None):
150 mock_conn.host = host
151 mock_conn.port = port
152 return mock_conn
153
154 result: dict = {}
155 with unittest.mock.patch(
156 "muse.core.transport._open_chunked_connection", side_effect=factory
157 ):
158 try:
159 result = transport.push_stream(
160 url="http://localhost:10003",
161 signing=signing,
162 objects=objects or [],
163 commits=commits or [],
164 snapshots=snapshots or [],
165 branch=branch,
166 force=force,
167 have=have or [],
168 local_head=local_head,
169 )
170 except TransportError:
171 pass
172
173 return mock_conn, result
174
175
176 def _decode_sends(sends: list[bytes]) -> list[bytes]:
177 """Decode raw chunk bytes (``{hex}\\r\\n{data}\\r\\n``) into frame payloads.
178 Terminal chunk (``0\\r\\n\\r\\n``) is excluded."""
179 frames = []
180 for raw in sends:
181 if raw == b"0\r\n\r\n":
182 continue
183 try:
184 frames.append(_parse_chunk(raw))
185 except (ValueError, IndexError):
186 pass
187 return frames
188
189
190 def _decode_mpack_frames(sends: list[bytes]) -> list[dict]:
191 """Decode all send() calls into decoded MPack frame dicts."""
192 result = []
193 for payload in _decode_sends(sends):
194 try:
195 obj = msgpack.unpackb(payload, raw=False)
196 if isinstance(obj, dict):
197 result.append(obj)
198 except Exception: # noqa: BLE001
199 pass
200 return result
201
202
203 # ---------------------------------------------------------------------------
204 # Headers: Transfer-Encoding, no Content-Length
205 # ---------------------------------------------------------------------------
206
207
208 class TestChunkedHeaders:
209 def test_transfer_encoding_chunked(self) -> None:
210 """push_stream must set Transfer-Encoding: chunked on the connection."""
211 t = HttpTransport()
212 conn, _ = _run_push(t)
213 assert conn.headers.get("transfer-encoding") == "chunked", (
214 f"Expected 'Transfer-Encoding: chunked', headers: {conn.headers}"
215 )
216
217 def test_no_content_length(self) -> None:
218 """push_stream must NOT set Content-Length — chunked encoding owns framing."""
219 t = HttpTransport()
220 conn, _ = _run_push(t)
221 assert "content-length" not in conn.headers, (
222 f"Content-Length must be absent for chunked streaming, "
223 f"found: {conn.headers.get('content-length')}"
224 )
225
226 def test_content_type_is_mpack(self) -> None:
227 """push_stream must set Content-Type: application/x-muse-mpack."""
228 t = HttpTransport()
229 conn, _ = _run_push(t)
230 assert conn.headers.get("content-type") == MPACK_CONTENT_TYPE
231
232 def test_accept_is_mpack(self) -> None:
233 """push_stream must set Accept: application/x-muse-mpack."""
234 t = HttpTransport()
235 conn, _ = _run_push(t)
236 assert conn.headers.get("accept") == MPACK_CONTENT_TYPE
237
238 def test_uses_post_method(self) -> None:
239 """push_stream must use POST method."""
240 t = HttpTransport()
241 conn, _ = _run_push(t)
242 assert conn.request_method == "POST"
243
244 def test_path_ends_with_push_stream(self) -> None:
245 """push_stream must POST to /push/stream."""
246 t = HttpTransport()
247 conn, _ = _run_push(t)
248 assert conn.request_path.endswith("/push/stream"), (
249 f"Expected path ending /push/stream, got {conn.request_path!r}"
250 )
251
252
253 # ---------------------------------------------------------------------------
254 # Signing: Authorization computed over H frame bytes only
255 # ---------------------------------------------------------------------------
256
257
258 class TestChunkedSigning:
259 def test_no_auth_header_when_unsigned(self) -> None:
260 """Without signing identity, no Authorization header is sent."""
261 t = HttpTransport()
262 conn, _ = _run_push(t, signing=None)
263 assert "authorization" not in conn.headers
264
265 def test_auth_header_present_when_signed(self) -> None:
266 """With signing identity, Authorization header is set."""
267 t = HttpTransport()
268 signing = _make_signing()
269 conn, _ = _run_push(t, signing=signing)
270 assert "authorization" in conn.headers, (
271 "Authorization header missing when signing identity provided"
272 )
273
274 def test_auth_signed_over_empty_body(self) -> None:
275 """MSign body_bytes must be None/empty for chunked streaming push.
276
277 Full-body signing is incompatible with streaming (body unknown upfront).
278 Body integrity is provided by the H frame's embedded Ed25519, object
279 content-addressing, and E frame counts. MSign covers identity + replay
280 via method/host/path/ts only; body_bytes must be None → SHA256("").
281 """
282 import muse.core.msign as _msign_mod
283 real_build = _msign_mod.build_msign_header # save before patching
284
285 t = HttpTransport()
286 signing = _make_signing()
287 captured_body: list[bytes | None] = []
288
289 def spy_build_msign(signing_id, method, url, body_bytes=None, **kw):
290 captured_body.append(body_bytes)
291 return real_build(signing_id, method, url, body_bytes, **kw)
292
293 with unittest.mock.patch(
294 "muse.core.msign.build_msign_header", side_effect=spy_build_msign
295 ):
296 _run_push(t, signing=signing)
297
298 assert captured_body, "build_msign_header was never called"
299 # For streaming push, body_bytes must be None (→ SHA256("") in canonical msg).
300 signed_body = captured_body[0]
301 assert signed_body is None, (
302 f"Auth body_bytes must be None for streaming push, got {signed_body!r:.64}"
303 )
304
305 def test_auth_not_signed_over_full_body(self) -> None:
306 """Authorization must NOT cover the full concatenated frame stream.
307
308 With multiple objects, the full body is much larger than H frame alone.
309 If body_bytes signed is larger than a realistic H frame (~2KB ceiling),
310 the old full-body signing path was NOT replaced.
311 """
312 import muse.core.msign as _msign_mod
313 real_build = _msign_mod.build_msign_header
314
315 t = HttpTransport()
316 signing = _make_signing()
317 captured_body: list[bytes] = []
318
319 def spy_build_msign(signing_id, method, url, body_bytes=None, **kw):
320 captured_body.append(body_bytes)
321 return real_build(signing_id, method, url, body_bytes, **kw)
322
323 # 10 objects × 4KB each = 40KB full body; H frame is <2KB
324 objects = []
325 for i in range(10):
326 content = f"object-{i}-".encode() * 400 # 4KB each
327 oid = long_id(hashlib.sha256(content).hexdigest())
328 objects.append({"object_id": oid, "content": content, "path": f"{i}.bin", "encoding": "raw"})
329
330 with unittest.mock.patch(
331 "muse.core.msign.build_msign_header", side_effect=spy_build_msign
332 ):
333 _run_push(t, signing=signing, objects=objects)
334
335 assert captured_body
336 signed_body = captured_body[0]
337 # For streaming push, body_bytes must be None → SHA256("") in canonical msg.
338 # The full 40KB body must NOT be buffered and signed.
339 assert signed_body is None, (
340 f"Authorization body_bytes must be None for streaming push, "
341 f"got {len(signed_body or b'')} bytes — full body was signed."
342 )
343
344
345 # ---------------------------------------------------------------------------
346 # Frame sequence and incremental sends
347 # ---------------------------------------------------------------------------
348
349
350 class TestChunkedFrameSequence:
351 def test_h_frame_first_send(self) -> None:
352 """H frame must be the first chunk sent on the connection."""
353 t = HttpTransport()
354 conn, _ = _run_push(t)
355 frames = _decode_mpack_frames(conn.sends)
356 assert frames, "No MPack frames decoded from sends"
357 assert frames[0].get("t") == "H", (
358 f"First frame must be H, got t={frames[0].get('t')!r}"
359 )
360
361 def test_terminal_chunk_is_last_send(self) -> None:
362 """Last send() call must be the terminal chunk ``0\\r\\n\\r\\n``."""
363 t = HttpTransport()
364 conn, _ = _run_push(t)
365 assert conn.sends, "No sends recorded"
366 assert conn.sends[-1] == b"0\r\n\r\n", (
367 f"Last send must be terminal chunk, got {conn.sends[-1]!r}"
368 )
369
370 def test_empty_push_has_h_c_e_terminal(self) -> None:
371 """Empty push (no objects) must send H, C, E, terminal — 4 chunks."""
372 t = HttpTransport()
373 conn, _ = _run_push(t)
374 frames = _decode_mpack_frames(conn.sends)
375 tags = [f.get("t") for f in frames]
376 assert "H" in tags and "C" in tags and "E" in tags, (
377 f"Expected H, C, E frames; got {tags}"
378 )
379 assert conn.sends[-1] == b"0\r\n\r\n"
380
381 def test_o_frames_between_h_and_c(self) -> None:
382 """O frames must appear after H and before C."""
383 t = HttpTransport()
384 content = b"object for sequencing test"
385 oid = long_id(hashlib.sha256(content).hexdigest())
386 objects = [{"object_id": oid, "content": content, "path": "f.bin", "encoding": "raw"}]
387 conn, _ = _run_push(t, objects=objects)
388 frames = _decode_mpack_frames(conn.sends)
389 tags = [f.get("t") for f in frames]
390 assert "O" in tags, f"No O frame sent; tags: {tags}"
391 h_i = tags.index("H")
392 o_i = tags.index("O")
393 c_i = tags.index("C")
394 assert h_i < o_i < c_i, f"Wrong order: {tags}"
395
396 def test_each_object_is_separate_send(self) -> None:
397 """Each O frame must be a separate send() call — not batched together."""
398 t = HttpTransport()
399 objects = []
400 for i in range(5):
401 content = f"separate-object-{i}".encode() * 50
402 oid = long_id(hashlib.sha256(content).hexdigest())
403 objects.append({"object_id": oid, "content": content, "path": f"{i}.bin", "encoding": "raw"})
404 conn, _ = _run_push(t, objects=objects)
405 o_frames = [f for f in _decode_mpack_frames(conn.sends) if f.get("t") == "O"]
406 assert len(o_frames) == 5, f"Expected 5 O frames, got {len(o_frames)}"
407
408 def test_send_count_matches_frame_count(self) -> None:
409 """Total send() calls == number of frames + 1 terminal chunk."""
410 t = HttpTransport()
411 n_objects = 3
412 objects = []
413 for i in range(n_objects):
414 content = f"count-test-{i}".encode()
415 oid = long_id(hashlib.sha256(content).hexdigest())
416 objects.append({"object_id": oid, "content": content, "path": f"{i}.bin", "encoding": "raw"})
417 conn, _ = _run_push(t, objects=objects)
418 # H + (N objects) + C + E + terminal
419 expected_sends = 1 + n_objects + 1 + 1 + 1
420 assert len(conn.sends) == expected_sends, (
421 f"Expected {expected_sends} send() calls, got {len(conn.sends)}"
422 )
423
424
425 # ---------------------------------------------------------------------------
426 # Memory: no single large allocation
427 # ---------------------------------------------------------------------------
428
429
430 class TestChunkedMemoryBounded:
431 def test_no_single_send_contains_all_frames(self) -> None:
432 """No single send() call should contain more than one frame.
433
434 Verifies that the implementation does NOT call b''.join(all_frames)
435 and send the result in one shot. Each send must decode to a single
436 valid MPack frame (or be the terminal chunk).
437 """
438 t = HttpTransport()
439 objects = []
440 for i in range(10):
441 content = f"mem-test-{i}".encode() * 100
442 oid = long_id(hashlib.sha256(content).hexdigest())
443 objects.append({"object_id": oid, "content": content, "path": f"{i}.bin", "encoding": "raw"})
444 conn, _ = _run_push(t, objects=objects)
445
446 for i, raw in enumerate(conn.sends):
447 if raw == b"0\r\n\r\n":
448 continue # terminal chunk is fine
449 # Each chunk payload must decode to exactly one MPack dict
450 payload = _parse_chunk(raw)
451 unpacked = msgpack.unpackb(payload, raw=False)
452 assert isinstance(unpacked, dict), (
453 f"send()[{i}] decoded to {type(unpacked)}, expected a single frame dict"
454 )
455
456 def test_peak_send_size_bounded_per_object(self) -> None:
457 """No individual send() call should exceed max(object_size) × 2.
458
459 The fudge factor of ×2 accounts for compression overhead and chunk
460 framing. If one send is enormous it means frames were concatenated.
461 """
462 t = HttpTransport()
463 max_obj_size = 8 * 1024 # 8KB per object
464 objects = []
465 for i in range(20):
466 content = (f"bounded-{i}:").encode() * (max_obj_size // 10)
467 oid = long_id(hashlib.sha256(content).hexdigest())
468 objects.append({"object_id": oid, "content": content, "path": f"{i}.bin", "encoding": "raw"})
469 conn, _ = _run_push(t, objects=objects)
470
471 for i, raw in enumerate(conn.sends):
472 if raw == b"0\r\n\r\n":
473 continue
474 assert len(raw) < max_obj_size * 2 + 512, (
475 f"send()[{i}] is {len(raw)} bytes — looks like frames were "
476 f"concatenated (expected < {max_obj_size * 2 + 512})"
477 )
478
479
480 # ---------------------------------------------------------------------------
481 # Connection type: HTTP vs HTTPS
482 # ---------------------------------------------------------------------------
483
484
485 class TestChunkedConnectionType:
486 def _run_with_url(self, url: str) -> tuple[str, int, bool]:
487 """Return (host, port, use_ssl) captured from _open_chunked_connection."""
488 captured: dict = {}
489
490 class CaptureConn(MockHTTPConnection):
491 def __init__(self, host, port, *, timeout=300, context=None):
492 super().__init__(host, port, timeout=timeout)
493 captured["host"] = host
494 captured["port"] = port
495 captured["use_ssl"] = context is not None # HTTPSConnection passes context
496
497 def factory(host, port, *, use_ssl, timeout, context=None):
498 captured["use_ssl"] = use_ssl
499 captured["host"] = host
500 captured["port"] = port
501 return MockHTTPConnection(host, port, timeout=timeout)
502
503 t = HttpTransport()
504 with unittest.mock.patch(
505 "muse.core.transport._open_chunked_connection", side_effect=factory
506 ):
507 try:
508 t.push_stream(
509 url=url,
510 signing=None,
511 objects=[],
512 commits=[],
513 snapshots=[],
514 branch="main",
515 force=False,
516 have=[],
517 )
518 except Exception: # noqa: BLE001
519 pass
520 return captured.get("host", ""), captured.get("port", 0), captured.get("use_ssl", False)
521
522 def test_http_url_uses_use_ssl_false(self) -> None:
523 """HTTP URL must open connection with use_ssl=False."""
524 _, _, use_ssl = self._run_with_url("http://localhost:10003")
525 assert use_ssl is False, f"Expected use_ssl=False for HTTP, got {use_ssl}"
526
527 def test_https_url_uses_use_ssl_true(self) -> None:
528 """HTTPS URL must open connection with use_ssl=True."""
529 _, _, use_ssl = self._run_with_url("https://staging.musehub.ai")
530 assert use_ssl is True, f"Expected use_ssl=True for HTTPS, got {use_ssl}"
531
532 def test_host_extracted_correctly(self) -> None:
533 """Host must be extracted from URL netloc."""
534 host, _, _ = self._run_with_url("http://localhost:10003")
535 assert host == "localhost"
536
537 def test_port_extracted_correctly(self) -> None:
538 """Port must be extracted from URL."""
539 _, port, _ = self._run_with_url("http://localhost:10003")
540 assert port == 10003
541
542 def test_connection_closed_after_success(self) -> None:
543 """Connection must be closed after a successful push."""
544 t = HttpTransport()
545 conn, _ = _run_push(t)
546 assert conn.closed, "Connection was not closed after push"
547
548 def test_connection_closed_after_error(self) -> None:
549 """Connection must be closed even when server returns an error frame."""
550 t = HttpTransport()
551 w = MPackStreamWriter()
552 error_resp = w.write_error(msg="conflict", code=409)
553 conn, _ = _run_push(t, response_body=error_resp)
554 assert conn.closed, "Connection was not closed after error response"
555
556
557 # ---------------------------------------------------------------------------
558 # Response parsing
559 # ---------------------------------------------------------------------------
560
561
562 class TestChunkedResponseParsing:
563 def test_r_frame_ok_returns_push_result(self) -> None:
564 """R frame with ok=True must return PushResult dict with ok=True."""
565 t = HttpTransport()
566 tip = long_id("b" * 64)
567 w = MPackStreamWriter()
568 response = w.write_result(ok=True, msg="pushed", heads={"main": tip}, head=tip)
569 conn, result = _run_push(t, response_body=response)
570 assert result.get("ok") is True, f"Expected ok=True, got {result}"
571
572 def test_r_frame_message_preserved(self) -> None:
573 """R frame message must be in the PushResult."""
574 t = HttpTransport()
575 w = MPackStreamWriter()
576 response = w.write_result(ok=True, msg="all good", head=long_id("c" * 64))
577 _, result = _run_push(t, response_body=response)
578 assert result.get("message") == "all good"
579
580 def test_x_frame_raises_transport_error(self) -> None:
581 """X frame in response must raise TransportError."""
582 t = HttpTransport()
583 w = MPackStreamWriter()
584 error_resp = w.write_error(msg="non-fast-forward", code=409)
585
586 def factory(host, port, *, use_ssl, timeout, context=None):
587 conn = MockHTTPConnection()
588 conn._response = MockHTTPResponse(error_resp)
589 return conn
590
591 with unittest.mock.patch(
592 "muse.core.transport._open_chunked_connection", side_effect=factory
593 ):
594 with pytest.raises(TransportError) as exc_info:
595 HttpTransport().push_stream(
596 url="http://localhost:10003",
597 signing=None,
598 objects=[],
599 commits=[],
600 snapshots=[],
601 branch="main",
602 force=False,
603 have=[],
604 )
605 assert exc_info.value.status_code == 409
606
607 def test_p_frame_does_not_raise(self) -> None:
608 """P (progress) frames must be consumed silently — no exception."""
609 t = HttpTransport()
610 w = MPackStreamWriter()
611 response = (
612 w.write_progress(msg="resolving objects", pct=25.0)
613 + w.write_progress(msg="writing objects", pct=75.0)
614 + w.write_result(ok=True, msg="done", head=long_id("d" * 64))
615 )
616 conn, result = _run_push(t, response_body=response)
617 assert result.get("ok") is True
618
619 def test_http_4xx_raises_transport_error(self) -> None:
620 """HTTP 4xx status must raise TransportError with that status code."""
621 t = HttpTransport()
622
623 def factory(host, port, *, use_ssl, timeout, context=None):
624 conn = MockHTTPConnection()
625 conn._response = MockHTTPResponse(b"forbidden", status=403)
626 return conn
627
628 with unittest.mock.patch(
629 "muse.core.transport._open_chunked_connection", side_effect=factory
630 ):
631 with pytest.raises(TransportError) as exc_info:
632 HttpTransport().push_stream(
633 url="http://localhost:10003",
634 signing=None,
635 objects=[],
636 commits=[],
637 snapshots=[],
638 branch="main",
639 force=False,
640 have=[],
641 )
642 assert exc_info.value.status_code == 403
643
644 def test_no_result_frame_raises_transport_error(self) -> None:
645 """Server closing stream without R frame must raise TransportError."""
646 t = HttpTransport()
647
648 def factory(host, port, *, use_ssl, timeout, context=None):
649 conn = MockHTTPConnection()
650 # Response with only a progress frame, no R frame
651 w = MPackStreamWriter()
652 conn._response = MockHTTPResponse(
653 w.write_progress(msg="working...", pct=50.0)
654 )
655 return conn
656
657 with unittest.mock.patch(
658 "muse.core.transport._open_chunked_connection", side_effect=factory
659 ):
660 with pytest.raises(TransportError):
661 HttpTransport().push_stream(
662 url="http://localhost:10003",
663 signing=None,
664 objects=[],
665 commits=[],
666 snapshots=[],
667 branch="main",
668 force=False,
669 have=[],
670 )
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago