Skip to content

Commit

Permalink
Add trace logging tests for protocols
Browse files Browse the repository at this point in the history
Protocols emit trace logs on `connection_made()` and `connection_lost`.
Add missing tests for these logs.
  • Loading branch information
laggardkernel committed Jun 19, 2021
1 parent c0b0095 commit 9904f7d
Showing 1 changed file with 49 additions and 0 deletions.
49 changes: 49 additions & 0 deletions tests/middleware/test_logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

import httpx
import pytest
import websockets

from tests.utils import run_server
from uvicorn import Config
Expand Down Expand Up @@ -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):
Expand Down

0 comments on commit 9904f7d

Please sign in to comment.