Enable more ruff

Collector/Collector.py

@@ -83,7 +83,7 @@ def refreshFromZip(self, zipFileName: str) -> None:
 
             # Now clean the signature directory, only deleting signatures and metadata
             for sigFile in os.listdir(self.sigCacheDir):
-                if sigFile.endswith(".signature") or sigFile.endswith(".metadata"):
+                if sigFile.endswith((".signature", ".metadata")):
                     os.remove(os.path.join(self.sigCacheDir, sigFile))
                 else:
                     print(
@@ -210,16 +210,15 @@ def search(self, crashInfo: CrashInfo) -> tuple[str | None, dict[str, Any] | Non
             sigFile = os.path.join(self.sigCacheDir, sigFile)
             if not os.path.isdir(sigFile):
                 with open(sigFile) as f:
-                    sigData = f.read()
-                    crashSig = CrashSignature(sigData)
-                    if crashSig.matches(crashInfo):
-                        metadataFile = sigFile.replace(".signature", ".metadata")
-                        metadata: dict[str, Any] | None = None
-                        if os.path.exists(metadataFile):
-                            with open(metadataFile) as m:
-                                metadata = json.loads(m.read())
+                    crashSig = CrashSignature(f.read())
+                if crashSig.matches(crashInfo):
+                    metadataFile = sigFile.replace(".signature", ".metadata")
+                    metadata: dict[str, Any] | None = None
+                    if os.path.exists(metadataFile):
+                        with open(metadataFile) as m:
+                            metadata = json.load(m)
 
-                        return (sigFile, metadata)
+                    return (sigFile, metadata)
 
         return (None, None)
 
@@ -772,22 +771,22 @@ def main(args: list[str] | None = None) -> int:
                 "Command line arguments:",
                 " ".join(args),
             )
-            print("")
+            print()
 
         if retJSON.get("env"):
             env = json.loads(retJSON["env"])
             print(
                 "Environment variables:",
                 " ".join(f"{k} = {v}" for (k, v) in env.items()),
             )
-            print("")
+            print()
 
         if retJSON.get("metadata"):
             metadata = json.loads(retJSON["metadata"])
             print("== Metadata ==")
             for k, v in metadata.items():
                 print(f"{k} = {v}")
-            print("")
+            print()
 
         print(retFile)
         return 0

FTB/AssertionHelper.py

@@ -282,7 +282,11 @@ def getSanitizedAssertionPattern(msgs: str | list[str]) -> str | list[str]:
                 matchPattern if replacementText is None else replacementText
             )
 
-            def _handleMatch(match: re.Match[str]) -> str:
+            def _handleMatch(
+                match: re.Match[str],
+                replacementPattern: str = replacementPattern,
+                bsPositions: list[int] = bsPositions,
+            ) -> str:
                 start = match.start(0)
                 end = match.end(0)
                 lengthDiff = len(replacementPattern) - len(match.group(0))

FTB/CoverageHelper.py

@@ -74,7 +74,7 @@ def merge_recursive(r: dict[str, Any], s: dict[str, Any]) -> None:
 
             minlen = min(len(rc), len(sc))
 
-            for idx in range(0, minlen):
+            for idx in range(minlen):
                 # There are multiple situations where coverage reports might disagree
                 # about which lines are coverable and which are not. Sometimes, GCOV
                 # reports this wrong in headers, but it can also happen when mixing

FTB/Running/GDB.py

@@ -87,7 +87,7 @@ def printImportantRegisters() -> None:
         ]
     elif isARM64():
         # ARM64 has GPRs from x0 to x30
-        regs = ["x" + str(x) for x in range(0, 31)]
+        regs = ["x" + str(x) for x in range(31)]
         regs.extend(["sp", "pc", "cpsr", "fpcsr", "fpcr"])
     else:
         regs = ["eax", "ebx", "ecx", "edx", "esi", "edi", "ebp", "esp", "eip"]

