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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

"""Track 5: Protocol validation and transport option acceptance.

Tests that protocol combinations are validated and transport options
are now accepted (previously raised NotImplementedError).
"""

import pytest

from eggfetch.compat.httpx import Client, AsyncClient, Timeout
from eggfetch.compat.httpx._transports import HTTPTransport, AsyncHTTPTransport
from eggfetch.compat.httpx._client import (
    _convert_socket_option,
    _validate_protocol_options,
    _validate_transport_options,
)


# ---------------------------------------------------------------------------
# Protocol validation
# ---------------------------------------------------------------------------

class TestProtocolValidation:
    def test_http1_true_http2_false_ok(self):
        client = Client(http1=True, http2=False)
        assert client._http1 is True
        assert client._http2 is False

    def test_http1_true_http2_true_ok(self):
        client = Client(http1=True, http2=True)
        assert client._http1 is True
        assert client._http2 is True

    def test_http1_false_http2_true_ok(self):
        """H2-only mode is now accepted (prior knowledge / ALPN-only)."""
        client = Client(http1=False, http2=True)
        assert client._http1 is False
        assert client._http2 is True

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

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

    def test_async_client_h2_only_ok(self):
        """AsyncClient H2-only mode is now accepted."""
        client = AsyncClient(http1=False, http2=True)
        assert client._http1 is False
        assert client._http2 is True

    def test_validate_protocol_direct(self):
        _validate_protocol_options(True, False)  # should not raise
        _validate_protocol_options(True, True)   # should not raise
        _validate_protocol_options(False, True)  # H2-only should not raise

    def test_validate_protocol_both_false(self):
        with pytest.raises(ValueError):
            _validate_protocol_options(False, False)

    def test_transport_h2_only_ok(self):
        """HTTPTransport H2-only mode is now accepted."""
        transport = HTTPTransport(http1=False, http2=True)
        assert transport._http1 is False
        assert transport._http2 is True

    def test_async_transport_h2_only_ok(self):
        """AsyncHTTPTransport H2-only mode is now accepted."""
        transport = AsyncHTTPTransport(http1=False, http2=True)
        assert transport._http1 is False
        assert transport._http2 is True


# ---------------------------------------------------------------------------
# Transport option acceptance (formerly rejected)
# ---------------------------------------------------------------------------

class TestTransportOptionsAccepted:
    def test_socket_option_bytearray_is_losslessly_converted(self):
        value = bytearray(b"\x01\x00\x00\x00")
        assert _convert_socket_option((1, 2, value)) == (1, 2, bytes(value))

    def test_uds_accepted(self):
        """UDS is now accepted in the transport constructor."""
        transport = HTTPTransport(uds="/tmp/test.sock")
        assert transport._uds == "/tmp/test.sock"

    def test_local_address_accepted(self):
        """local_address is now accepted in the transport constructor."""
        transport = HTTPTransport(local_address="127.0.0.1")
        assert transport._local_address == "127.0.0.1"

    def test_socket_options_accepted(self):
        """socket_options is now accepted in the transport constructor."""
        opts = [(6, 1, b"\x01\x00\x00\x00")]  # TCP_NODELAY
        transport = HTTPTransport(socket_options=opts)
        assert transport._socket_options == opts

    def test_async_uds_accepted(self):
        """UDS is now accepted in the async transport constructor."""
        transport = AsyncHTTPTransport(uds="/tmp/test.sock")
        assert transport._uds == "/tmp/test.sock"

    def test_async_local_address_accepted(self):
        """local_address is now accepted in the async transport constructor."""
        transport = AsyncHTTPTransport(local_address="127.0.0.1")
        assert transport._local_address == "127.0.0.1"

    def test_async_socket_options_accepted(self):
        """socket_options is now accepted in the async transport constructor."""
        opts = [(6, 1, b"\x01\x00\x00\x00")]  # TCP_NODELAY
        transport = AsyncHTTPTransport(socket_options=opts)
        assert transport._socket_options == opts

    def test_default_none_values_accepted(self):
        """Default None values should be accepted for signature compatibility."""
        transport = HTTPTransport(
            uds=None,
            local_address=None,
            socket_options=None,
        )
        assert transport._uds is None
        assert transport._local_address is None
        assert transport._socket_options is None

    def test_async_default_none_values_accepted(self):
        """Default None values should be accepted for signature compatibility."""
        transport = AsyncHTTPTransport(
            uds=None,
            local_address=None,
            socket_options=None,
        )
        assert transport._uds is None
        assert transport._local_address is None
        assert transport._socket_options is None

    def test_validate_transport_options_direct(self):
        _validate_transport_options()  # should not raise
        _validate_transport_options(uds=None, local_address=None,
                                   socket_options=None)  # should not raise

    def test_validate_transport_options_uds(self):
        _validate_transport_options(uds="/tmp/test.sock")  # should not raise

    def test_validate_transport_options_local_address(self):
        _validate_transport_options(local_address="127.0.0.1")  # should not raise

    def test_validate_transport_options_socket_options(self):
        _validate_transport_options(socket_options=[(6, 1, b"\x01")])  # should not raise

    def test_validate_transport_options_invalid_local_address(self):
        with pytest.raises(ValueError, match="invalid local_address"):
            _validate_transport_options(local_address="bad-format")

    def test_validate_transport_options_invalid_socket_options_type(self):
        with pytest.raises(TypeError, match="list of tuples"):
            _validate_transport_options(socket_options="not-a-list")

    def test_four_tuple_socket_option_is_bounded(self):
        """The facade rejects an unusable four-tuple at its safe boundary."""
        with pytest.raises(ValueError, match="four-element"):
            HTTPTransport(
                socket_options=[(6, 1, b"\x01", 0)],
            )

    def test_validate_transport_options_invalid_socket_option_triple(self):
        with pytest.raises(ValueError, match="triples"):
            _validate_transport_options(socket_options=[(1, 2)])  # only 2 elements