red-knot: adapt fuzz-parser to work with red-knot (#14566)

Co-authored-by: Alex Waygood <alex.waygood@gmail.com>

.github/workflows/ci.yaml

@@ -309,7 +309,7 @@ jobs:
           # Make executable, since artifact download doesn't preserve this
           chmod +x ${{ steps.download-cached-binary.outputs.download-path }}/ruff
 
-          python scripts/fuzz-parser/fuzz.py 0-500 --test-executable ${{ steps.download-cached-binary.outputs.download-path }}/ruff
+          python scripts/fuzz-parser/fuzz.py --bin ruff 0-500 --test-executable ${{ steps.download-cached-binary.outputs.download-path }}/ruff
 
   scripts:
     name: "test scripts"

.github/workflows/daily_fuzz.yaml

@@ -49,7 +49,7 @@ jobs:
         # but this is outweighed by the fact that a release build takes *much* longer to compile in CI
         run: cargo build --locked
       - name: Fuzz
-        run: python scripts/fuzz-parser/fuzz.py $(shuf -i 0-9999999999999999999 -n 1000) --test-executable target/debug/ruff
+        run: python scripts/fuzz-parser/fuzz.py --bin ruff $(shuf -i 0-9999999999999999999 -n 1000) --test-executable target/debug/ruff
 
   create-issue-on-failure:
     name: Create an issue if the daily fuzz surfaced any bugs

scripts/fuzz-parser/fuzz.py

@@ -1,30 +1,33 @@
 """
-Run the parser on randomly generated (but syntactically valid) Python source-code files.
+Run a Ruff executable on randomly generated (but syntactically valid)
+Python source-code files.
 
 To install all dependencies for this script into an environment using `uv`, run:
     uv pip install -r scripts/fuzz-parser/requirements.txt
 
 Example invocations of the script:
-- Run the fuzzer using seeds 0, 1, 2, 78 and 93 to generate the code:
+- Run the fuzzer on Ruff's parser using seeds 0, 1, 2, 78 and 93 to generate the code:
-  `python scripts/fuzz-parser/fuzz.py 0-2 78 93`
+  `python scripts/fuzz-parser/fuzz.py --bin ruff 0-2 78 93`
 - Run the fuzzer concurrently using seeds in range 0-10 inclusive,
   but only reporting bugs that are new on your branch:
-  `python scripts/fuzz-parser/fuzz.py 0-10 --new-bugs-only`
+  `python scripts/fuzz-parser/fuzz.py --bin ruff 0-10 --new-bugs-only`
 - Run the fuzzer concurrently on 10,000 different Python source-code files,
   using a random selection of seeds, and only print a summary at the end
   (the `shuf` command is Unix-specific):
-  `python scripts/fuzz-parser/fuzz.py $(shuf -i 0-1000000 -n 10000) --quiet
+  `python scripts/fuzz-parser/fuzz.py --bin ruff $(shuf -i 0-1000000 -n 10000) --quiet
 """
 
 from __future__ import annotations
 
 import argparse
 import concurrent.futures
-import os.path
+import enum
 import subprocess
+import tempfile
 from dataclasses import KW_ONLY, dataclass
 from functools import partial
-from typing import NewType
+from pathlib import Path
+from typing import NewType, assert_never
 
 from pysource_codegen import generate as generate_random_code
 from pysource_minimize import minimize as minimize_repro
@@ -36,7 +39,20 @@ Seed = NewType("Seed", int)
 ExitCode = NewType("ExitCode", int)
 
 
-def contains_bug(code: str, *, ruff_executable: str) -> bool:
+def redknot_contains_bug(code: str, *, red_knot_executable: Path) -> bool:
+    """Return `True` if the code triggers a panic in type-checking code."""
+    with tempfile.TemporaryDirectory() as tempdir:
+        Path(tempdir, "pyproject.toml").write_text('[project]\n\tname = "fuzz-input"')
+        Path(tempdir, "input.py").write_text(code)
+        completed_process = subprocess.run(
+            [red_knot_executable, "--current-directory", tempdir],
+            capture_output=True,
+            text=True,
+        )
+    return completed_process.returncode != 0 and completed_process.returncode != 1
+
+
+def ruff_contains_bug(code: str, *, ruff_executable: Path) -> bool:
     """Return `True` if the code triggers a parser error."""
     completed_process = subprocess.run(
         [ruff_executable, "check", "--config", "lint.select=[]", "--no-cache", "-"],
@@ -47,16 +63,33 @@ def contains_bug(code: str, *, ruff_executable: str) -> bool:
     return completed_process.returncode != 0
 
 
+def contains_bug(code: str, *, executable: Executable, executable_path: Path) -> bool:
+    """Return `True` if the code triggers an error."""
+    match executable:
+        case Executable.RUFF:
+            return ruff_contains_bug(code, ruff_executable=executable_path)
+        case Executable.RED_KNOT:
+            return redknot_contains_bug(code, red_knot_executable=executable_path)
+        case _ as unreachable:
+            assert_never(unreachable)
+
+
 def contains_new_bug(
-    code: str, *, test_executable: str, baseline_executable: str
+    code: str,
+    *,
+    executable: Executable,
+    test_executable_path: Path,
+    baseline_executable_path: Path,
 ) -> bool:
     """Return `True` if the code triggers a *new* parser error.
 
     A "new" parser error is one that exists with `test_executable`,
     but did not exist with `baseline_executable`.
     """
-    return contains_bug(code, ruff_executable=test_executable) and not contains_bug(
+    return contains_bug(
-        code, ruff_executable=baseline_executable
+        code, executable=executable, executable_path=test_executable_path
+    ) and not contains_bug(
+        code, executable=executable, executable_path=baseline_executable_path
     )
 
 
@@ -68,6 +101,8 @@ class FuzzResult:
     # If we found a bug, this will be the minimum Python code
     # required to trigger the bug. If not, it will be `None`.
     maybe_bug: MinimizedSourceCode | None
+    # The executable we're testing
+    executable: Executable
 
     def print_description(self, index: int, num_seeds: int) -> None:
         """Describe the results of fuzzing the parser with this seed."""
@@ -78,38 +113,47 @@ class FuzzResult:
             else colored(f"Ran fuzzer successfully on seed {self.seed}", "green")
         )
         print(f"{msg:<60} {progress:>15}", flush=True)
+
         if self.maybe_bug:
-            print(colored("The following code triggers a bug:", "red"))
+            match self.executable:
+                case Executable.RUFF:
+                    panic_message = "The following code triggers a parser bug:"
+                case Executable.RED_KNOT:
+                    panic_message = "The following code triggers a red-knot panic:"
+                case _ as unreachable:
+                    assert_never(unreachable)
+
+            print(colored(panic_message, "red"))
             print()
             print(self.maybe_bug)
             print(flush=True)
 
 
-def fuzz_code(
+def fuzz_code(seed: Seed, args: ResolvedCliArgs) -> FuzzResult:
-    seed: Seed,
-    *,
-    test_executable: str,
-    baseline_executable: str,
-    only_new_bugs: bool,
-) -> FuzzResult:
     """Return a `FuzzResult` instance describing the fuzzing result from this seed."""
     code = generate_random_code(seed)
     has_bug = (
         contains_new_bug(
             code,
-            test_executable=test_executable,
+            executable=args.executable,
-            baseline_executable=baseline_executable,
+            test_executable_path=args.test_executable_path,
+            baseline_executable_path=args.baseline_executable_path,
+        )
+        if args.baseline_executable_path is not None
+        else contains_bug(
+            code, executable=args.executable, executable_path=args.test_executable_path
         )
-        if only_new_bugs
-        else contains_bug(code, ruff_executable=test_executable)
     )
     if has_bug:
-        maybe_bug = MinimizedSourceCode(
+        callback = partial(
-            minimize_repro(code, partial(contains_bug, ruff_executable=test_executable))
+            contains_bug,
+            executable=args.executable,
+            executable_path=args.test_executable_path,
         )
+        maybe_bug = MinimizedSourceCode(minimize_repro(code, callback))
     else:
         maybe_bug = None
-    return FuzzResult(seed, maybe_bug)
+    return FuzzResult(seed, maybe_bug, args.executable)
 
 
 def run_fuzzer_concurrently(args: ResolvedCliArgs) -> list[FuzzResult]:
@@ -121,14 +165,7 @@ def run_fuzzer_concurrently(args: ResolvedCliArgs) -> list[FuzzResult]:
     bugs: list[FuzzResult] = []
     with concurrent.futures.ProcessPoolExecutor() as executor:
         fuzz_result_futures = [
-            executor.submit(
+            executor.submit(fuzz_code, seed, args) for seed in args.seeds
-                fuzz_code,
-                seed,
-                test_executable=args.test_executable,
-                baseline_executable=args.baseline_executable,
-                only_new_bugs=args.only_new_bugs,
-            )
-            for seed in args.seeds
         ]
         try:
             for i, future in enumerate(
@@ -155,12 +192,7 @@ def run_fuzzer_sequentially(args: ResolvedCliArgs) -> list[FuzzResult]:
     )
     bugs: list[FuzzResult] = []
     for i, seed in enumerate(args.seeds, start=1):
-        fuzz_result = fuzz_code(
+        fuzz_result = fuzz_code(seed, args)
-            seed,
-            test_executable=args.test_executable,
-            baseline_executable=args.baseline_executable,
-            only_new_bugs=args.only_new_bugs,
-        )
         if not args.quiet:
             fuzz_result.print_description(i, num_seeds)
         if fuzz_result.maybe_bug:
@@ -173,7 +205,7 @@ def main(args: ResolvedCliArgs) -> ExitCode:
         bugs = run_fuzzer_sequentially(args)
     else:
         bugs = run_fuzzer_concurrently(args)
-    noun_phrase = "New bugs" if args.only_new_bugs else "Bugs"
+    noun_phrase = "New bugs" if args.baseline_executable_path is not None else "Bugs"
     if bugs:
         print(colored(f"{noun_phrase} found in the following seeds:", "red"))
         print(*sorted(bug.seed for bug in bugs))
@@ -206,13 +238,18 @@ def parse_seed_argument(arg: str) -> int | range:
     return int(arg)
 
 
+class Executable(enum.StrEnum):
+    RUFF = "ruff"
+    RED_KNOT = "red_knot"
+
+
 @dataclass(slots=True)
 class ResolvedCliArgs:
     seeds: list[Seed]
     _: KW_ONLY
-    test_executable: str
+    executable: Executable
-    baseline_executable: str
+    test_executable_path: Path
-    only_new_bugs: bool
+    baseline_executable_path: Path | None
     quiet: bool
 
 
@@ -232,7 +269,7 @@ def parse_args() -> ResolvedCliArgs:
         action="store_true",
         help=(
             "Only report bugs if they exist on the current branch, "
-            "but *didn't* exist on the released version of Ruff "
+            "but *didn't* exist on the released version "
             "installed into the Python environment we're running in"
         ),
     )
@@ -244,21 +281,31 @@ def parse_args() -> ResolvedCliArgs:
     parser.add_argument(
         "--test-executable",
         help=(
-            "`ruff` executable to test. "
+            "Executable to test. "
             "Defaults to a fresh build of the currently checked-out branch."
         ),
+        type=Path,
     )
     parser.add_argument(
         "--baseline-executable",
         help=(
-            "`ruff` executable to compare results against. "
+            "Executable to compare results against. "
-            "Defaults to whatever `ruff` version is installed "
+            "Defaults to whatever version is installed "
             "in the Python environment."
         ),
+        type=Path,
+    )
+    parser.add_argument(
+        "--bin",
+        help="Which executable to test.",
+        required=True,
+        choices=[member.value for member in Executable],
     )
 
     args = parser.parse_args()
 
+    executable = Executable(args.bin)
+
     if args.baseline_executable:
         if not args.only_new_bugs:
             parser.error(
@@ -276,34 +323,43 @@ def parse_args() -> ResolvedCliArgs:
             )
     elif args.only_new_bugs:
         try:
-            ruff_version_proc = subprocess.run(
+            version_proc = subprocess.run(
-                ["ruff", "--version"], text=True, capture_output=True, check=True
+                [executable, "--version"], text=True, capture_output=True, check=True
             )
         except FileNotFoundError:
             parser.error(
                 "`--only-new-bugs` was specified without specifying a baseline "
-                "executable, and no released version of Ruff appears to be installed "
+                f"executable, and no released version of `{executable}` appears to be "
-                "in your Python environment"
+                "installed in your Python environment"
             )
         else:
             if not args.quiet:
-                ruff_version = ruff_version_proc.stdout.strip().split(" ")[1]
+                version = version_proc.stdout.strip().split(" ")[1]
                 print(
                     f"`--only-new-bugs` was specified without specifying a baseline "
-                    f"executable; falling back to using `ruff=={ruff_version}` as the "
+                    f"executable; falling back to using `{executable}=={version}` as "
-                    f"baseline (the version of Ruff installed in your current Python "
+                    f"the baseline (the version of `{executable}` installed in your "
-                    f"environment)"
+                    f"current Python environment)"
                 )
-        args.baseline_executable = "ruff"
 
     if not args.test_executable:
         print(
             "Running `cargo build --release` since no test executable was specified...",
             flush=True,
         )
+        cmd: list[str] = [
+            "cargo",
+            "build",
+            "--release",
+            "--locked",
+            "--color",
+            "always",
+            "--bin",
+            executable,
+        ]
         try:
             subprocess.run(
-                ["cargo", "build", "--release", "--locked", "--color", "always"],
+                cmd,
                 check=True,
                 capture_output=True,
                 text=True,
@@ -311,8 +367,8 @@ def parse_args() -> ResolvedCliArgs:
         except subprocess.CalledProcessError as e:
             print(e.stderr)
             raise
-        args.test_executable = os.path.join("target", "release", "ruff")
+        args.test_executable = Path("target", "release", executable)
-        assert os.path.exists(args.test_executable)
+        assert args.test_executable.is_file()
 
     seed_arguments: list[range | int] = args.seeds
     seen_seeds: set[int] = set()
@@ -324,10 +380,10 @@ def parse_args() -> ResolvedCliArgs:
 
     return ResolvedCliArgs(
         sorted(map(Seed, seen_seeds)),
-        only_new_bugs=args.only_new_bugs,
         quiet=args.quiet,
-        test_executable=args.test_executable,
+        executable=executable,
-        baseline_executable=args.baseline_executable,
+        test_executable_path=args.test_executable,
+        baseline_executable_path=args.baseline_executable,
     )