Support error codes from plugins in options

Mypy has options for enabling or disabling specific error codes. These
work fine, except that it is not possible to enable or disable error
codes from plugins, only mypy's original error codes.

The crux of the issue is that mypy validates and rejects unknown error
codes passed in the options before it loads plugins and learns about the
any error codes that might get registered.

There are many ways to solve this. This commit tries to find a pragmatic
solution where the relevant options parsing is deferred until after
plugin loading. Error code validation in the config parser, where
plugins are not loaded yet, is also skipped entirely, since the error
code options are re-validated later anyway.

This fixes #12987.

Author: sigvef

mypy/__main__.py

@@ -22,7 +22,7 @@ def console_entry() -> None:
         os.dup2(devnull, sys.stdout.fileno())
         sys.exit(2)
     except KeyboardInterrupt:
-        _, options = process_options(args=sys.argv[1:])
+        _, options, _ = process_options(args=sys.argv[1:])
         if options.show_traceback:
             sys.stdout.write(traceback.format_exc())
         formatter = FancyFormatter(sys.stdout, sys.stderr, False)

mypy/build.py

@@ -153,6 +153,7 @@ def build(
     stdout: TextIO | None = None,
     stderr: TextIO | None = None,
     extra_plugins: Sequence[Plugin] | None = None,
+    on_plugins_loaded: Callable[[], None] | None = None,
 ) -> BuildResult:
     """Analyze a program.
 
@@ -192,7 +193,7 @@ def default_flush_errors(
 
     try:
         result = _build(
-            sources, options, alt_lib_path, flush_errors, fscache, stdout, stderr, extra_plugins
+            sources, options, alt_lib_path, flush_errors, fscache, stdout, stderr, extra_plugins, on_plugins_loaded=on_plugins_loaded
         )
         result.errors = messages
         return result
@@ -216,6 +217,7 @@ def _build(
     stdout: TextIO,
     stderr: TextIO,
     extra_plugins: Sequence[Plugin],
+    on_plugins_loaded: Callable[[], None] | None = None
 ) -> BuildResult:
     if platform.python_implementation() == "CPython":
         # Run gc less frequently, as otherwise we can spent a large fraction of
@@ -239,6 +241,8 @@ def _build(
     errors = Errors(options, read_source=lambda path: read_py_file(path, cached_read))
     plugin, snapshot = load_plugins(options, errors, stdout, extra_plugins)
 
+    on_plugins_loaded()
+
     # Add catch-all .gitignore to cache dir if we created it
     cache_dir_existed = os.path.isdir(options.cache_dir)
 

mypy/config_parser.py

@@ -87,15 +87,6 @@ def complain(x: object, additional_info: str = "") -> Never:
         complain(v)
 
 
-def validate_codes(codes: list[str]) -> list[str]:
-    invalid_codes = set(codes) - set(error_codes.keys())
-    if invalid_codes:
-        raise argparse.ArgumentTypeError(
-            f"Invalid error code(s): {', '.join(sorted(invalid_codes))}"
-        )
-    return codes
-
-
 def validate_package_allow_list(allow_list: list[str]) -> list[str]:
     for p in allow_list:
         msg = f"Invalid allow list entry: {p}"
@@ -209,8 +200,8 @@ def split_commas(value: str) -> list[str]:
         [p.strip() for p in split_commas(s)]
     ),
     "enable_incomplete_feature": lambda s: [p.strip() for p in split_commas(s)],
-    "disable_error_code": lambda s: validate_codes([p.strip() for p in split_commas(s)]),
-    "enable_error_code": lambda s: validate_codes([p.strip() for p in split_commas(s)]),
+    "disable_error_code": lambda s: [p.strip() for p in split_commas(s)],
+    "enable_error_code": lambda s: [p.strip() for p in split_commas(s)],
     "package_root": lambda s: [p.strip() for p in split_commas(s)],
     "cache_dir": expand_path,
     "python_executable": expand_path,
@@ -234,8 +225,8 @@ def split_commas(value: str) -> list[str]:
         "always_false": try_split,
         "untyped_calls_exclude": lambda s: validate_package_allow_list(try_split(s)),
         "enable_incomplete_feature": try_split,
-        "disable_error_code": lambda s: validate_codes(try_split(s)),
-        "enable_error_code": lambda s: validate_codes(try_split(s)),
+        "disable_error_code": lambda s: try_split(s),
+        "enable_error_code": lambda s: try_split(s),
         "package_root": try_split,
         "exclude": str_or_array_as_list,
         "packages": try_split,

mypy/dmypy_server.py

@@ -136,7 +136,7 @@ def daemonize(
 
 
 def process_start_options(flags: list[str], allow_sources: bool) -> Options:
-    _, options = mypy.main.process_options(
+    _, options, _ = mypy.main.process_options(
         ["-i"] + flags, require_targets=False, server_options=True
     )
     if options.report_dirs:
@@ -326,7 +326,7 @@ def cmd_run(
             # capture stderr so the client can report it
             with redirect_stderr(stderr):
                 with redirect_stdout(stdout):
-                    sources, options = mypy.main.process_options(
+                    sources, options, _ = mypy.main.process_options(
                         ["-i"] + list(args),
                         require_targets=True,
                         server_options=True,

mypy/main.py

@@ -11,7 +11,7 @@
 from collections.abc import Sequence
 from gettext import gettext
 from io import TextIOWrapper
-from typing import IO, TYPE_CHECKING, Any, Final, NoReturn, TextIO
+from typing import IO, TYPE_CHECKING, Any, Callable, Final, NoReturn, TextIO
 
 from mypy import build, defaults, state, util
 from mypy.config_parser import (
@@ -86,7 +86,7 @@ def main(
         stdout.reconfigure(errors="backslashreplace")
 
     fscache = FileSystemCache()
-    sources, options = process_options(args, stdout=stdout, stderr=stderr, fscache=fscache)
+    sources, options, on_plugins_loaded = process_options(args, stdout=stdout, stderr=stderr, fscache=fscache)
     if clean_exit:
         options.fast_exit = False
 
@@ -124,7 +124,7 @@ def main(
         install_types(formatter, options, non_interactive=options.non_interactive)
         return
 
-    res, messages, blockers = run_build(sources, options, fscache, t0, stdout, stderr)
+    res, messages, blockers = run_build(sources, options, fscache, t0, stdout, stderr, on_plugins_loaded=on_plugins_loaded)
 
     if options.non_interactive:
         missing_pkgs = read_types_packages_to_install(options.cache_dir, after_run=True)
@@ -182,6 +182,7 @@ def run_build(
     t0: float,
     stdout: TextIO,
     stderr: TextIO,
+    on_plugins_loaded: Callable[[], None] | None = None
 ) -> tuple[build.BuildResult | None, list[str], bool]:
     formatter = util.FancyFormatter(
         stdout, stderr, options.hide_error_codes, hide_success=bool(options.output)
@@ -208,7 +209,7 @@ def flush_errors(filename: str | None, new_messages: list[str], serious: bool) -
     try:
         # Keep a dummy reference (res) for memory profiling afterwards, as otherwise
         # the result could be freed.
-        res = build.build(sources, options, None, flush_errors, fscache, stdout, stderr)
+        res = build.build(sources, options, None, flush_errors, fscache, stdout, stderr, on_plugins_loaded=on_plugins_loaded)
     except CompileError as e:
         blockers = True
         if not e.use_stdout:
@@ -1347,13 +1348,14 @@ def process_options(
     fscache: FileSystemCache | None = None,
     program: str = "mypy",
     header: str = HEADER,
-) -> tuple[list[BuildSource], Options]:
+) -> tuple[list[BuildSource], Options, Callable[[], None]]:
     """Parse command line arguments.
 
     If a FileSystemCache is passed in, and package_root options are given,
     call fscache.set_package_root() to set the cache's package root.
 
