diff --git a/maf_starter/critic.py b/maf_starter/critic.py new file mode 100644 index 0000000..a764a95 --- /dev/null +++ b/maf_starter/critic.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +from dataclasses import dataclass +import re +import sys +from typing import Literal + +from maf_starter.config import Settings + + +@dataclass(frozen=True) +class Finding: + check: str + severity: Literal["blocking", "advisory"] + file: str + line: int | None + detail: str + + +@dataclass(frozen=True) +class DiffFile: + path: str + added_lines: list[str] + + +@dataclass(frozen=True) +class CriticReport: + findings: list[Finding] + blocking_count: int + advisory_count: int + exit_code: int + + +_BARE_EXCEPT_PATTERN = re.compile(r"^\s*except\s*:\s*$") +_EXCEPTION_EXCEPT_PATTERN = re.compile(r"^\s*except\s+Exception(?:\s+as\s+\w+)?\s*:\s*$") +_LOG_TOKENS = ("logger.", "logging.", "print(", "emit_failure_telemetry(", "telemetry.") +_TYPED_FAILURE_TOKENS = ("FailureState", "McpException", "ValueError", "RuntimeError", "InvalidOperationException") + + +def parse_unified_diff(diff_text: str) -> list[DiffFile]: + files: list[DiffFile] = [] + current_path: str | None = None + current_added: list[str] = [] + + for line in diff_text.splitlines(): + if line.startswith("diff --git "): + if current_path is not None: + files.append(DiffFile(path=current_path, added_lines=current_added)) + current_path = None + current_added = [] + continue + if line.startswith("+++ b/"): + current_path = line[6:] + continue + if current_path is None: + continue + if line.startswith("+") and not line.startswith("+++"): + current_added.append(line[1:]) + + if current_path is not None: + files.append(DiffFile(path=current_path, added_lines=current_added)) + + return files + + +def check_bare_except(file_path: str, added_lines: list[str]) -> list[Finding]: + findings: list[Finding] = [] + for index, line in enumerate(added_lines): + if not (_BARE_EXCEPT_PATTERN.match(line) or _EXCEPTION_EXCEPT_PATTERN.match(line)): + continue + window = added_lines[index + 1 : index + 5] + has_reraise = any("raise" in candidate for candidate in window) + has_logging = any(any(token in candidate for token in _LOG_TOKENS) for candidate in window) + has_typed_failure = any(any(token in candidate for token in _TYPED_FAILURE_TOKENS) for candidate in window) + if has_reraise and has_logging and has_typed_failure: + continue + findings.append( + Finding( + check="bare-except", + severity="blocking", + file=file_path, + line=index + 1, + detail="Broad exception handler added without a typed, logged re-raise path.", + ) + ) + return findings + + +def check_missing_telemetry(file_path: str, added_lines: list[str]) -> list[Finding]: + findings: list[Finding] = [] + for index, line in enumerate(added_lines): + if "except" not in line: + continue + window = added_lines[index + 1 : index + 5] + has_logging = any(any(token in candidate for token in _LOG_TOKENS) for candidate in window) + if has_logging: + continue + findings.append( + Finding( + check="missing-telemetry", + severity="advisory", + file=file_path, + line=index + 1, + detail="Exception path does not emit structured telemetry or logging.", + ) + ) + return findings + + +def check_file_size_limit(file_path: str, added_line_count: int, limit: int = 400) -> list[Finding]: + if added_line_count <= limit: + return [] + return [ + Finding( + check="file-size-limit", + severity="advisory", + file=file_path, + line=None, + detail=f"Added line count {added_line_count} exceeds limit {limit}.", + ) + ] + + +def run_critic(diff_text: str, *, llm_pass: bool = False, settings: Settings | None = None) -> CriticReport: + if llm_pass: + raise NotImplementedError("LLM critic pass is reserved for a future phase.") + + try: + _ = settings + files = parse_unified_diff(diff_text) + if not files: + raise ValueError("No unified diff content was found.") + + findings: list[Finding] = [] + for diff_file in files: + findings.extend(check_bare_except(diff_file.path, diff_file.added_lines)) + findings.extend(check_missing_telemetry(diff_file.path, diff_file.added_lines)) + findings.extend(check_file_size_limit(diff_file.path, len(diff_file.added_lines))) + + blocking_count = sum(1 for finding in findings if finding.severity == "blocking") + advisory_count = sum(1 for finding in findings if finding.severity == "advisory") + return CriticReport( + findings=findings, + blocking_count=blocking_count, + advisory_count=advisory_count, + exit_code=1 if blocking_count > 0 else 0, + ) + except Exception as exc: # noqa: BLE001 + print(f"critic error: {exc}", file=sys.stderr) + finding = Finding( + check="critic-error", + severity="advisory", + file="", + line=None, + detail=str(exc), + ) + return CriticReport(findings=[finding], blocking_count=0, advisory_count=1, exit_code=0) diff --git a/maf_starter/critic_cli.py b/maf_starter/critic_cli.py new file mode 100644 index 0000000..b984d81 --- /dev/null +++ b/maf_starter/critic_cli.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import argparse +from pathlib import Path +import sys + +from maf_starter.critic import run_critic + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="critic", description="Run the Resilience First peer critic against a unified diff") + parser.add_argument("--diff", required=True, help="Diff file path or '-' to read from stdin") + parser.add_argument( + "--severity-gate", + choices=("blocking", "advisory"), + default="blocking", + help="Exit non-zero on blocking findings only, or on any finding in advisory mode", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + try: + if args.diff == "-": + diff_text = sys.stdin.read() + else: + diff_text = Path(args.diff).read_text(encoding="utf-8", errors="replace") + except (FileNotFoundError, OSError) as exc: + print(f"critic: unable to read diff: {exc}", file=sys.stderr) + return 2 + + report = run_critic(diff_text) + for finding in report.findings: + print(f"{finding.file}:{finding.line or '-'} [{finding.severity}] {finding.check} - {finding.detail}") + print(f"critic: {report.blocking_count} blocking, {report.advisory_count} advisory") + + if args.severity_gate == "advisory": + return 1 if (report.blocking_count + report.advisory_count) > 0 else 0 + return report.exit_code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_critic.py b/tests/test_critic.py new file mode 100644 index 0000000..0860e8d --- /dev/null +++ b/tests/test_critic.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import io +from pathlib import Path +import textwrap + +import pytest + +from maf_starter.critic import ( + check_bare_except, + check_file_size_limit, + check_missing_telemetry, + parse_unified_diff, + run_critic, +) +from maf_starter.critic_cli import build_parser, main + + +DIRTY_DIFF = textwrap.dedent( + """\ + diff --git a/app.py b/app.py + +++ b/app.py + @@ + +try: + + do_work() + +except Exception: + + pass + """ +) + +DIRTY_DIFF_WITH_TELEMETRY_GAP = textwrap.dedent( + """\ + diff --git a/app.py b/app.py + +++ b/app.py + @@ + +try: + + do_work() + +except ValueError: + + recover() + """ +) + +CLEAN_DIFF = textwrap.dedent( + """\ + diff --git a/app.py b/app.py + +++ b/app.py + @@ + +try: + + do_work() + +except Exception as exc: + + logger.error("failed", exc_info=exc) + + raise RuntimeError("typed failure") from exc + """ +) + + +def test_parse_unified_diff_returns_changed_files_and_added_lines() -> None: + diff_text = textwrap.dedent( + """\ + diff --git a/app.py b/app.py + +++ b/app.py + @@ + +first + diff --git a/worker.py b/worker.py + +++ b/worker.py + @@ + +second + """ + ) + + files = parse_unified_diff(diff_text) + + assert len(files) == 2 + assert files[0].path == "app.py" + assert files[0].added_lines == ["first"] + assert files[1].path == "worker.py" + assert files[1].added_lines == ["second"] + + +def test_check_bare_except_blocks_untyped_unlogged_handler() -> None: + findings = check_bare_except("app.py", ["except Exception:", " pass"]) + + assert len(findings) == 1 + assert findings[0].check == "bare-except" + assert findings[0].severity == "blocking" + + +def test_check_bare_except_allows_logged_typed_reraise() -> None: + findings = check_bare_except( + "app.py", + [ + "except Exception as exc:", + ' logger.error("boom", exc_info=exc)', + ' raise RuntimeError("typed failure") from exc', + ], + ) + + assert findings == [] + + +def test_check_missing_telemetry_returns_advisory_only() -> None: + findings = check_missing_telemetry("app.py", ["except ValueError:", " recover()"]) + + assert len(findings) == 1 + assert findings[0].check == "missing-telemetry" + assert findings[0].severity == "advisory" + + +def test_check_file_size_limit_flags_large_files() -> None: + findings = check_file_size_limit("big.py", 401, limit=400) + + assert len(findings) == 1 + assert findings[0].check == "file-size-limit" + assert findings[0].severity == "advisory" + + +def test_run_critic_blocks_dirty_diff() -> None: + report = run_critic(DIRTY_DIFF) + + assert report.exit_code == 1 + assert report.blocking_count == 1 + assert any(finding.check == "bare-except" for finding in report.findings) + + +def test_run_critic_passes_clean_diff() -> None: + report = run_critic(CLEAN_DIFF) + + assert report.exit_code == 0 + assert report.blocking_count == 0 + assert report.advisory_count == 0 + + +def test_run_critic_returns_advisory_error_on_malformed_input() -> None: + report = run_critic("not a diff at all") + + assert report.exit_code == 0 + assert report.blocking_count == 0 + assert report.advisory_count == 1 + assert report.findings[0].check == "critic-error" + + +def test_build_parser_defaults_to_blocking_gate() -> None: + args = build_parser().parse_args(["--diff", "-"]) + + assert args.diff == "-" + assert args.severity_gate == "blocking" + + +def test_main_returns_zero_for_clean_diff_file(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + diff_path = tmp_path / "clean.diff" + diff_path.write_text(CLEAN_DIFF, encoding="utf-8") + + exit_code = main(["--diff", str(diff_path)]) + + out = capsys.readouterr().out + assert exit_code == 0 + assert "critic: 0 blocking, 0 advisory" in out + + +def test_main_returns_one_for_blocking_diff_file(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + diff_path = tmp_path / "dirty.diff" + diff_path.write_text(DIRTY_DIFF, encoding="utf-8") + + exit_code = main(["--diff", str(diff_path)]) + + out = capsys.readouterr().out + assert exit_code == 1 + assert "[blocking] bare-except" in out + + +def test_main_returns_two_for_missing_diff_path(capsys: pytest.CaptureFixture[str]) -> None: + exit_code = main(["--diff", "missing.diff"]) + + err = capsys.readouterr().err + assert exit_code == 2 + assert "unable to read diff" in err + + +def test_main_reads_diff_from_stdin_and_supports_advisory_gate( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setattr("sys.stdin", io.StringIO(DIRTY_DIFF_WITH_TELEMETRY_GAP)) + + exit_code = main(["--diff", "-", "--severity-gate", "advisory"]) + + out = capsys.readouterr().out + assert exit_code == 1 + assert "critic: 0 blocking, 1 advisory" in out