don't autofix noqa'd errors, add --disable-noqa, add noqa test parameter

to eval_files

Author: jakkdl

flake8_trio/__init__.py

@@ -150,21 +150,24 @@ def from_source(cls, source: str) -> Plugin:
         return plugin
 
     def run(self) -> Iterable[Error]:
-        problems_ast = Flake8TrioRunner.run(self._tree, self.options)
-
-        cst_runner = Flake8TrioRunner_cst(self.options, self.module)
-        problems_cst = cst_runner.run()
-
         # 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:
+            self.options.disable_noqa = True
+
+        cst_runner = Flake8TrioRunner_cst(self.options, self.module)
+        # any noqa'd errors are suppressed upon being generated
+        yield from cst_runner.run()
+
+        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:
+            # access the stored noqas in cst_runner
+            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
@@ -184,6 +187,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 +339,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

@@ -18,6 +18,7 @@
     utility_visitors,
     utility_visitors_cst,
 )
+from .visitors.visitor_utility import NoqaHandler
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
@@ -113,22 +114,35 @@ class Flake8TrioRunner_cst(__CommonRunner):
     def __init__(self, options: Options, module: Module):
         super().__init__(options)
         self.options = options
+        self.noqas: dict[int, set[str]] = {}
+
+        utility_visitors = utility_visitors_cst.copy()
+        if self.options.disable_noqa:
+            utility_visitors.remove(NoqaHandler)
 
         # 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
+            v(self.state) for v in utility_visitors
         )
 
+        # 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
+        if not self.options.disable_noqa:
+            self.noqas = v.noqas  # type: ignore[reportUnboundVariable]

flake8_trio/visitors/flake8triovisitor.py

@@ -197,12 +197,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 +218,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,11 +234,15 @@ 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))
+        # this does not currently need to check for `noqa`s, as error() does that
+        # and no codes tries to autofix if there's no error. This might change if/when
+        # "printing an error" and "autofixing" becomes independent. See issue #192
         return code in self.options.autofix_codes
 
     @property

flake8_trio/visitors/visitor100.py

@@ -55,10 +55,13 @@ def leave_With(
         self, original_node: cst.With, updated_node: cst.With
     ) -> cst.BaseStatement | cst.FlattenSentinel[cst.BaseStatement]:
         if not self.has_checkpoint_stack.pop():
+            autofix = len(updated_node.items) == 1
             for res in self.node_dict[original_node]:
-                self.error(res.node, res.base, res.function)
+                autofix &= self.error(
+                    res.node, res.base, res.function
+                ) and self.should_autofix(res.node)
 
-            if self.should_autofix() and len(updated_node.items) == 1:
+            if autofix:
                 return flatten_preserving_comments(updated_node)
 
         return updated_node

flake8_trio/visitors/visitor101.py

@@ -60,6 +60,7 @@ def visit_FunctionDef(self, node: cst.FunctionDef):
             node, "contextmanager", "asynccontextmanager", "fixture"
         )
 
+    # trigger on leaving yield so any comments are parsed for noqas
     def visit_Yield(self, node: cst.Yield):
         if self._yield_is_error:
             self.error(node)

flake8_trio/visitors/visitor91x.py

@@ -118,14 +118,15 @@ def library(self) -> tuple[str, ...]:
         ...
 
     @abstractmethod
-    def should_autofix(self) -> bool:
+    def should_autofix(self, node: cst.CSTNode, code: str | None = None) -> bool:
         ...
 
     # instead of trying to exclude yields found in all the weird places from
     # setting self.add_statement, we instead clear it upon each new line.
     # Several of them *could* be handled, e.g. `if ...: yield`, but
     # that's uncommon enough we don't care about it.
     def visit_SimpleStatementLine(self, node: cst.SimpleStatementLine):
+        super().visit_SimpleStatementLine(node)
         self.add_statement = None
 
     # insert checkpoint before yield with a flattensentinel, if indicated
@@ -134,10 +135,12 @@ def leave_SimpleStatementLine(
         original_node: cst.SimpleStatementLine,
         updated_node: cst.SimpleStatementLine,
     ) -> cst.SimpleStatementLine | cst.FlattenSentinel[cst.SimpleStatementLine]:
+        _ = super().leave_SimpleStatementLine(original_node, updated_node)
+
         # possible TODO: generate an error if transforming+visiting is done in a
         # single pass and emit-error-on-transform can be enabled/disabled. The error can't
         # be generated in the yield/return since it doesn't know if it will be autofixed.
-        if self.add_statement is None or not self.should_autofix():
+        if self.add_statement is None or not self.should_autofix(original_node):
             return updated_node
         curr_add_statement = self.add_statement
         self.add_statement = None
