fix: improve types and type hinting

src/grizzly/adapter/adapter.py

@@ -11,7 +11,7 @@
 from ..common.utils import HARNESS_FILE
 
 if TYPE_CHECKING:
-    from collections.abc import Generator
+    from collections.abc import Generator, Iterable
 
     from sapphire import ServerMap
 
@@ -45,7 +45,7 @@ class Adapter(metaclass=ABCMeta):
         remaining: Can be used to indicate the number of TestCases remaining to process.
     """
 
-    IGNORE_FILES = ("desktop.ini", "thumbs.db")
+    IGNORE_FILES = frozenset(("desktop.ini", "thumbs.db"))
     # Maximum iterations between Target relaunches (<1 use default)
     RELAUNCH = 0
     # Maximum execution time per test (used as minimum timeout). The iteration is
@@ -117,9 +117,9 @@ def get_harness(self) -> bytes | None:
     @staticmethod
     def scan_path(
         path: str,
-        ignore: tuple[str, ...] = IGNORE_FILES,
+        ignore: Iterable[str] = IGNORE_FILES,
         recursive: bool = False,
-    ) -> Generator[str, None, None]:
+    ) -> Generator[str]:
         """Scan a path and yield the files within it. This is available as
         a helper method.
 
@@ -156,7 +156,7 @@ def generate(self, testcase: TestCase, server_map: ServerMap) -> None:
             None
         """
 
-    def on_served(self, testcase: TestCase, served: tuple[str, ...]) -> None:
+    def on_served(self, testcase: TestCase, served: Iterable[str]) -> None:
         """Optional. Automatically called after a test case is successfully served.
 
         Args:
@@ -167,7 +167,7 @@ def on_served(self, testcase: TestCase, served: tuple[str, ...]) -> None:
             None
         """
 
-    def on_timeout(self, testcase: TestCase, served: tuple[str, ...]) -> None:
+    def on_timeout(self, testcase: TestCase, served: Iterable[str]) -> None:
         """Optional. Automatically called if timeout occurs while attempting to
         serve a test case. By default it calls `self.on_served()`.
 

src/grizzly/adapter/no_op_adapter/__init__.py

@@ -30,8 +30,8 @@ def setup(self, input_path: str | None, server_map: ServerMap) -> None:
         in a row without closing the browser after each one.
 
         Args:
-            _input: Unused.
-            _server_map: Unused.
+            input: Unused in this example.
+            server_map: Unused in this example.
 
         Returns:
             None
@@ -55,7 +55,7 @@ def generate(self, testcase: TestCase, server_map: ServerMap) -> None:
 
         Args:
             testcase: TestCase to be populated.
-            _server_map: Unused in this example.
+            server_map: Unused in this example.
 
         Returns:
             None

src/grizzly/common/bugzilla.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 from shutil import rmtree
 from tempfile import mkdtemp
+from types import MappingProxyType
 from typing import TYPE_CHECKING
 from zipfile import ZipFile
 
@@ -42,7 +43,7 @@
     )
 )
 # TODO: support all target assets
-KNOWN_ASSETS = {"prefs": "prefs.js"}
+KNOWN_ASSETS = MappingProxyType({"prefs": "prefs.js"})
 LOG = getLogger(__name__)
 
 
@@ -119,9 +120,7 @@ def _unpack_archives(self) -> None:
                 entry.unlink()
             # TODO: add support for other archive types
 
-    def assets(
-        self, ignore: tuple[str] | None = None
-    ) -> Generator[tuple[str, Path], None, None]:
+    def assets(self, ignore: tuple[str] | None = None) -> Generator[tuple[str, Path]]:
         """Scan files for assets.
 
         Arguments:

src/grizzly/common/fuzzmanager.py

@@ -321,31 +321,29 @@ def cleanup(self) -> None:
         if self._sig_filename is not None:
             rmtree(self._sig_filename.parent)
 
-    def iter_crashes(
-        self, quality_filter: int | None = None
-    ) -> Generator[CrashEntry, None, None]:
+    def iter_crashes(self, quality_filter: int | None = None) -> Generator[CrashEntry]:
         """Fetch all crash IDs for this FuzzManager bucket.
         Only crashes with testcases are returned.
 
         Arguments:
-            quality_filter: Filter crashes by quality value (None for all)
+            quality_filter: Filter crashes by quality value (None for no filter)
 
         Yields:
             CrashEntry objects.
         """
 
         def _get_results(
             endpoint: str, params: dict[str, str] | None = None
-        ) -> Generator[dict[str, Any], None, None]:
+        ) -> Generator[dict[str, Any]]:
             """
             Function to get paginated results from FuzzManager
 
             Args:
                 endpoint: FuzzManager REST API to query (eg. "crashes").
                 params: Params to pass through to requests.get.
 