-    Returns a tuple of: a list of source files, an Options collected from flags.
+    Returns a tuple of: a list of source files, an Options collected from
+    flags, and a callback to be called once plugins have loaded.
     """
     stdout = stdout if stdout is not None else sys.stdout
     stderr = stderr if stderr is not None else sys.stderr
@@ -1456,9 +1458,14 @@ def set_strict_flags() -> None:
     validate_package_allow_list(options.untyped_calls_exclude)
     validate_package_allow_list(options.deprecated_calls_exclude)
 
-    options.process_error_codes(error_callback=parser.error)
     options.process_incomplete_features(error_callback=parser.error, warning_callback=print)
 
+    def on_plugins_loaded() -> None:
+        # Processing error codes after plugins have loaded since plugins may
+        # register custom error codes that we don't know about until plugins
+        # have loaded.
+        options.process_error_codes(error_callback=parser.error)
+
     # Compute absolute path for custom typeshed (if present).
     if options.custom_typeshed_dir is not None:
         options.abs_custom_typeshed_dir = os.path.abspath(options.custom_typeshed_dir)
@@ -1550,7 +1557,7 @@ def set_strict_flags() -> None:
         # exceptions of different types.
         except InvalidSourceList as e2:
             fail(str(e2), stderr, options)
-    return targets, options
+    return targets, options, on_plugins_loaded
 
 
 def process_package_roots(

mypy/test/helpers.py

@@ -343,7 +343,7 @@ def parse_options(
     if flags:
         flag_list = flags.group(1).split()
         flag_list.append("--no-site-packages")  # the tests shouldn't need an installed Python
-        targets, options = process_options(flag_list, require_targets=False)
+        targets, options, _ = process_options(flag_list, require_targets=False)
         if targets:
             # TODO: support specifying targets via the flags pragma
             raise RuntimeError("Specifying targets via the flags pragma is not supported.")

mypy/test/testargs.py

@@ -18,7 +18,7 @@
 class ArgSuite(Suite):
     def test_coherence(self) -> None:
         options = Options()
-        _, parsed_options = process_options([], require_targets=False)
+        _, parsed_options, _ = process_options([], require_targets=False)
         # FIX: test this too. Requires changing working dir to avoid finding 'setup.cfg'
         options.config_file = parsed_options.config_file
         assert_equal(options.snapshot(), parsed_options.snapshot())
@@ -31,12 +31,12 @@ def test_executable_inference(self) -> None:
 
         # test inference given one (infer the other)
         matching_version = base + [f"--python-version={sys_ver_str}"]
-        _, options = process_options(matching_version)
+        _, options, _ = process_options(matching_version)
         assert options.python_version == sys.version_info[:2]
         assert options.python_executable == sys.executable
 
         matching_version = base + [f"--python-executable={sys.executable}"]
-        _, options = process_options(matching_version)
+        _, options, _ = process_options(matching_version)
         assert options.python_version == sys.version_info[:2]
         assert options.python_executable == sys.executable
 
@@ -45,13 +45,13 @@ def test_executable_inference(self) -> None:
             f"--python-version={sys_ver_str}",
             f"--python-executable={sys.executable}",
         ]
-        _, options = process_options(matching_version)
+        _, options, _ = process_options(matching_version)
         assert options.python_version == sys.version_info[:2]
         assert options.python_executable == sys.executable
 
         # test that --no-site-packages will disable executable inference
         matching_version = base + [f"--python-version={sys_ver_str}", "--no-site-packages"]
-        _, options = process_options(matching_version)
+        _, options, _ = process_options(matching_version)
         assert options.python_version == sys.version_info[:2]
         assert options.python_executable is None
 

mypyc/build.py

@@ -126,7 +126,7 @@ def get_mypy_config(
     fscache: FileSystemCache | None,
 ) -> tuple[list[BuildSource], list[BuildSource], Options]:
     """Construct mypy BuildSources and Options from file and options lists"""
-    all_sources, options = process_options(mypy_options, fscache=fscache)
+    all_sources, options, _ = process_options(mypy_options, fscache=fscache)
     if only_compile_paths is not None:
         paths_set = set(only_compile_paths)
         mypyc_sources = [s for s in all_sources if s.path in paths_set]

test-data/unit/check-plugin-error-codes.test

@@ -0,0 +1,76 @@
+[case test_custom_error_code_from_plugin_is_targetable]
+# flags: --show-error-codes
+# mypy.ini: [mypy]
+# mypy.ini: plugins = my_dummy_plugin
+
+[file my_dummy_plugin.py]
+from __future__ import annotations
+from mypy.plugin import Plugin, FunctionContext
+from mypy.errorcodes import ErrorCode
+
+DUMMY_PLUGIN_ERROR = ErrorCode("dummy-plugin-error", "Dummy plugin error", "Plugin")
+
+class DummyPlugin(Plugin):
+    def get_function_hook(self, fullname: str):
+        return lambda ctx: ctx.api.fail("this is a plugin error", ctx.context, code=DUMMY_PLUGIN_ERROR)
+
+def plugin(version: str): return DummyPlugin
+
+[file main.py]
+def fn(x: int) -> int: ...
+fn("x")
+
+[out]
+main.py:2: error: this is a plugin error  [dummy-plugin-error]
+Found 1 error in 1 file (checked 1 source file)
+
+[case test_custom_error_code_can_be_disabled]
+# flags: --show-error-codes --disable-error-code=dummy-plugin-error
+# mypy.ini: [mypy]
+# mypy.ini: plugins = my_dummy_plugin
+
+[file my_dummy_plugin.py]
+from __future__ import annotations
+from mypy.plugin import Plugin, FunctionContext
+from mypy.errorcodes import ErrorCode
+
+DUMMY_PLUGIN_ERROR = ErrorCode("dummy-plugin-error", "Dummy plugin error", "Plugin")
+
+class DummyPlugin(Plugin):
+    def get_function_hook(self, fullname: str):
+        return lambda ctx: ctx.api.fail("this is a plugin error", ctx.context, code=DUMMY_PLUGIN_ERROR)
+
+def plugin(version: str): return DummyPlugin
+
+[file main.py]
+def fn(x: int) -> int: ...
+fn("x")  # no output expected when disabled
+
+[out]
+Success: no issues found in 1 source file
+
+[case test_custom_error_code_can_be_reenabled]
+# flags: --show-error-codes --disable-error-code=dummy-plugin-error --enable-error-code=dummy-plugin-error
+# mypy.ini: [mypy]
+# mypy.ini: plugins = my_dummy_plugin
+
+[file my_dummy_plugin.py]
+from __future__ import annotations
+from mypy.plugin import Plugin, FunctionContext
+from mypy.errorcodes import ErrorCode
+
+DUMMY_PLUGIN_ERROR = ErrorCode("dummy-plugin-error", "Dummy plugin error", "Plugin")
+
+class DummyPlugin(Plugin):
+    def get_function_hook(self, fullname: str):
+        return lambda ctx: ctx.api.fail("this is a plugin error", ctx.context, code=DUMMY_PLUGIN_ERROR)
+
+def plugin(version: str): return DummyPlugin
+
+[file main.py]
+def fn(x: int) -> int: ...
+fn("x")
+
+[out]
+main.py:2: error: this is a plugin error  [dummy-plugin-error]
+Found 1 error in 1 file (checked 1 source file)