Cover httptools_impl on mypy

setup.cfg

@@ -41,7 +41,8 @@ files =
     uvicorn/middleware/__init__.py,
     uvicorn/protocols/__init__.py,
     uvicorn/protocols/http/__init__.py,
-    uvicorn/protocols/websockets/__init__.py
+    uvicorn/protocols/websockets/__init__.py,
+    uvicorn/protocols/http/httptools_impl.py
 
 
 [mypy-tests.*]

uvicorn/_types.py

@@ -1,14 +1,44 @@
+import asyncio
 import types
-import typing
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Iterable,
+    MutableMapping,
+    Optional,
+    Protocol,
+    Tuple,
+    Type,
+    Union,
+)
+
+if TYPE_CHECKING:
+    from uvicorn.config import Config
+    from uvicorn.server_state import ServerState
 
 # WSGI
-Environ = typing.MutableMapping[str, typing.Any]
-ExcInfo = typing.Tuple[
-    typing.Type[BaseException], BaseException, typing.Optional[types.TracebackType]
-]
-StartResponse = typing.Callable[
-    [str, typing.Iterable[typing.Tuple[str, str]], typing.Optional[ExcInfo]], None
-]
-WSGIApp = typing.Callable[
-    [Environ, StartResponse], typing.Union[typing.Iterable[bytes], BaseException]
-]
+Environ = MutableMapping[str, Any]
+ExcInfo = Tuple[Type[BaseException], BaseException, Optional[types.TracebackType]]
+StartResponse = Callable[[str, Iterable[Tuple[str, str]], Optional[ExcInfo]], None]
+WSGIApp = Callable[[Environ, StartResponse], Union[Iterable[bytes], BaseException]]
+
+
+class WebProtocol(Protocol):
+    def __init__(
+        self,
+        config: "Config",
+        server_state: "ServerState",
+        on_connection_lost: Optional[Callable[..., Any]],
+        _loop: Optional[asyncio.AbstractEventLoop] = None,
+    ) -> None:
+        ...
+
+    def connection_made(self, transport) -> None:
+        ...
+
+    def data_received(self, data: bytes) -> None:
+        ...
+
+    def shutdown(self) -> None:
+        ...

uvicorn/config.py

@@ -9,6 +9,7 @@
 import sys
 from typing import Awaitable, Callable, Dict, List, Optional, Tuple, Type, Union
 
+from uvicorn._types import WebProtocol
 from uvicorn.logging import TRACE_LOG_LEVEL
 
 if sys.version_info < (3, 8):
@@ -134,6 +135,8 @@ def create_ssl_context(
 
 
 class Config:
+    ws_protocol_class: Type[WebProtocol]
+
     def __init__(
         self,
         app: Union[ASGIApplication, Callable, str],
@@ -143,7 +146,7 @@ def __init__(
         fd: Optional[int] = None,
         loop: LoopSetupType = "auto",
         http: Union[Type[asyncio.Protocol], HTTPProtocolType] = "auto",
-        ws: Union[Type[asyncio.Protocol], WSProtocolType] = "auto",
+        ws: Union[Type[WebProtocol], WSProtocolType] = "auto",
         ws_max_size: int = 16 * 1024 * 1024,
         ws_ping_interval: int = 20,
         ws_ping_timeout: int = 20,
@@ -337,7 +340,7 @@ def load(self) -> None:
 
         if isinstance(self.ws, str):
             ws_protocol_class = import_from_string(WS_PROTOCOLS[self.ws])
-            self.ws_protocol_class: Optional[Type[asyncio.Protocol]] = ws_protocol_class
+            self.ws_protocol_class = ws_protocol_class
         else:
             self.ws_protocol_class = self.ws
 
@@ -374,7 +377,7 @@ def load(self) -> None:
 
         if self.interface == "wsgi":
             self.loaded_app = WSGIMiddleware(self.loaded_app)
-            self.ws_protocol_class = None
+            self.ws_protocol_class = None  # type: ignore[assignment]
         elif self.interface == "asgi2":
             self.loaded_app = ASGI2Middleware(self.loaded_app)