fix(ci): TQ009 stops at an early exit that skips the reporting below it

A handler can report on the path that reaches the `raise` and still swallow on
one that leaves before it:

    except Exception as e:
        if "try pulling it first" in str(e):
            return
        pytest.fail(f"Error occurred: {e}")

The scan now stops at the first statement that can leave the handler without
reporting, so a `return`, `break` or `continue` above the reporting statement
makes the handler swallowing rather than clean. A `return` inside a nested
`def` is not one of those, since it leaves the nested body, so the walk stops
at a nested scope.

Two more sites, 42 to 44.
This commit is contained in:
ryan-crabbe-berri 2026-08-27 09:37:43 -07:00
parent ddb313a997
commit 8d821edfa5
3 changed files with 81 additions and 4 deletions

View file

@ -123,6 +123,7 @@ import sys
import tokenize
from collections.abc import Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from itertools import takewhile
from multiprocessing import Pool
from pathlib import Path
from types import MappingProxyType
@ -438,6 +439,16 @@ def _catches_assertion_error(handler: ast.ExceptHandler) -> bool:
return any(_dotted_name(node).rpartition(".")[2] in ASSERTION_ERROR_CATCHERS for node in named)
def _walk_within_scope(node: ast.AST) -> Iterator[ast.AST]:
"""`ast.walk` that stops at a nested function or class, whose body is a scope of its
own rather than more of the statements around it."""
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)):
return
yield node
for child in ast.iter_child_nodes(node):
yield from _walk_within_scope(child)
def _statement_reports(stmt: ast.stmt) -> bool:
"""Whether control leaving this statement has necessarily reported the failure.
A `raise` or a `pytest.fail` does. An `if` does only when both halves do, since the
@ -457,11 +468,27 @@ def _statement_reports(stmt: ast.stmt) -> bool:
return False
def _statement_escapes(stmt: ast.stmt) -> bool:
"""Whether control can leave the enclosing handler through this statement without
having reported. A `return`, `break` or `continue` on some branch does exactly that,
and it makes every reporting statement after it unreachable on that path. A nested
`def` is not walked into, since its `return` leaves the nested body, not the handler."""
return not _statement_reports(stmt) and any(
isinstance(node, (ast.Return, ast.Break, ast.Continue))
for node in _walk_within_scope(stmt)
)
def _reports_the_failure(body: Sequence[ast.stmt]) -> bool:
"""Re-raising, or failing the test, passes the failure on rather than eating it.
Every path out of the block has to do it: a `raise` reachable on one branch only
leaves the other branch swallowing, which is the shape the rule exists to catch."""
return any(_statement_reports(stmt) for stmt in body)
Every path out of the block has to do it, so the scan stops at the first statement
that can escape without reporting: a `raise` reachable on one branch only, or sitting
below an early `return`, leaves the other path swallowing, which is the shape the
rule exists to catch."""
return any(
_statement_reports(stmt)
for stmt in takewhile(lambda candidate: not _statement_escapes(candidate), body)
)
def _swallowing_handler(node: ast.Try) -> ast.ExceptHandler | None:

View file

@ -24,6 +24,6 @@
"limit": 11139
},
"TQ009": {
"limit": 42
"limit": 44
}
}

View file

@ -941,6 +941,56 @@ def test_a_report_after_a_swallowing_branch_still_counts(tmp_path):
assert _swallowed(tmp_path, body) == []
def test_an_early_return_below_a_raise_still_swallows(tmp_path):
body = (
" try:\n"
" assert f() == 3\n"
" except Exception as e:\n"
" if 'flaky' in str(e):\n"
" return\n"
" raise\n"
)
assert _swallowed(tmp_path, body) == ["TQ009"]
def test_a_continue_below_a_pytest_fail_still_swallows(tmp_path):
body = (
" for item in items:\n"
" try:\n"
" assert f(item) == 3\n"
" except Exception as e:\n"
" if 'skip' in str(e):\n"
" continue\n"
" pytest.fail(str(e))\n"
)
assert _swallowed(tmp_path, body) == ["TQ009"]
def test_a_break_below_a_raise_still_swallows(tmp_path):
body = (
" for item in items:\n"
" try:\n"
" assert f(item) == 3\n"
" except Exception:\n"
" if done:\n"
" break\n"
" raise\n"
)
assert _swallowed(tmp_path, body) == ["TQ009"]
def test_a_return_in_a_nested_def_does_not_escape_the_handler(tmp_path):
body = (
" try:\n"
" assert f() == 3\n"
" except Exception:\n"
" def _later():\n"
" return 1\n"
" raise\n"
)
assert _swallowed(tmp_path, body) == []
def test_a_swallow_outside_a_test_function_is_left_alone(tmp_path):
source = (
"def _helper():\n"