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

"""Tests for HTTPX-compatible transport protocols."""

from __future__ import annotations

import pytest

from eggfetch.compat.httpx import (
    BaseTransport,
    AsyncBaseTransport,
    HTTPTransport,
    AsyncHTTPTransport,
    MockTransport,
    Request,
    Response,
    URL,
)


class TestBaseTransport:
    def test_handle_request_not_implemented(self):
        transport = BaseTransport()
        request = Request("GET", "http://example.com/")
        with pytest.raises(NotImplementedError):
            transport.handle_request(request)

    def test_close_is_noop(self):
        transport = BaseTransport()
        transport.close()

    def test_context_manager(self):
        with BaseTransport() as transport:
            assert isinstance(transport, BaseTransport)

    def test_close_idempotent(self):
        transport = BaseTransport()
        transport.close()
        transport.close()


class TestAsyncBaseTransport:
    @pytest.mark.asyncio
    async def test_handle_async_request_not_implemented(self):
        transport = AsyncBaseTransport()
        request = Request("GET", "http://example.com/")
        with pytest.raises(NotImplementedError):
            await transport.handle_async_request(request)

    @pytest.mark.asyncio
    async def test_aclose_is_noop(self):
        transport = AsyncBaseTransport()
        await transport.aclose()

    @pytest.mark.asyncio
    async def test_async_context_manager(self):
        async with AsyncBaseTransport() as transport:
            assert isinstance(transport, AsyncBaseTransport)


class TestHTTPTransport:
    def test_constructor_defaults(self):
        transport = HTTPTransport()
        assert transport._verify is True
        assert transport._http1 is True
        assert transport._http2 is False

    def test_constructor_custom_params(self):
        transport = HTTPTransport(
            verify=False,
            http2=True,
            retries=3,
        )
        assert transport._verify is False
        assert transport._http2 is True
        assert transport._retries == 3

    def test_context_manager(self):
        with HTTPTransport() as transport:
            assert isinstance(transport, HTTPTransport)

    def test_close_idempotent(self):
        transport = HTTPTransport()
        transport.close()
        transport.close()


class TestAsyncHTTPTransport:
    def test_constructor_defaults(self):
        transport = AsyncHTTPTransport()
        assert transport._verify is True
        assert transport._http1 is True

    @pytest.mark.asyncio
    async def test_async_context_manager(self):
        async with AsyncHTTPTransport() as transport:
            assert isinstance(transport, AsyncHTTPTransport)

    @pytest.mark.asyncio
    async def test_aclose_idempotent(self):
        transport = AsyncHTTPTransport()
        await transport.aclose()
        await transport.aclose()


class TestTransportDispatch:
    def test_custom_transport_overrides_native(self):
        def handler(request):
            return Response(200, content=b"from transport")

        transport = MockTransport(handler)

        from eggfetch.compat.httpx import Client

        client = Client(transport=transport)
        response = client.get("http://example.com/")
        assert response.status_code == 200
        assert response.content == b"from transport"
        client.close()

    def test_transport_receives_full_request(self):
        received = []

        def handler(request):
            received.append(request)
            return Response(200, content=b"ok")

        from eggfetch.compat.httpx import Client

        client = Client(
            transport=MockTransport(handler),
            headers={"x-custom": "value"},
        )
        client.get("http://example.com/test")
        assert len(received) == 1
        req = received[0]
        assert req.method == "GET"
        assert "x-custom" in req.headers
        client.close()