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
|
import inspect
import socket
from logging import WARNING
import httpx
import pytest
from tests.utils import run_server
from uvicorn import Server
from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
from uvicorn.config import Config
from uvicorn.main import run
pytestmark = pytest.mark.anyio
async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
assert scope["type"] == "http"
await send({"type": "http.response.start", "status": 204, "headers": []})
await send({"type": "http.response.body", "body": b"", "more_body": False})
def _has_ipv6(host: str):
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.parametrize(
"host, url",
[
pytest.param(None, "http://127.0.0.1", id="default"),
pytest.param("localhost", "http://127.0.0.1", id="hostname"),
pytest.param(
"::1",
"http://[::1]",
id="ipv6",
marks=pytest.mark.skipif(not _has_ipv6("::1"), reason="IPV6 not enabled"),
),
],
)
async def test_run(host, url: str, unused_tcp_port: int):
config = Config(app=app, host=host, loop="asyncio", limit_max_requests=1, port=unused_tcp_port)
async with run_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"{url}:{unused_tcp_port}")
assert response.status_code == 204
async def test_run_multiprocess(unused_tcp_port: int):
config = Config(app=app, loop="asyncio", workers=2, limit_max_requests=1, port=unused_tcp_port)
async with run_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
assert response.status_code == 204
async def test_run_reload(unused_tcp_port: int):
config = Config(app=app, loop="asyncio", reload=True, limit_max_requests=1, port=unused_tcp_port)
async with run_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
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
async def test_exit_on_create_server_with_invalid_host() -> None:
with pytest.raises(SystemExit) as exc_info:
config = Config(app=app, host="illegal_host")
server = Server(config=config)
await server.serve()
assert exc_info.value.code == 1
|
