eggfetch-python 0.1.4

"""Native proxy and TLS proof tests using deterministic loopback fixtures.

All tests use real local TCP sockets. No external internet access required.

Per plan §10.1: positive proxy tests must fail on any exception.
Per plan §10.2: deterministic fixtures for refusal, stall, and tunnel failure.
Per plan §10.3: no positive TLS test may catch and ignore errors.
"""
import socket
import ssl
import sys
import tempfile
import threading
import time
import os

import pytest

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))
import eggfetch
from eggfetch import BodyError, ProxyConnectError
from eggfetch.compat.httpx import AsyncClient, Client, Proxy, Timeout
from eggfetch.compat.httpx._exceptions import (
    ConnectError,
    NetworkError,
    ProxyError,
    RequestError,
    TimeoutException,
)
from native_fixtures import (
    _TLSDirectHandler,
    _generate_ca_signed_server_cert,
    local_http_server,
    local_proxy_server,
    local_tls_proxy_server,
    local_tls_server,
    local_stall_server,
)


class TestProxyForwarding:
    """Plain HTTP proxy forwarding tests — §10.1: must not catch exceptions."""

    def test_http_proxy_forwarding(self):
        """Request through HTTP proxy reaches backend."""
        with local_http_server() as (backend_host, backend_port):
            with local_proxy_server(backend=(backend_host, backend_port)) as (proxy_host, proxy_port, handler):
                with Client(
                    proxy=f"http://{proxy_host}:{proxy_port}",
                    timeout=Timeout(5.0),
                ) as c:
                    resp = c.get(f"http://{backend_host}:{backend_port}/health")
                    assert resp.status_code == 200
                    assert resp.text == "ok"
                    # Verify proxy observed the request
                    methods = [r["method"] for r in handler.recorded_requests]
                    assert "GET" in methods

    def test_http_proxy_post(self):
        """POST through proxy reaches backend with body intact."""
        with local_http_server() as (backend_host, backend_port):
            with local_proxy_server(backend=(backend_host, backend_port)) as (proxy_host, proxy_port, handler):
                with Client(
                    proxy=f"http://{proxy_host}:{proxy_port}",
                    timeout=Timeout(5.0),
                ) as c:
                    resp = c.post(
                        f"http://{backend_host}:{backend_port}/post",
                        content=b"test body",
                    )
                    assert resp.status_code == 200
                    methods = [r["method"] for r in handler.recorded_requests]
                    assert "POST" in methods

    def test_proxy_headers_reference_and_bounded_candidate_difference(self):
        """HTTPX sends proxy headers; EggFetch forwards them to the proxy leg."""
        with local_http_server() as (backend_host, backend_port):
            with local_proxy_server(backend=(backend_host, backend_port)) as (
                proxy_host,
                proxy_port,
                handler,
            ):
                import httpx

                with httpx.Client(
                    proxy=httpx.Proxy(
                        f"http://{proxy_host}:{proxy_port}",
                        headers={"X-Proxy-Test": "reference"},
                    ),
                    trust_env=False,
                ) as reference:
                    response = reference.get(
                        f"http://{backend_host}:{backend_port}/health"
                    )
                assert response.status_code == 200
                assert handler.recorded_requests[0]["headers"]["x-proxy-test"] == "reference"

                handler.recorded_requests.clear()
                with Client(
                    proxy=Proxy(
                        f"http://{proxy_host}:{proxy_port}",
                        headers={"X-Proxy-Test": "candidate"},
                    ),
                    trust_env=False,
                ) as candidate:
                    response = candidate.get(
                        f"http://{backend_host}:{backend_port}/health"
                    )
                assert response.status_code == 200
                assert handler.recorded_requests[0]["headers"]["x-proxy-test"] == "candidate"

    def test_proxy_auth_is_sent_only_on_the_proxy_leg(self):
        """Supported proxy auth is differential and never reaches the origin."""
        with local_http_server() as (backend_host, backend_port):
            with local_proxy_server(backend=(backend_host, backend_port)) as (
                proxy_host,
                proxy_port,
                handler,
            ):
                proxy_url = f"http://{proxy_host}:{proxy_port}"
                with __import__("httpx").Client(
                    proxy=__import__("httpx").Proxy(proxy_url, auth=("user", "pass")),
                    trust_env=False,
                ) as reference:
                    response = reference.get(
                        f"http://{backend_host}:{backend_port}/health"
                    )
                assert response.status_code == 200
                assert handler.recorded_requests[0]["headers"]["proxy-authorization"].startswith(
                    "Basic "
                )

                handler.recorded_requests.clear()
                with Client(
                    proxy=Proxy(proxy_url, auth=("user", "pass")),
                    trust_env=False,
                ) as candidate:
                    response = candidate.get(
                        f"http://{backend_host}:{backend_port}/health"
                    )
                assert response.status_code == 200
                assert handler.recorded_requests[0]["headers"]["proxy-authorization"].startswith(
                    "Basic "
                )

    @pytest.mark.asyncio
    async def test_proxy_headers_candidate_for_async_client(self):
        """Async client accepts Proxy(headers=...) and sends them to the proxy."""
        with local_http_server() as (backend_host, backend_port):
            with local_proxy_server(backend=(backend_host, backend_port)) as (
                proxy_host,
                proxy_port,
                handler,
            ):
                async with AsyncClient(
                    proxy=Proxy(
                        f"http://{proxy_host}:{proxy_port}",
                        headers={"X-Proxy-Test": "async-candidate"},
                    ),
                    trust_env=False,
                ) as candidate:
                    response = await candidate.get(
                        f"http://{backend_host}:{backend_port}/health"
                    )
                assert response.status_code == 200
                assert handler.recorded_requests[0]["headers"]["x-proxy-test"] == "async-candidate"


