🚨 Cover server.py on mypy

setup.cfg

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

uvicorn/server.py

@@ -8,10 +8,17 @@
 import threading
 import time
 from email.utils import formatdate
-from typing import List
+from types import FrameType
+from typing import List, Optional, Set, Tuple, Union
 
 import click
 
+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
+
 from ._handlers.http import handle_http
 
 HANDLED_SIGNALS = (
@@ -21,35 +28,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
@@ -130,7 +139,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 asyncio.start_unix_server(  # type: ignore
                 handler, path=config.uds, ssl=config.ssl, backlog=config.backlog
             )
             os.chmod(config.uds, uds_perms)
@@ -207,7 +216,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:
@@ -216,7 +225,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()
@@ -238,7 +247,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.
@@ -287,7 +296,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: