From 27882195a47ba259e22dcc586ea50b30e0f1863d Mon Sep 17 00:00:00 2001 From: spencrr <23708360+spencrr@users.noreply.github.com> Date: Tue, 21 Apr 2026 11:17:08 -0700 Subject: [PATCH 1/2] [CI]: Add automated linter tool for repo-level rules --- .github/workflows/ci.yml | 6 +- .pre-commit-config.yaml | 7 + pyproject.toml | 24 ++- rampart/pytest_plugin/plugin.py | 12 +- tools/rampart_lint/__init__.py | 4 + tools/rampart_lint/__main__.py | 119 +++++++++++++++ tools/rampart_lint/driver.py | 103 +++++++++++++ tools/rampart_lint/rule.py | 32 ++++ tools/rampart_lint/rules/__init__.py | 4 + .../rules/rampart001_async_suffix.py | 33 ++++ .../rules/rampart002_keyword_only.py | 43 ++++++ tools/rampart_lint/tests/test_cli.py | 61 ++++++++ tools/rampart_lint/tests/test_driver.py | 144 ++++++++++++++++++ .../tests/test_rampart001_async_suffix.py | 78 ++++++++++ .../tests/test_rampart002_keyword_only.py | 94 ++++++++++++ tools/rampart_lint/tests/test_rule.py | 25 +++ 16 files changed, 781 insertions(+), 8 deletions(-) create mode 100644 tools/rampart_lint/__init__.py create mode 100644 tools/rampart_lint/__main__.py create mode 100644 tools/rampart_lint/driver.py create mode 100644 tools/rampart_lint/rule.py create mode 100644 tools/rampart_lint/rules/__init__.py create mode 100644 tools/rampart_lint/rules/rampart001_async_suffix.py create mode 100644 tools/rampart_lint/rules/rampart002_keyword_only.py create mode 100644 tools/rampart_lint/tests/test_cli.py create mode 100644 tools/rampart_lint/tests/test_driver.py create mode 100644 tools/rampart_lint/tests/test_rampart001_async_suffix.py create mode 100644 tools/rampart_lint/tests/test_rampart002_keyword_only.py create mode 100644 tools/rampart_lint/tests/test_rule.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b4035e3..c4d64e3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -46,6 +46,9 @@ jobs: - name: Pyright run: uv run pyright + - name: RAMPART lint + run: uv run python -m tools.rampart_lint check rampart/ + test: name: Test (Python ${{ matrix.python-version }}) runs-on: ubuntu-latest @@ -77,4 +80,5 @@ jobs: - name: Run tests # Coverage threshold (fail_under) is enforced via pyproject.toml [tool.coverage.report] - run: uv run pytest tests/unit --cov=rampart --cov-report=xml --cov-report=term-missing + # Test paths configured via testpaths in pyproject.toml + run: uv run pytest --cov=rampart --cov-report=xml --cov-report=term-missing diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 84fdeb2..881af72 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -14,3 +14,10 @@ repos: language: system types: [python] pass_filenames: false + - id: rampart-lint + name: rampart-lint (RAMPART001, RAMPART002) + entry: uv run python -m tools.rampart_lint check + language: system + types: [python] + files: ^rampart/ + pass_filenames: true diff --git a/pyproject.toml b/pyproject.toml index 92ef532..963e2d6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,15 +67,22 @@ skip_empty = true [tool.pyright] pythonVersion = "3.11" typeCheckingMode = "strict" -include = ["rampart", "tests"] +include = ["rampart", "tests", "tools"] [[tool.pyright.executionEnvironments]] root = "tests" extraPaths = ["."] reportPrivateUsage = false +[[tool.pyright.executionEnvironments]] +root = "tools" +extraPaths = ["."] +reportPrivateUsage = false + [tool.pytest.ini_options] asyncio_mode = "auto" +pythonpath = ["."] +testpaths = ["tests/unit", "tools/rampart_lint/tests"] markers = [ "harm(*categories): categorize test by harm type", "trial(n=, threshold=): statistical repetition of a test", @@ -87,6 +94,7 @@ filterwarnings = [ [tool.ruff.lint] select = ["ALL"] +external = ["RAMPART001", "RAMPART002"] [tool.ruff.lint.per-file-ignores] "tests/**" = [ @@ -105,6 +113,20 @@ select = ["ALL"] "ASYNC240", # pathlib in async tests "PERF401", "RUF015", "PTH123", # micro-optimizations / style ] +"tools/**/tests/**" = [ + "INP001", # test dirs don't need __init__.py + "S101", # assert is pytest's API + "D100", "D101", "D102", "D104", "D107", # no docstrings needed + "ANN001", "ANN201", "ANN202", "ANN401", # no type annotations needed + "PLR2004", # magic values in assertions are fine + "ARG001", "ARG002", # unused args (fixtures, stubs) + "PLC0415", # imports inside functions for isolation + "SLF001", # testing private members is valid + "TRY003", "EM101", "EM102", # exception message style +] +"tools/**" = [ + "T201", # print is the CLI's output mechanism +] [tool.ruff.lint.flake8-copyright] notice-rgx = "Copyright \\(c\\) Microsoft Corporation\\.\\s*\\n.*Licensed under the MIT license" diff --git a/rampart/pytest_plugin/plugin.py b/rampart/pytest_plugin/plugin.py index c7b5a67..d9482e3 100644 --- a/rampart/pytest_plugin/plugin.py +++ b/rampart/pytest_plugin/plugin.py @@ -241,8 +241,8 @@ def _create_trial_clones( @pytest.hookimpl(trylast=True) -def pytest_collection_modifyitems( - config: pytest.Config, # noqa: ARG001 — pytest hook signature +def pytest_collection_modifyitems( # noqa: RAMPART002 — pytest hook signature + config: pytest.Config, # noqa: ARG001 items: list[pytest.Item], ) -> None: """Clone trial-marked items and validate marker usage. @@ -469,9 +469,9 @@ def _evaluate_gates( ) -def pytest_sessionfinish( +def pytest_sessionfinish( # noqa: RAMPART002 — pytest hook signature session: pytest.Session, - exitstatus: int, # noqa: ARG001 — pytest hook signature + exitstatus: int, # noqa: ARG001 ) -> None: """Aggregate trial results, evaluate gates, and emit sinks. @@ -598,9 +598,9 @@ def _write_trial_group_lines( ) -def pytest_terminal_summary( +def pytest_terminal_summary( # noqa: RAMPART002 — pytest hook signature terminalreporter: TerminalReporter, - exitstatus: int, # noqa: ARG001 — pytest hook signature + exitstatus: int, # noqa: ARG001 config: pytest.Config, ) -> None: """Append RAMPART harm-category summary after pytest's standard output. diff --git a/tools/rampart_lint/__init__.py b/tools/rampart_lint/__init__.py new file mode 100644 index 0000000..81cc220 --- /dev/null +++ b/tools/rampart_lint/__init__.py @@ -0,0 +1,4 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""RAMPART custom lint rules package.""" diff --git a/tools/rampart_lint/__main__.py b/tools/rampart_lint/__main__.py new file mode 100644 index 0000000..bc3b0f8 --- /dev/null +++ b/tools/rampart_lint/__main__.py @@ -0,0 +1,119 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""CLI entry point for RAMPART's custom lint rules. + +Usage:: + + python -m tools.rampart_lint check rampart/ + python -m tools.rampart_lint list + python -m tools.rampart_lint explain RAMPART001 + +Exit code 0 = clean, 1 = violations found, 2 = usage error. +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path +from typing import TYPE_CHECKING + +from tools.rampart_lint.driver import RULES, check_file + +if TYPE_CHECKING: + from collections.abc import Callable + + +def _collect_paths(targets: list[Path]) -> list[Path]: + """Expand CLI targets into a sorted list of ``.py`` files.""" + paths: list[Path] = [] + for p in targets: + if p.is_file() and p.suffix == ".py": + paths.append(p) + elif p.is_dir(): + paths.extend(sorted(p.rglob("*.py"))) + else: + print(f"Skipping non-Python target: {p}", file=sys.stderr) + return paths + + +def _cmd_check(args: argparse.Namespace) -> int: + """Run lint checks on the given paths.""" + paths = _collect_paths(args.paths) + all_errors: list[str] = [] + for path in paths: + all_errors.extend(check_file(path)) + + for err in all_errors: + print(err) + + return 1 if all_errors else 0 + + +def _cmd_list(_args: argparse.Namespace) -> int: + """Print a summary table of all registered rules.""" + for rule in RULES: + print(f"{rule.code} {rule.description}") + return 0 + + +def _cmd_explain(args: argparse.Namespace) -> int: + """Print detailed information about a single rule.""" + for rule in RULES: + if rule.code == args.code: + node_names = ", ".join(t.__name__ for t in rule.node_types) + print(f"Code: {rule.code}") + print(f"Description: {rule.description}") + print(f"Class: {type(rule).__name__}") + print(f"Node types: {node_names}") + return 0 + print(f"Unknown rule code: {args.code}", file=sys.stderr) + return 2 + + +def _build_parser() -> argparse.ArgumentParser: + """Build the CLI argument parser.""" + parser = argparse.ArgumentParser( + prog="rampart_lint", + description="RAMPART custom lint rules (RAMPART001, RAMPART002, ...).", + ) + sub = parser.add_subparsers(dest="command") + + check = sub.add_parser("check", help="Lint files or directories.") + check.add_argument( + "paths", + nargs="+", + type=Path, + help="Files or directories to lint.", + ) + check.set_defaults(func=_cmd_check) + + list_cmd = sub.add_parser("list", help="List all registered rules.") + list_cmd.set_defaults(func=_cmd_list) + + explain = sub.add_parser("explain", help="Show details for a rule code.") + explain.add_argument( + "code", + type=str, + help="Rule code (e.g. RAMPART001).", + ) + explain.set_defaults(func=_cmd_explain) + + return parser + + +def main(argv: list[str] | None = None) -> int: + """Parse arguments and dispatch to the appropriate subcommand.""" + parser = _build_parser() + args = parser.parse_args(args=argv) + + if not args.command: + parser.print_help(sys.stderr) + return 2 + + func: Callable[[argparse.Namespace], int] = args.func + return func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/rampart_lint/driver.py b/tools/rampart_lint/driver.py new file mode 100644 index 0000000..49cdf19 --- /dev/null +++ b/tools/rampart_lint/driver.py @@ -0,0 +1,103 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""Lint engine: dispatch, suppression, and file checking.""" + +from __future__ import annotations + +import ast +from typing import TYPE_CHECKING + +from tools.rampart_lint.rule import LocatedNode +from tools.rampart_lint.rules.rampart001_async_suffix import AsyncSuffixRule +from tools.rampart_lint.rules.rampart002_keyword_only import KeywordOnlyRule + +if TYPE_CHECKING: + from pathlib import Path + + from tools.rampart_lint.rule import Rule + +RULES: list[Rule] = [ + AsyncSuffixRule(), + KeywordOnlyRule(), +] + + +def build_dispatch( + rules: list[Rule], +) -> dict[type[LocatedNode], list[Rule]]: + """Map each AST node type to the rules that inspect it. + + Raises: + ValueError: If two rules share the same code. + """ + seen_codes: dict[str, str] = {} + dispatch: dict[type[LocatedNode], list[Rule]] = {} + + for rule in rules: + owner = type(rule).__name__ + if rule.code in seen_codes: + msg = ( + f"Duplicate rule code {rule.code}: {seen_codes[rule.code]} and {owner}" + ) + raise ValueError(msg) + seen_codes[rule.code] = owner + + for node_type in rule.node_types: + dispatch.setdefault(node_type, []).append(rule) + + return dispatch + + +def is_suppressed(lines: list[str], lineno: int, code: str) -> bool: + """Check whether a ``# noqa`` comment suppresses the violation. + + Follows ruff semantics: the ``# noqa`` must appear on the same line as + the violation. Supports both targeted (``# noqa: RAMPART001``) and bare + (``# noqa``) suppression. + + Args: + lines: Source lines (0-indexed list). + lineno: 1-based line number from the AST node. + code: Rule code to look for (e.g. ``RAMPART001``). + """ + idx = lineno - 1 + if idx < 0 or idx >= len(lines): + return False + line = lines[idx] + if "# noqa" not in line: + return False + # Bare suppression (no colon after the directive) covers all codes. + noqa_pos = line.index("# noqa") + after = line[noqa_pos + len("# noqa") :] + stripped = after.lstrip() + if not stripped or stripped[0] != ":": + return True + # Targeted: check if our code appears after the colon. + return code in after + + +def check_file(path: Path) -> list[str]: + """Parse a file and run all matching rules against its AST.""" + source = path.read_text(encoding="utf-8") + try: + tree = ast.parse(source, filename=str(path)) + except SyntaxError: + return [f"{path}:0: could not parse file"] + + source_lines = source.splitlines() + errors: list[str] = [] + + dispatch = build_dispatch(RULES) + + for node in ast.walk(tree): + if not isinstance(node, LocatedNode): + continue + rules = dispatch.get(type(node)) + if rules is None: + continue + for rule in rules: + msg = rule.check(node) + if msg and not is_suppressed(source_lines, node.lineno, rule.code): + errors.append(f"{path}:{node.lineno}: {rule.code} {msg}") + + return errors diff --git a/tools/rampart_lint/rule.py b/tools/rampart_lint/rule.py new file mode 100644 index 0000000..3bcb2d0 --- /dev/null +++ b/tools/rampart_lint/rule.py @@ -0,0 +1,32 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""Rule protocol and shared utilities for rampart_lint.""" + +from __future__ import annotations + +import ast +from typing import TYPE_CHECKING, Protocol, TypeAlias + +if TYPE_CHECKING: + from collections.abc import Sequence + +# AST nodes that carry source location (lineno, col_offset, etc.). +# Using this instead of ast.AST gives typed access to .lineno +LocatedNode: TypeAlias = ast.stmt | ast.expr + + +class Rule(Protocol): + """A single lint rule that checks an AST node.""" + + code: str + description: str + node_types: Sequence[type[LocatedNode]] + + def check(self, node: LocatedNode) -> str | None: + """Return an error message if the node violates the rule, else None.""" + ... + + +def is_dunder(name: str) -> bool: + """Return True for Python dunder names like ``__init__``.""" + return name.startswith("__") and name.endswith("__") diff --git a/tools/rampart_lint/rules/__init__.py b/tools/rampart_lint/rules/__init__.py new file mode 100644 index 0000000..dba25c2 --- /dev/null +++ b/tools/rampart_lint/rules/__init__.py @@ -0,0 +1,4 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Individual lint rule implementations.""" diff --git a/tools/rampart_lint/rules/rampart001_async_suffix.py b/tools/rampart_lint/rules/rampart001_async_suffix.py new file mode 100644 index 0000000..443e7ff --- /dev/null +++ b/tools/rampart_lint/rules/rampart001_async_suffix.py @@ -0,0 +1,33 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""RAMPART001: async functions must end with ``_async`` (dunders exempt).""" + +from __future__ import annotations + +import ast +from typing import TYPE_CHECKING + +from tools.rampart_lint.rule import is_dunder + +if TYPE_CHECKING: + from collections.abc import Sequence + + from tools.rampart_lint.rule import LocatedNode + + +class AsyncSuffixRule: + """Async functions must end with the ``_async`` suffix.""" + + code = "RAMPART001" + description = "async function missing `_async` suffix" + node_types: Sequence[type[LocatedNode]] = (ast.AsyncFunctionDef,) + + def check(self, node: LocatedNode) -> str | None: + """Return an error if the async function lacks the ``_async`` suffix.""" + if not isinstance(node, ast.AsyncFunctionDef): + return None + if is_dunder(node.name): + return None + if node.name.endswith("_async"): + return None + return f"async function `{node.name}` missing `_async` suffix" diff --git a/tools/rampart_lint/rules/rampart002_keyword_only.py b/tools/rampart_lint/rules/rampart002_keyword_only.py new file mode 100644 index 0000000..007fbd1 --- /dev/null +++ b/tools/rampart_lint/rules/rampart002_keyword_only.py @@ -0,0 +1,43 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""RAMPART002: multi-param functions must use keyword-only ``*`` separator.""" + +from __future__ import annotations + +import ast +from typing import TYPE_CHECKING + +from tools.rampart_lint.rule import is_dunder + +if TYPE_CHECKING: + from collections.abc import Sequence + + from tools.rampart_lint.rule import LocatedNode + + +class KeywordOnlyRule: + """Functions with >1 non-self/cls param must enforce keyword-only args.""" + + code = "RAMPART002" + description = "function has multiple positional params — use `*`" + node_types: Sequence[type[LocatedNode]] = ( + ast.FunctionDef, + ast.AsyncFunctionDef, + ) + + def check(self, node: LocatedNode) -> str | None: + """Return an error if the function allows >1 positional param.""" + if not isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef): + return None + if is_dunder(node.name): + return None + + positional = list(node.args.posonlyargs) + list(node.args.args) + non_self = [a for a in positional if a.arg not in ("self", "cls")] + + if len(non_self) > 1: + return ( + f"function `{node.name}` has {len(non_self)} positional " + f"params — use `*` to enforce keyword-only arguments" + ) + return None diff --git a/tools/rampart_lint/tests/test_cli.py b/tools/rampart_lint/tests/test_cli.py new file mode 100644 index 0000000..ae68e63 --- /dev/null +++ b/tools/rampart_lint/tests/test_cli.py @@ -0,0 +1,61 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +from pathlib import Path + +import pytest + +from tools.rampart_lint.__main__ import main + + +class TestSubcommandCheck: + def test_no_args_returns_2(self) -> None: + assert main([]) == 2 + + def test_clean_file_returns_0(self, tmp_path: Path) -> None: + f = tmp_path / "ok.py" + f.write_text("async def go_async(): ...\n") + assert main(["check", str(f)]) == 0 + + def test_violations_returns_1(self, tmp_path: Path) -> None: + f = tmp_path / "bad.py" + f.write_text("async def go(): ...\n") + assert main(["check", str(f)]) == 1 + + def test_directory_input(self, tmp_path: Path) -> None: + sub = tmp_path / "pkg" + sub.mkdir() + (sub / "a.py").write_text("async def go(): ...\n") + (sub / "b.py").write_text("async def go_async(): ...\n") + assert main(["check", str(sub)]) == 1 + + def test_non_python_skipped(self, tmp_path: Path) -> None: + f = tmp_path / "readme.txt" + f.write_text("hello") + assert main(["check", str(f)]) == 0 + + +class TestSubcommandList: + def test_returns_0(self) -> None: + assert main(["list"]) == 0 + + def test_prints_all_codes(self, capsys: pytest.CaptureFixture[str]) -> None: + main(["list"]) + output = capsys.readouterr().out + assert "RAMPART001" in output + assert "RAMPART002" in output + + +class TestSubcommandExplain: + def test_known_code_returns_0(self) -> None: + assert main(["explain", "RAMPART001"]) == 0 + + def test_known_code_output(self, capsys: pytest.CaptureFixture[str]) -> None: + main(["explain", "RAMPART001"]) + output = capsys.readouterr().out + assert "RAMPART001" in output + assert "AsyncSuffixRule" in output + assert "AsyncFunctionDef" in output + + def test_unknown_code_returns_2(self) -> None: + assert main(["explain", "RAM999"]) == 2 diff --git a/tools/rampart_lint/tests/test_driver.py b/tools/rampart_lint/tests/test_driver.py new file mode 100644 index 0000000..ad37028 --- /dev/null +++ b/tools/rampart_lint/tests/test_driver.py @@ -0,0 +1,144 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +import ast +import textwrap +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from tools.rampart_lint.driver import build_dispatch, check_file, is_suppressed + +# -- is_suppressed ------------------------------------------------------------- + + +class TestIsSuppressed: + def test_same_line_targeted(self) -> None: + lines = ["async def fetch(): ... # noqa: RAMPART001"] + assert is_suppressed(lines, lineno=1, code="RAMPART001") is True + + def test_wrong_code(self) -> None: + lines = ["async def fetch(): ... # noqa: RAMPART002"] + assert is_suppressed(lines, lineno=1, code="RAMPART001") is False + + def test_different_line(self) -> None: + lines = [ + "# noqa: RAMPART001", + "async def fetch(): ...", + ] + assert is_suppressed(lines, lineno=2, code="RAMPART001") is False + + def test_bare_noqa_suppresses_all(self) -> None: + lines = ["async def fetch(): ... # noqa"] + assert is_suppressed(lines, lineno=1, code="RAMPART001") is True + + def test_bare_noqa_with_trailing_comment(self) -> None: + lines = ["async def fetch(): ... # noqa # some reason"] + assert is_suppressed(lines, lineno=1, code="RAMPART002") is True + + def test_multi_code_suppression(self) -> None: + lines = ["async def f(self, a, b): ... # noqa: RAMPART001, RAMPART002"] + assert is_suppressed(lines, lineno=1, code="RAMPART001") is True + assert is_suppressed(lines, lineno=1, code="RAMPART002") is True + + def test_first_line(self) -> None: + lines = ["async def fetch(): ... # noqa: RAMPART001"] + assert is_suppressed(lines, lineno=1, code="RAMPART001") is True + + def test_last_line(self) -> None: + lines = ["x = 1", "async def fetch(): ... # noqa: RAMPART001"] + assert is_suppressed(lines, lineno=2, code="RAMPART001") is True + + def test_empty_lines(self) -> None: + assert is_suppressed([], lineno=1, code="RAMPART001") is False + + def test_lineno_out_of_range(self) -> None: + lines = ["x = 1"] + assert is_suppressed(lines, lineno=99, code="RAMPART001") is False + + def test_noqa_substring_in_string_literal(self) -> None: + lines = ['x = "# noqa: RAMPART001"'] + # Known limitation: substring match, not comment parse. + # Documenting current behavior: it WILL suppress (matches ruff). + assert is_suppressed(lines, lineno=1, code="RAMPART001") is True + + +# -- build_dispatch ------------------------------------------------------------ + + +class TestBuildDispatch: + def test_duplicate_codes_raises(self) -> None: + rule_a = MagicMock(code="RAMPART001", node_types=(ast.AsyncFunctionDef,)) + rule_b = MagicMock(code="RAMPART001", node_types=(ast.FunctionDef,)) + type(rule_a).__name__ = "RuleA" + type(rule_b).__name__ = "RuleB" + + with pytest.raises(ValueError, match="Duplicate rule code RAMPART001"): + build_dispatch([rule_a, rule_b]) + + def test_empty_rules(self) -> None: + assert build_dispatch([]) == {} + + def test_two_rules_same_node_type(self) -> None: + rule_a = MagicMock( + code="RAMPART001", + node_types=(ast.AsyncFunctionDef,), + ) + rule_b = MagicMock( + code="RAMPART002", + node_types=(ast.AsyncFunctionDef, ast.FunctionDef), + ) + dispatch = build_dispatch([rule_a, rule_b]) + assert len(dispatch[ast.AsyncFunctionDef]) == 2 + assert len(dispatch[ast.FunctionDef]) == 1 + + +# -- check_file ---------------------------------------------------------------- + + +class TestCheckFile: + def test_finds_violations(self, tmp_path: Path) -> None: + f = tmp_path / "bad.py" + f.write_text("async def fetch(): ...\n") + errors = check_file(f) + assert len(errors) == 1 + assert "RAMPART001" in errors[0] + + def test_noqa_suppresses(self, tmp_path: Path) -> None: + f = tmp_path / "suppressed.py" + f.write_text("async def fetch(): ... # noqa: RAMPART001\n") + assert check_file(f) == [] + + def test_syntax_error(self, tmp_path: Path) -> None: + f = tmp_path / "broken.py" + f.write_text("def (\n") + errors = check_file(f) + assert len(errors) == 1 + assert "could not parse" in errors[0] + + def test_empty_file(self, tmp_path: Path) -> None: + f = tmp_path / "empty.py" + f.write_text("") + assert check_file(f) == [] + + def test_clean_file(self, tmp_path: Path) -> None: + f = tmp_path / "clean.py" + f.write_text( + textwrap.dedent("""\ + async def fetch_async(): ... + def process(self, *, a, b): ... + """), + ) + assert check_file(f) == [] + + def test_multiple_violations(self, tmp_path: Path) -> None: + f = tmp_path / "multi.py" + f.write_text( + textwrap.dedent("""\ + async def one(): ... + async def two(): ... + """), + ) + errors = check_file(f) + assert len(errors) == 2 diff --git a/tools/rampart_lint/tests/test_rampart001_async_suffix.py b/tools/rampart_lint/tests/test_rampart001_async_suffix.py new file mode 100644 index 0000000..3688a02 --- /dev/null +++ b/tools/rampart_lint/tests/test_rampart001_async_suffix.py @@ -0,0 +1,78 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +import ast +import textwrap + +from tools.rampart_lint.rule import LocatedNode +from tools.rampart_lint.rules.rampart001_async_suffix import AsyncSuffixRule + + +def _parse_first(source: str) -> LocatedNode: + """Parse source and return the first function/async-function node.""" + tree = ast.parse(textwrap.dedent(source)) + for node in ast.walk(tree): + if isinstance(node, ast.AsyncFunctionDef | ast.FunctionDef): + return node + msg = "no function found in source" + raise ValueError(msg) + + +rule = AsyncSuffixRule() + + +class TestCatches: + def test_plain_async_function(self) -> None: + node = _parse_first("async def fetch(): ...") + assert rule.check(node) is not None + + def test_private_async_function(self) -> None: + node = _parse_first("async def _fetch(): ...") + assert rule.check(node) is not None + + def test_async_in_middle_of_name(self) -> None: + node = _parse_first("async def _async_helper(): ...") + result = rule.check(node) + assert result is not None + assert "_async_helper" in result + + def test_nested_async_function(self) -> None: + source = """\ + def outer(): + async def inner(): ... + """ + tree = ast.parse(textwrap.dedent(source)) + violations = [ + rule.check(n) for n in ast.walk(tree) if isinstance(n, ast.AsyncFunctionDef) + ] + assert any(v is not None for v in violations) + + +class TestPasses: + def test_async_suffix(self) -> None: + node = _parse_first("async def fetch_async(): ...") + assert rule.check(node) is None + + def test_dunder_aenter(self) -> None: + node = _parse_first("async def __aenter__(): ...") + assert rule.check(node) is None + + def test_dunder_aexit(self) -> None: + node = _parse_first("async def __aexit__(): ...") + assert rule.check(node) is None + + def test_sync_function_ignored(self) -> None: + node = _parse_first("def fetch(): ...") + assert rule.check(node) is None + + def test_name_exactly_underscore_async(self) -> None: + node = _parse_first("async def _async(): ...") + assert rule.check(node) is None + + +class TestMetadata: + def test_code(self) -> None: + assert rule.code == "RAMPART001" + + def test_node_types(self) -> None: + assert ast.AsyncFunctionDef in rule.node_types diff --git a/tools/rampart_lint/tests/test_rampart002_keyword_only.py b/tools/rampart_lint/tests/test_rampart002_keyword_only.py new file mode 100644 index 0000000..42d2a0c --- /dev/null +++ b/tools/rampart_lint/tests/test_rampart002_keyword_only.py @@ -0,0 +1,94 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +import ast +import textwrap + +from tools.rampart_lint.rule import LocatedNode +from tools.rampart_lint.rules.rampart002_keyword_only import KeywordOnlyRule + + +def _parse_first(source: str) -> LocatedNode: + """Parse source and return the first function node.""" + tree = ast.parse(textwrap.dedent(source)) + for node in ast.walk(tree): + if isinstance(node, ast.AsyncFunctionDef | ast.FunctionDef): + return node + msg = "no function found in source" + raise ValueError(msg) + + +rule = KeywordOnlyRule() + + +class TestCatches: + def test_two_positional_method(self) -> None: + node = _parse_first("def f(self, a, b): ...") + result = rule.check(node) + assert result is not None + assert "2" in result + + def test_three_positional_standalone(self) -> None: + node = _parse_first("def f(a, b, c): ...") + result = rule.check(node) + assert result is not None + assert "3" in result + + def test_cls_plus_two_positional(self) -> None: + node = _parse_first("def f(cls, a, b): ...") + assert rule.check(node) is not None + + def test_async_function(self) -> None: + node = _parse_first("async def f(self, a, b): ...") + assert rule.check(node) is not None + + def test_positional_only_counted(self) -> None: + node = _parse_first("def f(a, b, /): ...") + assert rule.check(node) is not None + + +class TestPasses: + def test_single_param_method(self) -> None: + node = _parse_first("def f(self, a): ...") + assert rule.check(node) is None + + def test_single_param_standalone(self) -> None: + node = _parse_first("def f(a): ...") + assert rule.check(node) is None + + def test_zero_params(self) -> None: + node = _parse_first("def f(): ...") + assert rule.check(node) is None + + def test_keyword_only(self) -> None: + node = _parse_first("def f(self, *, a, b): ...") + assert rule.check(node) is None + + def test_dunder_init(self) -> None: + node = _parse_first("def __init__(self, a, b): ...") + assert rule.check(node) is None + + def test_dunder_eq(self) -> None: + node = _parse_first("def __eq__(self, other): ...") + assert rule.check(node) is None + + def test_vararg(self) -> None: + node = _parse_first("def f(self, *args): ...") + assert rule.check(node) is None + + def test_kwargs_only(self) -> None: + node = _parse_first("def f(self, **kwargs): ...") + assert rule.check(node) is None + + def test_cls_single_keyword_only(self) -> None: + node = _parse_first("def f(cls, *, a): ...") + assert rule.check(node) is None + + +class TestMetadata: + def test_code(self) -> None: + assert rule.code == "RAMPART002" + + def test_node_types(self) -> None: + assert ast.FunctionDef in rule.node_types + assert ast.AsyncFunctionDef in rule.node_types diff --git a/tools/rampart_lint/tests/test_rule.py b/tools/rampart_lint/tests/test_rule.py new file mode 100644 index 0000000..3a1fcb7 --- /dev/null +++ b/tools/rampart_lint/tests/test_rule.py @@ -0,0 +1,25 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +import pytest + +from tools.rampart_lint.rule import is_dunder + + +class TestIsDunder: + @pytest.mark.parametrize( + ("name", "expected"), + [ + ("__init__", True), + ("__aenter__", True), + ("__eq__", True), + ("____", True), + ("_private", False), + ("__leading", False), + ("trailing__", False), + ("normal", False), + ("", False), + ], + ) + def test_detection(self, name: str, expected: bool) -> None: # noqa: FBT001 + assert is_dunder(name) is expected From c895536d72863236d3e1328be2364a503e4a5dd5 Mon Sep 17 00:00:00 2001 From: spencrr <23708360+spencrr@users.noreply.github.com> Date: Tue, 21 Apr 2026 13:27:32 -0700 Subject: [PATCH 2/2] fixup asyncs to pass precommit --- rampart/attacks/_xpia.py | 2 +- rampart/core/execution.py | 12 ++++++------ rampart/core/injection.py | 6 +++--- rampart/pytest_plugin/_collection.py | 2 +- rampart/surfaces/onedrive.py | 6 +++--- tests/unit/attacks/test_xpia.py | 4 ++-- tests/unit/core/test_execution.py | 4 ++-- tests/unit/core/test_protocols.py | 4 ++-- tests/unit/pytest_plugin/test_collection.py | 10 +++++----- tests/unit/surfaces/test_onedrive.py | 10 +++++----- 10 files changed, 30 insertions(+), 30 deletions(-) diff --git a/rampart/attacks/_xpia.py b/rampart/attacks/_xpia.py index af2c8df..e0b376d 100644 --- a/rampart/attacks/_xpia.py +++ b/rampart/attacks/_xpia.py @@ -192,7 +192,7 @@ async def _activate_handles_async( # Concurrent: total = max of all wait times async with asyncio.TaskGroup() as tg: for handle in self._handles: - tg.create_task(handle.wait_until_ready()) + tg.create_task(handle.wait_until_ready_async()) def _build_attack_result( self, diff --git a/rampart/core/execution.py b/rampart/core/execution.py index c9b8b06..a3d4e24 100644 --- a/rampart/core/execution.py +++ b/rampart/core/execution.py @@ -69,7 +69,7 @@ class ExecutionEventHandler(ABC): """ @abstractmethod - async def on_event(self, *, event_data: ExecutionEventData) -> None: + async def on_event_async(self, *, event_data: ExecutionEventData) -> None: """Handle an execution lifecycle event. Args: @@ -228,7 +228,7 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result: _execute_async, after notifying handlers via ON_ERROR. """ start = time.monotonic() - await self._fire( + await self._fire_async( ExecutionEvent.ON_PRE_EXECUTE, adapter=adapter, elapsed=0.0, @@ -253,7 +253,7 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result: ) except Exception as exc: elapsed = time.monotonic() - start - await self._fire( + await self._fire_async( ExecutionEvent.ON_ERROR, adapter=adapter, elapsed=elapsed, @@ -263,7 +263,7 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result: elapsed = time.monotonic() - start result.duration_seconds = elapsed - await self._fire( + await self._fire_async( ExecutionEvent.ON_POST_EXECUTE, adapter=adapter, elapsed=elapsed, @@ -283,7 +283,7 @@ async def _execute_async(self, *, adapter: AgentAdapter) -> Result: """ ... - async def _fire( + async def _fire_async( self, event: ExecutionEvent, *, @@ -313,7 +313,7 @@ async def _fire( ) for handler in self._handlers: try: - await handler.on_event(event_data=event_data) + await handler.on_event_async(event_data=event_data) except Exception: # noqa: BLE001 — handler errors must not break execution logger.warning( "ExecutionEventHandler %s raised on %s — ignored.", diff --git a/rampart/core/injection.py b/rampart/core/injection.py index 85fceba..b6d219d 100644 --- a/rampart/core/injection.py +++ b/rampart/core/injection.py @@ -6,7 +6,7 @@ Two protocols serving two audiences: Surface is what surface authors implement; InjectionHandle is what execution strategies consume. -``sleep_until_ready`` is a helper function for surfaces that only need +``sleep_until_ready_async`` is a helper function for surfaces that only need a simple delay-based readiness wait. """ @@ -43,7 +43,7 @@ def surface_name(self) -> str: """The name of the surface this handle injects into (e.g., 'SharePoint').""" ... - async def wait_until_ready(self) -> None: + async def wait_until_ready_async(self) -> None: """Block until the injected content is visible to the agent. Implementations should raise `TimeoutError` if readiness @@ -65,7 +65,7 @@ async def __aexit__( ... -async def sleep_until_ready(delay: float) -> None: +async def sleep_until_ready_async(delay: float) -> None: """Sleep for `delay` seconds. Default readiness strategy for simple surfaces. Args: diff --git a/rampart/pytest_plugin/_collection.py b/rampart/pytest_plugin/_collection.py index b540bea..220e795 100644 --- a/rampart/pytest_plugin/_collection.py +++ b/rampart/pytest_plugin/_collection.py @@ -83,7 +83,7 @@ class ResultCollectionHandler(ExecutionEventHandler): collector is active (safe to use outside pytest). """ - async def on_event(self, *, event_data: ExecutionEventData) -> None: + async def on_event_async(self, *, event_data: ExecutionEventData) -> None: """Record result on post-execute. Ignore all other events. Args: diff --git a/rampart/surfaces/onedrive.py b/rampart/surfaces/onedrive.py index d239743..1858c54 100644 --- a/rampart/surfaces/onedrive.py +++ b/rampart/surfaces/onedrive.py @@ -13,7 +13,7 @@ from typing import TYPE_CHECKING, Self from rampart.core.errors import InfrastructureError -from rampart.core.injection import sleep_until_ready +from rampart.core.injection import sleep_until_ready_async if TYPE_CHECKING: import types @@ -196,14 +196,14 @@ def surface_name(self) -> str: """Identifies this injection as OneDrive for reporting.""" return "OneDrive" - async def wait_until_ready(self) -> None: + async def wait_until_ready_async(self) -> None: """Wait for the uploaded content to be indexed and discoverable. Note: Currently sleeps for `OneDriveSurface.indexing_delay` seconds. Future versions will poll the Graph API for content availability instead and raise `TimeoutError` if it doesn't appear within the `indexing_delay`. """ - await sleep_until_ready(delay=self._surface.indexing_delay) + await sleep_until_ready_async(delay=self._surface.indexing_delay) async def __aenter__(self) -> Self: """Upload payload to OneDrive. Raises InfrastructureError on failure.""" diff --git a/tests/unit/attacks/test_xpia.py b/tests/unit/attacks/test_xpia.py index 69580f8..7e62f19 100644 --- a/tests/unit/attacks/test_xpia.py +++ b/tests/unit/attacks/test_xpia.py @@ -177,7 +177,7 @@ async def test_handle_entered_and_exited(self) -> None: handle.__aenter__.assert_awaited_once() handle.__aexit__.assert_awaited_once() - handle.wait_until_ready.assert_awaited_once() + handle.wait_until_ready_async.assert_awaited_once() @pytest.mark.asyncio async def test_multiple_handles_all_cleaned(self) -> None: @@ -193,7 +193,7 @@ async def test_multiple_handles_all_cleaned(self) -> None: for h in (h1, h2): h.__aenter__.assert_awaited_once() h.__aexit__.assert_awaited_once() - h.wait_until_ready.assert_awaited_once() + h.wait_until_ready_async.assert_awaited_once() @pytest.mark.asyncio async def test_cleanup_on_evaluator_exception(self) -> None: diff --git a/tests/unit/core/test_execution.py b/tests/unit/core/test_execution.py index 811ad3b..1089c25 100644 --- a/tests/unit/core/test_execution.py +++ b/tests/unit/core/test_execution.py @@ -102,7 +102,7 @@ class _RecordingHandler(ExecutionEventHandler): def __init__(self) -> None: self.events: list[ExecutionEventData] = [] - async def on_event(self, *, event_data: ExecutionEventData) -> None: + async def on_event_async(self, *, event_data: ExecutionEventData) -> None: """Record the event data.""" self.events.append(event_data) @@ -110,7 +110,7 @@ async def on_event(self, *, event_data: ExecutionEventData) -> None: class _BrokenHandler(ExecutionEventHandler): """Handler that always raises.""" - async def on_event(self, *, event_data: ExecutionEventData) -> None: + async def on_event_async(self, *, event_data: ExecutionEventData) -> None: """Raise unconditionally to test handler safety.""" raise ValueError("handler broke") diff --git a/tests/unit/core/test_protocols.py b/tests/unit/core/test_protocols.py index 8a0bb69..96ac525 100644 --- a/tests/unit/core/test_protocols.py +++ b/tests/unit/core/test_protocols.py @@ -86,7 +86,7 @@ def payload_id(self) -> str | None: def surface_name(self) -> str: return "SharePoint" - async def wait_until_ready(self) -> None: + async def wait_until_ready_async(self) -> None: pass async def __aenter__(self) -> Self: @@ -114,7 +114,7 @@ def payload_id(self) -> str | None: def surface_name(self) -> str: return "test" - async def wait_until_ready(self) -> None: + async def wait_until_ready_async(self) -> None: pass async def __aenter__(self) -> Self: diff --git a/tests/unit/pytest_plugin/test_collection.py b/tests/unit/pytest_plugin/test_collection.py index 8d270b2..a08e5c4 100644 --- a/tests/unit/pytest_plugin/test_collection.py +++ b/tests/unit/pytest_plugin/test_collection.py @@ -83,7 +83,7 @@ async def test_records_on_post_execute_async(self) -> None: result=result, ) - await handler.on_event(event_data=event_data) + await handler.on_event_async(event_data=event_data) assert len(collector.results) == 1 assert collector.results[0].summary == "captured" @@ -98,7 +98,7 @@ async def test_ignores_pre_execute_async(self) -> None: handler = ResultCollectionHandler() event_data = _make_event_data(event=ExecutionEvent.ON_PRE_EXECUTE) - await handler.on_event(event_data=event_data) + await handler.on_event_async(event_data=event_data) assert collector.results == [] finally: @@ -112,7 +112,7 @@ async def test_ignores_on_error_async(self) -> None: handler = ResultCollectionHandler() event_data = _make_event_data(event=ExecutionEvent.ON_ERROR) - await handler.on_event(event_data=event_data) + await handler.on_event_async(event_data=event_data) assert collector.results == [] finally: @@ -127,7 +127,7 @@ async def test_noop_when_no_collector_active_async(self) -> None: result=result, ) - await handler.on_event(event_data=event_data) + await handler.on_event_async(event_data=event_data) @pytest.mark.asyncio async def test_noop_when_result_is_none_async(self) -> None: @@ -140,7 +140,7 @@ async def test_noop_when_result_is_none_async(self) -> None: result=None, ) - await handler.on_event(event_data=event_data) + await handler.on_event_async(event_data=event_data) assert collector.results == [] finally: diff --git a/tests/unit/surfaces/test_onedrive.py b/tests/unit/surfaces/test_onedrive.py index 12434ce..6b39a05 100644 --- a/tests/unit/surfaces/test_onedrive.py +++ b/tests/unit/surfaces/test_onedrive.py @@ -353,11 +353,11 @@ async def test_infrastructure_error_from_upload_not_double_wrapped(self) -> None class TestOneDriveInjectionWaitUntilReady: - """Test _OneDriveInjection.wait_until_ready wiring.""" + """Test _OneDriveInjection.wait_until_ready_async wiring.""" @pytest.mark.asyncio - async def test_delegates_to_sleep_until_ready(self) -> None: - """Verifies correct arguments are passed to sleep_until_ready.""" + async def test_delegates_to_sleep_until_ready_async(self) -> None: + """Verifies correct arguments are passed to sleep_until_ready_async.""" surface = OneDriveSurface( graph_client=MagicMock(), drive_id="d", @@ -367,9 +367,9 @@ async def test_delegates_to_sleep_until_ready(self) -> None: handle = surface.inject(payload=Payload(content="test")) with patch( - "rampart.surfaces.onedrive.sleep_until_ready", + "rampart.surfaces.onedrive.sleep_until_ready_async", new_callable=AsyncMock, ) as mock_sleep: - await handle.wait_until_ready() + await handle.wait_until_ready_async() mock_sleep.assert_awaited_once_with(delay=5.0)