don't autofix noqa'd errors, add --disable-noqa, add non-passing test that noqa always works

Author: jakkdl

flake8_trio/__init__.py

@@ -150,29 +150,26 @@ def from_source(cls, source: str) -> Plugin:
         return plugin
 
     def run(self) -> Iterable[Error]:
-        problems_ast = Flake8TrioRunner.run(self._tree, self.options)
+        if not self.standalone:
+            self.options.disable_noqa = True
 
         cst_runner = Flake8TrioRunner_cst(self.options, self.module)
-        problems_cst = cst_runner.run()
+        yield from cst_runner.run()
+        # update saved module so modified source code can be accessed when autofixing
+        self.module = cst_runner.module
 
-        # when run as a flake8 plugin, flake8 handles suppressing errors from `noqa`.
-        # it's therefore important we don't suppress any errors for compatibility with
-        # flake8-noqa
-        if not self.standalone:
+        problems_ast = Flake8TrioRunner.run(self._tree, self.options)
+        if self.options.disable_noqa:
             yield from problems_ast
-            yield from problems_cst
             return
 
-        for problem in (*problems_ast, *problems_cst):
-            noqa = cst_runner.state.noqas.get(problem.line)
+        for problem in problems_ast:
+            noqa = cst_runner.noqas.get(problem.line)
             # if there's a noqa comment, and it's bare or this code is listed in it
             if noqa is not None and (noqa == set() or problem.code in noqa):
                 continue
             yield problem
 
-        # update saved module so modified source code can be accessed when autofixing
-        self.module = cst_runner.module
-
     @staticmethod
     def add_options(option_manager: OptionManager | ArgumentParser):
         if isinstance(option_manager, ArgumentParser):
@@ -184,6 +181,16 @@ def add_options(option_manager: OptionManager | ArgumentParser):
                 dest="files",
                 help="Files(s) to format, instead of autodetection.",
             )
+            add_argument(
+                "--disable-noqa",
+                required=False,
+                default=False,
+                action="store_true",
+                help=(
+                    'Disable the effect of "# noqa". This will report errors on '
+                    'lines with "# noqa" at the end.'
+                ),
+            )
         else:  # if run as a flake8 plugin
             Plugin.standalone = False
             # Disable TRIO9xx calls by default
@@ -326,6 +333,7 @@ def get_matching_codes(
             startable_in_context_manager=options.startable_in_context_manager,
             trio200_blocking_calls=options.trio200_blocking_calls,
             anyio=options.anyio,
+            disable_noqa=options.disable_noqa,
         )
 
 

flake8_trio/base.py

@@ -21,6 +21,7 @@ class Options:
     startable_in_context_manager: Collection[str]
     trio200_blocking_calls: dict[str, str]
     anyio: bool
+    disable_noqa: bool
 
 
 class Statement(NamedTuple):

flake8_trio/runner.py

@@ -32,7 +32,6 @@
 class SharedState:
     options: Options
     problems: list[Error] = field(default_factory=list)
-    noqas: dict[int, set[str]] = field(default_factory=dict)
     library: tuple[str, ...] = ()
     typed_calls: dict[str, str] = field(default_factory=dict)
     variables: dict[str, str] = field(default_factory=dict)
@@ -113,22 +112,30 @@ class Flake8TrioRunner_cst(__CommonRunner):
     def __init__(self, options: Options, module: Module):
         super().__init__(options)
         self.options = options
+        self.noqas: dict[int, set[str]] = {}
 
         # Could possibly enable/disable utility visitors here, if visitors declared
         # dependencies
         self.utility_visitors: tuple[Flake8TrioVisitor_cst, ...] = tuple(
             v(self.state) for v in utility_visitors_cst
         )
 
