eggfetch-python 0.1.4

"""Tests for the eggfetch Python sync API."""

import http.server
import json
import threading
import time
import urllib.parse

import pytest

import eggfetch

from conftest import _ThreadingHTTPServer


# ---------------------------------------------------------------------------
# Local test server
# ---------------------------------------------------------------------------

class _Handler(http.server.BaseHTTPRequestHandler):
    """Minimal test server that echoes request details."""

    def do_GET(self):
        parsed = urllib.parse.urlparse(self.path)
        body = json.dumps({
            "method": "GET",
            "path": parsed.path,
            "query": parsed.query,
            "headers": dict(self.headers),
        }).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        raw = self.rfile.read(length) if length else b""
        body = json.dumps({
            "method": "POST",
            "path": self.path,
            "headers": dict(self.headers),
            "body": raw.decode(errors="replace"),
        }).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_PUT(self):
        self.do_POST()

    def do_PATCH(self):
        self.do_POST()

    def do_DELETE(self):
        self.do_GET()

    def do_HEAD(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.send_header("X-Echo", "head-ok")
        self.end_headers()

    def do_OPTIONS(self):
        self.send_response(200)
        self.send_header("Allow", "GET, POST, OPTIONS")
        self.end_headers()

    def log_message(self, format, *args):
        pass  # suppress logs during tests


@pytest.fixture(scope="module")
def server():
    """Start a local HTTP server for the test module."""
    srv = _ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
    port = srv.server_address[1]
    t = threading.Thread(target=srv.serve_forever, daemon=True)
    t.start()
    yield f"http://127.0.0.1:{port}"
    srv.shutdown()


# ---------------------------------------------------------------------------
# Package imports
# ---------------------------------------------------------------------------

class TestPackageImports:
    def test_import_version(self):
        assert isinstance(eggfetch.__version__, str)
        assert eggfetch.__version__ == "0.1.0"

    def test_imports_classes(self):
        assert hasattr(eggfetch, "Client")
        assert hasattr(eggfetch, "Response")
        assert hasattr(eggfetch, "Headers")
        assert hasattr(eggfetch, "Timeout")

    def test_imports_functions(self):
        for name in ("request", "get", "post", "put", "patch", "delete", "head", "options"):
            assert hasattr(eggfetch, name), f"missing {name}"

    def test_imports_exceptions(self):
        for name in (
            "EggfetchError", "RequestError", "InvalidUrl",
            "TimeoutException", "PoolTimeout", "ConnectTimeout",
            "ReadTimeout", "WriteTimeout", "NetworkError",
            "ProtocolError", "BodyError", "HTTPStatusError",
        ):
            assert hasattr(eggfetch, name), f"missing {name}"

    def test_exception_hierarchy(self):
        assert issubclass(eggfetch.RequestError, eggfetch.EggfetchError)
        assert issubclass(eggfetch.InvalidUrl, eggfetch.RequestError)
        assert issubclass(eggfetch.TimeoutException, eggfetch.RequestError)
        assert issubclass(eggfetch.HTTPStatusError, eggfetch.EggfetchError)


# ---------------------------------------------------------------------------
# Top-level helpers
# ---------------------------------------------------------------------------

class TestTopLevelGet:
    def test_get_returns_response(self, server):
        r = eggfetch.get(f"{server}/hello")
        assert isinstance(r, eggfetch.Response)
        assert r.status_code == 200

    def test_get_body(self, server):
        r = eggfetch.get(f"{server}/hello")
        data = json.loads(r.text)
        assert data["method"] == "GET"
        assert data["path"] == "/hello"

    def test_get_is_success(self, server):
        r = eggfetch.get(f"{server}/hello")
        assert r.is_success

    def test_get_url_property(self, server):
        r = eggfetch.get(f"{server}/hello")
        assert r.url == f"{server}/hello"


class TestTopLevelPost:
    def test_post_content(self, server):
        r = eggfetch.post(f"{server}/api", content=b"hello world")
        assert r.status_code == 200
        data = json.loads(r.text)
        assert data["method"] == "POST"
        assert data["body"] == "hello world"


class TestHeaders:
    def test_headers_reach_server(self, server):
        r = eggfetch.get(f"{server}/hello", headers={"X-Custom": "test-value"})
        data = json.loads(r.text)
        # HTTP headers are case-insensitive; the http crate normalizes to lowercase
        assert data["headers"].get("x-custom") == "test-value"

    def test_response_headers(self, server):
        r = eggfetch.get(f"{server}/hello")
        assert "content-type" in r.headers

    def test_headers_as_sequence_of_pairs(self, server):
        r = eggfetch.get(
            f"{server}/hello",
            headers=[("X-Pair", "val1"), ("X-Pair", "val2")],
        )
        data = json.loads(r.text)
        assert data["headers"].get("x-pair") == "val2"

    def test_request_headers_override_client_default(self, server):
        with eggfetch.Client(headers={"X-Foo": "from-client"}) as client:
            r = client.get(
                f"{server}/hello",
                headers={"X-Foo": "override", "X-Bar": "from-request"},
            )
            data = json.loads(r.text)
            # Request headers are sent; verify the request-specific header arrives.
            assert data["headers"].get("x-bar") == "from-request"

    def test_invalid_header_value_raises(self, server):
        with pytest.raises(eggfetch.RequestError):
            eggfetch.get(f"{server}/hello", headers={"X-Bad": "val\nue"})


class TestParams:
    def test_params_serialized(self, server):
        r = eggfetch.get(f"{server}/search", params={"q": "hello", "page": "1"})
        data = json.loads(r.text)
        assert "q=hello" in data["query"]
        assert "page=1" in data["query"]

    def test_params_sequence_of_pairs(self, server):
        r = eggfetch.get(
            f"{server}/search",
            params=[("q", "test"), ("q", "other")],
        )
        data = json.loads(r.text)
        assert "q=test" in data["query"]
        assert "q=other" in data["query"]

    def test_params_with_existing_query(self, server):
        r = eggfetch.get(
            f"{server}/search?existing=1",
            params={"q": "hello"},
        )
        data = json.loads(r.text)
        assert "existing=1" in data["query"]
        assert "q=hello" in data["query"]

    def test_params_none_ignored(self, server):
        r = eggfetch.get(f"{server}/search", params=None)
        data = json.loads(r.text)
        assert data["query"] == ""

    def test_invalid_params_type_raises(self, server):
        with pytest.raises(TypeError):
            eggfetch.get(f"{server}/search", params=123)


# ---------------------------------------------------------------------------
# Client
# ---------------------------------------------------------------------------

class TestClient:
    def test_client_context_manager(self, server):
        with eggfetch.Client() as client:
            r = client.get(f"{server}/hello")
            assert r.status_code == 200

    def test_client_reuses_connection(self, server):
        with eggfetch.Client() as client:
            r1 = client.get(f"{server}/hello")
            r2 = client.get(f"{server}/hello")
            assert r1.status_code == 200
            assert r2.status_code == 200

    def test_client_default_headers(self, server):
        with eggfetch.Client(headers={"X-Client-Header": "from-client"}) as client:
            r = client.get(f"{server}/hello")
            data = json.loads(r.text)
            # HTTP headers are case-insensitive; the http crate normalizes to lowercase
            assert data["headers"].get("x-client-header") == "from-client"

    def test_client_post(self, server):
        with eggfetch.Client() as client:
            r = client.post(f"{server}/api", content=b"client-post")
            data = json.loads(r.text)
            assert data["method"] == "POST"
            assert data["body"] == "client-post"

    def test_closed_client_raises(self, server):
        client = eggfetch.Client()
        client.close()
        with pytest.raises(ValueError, match="closed"):
            client.get(f"{server}/hello")

    def test_client_is_closed_property(self):
        client = eggfetch.Client()
        assert not client.is_closed
        client.close()
        assert client.is_closed


# ---------------------------------------------------------------------------
# Timeout
# ---------------------------------------------------------------------------

class TestTimeout:
    def test_scalar_timeout(self, server):
        r = eggfetch.get(f"{server}/hello", timeout=10.0)
        assert r.status_code == 200

    def test_timeout_none(self, server):
        r = eggfetch.get(f"{server}/hello", timeout=None)
        assert r.status_code == 200

    def test_request_timeout_overrides_client_default(self, server):
        with eggfetch.Client(timeout=0.001) as client:
            # Client timeout is very short; request-level override should succeed
            r = client.get(f"{server}/hello", timeout=10.0)
            assert r.status_code == 200

    def test_timeout_zero_is_valid_everywhere(self):
        """Zero timeouts are accepted consistently by all native paths."""
        t = eggfetch.Timeout(0)
        assert t.pool == 0.0
        assert t.connect == 0.0
        assert t.read == 0.0
        assert t.write == 0.0
        with pytest.raises(ValueError, match="non-negative"):
            eggfetch.Timeout(-1)

    def test_negative_timeout_raises(self, server):
        with pytest.raises(ValueError, match="non-negative"):
            eggfetch.get(f"{server}/hello", timeout=-5.0)


# ---------------------------------------------------------------------------
# Socket options validation
# ---------------------------------------------------------------------------

class TestSocketOptionsValidation:
    def test_four_element_forms_rejected_uniformly_with_value_error(self):
        """Both four-element shapes fail closed with ValueError.

        The null-pointer ``(level, option, None, optlen)`` form is outside
        the safe socket API and any other four-element shape is malformed;
        both must surface as ``ValueError`` (matching the HTTPX facade) so
        a single ``except ValueError`` catches every user mistake.
        """
        import socket

        null_pointer = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, None, 0)]
        with pytest.raises(ValueError, match="four-element"):
            eggfetch.Client(socket_options=null_pointer)

        malformed = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1, 0)]
        with pytest.raises(ValueError, match="four-element"):
            eggfetch.Client(socket_options=malformed)