FTB/Running/PersistentApplication.py

@@ -23,7 +23,7 @@
 import signal
 import subprocess
 import time
-from abc import ABCMeta
+from abc import ABCMeta, abstractmethod
 from enum import Enum, IntEnum, auto
 
 from FTB.Running.StreamCollector import StreamCollector
@@ -119,16 +119,19 @@ def __init__(
         self.spfpPrefix = ""
         self.spfpSuffix = ""  # To support <!-- -->
 
+    @abstractmethod
     def start(self, test: str | None = None) -> int | None:
         pass
 
+    @abstractmethod
     def stop(self) -> None:
         pass
 
+    @abstractmethod
     def runTest(self, test: str) -> int | None:
         pass
 
-    def status(self) -> int | None:
+    def status(self) -> int | None:  # noqa: B027
         pass
 
     def _crashed(self) -> bool:
@@ -278,17 +281,17 @@ def start(self, test: str | None = None) -> int | None:
                     f"{self.spfpPrefix}spfp-selftest{self.spfpSuffix}",
                     file=self.process.stdin,
                 )
-            except OSError:
+            except OSError as exc:
                 raise RuntimeError(
                     "SPFP Error: Selftest failed, application did not start properly."
-                )
+                ) from exc
 
             try:
                 response = self.responseQueue.get(
                     block=True, timeout=self.processingTimeout
                 )
-            except queue.Empty:
-                raise RuntimeError("SPFP Error: Selftest failed, no response.")
+            except queue.Empty as exc:
+                raise RuntimeError("SPFP Error: Selftest failed, no response.") from exc
 
             if response != "PASSED":
                 raise RuntimeError(
@@ -362,7 +365,7 @@ def runTest(self, test: str) -> int | None:
                 response = self.responseQueue.get(
                     block=True, timeout=self.processingTimeout
                 )
-            except queue.Empty:
+            except queue.Empty as exc:
                 if self.process.poll() is None:
                     # The process is still running, force it to stop and return timeout
                     # code
@@ -384,14 +387,14 @@ def runTest(self, test: str) -> int | None:
                     raise RuntimeError(
                         "SPFP Error: Application terminated with signal: "
                         f"{self.process.returncode}"
-                    )
+                    ) from exc
                 # The application exited, but didn't send us any message before
                 # doing so. We consider this a protocol violation and raise an
                 # exception.
                 raise RuntimeError(
                     "SPFP Error: Application exited without message. "
                     f"Exitcode: {self.process.returncode}"
-                )
+                ) from exc
 
             # Update stdout/err available for the last run
             assert self.outCollector is not None

FTB/Running/tests/test_persistent.py

@@ -150,7 +150,7 @@ def _check(spa):
             oldPid = spa.process.pid
             startTime = time.time()
 
-            for i in range(1, 10000):
+            for _ in range(1, 10000):
                 spa.runTest("aaa\naaaa")
 
             stopTime = time.time()

FTB/Signatures/CrashInfo.py

@@ -155,7 +155,7 @@ def __init__(self, *args: Any, **kwds: Any) -> None:
         super().__init__(*args, **kwds)
 
 