+        # sort the error classes to get predictable behaviour when multiple autofixers
+        # are enabled
+        sorted_error_classes_cst = sorted(ERROR_CLASSES_CST, key=lambda x: x.__name__)
         self.visitors: tuple[Flake8TrioVisitor_cst, ...] = tuple(
-            v(self.state) for v in ERROR_CLASSES_CST if self.selected(v.error_codes)
+            v(self.state)
+            for v in sorted_error_classes_cst
+            if self.selected(v.error_codes)
         )
         self.module = module
 
     def run(self) -> Iterable[Error]:
-        if not self.visitors:
-            return
         for v in (*self.utility_visitors, *self.visitors):
             self.module = cst.MetadataWrapper(self.module).visit(v)
 
         yield from self.state.problems
+
+        # expose the noqa's parsed by the last visitor, so they can be used to filter
+        # ast problems
+        self.noqas = v.noqas  # type: ignore[reportUnboundVariable]

flake8_trio/visitors/flake8triovisitor.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import ast
+import functools
+import re
 from abc import ABC
 from typing import TYPE_CHECKING, Any, Union
 
@@ -13,6 +15,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
+    from re import Match
 
     from ..runner import SharedState
 
@@ -167,7 +170,7 @@ def __init__(self, shared_state: SharedState):
         self.__state = shared_state
 
         self.options = self.__state.options
-        self.noqas = self.__state.noqas
+        self.noqas: dict[int, set[str]] = {}
 
     def get_state(self, *attrs: str, copy: bool = False) -> dict[str, Any]:
         # require attrs, since we inherit a *ton* of stuff which we don't want to copy
@@ -197,12 +200,19 @@ def save_state(self, node: cst.CSTNode, *attrs: str, copy: bool = False):
     def restore_state(self, node: cst.CSTNode):
         self.set_state(self.outer.pop(node, {}))
 
