Mypy uvicorn

requirements.txt

@@ -24,6 +24,7 @@ pytest
 pytest-cov
 requests
 seed-isort-config
+mypy
 
 # Documentation
 mkdocs

scripts/check

@@ -10,4 +10,5 @@ set -x
 
 ${PREFIX}black --check --diff --target-version=py36 $SOURCE_FILES
 ${PREFIX}flake8 $SOURCE_FILES
+${PREFIX}mypy $SOURCE_FILES
 ${PREFIX}isort --check --diff --project=uvicorn $SOURCE_FILES

setup.cfg

@@ -14,7 +14,7 @@ check_untyped_defs = True
 profile = black
 combine_as_imports = True
 known_first_party = uvicorn,tests
-known_third_party = click,does_not_exist,gunicorn,h11,httptools,pytest,requests,setuptools,urllib3,uvloop,watchgod,websockets,wsproto
+known_third_party = _pytest,click,does_not_exist,gunicorn,h11,httptools,pytest,requests,setuptools,typing_extensions,urllib3,uvloop,watchgod,websockets,wsproto
 
 [tool:pytest]
 addopts = --cov=uvicorn --cov=tests -rxXs

tests/client.py

@@ -5,6 +5,8 @@
 
 import requests
 
+from uvicorn._types import HTTPSendMessage
+
 
 class _HeaderDict(requests.packages.urllib3._collections.HTTPHeaderDict):
     def get_all(self, key, default):
@@ -75,7 +77,10 @@ async def receive():
                 body_bytes = body
             return {"type": "http.request", "body": body_bytes}
 
-        async def send(message):
+        raw_kwargs: typing.Dict[str, typing.Any] = {"body": io.BytesIO()}
+        response_started = False
+
+        async def send(message: HTTPSendMessage):
             nonlocal raw_kwargs, response_started
 
             if message["type"] == "http.response.start":
@@ -96,9 +101,6 @@ async def send(message):
                 if not more_body:
                     raw_kwargs["body"].seek(0)
 
-        response_started = False
-        raw_kwargs = {"body": io.BytesIO()}
-
         loop = asyncio.get_event_loop()
 
         try:
