eggfetch-python 0.1.4

Python sync and asyncio bindings for the eggfetch HTTP engine (Rust core via PyO3; Python users install from PyPI)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

"""Tests for the HTTP/2 client option."""

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

import pytest

import eggfetch

from conftest import _ThreadingHTTPServer


# ---------------------------------------------------------------------------
# Local test server (HTTP/1.1 only)
# ---------------------------------------------------------------------------

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 log_message(self, format, *args):
        pass  # suppress logs during tests


@pytest.fixture(scope="module")
def server():
    """Start a local HTTP/1.1 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()


# ---------------------------------------------------------------------------
# Sync client http2 option
# ---------------------------------------------------------------------------

class TestSyncClientHttp2:
    def test_client_rejects_no_protocols(self):
        with pytest.raises(ValueError, match="At least one of http1 or http2"):
            eggfetch.Client(http1=False, http2=False)

    def test_client_construction_http2_true(self):
        client = eggfetch.Client(http2=True)
        assert client is not None

    def test_client_construction_http2_false(self):
        client = eggfetch.Client(http2=False)
        assert client is not None

    def test_client_construction_no_http2(self):
        client = eggfetch.Client()
        assert client is not None

    def test_http2_client_makes_request(self, server):
        """Client with http2=True can still make requests (falls back to h1)."""
        client = eggfetch.Client(http2=True)
        r = client.get(f"{server}/hello")
        assert r.status_code == 200
        data = json.loads(r.text)
        assert data["method"] == "GET"
        assert data["path"] == "/hello"

    def test_http2_client_post(self, server):
        client = eggfetch.Client(http2=True)
        r = client.post(f"{server}/echo", content=b"hello")
        assert r.status_code == 200
        data = json.loads(r.text)
        assert data["method"] == "POST"

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


# ---------------------------------------------------------------------------
# Async client http2 option
# ---------------------------------------------------------------------------

class TestAsyncClientHttp2:
    def test_async_client_rejects_no_protocols(self):
        with pytest.raises(ValueError, match="At least one of http1 or http2"):
            eggfetch.AsyncClient(http1=False, http2=False)

    def test_async_client_construction_http2_true(self):
        client = eggfetch.AsyncClient(http2=True)
        assert client is not None

    def test_async_client_construction_http2_false(self):
        client = eggfetch.AsyncClient(http2=False)
        assert client is not None

    @pytest.mark.asyncio
    async def test_async_http2_client_makes_request(self, server):
        async with eggfetch.AsyncClient(http2=True) as client:
            r = await client.get(f"{server}/hello")
            assert r.status_code == 200
            data = json.loads(r.text)
            assert data["method"] == "GET"

    @pytest.mark.asyncio
    async def test_async_http2_client_post(self, server):
        async with eggfetch.AsyncClient(http2=True) as client:
            r = await client.post(f"{server}/echo", content=b"hello")
            assert r.status_code == 200
            data = json.loads(r.text)
            assert data["method"] == "POST"