Add 100% clean coverage (#2394)

* Add 100% clean coverage

* Add 100% clean coverage

* Add 100% clean coverage

* Add 100% clean coverage

Author: Kludex

docs/contributing.md

@@ -144,10 +144,10 @@ If tests are failing you will see this message under the coverage report:
 
 `=== 1 failed, 354 passed, 1 skipped, 1 xfailed in 37.08s ===`
 
-If tests succeed but coverage doesn't reach our current threshold, you will see this
+If tests succeed but coverage doesn't reach 100%, you will see this
 message under the coverage report:
 
-`Coverage failure: total of 88 is less than fail-under=95`
+`Coverage failure: total of 98 is less than fail-under=100`
 
 ## Releasing
 

pyproject.toml

@@ -9,9 +9,7 @@ description = "The lightning-fast ASGI server."
 readme = "README.md"
 license = "BSD-3-Clause"
 requires-python = ">=3.8"
-authors = [
-    { name = "Tom Christie", email = "[email protected]" },
-]
+authors = [{ name = "Tom Christie", email = "[email protected]" }]
 classifiers = [
     "Development Status :: 4 - Beta",
     "Environment :: Web Environment",
@@ -100,12 +98,13 @@ omit = ["uvicorn/workers.py", "uvicorn/__main__.py"]
 
 [tool.coverage.report]
 precision = 2
-fail_under = 98.35
+fail_under = 100
 show_missing = true
 skip_covered = true
 exclude_lines = [
     "pragma: no cover",
     "pragma: nocover",
+    "pragma: full coverage",
     "if TYPE_CHECKING:",
     "if typing.TYPE_CHECKING:",
     "raise NotImplementedError",

tests/supervisors/test_multiprocess.py

@@ -45,7 +45,7 @@ async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
 
 
 def run(sockets: list[socket.socket] | None) -> None:
-    while True:
+    while True:  # pragma: no cover
         time.sleep(1)
 
 

tests/supervisors/test_signal.py

@@ -60,7 +60,7 @@ async def forever_app(scope, receive, send):
         await send({"type": "http.response.body", "body": b"start", "more_body": True})
         # we never continue this one, so this request will time out
         await server_event.wait()
-        await send({"type": "http.response.body", "body": b"end", "more_body": False})
+        await send({"type": "http.response.body", "body": b"end", "more_body": False})  # pragma: full coverage
 
     config = Config(app=forever_app, reload=False, port=unused_tcp_port, timeout_graceful_shutdown=1)
     server: Server
@@ -90,7 +90,7 @@ async def test_sigint_deny_request_after_triggered(unused_tcp_port: int, caplog)
 
     async def app(scope, receive, send):
         await send({"type": "http.response.start", "status": 200, "headers": []})
-        await asyncio.sleep(1)
+        await asyncio.sleep(1)  # pragma: full coverage
 
     config = Config(app=app, reload=False, port=unused_tcp_port, timeout_graceful_shutdown=1)
     server: Server

tests/test_server.py

@@ -37,10 +37,10 @@ async def dummy_app(scope, receive, send):  # pragma: py-win32
     pass
 
 
-if sys.platform == "win32":
+if sys.platform == "win32":  # pragma: py-not-win32
     signals = [signal.SIGBREAK]
     signal_captures = [capture_signal_sync]
-else:
+else:  # pragma: py-win32
     signals = [signal.SIGTERM, signal.SIGINT]
     signal_captures = [capture_signal_sync, capture_signal_async]
 

uvicorn/_subprocess.py

@@ -70,7 +70,7 @@ def subprocess_started(
     """
     # Re-open stdin.
     if stdin_fileno is not None:
-        sys.stdin = os.fdopen(stdin_fileno)
+        sys.stdin = os.fdopen(stdin_fileno)  # pragma: full coverage
 
     # Logging needs to be setup again for each child.
     config.configure_logging()

uvicorn/config.py

@@ -124,7 +124,7 @@ def is_dir(path: Path) -> bool:
         if not path.is_absolute():
             path = path.resolve()
         return path.is_dir()
-    except OSError:
+    except OSError:  # pragma: full coverage
         return False
 
 
@@ -153,9 +153,9 @@ def resolve_reload_patterns(patterns_list: list[str], directories_list: list[str
 
     children = []
     for j in range(len(directories)):
-        for k in range(j + 1, len(directories)):
+        for k in range(j + 1, len(directories)):  # pragma: full coverage
             if directories[j] in directories[k].parents:
-                children.append(directories[k])  # pragma: py-darwin
+                children.append(directories[k])
             elif directories[k] in directories[j].parents:
                 children.append(directories[j])
 
@@ -298,12 +298,12 @@ def __init__(
                     if directory == reload_directory or directory in reload_directory.parents:
                         try:
                             self.reload_dirs.remove(reload_directory)
-                        except ValueError:
+                        except ValueError:  # pragma: full coverage
                             pass
 
             for pattern in self.reload_excludes:
                 if pattern in self.reload_includes:
-                    self.reload_includes.remove(pattern)
+                    self.reload_includes.remove(pattern)  # pragma: full coverage
 
             if not self.reload_dirs:
                 if reload_dirs:
@@ -332,7 +332,7 @@ def __init__(
         if forwarded_allow_ips is None:
             self.forwarded_allow_ips = os.environ.get("FORWARDED_ALLOW_IPS", "127.0.0.1")
         else:
-            self.forwarded_allow_ips = forwarded_allow_ips
+            self.forwarded_allow_ips = forwarded_allow_ips  # pragma: full coverage
 
         if self.reload and self.workers > 1:
             logger.warning('"workers" flag is ignored when reloading is enabled.')
@@ -485,7 +485,7 @@ def bind_socket(self) -> socket.socket:
                 sock.bind(path)
                 uds_perms = 0o666
                 os.chmod(self.uds, uds_perms)
-            except OSError as exc:
+            except OSError as exc:  # pragma: full coverage
                 logger.error(exc)
                 sys.exit(1)
 
@@ -503,7 +503,7 @@ def bind_socket(self) -> socket.socket:
             family = socket.AF_INET
             addr_format = "%s://%s:%d"
 
-            if self.host and ":" in self.host:  # pragma: py-win32
+            if self.host and ":" in self.host:  # pragma: full coverage
                 # It's an IPv6 address.
                 family = socket.AF_INET6
                 addr_format = "%s://[%s]:%d"
@@ -512,7 +512,7 @@ def bind_socket(self) -> socket.socket:
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             try:
                 sock.bind((self.host, self.port))
-            except OSError as exc:
+            except OSError as exc:  # pragma: full coverage
                 logger.error(exc)
                 sys.exit(1)
 

uvicorn/loops/asyncio.py

@@ -7,4 +7,4 @@
 
 def asyncio_setup(use_subprocess: bool = False) -> None:
     if sys.platform == "win32" and use_subprocess:
-        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())  # pragma: full coverage

uvicorn/main.py

@@ -576,7 +576,7 @@ def run(
         else:
             server.run()
     except KeyboardInterrupt:
-        pass
+        pass  # pragma: full coverage
     finally:
         if config.uds and os.path.exists(config.uds):
             os.remove(config.uds)  # pragma: py-win32

uvicorn/middleware/proxy_headers.py

@@ -37,7 +37,7 @@ def get_trusted_client_host(self, x_forwarded_for_hosts: list[str]) -> str | Non
             if host not in self.trusted_hosts:
                 return host
 
-        return None
+        return None  # pragma: full coverage
 
     async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
         if scope["type"] in ("http", "websocket"):

uvicorn/protocols/http/flow_control.py

@@ -16,7 +16,7 @@ def __init__(self, transport: asyncio.Transport) -> None:
         self._is_writable_event.set()
 
     async def drain(self) -> None:
-        await self._is_writable_event.wait()
+        await self._is_writable_event.wait()  # pragma: full coverage
 
     def pause_reading(self) -> None:
         if not self.read_paused:
@@ -29,12 +29,12 @@ def resume_reading(self) -> None:
             self._transport.resume_reading()
 
     def pause_writing(self) -> None:
-        if not self.write_paused:
+        if not self.write_paused:  # pragma: full coverage
             self.write_paused = True
             self._is_writable_event.clear()
 
     def resume_writing(self) -> None:
-        if self.write_paused:
+        if self.write_paused:  # pragma: full coverage
             self.write_paused = False
             self._is_writable_event.set()
 

uvicorn/protocols/http/h11_impl.py

@@ -263,7 +263,7 @@ def handle_events(self) -> None:
                 self.cycle.message_event.set()
 
     def handle_websocket_upgrade(self, event: h11.Request) -> None:
-        if self.logger.level <= TRACE_LOG_LEVEL:
+        if self.logger.level <= TRACE_LOG_LEVEL:  # pragma: full coverage
             prefix = "%s:%d - " % self.client if self.client else ""
             self.logger.log(TRACE_LOG_LEVEL, "%sUpgrading to WebSocket", prefix)
 
@@ -333,13 +333,13 @@ def pause_writing(self) -> None:
         """
         Called by the transport when the write buffer exceeds the high water mark.
         """
-        self.flow.pause_writing()
+        self.flow.pause_writing()  # pragma: full coverage
 
     def resume_writing(self) -> None:
         """
         Called by the transport when the write buffer drops below the low water mark.
         """
-        self.flow.resume_writing()
+        self.flow.resume_writing()  # pragma: full coverage
 
     def timeout_keep_alive_handler(self) -> None:
         """
@@ -441,10 +441,10 @@ async def send(self, message: ASGISendEvent) -> None:
         message_type = message["type"]
 
         if self.flow.write_paused and not self.disconnected:
-            await self.flow.drain()
+            await self.flow.drain()  # pragma: full coverage
 
         if self.disconnected:
-            return
+            return  # pragma: full coverage
 
         if not self.response_started:
             # Sending response status line and headers

uvicorn/protocols/http/httptools_impl.py

@@ -139,7 +139,7 @@ def _get_upgrade(self) -> bytes | None:
                 upgrade = value.lower()
         if b"upgrade" in connection:
             return upgrade
-        return None
+        return None  # pragma: full coverage
 
     def _should_upgrade_to_ws(self, upgrade: bytes | None) -> bool:
         if upgrade == b"websocket" and self.ws_protocol_class is not None:
@@ -193,7 +193,7 @@ def handle_websocket_upgrade(self) -> None:
     def send_400_response(self, msg: str) -> None:
         content = [STATUS_LINE[400]]
         for name, value in self.server_state.default_headers:
-            content.extend([name, b": ", value, b"\r\n"])
+            content.extend([name, b": ", value, b"\r\n"])  # pragma: full coverage
         content.extend(
             [
                 b"content-type: text/plain; charset=utf-8\r\n",
@@ -336,13 +336,13 @@ def pause_writing(self) -> None:
         """
         Called by the transport when the write buffer exceeds the high water mark.
         """
-        self.flow.pause_writing()
+        self.flow.pause_writing()  # pragma: full coverage
 
     def resume_writing(self) -> None:
         """
         Called by the transport when the write buffer drops below the low water mark.
         """
-        self.flow.resume_writing()
+        self.flow.resume_writing()  # pragma: full coverage
 
     def timeout_keep_alive_handler(self) -> None:
         """
@@ -441,10 +441,10 @@ async def send(self, message: ASGISendEvent) -> None:
         message_type = message["type"]
 
         if self.flow.write_paused and not self.disconnected:
-            await self.flow.drain()
+            await self.flow.drain()  # pragma: full coverage
 
         if self.disconnected:
-            return
+            return  # pragma: full coverage
 
         if not self.response_started:
             # Sending response status line and headers
@@ -477,7 +477,7 @@ async def send(self, message: ASGISendEvent) -> None:
 
             for name, value in headers:
                 if HEADER_RE.search(name):
-                    raise RuntimeError("Invalid HTTP header name.")
+                    raise RuntimeError("Invalid HTTP header name.")  # pragma: full coverage
                 if HEADER_VALUE_RE.search(value):
                     raise RuntimeError("Invalid HTTP header value.")
 

uvicorn/protocols/websockets/websockets_impl.py

@@ -242,7 +242,7 @@ async def run_asgi(self) -> None:
         """
         try:
             result = await self.app(self.scope, self.asgi_receive, self.asgi_send)  # type: ignore[func-returns-value]
-        except ClientDisconnected:
+        except ClientDisconnected:  # pragma: full coverage
             self.closed_event.set()
             self.transport.close()
         except BaseException:

uvicorn/protocols/websockets/wsproto_impl.py

@@ -44,7 +44,7 @@ def __init__(
         _loop: asyncio.AbstractEventLoop | None = None,
     ) -> None:
         if not config.loaded:
-            config.load()
+            config.load()  # pragma: full coverage
 
         self.config = config
         self.app = cast(ASGI3Application, config.loaded_app)
@@ -140,13 +140,13 @@ def pause_writing(self) -> None:
         """
         Called by the transport when the write buffer exceeds the high water mark.
         """
-        self.writable.clear()
+        self.writable.clear()  # pragma: full coverage
 
     def resume_writing(self) -> None:
         """
         Called by the transport when the write buffer drops below the low water mark.
         """
-        self.writable.set()
+        self.writable.set()  # pragma: full coverage
 
     def shutdown(self) -> None:
         if self.handshake_complete:
@@ -233,7 +233,7 @@ async def run_asgi(self) -> None:
         try:
             result = await self.app(self.scope, self.receive, self.send)  # type: ignore[func-returns-value]
         except ClientDisconnected:
-            self.transport.close()
+            self.transport.close()  # pragma: full coverage
         except BaseException:
             self.logger.exception("Exception in ASGI application\n")
             self.send_500_response()

uvicorn/server.py

@@ -112,7 +112,7 @@ def create_protocol(
         loop = asyncio.get_running_loop()
 
         listeners: Sequence[socket.SocketType]
-        if sockets is not None:
+        if sockets is not None:  # pragma: full coverage
             # Explicitly passed a list of open sockets.
             # We use this when the server is run from a Gunicorn worker.
 
@@ -147,7 +147,7 @@ def _share_socket(
             # Create a socket using UNIX domain socket.
             uds_perms = 0o666
             if os.path.exists(config.uds):
-                uds_perms = os.stat(config.uds).st_mode
+                uds_perms = os.stat(config.uds).st_mode  # pragma: full coverage
             server = await loop.create_unix_server(
                 create_protocol, path=config.uds, ssl=config.ssl, backlog=config.backlog
             )
@@ -180,7 +180,7 @@ def _share_socket(
         else:
             # We're most likely running multiple workers, so a message has already been
             # logged by `config.bind_socket()`.
-            pass
+            pass  # pragma: full coverage
 
         self.started = True
 
@@ -243,7 +243,7 @@ async def on_tick(self, counter: int) -> bool:
 
             # Callback to `callback_notify` once every `timeout_notify` seconds.
             if self.config.callback_notify is not None:
-                if current_time - self.last_notified > self.config.timeout_notify:
+                if current_time - self.last_notified > self.config.timeout_notify:  # pragma: full coverage
                     self.last_notified = current_time
                     await self.config.callback_notify()
 
@@ -261,7 +261,7 @@ async def shutdown(self, sockets: list[socket.socket] | None = None) -> None:
         for server in self.servers:
             server.close()
         for sock in sockets or []:
-            sock.close()
+            sock.close()  # pragma: full coverage
 
         # Request shutdown on all existing connections.
         for connection in list(self.server_state.connections):
@@ -330,6 +330,6 @@ def capture_signals(self) -> Generator[None, None, None]:
     def handle_exit(self, sig: int, frame: FrameType | None) -> None:
         self._captured_signals.append(sig)
         if self.should_exit and sig == signal.SIGINT:
-            self.force_exit = True
+            self.force_exit = True  # pragma: full coverage
         else:
             self.should_exit = True