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

"""Tests for MockTransport."""
from __future__ import annotations

import asyncio
import pytest
from eggfetch.compat.httpx import (
    Client,
    AsyncClient,
    MockTransport,
    Request,
    Response,
    _build_response,
)


class TestMockTransportSync:
    def test_basic_handler(self):
        def handler(request):
            return Response(200, content=b"Hello")

        with Client(transport=MockTransport(handler)) as client:
            resp = client.get("http://testserver/")
            assert resp.status_code == 200
            assert resp.content == b"Hello"

    def test_handler_receives_request(self):
        received = []

        def handler(request):
            received.append(request)
            return Response(200)

        with Client(transport=MockTransport(handler)) as client:
            client.post("http://testserver/data", content=b"body")

        assert len(received) == 1
        assert received[0].method == "POST"
        assert received[0].content == b"body"

    def test_response_request_attached(self):
        def handler(request):
            return Response(200, content=b"ok")

        with Client(transport=MockTransport(handler)) as client:
            resp = client.get("http://testserver/")
            assert resp.request is not None
            assert resp.request.method == "GET"

    def test_handler_exception_propagates(self):
        def handler(request):
            raise ValueError("test error")

        with Client(transport=MockTransport(handler)) as client:
            with pytest.raises(ValueError, match="test error"):
                client.get("http://testserver/")

    def test_mock_with_status_codes(self):
        def handler(request):
            if "404" in str(request.url):
                return Response(404, content=b"Not Found")
            return Response(200, content=b"OK")

        with Client(transport=MockTransport(handler)) as client:
            resp = client.get("http://testserver/404")
            assert resp.status_code == 404

    def test_close_idempotent(self):
        def handler(request):
            return Response(200)

        transport = MockTransport(handler)
        transport.close()
        transport.close()

    def test_context_manager(self):
        def handler(request):
            return Response(200)

        with MockTransport(handler) as transport:
            assert not transport._is_closed

    def test_handler_returns_json(self):
        def handler(request):
            return Response(200, json={"key": "value"})

        with Client(transport=MockTransport(handler)) as client:
            resp = client.get("http://testserver/")
            assert resp.json() == {"key": "value"}

    def test_sync_client_rejects_async_handler(self):
        async def handler(request):
            return Response(200)

        with pytest.raises(RuntimeError, match="async"):
            with Client(transport=MockTransport(handler)) as client:
                client.get("http://testserver/")

    def test_closed_transport_raises(self):
        def handler(request):
            return Response(200)

        transport = MockTransport(handler)
        transport.close()
        with pytest.raises(RuntimeError, match="closed"):
            transport.handle_request(Request("GET", "http://test/"))

    def test_streaming_response_via_handler(self):
        def handler(request):
            def body_iter():
                yield b"chunk1-"
                yield b"chunk2-"
                yield b"chunk3"

            return Response(200, stream=body_iter())

        with Client(transport=MockTransport(handler)) as client:
            with client.stream("GET", "http://testserver/") as resp:
                assert resp.status_code == 200
                content = resp.read()
                assert content == b"chunk1-chunk2-chunk3"


class TestMockTransportAsync:
    @pytest.mark.asyncio
    async def test_async_handler(self):
        async def handler(request):
            return Response(200, content=b"async")

        async with AsyncClient(
            async_transport=MockTransport(handler)
        ) as client:
            resp = await client.get("http://testserver/")
            assert resp.content == b"async"

    @pytest.mark.asyncio
    async def test_sync_handler_in_async_client(self):
        def handler(request):
            return Response(200, content=b"sync-in-async")

        async with AsyncClient(
            async_transport=MockTransport(handler)
        ) as client:
            resp = await client.get("http://testserver/")
            assert resp.content == b"sync-in-async"


class TestBuildResponse:
    def test_build_with_content(self):
        resp = _build_response(201, content=b"created")
        assert resp.status_code == 201
        assert resp.content == b"created"

    def test_build_with_text(self):
        resp = _build_response(200, text="hello")
        assert resp.text == "hello"

    def test_build_with_json(self):
        resp = _build_response(200, json={"a": 1})
        assert resp.json() == {"a": 1}

    def test_build_with_headers(self):
        resp = _build_response(200, headers={"x-test": "yes"})
        assert resp.headers["x-test"] == "yes"