-class CrashInfo(metaclass=ABCMeta):
+class CrashInfo(metaclass=ABCMeta):  # noqa: B024
     """
     Abstract base class that provides a method to instantiate the right sub class.
     It also supports generating a CrashSignature based on the stored information.
@@ -525,7 +525,7 @@ def createCrashSignature(
         # StackFramesSymptom is only supported in 1.2 and higher,
         # for everything else, use multiple stackFrame symptoms
         if minimumSupportedVersion < 12:
-            for idx in range(0, numFrames):
+            for idx in range(numFrames):
                 functionName = self.backtrace[idx]
                 if functionName != "??":
                     symptomArr.append(
@@ -541,7 +541,7 @@ def createCrashSignature(
         else:
             framesArray: list[str] = []
 
-            for idx in range(0, numFrames):
+            for idx in range(numFrames):
                 functionName = self.backtrace[idx]
                 if functionName != "??":
                     framesArray.append(functionName)
@@ -552,7 +552,7 @@ def createCrashSignature(
                         topStackMissCount += 1
 
             lastSymbolizedFrame = None
-            for frameIdx in range(0, len(framesArray)):
+            for frameIdx in range(len(framesArray)):
                 if framesArray[frameIdx] != "?":
                     lastSymbolizedFrame = frameIdx
 
@@ -712,7 +712,7 @@ def __init__(
         self.configuration = configuration
 
         # If crashData is given, use that to find the ASan trace, otherwise use stderr
-        asanOutput = crashData if crashData else stderr
+        asanOutput = crashData or stderr
         assert asanOutput is not None
 
         asanCrashAddressPattern = r"""(?x)