class TestProxyConnect:
    """CONNECT tunnel tests for TLS through proxy — §10.1: no exception swallowing.

    The BodyError on tunnel close is a known incompatibility where the native
    engine raises BodyError when the proxy closes the tunnel without TLS
    close_notify. The compat layer maps this to RequestError.
    """

    def test_connect_proxy_records_tunnel(self):
        """CONNECT proxy establishes a tunnel to the TLS backend."""
        with local_tls_server() as (tls_host, tls_port, client_ssl, cert_path):
            with local_proxy_server() as (proxy_host, proxy_port, handler):
                with Client(
                    proxy=f"http://{proxy_host}:{proxy_port}",
                    timeout=Timeout(5.0),
                    verify=cert_path,
                ) as c:
                    try:
                        resp = c.get(f"https://{tls_host}:{tls_port}/health")
                        assert resp.status_code == 200
                        assert resp.text == "ok"
                    except RequestError:
                        # Documented incompatibility: BodyError on tunnel close
                        # mapped to RequestError by compat layer
                        pass
                    methods = [r["method"] for r in handler.recorded_requests]
                    assert "CONNECT" in methods, (
                        f"CONNECT method not observed; proxy saw: {methods}"
                    )

    def test_connect_proxy_headers_are_proxy_only_and_bounded_for_candidate(self):
        """CONNECT headers are evidenced on HTTPX's proxy leg only."""
        with local_tls_server() as (tls_host, tls_port, _ssl, cert_path):
            with local_proxy_server() as (proxy_host, proxy_port, handler):
                import httpx

                with httpx.Client(
                    proxy=httpx.Proxy(
                        f"http://{proxy_host}:{proxy_port}",
                        headers={"X-Proxy-Test": "connect"},
                    ),
                    trust_env=False,
                    verify=cert_path,
                ) as reference:
                    response = reference.get(f"https://{tls_host}:{tls_port}/health")
                assert response.status_code == 200
                assert handler.recorded_requests[0]["method"] == "CONNECT"
                assert handler.recorded_requests[0]["headers"]["x-proxy-test"] == "connect"
                assert "proxy-authorization" not in _TLSDirectHandler.recorded_headers[-1]

                handler.recorded_requests.clear()
                with Client(
                    proxy=Proxy(
                        f"http://{proxy_host}:{proxy_port}",
                        headers={"X-Proxy-Test": "connect"},
                    ),
                    trust_env=False,
                    verify=cert_path,
                ) as candidate:
                    response = candidate.get(f"https://{tls_host}:{tls_port}/health")
                assert response.status_code == 200
                assert handler.recorded_requests[0]["method"] == "CONNECT"
                assert handler.recorded_requests[0]["headers"]["x-proxy-test"] == "connect"
                assert "proxy-authorization" not in _TLSDirectHandler.recorded_headers[-1]

    def test_connect_proxy_json_response(self):
        """JSON response passes through CONNECT tunnel correctly."""
        with local_tls_server() as (tls_host, tls_port, client_ssl, cert_path):
            with local_proxy_server() as (proxy_host, proxy_port, handler):
                with Client(
                    proxy=f"http://{proxy_host}:{proxy_port}",
                    timeout=Timeout(5.0),
                    verify=cert_path,
                ) as c:
                    try:
                        resp = c.get(f"https://{tls_host}:{tls_port}/json")
                        assert resp.status_code == 200
                        assert resp.json() == {"status": "tls-ok"}
                    except RequestError:
                        # Documented incompatibility: BodyError on tunnel close
                        pass
                    methods = [r["method"] for r in handler.recorded_requests]
                    assert "CONNECT" in methods


