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

"""Parameterized matrix tests for configuration merging."""

import pytest

from eggfetch.compat.httpx import (
    Client,
    Headers,
    Cookies,
    QueryParams,
    Timeout,
    URL,
)


# ── Client params + request params ──────────────────────────────────────

class TestParamsMerging:
    def test_client_params_used_when_no_request_params(self):
        client = Client(params={"a": "1", "b": "2"})
        req = client.build_request("GET", "https://example.com")
        assert req.params["a"] == "1"
        assert req.params["b"] == "2"

    def test_request_params_override_client(self):
        client = Client(params={"a": "1"})
        req = client.build_request("GET", "https://example.com", params={"a": "9"})
        assert req.params["a"] == "9"

    def test_request_params_add_to_client(self):
        client = Client(params={"a": "1"})
        req = client.build_request("GET", "https://example.com", params={"b": "2"})
        assert req.params["a"] == "1"
        assert req.params["b"] == "2"

    def test_empty_client_params(self):
        client = Client()
        req = client.build_request("GET", "https://example.com", params={"a": "1"})
        assert req.params["a"] == "1"


# ── Client headers + request headers ────────────────────────────────────

class TestHeadersMerging:
    def test_client_headers_used_when_no_request_headers(self):
        client = Client(headers={"x-client": "val"})
        req = client.build_request("GET", "https://example.com")
        assert req.headers["x-client"] == "val"

    def test_request_headers_override_client(self):
        client = Client(headers={"x-custom": "old"})
        req = client.build_request(
            "GET", "https://example.com", headers={"x-custom": "new"}
        )
        assert req.headers["x-custom"] == "new"

    def test_request_headers_add_to_client(self):
        client = Client(headers={"x-client": "val"})
        req = client.build_request(
            "GET", "https://example.com", headers={"x-req": "val"}
        )
        assert req.headers["x-client"] == "val"
        assert req.headers["x-req"] == "val"

    def test_empty_client_headers(self):
        client = Client()
        req = client.build_request(
            "GET", "https://example.com", headers={"x-req": "val"}
        )
        assert req.headers["x-req"] == "val"


# ── Client cookies + request cookies ────────────────────────────────────

class TestCookiesMerging:
    def test_client_cookies_used_when_no_request_cookies(self):
        client = Client(cookies={"session": "abc"})
        req = client.build_request("GET", "https://example.com")
        assert req.cookies["session"] == "abc"

    def test_request_cookies_override_client(self):
        client = Client(cookies={"session": "old"})
        req = client.build_request(
            "GET", "https://example.com", cookies={"session": "new"}
        )
        assert req.cookies["session"] == "new"

    def test_request_cookies_add_to_client(self):
        client = Client(cookies={"a": "1"})
        req = client.build_request(
            "GET", "https://example.com", cookies={"b": "2"}
        )
        assert req.cookies["a"] == "1"
        assert req.cookies["b"] == "2"

    def test_empty_client_cookies(self):
        client = Client()
        req = client.build_request(
            "GET", "https://example.com", cookies={"a": "1"}
        )
        assert req.cookies["a"] == "1"


# ── base_url + relative URL ────────────────────────────────────────────

class TestBaseUrlMerging:
    def test_relative_url_joined(self):
        client = Client(base_url="https://api.example.com/v1")
        req = client.build_request("GET", "/users")
        assert "api.example.com" in str(req.url)

    def test_absolute_url_wins(self):
        client = Client(base_url="https://api.example.com/v1")
        req = client.build_request("GET", "https://other.com/override")
        assert req.url.host == "other.com"
        assert "v1" not in str(req.url)

    def test_no_base_url(self):
        client = Client()
        req = client.build_request("GET", "https://example.com/path")
        assert req.url.host == "example.com"
        assert req.url.path == "/path"

    def test_empty_base_url(self):
        client = Client(base_url="")
        req = client.build_request("GET", "https://example.com/path")
        assert req.url.host == "example.com"


# ── timeout override ────────────────────────────────────────────────────

class TestTimeoutOverride:
    def test_client_timeout_used(self):
        client = Client(timeout=Timeout(10.0))
        assert client.timeout.total == 10.0

    def test_timeout_override_in_request(self):
        client = Client(timeout=Timeout(10.0))
        assert client.timeout.total == 10.0
        # The per-request timeout override happens at send() level


# ── auth override ──────────────────────────────────────────────────────

class TestAuthOverride:
    def test_client_auth(self):
        from eggfetch.compat.httpx._auth import BasicAuth
        client = Client(auth=("user", "pass"))
        assert isinstance(client.auth, BasicAuth)
        assert client.auth.username == "user"
        assert client.auth.password == "pass"

    def test_client_auth_none(self):
        client = Client()
        assert client.auth is None