httpcore/tests/client/test_async_client.py at 6845d95e5d09a19f3f4a1f0a0bf8856ff780df65 · taoufik07/httpcore · 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
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
import pytest

import httpcore


@pytest.mark.asyncio
async def test_get(server):
    url = "http://127.0.0.1:8000/"
    async with httpcore.AsyncClient() as client:
        response = await client.get(url)
    assert response.status_code == 200
    assert response.text == "Hello, world!"
    assert response.protocol == "HTTP/1.1"
    assert response.headers
    assert repr(response) == "<Response(200, 'OK')>"


@pytest.mark.asyncio
async def test_get_base_url(server):
    base_url = "http://127.0.0.1:8000/"
    async with httpcore.AsyncClient(base_url=base_url) as client:
        response = await client.get("/hello")
    assert response.status_code == 200
    assert response.text == "Hello, world!"
    assert str(response.url) == "http://127.0.0.1:8000/hello"


@pytest.mark.asyncio
async def test_post(server):
    url = "http://127.0.0.1:8000/"
    async with httpcore.AsyncClient() as client:
        response = await client.post(url, data=b"Hello, world!")
    assert response.status_code == 200


@pytest.mark.asyncio
async def test_stream_response(server):
    async with httpcore.AsyncClient() as client:
        response = await client.request("GET", "http://127.0.0.1:8000/", stream=True)
    assert response.status_code == 200
    body = await response.read()
    assert body == b"Hello, world!"
    assert response.content == b"Hello, world!"


@pytest.mark.asyncio
async def test_access_content_stream_response(server):
    async with httpcore.AsyncClient() as client:
        response = await client.request("GET", "http://127.0.0.1:8000/", stream=True)
    assert response.status_code == 200
    with pytest.raises(httpcore.ResponseNotRead):
        response.content


@pytest.mark.asyncio
async def test_stream_request(server):
    async def hello_world():
        yield b"Hello, "
        yield b"world!"

    async with httpcore.AsyncClient() as client:
        response = await client.request(
            "POST", "http://127.0.0.1:8000/", data=hello_world()
        )
    assert response.status_code == 200


@pytest.mark.asyncio
async def test_raise_for_status(server):
    async with httpcore.AsyncClient() as client:
        for status_code in (200, 400, 404, 500, 505):
            response = await client.request(
                "GET", f"http://127.0.0.1:8000/status/{status_code}"
            )

            if 400 <= status_code < 600:
                with pytest.raises(httpcore.exceptions.HttpError):
                    response.raise_for_status()
            else:
                assert response.raise_for_status() is None


@pytest.mark.asyncio
async def test_options(server):
    url = "http://127.0.0.1:8000/"
    async with httpcore.AsyncClient() as client:
        response = await client.options(url)
    assert response.status_code == 200
    assert response.text == "Hello, world!"


@pytest.mark.asyncio
async def test_head(server):
    url = "http://127.0.0.1:8000/"
    async with httpcore.AsyncClient() as client:
        response = await client.head(url)
    assert response.status_code == 200
    assert response.text == ""


@pytest.mark.asyncio
async def test_put(server):
    url = "http://127.0.0.1:8000/"
    async with httpcore.AsyncClient() as client:
        response = await client.put(url, data=b"Hello, world!")
    assert response.status_code == 200


@pytest.mark.asyncio
async def test_patch(server):
    url = "http://127.0.0.1:8000/"
    async with httpcore.AsyncClient() as client:
        response = await client.patch(url, data=b"Hello, world!")
    assert response.status_code == 200


@pytest.mark.asyncio
async def test_delete(server):
    url = "http://127.0.0.1:8000/"
    async with httpcore.AsyncClient() as client:
        response = await client.delete(url)
    assert response.status_code == 200
    assert response.text == "Hello, world!"