[ci] Enable ruff and address results

.pre-commit-config.yaml

@@ -20,6 +20,10 @@ repos:
     rev: 7.0.0
     hooks:
       - id: flake8
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.5.1
+    hooks:
+      - id: ruff
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.6.0
     hooks:

grizzly/args.py

@@ -13,6 +13,7 @@
 from os.path import exists
 from pathlib import Path
 from platform import system
+from types import MappingProxyType
 from typing import Iterable, List, Optional
 
 from FTB.ProgramConfiguration import ProgramConfiguration
@@ -47,18 +48,20 @@ def add_arguments(self, actions: Iterable[Action]) -> None:
 
 
 class CommonArgs:
-    IGNORABLE = ("log-limit", "memory", "timeout")
     DEFAULT_IGNORE = ("log-limit", "timeout")
-
-    def __init__(self) -> None:
-        # log levels for console logging
-        self._level_map = {
+    IGNORABLE = ("log-limit", "memory", "timeout")
+    # log levels for console logging
+    LEVEL_MAP = MappingProxyType(
+        {
             "CRIT": CRITICAL,
             "ERROR": ERROR,
             "WARN": WARNING,
             "INFO": INFO,
             "DEBUG": DEBUG,
         }
+    )
+
+    def __init__(self) -> None:
 
         self.parser = ArgumentParser(
             formatter_class=SortingHelpFormatter, conflict_handler="resolve"
@@ -71,17 +74,17 @@ def __init__(self) -> None:
         self.parser.add_argument("binary", type=Path, help="Firefox binary to run")
         self.parser.add_argument(
             "--log-level",
-            choices=sorted(self._level_map),
+            choices=sorted(self.LEVEL_MAP),
             default="INFO",
             help="Configure console logging (default: %(default)s)",
         )
 
-        # build 'asset' help string
+        # build 'asset' help string formatted as:
+        # target01: asset01, asset02. target02: asset03...
         assets = scan_target_assets()
-        asset_msg: List[str] = []
-        for target in sorted(assets):
-            if assets[target]:
-                asset_msg.append(f"{target}: {', '.join(sorted(assets[target]))}.")
+        asset_msg = "".join(
+            f"{x}: {', '.join(sorted(assets[x]))}." for x in sorted(assets) if assets[x]
+        )
 
         self.launcher_grp = self.parser.add_argument_group("Launcher Arguments")
         self.launcher_grp.add_argument(
@@ -90,7 +93,7 @@ def __init__(self) -> None:
             default=[],
             metavar=("ASSET", "PATH"),
             nargs=2,
-            help=f"Specify target specific asset files. {''.join(asset_msg)}",
+            help=f"Specify target specific asset files. {asset_msg}",
         )
         self.launcher_grp.add_argument(
             "-e",
@@ -295,7 +298,7 @@ def sanity_check(self, args: Namespace) -> None:
         if args.launch_attempts < 1:
             self.parser.error("--launch-attempts must be >= 1")
 
-        args.log_level = self._level_map[args.log_level]
+        args.log_level = self.LEVEL_MAP[args.log_level]
 
         if args.log_limit < 0:
             self.parser.error("--log-limit must be >= 0")

grizzly/common/runner.py

@@ -344,10 +344,9 @@ def run(
         # add all files that were served (includes, etc...) to test
         existing = set(testcase.required)
         for url, local_file in served.items():
-            if url in existing:
-                continue
-            # use copy here so include files are copied
-            testcase.add_from_file(local_file, file_name=url, copy=True)
+            if url not in existing:
+                # copy include files
+                testcase.add_from_file(local_file, file_name=url, copy=True)
         # record use of https in testcase
         testcase.https = self._server.scheme == "https"
         if result.timeout:

grizzly/common/status_reporter.py

@@ -147,7 +147,7 @@ def __len__(self) -> int:
 
     def __str__(self) -> str:
         return "\n".join(
-            [f"Log: '{self.log_file.name}'"] + self.prev_lines + self.lines
+            [f"Log: '{self.log_file.name}'", *self.prev_lines, *self.lines]
         )
 
 
@@ -600,7 +600,7 @@ def summary(
 
         if self.tracebacks:
             txt = self._merge_tracebacks(self.tracebacks, self.SUMMARY_LIMIT - len(msg))
-            msg = "".join((msg, txt))
+            msg = f"{msg}{txt}"
         return msg
 
 

grizzly/reduce/strategies/beautify.py

@@ -11,7 +11,7 @@
 from abc import ABC, abstractmethod
 from logging import getLogger
 from pathlib import Path
-from typing import Generator, List, Match, Optional, Set, Tuple, cast
+from typing import FrozenSet, Generator, List, Match, Optional, Tuple, cast
 
 from lithium.testcases import TestcaseLine
 
@@ -64,8 +64,8 @@ class _BeautifyStrategy(Strategy, ABC):
         tag_name: Tag name to search for in other (non-native) extensions.
     """
 
-    all_extensions: Set[str]
-    ignore_files = {TEST_INFO, "prefs.js"}
+    all_extensions: FrozenSet[str]
+    ignore_files = frozenset((TEST_INFO, "prefs.js"))
     import_available: bool
     import_name: str
     native_extension: str
@@ -257,10 +257,7 @@ def __iter__(self) -> Generator[List[TestCase], None, None]:
                 lith_tc.dump(file)
                 continue
 
-            testcases: List[TestCase] = []
-            for test in sorted(self._testcase_root.iterdir()):
-                testcases.append(TestCase.load(test))
-            yield testcases
+            yield [TestCase.load(x) for x in sorted(self._testcase_root.iterdir())]
 
             assert self._current_feedback is not None, "No feedback for last iteration"
             if self._current_feedback:
@@ -279,7 +276,7 @@ class CSSBeautify(_BeautifyStrategy):
     definitions.
     """
 
-    all_extensions = {".css", ".htm", ".html", ".xhtml"}
+    all_extensions = frozenset((".css", ".htm", ".html", ".xhtml"))
     import_available = HAVE_CSSBEAUTIFIER
     import_name = "cssbeautifier"
     name = "cssbeautify"
@@ -318,7 +315,7 @@ class JSBeautify(_BeautifyStrategy):
     compound statements.
     """
 
-    all_extensions = {".js", ".htm", ".html", ".xhtml"}
+    all_extensions = frozenset((".js", ".htm", ".html", ".xhtml"))
     import_available = HAVE_JSBEAUTIFIER
     import_name = "jsbeautifier"
     name = "jsbeautify"

grizzly/reduce/strategies/lithium.py

@@ -47,10 +47,12 @@ def __init__(self, testcases: List[TestCase]) -> None:
         self._current_feedback: Optional[bool] = None
         self._current_reducer: Optional[ReductionIterator] = None
         self._files_to_reduce: List[Path] = []
-        local_tests: List[TestCase] = []
-        for test in sorted(self._testcase_root.iterdir()):
-            local_tests.append(TestCase.load(test, catalog=True))
-        self.rescan_files_to_reduce(local_tests)
+        self.rescan_files_to_reduce(
+            [
+                TestCase.load(x, catalog=True)
+                for x in sorted(self._testcase_root.iterdir())
+            ]
+        )
 
     def rescan_files_to_reduce(self, testcases: List[TestCase]) -> None:
         """Repopulate the private `files_to_reduce` attribute by scanning the testcase
@@ -128,9 +130,10 @@ def __iter__(self) -> Generator[List[TestCase], None, None]:
 
             for reduction in self._current_reducer:
                 reduction.dump()
-                testcases: List[TestCase] = []
-                for test in sorted(self._testcase_root.iterdir()):
-                    testcases.append(TestCase.load(test, catalog=True))
+                testcases = [
+                    TestCase.load(x, catalog=True)
+                    for x in sorted(self._testcase_root.iterdir())
+                ]
                 LOG.info("[%s] %s", self.name, self._current_reducer.description)
                 yield testcases
                 if not self._current_feedback:

grizzly/reduce/strategies/testcases.py

@@ -65,9 +65,7 @@ def __iter__(self) -> Generator[List[TestCase], None, None]:
         """
         assert self._current_feedback is None
         idx = 0
-        testcases: List[TestCase] = []
-        for test in sorted(self._testcase_root.iterdir()):
-            testcases.append(TestCase.load(test))
+        testcases = [TestCase.load(x) for x in sorted(self._testcase_root.iterdir())]
         n_testcases = len(testcases)
         while True:
             if n_testcases <= 1:

grizzly/reduce/test_strategies.py

@@ -31,10 +31,7 @@ def test_strategy_tc_load(tmp_path, is_hang):
 
     class _TestStrategy(Strategy):
         def __iter__(self):
-            testcases = []
-            for test in sorted(self._testcase_root.iterdir()):
-                testcases.append(TestCase.load(test))
-            yield testcases
+            yield [TestCase.load(x) for x in sorted(self._testcase_root.iterdir())]
 
         def update(self, success):
             pass

grizzly/test_args.py

@@ -22,7 +22,7 @@ def test_common_args_01(mocker, tmp_path, extra_args, results):
     mocker.patch("grizzly.args.scan_plugins", return_value=["targ"])
     fake_bin = tmp_path / "fake.bin"
     fake_bin.touch()
-    cmd = [str(fake_bin), "--platform", "targ"] + extra_args
+    cmd = [str(fake_bin), "--platform", "targ", *extra_args]
     args = vars(CommonArgs().parse_args(argv=cmd))
     assert args
     for arg_name, expected_value in results.items():
@@ -138,7 +138,7 @@ def test_common_args_03(capsys, mocker, tmp_path, args, msg, targets):
     fake_bin.touch()
     fake_bin.with_suffix(fake_bin.suffix + ".fuzzmanagerconf").touch()
     with raises(SystemExit):
-        CommonArgs().parse_args(argv=[str(fake_bin)] + args)
+        CommonArgs().parse_args(argv=[str(fake_bin), *args])
     assert msg in capsys.readouterr()[-1]
 
 
@@ -209,7 +209,7 @@ def test_grizzly_args_01(mocker, tmp_path, extra_args, results):
     )
     fake_bin = tmp_path / "fake.bin"
     fake_bin.touch()
-    cmd = [str(fake_bin), "adpt", "--platform", "targ"] + extra_args
+    cmd = [str(fake_bin), "adpt", "--platform", "targ", *extra_args]
     args = vars(GrizzlyArgs().parse_args(argv=cmd))
     assert args
     for arg_name, expected_value in results.items():
@@ -269,7 +269,7 @@ def test_grizzly_args_04(capsys, mocker, tmp_path, args, msg):
     fake_bin.touch()
     with raises(SystemExit):
         GrizzlyArgs().parse_args(
-            argv=[str(fake_bin), "adpt", "--platform", "targ"] + args
+            argv=[str(fake_bin), "adpt", "--platform", "targ", *args]
         )
     assert msg in capsys.readouterr()[-1]
 
@@ -297,6 +297,6 @@ def test_grizzly_args_05(capsys, mocker, tmp_path, args, msg):
     fake_bin.touch()
     with raises(SystemExit):
         GrizzlyArgs().parse_args(
-            argv=[str(fake_bin), "adpt", "--platform", "targ"] + args
+            argv=[str(fake_bin), "adpt", "--platform", "targ", *args]
         )
     assert msg in capsys.readouterr()[-1]

grizzly/test_main.py

@@ -54,7 +54,7 @@ def test_main_01(mocker, session_setup, adpt_relaunch, extra_args):
     adapter.RELAUNCH = adpt_relaunch
 
     # use __file__ as "binary" since it is not used
-    cmd = [__file__, "adpt", "--platform", "targ"] + extra_args
+    cmd = [__file__, "adpt", "--platform", "targ", *extra_args]
     args = GrizzlyArgs().parse_args(cmd)
     session_obj.status.results.total = 1 if args.smoke_test else 0
 

loki/loki.py

@@ -22,7 +22,7 @@
 
 
 class Loki:
-    BYTE_ORDERS = {"<", ">", "@", "!", "="}
+    BYTE_ORDERS = frozenset(("<", ">", "@", "!", "="))
 
     __slots__ = ("aggr", "byte_order")
 

pyproject.toml

@@ -63,4 +63,16 @@ filterwarnings = [
 ]
 log_level = "DEBUG"
 
+[tool.ruff.lint]
+select = [
+    # Pyflakes
+    "F",
+    # Flynt
+    "FLY",
+    # Perflint
+    "PERF",
+    # Ruff-specific rules
+    "RUF",
+]
+
 [tool.setuptools_scm]

sapphire/core.py

@@ -10,7 +10,7 @@
 from socket import SO_REUSEADDR, SOL_SOCKET, gethostname, socket
 from ssl import PROTOCOL_TLS_SERVER, SSLContext, SSLSocket
 from time import perf_counter, sleep
-from typing import Any, Callable, Dict, Iterable, Optional, Tuple, Union, cast
+from typing import Any, Callable, Iterable, Mapping, Optional, Tuple, Union, cast
 
 from .certificate_bundle import CertificateBundle
 from .connection_manager import ConnectionManager
@@ -200,7 +200,7 @@ def serve_path(
         forever: bool = False,
         required_files: Optional[Iterable[str]] = None,
         server_map: Optional[ServerMap] = None,
-    ) -> Tuple[Served, Dict[str, Path]]:
+    ) -> Tuple[Served, Mapping[str, Path]]:
         """Serve files in path.
 
         The status codes include:
@@ -216,7 +216,7 @@ def serve_path(
                 This is meant to be used with continue_cb.
             required_files: Files that need to be served in order to exit the
                 serve loop.
-            server_map (ServerMap):
+            server_map: Map of server includes, dynamic requests and redirects.
 
         Returns:
             Status code and files served.

sapphire/job.py

@@ -13,7 +13,8 @@
 from pathlib import Path
 from queue import Queue
 from threading import Event, Lock
-from typing import Any, Dict, Iterable, NamedTuple, Optional, Set, Tuple, Union, cast
+from types import MappingProxyType
+from typing import Any, Iterable, Mapping, NamedTuple, Optional, Set, Tuple, Union, cast
 
 from .server_map import DynamicResource, FileResource, RedirectResource, ServerMap
 
@@ -50,13 +51,15 @@ class ServedTracker(NamedTuple):
 class Job:
     # MIME_MAP is used to support new or uncommon mime types.
     # Definitions in here take priority over mimetypes.guess_type().
-    MIME_MAP = {
-        ".avif": "image/avif",
-        ".bmp": "image/bmp",
-        ".ico": "image/x-icon",
-        ".wave": "audio/x-wav",
-        ".webp": "image/webp",
-    }
+    MIME_MAP = MappingProxyType(
+        {
+            ".avif": "image/avif",
+            ".bmp": "image/bmp",
+            ".ico": "image/x-icon",
+            ".wave": "audio/x-wav",
+            ".webp": "image/webp",
+        }
+    )
 
     __slots__ = (
         "_complete",
@@ -260,7 +263,7 @@ def remove_pending(self, file_name: str) -> bool:
             return not self._pending.files
 
     @property
-    def served(self) -> Dict[str, Path]:
+    def served(self) -> Mapping[str, Path]:
         """Served files.
 
         Args:
@@ -270,7 +273,7 @@ def served(self) -> Dict[str, Path]:
             Mapping of URLs to files on disk.
         """
         with self._served.lock:
-            return {entry.url: entry.target for entry in self._served.files}
+            return MappingProxyType({x.url: x.target for x in self._served.files})
 
     @property
     def status(self) -> Served: