"""Tests for muse.core.transport — HttpTransport and response parsers."""

from __future__ import annotations

import json
import signal
import socket
import threading
import time
import unittest.mock
import urllib.error
import urllib.request
from io import BytesIO

import msgpack
import pytest

from muse.core._types import MsgpackDict
from muse.core.pack import PackBundle, RemoteInfo
from muse.core.transport import (
    HttpTransport,
    SigningIdentity,
    TransportError,
    _parse_bundle,
    _parse_push_result,
    _parse_remote_info,
    build_msign_header,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_signing() -> "SigningIdentity":
    """Generate a fresh Ed25519 SigningIdentity for tests."""
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    from muse.core.transport import SigningIdentity
    return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())


def _mock_response(
    body: bytes,
    status: int = 200,
    content_type: str = "application/x-msgpack",
    streaming: bool = False,
) -> unittest.mock.MagicMock:
    """Return a mock urllib response context manager.

    When *streaming* is True, ``read()`` returns *body* on the first call and
    ``b""`` on all subsequent calls — matching the chunked-read loop used by
    :meth:`~muse.core.transport.HttpTransport._execute_fetch`.
    """
    resp = unittest.mock.MagicMock()
    if streaming:
        resp.read.side_effect = [body, b""]
    else:
        resp.read.return_value = body
    resp.headers = {"Content-Type": content_type}
    resp.__enter__ = lambda s: s
    resp.__exit__ = unittest.mock.MagicMock(return_value=False)
    return resp


def _mp(data: MsgpackDict) -> bytes:
    """Encode data as msgpack."""
    return msgpack.packb(data, use_bin_type=True)


def _http_error(code: int, body: bytes = b"") -> urllib.error.HTTPError:
    return urllib.error.HTTPError(
        url="https://example.com",
        code=code,
        msg=str(code),
        hdrs=None,
        fp=BytesIO(body),
    )


# ---------------------------------------------------------------------------
# _parse_remote_info
# ---------------------------------------------------------------------------


class TestParseRemoteInfo:
    def test_valid_response(self) -> None:
        raw = _mp(
            {
                "repo_id": "r123",
                "domain": "midi",
                "default_branch": "main",
                "branch_heads": {"main": "abc123", "dev": "def456"},
            }
        )
        info = _parse_remote_info(raw)
        assert info["repo_id"] == "r123"
        assert info["domain"] == "midi"
        assert info["default_branch"] == "main"
        assert info["branch_heads"] == {"main": "abc123", "dev": "def456"}

    def test_invalid_msgpack_raises_transport_error(self) -> None:
        with pytest.raises(TransportError):
            _parse_remote_info(b"\xff\xff\xff\xff\xff invalid")

    def test_non_dict_response_returns_defaults(self) -> None:
        raw = _mp([1, 2, 3])
        info = _parse_remote_info(raw)
        assert info["repo_id"] == ""
        assert info["branch_heads"] == {}

    def test_missing_fields_get_defaults(self) -> None:
        raw = _mp({"repo_id": "x"})
        info = _parse_remote_info(raw)
        assert info["repo_id"] == "x"
        assert info["domain"] == "midi"
        assert info["default_branch"] == "main"
        assert info["branch_heads"] == {}

    def test_non_string_branch_heads_excluded(self) -> None:
        raw = _mp({"branch_heads": {"main": "abc", "bad": 123}})
        info = _parse_remote_info(raw)
        assert "main" in info["branch_heads"]
        assert "bad" not in info["branch_heads"]

    def test_pack_origin_populated_when_present(self) -> None:
        raw = _mp(
            {
                "repo_id": "r1",
                "domain": "code",
                "default_branch": "main",
                "branch_heads": {},
                "pack_origin": "https://worker.example.workers.dev",
            }
        )
        info = _parse_remote_info(raw)
        assert info.get("pack_origin") == "https://worker.example.workers.dev"

    def test_pack_origin_absent_when_not_in_response(self) -> None:
        raw = _mp({"repo_id": "r1", "branch_heads": {}})
        info = _parse_remote_info(raw)
        assert "pack_origin" not in info

    def test_pack_origin_whitespace_stripped(self) -> None:
        raw = _mp(
            {"repo_id": "r1", "branch_heads": {}, "pack_origin": "  https://w.example.dev  "}
        )
        info = _parse_remote_info(raw)
        assert info.get("pack_origin") == "https://w.example.dev"

    def test_pack_origin_none_value_not_set(self) -> None:
        raw = _mp({"repo_id": "r1", "branch_heads": {}, "pack_origin": None})
        info = _parse_remote_info(raw)
        assert "pack_origin" not in info

    def test_pack_origin_empty_string_not_set(self) -> None:
        raw = _mp({"repo_id": "r1", "branch_heads": {}, "pack_origin": ""})
        info = _parse_remote_info(raw)
        assert "pack_origin" not in info

    def test_pack_origin_whitespace_only_not_set(self) -> None:
        raw = _mp({"repo_id": "r1", "branch_heads": {}, "pack_origin": "   "})
        info = _parse_remote_info(raw)
        assert "pack_origin" not in info


