chore(api): disallow builtins getattr variants in new code (#38250)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.github/workflows/style.yml

@@ -29,6 +29,8 @@ jobs:
         with:
           files: |
             api/**
+            scripts/check_no_new_getattr.py
+            scripts/ast_grep_rules/no_new_getattr.yml
             .github/workflows/style.yml
 
       - name: Setup UV and Python
@@ -51,6 +53,18 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv run --project api --dev python api/dev/lint_response_contracts.py --fail-on-mismatch
 
+      - name: Fetch merge target ref for getattr guard
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: git fetch --no-tags --depth=1 origin +refs/heads/main:refs/remotes/origin/main
+
+      - name: Bind merge target branch for getattr guard
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: git show-ref --verify --quiet refs/heads/main || git branch main origin/main
+
+      - name: Run No New Getattr Guard
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: uv run --project api python scripts/check_no_new_getattr.py --mode ci --merge-target main
+
       - name: Run Type Checks
         if: steps.changed-files.outputs.any_changed == 'true'
         env:

api/tests/unit_tests/commands/test_check_no_new_getattr.py

@@ -0,0 +1,768 @@
+"""Contract tests for the future no-new-getattr CLI wrapper."""
+
+from __future__ import annotations
+
+import importlib.util
+import re
+import subprocess
+import sys
+import textwrap
+import types
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+SCRIPT_PATH = REPO_ROOT / "scripts" / "check_no_new_getattr.py"
+
+
+def load_guard_module() -> types.ModuleType:
+    spec = importlib.util.spec_from_file_location("check_no_new_getattr_under_test", SCRIPT_PATH)
+    assert spec is not None
+    assert spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def git(repo: Path, *args: str) -> str:
+    completed = subprocess.run(
+        ["git", *args],
+        cwd=repo,
+        text=True,
+        capture_output=True,
+        check=True,
+    )
+    return completed.stdout.strip()
+
+
+def run_script(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["python3", str(SCRIPT_PATH), *args],
+        cwd=repo,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+
+def write_repo_file(repo: Path, relative_path: str, content: str) -> None:
+    path = repo / relative_path
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(textwrap.dedent(content).lstrip(), encoding="utf-8")
+
+
+def commit_all(repo: Path, message: str) -> None:
+    git(repo, "add", ".")
+    git(repo, "commit", "-m", message)
+
+
+def init_repo(repo: Path) -> None:
+    git(repo, "init", "-b", "main")
+    git(repo, "config", "user.name", "Tester")
+    git(repo, "config", "user.email", "tester@example.com")
+
+
+def checkout_feature_branch(repo: Path) -> None:
+    git(repo, "checkout", "-b", "feature/test-branch")
+
+
+def stderr_lines(result: subprocess.CompletedProcess[str]) -> list[str]:
+    return [line for line in result.stderr.splitlines() if line.strip()]
+
+
+def assert_has_actionable_violation(stderr: str, path: str) -> None:
+    assert re.search(rf"{re.escape(path)}:\d+:", stderr), stderr
+    assert "no-new-getattr" in stderr
+
+
+def test_resolve_ast_grep_command_prefers_ast_grep(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_guard_module()
+    monkeypatch.setattr(
+        module.shutil,
+        "which",
+        lambda name: f"/usr/bin/{name}" if name == "ast-grep" else None,
+    )
+
+    assert module.resolve_ast_grep_command() == ["ast-grep"]
+
+
+def test_resolve_ast_grep_command_falls_back_to_uvx(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_guard_module()
+    monkeypatch.setattr(
+        module.shutil,
+        "which",
+        lambda name: "/usr/bin/uvx" if name == "uvx" else None,
+    )
+
+    assert module.resolve_ast_grep_command() == ["uvx", "--from", "ast-grep-cli", "ast-grep"]
+
+
+def test_resolve_ast_grep_command_never_uses_sg(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_guard_module()
+    monkeypatch.setattr(
+        module.shutil,
+        "which",
+        lambda name: f"/usr/bin/{name}" if name in {"sg", "uvx"} else None,
+    )
+
+    assert module.resolve_ast_grep_command() == ["uvx", "--from", "ast-grep-cli", "ast-grep"]
+
+
+def test_resolve_ast_grep_command_rejects_sg_only(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_guard_module()
+    monkeypatch.setattr(
+        module.shutil,
+        "which",
+        lambda name: "/usr/bin/sg" if name == "sg" else None,
+    )
+
+    with pytest.raises(RuntimeError, match="ast-grep executable not found"):
+        module.resolve_ast_grep_command()
+
+
+def test_resolve_ast_grep_command_raises_without_explicit_binary(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_guard_module()
+    monkeypatch.setattr(module.shutil, "which", lambda _name: None)
+
+    with pytest.raises(RuntimeError, match="ast-grep executable not found"):
+        module.resolve_ast_grep_command()
+
+
+def test_style_workflow_wires_no_new_getattr_guard() -> None:
+    workflow = (REPO_ROOT / ".github" / "workflows" / "style.yml").read_text(encoding="utf-8")
+    python_style_job = re.search(
+        r"(?ms)^  python-style:\n(?P<job>.*?)(?=^  [a-z0-9-]+:\n|\Z)",
+        workflow,
+    )
+    assert python_style_job is not None
+
+    job_text = python_style_job.group("job")
+    checkout_step = re.search(
+        r"(?ms)^      - name: Checkout code\n(?P<step>.*?)(?=^      - name: |\Z)",
+        job_text,
+    )
+    assert checkout_step is not None
+
+    changed_files_step = re.search(
+        r"(?ms)^      - name: Check changed files\n.*?^          files: \|\n(?P<files>(?:^            \S[^\n]*\n)+)",
+        job_text,
+    )
+    assert changed_files_step is not None
+
+    files_block = changed_files_step.group("files")
+    assert "api/**\n" in files_block
+    assert "scripts/check_no_new_getattr.py\n" in files_block
+    assert "scripts/ast_grep_rules/no_new_getattr.yml\n" in files_block
+    assert ".github/workflows/style.yml\n" in files_block
+
+    guard_command = "scripts/check_no_new_getattr.py --mode ci --merge-target main"
+    assert guard_command in job_text
+
+    guard_step = re.search(
+        rf"(?ms)^      - name: Run No New Getattr Guard\n"
+        rf"(?P<step>.*?{re.escape(guard_command)}.*?)(?=^      - name: |\Z)",
+        job_text,
+    )
+    assert guard_step is not None
+
+    pre_guard_text = job_text[: guard_step.start()]
+    step_pattern = r"(?ms)^      - name: [^\n]*\n(?P<step>.*?)(?=^      - name: |\Z)"
+    fetch_step_text = next(
+        (
+            match.group("step")
+            for match in re.finditer(step_pattern, pre_guard_text)
+            if any(
+                re.search(pattern, line)
+                for line in match.group("step").splitlines()
+                for pattern in (
+                    r"git fetch .*refs/heads/main:refs/remotes/origin/main",
+                    r"git fetch .*main:refs/remotes/origin/main",
+                    r"git fetch .*refs/remotes/origin/main",
+                )
+            )
+        ),
+        "",
+    )
+    assert fetch_step_text
+    assert "git fetch" in fetch_step_text
+    assert "origin" in fetch_step_text
+    assert any(
+        re.search(pattern, line)
+        for line in fetch_step_text.splitlines()
+        for pattern in (
+            r"git fetch .*refs/heads/main:refs/remotes/origin/main",
+            r"git fetch .*main:refs/remotes/origin/main",
+            r"git fetch .*refs/remotes/origin/main",
+        )
+    )
+
+    bind_step = re.search(
+        r"(?ms)^      - name: Bind merge target branch for getattr guard\n(?P<step>.*?)(?=^      - name: |\Z)",
+        pre_guard_text,
+    )
+    assert bind_step is not None
+    bind_step_text = bind_step.group("step")
+    assert any(
+        command in bind_step_text
+        for command in (
+            "git branch main origin/main",
+            "git checkout -B main origin/main",
+            "git switch -C main origin/main",
+            "git update-ref refs/heads/main refs/remotes/origin/main",
+        )
+    )
+
+
+def test_ci_mode_passes_when_only_legacy_getattr_exists(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/legacy.py",
+        """
+        def read_value(obj):
+            return getattr(obj, "legacy_name", None)
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/other.py",
+        """
+        def meaning() -> int:
+            return 42
+        """,
+    )
+    commit_all(tmp_path, "unrelated change")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 0, result.stderr
+
+
+def test_ci_mode_fails_for_new_file_with_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/new_usage.py",
+        """
+        def read_value(obj):
+            return getattr(obj, "new_name", None)
+        """,
+    )
+    commit_all(tmp_path, "add new getattr usage")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/new_usage.py")
+
+
+def test_ci_mode_fails_for_new_file_with_two_arg_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/new_usage.py",
+        """
+        def read_value(obj):
+            return getattr(obj, "dynamic_name")
+        """,
+    )
+    commit_all(tmp_path, "add new two-arg getattr usage")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/new_usage.py")
+
+
+def test_ci_mode_fails_for_new_file_with_builtins_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/new_usage.py",
+        """
+        import builtins
+
+
+        def read_value(obj):
+            return builtins.getattr(obj, "dynamic_name", None)
+        """,
+    )
+    commit_all(tmp_path, "add new builtins getattr usage")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/new_usage.py")
+
+
+def test_ci_mode_fails_for_new_file_with_two_arg_builtins_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/new_usage.py",
+        """
+        import builtins
+
+
+        def read_value(obj):
+            return builtins.getattr(obj, "dynamic_name")
+        """,
+    )
+    commit_all(tmp_path, "add new two-arg builtins getattr usage")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/new_usage.py")
+
+
+def test_ci_mode_fails_for_new_file_with_dunder_builtins_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/new_usage.py",
+        """
+        def read_value(obj):
+            return __builtins__.getattr(obj, "dynamic_name", None)
+        """,
+    )
+    commit_all(tmp_path, "add new dunder builtins getattr usage")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/new_usage.py")
+
+
+def test_ci_mode_fails_for_new_file_with_two_arg_dunder_builtins_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/new_usage.py",
+        """
+        def read_value(obj):
+            return __builtins__.getattr(obj, "dynamic_name")
+        """,
+    )
+    commit_all(tmp_path, "add new two-arg dunder builtins getattr usage")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/new_usage.py")
+
+
+def test_ci_mode_uses_merge_base_against_main_not_just_head_parent(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/introduced_earlier.py",
+        """
+        def read_value(obj):
+            return getattr(obj, "introduced_in_first_feature_commit", None)
+        """,
+    )
+    commit_all(tmp_path, "introduce violating getattr in first feature commit")
+
+    write_repo_file(
+        tmp_path,
+        "pkg/other.py",
+        """
+        def meaning() -> int:
+            return 42
+        """,
+    )
+    commit_all(tmp_path, "later feature commit does not touch violating file")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/introduced_earlier.py")
+
+
+def test_pre_commit_mode_reads_staged_content_only(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        def read_value(obj):
+            return obj.value
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        def read_value(obj):
+            return obj.value + 1
+        """,
+    )
+    git(tmp_path, "add", "pkg/module.py")
+
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        def read_value(obj):
+            return getattr(obj, "value", None)
+        """,
+    )
+
+    result = run_script(tmp_path, "--mode", "pre-commit")
+
+    assert result.returncode == 0, result.stderr
+
+
+def test_pre_commit_mode_fails_for_staged_two_arg_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        def read_value(obj):
+            return obj.value
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        def read_value(obj):
+            return getattr(obj, "value")
+        """,
+    )
+    git(tmp_path, "add", "pkg/module.py")
+
+    result = run_script(tmp_path, "--mode", "pre-commit")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/module.py")
+
+
+def test_pre_commit_mode_fails_for_staged_builtins_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        def read_value(obj):
+            return obj.value
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        import builtins
+
+
+        def read_value(obj):
+            return builtins.getattr(obj, "value", None)
+        """,
+    )
+    git(tmp_path, "add", "pkg/module.py")
+
+    result = run_script(tmp_path, "--mode", "pre-commit")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/module.py")
+
+
+def test_pre_commit_mode_fails_for_staged_two_arg_builtins_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        def read_value(obj):
+            return obj.value
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+
+    write_repo_file(
+        tmp_path,
+        "pkg/module.py",
+        """
+        import builtins
+
+
+        def read_value(obj):
+            return builtins.getattr(obj, "value")
+        """,
+    )
+    git(tmp_path, "add", "pkg/module.py")
+
+    result = run_script(tmp_path, "--mode", "pre-commit")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/module.py")
+
+
+def test_modified_hunk_with_same_getattr_count_is_allowed(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/sample.py",
+        """
+        def resolve_name(user):
+            name = getattr(user, "display_name", None)
+            if name:
+                return name
+            return "unknown"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/sample.py",
+        """
+        def resolve_name(user):
+            name = getattr(user, "display_name", None)
+            if name:
+                return name.strip()
+            return "unknown user"
+        """,
+    )
+    commit_all(tmp_path, "touch legacy getattr hunk")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 0, result.stderr
+
+
+def test_modified_hunk_with_decreased_getattr_count_is_allowed(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/sample.py",
+        """
+        def resolve_name(user):
+            primary = getattr(user, "display_name", None)
+            return primary or getattr(user, "username", None)
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/sample.py",
+        """
+        def resolve_name(user):
+            return getattr(user, "display_name", None)
+        """,
+    )
+    commit_all(tmp_path, "remove one legacy getattr")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 0, result.stderr
+
+
+def test_modified_hunk_with_increased_getattr_count_fails(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/sample.py",
+        """
+        def resolve_name(user):
+            return getattr(user, "display_name", None)
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/sample.py",
+        """
+        def resolve_name(user):
+            primary = getattr(user, "display_name", None)
+            return primary or getattr(user, "username", None)
+        """,
+    )
+    commit_all(tmp_path, "add one more getattr")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/sample.py")
+    assert "net-new getattr" in result.stderr
+
+
+def test_inline_noqa_suppression_with_explanatory_text_skips_added_getattr(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def read_value(obj):
+            return getattr(obj, "dynamic_name", None)  # noqa: no-new-getattr needed for plugin-defined attributes
+        """,
+    )
+    commit_all(tmp_path, "add suppressed getattr")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert "no-new-getattr needed for plugin-defined attributes" in (tmp_path / "pkg/existing.py").read_text(
+        encoding="utf-8"
+    )
+    assert result.returncode == 0, stderr_lines(result)
+
+
+def test_inline_noqa_without_explanatory_text_is_not_sufficient(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def stable() -> str:
+            return "ok"
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "pkg/existing.py",
+        """
+        def read_value(obj):
+            return getattr(obj, "dynamic_name", None)  # noqa: no-new-getattr
+        """,
+    )
+    commit_all(tmp_path, "add bare noqa getattr")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 1
+    assert_has_actionable_violation(result.stderr, "pkg/existing.py")
+
+
+def test_non_python_file_with_getattr_text_does_not_fail_guard(tmp_path: Path) -> None:
+    init_repo(tmp_path)
+    write_repo_file(
+        tmp_path,
+        "docs/example.txt",
+        """
+        Existing documentation.
+        """,
+    )
+    commit_all(tmp_path, "baseline")
+    checkout_feature_branch(tmp_path)
+
+    write_repo_file(
+        tmp_path,
+        "docs/example.txt",
+        """
+        getattr(obj, "dynamic_name", None)
+        """,
+    )
+    commit_all(tmp_path, "document getattr example")
+
+    result = run_script(tmp_path, "--mode", "ci", "--merge-target", "main")
+
+    assert result.returncode == 0, result.stderr

scripts/ast_grep_rules/no_new_getattr.yml

@@ -0,0 +1,12 @@
+id: no-new-getattr
+language: python
+rule:
+  any:
+    - pattern: getattr($OBJ, $NAME)
+    - pattern: getattr($OBJ, $NAME, $$$REST)
+    - pattern: builtins.getattr($OBJ, $NAME)
+    - pattern: builtins.getattr($OBJ, $NAME, $$$REST)
+    - pattern: __builtins__.getattr($OBJ, $NAME)
+    - pattern: __builtins__.getattr($OBJ, $NAME, $$$REST)
+message: net-new getattr() in added code
+severity: error

scripts/check_no_new_getattr.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""Block net-new getattr() usage in changed Python hunks."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SCAN_ROOT = Path.cwd()
+RULE_PATH = REPO_ROOT / "scripts" / "ast_grep_rules" / "no_new_getattr.yml"
+HUNK_PATTERN = re.compile(
+    r"^@@ -(?P<old_start>\d+)(?:,(?P<old_count>\d+))? \+(?P<new_start>\d+)(?:,(?P<new_count>\d+))? @@"
+)
+SUPPRESSION_PATTERN = re.compile(r"# noqa: no-new-getattr(?:\s+(?P<reason>\S.*))?\s*$")
+
+
+@dataclass(frozen=True)
+class Hunk:
+    old_start: int
+    old_count: int
+    new_start: int
+    new_count: int
+
+    def contains_old(self, line_number: int) -> bool:
+        return self._contains(line_number, self.old_start, self.old_count)
+
+    def contains_new(self, line_number: int) -> bool:
+        return self._contains(line_number, self.new_start, self.new_count)
+
+    @staticmethod
+    def _contains(line_number: int, start: int, count: int) -> bool:
+        if count == 0:
+            return False
+        return start <= line_number < start + count
+
+
+@dataclass(frozen=True)
+class Match:
+    line_number: int
+    source_line: str
+
+
+@dataclass(frozen=True)
+class Violation:
+    path: str
+    line_number: int
+    message: str
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--mode", choices=("pre-commit", "ci"), required=True)
+    parser.add_argument("--merge-target", default="main")
+    return parser.parse_args()
+
+
+def resolve_ast_grep_command() -> list[str]:
+    if shutil.which("ast-grep"):
+        return ["ast-grep"]
+    if shutil.which("uvx"):
+        return ["uvx", "--from", "ast-grep-cli", "ast-grep"]
+    raise RuntimeError("ast-grep executable not found")
+
+
+def git_output(*args: str, allow_missing: bool = False) -> str:
+    completed = subprocess.run(
+        ["git", *args],
+        cwd=SCAN_ROOT,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    if completed.returncode == 0:
+        return completed.stdout
+    if allow_missing:
+        return ""
+    raise RuntimeError(completed.stderr.strip() or completed.stdout.strip() or "git command failed")
+
+
+def collect_diff_text(args: argparse.Namespace) -> str:
+    if args.mode == "pre-commit":
+        return git_output("diff", "--cached", "--unified=0", "--diff-filter=AM", "--no-ext-diff")
+    merge_base = git_output("merge-base", args.merge_target, "HEAD").strip()
+    return git_output(
+        "diff",
+        "--unified=0",
+        "--diff-filter=AM",
+        "--no-ext-diff",
+        f"{merge_base}..HEAD",
+    )
+
+
+def parse_changed_hunks(diff_text: str) -> dict[str, list[Hunk]]:
+    changed_hunks: dict[str, list[Hunk]] = {}
+    current_path: str | None = None
+
+    for line in diff_text.splitlines():
+        if line.startswith("diff --git "):
+            parts = line.split()
+            current_path = parts[3][2:]
+            changed_hunks.setdefault(current_path, [])
+            continue
+
+        if current_path is None:
+            continue
+
+        hunk_match = HUNK_PATTERN.match(line)
+        if not hunk_match:
+            continue
+
+        changed_hunks[current_path].append(
+            Hunk(
+                old_start=int(hunk_match.group("old_start")),
+                old_count=parse_count(hunk_match.group("old_count")),
+                new_start=int(hunk_match.group("new_start")),
+                new_count=parse_count(hunk_match.group("new_count")),
+            )
+        )
+
+    return {path: hunks for path, hunks in changed_hunks.items() if hunks}
+
+
+def parse_count(value: str | None) -> int:
+    if value is None:
+        return 1
+    return int(value)
+
+
+def is_python_source_path(path: str) -> bool:
+    return Path(path).suffix in {".py", ".pyi"}
+
+
+def load_file_versions(path: str, args: argparse.Namespace) -> tuple[str, str]:
+    if args.mode == "pre-commit":
+        return (
+            git_output("show", f"HEAD:{path}", allow_missing=True),
+            git_output("show", f":{path}"),
+        )
+
+    merge_base = git_output("merge-base", args.merge_target, "HEAD").strip()
+    return (
+        git_output("show", f"{merge_base}:{path}", allow_missing=True),
+        git_output("show", f"HEAD:{path}"),
+    )
+
+
+def run_ast_grep(source: str) -> list[Match]:
+    if not source.strip():
+        return []
+
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", encoding="utf-8") as temp_file:
+        temp_file.write(source)
+        temp_file.flush()
+        completed = subprocess.run(
+            [
+                *resolve_ast_grep_command(),
+                "scan",
+                "--rule",
+                str(RULE_PATH),
+                "--json=compact",
+                temp_file.name,
+            ],
+            cwd=REPO_ROOT,
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+
+    if completed.returncode not in (0, 1):
+        raise RuntimeError(completed.stderr.strip() or completed.stdout.strip() or "ast-grep command failed")
+
+    if not completed.stdout.strip():
+        return []
+
+    lines = source.splitlines()
+    raw_matches = json.loads(completed.stdout)
+    matches: list[Match] = []
+    for raw_match in raw_matches:
+        line_number = int(raw_match["range"]["start"]["line"]) + 1
+        source_line = lines[line_number - 1] if line_number - 1 < len(lines) else ""
+        matches.append(Match(line_number=line_number, source_line=source_line))
+    return matches
+
+
+def is_suppressed(source_line: str) -> bool:
+    match = SUPPRESSION_PATTERN.search(source_line)
+    if not match:
+        return False
+    reason = match.group("reason")
+    return reason is not None and bool(reason.strip())
+
+
+def collect_hunk_violations(
+    path: str, old_matches: list[Match], new_matches: list[Match], hunk: Hunk
+) -> list[Violation]:
+    old_in_hunk = [
+        match for match in old_matches if hunk.contains_old(match.line_number) and not is_suppressed(match.source_line)
+    ]
+    new_in_hunk = [
+        match for match in new_matches if hunk.contains_new(match.line_number) and not is_suppressed(match.source_line)
+    ]
+    surplus = len(new_in_hunk) - len(old_in_hunk)
+    if surplus <= 0:
+        return []
+
+    return [
+        Violation(path=path, line_number=match.line_number, message="no-new-getattr net-new getattr() in added code")
+        for match in new_in_hunk[-surplus:]
+    ]
+
+
+def find_net_new_getattr_violations(changed_hunks: dict[str, list[Hunk]], args: argparse.Namespace) -> list[Violation]:
+    violations: list[Violation] = []
+    for path, hunks in changed_hunks.items():
+        if not is_python_source_path(path):
+            continue
+        old_source, new_source = load_file_versions(path, args)
+        old_matches = run_ast_grep(old_source)
+        new_matches = run_ast_grep(new_source)
+        for hunk in hunks:
+            violations.extend(collect_hunk_violations(path, old_matches, new_matches, hunk))
+    return sorted(violations, key=lambda item: (item.path, item.line_number))
+
+
+def print_violations(violations: list[Violation]) -> None:
+    for violation in violations:
+        print(f"{violation.path}:{violation.line_number}: {violation.message}", file=sys.stderr)
+
+
+def main() -> int:
+    try:
+        args = parse_args()
+        changed_hunks = parse_changed_hunks(collect_diff_text(args))
+        violations = find_net_new_getattr_violations(changed_hunks, args)
+    except RuntimeError as exc:
+        print(str(exc), file=sys.stderr)
+        return 2
+
+    if violations:
+        print_violations(violations)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())