-            Returns:
-                Objects (dict) returned by FuzzManager.
+            Yields:
+                Data returned by FuzzManager.
             """
             LOG.debug("first request to /%s/", endpoint)
 
@@ -438,7 +436,7 @@ def signature_path(self) -> Path:
 @contextmanager
 def load_fm_data(
     crash_id: int, load_bucket: bool = False
-) -> Generator[tuple[CrashEntry, Bucket | None], None, None]:
+) -> Generator[tuple[CrashEntry, Bucket | None]]:
     """Load CrashEntry including Bucket from FuzzManager.
 
     Arguments:

src/grizzly/common/status.py

@@ -105,7 +105,7 @@ def __init__(self, pid: int) -> None:
         self._desc: dict[str, str] = {}
         self.pid = pid
 
-    def __iter__(self) -> Generator[ResultEntry, None, None]:
+    def __iter__(self) -> Generator[ResultEntry]:
         """Yield all result data.
 
         Args:
@@ -120,7 +120,7 @@ def __iter__(self) -> Generator[ResultEntry, None, None]:
 
     def blockers(
         self, iterations: int, iters_per_result: int = 100
-    ) -> Generator[ResultEntry, None, None]:
+    ) -> Generator[ResultEntry]:
         """Any result with an iterations-per-result ratio of less than or equal the
         given limit are considered 'blockers'. Results with a count <= 1 are not
         included.