# ---------------------------------------------------------------------------
# build_msign_header — module-level signing utility
# ---------------------------------------------------------------------------


class TestBuildMsignHeader:
    def _make_signing(self) -> SigningIdentity:
        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

        return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())

    def test_header_format(self) -> None:
        header = build_msign_header(self._make_signing(), "GET", "https://example.com/path", None)
        assert header.startswith('MSign handle="testuser"')
        assert " ts=" in header
        assert " sig=" in header

    def test_timestamp_is_recent(self) -> None:
        import time

        before = int(time.time())
        header = build_msign_header(self._make_signing(), "GET", "https://example.com/p", None)
        after = int(time.time())
        ts_part = next(p for p in header.split() if p.startswith("ts="))
        ts = int(ts_part[3:])
        assert before <= ts <= after + 1

    def test_signature_is_verifiable(self) -> None:
        """The sig= value must verify against the Ed25519 public key for the canonical input."""
        import base64
        import hashlib

        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

        private_key = Ed25519PrivateKey.generate()
        signing = SigningIdentity(handle="testuser", private_key=private_key)
        method = "POST"
        url = "https://hub.example.com/owner/repo/push"
        body = b"some body data"

        header = build_msign_header(signing, method, url, body)

        parts: dict[str, str] = {}
        for part in header[len("MSign "):].split():
            k, _, v = part.partition("=")
            parts[k] = v.strip('"')

        ts = int(parts["ts"])
        sig_bytes = base64.urlsafe_b64decode(parts["sig"] + "==")

        body_hash = hashlib.sha256(body).hexdigest()
        canonical = f"{method}\n/owner/repo/push\n{ts}\n{body_hash}".encode()

        # raises cryptography.exceptions.InvalidSignature on failure
        private_key.public_key().verify(sig_bytes, canonical)

    def test_query_string_included_in_canonical(self) -> None:
        """Query parameters must be part of the signed path, not dropped."""
        import base64
        import hashlib

        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

        private_key = Ed25519PrivateKey.generate()
        signing = SigningIdentity(handle="u", private_key=private_key)
        url = "https://hub.example.com/path?foo=bar&baz=1"

        header = build_msign_header(signing, "GET", url, None)

        parts: dict[str, str] = {}
        for part in header[len("MSign "):].split():
            k, _, v = part.partition("=")
            parts[k] = v.strip('"')

        ts = int(parts["ts"])
        sig_bytes = base64.urlsafe_b64decode(parts["sig"] + "==")
        body_hash = hashlib.sha256(b"").hexdigest()
        canonical = f"GET\n/path?foo=bar&baz=1\n{ts}\n{body_hash}".encode()

        private_key.public_key().verify(sig_bytes, canonical)

    def test_none_body_treated_as_empty_bytes(self) -> None:
        """None and b'' must produce the same SHA-256 body hash in the canonical form."""
        import base64
        import hashlib

        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

        private_key = Ed25519PrivateKey.generate()
        signing = SigningIdentity(handle="u", private_key=private_key)
        url = "https://hub.example.com/path"

        header = build_msign_header(signing, "GET", url, None)

        parts: dict[str, str] = {}
        for part in header[len("MSign "):].split():
            k, _, v = part.partition("=")
            parts[k] = v.strip('"')

        ts = int(parts["ts"])
        sig_bytes = base64.urlsafe_b64decode(parts["sig"] + "==")
        # body=None → b"" → sha256("") is the canonical body hash
        body_hash = hashlib.sha256(b"").hexdigest()
        canonical = f"GET\n/path\n{ts}\n{body_hash}".encode()

        private_key.public_key().verify(sig_bytes, canonical)

    def test_different_methods_produce_different_headers(self) -> None:
        """Two calls with different methods on the same URL must differ (method is in canonical)."""

        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

        private_key = Ed25519PrivateKey.generate()
        signing = SigningIdentity(handle="u", private_key=private_key)

        with unittest.mock.patch("muse.core.transport.time") as mt:
            mt.time.return_value = 1_700_000_000
            h_get = build_msign_header(signing, "GET", "https://example.com/x", b"")
            h_post = build_msign_header(signing, "POST", "https://example.com/x", b"")

        assert h_get != h_post

    def test_different_bodies_produce_different_sigs(self) -> None:
        """Body content must influence the signature (body hash is in canonical)."""

        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

        private_key = Ed25519PrivateKey.generate()
        signing = SigningIdentity(handle="u", private_key=private_key)

        with unittest.mock.patch("muse.core.transport.time") as mt:
            mt.time.return_value = 1_700_000_000
            h1 = build_msign_header(signing, "POST", "https://example.com/x", b"body-a")
            h2 = build_msign_header(signing, "POST", "https://example.com/x", b"body-b")

        assert h1 != h2

    def test_handle_embedded_in_header(self) -> None:
        """The MSign header must carry the signing identity's handle."""

        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

        private_key = Ed25519PrivateKey.generate()
        signing = SigningIdentity(handle="my-agent-42", private_key=private_key)
        header = build_msign_header(signing, "GET", "https://example.com/x", None)
        assert 'handle="my-agent-42"' in header