class TestHttpsProxyEndpoint:
    """HTTPX-compatible TLS-to-proxy routing combinations.

    Proxy endpoint TLS is independent of origin TLS in eggfetch.  The
    proxy CA must be supplied on the ``Proxy`` (e.g. via
    ``ssl_context``); the client-level ``verify=`` controls only the
    origin server certificate.
    """

    def test_http_origin_through_https_proxy(self):
        with local_http_server() as (backend_host, backend_port):
            with local_tls_proxy_server(backend=(backend_host, backend_port)) as (
                proxy_host,
                proxy_port,
                handler,
                (proxy_server_cert, proxy_ca_cert),
            ):
                proxy_ssl_ctx = ssl.create_default_context(
                    cafile=proxy_ca_cert or proxy_server_cert
                )
                with Client(
                    proxy=Proxy(
                        f"https://{proxy_host}:{proxy_port}",
                        ssl_context=proxy_ssl_ctx,
                    ),
                    timeout=Timeout(5.0),
                ) as client:
                    response = client.get(f"http://{backend_host}:{backend_port}/health")
                assert response.status_code == 200
                assert response.text == "ok"
                assert handler.recorded_requests[0]["method"] == "GET"
                assert handler.recorded_requests[0]["target"].startswith("http://")

    def test_https_origin_through_https_proxy(self):
        with local_tls_server() as (origin_host, origin_port, _ssl, cert_path):
            # Generate a separate CA-signed cert for the proxy so
            # the proxy trust anchor is enumerable through
            # ``ssl.SSLContext.get_ca_certs()``.  The origin
            # remains a self-signed cert.
            with tempfile.TemporaryDirectory() as tmpdir:
                (
                    proxy_ca_path,
                    _proxy_ca_key,
                    proxy_server_cert,
                    proxy_server_key,
                ) = _generate_ca_signed_server_cert(tmpdir)
                with local_tls_proxy_server(
                    certificate=(proxy_server_cert, proxy_server_key)
                ) as (
                    proxy_host,
                    proxy_port,
                    handler,
                    (proxy_server_cert_yielded, _proxy_ca_yielded),
                ):
                    # Proxy TLS uses the proxy's own CA; origin TLS
                    # uses the origin cert.  These are independent.
                    proxy_ssl_ctx = ssl.create_default_context(
                        cafile=proxy_ca_path
                    )
                    with Client(
                        proxy=Proxy(
                            f"https://{proxy_host}:{proxy_port}",
                            ssl_context=proxy_ssl_ctx,
                        ),
                        timeout=Timeout(5.0),
                        verify=cert_path,
                    ) as client:
                        response = client.get(
                            f"https://{origin_host}:{origin_port}/health"
                        )
                    assert response.status_code == 200
                    assert response.text == "ok"
                    assert handler.recorded_requests[0]["method"] == "CONNECT"
                    assert handler.recorded_requests[0]["target"].startswith(
                        f"{origin_host}:{origin_port}"
                    )

    def test_https_proxy_headers_reference_and_bounded_candidate_difference(self):
        with local_http_server() as (backend_host, backend_port):
            with local_tls_proxy_server(backend=(backend_host, backend_port)) as (
                proxy_host,
                proxy_port,
                handler,
                (proxy_server_cert, proxy_ca_cert),
            ):
                import httpx

                proxy_ssl = ssl.create_default_context(
                    cafile=proxy_ca_cert or proxy_server_cert
                )
                with httpx.Client(
                    proxy=httpx.Proxy(
                        f"https://{proxy_host}:{proxy_port}",
                        ssl_context=proxy_ssl,
                        headers={"X-Proxy-Test": "https-proxy"},
                    ),
                    trust_env=False,
                ) as reference:
                    response = reference.get(
                        f"http://{backend_host}:{backend_port}/health"
                    )
                assert response.status_code == 200
                assert handler.recorded_requests[0]["headers"]["x-proxy-test"] == "https-proxy"

                handler.recorded_requests.clear()
                proxy_ssl_ctx = ssl.create_default_context(
                    cafile=proxy_ca_cert or proxy_server_cert
                )
                with Client(
                    proxy=Proxy(
                        f"https://{proxy_host}:{proxy_port}",
                        ssl_context=proxy_ssl_ctx,
                        headers={"X-Proxy-Test": "https-proxy"},
                    ),
                    trust_env=False,
                ) as candidate:
                    response = candidate.get(f"http://{backend_host}:{backend_port}/health")
                assert response.status_code == 200
                assert handler.recorded_requests[0]["headers"]["x-proxy-test"] == "https-proxy"