# ---------------------------------------------------------------------------
# Limits validation
# ---------------------------------------------------------------------------

class TestLimitsValidation:
    def test_keepalive_expiry_rejects_non_finite_and_negative(self):
        with pytest.raises(ValueError):
            eggfetch.Limits(keepalive_expiry=-1.0)
        with pytest.raises(ValueError):
            eggfetch.Limits(keepalive_expiry=float("nan"))
        with pytest.raises(ValueError):
            eggfetch.Limits(keepalive_expiry=float("inf"))

    def test_keepalive_expiry_accepts_zero(self):
        limits = eggfetch.Limits(keepalive_expiry=0.0)
        assert limits.keepalive_expiry == 0.0


# ---------------------------------------------------------------------------
# Error mapping
# ---------------------------------------------------------------------------

class TestErrors:
    def test_invalid_url(self):
        with pytest.raises(ValueError):
            eggfetch.get("not-a-url")

    def test_unsupported_scheme(self):
        with pytest.raises(eggfetch.RequestError, match="not supported"):
            eggfetch.get("ftp://example.com")

    def test_raise_for_status_4xx(self, server):
        # The test server always returns 200, so we construct a response manually
        # by testing that raise_for_status works on a normal response
        r = eggfetch.get(f"{server}/hello")
        r.raise_for_status()  # should not raise

    def test_raise_for_status_failure(self):
        # Create a response object and test raise_for_status manually
        # Since we can't easily get a 4xx from our test server, test the exception type
        assert issubclass(eggfetch.HTTPStatusError, eggfetch.EggfetchError)


