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
|
import pytest
import httpx
@pytest.mark.anyio
async def test_read_timeout(server):
timeout = httpx.Timeout(None, read=1e-6)
async with httpx.AsyncClient(timeout=timeout) as client:
with pytest.raises(httpx.ReadTimeout):
await client.get(server.url.copy_with(path="/slow_response"))
@pytest.mark.anyio
async def test_write_timeout(server):
timeout = httpx.Timeout(None, write=1e-6)
async with httpx.AsyncClient(timeout=timeout) as client:
with pytest.raises(httpx.WriteTimeout):
data = b"*" * 1024 * 1024 * 100
await client.put(server.url.copy_with(path="/slow_response"), content=data)
@pytest.mark.anyio
@pytest.mark.network
async def test_connect_timeout(server):
timeout = httpx.Timeout(None, connect=1e-6)
async with httpx.AsyncClient(timeout=timeout) as client:
with pytest.raises(httpx.ConnectTimeout):
# See https://stackoverflow.com/questions/100841/
await client.get("http://10.255.255.1/")
@pytest.mark.anyio
async def test_pool_timeout(server):
limits = httpx.Limits(max_connections=1)
timeout = httpx.Timeout(None, pool=1e-4)
async with httpx.AsyncClient(limits=limits, timeout=timeout) as client:
with pytest.raises(httpx.PoolTimeout):
async with client.stream("GET", server.url):
await client.get(server.url)
@pytest.mark.anyio
async def test_async_client_new_request_send_timeout(server):
timeout = httpx.Timeout(1e-6)
async with httpx.AsyncClient(timeout=timeout) as client:
with pytest.raises(httpx.TimeoutException):
await client.send(
httpx.Request("GET", server.url.copy_with(path="/slow_response"))
)
|
