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
|
from __future__ import annotations
import asyncio
import contextlib
import logging
import signal
import sys
from typing import Callable, ContextManager, Generator
import httpx
import pytest
from tests.utils import run_server
from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
from uvicorn.config import Config
from uvicorn.protocols.http.h11_impl import H11Protocol
from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
from uvicorn.server import Server
pytestmark = pytest.mark.anyio
# asyncio does NOT allow raising in signal handlers, so to detect
# raised signals raised a mutable `witness` receives the signal
@contextlib.contextmanager
def capture_signal_sync(sig: signal.Signals) -> Generator[list[int], None, None]:
"""Replace `sig` handling with a normal exception via `signal"""
witness: list[int] = []
original_handler = signal.signal(sig, lambda signum, frame: witness.append(signum))
yield witness
signal.signal(sig, original_handler)
@contextlib.contextmanager
def capture_signal_async(sig: signal.Signals) -> Generator[list[int], None, None]: # pragma: py-win32
"""Replace `sig` handling with a normal exception via `asyncio"""
witness: list[int] = []
original_handler = signal.getsignal(sig)
asyncio.get_running_loop().add_signal_handler(sig, witness.append, sig)
yield witness
signal.signal(sig, original_handler)
async def dummy_app(scope, receive, send): # pragma: py-win32
pass
async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
assert scope["type"] == "http"
await send({"type": "http.response.start", "status": 200, "headers": []})
await send({"type": "http.response.body", "body": b"", "more_body": False})
if sys.platform == "win32": # pragma: py-not-win32
signals = [signal.SIGBREAK]
signal_captures = [capture_signal_sync]
else: # pragma: py-win32
signals = [signal.SIGTERM, signal.SIGINT]
signal_captures = [capture_signal_sync, capture_signal_async]
@pytest.mark.parametrize("exception_signal", signals)
@pytest.mark.parametrize("capture_signal", signal_captures)
async def test_server_interrupt(
exception_signal: signal.Signals, capture_signal: Callable[[signal.Signals], ContextManager[None]]
): # pragma: py-win32
"""Test interrupting a Server that is run explicitly inside asyncio"""
async def interrupt_running(srv: Server):
while not srv.started:
await asyncio.sleep(0.01)
signal.raise_signal(exception_signal)
server = Server(Config(app=dummy_app, loop="asyncio"))
asyncio.create_task(interrupt_running(server))
with capture_signal(exception_signal) as witness:
await server.serve()
assert witness
# set by the server's graceful exit handler
assert server.should_exit
async def test_request_than_limit_max_requests_warn_log(
unused_tcp_port: int, http_protocol_cls: type[H11Protocol | HttpToolsProtocol], caplog: pytest.LogCaptureFixture
):
caplog.set_level(logging.WARNING, logger="uvicorn.error")
config = Config(app=app, limit_max_requests=1, port=unused_tcp_port, http=http_protocol_cls)
async with run_server(config):
async with httpx.AsyncClient() as client:
tasks = [client.get(f"http://127.0.0.1:{unused_tcp_port}") for _ in range(2)]
responses = await asyncio.gather(*tasks)
assert len(responses) == 2
assert "Maximum request limit of 1 exceeded. Terminating process." in caplog.text
|