@@ -122,7 +124,8 @@ def __init__(
         self.headers.update({"user-agent": "testclient"})
         self.base_url = base_url
 
-    def request(self, method: str, url: str, **kwargs) -> requests.Response:
+    def request(self, *args, **kwargs) -> requests.Response:
+        method, url = args
         url = urljoin(self.base_url, url)
         return super().request(method, url, **kwargs)
 

tests/middleware/test_debug.py

@@ -3,11 +3,12 @@
 import pytest
 
 from tests.client import TestClient
+from uvicorn._types import ASGI3App, Receive, Scope, Send
 from uvicorn.middleware.debug import DebugMiddleware
 
 
-def test_debug_text():
-    async def app(scope, receive, send):
+def test_debug_text() -> None:
+    async def app(scope: Scope, receive: Receive, send: Send) -> ASGI3App:
         raise RuntimeError("Something went wrong")
 
     app = DebugMiddleware(app)
@@ -18,8 +19,8 @@ async def app(scope, receive, send):
     assert "RuntimeError" in response.text
 
 
-def test_debug_html():
-    async def app(scope, receive, send):
+def test_debug_html() -> None:
+    async def app(scope: Scope, receive: Receive, send: Send) -> ASGI3App:
         raise RuntimeError("Something went wrong")
 
     app = DebugMiddleware(app)
@@ -30,8 +31,8 @@ async def app(scope, receive, send):
     assert "RuntimeError" in response.text
 
 
-def test_debug_after_response_sent():
-    async def app(scope, receive, send):
+def test_debug_after_response_sent() -> None:
+    async def app(scope: Scope, receive: Receive, send: Send) -> ASGI3App:
         await send({"type": "http.response.start", "status": 204, "headers": []})
         await send({"type": "http.response.body", "body": b"", "more_body": False})
         raise RuntimeError("Something went wrong")
@@ -43,8 +44,8 @@ async def app(scope, receive, send):
     assert response.content == b""
 
 
-def test_debug_not_http():
-    async def app(scope, send, receive):
+def test_debug_not_http() -> None:
+    async def app(scope: Scope, receive: Receive, send: Send) -> ASGI3App:
         raise RuntimeError("Something went wrong")
 
     app = DebugMiddleware(app)

tests/middleware/test_message_logger.py

@@ -1,13 +1,15 @@
 import pytest
+from _pytest.logging import LogCaptureFixture
 
 from tests.client import TestClient
+from uvicorn._types import ASGI3App, Receive, Scope, Send
 from uvicorn.middleware.message_logger import MessageLoggerMiddleware
 
 TRACE_LOG_LEVEL = 5
 
 
-def test_message_logger(caplog):
-    async def app(scope, receive, send):
+def test_message_logger(caplog: LogCaptureFixture) -> None:
+    async def app(scope: Scope, receive: Receive, send: Send) -> ASGI3App:
         await receive()
         await send({"type": "http.response.start", "status": 200, "headers": []})
         await send({"type": "http.response.body", "body": b"", "more_body": False})
@@ -27,8 +29,8 @@ async def app(scope, receive, send):
     assert sum(["ASGI [1] Raised exception" in message for message in messages]) == 0
 
 
-def test_message_logger_exc(caplog):
-    async def app(scope, receive, send):
+def test_message_logger_exc(caplog: LogCaptureFixture) -> None:
+    async def app(scope: Scope, receive: Receive, send: Send) -> ASGI3App:
         raise RuntimeError()
 
     caplog.set_level(TRACE_LOG_LEVEL, logger="uvicorn.asgi")

tests/middleware/test_proxy_headers.py

@@ -1,9 +1,10 @@
 from tests.client import TestClient
 from tests.response import Response
+from uvicorn._types import Receive, Scope, Send
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 
 
-async def app(scope, receive, send):
+async def app(scope: Scope, receive: Receive, send: Send) -> None:
     scheme = scope["scheme"]
     host, port = scope["client"]
     addr = "%s://%s:%d" % (scheme, host, port)
@@ -14,15 +15,15 @@ async def app(scope, receive, send):
 app = ProxyHeadersMiddleware(app, trusted_hosts="*")
 
 
-def test_proxy_headers():
+def test_proxy_headers() -> None:
     client = TestClient(app)
     headers = {"X-Forwarded-Proto": "https", "X-Forwarded-For": "1.2.3.4"}
     response = client.get("/", headers=headers)
     assert response.status_code == 200
     assert response.text == "Remote: https://1.2.3.4:0"
 
 
-def test_proxy_headers_no_port():
+def test_proxy_headers_no_port() -> None:
     client = TestClient(app)
     headers = {"X-Forwarded-Proto": "https", "X-Forwarded-For": "1.2.3.4"}
     response = client.get("/", headers=headers)

tests/middleware/test_trace_logging.py

@@ -5,8 +5,10 @@
 
 import pytest
 import requests
+from _pytest.capture import CaptureFixture
 
 from uvicorn import Config, Server
+from uvicorn._types import Receive, Scope, Send
 
 test_logging_config = {
     "version": 1,
@@ -52,18 +54,18 @@
     sys.platform.startswith("win") or platform.python_implementation() == "PyPy",
     reason="Skipping test on Windows and PyPy",
 )
-def test_trace_logging(capsys):
+def test_trace_logging(capsys: CaptureFixture) -> None:
     class App:
-        def __init__(self, scope):
+        def __init__(self, scope: Scope):
             if scope["type"] != "http":
                 raise Exception()
 
-        async def __call__(self, receive, send):
+        async def __call__(self, receive: Receive, send: Send) -> None:
             await send({"type": "http.response.start", "status": 204, "headers": []})
             await send({"type": "http.response.body", "body": b"", "more_body": False})
 
     class CustomServer(Server):
-        def install_signal_handlers(self):
+        def install_signal_handlers(self) -> None:
             pass
 
     config = Config(
@@ -91,18 +93,18 @@ def install_signal_handlers(self):
     reason="Skipping test on Windows and PyPy",
 )
 @pytest.mark.parametrize("http_protocol", [("h11"), ("httptools")])
-def test_access_logging(capsys, http_protocol):
+def test_access_logging(capsys: CaptureFixture, http_protocol: str) -> None:
     class App:
-        def __init__(self, scope):
+        def __init__(self, scope: Scope) -> None:
             if scope["type"] != "http":
                 raise Exception()
 
-        async def __call__(self, receive, send):
+        async def __call__(self, receive: Receive, send: Send) -> None:
             await send({"type": "http.response.start", "status": 204, "headers": []})
             await send({"type": "http.response.body", "body": b"", "more_body": False})
 
     class CustomServer(Server):
-        def install_signal_handlers(self):
+        def install_signal_handlers(self) -> None:
             pass
 
     config = Config(

tests/middleware/test_wsgi.py

@@ -1,12 +1,13 @@
 import sys
+from typing import Callable, List, Type
 
 import pytest
 
 from tests.client import TestClient
 from uvicorn.middleware.wsgi import WSGIMiddleware
 
 
-def hello_world(environ, start_response):
+def hello_world(environ: dict, start_response: Callable) -> List[bytes]:
     status = "200 OK"
     output = b"Hello World!\n"
     headers = [
@@ -17,7 +18,7 @@ def hello_world(environ, start_response):
     return [output]
 
 
-def echo_body(environ, start_response):
+def echo_body(environ: dict, start_response: Callable) -> List[bytes]:
     status = "200 OK"
     output = environ["wsgi.input"].read()
     headers = [
@@ -28,11 +29,11 @@ def echo_body(environ, start_response):
     return [output]
 
 
-def raise_exception(environ, start_response):
+def raise_exception(environ: dict, start_response: Callable) -> Type[Exception]:
     raise RuntimeError("Something went wrong")
 
 
-def return_exc_info(environ, start_response):
+def return_exc_info(environ: dict, start_response: Callable) -> List[bytes]:
     try:
         raise RuntimeError("Something went wrong")
     except RuntimeError:
@@ -46,23 +47,23 @@ def return_exc_info(environ, start_response):
         return [output]
 
 
-def test_wsgi_get():
+def test_wsgi_get() -> None:
     app = WSGIMiddleware(hello_world)
     client = TestClient(app)
     response = client.get("/")
     assert response.status_code == 200
     assert response.text == "Hello World!\n"
 
 
-def test_wsgi_post():
+def test_wsgi_post() -> None:
     app = WSGIMiddleware(echo_body)
     client = TestClient(app)
     response = client.post("/", json={"example": 123})
     assert response.status_code == 200
     assert response.text == '{"example": 123}'
 
 
-def test_wsgi_exception():
+def test_wsgi_exception() -> None:
     # Note that we're testing the WSGI app directly here.
     # The HTTP protocol implementations would catch this error and return 500.
     app = WSGIMiddleware(raise_exception)
@@ -71,7 +72,7 @@ def test_wsgi_exception():
         client.get("/")
 
 
-def test_wsgi_exc_info():
+def test_wsgi_exc_info() -> None:
     # Note that we're testing the WSGI app directly here.
     # The HTTP protocol implementations would catch this error and return 500.
     app = WSGIMiddleware(return_exc_info)

tests/protocols/test_websocket.py

@@ -167,7 +167,7 @@ async def websocket_connect(self, message):
             path = self.scope.get("path")
             raw_path = self.scope.get("raw_path")
             assert path == "/one/two"
-            assert raw_path == "/one%2Ftwo"
+            assert raw_path == b"/one%2Ftwo"
             await self.send({"type": "websocket.accept"})
 
     async def open_connection(url):

tests/test_auto_detection.py

@@ -20,7 +20,7 @@
     import websockets
 except ImportError:  # pragma: no cover
     # Note that we skip the websocket tests completely in this case.
-    websockets = None
+    websockets = None  # type: ignore
 
 
 # TODO: Add pypy to our testing matrix, and assert we get the correct classes
@@ -46,6 +46,8 @@ def test_http_auto():
 def test_websocket_auto():
     config = Config(app=None)
     server_state = ServerState()
-    protocol = AutoWebSocketsProtocol(config=config, server_state=server_state)
+    protocol = AutoWebSocketsProtocol(  # type: ignore
+        config=config, server_state=server_state
+    )
     expected_websockets = "WSProtocol" if websockets is None else "WebSocketProtocol"
     assert type(protocol).__name__ == expected_websockets

tests/test_main.py

@@ -1,6 +1,7 @@
 import asyncio
 import threading
 import time
+from typing import Optional, Union
 
 import requests
 
@@ -100,7 +101,7 @@ def install_signal_handlers(self):
     config = Config(app=App, loop="asyncio", workers=2, limit_max_requests=1)
     server = CustomServer(config=config)
     sock = config.bind_socket()
-    exc = True
+    exc: Optional[Union[Exception, bool]] = True
 
     def safe_run():
         nonlocal exc, server