forked from encode/uvicorn
/
test_main.py
147 lines (121 loc) · 4.51 KB
/
test_main.py
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
import asyncio
import inspect
import socket
from logging import WARNING
import httpx
import pytest
from tests.utils import run_server
from uvicorn import Server
from uvicorn.config import Config
from uvicorn.main import run
async def app(scope, receive, send):
assert scope["type"] == "http"
await send({"type": "http.response.start", "status": 204, "headers": []})
await send({"type": "http.response.body", "body": b"", "more_body": False})
@pytest.mark.anyio
async def test_return_close_header():
config = Config(app=app, host="localhost", loop="asyncio", limit_max_requests=1)
async with run_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(
"http://127.0.0.1:8000", headers={"connection": "close"}
)
assert response.status_code == 204
assert (
"connection" in response.headers and response.headers["connection"] == "close"
)
def _has_ipv6(host):
sock = None
has_ipv6 = False
if socket.has_ipv6:
try:
sock = socket.socket(socket.AF_INET6)
sock.bind((host, 0))
has_ipv6 = True
except Exception: # pragma: no cover
pass
if sock:
sock.close()
return has_ipv6
@pytest.mark.anyio
@pytest.mark.parametrize(
"host, url",
[
pytest.param(None, "http://127.0.0.1:8000", id="default"),
pytest.param("localhost", "http://127.0.0.1:8000", id="hostname"),
pytest.param(
"::1",
"http://[::1]:8000",
id="ipv6",
marks=pytest.mark.skipif(not _has_ipv6("::1"), reason="IPV6 not enabled"),
),
],
)
async def test_run(host, url):
config = Config(app=app, host=host, loop="asyncio", limit_max_requests=1)
async with run_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(url)
assert response.status_code == 204
@pytest.mark.anyio
async def test_run_multiprocess():
config = Config(app=app, loop="asyncio", workers=2, limit_max_requests=1)
async with run_server(config):
async with httpx.AsyncClient() as client:
response = await client.get("http://127.0.0.1:8000")
assert response.status_code == 204
@pytest.mark.anyio
async def test_run_reload():
config = Config(app=app, loop="asyncio", reload=True, limit_max_requests=1)
async with run_server(config):
async with httpx.AsyncClient() as client:
response = await client.get("http://127.0.0.1:8000")
assert response.status_code == 204
def test_run_invalid_app_config_combination(caplog: pytest.LogCaptureFixture) -> None:
with pytest.raises(SystemExit) as exit_exception:
run(app, reload=True)
assert exit_exception.value.code == 1
assert caplog.records[-1].name == "uvicorn.error"
assert caplog.records[-1].levelno == WARNING
assert caplog.records[-1].message == (
"You must pass the application as an import string to enable "
"'reload' or 'workers'."
)
def test_run_startup_failure(caplog: pytest.LogCaptureFixture) -> None:
async def app(scope, receive, send):
assert scope["type"] == "lifespan"
message = await receive()
if message["type"] == "lifespan.startup":
raise RuntimeError("Startup failed")
with pytest.raises(SystemExit) as exit_exception:
run(app, lifespan="on")
assert exit_exception.value.code == 3
def test_run_match_config_params() -> None:
config_params = {
key: repr(value)
for key, value in inspect.signature(Config.__init__).parameters.items()
if key not in ("self", "timeout_notify", "callback_notify")
}
run_params = {
key: repr(value)
for key, value in inspect.signature(run).parameters.items()
if key not in ("app_dir",)
}
assert config_params == run_params
@pytest.mark.anyio
async def test_run_multiprocess_with_sockets():
config = Config(app=app, workers=2, limit_max_requests=1)
with socket.socket() as sock:
sock.bind(("localhost", 0))
async with run_server(config, sockets=[sock]) as server:
while not server.started:
await asyncio.sleep(0.1)
await asyncio.sleep(0.1)
@pytest.mark.anyio
async def test_run_invalid_host() -> None:
with pytest.raises(SystemExit) as e:
config = Config(app=app, host="illegal_host")
server = Server(config=config)
await server.serve()
assert e.type == SystemExit
assert e.value.code == 1