@@ -196,8 +199,8 @@ def __init__(
     def library(self) -> tuple[str, ...]:
         return self.__library if self.__library else ("trio",)
 
-    def should_autofix(self) -> bool:
-        return True
+    def should_autofix(self, node: cst.CSTNode, code: str | None = None) -> bool:
+        return not self.noautofix
 
     def leave_Yield(
         self,
@@ -206,7 +209,9 @@ def leave_Yield(
     ) -> cst.Yield:
         # Needs to be passed *original* node, since updated node is a copy
         # which loses identity equality
-        if original_node in self.nodes_needing_checkpoint and not self.noautofix:
+        if original_node in self.nodes_needing_checkpoint and self.should_autofix(
+            original_node
+        ):
             self.add_statement = checkpoint_statement(self.library[0])
         return updated_node
 
@@ -239,8 +244,10 @@ def __init__(self, *args: Any, **kwargs: Any):
         self.loop_state = LoopState()
         self.try_state = TryState()
 
-    def should_autofix(self, code: str | None = None) -> bool:
-        return super().should_autofix("TRIO911" if self.has_yield else "TRIO910")
+    def should_autofix(self, node: cst.CSTNode, code: str | None = None) -> bool:
+        return not self.noautofix and super().should_autofix(
+            node, "TRIO911" if self.has_yield else "TRIO910"
+        )
 
     def checkpoint_statement(self) -> cst.SimpleStatementLine:
         return checkpoint_statement(self.library[0])
@@ -290,7 +297,7 @@ def leave_FunctionDef(
             self.async_function
             # updated_node does not have a position, so we must send original_node
             and self.check_function_exit(original_node)
-            and self.should_autofix()
+            and self.should_autofix(original_node)
             and isinstance(updated_node.body, cst.IndentedBlock)
         ):
             # insert checkpoint at the end of body
@@ -327,17 +334,20 @@ def check_function_exit(
             # Add this as a node potentially needing checkpoints only if it
             # missing checkpoints solely depends on whether the artificial statement is
             # "real"
-            if len(self.uncheckpointed_statements) == 1 and not self.noautofix:
+            if len(self.uncheckpointed_statements) == 1 and self.should_autofix(
+                original_node
+            ):
                 self.loop_state.nodes_needing_checkpoints.append(original_node)
                 return False
 
+        any_errors = False
         # raise the actual errors
         for statement in self.uncheckpointed_statements:
             if statement == ARTIFICIAL_STATEMENT:
                 continue
-            self.error_91x(original_node, statement)
+            any_errors |= self.error_91x(original_node, statement)
 
-        return True
+        return any_errors
 
     def leave_Return(
         self, original_node: cst.Return, updated_node: cst.Return
@@ -357,15 +367,15 @@ def error_91x(
         self,
         node: cst.Return | cst.FunctionDef | cst.Yield,
         statement: Statement,
-    ):
+    ) -> bool:
         assert statement != ARTIFICIAL_STATEMENT
 
         if isinstance(node, cst.FunctionDef):
             msg = "exit"
         else:
             msg = node.__class__.__name__.lower()
 
-        self.error(
+        return self.error(
             node,
             msg,
             statement,
@@ -403,7 +413,9 @@ def leave_Yield(
             return updated_node
         self.has_yield = True
 
-        if self.check_function_exit(original_node) and not self.noautofix:
+        if self.check_function_exit(original_node) and self.should_autofix(
+            original_node
+        ):
             self.add_statement = self.checkpoint_statement()
 
         # mark as requiring checkpoint after
@@ -596,8 +608,7 @@ def leave_While_body(self, node: cst.For | cst.While):
             ):
                 if stmt == ARTIFICIAL_STATEMENT:
                     continue
-                any_error = True
-                self.error_91x(err_node, stmt)
+                any_error |= self.error_91x(err_node, stmt)
 
         # if there's no errors from artificial statements, we don't need to insert
         # the potential checkpoints
@@ -664,7 +675,7 @@ def leave_While(
         | cst.FlattenSentinel[cst.For | cst.While]
         | cst.RemovalSentinel
     ):
-        if self.loop_state.nodes_needing_checkpoints and self.should_autofix():
+        if self.loop_state.nodes_needing_checkpoints:
             transformer = InsertCheckpointsInLoopBody(
                 self.loop_state.nodes_needing_checkpoints, self.library
             )

pyproject.toml

@@ -28,7 +28,7 @@ warn_unused_ignores = false
 [tool.pyright]
 exclude = ["**/node_modules", "**/__pycache__", "**/.*"]
 reportCallInDefaultInitializer = true
-reportImplicitStringConcatenation = true
+reportImplicitStringConcatenation = false  # black generates implicit string concats
 reportMissingSuperCall = true
 reportPropertyTypeMismatch = true
 reportShadowedImports = true