gabriel / muse public
test_core_transport.py python
632 lines 26.0 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """Tests for muse.core.transport — HttpTransport and response parsers."""
2
3 from __future__ import annotations
4
5 import json
6 import signal
7 import socket
8 import threading
9 import time
10 import unittest.mock
11
12 import msgpack
13 import pytest
14
15 from muse.core.types import MsgpackDict, b64url_decode, blob_id
16 from muse.core.pack import MPackBundle, RemoteInfo
17 from muse.core.msign import build_msign_header
18 from muse.core.transport import (
19 _Request,
20 HttpTransport,
21 SigningIdentity,
22 TransportError,
23 _parse_bundle,
24 _parse_push_result,
25 _parse_remote_info,
26 )
27
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33
34 def _make_signing() -> "SigningIdentity":
35 """Generate a fresh Ed25519 SigningIdentity for tests."""
36 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
37 from muse.core.transport import SigningIdentity
38 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
39
40
41 def _mock_response(
42 body: bytes,
43 status: int = 200,
44 content_type: str = "application/x-msgpack",
45 ) -> unittest.mock.MagicMock:
46 """Return a mock httpx response."""
47 resp = unittest.mock.MagicMock()
48 resp.content = body
49 resp.status_code = status
50 resp.headers = {"Content-Type": content_type}
51 return resp
52
53
54 def _mp(data: MsgpackDict) -> bytes:
55 """Encode data as msgpack."""
56 return msgpack.packb(data, use_bin_type=True)
57
58
59
60 # ---------------------------------------------------------------------------
61 # _parse_remote_info
62 # ---------------------------------------------------------------------------
63
64
65 class TestParseRemoteInfo:
66 def test_valid_response(self) -> None:
67 raw = _mp(
68 {
69 "repo_id": "r123",
70 "domain": "midi",
71 "default_branch": "main",
72 "branch_heads": {"main": "abc123", "dev": "def456"},
73 }
74 )
75 info = _parse_remote_info(raw)
76 assert info["repo_id"] == "r123"
77 assert info["domain"] == "midi"
78 assert info["default_branch"] == "main"
79 assert info["branch_heads"] == {"main": "abc123", "dev": "def456"}
80
81 def test_invalid_msgpack_raises_transport_error(self) -> None:
82 with pytest.raises(TransportError):
83 _parse_remote_info(b"\xff\xff\xff\xff\xff invalid")
84
85 def test_non_dict_response_returns_defaults(self) -> None:
86 raw = _mp([1, 2, 3])
87 info = _parse_remote_info(raw)
88 assert info["repo_id"] == ""
89 assert info["branch_heads"] == {}
90
91 def test_missing_fields_get_defaults(self) -> None:
92 raw = _mp({"repo_id": "x"})
93 info = _parse_remote_info(raw)
94 assert info["repo_id"] == "x"
95 assert info["domain"] == "midi"
96 assert info["default_branch"] == "main"
97 assert info["branch_heads"] == {}
98
99 def test_non_string_branch_heads_excluded(self) -> None:
100 raw = _mp({"branch_heads": {"main": "abc", "bad": 123}})
101 info = _parse_remote_info(raw)
102 assert "main" in info["branch_heads"]
103 assert "bad" not in info["branch_heads"]
104
105
106
107 # ---------------------------------------------------------------------------
108 # build_msign_header — module-level signing utility
109 # ---------------------------------------------------------------------------
110
111
112 class TestBuildMsignHeader:
113 def _make_signing(self) -> SigningIdentity:
114 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
115
116 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
117
118 def test_header_format(self) -> None:
119 header = build_msign_header(self._make_signing(), "GET", "https://example.com/path", None)
120 assert header.startswith('MSign handle="testuser"')
121 assert " ts=" in header
122 assert " sig=" in header
123
124 def test_timestamp_is_recent(self) -> None:
125 import time
126
127 before = int(time.time())
128 header = build_msign_header(self._make_signing(), "GET", "https://example.com/p", None)
129 after = int(time.time())
130 ts_part = next(p for p in header.split() if p.startswith("ts="))
131 ts = int(ts_part[3:])
132 assert before <= ts <= after + 1
133
134 def test_signature_is_verifiable(self) -> None:
135 """The sig= value must verify against the Ed25519 public key for the canonical input."""
136 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
137
138 private_key = Ed25519PrivateKey.generate()
139 signing = SigningIdentity(handle="testuser", private_key=private_key)
140 method = "POST"
141 url = "https://hub.example.com/owner/repo/push"
142 body = b"some body data"
143
144 header = build_msign_header(signing, method, url, body)
145
146 parts: dict[str, str] = {}
147 for part in header[len("MSign "):].split():
148 k, _, v = part.partition("=")
149 parts[k] = v.strip('"')
150
151 ts = int(parts["ts"])
152 sig_bytes = b64url_decode(parts["sig"])
153
154 body_hash = blob_id(body)
155 canonical = f"ed25519\n{method}\nhub.example.com\n/owner/repo/push\n{ts}\n{body_hash}".encode()
156
157 # raises cryptography.exceptions.InvalidSignature on failure
158 private_key.public_key().verify(sig_bytes, canonical)
159
160 def test_query_string_included_in_canonical(self) -> None:
161 """Query parameters must be part of the signed path, not dropped."""
162 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
163
164 private_key = Ed25519PrivateKey.generate()
165 signing = SigningIdentity(handle="u", private_key=private_key)
166 url = "https://hub.example.com/path?foo=bar&baz=1"
167
168 header = build_msign_header(signing, "GET", url, None)
169
170 parts: dict[str, str] = {}
171 for part in header[len("MSign "):].split():
172 k, _, v = part.partition("=")
173 parts[k] = v.strip('"')
174
175 ts = int(parts["ts"])
176 sig_bytes = b64url_decode(parts["sig"])
177 body_hash = blob_id(b"")
178 canonical = f"ed25519\nGET\nhub.example.com\n/path?foo=bar&baz=1\n{ts}\n{body_hash}".encode()
179
180 private_key.public_key().verify(sig_bytes, canonical)
181
182 def test_none_body_treated_as_empty_bytes(self) -> None:
183 """None and b'' must produce the same SHA-256 body hash in the canonical form."""
184 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
185
186 private_key = Ed25519PrivateKey.generate()
187 signing = SigningIdentity(handle="u", private_key=private_key)
188 url = "https://hub.example.com/path"
189
190 header = build_msign_header(signing, "GET", url, None)
191
192 parts: dict[str, str] = {}
193 for part in header[len("MSign "):].split():
194 k, _, v = part.partition("=")
195 parts[k] = v.strip('"')
196
197 ts = int(parts["ts"])
198 sig_bytes = b64url_decode(parts["sig"])
199 # body=None → b"" → blob_id(b"") is the canonical body hash
200 body_hash = blob_id(b"")
201 canonical = f"ed25519\nGET\nhub.example.com\n/path\n{ts}\n{body_hash}".encode()
202
203 private_key.public_key().verify(sig_bytes, canonical)
204
205 def test_different_methods_produce_different_headers(self) -> None:
206 """Two calls with different methods on the same URL must differ (method is in canonical)."""
207
208 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
209
210 private_key = Ed25519PrivateKey.generate()
211 signing = SigningIdentity(handle="u", private_key=private_key)
212
213 with unittest.mock.patch("muse.core.msign.time") as mt:
214 mt.time.return_value = 1_700_000_000
215 h_get = build_msign_header(signing, "GET", "https://example.com/x", b"")
216 h_post = build_msign_header(signing, "POST", "https://example.com/x", b"")
217
218 assert h_get != h_post
219
220 def test_different_bodies_produce_different_sigs(self) -> None:
221 """Body content must influence the signature (body hash is in canonical)."""
222
223 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
224
225 private_key = Ed25519PrivateKey.generate()
226 signing = SigningIdentity(handle="u", private_key=private_key)
227
228 with unittest.mock.patch("muse.core.msign.time") as mt:
229 mt.time.return_value = 1_700_000_000
230 h1 = build_msign_header(signing, "POST", "https://example.com/x", b"body-a")
231 h2 = build_msign_header(signing, "POST", "https://example.com/x", b"body-b")
232
233 assert h1 != h2
234
235 def test_handle_embedded_in_header(self) -> None:
236 """The MSign header must carry the signing identity's handle."""
237
238 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
239
240 private_key = Ed25519PrivateKey.generate()
241 signing = SigningIdentity(handle="my-agent-42", private_key=private_key)
242 header = build_msign_header(signing, "GET", "https://example.com/x", None)
243 assert 'handle="my-agent-42"' in header
244
245
246 # ---------------------------------------------------------------------------
247 # _parse_bundle
248 # ---------------------------------------------------------------------------
249
250
251 class TestParseBundle:
252 def test_empty_msgpack_object_returns_empty_bundle(self) -> None:
253 bundle = _parse_bundle(_mp({}))
254 assert bundle == {}
255
256 def test_non_dict_returns_empty_bundle(self) -> None:
257 bundle = _parse_bundle(_mp([]))
258 assert bundle == {}
259
260 def test_commits_extracted(self) -> None:
261 raw = _mp(
262 {
263 "commits": [
264 {
265 "commit_id": "c1",
266 "repo_id": "r1",
267 "branch": "main",
268 "snapshot_id": "1" * 64,
269 "message": "test",
270 "committed_at": "2026-01-01T00:00:00+00:00",
271 "parent_commit_id": None,
272 "parent2_commit_id": None,
273 "author": "bob",
274 "metadata": {},
275 }
276 ]
277 }
278 )
279 bundle = _parse_bundle(raw)
280 commits = bundle.get("commits") or []
281 assert len(commits) == 1
282 assert commits[0]["commit_id"] == "c1"
283
284 def test_objects_extracted(self) -> None:
285 raw = _mp(
286 {
287 "objects": [
288 {
289 "object_id": "abc123",
290 "content": b"hello",
291 }
292 ]
293 }
294 )
295 bundle = _parse_bundle(raw)
296 objs = bundle.get("objects") or []
297 assert len(objs) == 1
298 assert objs[0]["object_id"] == "abc123"
299 assert objs[0]["content"] == b"hello"
300
301 def test_object_missing_content_excluded(self) -> None:
302 raw = _mp({"objects": [{"object_id": "abc"}]})
303 bundle = _parse_bundle(raw)
304 assert (bundle.get("objects") or []) == []
305
306 def test_branch_heads_extracted(self) -> None:
307 raw = _mp({"branch_heads": {"main": "abc123"}})
308 bundle = _parse_bundle(raw)
309 assert bundle.get("branch_heads") == {"main": "abc123"}
310
311
312 # ---------------------------------------------------------------------------
313 # _parse_push_result
314 # ---------------------------------------------------------------------------
315
316
317 class TestParsePushResult:
318 def test_success_response(self) -> None:
319 raw = _mp({"ok": True, "message": "pushed", "branch_heads": {"main": "abc"}})
320 result = _parse_push_result(raw)
321 assert result["ok"] is True
322 assert result["message"] == "pushed"
323 assert result["branch_heads"] == {"main": "abc"}
324
325 def test_failure_response(self) -> None:
326 raw = _mp({"ok": False, "message": "rejected", "branch_heads": {}})
327 result = _parse_push_result(raw)
328 assert result["ok"] is False
329 assert result["message"] == "rejected"
330
331 def test_non_msgpack_raises_transport_error(self) -> None:
332 with pytest.raises(TransportError):
333 _parse_push_result(b"\xff\xff invalid msgpack")
334
335 def test_missing_ok_defaults_false(self) -> None:
336 raw = _mp({"message": "hm", "branch_heads": {}})
337 result = _parse_push_result(raw)
338 assert result["ok"] is False
339
340
341 # ---------------------------------------------------------------------------
342 # HttpTransport — mocked urlopen
343 # ---------------------------------------------------------------------------
344
345
346 def _mock_httpx_client_resp(body: bytes, status: int = 200) -> unittest.mock.MagicMock:
347 """Return a mock httpx client whose .request() returns a response with body/status.
348
349 Supports context-manager usage: ``with _httpx_mod.Client(...) as client:``.
350 """
351 resp = unittest.mock.MagicMock()
352 resp.status_code = status
353 resp.content = body
354 resp.text = body.decode("utf-8", errors="replace")
355 client = unittest.mock.MagicMock()
356 client.is_closed = False
357 client.request = unittest.mock.MagicMock(return_value=resp)
358 client.__enter__ = unittest.mock.MagicMock(return_value=client)
359 client.__exit__ = unittest.mock.MagicMock(return_value=False)
360 return client
361
362
363 class TestHttpTransportFetchRemoteInfo:
364 def test_calls_correct_endpoint(self) -> None:
365 body = _mp({
366 "repo_id": "r1", "domain": "midi",
367 "default_branch": "main", "branch_heads": {"main": "abc"},
368 })
369 client = _mock_httpx_client_resp(body)
370 with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod:
371 mock_mod.Client = unittest.mock.MagicMock(return_value=client)
372 info = HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
373 url_called = client.request.call_args[0][1]
374 assert url_called == "https://hub.example.com/repos/r1/refs"
375 assert info["repo_id"] == "r1"
376
377 def test_msign_header_sent(self) -> None:
378 body = _mp({"repo_id": "r1", "domain": "midi", "default_branch": "main", "branch_heads": {}})
379 client = _mock_httpx_client_resp(body)
380 signing = _make_signing()
381 with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod:
382 mock_mod.Client = unittest.mock.MagicMock(return_value=client)
383 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", signing)
384 headers = client.request.call_args.kwargs.get("headers", {})
385 auth = headers.get("Authorization") or headers.get("authorization")
386 assert auth and auth.startswith("MSign handle=\"testuser\"")
387
388 def test_no_token_no_auth_header(self) -> None:
389 body = _mp({"repo_id": "r1", "domain": "midi", "default_branch": "main", "branch_heads": {}})
390 client = _mock_httpx_client_resp(body)
391 with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod:
392 mock_mod.Client = unittest.mock.MagicMock(return_value=client)
393 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
394 headers = client.request.call_args.kwargs.get("headers", {})
395 auth = headers.get("Authorization") or headers.get("authorization")
396 assert auth is None
397
398 def test_http_401_raises_transport_error(self) -> None:
399 client = _mock_httpx_client_resp(b"Unauthorized", status=401)
400 with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod:
401 mock_mod.Client = unittest.mock.MagicMock(return_value=client)
402 with pytest.raises(TransportError) as exc_info:
403 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
404 assert exc_info.value.status_code == 401
405
406 def test_http_404_raises_transport_error(self) -> None:
407 client = _mock_httpx_client_resp(b"Not Found", status=404)
408 with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod:
409 mock_mod.Client = unittest.mock.MagicMock(return_value=client)
410 with pytest.raises(TransportError) as exc_info:
411 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
412 assert exc_info.value.status_code == 404
413
414 def test_http_500_raises_transport_error(self) -> None:
415 client = _mock_httpx_client_resp(b"Internal Error", status=500)
416 with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod:
417 mock_mod.Client = unittest.mock.MagicMock(return_value=client)
418 with pytest.raises(TransportError) as exc_info:
419 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
420 assert exc_info.value.status_code == 500
421
422 def test_url_error_raises_transport_error_with_code_0(self) -> None:
423 client = unittest.mock.MagicMock()
424 client.is_closed = False
425 client.request = unittest.mock.MagicMock(side_effect=Exception("Name or service not known"))
426 with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod:
427 mock_mod.Client = unittest.mock.MagicMock(return_value=client)
428 with pytest.raises(TransportError) as exc_info:
429 HttpTransport().fetch_remote_info("https://bad.host/r", None)
430 assert exc_info.value.status_code == 0
431
432 def test_trailing_slash_stripped_from_url(self) -> None:
433 body = _mp({"repo_id": "r", "domain": "midi", "default_branch": "main", "branch_heads": {}})
434 client = _mock_httpx_client_resp(body)
435 with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod:
436 mock_mod.Client = unittest.mock.MagicMock(return_value=client)
437 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1/", None)
438 url_called = client.request.call_args[0][1]
439 assert url_called == "https://hub.example.com/repos/r1/refs"
440
441
442 # ---------------------------------------------------------------------------
443 # HttpTransport._build_request — credential security and loopback allowlist
444 # ---------------------------------------------------------------------------
445
446
447 class TestBuildRequest:
448 """_build_request enforces HTTPS for non-loopback URLs with signing identity."""
449
450 def _build(self, url: str, with_signing: bool = True) -> "_Request":
451 signing = _make_signing() if with_signing else None
452 with unittest.mock.patch("muse.core.hub_trust.check_and_pin"):
453 return HttpTransport()._build_request("GET", url, signing)
454
455 # ── Loopback hosts allowed over plain HTTP ────────────────────────────
456
457 def test_localhost_http_with_token_allowed(self) -> None:
458 req = self._build("https://localhost:1337/repo/refs")
459 assert req.headers.get("Authorization", "").startswith("MSign ")
460
461 def test_127_0_0_1_http_with_token_allowed(self) -> None:
462 req = self._build("http://127.0.0.1:10003/repo/refs")
463 assert req.headers.get("Authorization", "").startswith("MSign ")
464
465 def test_ipv6_loopback_http_with_token_allowed(self) -> None:
466 req = self._build("http://[::1]:10003/repo/refs")
467 assert req.headers.get("Authorization", "").startswith("MSign ")
468
469 def test_host_docker_internal_http_with_token_allowed(self) -> None:
470 """host.docker.internal is Docker Desktop's alias for the host loopback.
471
472 Agent swarms run inside Docker and use http://host.docker.internal:10003
473 to reach a local MuseHub instance. Credentials must be sent over this
474 plain-HTTP connection — the traffic never leaves the machine.
475 """
476 req = self._build("http://host.docker.internal:10003/gabriel/repo/refs")
477 assert req.headers.get("Authorization", "").startswith("MSign ")
478
479 def test_https_any_host_with_token_allowed(self) -> None:
480 req = self._build("https://musehub.ai/gabriel/repo/refs")
481 assert req.headers.get("Authorization", "").startswith("MSign ")
482
483 # ── Non-loopback HTTP with token must be rejected ─────────────────────
484
485 def test_non_loopback_http_token_raises_transport_error(self) -> None:
486 with pytest.raises(TransportError, match="non-HTTPS"):
487 self._build("http://musehub.ai/gabriel/repo/refs")
488
489 def test_arbitrary_hostname_http_token_raises(self) -> None:
490 with pytest.raises(TransportError, match="non-HTTPS"):
491 self._build("http://attacker.example.com/steal")
492
493 def test_localhost_lookalike_http_token_raises(self) -> None:
494 """'localhost.attacker.example.com' must NOT be mistaken for the loopback interface."""
495 with pytest.raises(TransportError, match="non-HTTPS"):
496 self._build("http://localhost.attacker.example.com/repo/refs")
497
498 def test_host_docker_internal_lookalike_http_token_raises(self) -> None:
499 """'host.docker.internal.attacker.example.com' must not bypass the check."""
500 with pytest.raises(TransportError, match="non-HTTPS"):
501 self._build("http://host.docker.internal.attacker.example.com/repo")
502
503 # ── No token — scheme restriction does not apply ──────────────────────
504
505 def test_non_loopback_http_without_token_allowed(self) -> None:
506 req = self._build("http://musehub.ai/public/repo/refs", with_signing=False)
507 assert "Authorization" not in req.headers
508
509 # ── Request structure ─────────────────────────────────────────────────
510
511 def test_accept_header_always_set(self) -> None:
512 req = self._build("https://musehub.ai/repo/refs")
513 assert "msgpack" in req.headers.get("Accept", "")
514
515 def test_method_preserved(self) -> None:
516 req = HttpTransport()._build_request("POST", "https://musehub.ai/x", None)
517 assert req.method == "POST"
518
519 def test_body_sets_content_type(self) -> None:
520 req = HttpTransport()._build_request(
521 "POST", "https://musehub.ai/x", None, body_bytes=b"data"
522 )
523 assert req.headers.get("Content-Type") == "application/x-msgpack"
524
525 def test_no_body_omits_content_type(self) -> None:
526 req = self._build("https://musehub.ai/x", with_signing=False)
527 assert "Content-Type" not in req.headers
528
529
530 # ---------------------------------------------------------------------------
531 # SIGPIPE regression — large push body with early-close server
532 # ---------------------------------------------------------------------------
533
534
535 def _early_close_server(
536 server_sock: socket.socket,
537 response_code: int,
538 resp_body: bytes,
539 ) -> None:
540 """Accept one connection, read HTTP headers, send response, close immediately.
541
542 Simulates the scenario where the server sends a 4xx/5xx response while
543 the client is still uploading a large request body. Without the
544 ``_ignore_sigpipe`` guard, the client process dies with exit code 141
545 (SIGPIPE) instead of raising ``TransportError``.
546 """
547 try:
548 conn, _ = server_sock.accept()
549 conn.settimeout(5.0)
550 try:
551 buf = b""
552 deadline = time.time() + 5
553 while time.time() < deadline:
554 try:
555 chunk = conn.recv(4096)
556 if not chunk:
557 break
558 buf += chunk
559 if b"\r\n\r\n" in buf:
560 break
561 except socket.timeout:
562 break
563 status_line = f"HTTP/1.1 {response_code} Error\r\n"
564 resp_headers = (
565 "Content-Type: application/json\r\n"
566 f"Content-Length: {len(resp_body)}\r\n"
567 "Connection: close\r\n\r\n"
568 )
569 conn.sendall((status_line + resp_headers).encode() + resp_body)
570 finally:
571 conn.close()
572 except Exception:
573 pass
574 finally:
575 server_sock.close()
576
577
578 class TestSigpipeRegression:
579 """Regression tests for SIGPIPE on large push bodies.
580
581 ``muse/cli/app.py`` sets ``SIGPIPE = SIG_DFL`` so that piping output to
582 ``head``/``grep``/``jq`` exits cleanly. Without the ``_ignore_sigpipe``
583 guard the push command dies with exit 141 when the server closes the
584 connection while the client is still uploading a large body.
585 """
586
587 def _run_early_close_scenario(self, payload_mb: float, response_code: int) -> None:
588 """Assert that TransportError is raised, not a process-killing SIGPIPE."""
589 # Reproduce app.py's startup action — SIG_DFL kills the process on SIGPIPE.
590 if hasattr(signal, "SIGPIPE"):
591 original = signal.signal(signal.SIGPIPE, signal.SIG_DFL)
592 else:
593 original = None
594
595 body = b"X" * int(payload_mb * 1024 * 1024)
596 resp_body = b'{"detail":"test error"}'
597
598 server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
599 server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
600 server_sock.bind(("127.0.0.1", 0))
601 server_sock.listen(1)
602 port = server_sock.getsockname()[1]
603
604 t = threading.Thread(
605 target=_early_close_server,
606 args=(server_sock, response_code, resp_body),
607 daemon=True,
608 )
609 t.start()
610
611 try:
612 url = f"http://127.0.0.1:{port}/owner/repo/push"
613 transport = HttpTransport()
614 req = transport._build_request("POST", url, None, body, "application/x-msgpack")
615 with pytest.raises(TransportError):
616 transport._execute(req)
617 finally:
618 t.join(timeout=2)
619 if original is not None and hasattr(signal, "SIGPIPE"):
620 signal.signal(signal.SIGPIPE, original)
621
622 def test_sigpipe_not_fatal_409_large_body(self) -> None:
623 """20 MB body + server closes after headers → TransportError, not exit 141."""
624 self._run_early_close_scenario(payload_mb=20.0, response_code=409)
625
626 def test_sigpipe_not_fatal_401_large_body(self) -> None:
627 """15 MB body + server sends 401 early → TransportError, not SIGPIPE crash."""
628 self._run_early_close_scenario(payload_mb=15.0, response_code=401)
629
630 def test_sigpipe_not_fatal_500_large_body(self) -> None:
631 """10 MB body + server crashes (500) → TransportError, not SIGPIPE crash."""
632 self._run_early_close_scenario(payload_mb=10.0, response_code=500)
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago