fix(suppressions): bind multiline inline ignores to decorated declaration headers consistently

Author: orenlab

codeclone/extractor.py

@@ -4,9 +4,11 @@
 from __future__ import annotations
 
 import ast
+import io
 import math
 import os
 import signal
+import tokenize
 from contextlib import contextmanager
 from hashlib import sha1 as _sha1
 from typing import TYPE_CHECKING, Literal
@@ -152,6 +154,98 @@ def _stmt_count(node: ast.AST) -> int:
     return len(body) if isinstance(body, list) else 0
 
 
+def _source_tokens(source: str) -> tuple[tokenize.TokenInfo, ...]:
+    try:
+        return tuple(tokenize.generate_tokens(io.StringIO(source).readline))
+    except tokenize.TokenError:
+        return ()
+
+
+def _declaration_token_name(node: ast.AST) -> str:
+    if isinstance(node, ast.ClassDef):
+        return "class"
+    if isinstance(node, ast.AsyncFunctionDef):
+        return "async"
+    return "def"
+
+
+def _declaration_token_index(
+    *,
+    source_tokens: tuple[tokenize.TokenInfo, ...],
+    start_line: int,
+    start_col: int,
+    declaration_token: str,
+) -> int | None:
+    for idx, token in enumerate(source_tokens):
+        if token.start != (start_line, start_col):
+            continue
+        if token.type == tokenize.NAME and token.string == declaration_token:
+            return idx
+    return None
+
+
+def _scan_declaration_colon_line(
+    *,
+    source_tokens: tuple[tokenize.TokenInfo, ...],
+    start_index: int,
+) -> int | None:
+    nesting = 0
+    for token in source_tokens[start_index + 1 :]:
+        if token.type == tokenize.OP:
+            if token.string in "([{":
+                nesting += 1
+                continue
+            if token.string in ")]}":
+                if nesting > 0:
+                    nesting -= 1
+                continue
+            if token.string == ":" and nesting == 0:
+                return token.start[0]
+        if token.type == tokenize.NEWLINE and nesting == 0:
+            return None
+    return None
+
+
+def _fallback_declaration_end_line(node: ast.AST, *, start_line: int) -> int:
+    body = getattr(node, "body", None)
+    if not isinstance(body, list) or not body:
+        return start_line
+
+    first_body_line = int(getattr(body[0], "lineno", 0))
+    if first_body_line <= 0 or first_body_line == start_line:
+        return start_line
+    return max(start_line, first_body_line - 1)
+
+
+def _declaration_end_line(
+    node: ast.AST,
+    *,
+    source_tokens: tuple[tokenize.TokenInfo, ...],
+) -> int:
+    start_line = int(getattr(node, "lineno", 0))
+    start_col = int(getattr(node, "col_offset", 0))
+    if start_line <= 0:
+        return 0
+
+    declaration_token = _declaration_token_name(node)
+    start_index = _declaration_token_index(
+        source_tokens=source_tokens,
+        start_line=start_line,
+        start_col=start_col,
+        declaration_token=declaration_token,
+    )
+    if start_index is None:
+        return _fallback_declaration_end_line(node, start_line=start_line)
+
+    colon_line = _scan_declaration_colon_line(
+        source_tokens=source_tokens,
+        start_index=start_index,
+    )
+    if colon_line is not None:
+        return colon_line
+    return _fallback_declaration_end_line(node, start_line=start_line)
+
+
 class _QualnameCollector(ast.NodeVisitor):
     __slots__ = (
         "class_count",
@@ -577,6 +671,7 @@ def _collect_declaration_targets(
     filepath: str,
     module_name: str,
     collector: _QualnameCollector,
+    source_tokens: tuple[tokenize.TokenInfo, ...],
 ) -> tuple[DeclarationTarget, ...]:
     declarations: list[DeclarationTarget] = []
 
@@ -585,6 +680,10 @@ def _collect_declaration_targets(
         end = int(getattr(node, "end_lineno", 0))
         if start <= 0 or end <= 0:
             continue
+        declaration_end_line = _declaration_end_line(
+            node,
+            source_tokens=source_tokens,
+        )
         kind: Literal["function", "method"] = (
             "method" if "." in local_name else "function"
         )
@@ -595,6 +694,7 @@ def _collect_declaration_targets(
                 start_line=start,
                 end_line=end,
                 kind=kind,
+                declaration_end_line=declaration_end_line,
             )
         )
 
@@ -603,13 +703,18 @@ def _collect_declaration_targets(
         end = int(getattr(class_node, "end_lineno", 0))
         if start <= 0 or end <= 0:
             continue
+        declaration_end_line = _declaration_end_line(
+            class_node,
+            source_tokens=source_tokens,
+        )
         declarations.append(
             DeclarationTarget(
                 filepath=filepath,
                 qualname=f"{module_name}:{class_qualname}",
                 start_line=start,
                 end_line=end,
                 kind="class",
+                declaration_end_line=declaration_end_line,
             )
         )
 
@@ -657,6 +762,7 @@ def extract_units_and_stats_from_source(
     collector = _QualnameCollector()
     collector.visit(tree)
     source_lines = source.splitlines()
+    source_tokens = _source_tokens(source)
     source_line_count = len(source_lines)
 
     is_test_file = is_test_filepath(filepath)
@@ -676,6 +782,7 @@ def extract_units_and_stats_from_source(
         filepath=filepath,
         module_name=module_name,
         collector=collector,
+        source_tokens=source_tokens,
     )
     suppression_bindings = bind_suppressions_to_declarations(
         directives=suppression_directives,

codeclone/suppressions.py

@@ -63,6 +63,7 @@ class DeclarationTarget:
     start_line: int
     end_line: int
     kind: DeclarationKind
+    declaration_end_line: int | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -176,9 +177,10 @@ def bind_suppressions_to_declarations(
 
     bindings: list[SuppressionBinding] = []
     for target in declarations:
+        inline_binding_line = target.declaration_end_line or target.start_line
         bound_rules = _merge_rules(
             leading_rules_by_line.get(target.start_line - 1, ()),
-            inline_rules_by_line.get(target.start_line, ()),
+            inline_rules_by_line.get(inline_binding_line, ()),
         )
         if not bound_rules:
             continue

tests/test_cli_inprocess.py

@@ -3803,6 +3803,71 @@ def fn(x):
     assert any("sf" in entry for entry in files_after.values())
 
 
+def test_cli_dead_code_suppression_is_stable_between_plain_and_json_runs(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _write_python_module(
+        tmp_path,
+        "models.py",
+        """\
+class Settings:  # codeclone: ignore[dead-code]
+    @validator("field")
+    @classmethod
+    def validate_config_version(
+        cls,
+        value: str | None,
+    ) -> str | None:  # codeclone: ignore[dead-code]
+        return value
+""",
+    )
+    json_out = tmp_path / "report.json"
+    cache_path = tmp_path / "cache.json"
+
+    _patch_parallel(monkeypatch)
+    _run_main(
+        monkeypatch,
+        [
+            str(tmp_path),
+            "--cache-path",
+            str(cache_path),
+            "--fail-dead-code",
+            "--no-progress",
+        ],
+    )
+
+    cache_payload = json.loads(cache_path.read_text("utf-8"))
+    files_before = cache_payload["payload"]["files"]
+    assert all("sf" not in entry for entry in files_before.values())
+
+    _run_main(
+        monkeypatch,
+        [
+            str(tmp_path),
+            "--cache-path",
+            str(cache_path),
+            "--fail-dead-code",
+            "--json",
+            str(json_out),
+            "--no-progress",
+        ],
+    )
+    payload = json.loads(json_out.read_text("utf-8"))
+    dead_code = payload["metrics"]["families"]["dead_code"]
+    assert dead_code["summary"] == {"total": 0, "high_confidence": 0, "suppressed": 2}
+
+    _run_main(
+        monkeypatch,
+        [
+            str(tmp_path),
+            "--cache-path",
+            str(cache_path),
+            "--fail-dead-code",
+            "--no-progress",
+        ],
+    )
+
+
 @pytest.mark.parametrize(
     ("reason", "expected"),
     [

tests/test_extractor.py

@@ -66,6 +66,57 @@ def foo():
     assert segments == []
 
 
+def test_source_tokens_returns_empty_on_tokenize_error() -> None:
+    assert extractor._source_tokens('"""') == ()
+
+
+def test_declaration_token_index_returns_none_when_start_token_is_missing() -> None:
+    tokens = extractor._source_tokens("value = 1\n")
+    assert (
+        extractor._declaration_token_index(
+            source_tokens=tokens,
+            start_line=1,
+            start_col=0,
+            declaration_token="def",
+        )
+        is None
+    )
+
+
+def test_scan_declaration_colon_line_returns_none_when_header_is_incomplete() -> None:
+    tokens = extractor._source_tokens("def broken\n")
+    assert (
+        extractor._scan_declaration_colon_line(
+            source_tokens=tokens,
+            start_index=0,
+        )
+        is None
+    )
+
+
+def test_declaration_end_line_falls_back_without_tokens() -> None:
+    node = ast.parse(
+        """
+class Demo:
+    pass
+"""
+    ).body[0]
+    assert isinstance(node, ast.ClassDef)
+    assert extractor._declaration_end_line(node, source_tokens=()) == 2
+
+
+def test_declaration_end_line_returns_zero_for_invalid_start_line() -> None:
+    node = ast.parse(
+        """
+def broken():
+    return 1
+"""
+    ).body[0]
+    assert isinstance(node, ast.FunctionDef)
+    node.lineno = 0
+    assert extractor._declaration_end_line(node, source_tokens=()) == 0
+
+
 def test_init_function_is_ignored_for_blocks() -> None:
     src = """
 class A:
@@ -675,6 +726,36 @@ def alive(self):
     assert tuple(item.qualname for item in dead) == ("pkg.mod:Service.alive",)
 
 
+def test_dead_code_binds_inline_suppression_on_multiline_decorated_method() -> None:
+    src = """
+class Settings:  # codeclone: ignore[dead-code]
+    @validator("field")
+    @classmethod
+    def validate_config_version(
+        cls,
+        value: str | None,
+    ) -> str | None:  # codeclone: ignore[dead-code]
+        return value
+
+    def orphan(self) -> int:
+        return 1
+"""
+    _, _, _, _, file_metrics, _ = extractor.extract_units_and_stats_from_source(
+        source=src,
+        filepath="pkg/mod.py",
+        module_name="pkg.mod",
+        cfg=NormalizationConfig(),
+        min_loc=1,
+        min_stmt=1,
+    )
+    dead = find_unused(
+        definitions=file_metrics.dead_candidates,
+        referenced_names=file_metrics.referenced_names,
+        referenced_qualnames=file_metrics.referenced_qualnames,
+    )
+    assert tuple(item.qualname for item in dead) == ("pkg.mod:Settings.orphan",)
+
+
 def test_collect_dead_candidates_and_extract_skip_classes_without_lineno(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: