diff --git a/tests/middleware/test_logging.py b/tests/middleware/test_logging.py index 0434b9f8b4..62fb8f626e 100644 --- a/tests/middleware/test_logging.py +++ b/tests/middleware/test_logging.py @@ -3,6 +3,7 @@ import httpx import pytest +import websockets from tests.utils import run_server from uvicorn import Config @@ -45,6 +46,54 @@ async def test_trace_logging(caplog): assert "ASGI [2] Completed" in messages.pop(0) +@pytest.mark.asyncio +@pytest.mark.parametrize("protocol", [("h11"), ("httptools")]) +async def test_trace_logging_on_http_protocol(protocol, caplog): + config = Config(app=app, log_level="trace", http=protocol) + with caplog_for_logger(caplog, "uvicorn.protocol"): + 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 + messages = [ + record.message + for record in caplog.records + if record.name == "uvicorn.error" + ] + assert any(" - Connection made" in message for message in messages) + assert any(" - Connection lost" in message for message in messages) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("protocol", [("websockets"), ("wsproto")]) +async def test_trace_logging_on_ws_protocol(protocol, caplog): + async def websocket_app(scope, receive, send): + assert scope["type"] == "websocket" + while True: + message = await receive() + if message["type"] == "websocket.connect": + await send({"type": "websocket.accept"}) + elif message["type"] == "websocket.disconnect": + break + + async def open_connection(url): + async with websockets.connect(url) as websocket: + return websocket.open + + config = Config(app=websocket_app, log_level="trace", ws=protocol) + with caplog_for_logger(caplog, "uvicorn.protocol"): + async with run_server(config): + is_open = await open_connection("ws://127.0.0.1:8000") + assert is_open + messages = [ + record.message + for record in caplog.records + if record.name == "uvicorn.error" + ] + assert any(" - Connection made" in message for message in messages) + assert any(" - Connection lost" in message for message in messages) + + @pytest.mark.asyncio @pytest.mark.parametrize("use_colors", [(True), (False), (None)]) async def test_access_logging(use_colors, caplog):