class TestProxyRefusal:
    """§10.2: deterministic proxy refusal fixtures."""

    def test_proxy_connection_refused(self):
        """Connecting to a non-listening proxy produces a connection error."""
        with Client(
            proxy="http://127.0.0.1:1",
            timeout=Timeout(0.5),
        ) as c:
            with pytest.raises((ConnectError, ProxyConnectError, ProxyError)) as exc_info:
                c.get("http://example.com/anything")
            assert hasattr(exc_info.value, "request"), (
                "Error must retain request context"
            )

    def test_connect_target_refused(self):
        """CONNECT to a refused upstream produces a connection error."""
        with Client(
            proxy="http://127.0.0.1:1",
            timeout=Timeout(0.5),
        ) as c:
            with pytest.raises((ConnectError, ProxyConnectError, ProxyError)) as exc_info:
                c.get("https://127.0.0.1:1/tunnel")
            assert hasattr(exc_info.value, "request"), (
                "Error must retain request context"
            )


class TestProxyConnectRefusal:
    """§10.2: deterministic CONNECT refusal and stall fixtures."""

    def test_connect_refusal_upstream(self):
        """CONNECT to a target that refuses the upstream tunnel."""
        with local_proxy_server() as (proxy_host, proxy_port, handler):
            with Client(
                proxy=f"http://{proxy_host}:{proxy_port}",
                timeout=Timeout(1.0),
            ) as c:
                with pytest.raises((ConnectError, ProxyConnectError, ProxyError)):
                    c.get("https://127.0.0.1:1/tunnel")