# ---------------------------------------------------------------------------
# _parse_bundle
# ---------------------------------------------------------------------------


class TestParseBundle:
    def test_empty_msgpack_object_returns_empty_bundle(self) -> None:
        bundle = _parse_bundle(_mp({}))
        assert bundle == {}

    def test_non_dict_returns_empty_bundle(self) -> None:
        bundle = _parse_bundle(_mp([]))
        assert bundle == {}

    def test_commits_extracted(self) -> None:
        raw = _mp(
            {
                "commits": [
                    {
                        "commit_id": "c1",
                        "repo_id": "r1",
                        "branch": "main",
                        "snapshot_id": "1" * 64,
                        "message": "test",
                        "committed_at": "2026-01-01T00:00:00+00:00",
                        "parent_commit_id": None,
                        "parent2_commit_id": None,
                        "author": "bob",
                        "metadata": {},
                    }
                ]
            }
        )
        bundle = _parse_bundle(raw)
        commits = bundle.get("commits") or []
        assert len(commits) == 1
        assert commits[0]["commit_id"] == "c1"

    def test_objects_extracted(self) -> None:
        raw = _mp(
            {
                "objects": [
                    {
                        "object_id": "abc123",
                        "content": b"hello",
                    }
                ]
            }
        )
        bundle = _parse_bundle(raw)
        objs = bundle.get("objects") or []
        assert len(objs) == 1
        assert objs[0]["object_id"] == "abc123"
        assert objs[0]["content"] == b"hello"

    def test_object_missing_content_excluded(self) -> None:
        raw = _mp({"objects": [{"object_id": "abc"}]})
        bundle = _parse_bundle(raw)
        assert (bundle.get("objects") or []) == []

    def test_branch_heads_extracted(self) -> None:
        raw = _mp({"branch_heads": {"main": "abc123"}})
        bundle = _parse_bundle(raw)
        assert bundle.get("branch_heads") == {"main": "abc123"}


# ---------------------------------------------------------------------------
# _parse_push_result
# ---------------------------------------------------------------------------


class TestParsePushResult:
    def test_success_response(self) -> None:
        raw = _mp({"ok": True, "message": "pushed", "branch_heads": {"main": "abc"}})
        result = _parse_push_result(raw)
        assert result["ok"] is True
        assert result["message"] == "pushed"
        assert result["branch_heads"] == {"main": "abc"}

    def test_failure_response(self) -> None:
        raw = _mp({"ok": False, "message": "rejected", "branch_heads": {}})
        result = _parse_push_result(raw)
        assert result["ok"] is False
        assert result["message"] == "rejected"

    def test_non_msgpack_raises_transport_error(self) -> None:
        with pytest.raises(TransportError):
            _parse_push_result(b"\xff\xff invalid msgpack")

    def test_missing_ok_defaults_false(self) -> None:
        raw = _mp({"message": "hm", "branch_heads": {}})
        result = _parse_push_result(raw)
        assert result["ok"] is False


