Upstream basic support for red knot

mypy_primer/globals.py

@@ -80,7 +80,7 @@ def parse_options(argv: list[str]) -> _Args:
     type_checker_group.add_argument(
         "--type-checker",
         default="mypy",
-        choices=["mypy", "pyright"],
+        choices=["mypy", "pyright", "knot"],
         help="type checker to use",
     )
     type_checker_group.add_argument(

mypy_primer/main.py

@@ -8,7 +8,7 @@
 import traceback
 from dataclasses import replace
 from pathlib import Path
-from typing import Awaitable, Callable, Iterator, TypeVar
+from typing import Any, Awaitable, Callable, Iterator, TypeVar
 
 from mypy_primer.git_utils import (
     RevisionLike,
@@ -18,24 +18,28 @@
 from mypy_primer.globals import _Args, parse_options_and_set_ctx
 from mypy_primer.model import Project, TypeCheckResult
 from mypy_primer.projects import get_projects
-from mypy_primer.type_checker import setup_mypy, setup_pyright, setup_typeshed
+from mypy_primer.type_checker import setup_knot, setup_mypy, setup_pyright, setup_typeshed
 from mypy_primer.utils import Style, debug_print, get_npm, line_count, run, strip_colour_code
 
 T = TypeVar("T")
 
 
 def setup_type_checker(ARGS: _Args, *, revision_like: RevisionLike, suffix: str) -> Awaitable[Path]:
     setup_fn: Callable[..., Awaitable[Path]]
+    kwargs: dict[str, Any]
+
     if ARGS.type_checker == "mypy":
         setup_fn = setup_mypy
-        arg_names = ["repo", "mypyc_compile_level"]
+        kwargs = {"repo": ARGS.repo, "mypyc_compile_level": ARGS.mypyc_compile_level}
     elif ARGS.type_checker == "pyright":
         setup_fn = setup_pyright
-        arg_names = ["repo"]
+        kwargs = {"repo": ARGS.repo}
+    elif ARGS.type_checker == "knot":
+        setup_fn = setup_knot
+        kwargs = {"repo": ARGS.repo}
     else:
         raise ValueError(f"Unknown type checker {ARGS.type_checker}")
 
-    kwargs = {arg_name: getattr(ARGS, arg_name) for arg_name in arg_names}
     return setup_fn(
         ARGS.base_dir / f"{ARGS.type_checker}_{suffix}", revision_like=revision_like, **kwargs
     )
@@ -97,8 +101,12 @@ def select_projects(ARGS: _Args) -> list[Project]:
         for p in get_projects()
         if not (p.min_python_version and sys.version_info < p.min_python_version)
     )
+
     if ARGS.type_checker == "pyright":
         project_iter = iter(p for p in project_iter if p.pyright_cmd is not None)
+    # if ARGS.type_checker == "knot":
+    #     project_iter = iter(p for p in project_iter if p.knot_cmd is not None)
+
     if ARGS.project_selector:
         project_iter = iter(
             p for p in project_iter if re.search(ARGS.project_selector, p.location, flags=re.I)

mypy_primer/model.py

@@ -27,6 +27,7 @@ class Project:
 
     mypy_cmd: str
     pyright_cmd: str | None
+    knot_cmd: str | None = None  # TODO: remove this default
     paths: list[str] | None = None
 
     install_cmd: str | None = None
@@ -174,14 +175,14 @@ async def run_mypy(
         self, mypy: Path, typeshed_dir: Path | None, prepend_path: Path | None
     ) -> TypeCheckResult:
         env = os.environ.copy()
-        env["MYPY_FORCE_COLOR"] = "1"
+        additional_flags = ctx.get().additional_flags.copy()
 
         mypy_path = []  # TODO: this used to be exposed, could be useful to expose it again
-        additional_flags = ctx.get().additional_flags.copy()
         if typeshed_dir is not None:
             additional_flags.append(f"--custom-typeshed-dir={quote_path(typeshed_dir)}")
             mypy_path += list(map(str, typeshed_dir.glob("stubs/*")))
 
+        env["MYPY_FORCE_COLOR"] = "1"
         if "MYPYPATH" in env:
             mypy_path = env["MYPYPATH"].split(os.pathsep) + mypy_path
         env["MYPYPATH"] = os.pathsep.join(mypy_path)
@@ -253,6 +254,7 @@ async def run_pyright(
     ) -> TypeCheckResult:
         env = os.environ.copy()
         additional_flags = ctx.get().additional_flags.copy()
+
         if typeshed_dir is not None:
             additional_flags.append(f"--typeshedpath {quote_path(typeshed_dir)}")
         if prepend_path is not None:
@@ -283,13 +285,72 @@ async def run_pyright(
             runtime=runtime,
         )
 
+    def get_knot_cmd(self, knot: Path, additional_flags: Sequence[str] = ()) -> str:
+        knot_cmd = self.knot_cmd
+        if knot_cmd is None:
+            knot_cmd = "{knot} check {paths}" if self.paths else "{knot} check"
+        assert "{knot}" in knot_cmd
+        if additional_flags:
+            knot_cmd += " " + " ".join(additional_flags)
+
+        knot_cmd = knot_cmd.format_map(_FormatMap(knot=knot, paths=self.paths))
+
+        knot_cmd += f" --python {quote_path(self.venv.dir)} --output-format concise"
+        return knot_cmd
+
+    async def run_knot(
+        self, knot: Path, typeshed_dir: Path | None, prepend_path: Path | None
+    ) -> TypeCheckResult:
+        env = os.environ.copy()
+        additional_flags = ctx.get().additional_flags.copy()
+
+        if typeshed_dir is not None:
+            additional_flags += ["--typeshed", quote_path(typeshed_dir)]
+        if prepend_path is not None:
+            env["MYPY_PRIMER_PREPEND_PATH"] = str(prepend_path)
+
+        env["CLICOLOR_FORCE"] = "1"
+
+        knot_cmd = self.get_knot_cmd(knot, additional_flags)
+        proc, runtime = await run(
+            knot_cmd,
+            shell=True,
+            output=True,
+            check=False,
+            cwd=ctx.get().projects_dir / self.name,
+            env=env,
+        )
+        if ctx.get().debug:
+            debug_print(f"{Style.BLUE}{knot} on {self.name} took {runtime:.2f}s{Style.RESET}")
+
+        if proc.returncode not in (0, 1):
+            debug_print(proc.stderr + proc.stdout)
+            if proc.returncode == 2:
+                raise RuntimeError(
+                    "Red Knot exited with code 2 which may indicate an internal problem (e.g. IO error)"
+                )
+            else:
+                raise RuntimeError("Red Knot did not exit with code 0, 1 or 2. Panic?")
+
+        output = proc.stderr + proc.stdout
+
+        return TypeCheckResult(
+            knot_cmd,
+            output=output,
+            success=not bool(proc.returncode),
+            expected_success="knot" in self.expected_success,
+            runtime=runtime,
+        )
+
     async def run_typechecker(
         self, type_checker: Path, typeshed_dir: Path | None, *, prepend_path: Path | None
     ) -> TypeCheckResult:
         if ctx.get().type_checker == "mypy":
             return await self.run_mypy(type_checker, typeshed_dir, prepend_path)
         elif ctx.get().type_checker == "pyright":
             return await self.run_pyright(type_checker, typeshed_dir, prepend_path)
+        elif ctx.get().type_checker == "knot":
+            return await self.run_knot(type_checker, typeshed_dir, prepend_path)
         else:
             raise ValueError(f"Unknown type checker: {ctx.get().type_checker}")
 

mypy_primer/type_checker.py

@@ -98,6 +98,41 @@ async def setup_pyright(
     return pyright_exe
 
 
+async def setup_knot(
+    knot_dir: Path,
+    revision_like: RevisionLike,
+    *,
+    repo: str | None,
+) -> Path:
+    knot_dir.mkdir(parents=True, exist_ok=True)
+
+    if repo is None:
+        repo = "https://github.com/astral-sh/ruff"
+    repo_dir = await ensure_repo_at_revision(repo, knot_dir, revision_like)
+
+    cargo_target_dir = knot_dir / "target"
+    if not os.environ.get("MYPY_PRIMER_NO_REBUILD", False):
+        env = os.environ.copy()
+        env["CARGO_TARGET_DIR"] = str(cargo_target_dir)
+
+        try:
+            await run(
+                ["cargo", "build", "--bin", "red_knot"],
+                cwd=repo_dir,
+                env=env,
+                output=True,
+            )
+        except subprocess.CalledProcessError as e:
+            print("Error while building 'knot'")
+            print(e.stdout)
+            print(e.stderr)
+            raise e
+
+    knot_exe = cargo_target_dir / "debug" / "red_knot"
+    assert knot_exe.exists()
+    return knot_exe
+
+
 async def setup_typeshed(parent_dir: Path, *, repo: str, revision_like: RevisionLike) -> Path:
     if parent_dir.exists():
         shutil.rmtree(parent_dir)