@@ -951,7 +951,7 @@ def __init__(
         self.configuration = configuration
 
         # If crashData is given, use that to find the LSan trace, otherwise use stderr
-        lsanOutput = crashData if crashData else stderr
+        lsanOutput = crashData or stderr
         assert lsanOutput is not None
         lsanErrorPattern = "ERROR: LeakSanitizer:"
         lsanPatternSeen = False
@@ -1048,7 +1048,7 @@ def __init__(
         self.configuration = configuration
 
         # If crashData is given, use that to find the UBSan trace, otherwise use stderr
-        ubsanOutput = crashData if crashData else stderr
+        ubsanOutput = crashData or stderr
         assert ubsanOutput is not None
         ubsanErrorPattern = r":\d+:\d+:\s+runtime\s+error:\s+"
         ubsanPatternSeen = False
@@ -1538,7 +1538,7 @@ def calculateARMDerefOpAddress(
 
             # ARM assembly has nested comma-separated operands, so we need to merge
             # those inside  brackets back together before proceeding.
-            for i in range(0, len(parts)):
+            for i in range(len(parts)):
                 if i >= len(parts):
                     break
                 if (
@@ -1559,9 +1559,7 @@ def calculateARMDerefOpAddress(
                 if instruction == "brk":
                     # This is an explicit breakpoint / trap
                     return RegisterHelper.getInstructionPointer(registerMap)
-            elif len(parts) == 2 and (
-                instruction.startswith("ldr") or instruction.startswith("str")
-            ):
+            elif len(parts) == 2 and (instruction.startswith(("ldr", "str"))):
                 # Load/Store instruction
                 match = re.match("^\\s*\\[(.*)\\]$", parts[1])
                 if match is not None:
@@ -1991,7 +1989,7 @@ def __init__(
         self.configuration = configuration
 
         # If crashData is given, use that to find the ASan trace, otherwise use stderr
-        tsanOutput = crashData if crashData else stderr
+        tsanOutput = crashData or stderr
         assert tsanOutput is not None
 
         tsanWarningPattern = r"""WARNING: ThreadSanitizer:.*\s.+?\s+\(pid=\d+\)"""
@@ -2160,7 +2158,7 @@ def __init__(
 
         # If crashData is given, use that to find the Valgrind trace, otherwise use
         # stderr
-        vgdOutput = crashData if crashData else stderr
+        vgdOutput = crashData or stderr
         assert vgdOutput is not None
         stackPattern = re.compile(
             r"""
@@ -2211,7 +2209,7 @@ def createShortSignature(self) -> str:
         @return: A string representing this crash (short signature)
         """
 
-        logData = self.rawCrashData if self.rawCrashData else self.rawStderr
+        logData = self.rawCrashData or self.rawStderr
         for line in logData:
             m = re.match(ValgrindCrashInfo.MSG_REGEX, line)
             if m and m.group("msg"):

FTB/Signatures/CrashSignature.py

@@ -53,7 +53,7 @@ def __init__(self, rawSignature: str) -> None:
         try:
             obj = json.loads(rawSignature)
         except ValueError as e:
-            raise RuntimeError(f"Invalid JSON: {e}")
+            raise RuntimeError(f"Invalid JSON: {e}") from e
 
         # Get the symptoms objects (mandatory)
         if "symptoms" in obj:
@@ -254,12 +254,7 @@ def getSignatureUnifiedDiffTuples(
         signatureDiff = difflib.unified_diff(oldLines, newLines, n=context)
 
         for diffLine in signatureDiff:
-            if (
-                diffLine.startswith("+++")
-                or diffLine.startswith("---")
-                or diffLine.startswith("@@")
-                or not diffLine.strip()
-            ):
+            if diffLine.startswith(("+++", "---", "@@")) or not diffLine.strip():
                 continue
 
             diffTuples.append((diffLine[0], diffLine[1:]))

FTB/Signatures/Matchers.py

@@ -48,7 +48,7 @@ def __init__(self, obj: str | bytes | dict[str, Any]) -> None:
                 try:
                     self.compiledValue = re.compile(self.value)
                 except re.error as e:
-                    raise RuntimeError(f"Error in regular expression: {e}")
+                    raise RuntimeError(f"Error in regular expression: {e}") from e
         else:
             value = JSONHelper.getStringChecked(obj, "value", True)
             assert value is not None
@@ -63,7 +63,7 @@ def __init__(self, obj: str | bytes | dict[str, Any]) -> None:
                     try:
                         self.compiledValue = re.compile(self.value)
                     except re.error as e:
-                        raise RuntimeError(f"Error in regular expression: {e}")
+                        raise RuntimeError(f"Error in regular expression: {e}") from e
                 else:
                     raise RuntimeError(f"Unknown match operator specified: {matchType}")
 
@@ -122,19 +122,19 @@ def __init__(self, obj: str | bytes | int) -> None:
                     matchType = numberMatchComponents[0]
                     try:
                         self.matchType = NumberMatchType(matchType)
-                    except ValueError:
+                    except ValueError as exc:
                         raise RuntimeError(
                             f"Unknown match operator specified: {matchType}"
-                        )
+                        ) from exc
 
                 try:
                     value = numberMatchComponents[numIdx]
                     base = 16 if value.startswith("0x") else 10
                     self.value = int(numberMatchComponents[numIdx], base)
-                except ValueError:
+                except ValueError as exc:
                     raise RuntimeError(
                         f"Invalid number specified: {numberMatchComponents[numIdx]}"
-                    )
+                    ) from exc
             else:
                 # We're trying to match the fact that we cannot calculate a crash
                 # address

FTB/Signatures/RegisterHelper.py

@@ -56,7 +56,7 @@
     "cpsr",
 ]
 
-arm64Registers = ["x" + str(x) for x in range(0, 31)] + [
+arm64Registers = ["x" + str(x) for x in range(31)] + [
     "sp",
     "pc",
     "cpsr",

Reporter/Reporter.py

@@ -153,7 +153,7 @@ def sentry_init() -> None:
         sentry_fuzzing_config.init()
 
 
-class Reporter(ABC):
+class Reporter(ABC):  # noqa: B024
     def __init__(
         self,
         sigCacheDir: str | None = None,

TaskStatusReporter/TaskStatusReporter.py

@@ -39,9 +39,8 @@
 class TaskStatusReporter(Reporter):
     @functools.wraps(Reporter.__init__)
     def __init__(self, *args, **kwargs):
-        kwargs.setdefault(
-            "tool", "N/A"
-        )  # tool is required by remote_checks, but unused by TaskStatusReporter
+        # 'tool' is required by remote_checks, but unused by TaskStatusReporter
+        kwargs.setdefault("tool", "N/A")
         super().__init__(*args, **kwargs)
 
     @staticmethod