Enable more linting

pyproject.toml

@@ -18,6 +18,7 @@ omit = [
 exclude_lines = [
     "except ImportError:",
     "if __name__ == .__main__.:",
+    "if TYPE_CHECKING:",
     "pragma: no cover",
 ]
 
@@ -56,6 +57,8 @@ log_level = "DEBUG"
 
 [tool.ruff.lint]
 select = [
+    # flake8-comprehensions
+    "C4",
     # pycodestyle
     "E",
     # Pyflakes
@@ -64,11 +67,18 @@ select = [
     "FLY",
     # Perflint
     "PERF",
+    # flake8-simplify
+    "SIM",
+    # flake8-type-checking
+    "TCH",
     # Ruff-specific rules
     "RUF",
     # pycodestyle
     "W",
 ]
-ignore = ["PERF203"]
+ignore = [
+    "PERF203",
+    "SIM117",
+]
 
 [tool.setuptools_scm]

src/ffpuppet/bootstrapper.py

@@ -8,7 +8,7 @@
 from select import select
 from socket import SO_REUSEADDR, SOL_SOCKET, socket
 from time import sleep, time
-from typing import Any, Callable
+from typing import Callable
 
 from .exceptions import BrowserTerminatedError, BrowserTimeoutError, LaunchError
 
@@ -60,7 +60,7 @@ def __init__(self, sock: socket) -> None:
     def __enter__(self) -> Bootstrapper:
         return self
 
-    def __exit__(self, *exc: Any) -> None:
+    def __exit__(self, *exc: object) -> None:
         self.close()
 
     def close(self) -> None:

src/ffpuppet/core.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+from contextlib import suppress
 from enum import IntEnum, unique
 from logging import getLogger
 from os import X_OK, access, getenv
@@ -17,21 +18,17 @@
 from shutil import copyfileobj
 from subprocess import Popen, check_output
 from sys import executable
-from typing import Any, Generator, Pattern
+from typing import Generator, Pattern
 from urllib.request import pathname2url
 
-try:
+with suppress(ImportError):
     # pylint: disable=ungrouped-imports
     from subprocess import CREATE_NEW_PROCESS_GROUP  # type: ignore[attr-defined]
 
     CREATE_SUSPENDED = 0x00000004
-except ImportError:
-    pass
 
-try:
+with suppress(ImportError):
     from xvfbwrapper import Xvfb
-except ImportError:
-    pass
 
 from .bootstrapper import Bootstrapper
 from .checks import CheckLogContents, CheckLogSize, CheckMemoryUsage
@@ -156,7 +153,7 @@ def __init__(
     def __enter__(self) -> FFPuppet:
         return self
 
-    def __exit__(self, *exc: Any) -> None:
+    def __exit__(self, *exc: object) -> None:
         self.clean_up()
 
     def _benign_sanitizer_report(self, report: Path) -> bool:
@@ -610,10 +607,9 @@ def dump_coverage(self, timeout: int = 15) -> None:
         """
         if system() != "Linux":  # pragma: no cover
             raise NotImplementedError("dump_coverage() is not available")
-        if self._proc_tree is not None:
-            if not self._proc_tree.dump_coverage(timeout=timeout):
-                LOG.warning("Timeout writing coverage data")
-                self.close()
+        if self._proc_tree and not self._proc_tree.dump_coverage(timeout=timeout):
+            LOG.warning("Timeout writing coverage data")
+            self.close()
 
     def get_pid(self) -> int | None:
         """Get the browser parent process ID.
@@ -917,8 +913,7 @@ def wait(self, timeout: float | None = None) -> bool:
             True if processes exit before timeout expires otherwise False.
         """
         assert timeout is None or timeout >= 0
-        if self._proc_tree:
-            if self._proc_tree.wait_procs(timeout=timeout) > 0:
-                LOG.debug("wait(timeout=%0.2f) timed out", timeout)
-                return False
+        if self._proc_tree and self._proc_tree.wait_procs(timeout=timeout) > 0:
+            LOG.debug("wait(timeout=%0.2f) timed out", timeout)
+            return False
         return True

src/ffpuppet/helpers.py

@@ -4,6 +4,7 @@
 """ffpuppet helper utilities"""
 from __future__ import annotations
 
+from contextlib import suppress
 from logging import getLogger
 from os import W_OK, access, chmod, environ
 from pathlib import Path
@@ -210,26 +211,22 @@ def files_in_use(files: Iterable[Path]) -> Generator[tuple[Path, int, str]]:
         if system() == "Windows":
             for open_file, pids in pids_by_file().items():
                 for check_file in files:
-                    try:
+                    # samefile() can raise if either file cannot be accessed
+                    # this is triggered on Windows if a file is missing
+                    with suppress(OSError):
                         if check_file.samefile(open_file):
                             for pid in pids:
                                 yield open_file, pid, Process(pid).name()
-                    except OSError:  # pragma: no cover
-                        # samefile() can raise if either file cannot be accessed
-                        # this is triggered on Windows if a file is missing
-                        pass
         else:
             for proc in process_iter(["pid", "name", "open_files"]):
                 if not proc.info["open_files"]:
                     continue
                 for open_file in (Path(x.path) for x in proc.info["open_files"]):
                     for check_file in files:
-                        try:
+                        # samefile() can raise if either file cannot be accessed
+                        with suppress(OSError):
                             if check_file.samefile(open_file):
                                 yield open_file, proc.info["pid"], proc.info["name"]
-                        except OSError:  # pragma: no cover
-                            # samefile() can raise if either file cannot be accessed
-                            pass
 
 
 def onerror(

src/ffpuppet/main.py

@@ -39,7 +39,7 @@ def dump_to_console(log_dir: Path, log_quota: int = 0x8000) -> str:
         Merged log data to be displayed on the console.
     """
 
-    logs = list(x for x in log_dir.iterdir() if x.is_file())
+    logs = [x for x in log_dir.iterdir() if x.is_file()]
     if not logs:
         return ""
     # display stdout and stderr last to avoid the need to scroll back

src/ffpuppet/minidump_parser.py

@@ -38,7 +38,7 @@ def __init__(self, symbols: Path | None = None) -> None:
     def __enter__(self) -> MinidumpParser:
         return self
 
-    def __exit__(self, *exc: Any) -> None:
+    def __exit__(self, *exc: object) -> None:
         self.close()
 
     def _cmd(self, src: Path) -> list[str]:

src/ffpuppet/process_tree.py

@@ -4,13 +4,13 @@
 """ffpuppet process tree module"""
 from __future__ import annotations
 
+from contextlib import suppress
 from logging import getLogger
 from os import getenv
 from pathlib import Path
 from platform import system
-from subprocess import Popen
 from time import perf_counter, sleep
-from typing import Callable, Generator, Iterable, List, Tuple, cast
+from typing import TYPE_CHECKING, Callable, Generator, Iterable, List, Tuple, cast
 
 try:
     from signal import SIGUSR1, Signals
@@ -23,6 +23,9 @@
 
 from .exceptions import TerminateError
 
+if TYPE_CHECKING:
+    from subprocess import Popen
+
 LOG = getLogger(__name__)
 
 
@@ -60,14 +63,12 @@ def _safe_wait_procs(
     deadline = None if timeout is None else perf_counter() + timeout
     while True:
         remaining = None if deadline is None else max(deadline - perf_counter(), 0)
-        try:
+        with suppress(AccessDenied):
             # Python 3.8 is not compatible with __future__.annotations in cast()
             return cast(
                 Tuple[List[Process], List[Process]],
                 wait_procs(procs, timeout=remaining, callback=callback),
             )
-        except AccessDenied:
-            pass
         if deadline is not None and deadline <= perf_counter():
             break
         sleep(0.25)
@@ -98,11 +99,9 @@ def _writing_coverage(procs: list[Process]) -> bool:
         True if processes with open .gcda files are found.
     """
     for proc in procs:
-        try:
+        with suppress(AccessDenied, NoSuchProcess):
             if any(x for x in proc.open_files() if x.path.endswith(".gcda")):
                 return True
-        except (AccessDenied, NoSuchProcess):  # pragma: no cover
-            pass
     return False
 
 
@@ -176,11 +175,9 @@ def dump_coverage(self, timeout: int = 15, idle_wait: int = 2) -> bool:
         signaled = 0
         # send COVERAGE_SIGNAL (SIGUSR1) to browser processes
         for proc in self.processes():
-            try:
+            with suppress(AccessDenied, NoSuchProcess):
                 proc.send_signal(COVERAGE_SIGNAL)
                 signaled += 1
-            except (AccessDenied, NoSuchProcess):  # pragma: no cover
-                pass
         # no processes signaled
         if signaled == 0:
             LOG.debug("coverage signal not sent, no browser processes found")
@@ -296,10 +293,8 @@ def processes(self, recursive: bool = False) -> list[Process]:
             procs.append(self.launcher)
         if self._poll(self.parent) is None:
             procs.append(self.parent)
-        try:
+        with suppress(AccessDenied, NoSuchProcess):
             procs.extend(self.parent.children(recursive=recursive))
-        except (AccessDenied, NoSuchProcess):  # pragma: no cover
-            pass
         return procs
 
     def terminate(self) -> None:
@@ -318,12 +313,10 @@ def terminate(self) -> None:
 
         # try terminating the parent process first, this should be all that is needed
         if self._poll(self.parent) is None:
-            try:
+            with suppress(AccessDenied, NoSuchProcess, TimeoutExpired):
                 LOG.debug("attempting to terminate parent (%d)", self.parent.pid)
                 self.parent.terminate()
                 self.parent.wait(timeout=10)
-            except (AccessDenied, NoSuchProcess, TimeoutExpired):  # pragma: no cover
-                pass
             procs = _safe_wait_procs(procs, timeout=0)[1]
 
         use_kill = False
@@ -335,10 +328,11 @@ def terminate(self) -> None:
             )
             # iterate over processes and call terminate()/kill()
             for proc in procs:
-                try:
-                    proc.kill() if use_kill else proc.terminate()
-                except (AccessDenied, NoSuchProcess):  # pragma: no cover
-                    pass
+                with suppress(AccessDenied, NoSuchProcess):
+                    if use_kill:
+                        proc.kill()
+                    else:
+                        proc.terminate()
             # wait for processes to terminate
             procs = _safe_wait_procs(procs, timeout=30)[1]
             if use_kill:
@@ -348,10 +342,8 @@ def terminate(self) -> None:
         if procs:
             LOG.warning("Processes still running: %d", len(procs))
             for proc in procs:
-                try:
+                with suppress(AccessDenied, NoSuchProcess):
                     LOG.warning("-> %d: %s (%s)", proc.pid, proc.name(), proc.status())
-                except (AccessDenied, NoSuchProcess):  # pragma: no cover
-                    pass
             raise TerminateError("Failed to terminate processes")
 
     def wait(self, timeout: int = 300) -> int:

src/ffpuppet/profile.py

@@ -11,7 +11,6 @@
 from subprocess import STDOUT, CalledProcessError, check_output
 from tempfile import mkdtemp
 from time import time
-from typing import Any
 from xml.etree import ElementTree
 
 from .helpers import certutil_available, certutil_find, onerror
@@ -61,7 +60,7 @@ def __init__(
     def __enter__(self) -> Profile:
         return self
 
-    def __exit__(self, *exc: Any) -> None:
+    def __exit__(self, *exc: object) -> None:
         self.remove()
 
     def __str__(self) -> str:

src/ffpuppet/puppet_logger.py

@@ -4,6 +4,7 @@
 """browser and debugger log management"""
 from __future__ import annotations
 
+from contextlib import suppress
 from logging import getLogger
 from mmap import ACCESS_READ, mmap
 from os import getpid, stat
@@ -12,7 +13,7 @@
 from shutil import copy2, copyfileobj, copytree, rmtree
 from subprocess import STDOUT, CalledProcessError, check_output
 from tempfile import NamedTemporaryFile, mkdtemp
-from typing import IO, Any, Iterator, KeysView
+from typing import IO, Iterator, KeysView
 
 from .helpers import onerror, warn_open
 
@@ -44,7 +45,7 @@ def __init__(self, base_path: str | None = None) -> None:
     def __enter__(self) -> PuppetLogger:
         return self
 
-    def __exit__(self, *exc: Any) -> None:
+    def __exit__(self, *exc: object) -> None:
         self.clean_up()
 
     def add_log(self, log_id: str, logfp: IO[bytes] | None = None) -> IO[bytes]:
@@ -275,15 +276,13 @@ def save_logs(
             rr_trace = self.path / self.PATH_RR / "latest-trace"
             if rr_trace.is_dir():
                 # check logs for rr related issues
-                try:
+                # OSError: in case the file does not exist
+                # ValueError: cannot mmap an empty file on Windows
+                with suppress(OSError, ValueError):
                     with (dest / "log_stderr.txt").open("rb") as lfp:
                         with mmap(lfp.fileno(), 0, access=ACCESS_READ) as lmm:
                             if lmm.find(b"=== Start rr backtrace:") != -1:
                                 LOG.warning("rr traceback detected in stderr log")
-                except (OSError, ValueError):  # pragma: no cover
-                    # OSError: in case the file does not exist
-                    # ValueError: cannot mmap an empty file on Windows
-                    pass
                 if rr_pack and not self._rr_packed:
                     LOG.debug("packing rr trace")
                     try: