Change ruff rules (#2251)

* Change ruff rules

* fix type checker

Author: Kludex

pyproject.toml

@@ -61,7 +61,8 @@ path = "uvicorn/__init__.py"
 include = ["/uvicorn"]
 
 [tool.ruff]
-select = ["E", "F", "I"]
+line-length = 120
+select = ["E", "F", "I", "FA", "UP"]
 ignore = ["B904", "B028"]
 
 [tool.ruff.lint.isort]

tests/conftest.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import contextlib
 import importlib.util
 import os
@@ -9,6 +11,7 @@
 from tempfile import TemporaryDirectory
 from threading import Thread
 from time import sleep
+from typing import Any
 from uuid import uuid4
 
 import pytest
@@ -38,14 +41,14 @@
 
 
 @pytest.fixture
-def tls_certificate_authority() -> "trustme.CA":
+def tls_certificate_authority() -> trustme.CA:
     if not HAVE_TRUSTME:
         pytest.skip("trustme not installed")  # pragma: no cover
     return trustme.CA()
 
 
 @pytest.fixture
-def tls_certificate(tls_certificate_authority: "trustme.CA") -> "trustme.LeafCert":
+def tls_certificate(tls_certificate_authority: trustme.CA) -> trustme.LeafCert:
     return tls_certificate_authority.issue_cert(
         "localhost",
         "127.0.0.1",
@@ -54,13 +57,13 @@ def tls_certificate(tls_certificate_authority: "trustme.CA") -> "trustme.LeafCer
 
 
 @pytest.fixture
-def tls_ca_certificate_pem_path(tls_certificate_authority: "trustme.CA"):
+def tls_ca_certificate_pem_path(tls_certificate_authority: trustme.CA):
     with tls_certificate_authority.cert_pem.tempfile() as ca_cert_pem:
         yield ca_cert_pem
 
 
 @pytest.fixture
-def tls_ca_certificate_private_key_path(tls_certificate_authority: "trustme.CA"):
+def tls_ca_certificate_private_key_path(tls_certificate_authority: trustme.CA):
     with tls_certificate_authority.private_key_pem.tempfile() as private_key:
         yield private_key
 
@@ -82,25 +85,25 @@ def tls_certificate_private_key_encrypted_path(tls_certificate):
 
 
 @pytest.fixture
-def tls_certificate_private_key_path(tls_certificate: "trustme.CA"):
+def tls_certificate_private_key_path(tls_certificate: trustme.CA):
     with tls_certificate.private_key_pem.tempfile() as private_key:
         yield private_key
 
 
 @pytest.fixture
-def tls_certificate_key_and_chain_path(tls_certificate: "trustme.LeafCert"):
+def tls_certificate_key_and_chain_path(tls_certificate: trustme.LeafCert):
     with tls_certificate.private_key_and_cert_chain_pem.tempfile() as cert_pem:
         yield cert_pem
 
 
 @pytest.fixture
-def tls_certificate_server_cert_path(tls_certificate: "trustme.LeafCert"):
+def tls_certificate_server_cert_path(tls_certificate: trustme.LeafCert):
     with tls_certificate.cert_chain_pems[0].tempfile() as cert_pem:
         yield cert_pem
 
 
 @pytest.fixture
-def tls_ca_ssl_context(tls_certificate_authority: "trustme.CA") -> ssl.SSLContext:
+def tls_ca_ssl_context(tls_certificate_authority: trustme.CA) -> ssl.SSLContext:
     ssl_ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
     tls_certificate_authority.configure_trust(ssl_ctx)
     return ssl_ctx
@@ -172,7 +175,7 @@ def anyio_backend() -> str:
 
 
 @pytest.fixture(scope="function")
-def logging_config() -> dict:
+def logging_config() -> dict[str, Any]:
     return deepcopy(LOGGING_CONFIG)
 
 
@@ -250,9 +253,7 @@ def unused_tcp_port() -> int:
     params=[
         pytest.param(
             "uvicorn.protocols.websockets.wsproto_impl:WSProtocol",
-            marks=pytest.mark.skipif(
-                not importlib.util.find_spec("wsproto"), reason="wsproto not installed."
-            ),
+            marks=pytest.mark.skipif(not importlib.util.find_spec("wsproto"), reason="wsproto not installed."),
             id="wsproto",
         ),
         pytest.param(

tests/middleware/test_logging.py

@@ -60,9 +60,7 @@ async def test_trace_logging(caplog, logging_config, unused_tcp_port: int):
             async with httpx.AsyncClient() as client:
                 response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
         assert response.status_code == 204
-        messages = [
-            record.message for record in caplog.records if record.name == "uvicorn.asgi"
-        ]
+        messages = [record.message for record in caplog.records if record.name == "uvicorn.asgi"]
         assert "ASGI [1] Started scope=" in messages.pop(0)
         assert "ASGI [1] Raised exception" in messages.pop(0)
         assert "ASGI [2] Started scope=" in messages.pop(0)
@@ -72,9 +70,7 @@ async def test_trace_logging(caplog, logging_config, unused_tcp_port: int):
 
 
 @pytest.mark.anyio
-async def test_trace_logging_on_http_protocol(
-    http_protocol_cls, caplog, logging_config, unused_tcp_port: int
-):
+async def test_trace_logging_on_http_protocol(http_protocol_cls, caplog, logging_config, unused_tcp_port: int):
     config = Config(
         app=app,
         log_level="trace",
@@ -87,11 +83,7 @@ async def test_trace_logging_on_http_protocol(
             async with httpx.AsyncClient() as client:
                 response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
         assert response.status_code == 204
-        messages = [
-            record.message
-            for record in caplog.records
-            if record.name == "uvicorn.error"
-        ]
+        messages = [record.message for record in caplog.records if record.name == "uvicorn.error"]
         assert any(" - HTTP connection made" in message for message in messages)
         assert any(" - HTTP connection lost" in message for message in messages)
 
@@ -127,11 +119,7 @@ async def open_connection(url):
         async with run_server(config):
             is_open = await open_connection(f"ws://127.0.0.1:{unused_tcp_port}")
         assert is_open
-        messages = [
-            record.message
-            for record in caplog.records
-            if record.name == "uvicorn.error"
-        ]
+        messages = [record.message for record in caplog.records if record.name == "uvicorn.error"]
         assert any(" - Upgrading to WebSocket" in message for message in messages)
         assert any(" - WebSocket connection made" in message for message in messages)
         assert any(" - WebSocket connection lost" in message for message in messages)
@@ -140,39 +128,27 @@ async def open_connection(url):
 @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):
-    config = Config(
-        app=app, use_colors=use_colors, log_config=logging_config, port=unused_tcp_port
-    )
+    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):
             async with httpx.AsyncClient() as client:
                 response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
 
         assert response.status_code == 204
-        messages = [
-            record.message
-            for record in caplog.records
-            if record.name == "uvicorn.access"
-        ]
+        messages = [record.message for record in caplog.records if record.name == "uvicorn.access"]
         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
-):
-    config = Config(
-        app=app, use_colors=use_colors, log_config=logging_config, port=unused_tcp_port
-    )
+async def test_default_logging(use_colors, caplog, 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):
             async with httpx.AsyncClient() as client:
                 response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
         assert response.status_code == 204
-        messages = [
-            record.message for record in caplog.records if "uvicorn" in record.name
-        ]
+        messages = [record.message for record in caplog.records if "uvicorn" in record.name]
         assert "Started server process" in messages.pop(0)
         assert "Waiting for application startup" in messages.pop(0)
         assert "ASGI 'lifespan' protocol appears unsupported" in messages.pop(0)
@@ -184,19 +160,14 @@ async def test_default_logging(
 
 @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, short_socket_name, 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):
             ...
 
     messages = [record.message for record in caplog.records if "uvicorn" in record.name]
-    assert (
-        f"Uvicorn running on unix socket {short_socket_name} (Press CTRL+C to quit)"
-        in messages
-    )
+    assert f"Uvicorn running on unix socket {short_socket_name} (Press CTRL+C to quit)" in messages
 
 
 @pytest.mark.anyio
@@ -227,11 +198,7 @@ async def app(scope, receive, send):
                 response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
 
         assert response.status_code == 599
-        messages = [
-            record.message
-            for record in caplog.records
-            if record.name == "uvicorn.access"
-        ]
+        messages = [record.message for record in caplog.records if record.name == "uvicorn.access"]
         assert '"GET / HTTP/1.1" 599' in messages.pop()
 
 

tests/middleware/test_message_logger.py

@@ -26,9 +26,7 @@ async def app(scope, receive, send):
         assert sum(["ASGI [1] Send" in message for message in messages]) == 2
         assert sum(["ASGI [1] Receive" in message for message in messages]) == 1
         assert sum(["ASGI [1] Completed" in message for message in messages]) == 1
-        assert (
-            sum(["ASGI [1] Raised exception" in message for message in messages]) == 0
-        )
+        assert sum(["ASGI [1] Raised exception" in message for message in messages]) == 0
 
 
 @pytest.mark.anyio
@@ -48,6 +46,4 @@ async def app(scope, receive, send):
         assert sum(["ASGI [1] Send" in message for message in messages]) == 0
         assert sum(["ASGI [1] Receive" in message for message in messages]) == 0
         assert sum(["ASGI [1] Completed" in message for message in messages]) == 0
-        assert (
-            sum(["ASGI [1] Raised exception" in message for message in messages]) == 1
-        )
+        assert sum(["ASGI [1] Raised exception" in message for message in messages]) == 1

tests/middleware/test_proxy_headers.py

@@ -1,4 +1,6 @@
-from typing import TYPE_CHECKING, List, Type, Union
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 import httpx
 import pytest
@@ -45,13 +47,9 @@ async def app(
         ("192.168.0.1", "Remote: http://127.0.0.1:123"),
     ],
 )
-async def test_proxy_headers_trusted_hosts(
-    trusted_hosts: Union[List[str], str], response_text: str
-) -> None:
+async def test_proxy_headers_trusted_hosts(trusted_hosts: list[str] | str, response_text: str) -> None:
     app_with_middleware = ProxyHeadersMiddleware(app, trusted_hosts=trusted_hosts)
-    async with httpx.AsyncClient(
-        app=app_with_middleware, base_url="http://testserver"
-    ) as client:
+    async with httpx.AsyncClient(app=app_with_middleware, base_url="http://testserver") as client:
         headers = {"X-Forwarded-Proto": "https", "X-Forwarded-For": "1.2.3.4"}
         response = await client.get("/", headers=headers)
 
@@ -79,13 +77,9 @@ async def test_proxy_headers_trusted_hosts(
         (["192.168.0.2", "127.0.0.1"], "Remote: https://10.0.2.1:0"),
     ],
 )
-async def test_proxy_headers_multiple_proxies(
-    trusted_hosts: Union[List[str], str], response_text: str
-) -> None:
+async def test_proxy_headers_multiple_proxies(trusted_hosts: list[str] | str, response_text: str) -> None:
     app_with_middleware = ProxyHeadersMiddleware(app, trusted_hosts=trusted_hosts)
-    async with httpx.AsyncClient(
-        app=app_with_middleware, base_url="http://testserver"
-    ) as client:
+    async with httpx.AsyncClient(app=app_with_middleware, base_url="http://testserver") as client:
         headers = {
             "X-Forwarded-Proto": "https",
             "X-Forwarded-For": "1.2.3.4, 10.0.2.1, 192.168.0.2",
@@ -99,9 +93,7 @@ async def test_proxy_headers_multiple_proxies(
 @pytest.mark.anyio
 async def test_proxy_headers_invalid_x_forwarded_for() -> None:
     app_with_middleware = ProxyHeadersMiddleware(app, trusted_hosts="*")
-    async with httpx.AsyncClient(
-        app=app_with_middleware, base_url="http://testserver"
-    ) as client:
+    async with httpx.AsyncClient(app=app_with_middleware, base_url="http://testserver") as client:
         headers = httpx.Headers(
             {
                 "X-Forwarded-Proto": "https",
@@ -127,12 +119,14 @@ async def test_proxy_headers_invalid_x_forwarded_for() -> None:
 async def test_proxy_headers_websocket_x_forwarded_proto(
     x_forwarded_proto: str,
     addr: str,
-    ws_protocol_cls: "Type[WSProtocol | WebSocketProtocol]",
-    http_protocol_cls: "Type[H11Protocol | HttpToolsProtocol]",
+    ws_protocol_cls: type[WSProtocol | WebSocketProtocol],
+    http_protocol_cls: type[H11Protocol | HttpToolsProtocol],
     unused_tcp_port: int,
 ) -> None:
-    async def websocket_app(scope, receive, send):
+    async def websocket_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
+        assert scope["type"] == "websocket"
         scheme = scope["scheme"]
+        assert scope["client"] is not None
         host, port = scope["client"]
         addr = "%s://%s:%d" % (scheme, host, port)
         await send({"type": "websocket.accept"})

tests/middleware/test_wsgi.py

@@ -1,6 +1,8 @@
+from __future__ import annotations
+
 import io
 import sys
-from typing import AsyncGenerator, Callable, List
+from typing import AsyncGenerator, Callable
 
 import a2wsgi
 import httpx
@@ -10,7 +12,7 @@
 from uvicorn.middleware import wsgi
 
 
-def hello_world(environ: Environ, start_response: StartResponse) -> List[bytes]:
+def hello_world(environ: Environ, start_response: StartResponse) -> list[bytes]:
     status = "200 OK"
     output = b"Hello World!\n"
     headers = [
@@ -21,7 +23,7 @@ def hello_world(environ: Environ, start_response: StartResponse) -> List[bytes]:
     return [output]
 
 
-def echo_body(environ: Environ, start_response: StartResponse) -> List[bytes]:
+def echo_body(environ: Environ, start_response: StartResponse) -> list[bytes]:
     status = "200 OK"
     output = environ["wsgi.input"].read()
     headers = [
@@ -32,11 +34,11 @@ def echo_body(environ: Environ, start_response: StartResponse) -> List[bytes]:
     return [output]
 
 
-def raise_exception(environ: Environ, start_response: StartResponse) -> List[bytes]:
+def raise_exception(environ: Environ, start_response: StartResponse) -> list[bytes]:
     raise RuntimeError("Something went wrong")
 
 
-def return_exc_info(environ: Environ, start_response: StartResponse) -> List[bytes]:
+def return_exc_info(environ: Environ, start_response: StartResponse) -> list[bytes]:
     try:
         raise RuntimeError("Something went wrong")
     except RuntimeError:
@@ -110,16 +112,14 @@ async def test_wsgi_exc_info(wsgi_middleware: Callable) -> None:
         app=app,
         raise_app_exceptions=False,
     )
-    async with httpx.AsyncClient(
-        transport=transport, base_url="http://testserver"
-    ) as client:
+    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
         response = await client.get("/")
     assert response.status_code == 500
     assert response.text == "Internal Server Error"
 
 
 def test_build_environ_encoding() -> None:
-    scope: "HTTPScope" = {
+    scope: HTTPScope = {
         "asgi": {"version": "3.0", "spec_version": "2.0"},
         "scheme": "http",
         "raw_path": b"/\xe6\x96\x87%2Fall",
@@ -134,12 +134,12 @@ def test_build_environ_encoding() -> None:
         "headers": [(b"key", b"value1"), (b"key", b"value2")],
         "extensions": {},
     }
-    message: "HTTPRequestEvent" = {
+    message: HTTPRequestEvent = {
         "type": "http.request",
         "body": b"",
         "more_body": False,
     }
     environ = wsgi.build_environ(scope, message, io.BytesIO(b""))
-    assert environ["SCRIPT_NAME"] == "/文".encode("utf8").decode("latin-1")
-    assert environ["PATH_INFO"] == "/all".encode("utf8").decode("latin-1")
+    assert environ["SCRIPT_NAME"] == "/文".encode().decode("latin-1")
+    assert environ["PATH_INFO"] == b"/all".decode("latin-1")
     assert environ["HTTP_KEY"] == "value1,value2"