# ---------------------------------------------------------------------------
# Unsupported kwargs
# ---------------------------------------------------------------------------

class TestUnsupportedKwargs:
    def test_unsupported_kwarg_top_level(self, server):
        with pytest.raises(TypeError):
            eggfetch.get(f"{server}/hello", json={"key": "value"})

    def test_unsupported_kwarg_client(self, server):
        with eggfetch.Client() as client:
            with pytest.raises(TypeError):
                client.get(f"{server}/hello", json={"key": "value"})


# ---------------------------------------------------------------------------
# Body kwargs: content, data, json
# ---------------------------------------------------------------------------

class TestContent:
    def test_content_dict_rejected(self, server):
        with pytest.raises(TypeError, match="content must be bytes, str"):
            eggfetch.post(f"{server}/api", content={"key": "value"})

    def test_content_bytes(self, server):
        r = eggfetch.post(f"{server}/api", content=b"raw bytes")
        data = json.loads(r.text)
        assert data["body"] == "raw bytes"

    def test_content_str(self, server):
        r = eggfetch.post(f"{server}/api", content="string body")
        data = json.loads(r.text)
        assert data["body"] == "string body"

    def test_content_bytearray(self, server):
        r = eggfetch.post(f"{server}/api", content=bytearray(b"ba data"))
        data = json.loads(r.text)
        assert data["body"] == "ba data"

    def test_content_no_auto_content_type(self, server):
        r = eggfetch.post(f"{server}/api", content=b"raw")
        data = json.loads(r.text)
        # raw content should not set content-type automatically
        assert data["headers"].get("content-type") is None or \
               data["headers"].get("content-type") != "application/x-www-form-urlencoded"


