fix: a few more minor nits

Author: tysmith

src/grizzly/services/core.py

@@ -1,18 +1,23 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
+from __future__ import annotations
+
 import asyncio
-from collections.abc import Mapping
 from enum import Enum
 from logging import getLogger
-from pathlib import Path
-from typing import cast
+from typing import TYPE_CHECKING, cast
 
 from sapphire import ServerMap, create_listening_socket
 
-from .base import BaseService
 from .webtransport.core import WebTransportServer
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+    from pathlib import Path
+
+    from .base import BaseService
+
 LOG = getLogger(__name__)
 
 
@@ -29,7 +34,7 @@ def __init__(self, services: Mapping[ServiceName, BaseService]):
         """Initialize new WebServices instance
 
         Args:
-            services (dict of ServiceName: BaseService): Collection of services.
+            services: Collection of services.
         """
         self.services = services
 
@@ -46,10 +51,10 @@ async def is_running(self, timeout: float = 20) -> bool:
         """Polls all available services to ensure they are running and accessible.
 
         Args:
-            timeout (int): Total time to wait.
+            timeout: Total time to wait.
 
         Returns:
-            bool: Indicates if all services started successfully.
+            Indicates if all services started successfully.
         """
         tasks = {}
         for name, service in self.services.items():
@@ -79,12 +84,12 @@ def map_locations(self, server_map: ServerMap) -> None:
             )
 
     @classmethod
-    def start_services(cls, cert: Path, key: Path) -> "WebServices":
+    def start_services(cls, cert: Path, key: Path) -> WebServices:
         """Start all available services
 
         Args:
-            cert (Path): Path to the certificate file
-            key (Path): Path to the certificate's private key
+            cert: Certificate file
+            key: Certificate's private key
         """
         services = {}
         # Start WebTransport service

src/grizzly/services/webtransport/core.py

@@ -6,10 +6,9 @@
 import asyncio
 from asyncio import AbstractEventLoop
 from logging import getLogger
-from os.path import abspath, dirname, join
+from pathlib import Path
 from platform import system
 from threading import Thread
-from typing import TYPE_CHECKING
 
 from aioquic.asyncio import serve  # type: ignore[attr-defined]
 from aioquic.h3.connection import H3_ALPN
@@ -23,16 +22,12 @@
     _connect_to_server,
 )
 
-if TYPE_CHECKING:
-    from pathlib import Path
-
 # Override global _doc_root
-webtransport_h3_server._doc_root = join(  # pylint: disable=protected-access
-    dirname(abspath(__file__)),
-    "wpt_h3_server",
-    "handlers",
+webtransport_h3_server._doc_root = str(  # pylint: disable=protected-access
+    (Path(__file__).parent / "wpt_h3_server" / "handlers").resolve()
 )
 
+
 LOG = getLogger(__name__)
 
 
@@ -42,16 +37,16 @@ def __init__(self, port: int, cert: Path, key: Path) -> None:
 
         Args:
             port: The port on which to listen on.
-            cert: The path to the certificate file.
-            key: The path to the certificate's private key.
+            cert: Certificate file.
+            key: Certificate's private key.
         """
         self._port = port
         self._cert = cert
         self._key = key
 
         self._loop: AbstractEventLoop | None = None
         self._server_thread: Thread | None = None
-        self._started: bool = False
+        self._started = False
 
     @property
     def location(self) -> str:
@@ -84,7 +79,7 @@ def _start_service() -> None:
                 max_datagram_frame_size=65536,
             )
 
-            LOG.debug("Starting WebTransport service on port %s", self.port)
+            LOG.debug("Starting WebTransport service on port %d", self.port)
             configuration.load_cert_chain(self._cert, self._key)
             ticket_store = SessionTicketStore()
 
@@ -117,14 +112,15 @@ def _start_service() -> None:
     def cleanup(self) -> None:
         """Stop the server."""
 
-        async def _stop_loop() -> None:
-            if self._loop is not None:
-                self._loop.stop()
-
         if self._started:
+
+            async def _stop_loop() -> None:
+                if self._loop is not None:
+                    self._loop.stop()
+
             if self._loop is not None:
                 asyncio.run_coroutine_threadsafe(_stop_loop(), self._loop)
             if self._server_thread is not None:
                 self._server_thread.join()
-            LOG.debug("Stopped WebTransport service on port %s", self._port)
-        self._started = False
+            LOG.debug("Stopped WebTransport service on port %d", self._port)
+            self._started = False

src/grizzly/services/webtransport/test_webtransport.py

@@ -4,7 +4,7 @@
 # pylint: disable=protected-access
 import asyncio
 
-import pytest
+from pytest import raises
 
 from sapphire import CertificateBundle
 
@@ -32,7 +32,7 @@ def test_webtransport_01():
         web_transport.cleanup()
 
         assert not web_transport._started
-        with pytest.raises(asyncio.TimeoutError):
+        with raises(asyncio.TimeoutError):
             asyncio.run(asyncio.wait_for(web_transport.is_ready(), timeout=1.0))
     finally:
         cert.cleanup()