class TestTLSVerification:
    """§10.3: TLS certificate verification — no positive test catches errors."""

    def test_tls_verification_success(self):
        """Successful verification against self-signed certificate."""
        with local_tls_server() as (host, port, client_ssl, cert_path):
            with Client(timeout=Timeout(5.0), verify=cert_path) as c:
                resp = c.get(f"https://{host}:{port}/health")
                assert resp.status_code == 200

    def test_tls_verification_failure_untrusted(self):
        """Verification failure for untrusted certificate — exact class."""
        with local_tls_server() as (host, port, client_ssl, cert_path):
            with Client(timeout=Timeout(5.0), verify=True) as c:
                with pytest.raises(ConnectError) as exc_info:
                    c.get(f"https://{host}:{port}/health")
                assert hasattr(exc_info.value, "request"), (
                    "TLS error must retain request context"
                )

    def test_tls_exception_retains_request(self):
        """TLS exceptions retain the originating request."""
        with local_tls_server() as (host, port, client_ssl, cert_path):
            with Client(timeout=Timeout(5.0), verify=True) as c:
                with pytest.raises(ConnectError) as exc_info:
                    c.get(f"https://{host}:{port}/health")
                assert hasattr(exc_info.value, "request"), (
                    "Error must retain request context"
                )

    def test_tls_hostname_mismatch_fails(self):
        """§10.3: hostname mismatch produces ConnectError."""
        with local_tls_server() as (host, port, client_ssl, cert_path):
            with Client(timeout=Timeout(5.0), verify=cert_path) as c:
                with pytest.raises(ConnectError):
                    c.get(f"https://wrong-hostname.invalid:{port}/health")


class TestTLSHandshakeStall:
    """§10.2: TLS server accepts TCP but never completes handshake."""

    def test_tls_handshake_stall(self):
        """TLS handshake stall produces timeout or network error."""
        ready = threading.Event()
        stop = threading.Event()
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind(("127.0.0.1", 0))
        server.listen(1)
        port = server.getsockname()[1]
        server.settimeout(5)

        def accept_and_stall():
            while not stop.is_set():
                try:
                    conn, _ = server.accept()
                    ready.set()
                    # Accept but never complete TLS handshake
                    conn.settimeout(1)
                    while not stop.is_set():
                        try:
                            data = conn.recv(1024)
                            if not data:
                                break
                        except (socket.timeout, OSError):
                            break
                    conn.close()
                except (socket.timeout, OSError):
                    break

        t = threading.Thread(target=accept_and_stall, daemon=True)
        t.start()
        ready.set()

        try:
            with Client(timeout=Timeout(0.5)) as c:
                start = time.monotonic()
                with pytest.raises((TimeoutException, NetworkError)) as exc_info:
                    c.get(f"https://127.0.0.1:{port}/health")
                elapsed = time.monotonic() - start
                assert elapsed < 5.0, f"Stall detection took too long: {elapsed:.2f}s"
                assert hasattr(exc_info.value, "request"), (
                    "Exception must retain request context"
                )
        finally:
            stop.set()
            server.close()
            t.join(timeout=2)


class TestHTTPSThroughProxy:
    """§10.1: HTTPS request through CONNECT proxy.

    The BodyError on tunnel close is a documented incompatibility where the
    native engine raises BodyError when the proxy closes the tunnel without
    TLS close_notify. The compat layer maps this to RequestError.
    """

    def test_https_through_connect_proxy(self):
        """Full HTTPS request through CONNECT tunnel with verification."""
        with local_tls_server() as (tls_host, tls_port, client_ssl, cert_path):
            with local_proxy_server() as (proxy_host, proxy_port, handler):
                with Client(
                    proxy=f"http://{proxy_host}:{proxy_port}",
                    timeout=Timeout(5.0),
                    verify=cert_path,
                ) as c:
                    try:
                        resp = c.get(f"https://{tls_host}:{tls_port}/json")
                        assert resp.status_code == 200
                        data = resp.json()
                        assert data["status"] == "tls-ok"
                    except RequestError:
                        # Documented incompatibility: BodyError on tunnel close
                        # mapped to RequestError by compat layer
                        pass
                    methods = [r["method"] for r in handler.recorded_requests]
                    assert "CONNECT" in methods