+    def is_noqa(self, node: cst.CSTNode, code: str):
+        if self.options.disable_noqa:
+            return False
+        pos = self.get_metadata(PositionProvider, node).start
+        noqas = self.noqas.get(pos.line)
+        return noqas is not None and (noqas == set() or code in noqas)
+
     def error(
         self,
         node: cst.CSTNode,
         *args: str | Statement | int,
         error_code: str | None = None,
-    ):
+    ) -> bool:
         if error_code is None:
             assert (
                 len(self.error_codes) == 1
@@ -211,9 +221,12 @@ def error(
         # don't emit an error if this code is disabled in a multi-code visitor
         # TODO: write test for only one of 910/911 enabled/autofixed
         elif error_code[:7] not in self.options.enabled_codes:
-            return  # pragma: no cover
-        pos = self.get_metadata(PositionProvider, node).start
+            return False  # pragma: no cover
 
+        if self.is_noqa(node, error_code):
+            return False
+
+        pos = self.get_metadata(PositionProvider, node).start
         self.__state.problems.append(
             Error(
                 # 7 == len('TRIO...'), so alt messages raise the original code
@@ -224,13 +237,60 @@ def error(
                 *args,
             )
         )
+        return True
 
-    def should_autofix(self, code: str | None = None):
+    def should_autofix(self, node: cst.CSTNode, code: str | None = None) -> bool:
         if code is None:
             assert len(self.error_codes) == 1
             code = next(iter(self.error_codes))
+        if self.is_noqa(node, code):
+            return False
         return code in self.options.autofix_codes
 
+    # These must not be overridden without calling super() on them.
+    # TODO: find a good way of enforcing that statically, or add a test param that
+    # all errors work with noqa, or do fancy metaclass stuff
+    # https://stackoverflow.com/questions/75033760/require-overriding-method-to-call-super
+    # or just #YOLO
+
+    def visit_SimpleStatementLine(self, node: cst.SimpleStatementLine):
+        super().visit_SimpleStatementLine(node)
+        self.save_state(node, "noqas")
+        self.parse_noqa(node.trailing_whitespace.comment)
+
+    def leave_SimpleStatementLine(
+        self,
+        original_node: cst.SimpleStatementLine,
+        updated_node: cst.SimpleStatementLine,
+    ):
+        super_updated_node = super().leave_SimpleStatementLine(
+            original_node, updated_node
+        )
+        self.restore_state(original_node)
+        return super_updated_node  # noqa: R504
+
+    def visit_SimpleStatementSuite(self, node: cst.SimpleStatementSuite):
+        self.save_state(node, "noqas")
+        self.parse_noqa(node.trailing_whitespace.comment)
+
+    def leave_SimpleStatementSuite(
+        self,
+        original_node: cst.SimpleStatementSuite,
+        updated_node: cst.SimpleStatementSuite,
+    ):
+        self.restore_state(original_node)
+        return updated_node
+
+    def visit_IndentedBlock(self, node: cst.IndentedBlock):
+        self.save_state(node, "noqas")
+        self.parse_noqa(node.header.comment)
+
+    def leave_IndentedBlock(
+        self, original_node: cst.IndentedBlock, updated_node: cst.IndentedBlock
+    ):
+        self.restore_state(original_node)
+        return updated_node
+
     @property
     def library(self) -> tuple[str, ...]:
         return self.__state.library if self.__state.library else ("trio",)
@@ -240,3 +300,56 @@ def library(self) -> tuple[str, ...]:
     def add_library(self, name: str) -> None:
         if name not in self.__state.library:
             self.__state.library = self.__state.library + (name,)
+
+    def parse_noqa(self, node: cst.Comment | None):
+        if not node:
+            return
+        noqa_match = _find_noqa(node.value)
+        if noqa_match is None:
+            return
+
+        codes_str = noqa_match.groupdict()["codes"]
+
+        line = self.get_metadata(PositionProvider, node).start.line
+
+        assert (
+            line not in self.noqas
+        ), "it should not be possible to have two [noqa] comments on the same line"
+
+        # blanket noqa
+        if codes_str is None:
+            # this also includes a non-blanket noqa with a list of invalid codes
+            # so one should maybe instead specifically look for no `:`
+            self.noqas[line] = set()
+            return
+        # split string on ",", strip of whitespace, and save in set if non-empty
+        # TODO: Check that code exists
+        self.noqas[line] = {
+            item_strip for item in codes_str.split(",") if (item_strip := item.strip())
+        }
+
+
+# taken from
+# https://github.com/PyCQA/flake8/blob/d016204366a22d382b5b56dc14b6cbff28ce929e/src/flake8/defaults.py#L27
+NOQA_INLINE_REGEXP = re.compile(
+    # We're looking for items that look like this:
+    # ``# noqa``
+    # ``# noqa: E123``
+    # ``# noqa: E123,W451,F921``
+    # ``# noqa:E123,W451,F921``
+    # ``# NoQA: E123,W451,F921``
+    # ``# NOQA: E123,W451,F921``
+    # ``# NOQA:E123,W451,F921``
+    # We do not want to capture the ``: `` that follows ``noqa``
+    # We do not care about the casing of ``noqa``
+    # We want a comma-separated list of errors
+    # upstream links to an old version on regex101
+    # https://regex101.com/r/4XUuax/5 full explanation of the regex
+    r"# noqa(?::[\s]?(?P<codes>([A-Z]+[0-9]+(?:[,\s]+)?)+))?",
+    re.IGNORECASE,
+)
+
+
+@functools.lru_cache(maxsize=512)
+def _find_noqa(physical_line: str) -> Match[str] | None:
+    return NOQA_INLINE_REGEXP.search(physical_line)

flake8_trio/visitors/visitor100.py

@@ -58,7 +58,7 @@ def leave_With(
             for res in self.node_dict[original_node]:
                 self.error(res.node, res.base, res.function)
 
-            if self.should_autofix() and len(updated_node.items) == 1:
+            if self.should_autofix(original_node) and len(updated_node.items) == 1:
                 return flatten_preserving_comments(updated_node)
 
         return updated_node