class TestFormData:
    def test_form_dict(self, server):
        r = eggfetch.post(f"{server}/api", data={"a": "1", "b": "2"})
        data = json.loads(r.text)
        assert "a=1" in data["body"]
        assert "b=2" in data["body"]
        assert data["headers"].get("content-type") == "application/x-www-form-urlencoded"

    def test_form_sequence_of_pairs(self, server):
        r = eggfetch.post(
            f"{server}/api",
            data=[("a", "1"), ("a", "2")],
        )
        data = json.loads(r.text)
        assert "a=1" in data["body"]
        assert "a=2" in data["body"]

    def test_form_percent_encoding(self, server):
        r = eggfetch.post(f"{server}/api", data={"key": "hello world"})
        data = json.loads(r.text)
        assert "key=hello+world" in data["body"] or "key=hello%20world" in data["body"]

    def test_form_preserves_user_content_type(self, server):
        r = eggfetch.post(
            f"{server}/api",
            headers={"Content-Type": "custom/type"},
            data={"a": "1"},
        )
        data = json.loads(r.text)
        assert data["headers"].get("content-type") == "custom/type"


class TestJsonBody:
    def test_json_dict(self, server):
        r = eggfetch.post(f"{server}/api", json={"hello": "world"})
        data = json.loads(r.text)
        body = json.loads(data["body"])
        assert body == {"hello": "world"}
        assert data["headers"].get("content-type") == "application/json"

    def test_json_list(self, server):
        r = eggfetch.post(f"{server}/api", json=[1, 2, 3])
        data = json.loads(r.text)
        body = json.loads(data["body"])
        assert body == [1, 2, 3]

    def test_json_nested(self, server):
        r = eggfetch.post(f"{server}/api", json={"a": {"b": [1, 2]}})
        data = json.loads(r.text)
        body = json.loads(data["body"])
        assert body == {"a": {"b": [1, 2]}}

    def test_json_preserves_user_content_type(self, server):
        r = eggfetch.post(
            f"{server}/api",
            headers={"Content-Type": "custom/json"},
            json={"a": 1},
        )
        data = json.loads(r.text)
        assert data["headers"].get("content-type") == "custom/json"

    def test_json_unserializable_raises(self, server):
        with pytest.raises(TypeError):
            eggfetch.post(f"{server}/api", json=object())


class TestBodyConflict:
    def test_content_and_json_raises(self, server):
        with pytest.raises(TypeError, match="only one of content, data, or json"):
            eggfetch.post(f"{server}/api", content=b"raw", json={"a": 1})

    def test_content_and_data_raises(self, server):
        with pytest.raises(TypeError, match="only one of content, data, or json"):
            eggfetch.post(f"{server}/api", content=b"raw", data={"a": "1"})

    def test_data_and_json_raises(self, server):
        with pytest.raises(TypeError, match="only one of content, data, or json"):
            eggfetch.post(f"{server}/api", data={"a": "1"}, json={"b": 2})

    def test_all_three_raises(self, server):
        with pytest.raises(TypeError, match="only one of content, data, or json"):
            eggfetch.post(
                f"{server}/api",
                content=b"raw",
                data={"a": "1"},
                json={"b": 2},
            )


# ---------------------------------------------------------------------------
# Client body kwargs
# ---------------------------------------------------------------------------

class TestClientBodyKwargs:
    def test_client_json(self, server):
        with eggfetch.Client() as client:
            r = client.post(f"{server}/api", json={"key": "value"})
            data = json.loads(r.text)
            body = json.loads(data["body"])
            assert body == {"key": "value"}
            assert data["headers"].get("content-type") == "application/json"

    def test_client_form_data(self, server):
        with eggfetch.Client() as client:
            r = client.post(f"{server}/api", data={"a": "1"})
            data = json.loads(r.text)
            assert "a=1" in data["body"]

    def test_client_content(self, server):
        with eggfetch.Client() as client:
            r = client.post(f"{server}/api", content=b"raw bytes")
            data = json.loads(r.text)
            assert data["body"] == "raw bytes"

    def test_client_put_json(self, server):
        with eggfetch.Client() as client:
            r = client.put(f"{server}/api", json={"updated": True})
            data = json.loads(r.text)
            body = json.loads(data["body"])
            assert body == {"updated": True}

    def test_client_patch_json(self, server):
        with eggfetch.Client() as client:
            r = client.patch(f"{server}/api", json={"patched": 1})
            data = json.loads(r.text)
            body = json.loads(data["body"])
            assert body == {"patched": 1}