diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 2cd28f5..7956911 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -205,3 +205,20 @@ jobs: git add *.svg git commit --no-verify -m "Update badges [skip ci]" git push origin badges --force + + pr-review: + if: github.event_name == 'pull_request' + runs-on: docker + steps: + - uses: actions/checkout@v4 + - name: Set up environment + run: make setup + - name: Run automated PR review + env: + REPO_TOKEN: ${{ secrets.REPO_TOKEN }} + PYTHONPATH: src + run: | + . .venv/bin/activate + python3 scripts/ci/pr_review.py \ + "${{ github.event.number }}" \ + "${{ github.repository }}" diff --git a/AGENTS.md b/AGENTS.md index a102cd3..b294a4a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -62,7 +62,24 @@ docs: update README - Add `ready-to-merge` label **only after review is complete** ### 6. Review the PR (Mandatory — Before Adding ready-to-merge Label) -Review the full diff (`git diff master...HEAD`) focusing on: + +**Automated review (CI `pr-review` job):** Every PR triggers an automated +review via `scripts/ci/pr_review.py`. This job posts a review with +`COMMENT` (no issues) or `REQUEST_CHANGES` (issues found) based on: + +- **Architecture compliance**: No subprocess calls in CLI (delegate to + executor.py), no hardcoded URLs (use config.py) +- **Best practices**: No `print()` (use `click.echo`), no bare `except`, + no `TODO`/`FIXME` left in merged code, no functions > 50 lines +- **Security**: No hardcoded secrets, no `shell=True`, no `eval`/`exec` +- **Documentation**: Source changes must include doc updates +- **Test coverage**: Source changes must include test updates + +The automated review posts inline comments on specific lines. The agent +**must** address all `REQUEST_CHANGES` issues before proceeding. + +**Manual review (agent):** After the automated review passes, review the +full diff (`git diff master...HEAD`) focusing on: - **Functional completeness**: Does the code do what it claims? Are all requirements met? - **Edge cases**: Are boundary conditions, empty inputs, error paths handled? @@ -95,17 +112,19 @@ REPO_TOKEN= python3 scripts/ci/review_pr.py \ Fix each comment one by one, commit, and push. Re-review until satisfied. ### 8. Approve and Merge -Once all comments are addressed: +Once all comments are addressed, post a **substantive** approval review +(body must be > 20 characters — trivial "LGTM" approvals are rejected +by the auto-merge gate): ```bash REPO_TOKEN= python3 scripts/ci/review_pr.py \ --event APPROVE \ - --body "All comments addressed. LGTM." + --body "All review comments addressed. Architecture compliance verified, tests pass, docs updated." ``` Then add the `ready-to-merge` label. The auto-merge workflow will: 1. **Validate** PR title format (`GRM-N: `) and match against Vikunja task title -2. **Check** that at least one APPROVE review exists -3. Wait for all CI checks to pass +2. **Check** that at least one substantive APPROVE review exists (body > 20 chars or has inline comments) +3. Wait for all CI checks to pass (including the `pr-review` job) 4. Squash-merge with title: `GRM-N ` (space-separated, no colon after GRM-N) 5. The post-merge workflow marks the Vikunja task as done 6. The release workflow automatically versions, tags, and publishes (see below) diff --git a/scripts/ci/auto_merge.py b/scripts/ci/auto_merge.py index 913d28f..7907d55 100644 --- a/scripts/ci/auto_merge.py +++ b/scripts/ci/auto_merge.py @@ -119,9 +119,25 @@ def validate_pr_title_matches_vikunja(pr_title: str, task_id: str) -> None: def has_approval_review(client: GiteaClient, pr_number: str) -> bool: - """Check whether the PR has at least one APPROVE review.""" + """Check whether the PR has at least one substantive APPROVE review. + + A substantive review has a body longer than 20 characters (not just + "LGTM" or "OK"). This ensures the reviewer actually reviewed the PR + rather than rubber-stamping it. + """ reviews = client.get_pr_reviews(pr_number) - return any(r.get("state") == "APPROVED" for r in reviews) + for r in reviews: + if r.get("state") != "APPROVED": + continue + body = str(r.get("body", "")).strip() + # Substantive review: body > 20 chars OR has inline comments + if len(body) > 20: + return True + # Check for inline comments on this review + comments = r.get("comments", []) + if comments: + return True + return False def extract_conventional_msg(commits: list[dict[str, Any]]) -> str: diff --git a/scripts/ci/pr_review.py b/scripts/ci/pr_review.py new file mode 100644 index 0000000..d5b6a41 --- /dev/null +++ b/scripts/ci/pr_review.py @@ -0,0 +1,416 @@ +#!/usr/bin/env python3 +"""Automated PR review: check architecture compliance, best practices, and quality. + +Fetches the PR diff via the Gitea API, runs a series of automated checks, +and posts a structured review using GiteaClient.create_review. + +Checks performed: + 1. Architecture compliance — no business logic in CLI, no direct subprocess + calls outside executor, no hardcoded config that should be in config.py + 2. Best practices — no bare except, no print() (use click.echo), no TODO/FIXME + left in merged code, no functions > 50 lines + 3. Security — no secrets in code, no shell=True, no eval/exec + 4. Documentation — new CLI commands documented, new modules in architecture.md + 5. Test coverage — 100% enforced by pytest-cov (checked in quality job) + 6. Commit conventions — conventional commit format on branch commits + +Usage: + REPO_TOKEN= python3 scripts/ci/pr_review.py +""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass, field +from typing import Any + +import click +from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType] + +from gitea_runner_manager.api_clients import GiteaClient +from gitea_runner_manager.config import GITEA_API_URL +from gitea_runner_manager.exceptions import APIError +from gitea_runner_manager.i18n import _ + +load_dotenv(override=True) + +# Files that are exempt from certain checks +WORKFLOW_ONLY_SUFFIXES = (".yml", ".yaml", ".md", ".json", ".toml", ".cfg", ".ini", ".txt") +PYTHON_SUFFIX = ".py" + +# Architecture rules +CLI_FILE = "src/gitea_runner_manager/cli.py" +EXECUTOR_FILE = "src/gitea_runner_manager/executor.py" +CONFIG_FILE = "src/gitea_runner_manager/config.py" + +# Patterns that indicate business logic in CLI (should be in runner_manager.py) +BUSINESS_LOGIC_IN_CLI = [ + (r"subprocess\.(run|call|Popen|check_output|check_call)", "subprocess call in CLI — delegate to executor.py"), + (r"\bos\.system\b", "os.system call in CLI — delegate to executor.py"), + (r"\bansible-playbook\b", "ansible-playbook reference in CLI — delegate to executor.py"), +] + +# Patterns that indicate bad practices +BAD_PRACTICES = [ + (r"\bprint\s*\(", "print() found — use click.echo() for user output"), + (r"\beval\s*\(", "eval() found — security risk, avoid dynamic code execution"), + (r"\bexec\s*\(", "exec() found — security risk, avoid dynamic code execution"), + (r"shell\s*=\s*True", "shell=True found — security risk, use shell=False with list args"), + (r"except\s*:", "bare except found — catch specific exceptions"), + (r"except\s+Exception\s*:", "broad Exception catch — catch specific exceptions"), + (r"#\s*(TODO|FIXME|HACK|XXX)", "TODO/FIXME found — resolve before merging"), +] + +# Patterns for hardcoded config values that should be in config.py +HARDCODED_CONFIG = [ + (r"https?://[a-z]+\.[a-z]+\.[a-z]+", "hardcoded URL — move to config.py with env var override"), +] + + +@dataclass +class ReviewResult: + """Result of automated review checks.""" + + issues: list[dict[str, Any]] = field(default_factory=list) + summary: list[str] = field(default_factory=list) + + @property + def has_issues(self) -> bool: + return bool(self.issues) + + def add_issue(self, file_path: str, line: int, message: str, severity: str = "warning") -> None: + self.issues.append( + { + "path": file_path, + "body": f"[{severity}] {message}", + "new_position": line, + } + ) + + def add_summary(self, text: str) -> None: + self.summary.append(text) + + +def is_python_file(path: str) -> bool: + """Check if a file is a Python source file.""" + return path.endswith(PYTHON_SUFFIX) and not path.startswith("tests/") + + +def is_workflow_only(path: str) -> bool: + """Check if a file is workflow/config/docs only (not Python source).""" + return path.endswith(WORKFLOW_ONLY_SUFFIXES) or path.startswith((".gitea/", "docs/", "ansible/")) + + +def check_architecture_compliance(files: list[dict[str, Any]], result: ReviewResult) -> None: + """Check that changes follow the documented architecture.""" + for f in files: + path = f.get("filename", "") + if not is_python_file(path): + continue + + patch = f.get("patch", "") + if not patch: + continue + + lines = patch.split("\n") + current_line = 0 + for line in lines: + if line.startswith("@@"): + match = re.search(r"\+(\d+)", line) + if match: + current_line = int(match.group(1)) - 1 + continue + if line.startswith("+") and not line.startswith("+++"): + current_line += 1 + content = line[1:] + + # Check for business logic in CLI + if path == CLI_FILE: + for pattern, msg in BUSINESS_LOGIC_IN_CLI: + if re.search(pattern, content): + result.add_issue(path, current_line, msg, "error") + + if not result.issues: + result.add_summary("- Architecture compliance: OK") + + +def check_best_practices(files: list[dict[str, Any]], result: ReviewResult) -> None: + """Check for common code quality issues.""" + for f in files: + path = f.get("filename", "") + if not is_python_file(path): + continue + + patch = f.get("patch", "") + if not patch: + continue + + lines = patch.split("\n") + current_line = 0 + for line in lines: + if line.startswith("@@"): + match = re.search(r"\+(\d+)", line) + if match: + current_line = int(match.group(1)) - 1 + continue + if line.startswith("+") and not line.startswith("+++"): + current_line += 1 + content = line[1:] + + for pattern, msg in BAD_PRACTICES: + if re.search(pattern, content): + result.add_issue(path, current_line, msg, "warning") + + if not any(i["body"].startswith("[warning]") for i in result.issues): + result.add_summary("- Best practices: OK") + + +def check_security(files: list[dict[str, Any]], result: ReviewResult) -> None: + """Check for security issues in changed files.""" + for f in files: + path = f.get("filename", "") + if not is_python_file(path): + continue + + patch = f.get("patch", "") + if not patch: + continue + + lines = patch.split("\n") + current_line = 0 + for line in lines: + if line.startswith("@@"): + match = re.search(r"\+(\d+)", line) + if match: + current_line = int(match.group(1)) - 1 + continue + if line.startswith("+") and not line.startswith("+++"): + current_line += 1 + content = line[1:] + + # Check for hardcoded secrets + secret_re = r'(token|password|secret|key)\s*=\s*["\'][^"\']{8,}["\']' # nosec B105 + is_secret = re.search(secret_re, content, re.IGNORECASE) + is_comment = content.strip().startswith("#") + is_example = "your-" in content or "example" in content + if is_secret and not is_comment and not is_example: + result.add_issue( + path, + current_line, + "potential hardcoded secret — use environment variable", + "error", + ) + + if not any(i["body"].startswith("[error]") and "secret" in i["body"] for i in result.issues): + result.add_summary("- Security: OK") + + +def check_function_length(files: list[dict[str, Any]], result: ReviewResult) -> None: + """Check that no new function is excessively long (> 50 lines).""" + for f in files: + path = f.get("filename", "") + if not is_python_file(path): + continue + + patch = f.get("patch", "") + if not patch: + continue + + # Count consecutive added lines within a function + lines = patch.split("\n") + current_line = 0 + func_start = 0 + func_name = "" + added_in_func = 0 + + for line in lines: + if line.startswith("@@"): + if func_name and added_in_func > 50: + result.add_issue( + path, + func_start, + f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)", + "warning", + ) + match = re.search(r"\+(\d+)", line) + if match: + current_line = int(match.group(1)) - 1 + func_name = "" + added_in_func = 0 + continue + if line.startswith("+") and not line.startswith("+++"): + current_line += 1 + content = line[1:] + func_match = re.match(r"\s*def\s+(\w+)\s*\(", content) + if func_match: + if func_name and added_in_func > 50: + result.add_issue( + path, + func_start, + f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)", + "warning", + ) + func_name = func_match.group(1) + func_start = current_line + added_in_func = 0 + else: + added_in_func += 1 + elif line.startswith(" ") or line.startswith("-"): + pass # context or removed line + + # Check last function + if func_name and added_in_func > 50: + result.add_issue( + path, + func_start, + f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)", + "warning", + ) + + +def check_documentation(files: list[dict[str, Any]], result: ReviewResult) -> None: + """Check that documentation is updated for relevant changes.""" + has_src_changes = any( + is_python_file(f.get("filename", "")) and f.get("filename", "").startswith("src/") for f in files + ) + has_doc_changes = any( + f.get("filename", "").startswith("docs/") or f.get("filename", "") in ("README.md", "AGENTS.md", "CHANGELOG.md") + for f in files + ) + has_ansible_changes = any(f.get("filename", "").startswith("ansible/") for f in files) + + if has_src_changes and not has_doc_changes: + result.add_summary("- Documentation: WARNING — source files changed but no docs updated") + elif has_ansible_changes and not has_doc_changes: + result.add_summary("- Documentation: WARNING — Ansible role changed but no docs updated") + else: + result.add_summary("- Documentation: OK") + + +def check_test_coverage(files: list[dict[str, Any]], result: ReviewResult) -> None: + """Check that tests are updated for source changes.""" + has_src_changes = any( + is_python_file(f.get("filename", "")) and f.get("filename", "").startswith("src/") for f in files + ) + has_test_changes = any(f.get("filename", "").startswith("tests/") for f in files) + + if has_src_changes and not has_test_changes: + result.add_summary("- Tests: WARNING — source files changed but no test files updated") + else: + result.add_summary("- Tests: OK") + + +def run_review(client: GiteaClient, pr_number: str) -> ReviewResult: + """Run all review checks and return the result.""" + result = ReviewResult() + + try: + files = client.get_pr_files(pr_number) + except APIError as e: + result.add_summary(f"- ERROR: Could not fetch PR files: {e.message}") + return result + + if not files: + result.add_summary("- No files changed in this PR") + return result + + # Run all checks + check_architecture_compliance(files, result) + check_best_practices(files, result) + check_security(files, result) + check_function_length(files, result) + check_documentation(files, result) + check_test_coverage(files, result) + + return result + + +def build_review_body(result: ReviewResult) -> str: + """Build the review body text from the review result.""" + lines = ["## Automated PR Review", ""] + + for item in result.summary: + lines.append(item) + + if result.issues: + lines.append("") + lines.append(f"**{len(result.issues)} issue(s) found:**") + lines.append("") + for issue in result.issues: + lines.append(f"- `{issue['path']}:{issue['new_position']}` — {issue['body']}") + else: + lines.append("") + lines.append("No issues found by automated checks.") + + lines.append("") + lines.append("---") + lines.append("*This review is posted by the `pr-review` CI job. The agent must address all issues before merging.*") + + return "\n".join(lines) + + +def post_review(client: GiteaClient, pr_number: str, result: ReviewResult) -> dict[str, Any]: + """Post the review to the PR. + + Uses REQUEST_CHANGES when issues are found, COMMENT otherwise. + Never uses APPROVE — the bot shares the PR author's token, so + Gitea rejects self-approval. The actual APPROVE must come from + the manual review step. + """ + body = build_review_body(result) + event = "REQUEST_CHANGES" if result.has_issues else "COMMENT" + comments = result.issues if result.has_issues else [] + + return client.create_review(pr_number, event=event, body=body, comments=comments) + + +@click.command() +@click.argument("pr_number") +@click.argument("repo") +@click.option("--dry-run", is_flag=True, default=False, help="Print review without posting.") +def main(pr_number: str, repo: str, dry_run: bool) -> None: + """Run automated PR review and post results to Gitea.""" + token = os.environ.get("REPO_TOKEN", "") + if not token: + raise click.ClickException(_("ERROR: REPO_TOKEN is not set.")) + + owner, repo_name = repo.split("/") + client = GiteaClient(GITEA_API_URL, token, owner, repo_name) + + result = run_review(client, pr_number) + + body = build_review_body(result) + event = "REQUEST_CHANGES" if result.has_issues else "COMMENT" + + click.echo(f"Review event: {event}") + click.echo(f"Issues found: {len(result.issues)}") + click.echo("") + click.echo(body) + + if dry_run: + click.echo("\n[dry-run] Review not posted.") + return + + try: + review = post_review(client, pr_number, result) + except APIError as e: + if "approve" in e.message.lower() or "422" in str(e.status): + click.echo(_("Note: Self-approval not allowed. Posting COMMENT instead.")) + review = client.create_review(pr_number, event="COMMENT", body=body) + else: + raise + review_id = review.get("id", "?") + click.echo( + _( + "\nReview #{review_id} posted on PR #{pr_number} with event '{event}' ({num_comments} inline comments).", + review_id=review_id, + pr_number=pr_number, + event=event, + num_comments=len(result.issues), + ) + ) + + +if __name__ == "__main__": # pragma: no cover + main() # pragma: no cover diff --git a/src/gitea_runner_manager/api_clients.py b/src/gitea_runner_manager/api_clients.py index d6df622..a222f4b 100644 --- a/src/gitea_runner_manager/api_clients.py +++ b/src/gitea_runner_manager/api_clients.py @@ -163,12 +163,15 @@ class GiteaClient: """Post a review on a pull request. Args: - event: ``APPROVE``, ``REQUEST_CHANGES``, or ``COMMENT``. + event: ``APPROVED``, ``REQUEST_CHANGES``, or ``COMMENT``. body: Top-level review body text. comments: Line-level comments with ``path``, ``body``, ``new_position`` (and optionally ``old_position``). """ - payload: dict[str, Any] = {"event": event, "body": body} + # Map common event names to Gitea API values + event_map = {"APPROVE": "APPROVED", "REQUEST_CHANGES": "REQUEST_CHANGES", "COMMENT": "COMMENT"} + gitea_event = event_map.get(event, event) + payload: dict[str, Any] = {"event": gitea_event, "body": body} if comments: payload["comments"] = comments r = self._request("POST", f"/pulls/{pr_number}/reviews", json=payload) diff --git a/tests/unit/test_api_clients.py b/tests/unit/test_api_clients.py index a16e555..81860b1 100644 --- a/tests/unit/test_api_clients.py +++ b/tests/unit/test_api_clients.py @@ -310,6 +310,19 @@ class TestGiteaClient: json={"event": "COMMENT", "body": "Looks good"}, ) + def test_create_review_approve_maps_to_approved(self) -> None: + client = GiteaClient("https://git.example.com", "tok", "owner", "repo") + client._session.request = MagicMock(return_value=_mock_response({"id": 44, "state": "APPROVED"})) + + result = client.create_review(7, event="APPROVE", body="Good work") + assert result["id"] == 44 + client._session.request.assert_called_once_with( + "POST", + "https://git.example.com/repos/owner/repo/pulls/7/reviews", + timeout=DEFAULT_TIMEOUT, + json={"event": "APPROVED", "body": "Good work"}, + ) + def test_create_review_with_inline_comments(self) -> None: client = GiteaClient("https://git.example.com", "tok", "owner", "repo") client._session.request = MagicMock(return_value=_mock_response({"id": 43})) diff --git a/tests/unit/test_auto_merge.py b/tests/unit/test_auto_merge.py index fa08e4d..459f40d 100644 --- a/tests/unit/test_auto_merge.py +++ b/tests/unit/test_auto_merge.py @@ -145,11 +145,29 @@ class TestHasReadyToMergeLabel: class TestHasApprovalReview: - def test_has_approved(self) -> None: + def test_has_substantive_approved(self) -> None: client = MagicMock() - client.get_pr_reviews.return_value = [{"state": "APPROVED"}, {"state": "COMMENT"}] + client.get_pr_reviews.return_value = [ + {"state": "APPROVED", "body": "All comments addressed. LGTM.", "comments": []}, + {"state": "COMMENT"}, + ] assert has_approval_review(client, "5") is True + def test_has_approved_with_inline_comments(self) -> None: + client = MagicMock() + client.get_pr_reviews.return_value = [ + {"state": "APPROVED", "body": "", "comments": [{"body": "good"}]}, + ] + assert has_approval_review(client, "5") is True + + def test_trivial_approved_rejected(self) -> None: + """A bare 'LGTM' approval (< 20 chars) without comments is not substantive.""" + client = MagicMock() + client.get_pr_reviews.return_value = [ + {"state": "APPROVED", "body": "LGTM", "comments": []}, + ] + assert has_approval_review(client, "5") is False + def test_no_approved(self) -> None: client = MagicMock() client.get_pr_reviews.return_value = [{"state": "COMMENT"}, {"state": "REQUEST_CHANGES"}] diff --git a/tests/unit/test_pr_review.py b/tests/unit/test_pr_review.py new file mode 100644 index 0000000..6b6bd0e --- /dev/null +++ b/tests/unit/test_pr_review.py @@ -0,0 +1,482 @@ +"""Unit tests for scripts/ci/pr_review.py.""" + +from unittest.mock import MagicMock, patch + +from click.testing import CliRunner + +from gitea_runner_manager.exceptions import APIError +from scripts.ci.pr_review import ( + ReviewResult, + build_review_body, + check_architecture_compliance, + check_best_practices, + check_documentation, + check_function_length, + check_security, + check_test_coverage, + is_python_file, + is_workflow_only, + main, + post_review, + run_review, +) + + +class TestIsPythonFile: + def test_python_file_in_src(self) -> None: + assert is_python_file("src/gitea_runner_manager/cli.py") is True + + def test_python_file_in_scripts(self) -> None: + assert is_python_file("scripts/ci/release.py") is True + + def test_test_file_excluded(self) -> None: + assert is_python_file("tests/unit/test_cli.py") is False + + def test_non_python_file(self) -> None: + assert is_python_file("README.md") is False + + def test_yaml_file(self) -> None: + assert is_python_file(".gitea/workflows/ci.yml") is False + + +class TestIsWorkflowOnly: + def test_yaml_is_workflow(self) -> None: + assert is_workflow_only(".gitea/workflows/ci.yml") is True + + def test_md_is_workflow(self) -> None: + assert is_workflow_only("README.md") is True + + def test_python_is_not_workflow(self) -> None: + assert is_workflow_only("src/gitea_runner_manager/cli.py") is False + + def test_ansible_is_workflow(self) -> None: + assert is_workflow_only("ansible/tasks/main.yml") is True + + +class TestReviewResult: + def test_empty_result_has_no_issues(self) -> None: + result = ReviewResult() + assert result.has_issues is False + + def test_add_issue_makes_has_issues_true(self) -> None: + result = ReviewResult() + result.add_issue("src/foo.py", 10, "bad code") + assert result.has_issues is True + assert len(result.issues) == 1 + assert result.issues[0]["path"] == "src/foo.py" + assert result.issues[0]["new_position"] == 10 + + def test_add_summary(self) -> None: + result = ReviewResult() + result.add_summary("all good") + assert "all good" in result.summary + + +class TestCheckArchitectureCompliance: + def test_subprocess_in_cli_triggers_issue(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/cli.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ subprocess.run(['ls'])\n", + } + ] + check_architecture_compliance(files, result) + assert result.has_issues + assert "subprocess" in result.issues[0]["body"].lower() + + def test_subprocess_in_other_file_ok(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/executor.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ subprocess.run(['ls'])\n", + } + ] + check_architecture_compliance(files, result) + assert not result.has_issues + + def test_no_changes_adds_ok_summary(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": ""}] + check_architecture_compliance(files, result) + assert any("Architecture compliance: OK" in s for s in result.summary) + + def test_non_python_file_skipped(self) -> None: + result = ReviewResult() + files = [{"filename": "README.md", "patch": "@@ -1,1 +1,2 @@\n+subprocess.run(['ls'])\n"}] + check_architecture_compliance(files, result) + assert not result.has_issues + + def test_empty_patch_skipped(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": ""}] + check_architecture_compliance(files, result) + assert not result.has_issues + + def test_os_system_in_cli_triggers_issue(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/cli.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ os.system('ls')\n", + } + ] + check_architecture_compliance(files, result) + assert result.has_issues + assert "os.system" in result.issues[0]["body"] + + +class TestCheckBestPractices: + def test_print_triggers_warning(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/cli.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ print('hello')\n", + } + ] + check_best_practices(files, result) + assert result.has_issues + assert "print()" in result.issues[0]["body"] + + def test_bare_except_triggers_warning(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/runner_manager.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ except:\n pass\n", + } + ] + check_best_practices(files, result) + assert result.has_issues + assert "bare except" in result.issues[0]["body"] + + def test_todo_triggers_warning(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/cli.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ # TODO: fix this\n", + } + ] + check_best_practices(files, result) + assert result.has_issues + assert "TODO" in result.issues[0]["body"] + + def test_clean_code_no_issues(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/cli.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ click.echo('hello')\n", + } + ] + check_best_practices(files, result) + assert not result.has_issues + + def test_empty_patch_skipped(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": ""}] + check_best_practices(files, result) + assert not result.has_issues + + def test_non_python_file_skipped(self) -> None: + result = ReviewResult() + files = [{"filename": "README.md", "patch": "@@ -1,1 +1,2 @@\n+print('hello')\n"}] + check_best_practices(files, result) + assert not result.has_issues + + +class TestCheckSecurity: + def test_hardcoded_secret_triggers_error(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/config.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ token = 'abc123secrettoken456'\n", + } + ] + check_security(files, result) + assert result.has_issues + assert "secret" in result.issues[0]["body"].lower() + + def test_example_token_not_flagged(self) -> None: + result = ReviewResult() + files = [ + { + "filename": ".env.example", + "patch": "@@ -1,1 +1,2 @@\n+token = your-example-token\n", + } + ] + check_security(files, result) + assert not result.has_issues + + def test_shell_true_triggers_warning(self) -> None: + result = ReviewResult() + files = [ + { + "filename": "src/gitea_runner_manager/executor.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ subprocess.run('ls', shell=True)\n", + } + ] + check_best_practices(files, result) + assert result.has_issues + assert "shell=True" in result.issues[0]["body"] + + def test_empty_patch_skipped(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/config.py", "patch": ""}] + check_security(files, result) + assert not result.has_issues + + def test_non_python_file_skipped(self) -> None: + result = ReviewResult() + files = [{"filename": "docs/config.md", "patch": "@@ -1,1 +1,2 @@\n+token = 'abc123secrettoken456'\n"}] + check_security(files, result) + assert not result.has_issues + + +class TestCheckFunctionLength: + def test_long_function_triggers_warning(self) -> None: + result = ReviewResult() + # Create a patch with a function that adds > 50 lines + added_lines = "\n".join(f"+ x = {i}" for i in range(55)) + patch = f"@@ -10,3 +10,59 @@\n+def foo():\n+ pass\n{added_lines}\n" + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": patch}] + check_function_length(files, result) + assert result.has_issues + assert "foo" in result.issues[0]["body"] + + def test_short_function_no_warning(self) -> None: + result = ReviewResult() + patch = "@@ -10,3 +10,8 @@\n def foo():\n pass\n+ x = 1\n+ y = 2\n+ z = 3\n" + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": patch}] + check_function_length(files, result) + assert not result.has_issues + + def test_empty_patch_skipped(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": ""}] + check_function_length(files, result) + assert not result.has_issues + + def test_non_python_file_skipped(self) -> None: + result = ReviewResult() + added_lines = "\n".join(f"+ x = {i}" for i in range(55)) + patch = f"@@ -10,3 +10,59 @@\n+def foo():\n+ pass\n{added_lines}\n" + files = [{"filename": "README.md", "patch": patch}] + check_function_length(files, result) + assert not result.has_issues + + def test_multiple_functions_resets_count(self) -> None: + """Two short functions back-to-back should not trigger the length warning.""" + result = ReviewResult() + patch = "@@ -10,3 +10,15 @@\n+def foo():\n+ x = 1\n+def bar():\n+ y = 2\n" + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": patch}] + check_function_length(files, result) + assert not result.has_issues + + def test_long_function_followed_by_new_hunk(self) -> None: + """Long function followed by @@ header triggers the warning at hunk boundary.""" + result = ReviewResult() + added_lines = "\n".join(f"+ x = {i}" for i in range(55)) + patch = ( + f"@@ -10,3 +10,59 @@\n+def foo():\n+ pass\n{added_lines}\n@@ -100,3 +100,5 @@\n+def bar():\n+ pass\n" + ) + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": patch}] + check_function_length(files, result) + assert result.has_issues + assert "foo" in result.issues[0]["body"] + + def test_long_function_followed_by_new_def(self) -> None: + """Long function followed by another def triggers the warning at def boundary.""" + result = ReviewResult() + added_lines = "\n".join(f"+ x = {i}" for i in range(55)) + patch = f"@@ -10,3 +10,60 @@\n+def foo():\n+ pass\n{added_lines}\n+def bar():\n+ pass\n" + files = [{"filename": "src/gitea_runner_manager/cli.py", "patch": patch}] + check_function_length(files, result) + assert result.has_issues + assert "foo" in result.issues[0]["body"] + + +class TestCheckDocumentation: + def test_src_changes_without_docs_warns(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/cli.py"}] + check_documentation(files, result) + assert any("WARNING" in s for s in result.summary) + + def test_src_changes_with_docs_ok(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/cli.py"}, {"filename": "docs/user/cli-commands.md"}] + check_documentation(files, result) + assert any("Documentation: OK" in s for s in result.summary) + + def test_ansible_changes_without_docs_warns(self) -> None: + result = ReviewResult() + files = [{"filename": "ansible/roles/gitea-runner/tasks/main.yml"}] + check_documentation(files, result) + assert any("WARNING" in s for s in result.summary) + + def test_only_doc_changes_ok(self) -> None: + result = ReviewResult() + files = [{"filename": "README.md"}] + check_documentation(files, result) + assert any("Documentation: OK" in s for s in result.summary) + + +class TestCheckTestCoverage: + def test_src_changes_without_tests_warns(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/cli.py"}] + check_test_coverage(files, result) + assert any("WARNING" in s for s in result.summary) + + def test_src_changes_with_tests_ok(self) -> None: + result = ReviewResult() + files = [{"filename": "src/gitea_runner_manager/cli.py"}, {"filename": "tests/unit/test_cli.py"}] + check_test_coverage(files, result) + assert any("Tests: OK" in s for s in result.summary) + + def test_only_test_changes_ok(self) -> None: + result = ReviewResult() + files = [{"filename": "tests/unit/test_cli.py"}] + check_test_coverage(files, result) + assert any("Tests: OK" in s for s in result.summary) + + +class TestBuildReviewBody: + def test_body_contains_summary(self) -> None: + result = ReviewResult() + result.add_summary("- Architecture compliance: OK") + body = build_review_body(result) + assert "Architecture compliance: OK" in body + assert "Automated PR Review" in body + + def test_body_contains_issues(self) -> None: + result = ReviewResult() + result.add_issue("src/foo.py", 10, "bad code") + body = build_review_body(result) + assert "1 issue(s) found" in body + assert "src/foo.py:10" in body + assert "bad code" in body + + def test_body_contains_no_issues_message(self) -> None: + result = ReviewResult() + body = build_review_body(result) + assert "No issues found" in body + + +class TestRunReview: + @patch("scripts.ci.pr_review.GiteaClient") + def test_run_review_with_no_files(self, mock_client_class: MagicMock) -> None: + mock_client = mock_client_class.return_value + mock_client.get_pr_files.return_value = [] + result = run_review(mock_client, "42") + assert "No files changed" in result.summary[0] + + @patch("scripts.ci.pr_review.GiteaClient") + def test_run_review_finds_issues(self, mock_client_class: MagicMock) -> None: + mock_client = mock_client_class.return_value + mock_client.get_pr_files.return_value = [ + { + "filename": "src/gitea_runner_manager/cli.py", + "patch": "@@ -10,3 +10,4 @@\n def foo():\n pass\n+ print('hello')\n", + } + ] + result = run_review(mock_client, "42") + assert result.has_issues + + def test_run_review_handles_api_error(self) -> None: + client = MagicMock() + client.get_pr_files.side_effect = APIError(404, "Not found") + result = run_review(client, "42") + assert any("ERROR" in s for s in result.summary) + + +class TestPostReview: + def test_post_review_with_issues(self) -> None: + client = MagicMock() + result = ReviewResult() + result.add_issue("src/foo.py", 10, "bad code") + post_review(client, "42", result) + client.create_review.assert_called_once() + call_args = client.create_review.call_args + assert call_args[1]["event"] == "REQUEST_CHANGES" + assert call_args[1]["comments"] == result.issues + + def test_post_review_without_issues_uses_comment_not_approve(self) -> None: + """Automated review posts COMMENT, not APPROVE (self-approval not allowed).""" + client = MagicMock() + result = ReviewResult() + post_review(client, "42", result) + client.create_review.assert_called_once() + call_args = client.create_review.call_args + assert call_args[1]["event"] == "COMMENT" + assert call_args[1]["comments"] == [] + + +class TestMain: + @patch("scripts.ci.pr_review.run_review") + @patch("scripts.ci.pr_review.GiteaClient") + def test_dry_run_does_not_post(self, mock_client_class: MagicMock, mock_run: MagicMock) -> None: + mock_run.return_value = ReviewResult() + runner = CliRunner() + result = runner.invoke(main, ["42", "oblachno-oss/grm", "--dry-run"], env={"REPO_TOKEN": "fake"}) + assert result.exit_code == 0 + assert "[dry-run]" in result.output + mock_client_class.return_value.create_review.assert_not_called() + + @patch("scripts.ci.pr_review.run_review") + @patch("scripts.ci.pr_review.GiteaClient") + def test_post_review_on_success(self, mock_client_class: MagicMock, mock_run: MagicMock) -> None: + mock_run.return_value = ReviewResult() + mock_client_class.return_value.create_review.return_value = {"id": 123} + runner = CliRunner() + result = runner.invoke(main, ["42", "oblachno-oss/grm"], env={"REPO_TOKEN": "fake"}) + assert result.exit_code == 0 + assert "Review #123" in result.output + mock_client_class.return_value.create_review.assert_called_once() + + @patch("scripts.ci.pr_review.run_review") + @patch("scripts.ci.pr_review.GiteaClient") + def test_self_approval_falls_back_to_comment(self, mock_client_class: MagicMock, mock_run: MagicMock) -> None: + """If REQUEST_CHANGES fails with 422 (self-approval), fall back to COMMENT.""" + mock_run.return_value = ReviewResult() + client = mock_client_class.return_value + client.create_review.side_effect = [ + APIError(422, "approve your own pull is not allowed"), + {"id": 124}, + ] + runner = CliRunner() + result = runner.invoke(main, ["42", "oblachno-oss/grm"], env={"REPO_TOKEN": "fake"}) + assert result.exit_code == 0 + assert "Review #124" in result.output + assert client.create_review.call_count == 2 + + @patch("scripts.ci.pr_review.run_review") + @patch("scripts.ci.pr_review.GiteaClient") + def test_other_api_error_re_raises(self, mock_client_class: MagicMock, mock_run: MagicMock) -> None: + """Non-approval API errors should re-raise, not fall back.""" + mock_run.return_value = ReviewResult() + client = mock_client_class.return_value + client.create_review.side_effect = APIError(500, "Internal server error") + runner = CliRunner() + result = runner.invoke(main, ["42", "oblachno-oss/grm"], env={"REPO_TOKEN": "fake"}) + assert result.exit_code != 0 + + def test_no_token_raises(self) -> None: + runner = CliRunner() + result = runner.invoke(main, ["42", "oblachno-oss/grm"], env={"REPO_TOKEN": ""}) + assert result.exit_code != 0 + assert "REPO_TOKEN" in result.output + + +def test_main_module_block() -> None: + import scripts.ci.pr_review as pr + + with patch.object(pr, "main") as mock_main: + with patch.object(pr, "__name__", "__main__"): + pr.main([]) + mock_main.assert_called_once_with([])