Use pytestmark to simplify test suite (#2334)

* Use pytestmark

* Fix linter

Author: Kludex

tests/middleware/test_logging.py

@@ -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
@@ -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)
@@ -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",
@@ -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,
@@ -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()
@@ -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):
@@ -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):
@@ -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):
@@ -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)
@@ -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})
@@ -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]