httpx/tests/client/test_network_backend.py at 0ec032d724df25cee80453626bb0450ea3d59db0 · encode/httpx · GitHub

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
import typing

import httpcore
import pytest

import httpx


def test_network_backend():
    class Backend(httpcore.NetworkBackend):
        def connect_tcp(
            self,
            host: str,
            port: int,
            timeout: typing.Optional[float] = None,
            local_address: typing.Optional[str] = None,
            socket_options: typing.Optional[
                typing.Iterable[httpcore.SOCKET_OPTION]
            ] = None,
        ) -> httpcore.NetworkStream:
            return Stream()

    class Stream(httpcore.NetworkStream):
        body = b"\r\n".join(
            [
                b"HTTP/1.1 200 OK",
                b"",
                b"From Backend!",
            ]
        )

        def read(self, max_bytes: int, timeout: typing.Optional[float] = None) -> bytes:
            body = self.body
            if body:
                self.body = b""
            return body

        def write(self, buffer: bytes, timeout: typing.Optional[float] = None) -> None:
            pass

        def close(self) -> None:
            pass

    backend = Backend()
    transport = httpx.HTTPTransport(network_backend=backend)
    with httpx.Client(transport=transport) as client:
        response = client.get("http://www.example.org")
    assert response.status_code == 200
    assert response.text == "From Backend!"


@pytest.mark.anyio
async def test_async_network_backend():
    class AsyncBackend(httpcore.AsyncNetworkBackend):
        async def connect_tcp(
            self,
            host: str,
            port: int,
            timeout: typing.Optional[float] = None,
            local_address: typing.Optional[str] = None,
            socket_options: typing.Optional[
                typing.Iterable[httpcore.SOCKET_OPTION]
            ] = None,
        ) -> httpcore.AsyncNetworkStream:
            return AsyncStream()

    class AsyncStream(httpcore.AsyncNetworkStream):
        body = b"\r\n".join(
            [
                b"HTTP/1.1 200 OK",
                b"",
                b"From Async Backend!",
            ]
        )

        async def read(
            self, max_bytes: int, timeout: typing.Optional[float] = None
        ) -> bytes:
            body = self.body
            if body:
                self.body = b""
            return body

        async def write(
            self, buffer: bytes, timeout: typing.Optional[float] = None
        ) -> None:
            pass

        async def aclose(self) -> None:
            pass

    backend = AsyncBackend()
    transport = httpx.AsyncHTTPTransport(network_backend=backend)
    async with httpx.AsyncClient(transport=transport) as client:
        response = await client.get("http://www.example.org")
    assert response.status_code == 200
    assert response.text == "From Async Backend!"