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