# ---------------------------------------------------------------------------
# HttpTransport — mocked urlopen
# ---------------------------------------------------------------------------


class TestHttpTransportFetchRemoteInfo:
    def test_calls_correct_endpoint(self) -> None:
        body = _mp(
            {
                "repo_id": "r1",
                "domain": "midi",
                "default_branch": "main",
                "branch_heads": {"main": "abc"},
            }
        )
        mock_resp = _mock_response(body)
        with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
            transport = HttpTransport()
            info = transport.fetch_remote_info("https://hub.example.com/repos/r1", None)
        req = m.call_args[0][0]
        assert req.full_url == "https://hub.example.com/repos/r1/refs"
        assert info["repo_id"] == "r1"

    def test_msign_header_sent(self) -> None:
        body = _mp(
            {"repo_id": "r1", "domain": "midi", "default_branch": "main", "branch_heads": {}}
        )
        mock_resp = _mock_response(body)
        signing = _make_signing()
        with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
            HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", signing)
        req = m.call_args[0][0]
        assert req.get_header("Authorization").startswith("MSign handle=\"testuser\"")

    def test_no_token_no_auth_header(self) -> None:
        body = _mp(
            {"repo_id": "r1", "domain": "midi", "default_branch": "main", "branch_heads": {}}
        )
        mock_resp = _mock_response(body)
        with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
            HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
        req = m.call_args[0][0]
        assert req.get_header("Authorization") is None

    def test_http_401_raises_transport_error(self) -> None:
        with unittest.mock.patch(
            "muse.core.transport._open_url", side_effect=_http_error(401, b"Unauthorized")
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
        assert exc_info.value.status_code == 401

    def test_http_404_raises_transport_error(self) -> None:
        with unittest.mock.patch(
            "muse.core.transport._open_url", side_effect=_http_error(404)
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
        assert exc_info.value.status_code == 404

    def test_http_500_raises_transport_error(self) -> None:
        with unittest.mock.patch(
            "muse.core.transport._open_url", side_effect=_http_error(500, b"Internal Error")
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
        assert exc_info.value.status_code == 500

    def test_url_error_raises_transport_error_with_code_0(self) -> None:
        with unittest.mock.patch(
            "muse.core.transport._open_url",
            side_effect=urllib.error.URLError("Name or service not known"),
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_remote_info("https://bad.host/r", None)
        assert exc_info.value.status_code == 0

    def test_trailing_slash_stripped_from_url(self) -> None:
        body = _mp(
            {"repo_id": "r", "domain": "midi", "default_branch": "main", "branch_heads": {}}
        )
        mock_resp = _mock_response(body)
        with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
            HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1/", None)
        req = m.call_args[0][0]
        assert req.full_url == "https://hub.example.com/repos/r1/refs"


class TestHttpTransportFetchPack:
    def test_posts_to_fetch_endpoint(self) -> None:
        bundle_body = _mp(
            {
                "commits": [],
                "snapshots": [],
                "objects": [],
                "branch_heads": {"main": "abc"},
            }
        )
        # fetch_pack uses _execute_fetch which reads in a chunked loop until
        # read() returns b"" — use streaming=True so the mock terminates.
        mock_resp = _mock_response(bundle_body, streaming=True)
        with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
            transport = HttpTransport()
            bundle = transport.fetch_pack(
                "https://hub.example.com/repos/r1",
                _make_signing(),
                want=["abc"],
                have=["def"],
            )
        req = m.call_args[0][0]
        assert req.full_url == "https://hub.example.com/repos/r1/fetch"
        sent = msgpack.unpackb(req.data, raw=False)
        assert sent["want"] == ["abc"]
        assert sent["have"] == ["def"]
        assert bundle.get("branch_heads") == {"main": "abc"}

    def test_http_409_raises_transport_error(self) -> None:
        with unittest.mock.patch(
            "muse.core.transport._open_url", side_effect=_http_error(409)
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_pack("https://hub.example.com/r", None, [], [])
        assert exc_info.value.status_code == 409


class TestHttpTransportPushPack:
    def test_posts_to_push_endpoint(self) -> None:
        push_body = _mp({"ok": True, "message": "ok", "branch_heads": {"main": "new"}})
        mock_resp = _mock_response(push_body)
        bundle: PackBundle = {"commits": [], "snapshots": [], "objects": []}
        with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
            result = HttpTransport().push_pack(
                "https://hub.example.com/repos/r1", _make_signing(), bundle, "main", False
            )
        req = m.call_args[0][0]
        assert req.full_url == "https://hub.example.com/repos/r1/push"
        sent = msgpack.unpackb(req.data, raw=False)
        assert sent["branch"] == "main"
        assert sent["force"] is False
        assert result["ok"] is True

    def test_force_flag_sent(self) -> None:
        push_body = _mp({"ok": True, "message": "", "branch_heads": {}})
        mock_resp = _mock_response(push_body)
        bundle: PackBundle = {}
        with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
            HttpTransport().push_pack("https://hub.example.com/r", None, bundle, "main", True)
        req = m.call_args[0][0]
        sent = msgpack.unpackb(req.data, raw=False)
        assert sent["force"] is True

    def test_push_rejected_raises_transport_error(self) -> None:
        with unittest.mock.patch(
            "muse.core.transport._open_url", side_effect=_http_error(409, b"non-fast-forward")
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().push_pack(
                    "https://hub.example.com/r", None, {}, "main", False
                )
        assert exc_info.value.status_code == 409


# ---------------------------------------------------------------------------
# HttpTransport._build_request — credential security and loopback allowlist
# ---------------------------------------------------------------------------


class TestBuildRequest:
    """_build_request enforces HTTPS for non-loopback URLs with signing identity."""

    def _build(self, url: str, with_signing: bool = True) -> urllib.request.Request:
        signing = _make_signing() if with_signing else None
        # Stub TOFU so unit tests don't make real TLS connections to verify
        # certificate fingerprints — that's network state, not request structure.
        with unittest.mock.patch("muse.core.hub_trust.check_and_pin"):
            return HttpTransport()._build_request("GET", url, signing)

    # ── Loopback hosts allowed over plain HTTP ────────────────────────────

    def test_localhost_http_with_token_allowed(self) -> None:
        req = self._build("http://localhost:10003/repo/refs")
        assert req.get_header("Authorization").startswith("MSign ")

    def test_127_0_0_1_http_with_token_allowed(self) -> None:
        req = self._build("http://127.0.0.1:10003/repo/refs")
        assert req.get_header("Authorization").startswith("MSign ")

    def test_ipv6_loopback_http_with_token_allowed(self) -> None:
        req = self._build("http://[::1]:10003/repo/refs")
        assert req.get_header("Authorization").startswith("MSign ")

    def test_host_docker_internal_http_with_token_allowed(self) -> None:
        """host.docker.internal is Docker Desktop's alias for the host loopback.

        Agent swarms run inside Docker and use http://host.docker.internal:10003
        to reach a local MuseHub instance.  Credentials must be sent over this
        plain-HTTP connection — the traffic never leaves the machine.
        """
        req = self._build("http://host.docker.internal:10003/gabriel/repo/refs")
        assert req.get_header("Authorization").startswith("MSign ")

    def test_https_any_host_with_token_allowed(self) -> None:
        req = self._build("https://musehub.ai/gabriel/repo/refs")
        assert req.get_header("Authorization").startswith("MSign ")

    # ── Non-loopback HTTP with token must be rejected ─────────────────────

    def test_non_loopback_http_token_raises_transport_error(self) -> None:
        with pytest.raises(TransportError, match="non-HTTPS"):
            self._build("http://musehub.ai/gabriel/repo/refs")

    def test_arbitrary_hostname_http_token_raises(self) -> None:
        with pytest.raises(TransportError, match="non-HTTPS"):
            self._build("http://evil.example.com/steal")

    def test_localhost_lookalike_http_token_raises(self) -> None:
        """'localhost.evil.com' must NOT be mistaken for the loopback interface."""
        with pytest.raises(TransportError, match="non-HTTPS"):
            self._build("http://localhost.evil.com/repo/refs")

    def test_host_docker_internal_lookalike_http_token_raises(self) -> None:
        """'host.docker.internal.evil.com' must not bypass the check."""
        with pytest.raises(TransportError, match="non-HTTPS"):
            self._build("http://host.docker.internal.evil.com/repo")

    # ── No token — scheme restriction does not apply ──────────────────────

    def test_non_loopback_http_without_token_allowed(self) -> None:
        req = self._build("http://musehub.ai/public/repo/refs", with_signing=False)
        assert req.get_header("Authorization") is None

    # ── Request structure ─────────────────────────────────────────────────

    def test_accept_header_always_set(self) -> None:
        req = self._build("https://musehub.ai/repo/refs")
        assert "msgpack" in req.get_header("Accept")

    def test_method_preserved(self) -> None:
        req = HttpTransport()._build_request("POST", "https://musehub.ai/x", None)
        assert req.get_method() == "POST"

    def test_body_sets_content_type(self) -> None:
        req = HttpTransport()._build_request(
            "POST", "https://musehub.ai/x", None, body_bytes=b"data"
        )
        assert req.get_header("Content-type") == "application/x-msgpack"

    def test_no_body_omits_content_type(self) -> None:
        req = self._build("https://musehub.ai/x", with_signing=False)
        assert req.get_header("Content-type") is None


# ---------------------------------------------------------------------------
# SIGPIPE regression — large push body with early-close server
# ---------------------------------------------------------------------------


def _early_close_server(
    server_sock: socket.socket,
    response_code: int,
    resp_body: bytes,
) -> None:
    """Accept one connection, read HTTP headers, send response, close immediately.

    Simulates the scenario where the server sends a 4xx/5xx response while
    the client is still uploading a large request body.  Without the
    ``_ignore_sigpipe`` guard in ``_open_url``, the client process dies with
    exit code 141 (SIGPIPE) instead of raising ``TransportError``.
    """
    try:
        conn, _ = server_sock.accept()
        conn.settimeout(5.0)
        try:
            buf = b""
            deadline = time.time() + 5
            while time.time() < deadline:
                try:
                    chunk = conn.recv(4096)
                    if not chunk:
                        break
                    buf += chunk
                    if b"\r\n\r\n" in buf:
                        break
                except socket.timeout:
                    break
            status_line = f"HTTP/1.1 {response_code} Error\r\n"
            resp_headers = (
                "Content-Type: application/json\r\n"
                f"Content-Length: {len(resp_body)}\r\n"
                "Connection: close\r\n\r\n"
            )
            conn.sendall((status_line + resp_headers).encode() + resp_body)
        finally:
            conn.close()
    except Exception:
        pass
    finally:
        server_sock.close()


class TestSigpipeRegression:
    """Regression tests for SIGPIPE on large push bodies.

    ``muse/cli/app.py`` sets ``SIGPIPE = SIG_DFL`` so that piping output to
    ``head``/``grep``/``jq`` exits cleanly.  Without the ``_ignore_sigpipe``
    guard in ``_open_url`` the push command dies with exit 141 when the server
    closes the connection while the client is still uploading a large body.
    """

    def _run_early_close_scenario(self, payload_mb: float, response_code: int) -> None:
        """Assert that TransportError is raised, not a process-killing SIGPIPE."""
        # Reproduce app.py's startup action — SIG_DFL kills the process on SIGPIPE.
        if hasattr(signal, "SIGPIPE"):
            original = signal.signal(signal.SIGPIPE, signal.SIG_DFL)
        else:
            original = None

        body = b"X" * int(payload_mb * 1024 * 1024)
        resp_body = b'{"detail":"test error"}'

        server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_sock.bind(("127.0.0.1", 0))
        server_sock.listen(1)
        port = server_sock.getsockname()[1]

        t = threading.Thread(
            target=_early_close_server,
            args=(server_sock, response_code, resp_body),
            daemon=True,
        )
        t.start()

        try:
            url = f"http://127.0.0.1:{port}/owner/repo/push"
            transport = HttpTransport()
            req = transport._build_request("POST", url, None, body, "application/x-msgpack")
            with pytest.raises(TransportError):
                transport._execute(req)
        finally:
            t.join(timeout=2)
            if original is not None and hasattr(signal, "SIGPIPE"):
                signal.signal(signal.SIGPIPE, original)

    def test_sigpipe_not_fatal_409_large_body(self) -> None:
        """20 MB body + server closes after headers → TransportError, not exit 141."""
        self._run_early_close_scenario(payload_mb=20.0, response_code=409)

    def test_sigpipe_not_fatal_401_large_body(self) -> None:
        """15 MB body + server sends 401 early → TransportError, not SIGPIPE crash."""
        self._run_early_close_scenario(payload_mb=15.0, response_code=401)

    def test_sigpipe_not_fatal_500_large_body(self) -> None:
        """10 MB body + server crashes (500) → TransportError, not SIGPIPE crash."""
        self._run_early_close_scenario(payload_mb=10.0, response_code=500)