Skip to content

Use pytestmark to simplify test suite #2334

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
May 14, 2024
Merged

Use pytestmark to simplify test suite #2334

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
78afc2d
Use pytestmark
Kludex May 14, 2024
775597f
Fix linter
Kludex May 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 19 additions & 21 deletions tests/middleware/test_logging.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

from tests.utils import run_server
from uvicorn import Config
from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope

if typing.TYPE_CHECKING:
import sys
Expand All @@ -27,9 +28,11 @@

WSProtocol: TypeAlias = "type[WebSocketProtocol | _WSProtocol]"

pytestmark = pytest.mark.anyio


@contextlib.contextmanager
def caplog_for_logger(caplog, logger_name):
def caplog_for_logger(caplog: pytest.LogCaptureFixture, logger_name: str) -> typing.Iterator[pytest.LogCaptureFixture]:
logger = logging.getLogger(logger_name)
logger.propagate, old_propagate = False, logger.propagate
logger.addHandler(caplog.handler)
Expand All @@ -40,14 +43,13 @@ def caplog_for_logger(caplog, logger_name):
logger.propagate = old_propagate


async def app(scope, receive, send):
async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
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_trace_logging(caplog, logging_config, unused_tcp_port: int):
async def test_trace_logging(caplog: pytest.LogCaptureFixture, logging_config, unused_tcp_port: int):
config = Config(
app=app,
log_level="trace",
Expand All @@ -69,7 +71,6 @@ async def test_trace_logging(caplog, logging_config, unused_tcp_port: int):
assert "ASGI [2] Completed" in messages.pop(0)


@pytest.mark.anyio
async def test_trace_logging_on_http_protocol(http_protocol_cls, caplog, logging_config, unused_tcp_port: int):
config = Config(
app=app,
Expand All @@ -88,14 +89,13 @@ async def test_trace_logging_on_http_protocol(http_protocol_cls, caplog, logging
assert any(" - HTTP connection lost" in message for message in messages)


@pytest.mark.anyio
async def test_trace_logging_on_ws_protocol(
ws_protocol_cls: WSProtocol,
caplog,
logging_config,
unused_tcp_port: int,
):
async def websocket_app(scope, receive, send):
async def websocket_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
assert scope["type"] == "websocket"
while True:
message = await receive()
Expand Down Expand Up @@ -125,9 +125,8 @@ async def open_connection(url):
assert any(" - WebSocket connection lost" in message for message in messages)


@pytest.mark.anyio
@pytest.mark.parametrize("use_colors", [(True), (False), (None)])
async def test_access_logging(use_colors, caplog, logging_config, unused_tcp_port: int):
async def test_access_logging(use_colors: bool, caplog: pytest.LogCaptureFixture, logging_config, unused_tcp_port: int):
config = Config(app=app, use_colors=use_colors, log_config=logging_config, port=unused_tcp_port)
with caplog_for_logger(caplog, "uvicorn.access"):
async with run_server(config):
Expand All @@ -139,9 +138,10 @@ async def test_access_logging(use_colors, caplog, logging_config, unused_tcp_por
assert '"GET / HTTP/1.1" 204' in messages.pop()


@pytest.mark.anyio
@pytest.mark.parametrize("use_colors", [(True), (False)])
async def test_default_logging(use_colors, caplog, logging_config, unused_tcp_port: int):
async def test_default_logging(
use_colors: bool, caplog: pytest.LogCaptureFixture, logging_config, unused_tcp_port: int
):
config = Config(app=app, use_colors=use_colors, log_config=logging_config, port=unused_tcp_port)
with caplog_for_logger(caplog, "uvicorn.access"):
async with run_server(config):
Expand All @@ -158,9 +158,10 @@ async def test_default_logging(use_colors, caplog, logging_config, unused_tcp_po
assert "Shutting down" in messages.pop(0)


@pytest.mark.anyio
@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
async def test_running_log_using_uds(caplog, short_socket_name, unused_tcp_port: int): # pragma: py-win32
async def test_running_log_using_uds(
caplog: pytest.LogCaptureFixture, short_socket_name: str, unused_tcp_port: int
): # pragma: py-win32
config = Config(app=app, uds=short_socket_name, port=unused_tcp_port)
with caplog_for_logger(caplog, "uvicorn.access"):
async with run_server(config):
Expand All @@ -170,9 +171,8 @@ async def test_running_log_using_uds(caplog, short_socket_name, unused_tcp_port:
assert f"Uvicorn running on unix socket {short_socket_name} (Press CTRL+C to quit)" in messages


@pytest.mark.anyio
@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
async def test_running_log_using_fd(caplog, unused_tcp_port: int): # pragma: py-win32
async def test_running_log_using_fd(caplog: pytest.LogCaptureFixture, unused_tcp_port: int): # pragma: py-win32
with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
fd = sock.fileno()
config = Config(app=app, fd=fd, port=unused_tcp_port)
Expand All @@ -184,9 +184,8 @@ async def test_running_log_using_fd(caplog, unused_tcp_port: int): # pragma: py
assert f"Uvicorn running on socket {sockname} (Press CTRL+C to quit)" in messages


@pytest.mark.anyio
async def test_unknown_status_code(caplog, unused_tcp_port: int):
async def app(scope, receive, send):
async def test_unknown_status_code(caplog: pytest.LogCaptureFixture, unused_tcp_port: int):
async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
assert scope["type"] == "http"
await send({"type": "http.response.start", "status": 599, "headers": []})
await send({"type": "http.response.body", "body": b"", "more_body": False})
Expand All @@ -202,11 +201,10 @@ async def app(scope, receive, send):
assert '"GET / HTTP/1.1" 599' in messages.pop()


@pytest.mark.anyio
async def test_server_start_with_port_zero(caplog: pytest.LogCaptureFixture):
config = Config(app=app, port=0)
async with run_server(config) as server:
server = server.servers[0]
async with run_server(config) as _server:
server = _server.servers[0]
sock = server.sockets[0]
host, port = sock.getsockname()
messages = [record.message for record in caplog.records if "uvicorn" in record.name]
Expand Down
Loading