🚨 Cover server.py on mypy

setup.cfg

@@ -29,6 +29,7 @@ files =
     uvicorn/logging.py,
     uvicorn/middleware/asgi2.py,
     uvicorn/_handlers,
+    uvicorn/server.py,
     uvicorn/__init__.py,
     uvicorn/__main__.py,
     uvicorn/subprocess.py,

uvicorn/config.py

@@ -7,7 +7,7 @@
 import socket
 import ssl
 import sys
-from typing import Callable, Dict, List, Optional, Tuple, Type, Union
+from typing import Awaitable, Callable, Dict, List, Optional, Tuple, Type, Union
 
 from uvicorn.logging import TRACE_LOG_LEVEL
 
@@ -169,7 +169,7 @@ def __init__(
         backlog: int = 2048,
         timeout_keep_alive: int = 5,
         timeout_notify: int = 30,
-        callback_notify: Callable[..., None] = None,
+        callback_notify: Callable[..., Awaitable[None]] = None,
         ssl_keyfile: Optional[str] = None,
         ssl_certfile: Optional[Union[str, os.PathLike]] = None,
         ssl_keyfile_password: Optional[str] = None,
@@ -219,7 +219,7 @@ def __init__(
         self.ssl_ca_certs = ssl_ca_certs
         self.ssl_ciphers = ssl_ciphers
         self.headers: List[List[str]] = headers or []
-        self.encoded_headers: Optional[List[Tuple[bytes, bytes]]] = None
+        self.encoded_headers: List[Tuple[bytes, bytes]] = []
         self.factory = factory
 
         self.loaded = False

uvicorn/server.py

@@ -8,11 +8,25 @@
 import threading
 import time
 from email.utils import formatdate
-from typing import List
+from types import FrameType
+from typing import Any, List, Optional, Set, Tuple, Union
 
 import click
 
-from ._handlers.http import handle_http
+from uvicorn._handlers.http import handle_http
+from uvicorn.config import Config
+from uvicorn.protocols.http.h11_impl import H11Protocol
+from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
+from uvicorn.protocols.websockets.websockets_impl import WebSocketProtocol
+from uvicorn.protocols.websockets.wsproto_impl import WSProtocol
+
+if sys.platform != "win32":
+    from asyncio import start_unix_server as _start_unix_server
+else:
+
+    async def _start_unix_server(*args: Any, **kwargs: Any) -> Any:
+        raise NotImplementedError("Cannot start a unix server on win32")
+
 
 HANDLED_SIGNALS = (
     signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.
@@ -21,35 +35,37 @@
 
 logger = logging.getLogger("uvicorn.error")
 
+Protocols = Union[H11Protocol, HttpToolsProtocol, WSProtocol, WebSocketProtocol]
+
 
 class ServerState:
     """
     Shared servers state that is available between all protocol instances.
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         self.total_requests = 0
-        self.connections = set()
-        self.tasks = set()
-        self.default_headers = []
+        self.connections: Set[Protocols] = set()
+        self.tasks: Set[asyncio.Task] = set()
+        self.default_headers: List[Tuple[bytes, bytes]] = []
 
 
 class Server:
-    def __init__(self, config):
+    def __init__(self, config: Config) -> None:
         self.config = config
         self.server_state = ServerState()
 
         self.started = False
         self.should_exit = False
         self.force_exit = False
-        self.last_notified = 0
+        self.last_notified = 0.0
 
-    def run(self, sockets=None):
+    def run(self, sockets: Optional[List[socket.socket]] = None) -> None:
         self.config.setup_event_loop()
         loop = asyncio.get_event_loop()
         loop.run_until_complete(self.serve(sockets=sockets))
 
-    async def serve(self, sockets=None):
+    async def serve(self, sockets: Optional[List[socket.socket]] = None) -> None:
         process_id = os.getpid()
 
         config = self.config
@@ -126,7 +142,7 @@ def _share_socket(sock: socket.SocketType) -> socket.SocketType:
             uds_perms = 0o666
             if os.path.exists(config.uds):
                 uds_perms = os.stat(config.uds).st_mode
-            server = await asyncio.start_unix_server(
+            server = await _start_unix_server(
                 handler, path=config.uds, ssl=config.ssl, backlog=config.backlog
             )
             os.chmod(config.uds, uds_perms)
@@ -203,7 +219,7 @@ def _log_started_message(self, listeners: List[socket.SocketType]) -> None:
                 extra={"color_message": color_message},
             )
 
-    async def main_loop(self):
+    async def main_loop(self) -> None:
         counter = 0
         should_exit = await self.on_tick(counter)
         while not should_exit:
@@ -212,7 +228,7 @@ async def main_loop(self):
             await asyncio.sleep(0.1)
             should_exit = await self.on_tick(counter)
 
-    async def on_tick(self, counter) -> bool:
+    async def on_tick(self, counter: int) -> bool:
         # Update the default headers, once per second.
         if counter % 10 == 0:
             current_time = time.time()
@@ -240,7 +256,7 @@ async def on_tick(self, counter) -> bool:
             return self.server_state.total_requests >= self.config.limit_max_requests
         return False
 
-    async def shutdown(self, sockets=None):
+    async def shutdown(self, sockets: Optional[List[socket.socket]] = None) -> None:
         logger.info("Shutting down")
 
         # Stop accepting new connections.
@@ -289,7 +305,8 @@ def install_signal_handlers(self) -> None:
             for sig in HANDLED_SIGNALS:
                 signal.signal(sig, self.handle_exit)
 
-    def handle_exit(self, sig, frame):
+    def handle_exit(self, sig: signal.Signals, frame: FrameType) -> None:
+
         if self.should_exit:
             self.force_exit = True
         else: