diff --git a/.github/workflows/style.yml b/.github/workflows/style.yml index 1cc46205d14..0c34de4a7b9 100644 --- a/.github/workflows/style.yml +++ b/.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: diff --git a/api/tests/unit_tests/commands/test_check_no_new_getattr.py b/api/tests/unit_tests/commands/test_check_no_new_getattr.py new file mode 100644 index 00000000000..c8fadfe982b --- /dev/null +++ b/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.*?)(?=^ [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.*?)(?=^ - 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(?:^ \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.*?{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.*?)(?=^ - 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.*?)(?=^ - 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 diff --git a/scripts/ast_grep_rules/no_new_getattr.yml b/scripts/ast_grep_rules/no_new_getattr.yml new file mode 100644 index 00000000000..60832ff2ed1 --- /dev/null +++ b/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 diff --git a/scripts/check_no_new_getattr.py b/scripts/check_no_new_getattr.py new file mode 100644 index 00000000000..102593487e3 --- /dev/null +++ b/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\d+)(?:,(?P\d+))? \+(?P\d+)(?:,(?P\d+))? @@" +) +SUPPRESSION_PATTERN = re.compile(r"# noqa: no-new-getattr(?:\s+(?P\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())