@@ -429,7 +429,7 @@ def __init__(
         self.start_time = start_time
         self.test_name: str | None = None
 
-    def profile_entries(self) -> Generator[ProfileEntry, None, None]:
+    def profile_entries(self) -> Generator[ProfileEntry]:
         """Used to retrieve profiling data.
 
         Args:
@@ -513,7 +513,7 @@ def __init__(
     @classmethod
     def load_all(
         cls, db_file: Path, time_limit: float = 300
-    ) -> Generator[ReadOnlyStatus, None, None]:
+    ) -> Generator[ReadOnlyStatus]:
         """Load all status reports found in `db_file`.
 
         Args:
@@ -680,7 +680,7 @@ def _init_db(db_file: Path, pid: int, life_time: float) -> None:
                 cur.execute("""DELETE FROM status WHERE pid = ?;""", (pid,))
 
     @contextmanager
-    def measure(self, name: str) -> Generator[None, None, None]:
+    def measure(self, name: str) -> Generator[None]:
         """Used to simplify collecting profiling data.
 
         Args:
@@ -1061,7 +1061,7 @@ def report(self, force: bool = False, report_rate: float = REPORT_RATE) -> bool:
     @classmethod
     def load_all(
         cls, db_file: Path, time_limit: float = 300
-    ) -> Generator[ReductionStatus, None, None]:
+    ) -> Generator[ReductionStatus]:
         """Load all reduction status reports found in `db_file`.
 
         Args:
@@ -1247,7 +1247,7 @@ def record(
             self.report(force=True)
 
     @contextmanager
-    def measure(self, name: str, report: bool = True) -> Generator[None, None, None]:
+    def measure(self, name: str, report: bool = True) -> Generator[None]:
         """Time and record the period leading up to a reduction milestone.
         eg. a strategy being run.
 

src/grizzly/common/status_reporter.py

@@ -629,7 +629,7 @@ def __init__(
         self._vsep = vsep
         self._hsep = hsep
 
-    def format_rows(self, rows: Iterable[ReductionStep]) -> Generator[str, None, None]:
+    def format_rows(self, rows: Iterable[ReductionStep]) -> Generator[str]:
         """Format rows as a table and return a line generator.
 
         Arguments:

src/grizzly/common/storage.py

@@ -17,7 +17,7 @@
 from .utils import grz_tmp, package_version
 
 if TYPE_CHECKING:
-    from collections.abc import Generator, Iterator
+    from collections.abc import Generator
 
 __all__ = ("TestCase", "TestCaseLoadFailure", "TestFileExists")
 __author__ = "Tyson Smith"
@@ -121,7 +121,7 @@ def __enter__(self) -> TestCase:
     def __exit__(self, *exc: object) -> None:
         self.cleanup()
 
-    def __iter__(self) -> Iterator[str]:
+    def __iter__(self) -> Generator[str]:
         """TestCase contents.
 
         Args:
@@ -478,7 +478,7 @@ def load_meta(
         return (entry_point, info)
 
     @property
-    def optional(self) -> Generator[str, None, None]:
+    def optional(self) -> Generator[str]:
         """Get file paths of optional files.
 
         Args:
@@ -511,7 +511,7 @@ def read_info(path: Path) -> dict[str, Any]:
         return info or {}
 
     @property
-    def required(self) -> Generator[str, None, None]:
+    def required(self) -> Generator[str]:
         """Get file paths of required files.
 
         Args:

src/grizzly/reduce/strategies/__init__.py

@@ -19,12 +19,13 @@
 from pathlib import Path
 from shutil import rmtree
 from tempfile import mkdtemp
+from types import MappingProxyType
 from typing import TYPE_CHECKING, cast
 
 from ...common.utils import grz_tmp
 
 if TYPE_CHECKING:
-    from collections.abc import Iterable, Iterator
+    from collections.abc import Iterable, Iterator, Sequence
 
     from ...common.storage import TestCase
 
@@ -58,7 +59,7 @@ class Strategy(ABC):
 
     name: str
 
-    def __init__(self, testcases: list[TestCase]) -> None:
+    def __init__(self, testcases: Sequence[TestCase]) -> None:
         """Initialize strategy instance.
 
         Arguments:
@@ -126,9 +127,9 @@ def get_tried(self) -> frozenset[tuple[tuple[str, bytes], ...]]:
         return frozenset(self._tried)
 
     def dump_testcases(
-        self, testcases: list[TestCase], recreate_tcroot: bool = False
+        self, testcases: Sequence[TestCase], recreate_tcroot: bool = False
     ) -> None:
-        """Dump a testcase list to the testcase root on disk.
+        """Dump provided testcases to the testcase root on disk.
 
         Arguments:
             testcases: Testcases to dump.
@@ -195,7 +196,7 @@ def cleanup(self) -> None:
         rmtree(self._testcase_root)
 
 
-def _load_strategies() -> dict[str, type[Strategy]]:
+def _load_strategies() -> MappingProxyType[str, type[Strategy]]:
     """STRATEGIES is created at the end of this file.
 
     Returns:
@@ -217,7 +218,7 @@ def _load_strategies() -> dict[str, type[Strategy]]:
             f"Unknown entry in DEFAULT_STRATEGIES: {strategy} "
             f"(STRATEGIES: [{','.join(strategies)}])"
         )
-    return strategies
+    return MappingProxyType(strategies)
 
 
 def _contains_dd(path: Path) -> bool:

src/grizzly/reduce/strategies/beautify.py

@@ -41,7 +41,7 @@
 LOG = getLogger(__name__)
 
 
-def _split_normal_lines(data: bytes) -> Generator[bytes, None, None]:
+def _split_normal_lines(data: bytes) -> Generator[bytes]:
     """Like str.splitlines but only respect \n, \r\n, and \r .. leave other
     potential line break characters intact.
 
@@ -130,7 +130,7 @@ def beautify_bytes(cls, data: bytes) -> bytes:
 
     def _chunks_to_beautify(
         self, before: bytes, to_beautify: bytes, file: Path
-    ) -> Generator[tuple[int, int], None, None]:
+    ) -> Generator[tuple[int, int]]:
         """Iterate over `to_beautify` and find chunks of style/script to beautify.
 
         Arguments:
@@ -183,7 +183,7 @@ def _chunks_to_beautify(
             yield (chunk_start, chunk_start + tag_end.start(0))
             search_start = chunk_start + tag_end.end(0)
 
-    def __iter__(self) -> Generator[list[TestCase], None, None]:
+    def __iter__(self) -> Generator[list[TestCase]]:
         """Iterate over potential beautifications of testcases according to this
         strategy.
 

src/grizzly/reduce/strategies/lithium.py

@@ -18,7 +18,7 @@
 from . import Strategy, _contains_dd
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator
+    from collections.abc import Generator
     from pathlib import Path
 
 LOG = getLogger(__name__)
@@ -83,7 +83,7 @@ def update(self, success: bool) -> None:
             self._current_reducer.feedback(success)
         self._current_feedback = success
 
-    def __iter__(self) -> Iterator[list[TestCase]]:
+    def __iter__(self) -> Generator[list[TestCase]]:
         """Iterate over potential reductions of testcases according to this strategy.
 
         The caller should evaluate each testcase set yielded, and call `update` with the
@@ -185,7 +185,7 @@ def __init__(self, testcases: list[TestCase]) -> None:
         # just once per Grizzly TestCase set is enough.
         self._files_to_reduce = self._files_to_reduce[:1]
 
-    def __iter__(self) -> Iterator[list[TestCase]]:
+    def __iter__(self) -> Generator[list[TestCase]]:
         yield from super().__iter__()
 
 

src/grizzly/reduce/strategies/testcases.py

@@ -57,7 +57,7 @@ def update(self, success: bool) -> None:
         assert self._current_served is None
         self._current_feedback = success
 
-    def __iter__(self) -> Generator[list[TestCase], None, None]:
+    def __iter__(self) -> Generator[list[TestCase]]:
         """Iterate over potential reductions of testcases according to this strategy.
 
         The caller should evaluate each testcase set yielded, and call `update` with the

src/grizzly/replay/replay.py

@@ -44,7 +44,7 @@
 
 if TYPE_CHECKING:
     from argparse import Namespace
-    from collections.abc import Iterable
+    from collections.abc import Iterable, Sequence
 
 __author__ = "Tyson Smith"
 __credits__ = ["Tyson Smith"]
@@ -180,7 +180,7 @@ def expect_hang(
     @classmethod
     def load_testcases(
         cls,
-        paths: Iterable[Path],
+        paths: Sequence[Path],
         catalog: bool = False,
         entry_point: Path | None = None,
     ) -> tuple[list[TestCase], AssetManager | None, dict[str, str] | None]: