diff --git a/.coverage b/.coverage new file mode 100644 index 00000000..dde2c7c7 Binary files /dev/null and b/.coverage differ diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 00000000..1b9444d6 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["scripts/ci"] + +[tool.coverage.run] +source = ["pr_review_merge_scheduler", "opencode_review_normalize_output"] +branch = false + +[tool.coverage.report] +fail_under = 100 +show_missing = true + +[tool.interrogate] +fail-under = 100 +exclude = ["tests", "setup.py"] +ignore-init-method = false +ignore-init-module = false +ignore-magic = false +ignore-semiprivate = false +ignore-private = false +ignore-property-decorators = false +ignore-module = false +ignore-nested-functions = false diff --git a/scripts/ci/__pycache__/opencode_review_normalize_output.cpython-312.pyc b/scripts/ci/__pycache__/opencode_review_normalize_output.cpython-312.pyc new file mode 100644 index 00000000..549470e2 Binary files /dev/null and b/scripts/ci/__pycache__/opencode_review_normalize_output.cpython-312.pyc differ diff --git a/scripts/ci/__pycache__/pr_review_merge_scheduler.cpython-312.pyc b/scripts/ci/__pycache__/pr_review_merge_scheduler.cpython-312.pyc new file mode 100644 index 00000000..3ff137c3 Binary files /dev/null and b/scripts/ci/__pycache__/pr_review_merge_scheduler.cpython-312.pyc differ diff --git a/scripts/ci/opencode_review_normalize_output.py b/scripts/ci/opencode_review_normalize_output.py index 32145f8f..c3be9821 100755 --- a/scripts/ci/opencode_review_normalize_output.py +++ b/scripts/ci/opencode_review_normalize_output.py @@ -274,5 +274,5 @@ def main(argv: list[str]) -> int: return 4 -if __name__ == "__main__": +if __name__ == "__main__": # pragma: no cover raise SystemExit(main(sys.argv)) diff --git a/scripts/ci/pr_review_merge_scheduler.py b/scripts/ci/pr_review_merge_scheduler.py index cf4805f0..5f64a850 100644 --- a/scripts/ci/pr_review_merge_scheduler.py +++ b/scripts/ci/pr_review_merge_scheduler.py @@ -1,4 +1,5 @@ #!/usr/bin/env python3 +"""Schedule OpenCode reviews and auto-merges for open pull requests.""" from __future__ import annotations import argparse @@ -69,12 +70,15 @@ @dataclass class Decision: + """Holds the scheduling decision for a single pull request.""" + pr: int action: str reason: str def run(args: list[str], *, stdin: str | None = None) -> str: + """Run a subprocess and return its stdout, raising RuntimeError on failure.""" process = subprocess.run(args, input=stdin, capture_output=True, text=True) if process.returncode != 0: raise RuntimeError( @@ -84,6 +88,7 @@ def run(args: list[str], *, stdin: str | None = None) -> str: def split_repo(repo: str) -> tuple[str, str]: + """Split an 'owner/name' repository string into a (owner, name) tuple.""" try: owner, name = repo.split("/", 1) except ValueError as exc: @@ -94,6 +99,7 @@ def split_repo(repo: str) -> tuple[str, str]: def gh_graphql(query: str, **fields: str | int) -> dict[str, Any]: + """Execute a GitHub GraphQL query via the gh CLI and return the parsed JSON response.""" cmd = ["gh", "api", "graphql", "-F", "query=@-"] for key, value in fields.items(): flag = "-F" if isinstance(value, int) else "-f" @@ -102,6 +108,7 @@ def gh_graphql(query: str, **fields: str | int) -> dict[str, Any]: def fetch_open_prs(repo: str, max_prs: int) -> list[dict[str, Any]]: + """Fetch up to max_prs open pull requests from the repository using pagination.""" owner, name = split_repo(repo) prs: list[dict[str, Any]] = [] cursor: str | None = None @@ -126,12 +133,14 @@ def fetch_open_prs(repo: str, max_prs: int) -> list[dict[str, Any]]: def context_nodes(pr: dict[str, Any]) -> list[dict[str, Any]]: + """Return the list of status-check context nodes from a PR's statusCheckRollup.""" rollup = pr.get("statusCheckRollup") or {} contexts = rollup.get("contexts") or {} return contexts.get("nodes") or [] def is_opencode_context(node: dict[str, Any]) -> bool: + """Return True if the status-check node belongs to the OpenCode review workflow.""" if node.get("__typename") == "CheckRun": workflow = ( ((node.get("checkSuite") or {}).get("workflowRun") or {}).get("workflow") @@ -142,6 +151,7 @@ def is_opencode_context(node: dict[str, Any]) -> bool: def opencode_in_progress(pr: dict[str, Any]) -> bool: + """Return True if an OpenCode review is currently queued or running for the PR.""" for node in context_nodes(pr): if not is_opencode_context(node): continue @@ -152,19 +162,23 @@ def opencode_in_progress(pr: dict[str, Any]) -> bool: def unresolved_thread_count(pr: dict[str, Any]) -> int: + """Return the number of non-outdated, unresolved review threads on the PR.""" threads = ((pr.get("reviewThreads") or {}).get("nodes") or []) return sum(1 for thread in threads if not thread.get("isResolved") and not thread.get("isOutdated")) def review_author_login(review: dict[str, Any]) -> str: + """Return the lowercased login of the review author, or empty string if absent.""" return ((review.get("author") or {}).get("login") or "").lower() def is_opencode_review(review: dict[str, Any]) -> bool: + """Return True if the review was submitted by the opencode-agent bot.""" return review_author_login(review) == "opencode-agent" def current_head_review_state(pr: dict[str, Any], state: str) -> bool: + """Return True if the latest opencode-agent review on the current head has the given state.""" head = pr.get("headRefOid") for review in reversed((pr.get("reviews") or {}).get("nodes") or []): if not is_opencode_review(review): @@ -178,14 +192,17 @@ def current_head_review_state(pr: dict[str, Any], state: str) -> bool: def has_current_head_approval(pr: dict[str, Any]) -> bool: + """Return True if the opencode-agent approved the current head commit.""" return current_head_review_state(pr, "APPROVED") def has_current_head_changes_requested(pr: dict[str, Any]) -> bool: + """Return True if the opencode-agent requested changes on the current head commit.""" return current_head_review_state(pr, "CHANGES_REQUESTED") def enable_auto_merge(repo: str, pr: dict[str, Any], *, dry_run: bool) -> None: + """Enable auto-merge for the PR via gh CLI, unless dry_run is True.""" number = str(pr["number"]) head = pr["headRefOid"] if dry_run: @@ -194,6 +211,7 @@ def enable_auto_merge(repo: str, pr: dict[str, Any], *, dry_run: bool) -> None: def dispatch_opencode_review(repo: str, workflow: str, pr: dict[str, Any], *, dry_run: bool) -> None: + """Trigger the OpenCode review workflow for the PR, unless dry_run is True.""" if dry_run: return run( @@ -230,6 +248,7 @@ def inspect_pr( workflow: str, base_branch: str, ) -> Decision: + """Determine what action to take for a single open pull request.""" number = pr["number"] head_repo = (pr.get("headRepository") or {}).get("nameWithOwner") base_ref = pr.get("baseRefName") @@ -273,6 +292,7 @@ def print_summary( base_branch: str, project_flow: str, ) -> None: + """Print a per-PR decision log and a JSON summary to stdout.""" counts: dict[str, int] = {} for decision in decisions: counts[decision.action] = counts.get(decision.action, 0) + 1 @@ -292,6 +312,7 @@ def print_summary( def self_test() -> None: + """Run built-in assertions to verify core logic is correct.""" sample = { "number": 1, "headRefOid": "abc", @@ -340,6 +361,7 @@ def self_test() -> None: def parse_args(argv: list[str]) -> argparse.Namespace: + """Parse command-line arguments and return the resulting Namespace.""" parser = argparse.ArgumentParser() parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY", "")) parser.add_argument("--base-branch", default=os.environ.get("DEFAULT_BRANCH", "")) @@ -354,6 +376,7 @@ def parse_args(argv: list[str]) -> argparse.Namespace: def main(argv: list[str]) -> int: + """Entry point: parse args, fetch PRs, inspect each, print summary.""" args = parse_args(argv) if args.self_test: self_test() @@ -386,7 +409,7 @@ def main(argv: list[str]) -> int: return 0 -if __name__ == "__main__": +if __name__ == "__main__": # pragma: no cover try: raise SystemExit(main(sys.argv[1:])) except RuntimeError as exc: diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/__pycache__/__init__.cpython-312.pyc b/tests/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 00000000..c98a0059 Binary files /dev/null and b/tests/__pycache__/__init__.cpython-312.pyc differ diff --git a/tests/__pycache__/test_opencode_review_normalize_output.cpython-312-pytest-9.1.1.pyc b/tests/__pycache__/test_opencode_review_normalize_output.cpython-312-pytest-9.1.1.pyc new file mode 100644 index 00000000..0012957c Binary files /dev/null and b/tests/__pycache__/test_opencode_review_normalize_output.cpython-312-pytest-9.1.1.pyc differ diff --git a/tests/__pycache__/test_pr_review_merge_scheduler.cpython-312-pytest-9.1.1.pyc b/tests/__pycache__/test_pr_review_merge_scheduler.cpython-312-pytest-9.1.1.pyc new file mode 100644 index 00000000..19e7e78a Binary files /dev/null and b/tests/__pycache__/test_pr_review_merge_scheduler.cpython-312-pytest-9.1.1.pyc differ diff --git a/tests/test_opencode_review_normalize_output.py b/tests/test_opencode_review_normalize_output.py new file mode 100644 index 00000000..d2242eb5 --- /dev/null +++ b/tests/test_opencode_review_normalize_output.py @@ -0,0 +1,458 @@ +"""Tests for scripts/ci/opencode_review_normalize_output.py.""" +from __future__ import annotations + +import json +import sys +from io import StringIO +from pathlib import Path + +import pytest + +from opencode_review_normalize_output import ( + admits_missing_structural_review, + check_structural_approval, + iter_json_objects, + main, + mentions_changed_file_evidence, + valid_control, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_VALID_KWARGS = { + "expected_head_sha": "sha1", + "expected_run_id": "run1", + "expected_run_attempt": "1", +} + + +def _valid_approve_value(**overrides: object) -> dict: + """Return a minimal valid APPROVE control block.""" + base: dict = { + "head_sha": "sha1", + "run_id": "run1", + "run_attempt": "1", + "result": "APPROVE", + "reason": "Reviewed scripts/ci/pr_review_merge_scheduler.py thoroughly.", + "summary": "All changes in scripts/ci/pr_review_merge_scheduler.py look correct.", + "findings": [], + } + base.update(overrides) + return base + + +def _valid_request_changes_value(**overrides: object) -> dict: + """Return a minimal valid REQUEST_CHANGES control block with one finding.""" + finding: dict = { + "path": "scripts/ci/pr_review_merge_scheduler.py", + "line": 10, + "severity": "HIGH", + "title": "Bug in function", + "problem": "The function does X incorrectly.", + "root_cause": "Missing validation.", + "fix_direction": "Add validation.", + "regression_test_direction": "Add a test for the edge case.", + "suggested_diff": "- old\n+ new", + } + base: dict = { + "head_sha": "sha1", + "run_id": "run1", + "run_attempt": "1", + "result": "REQUEST_CHANGES", + "reason": "Found issues in scripts/ci/pr_review_merge_scheduler.py.", + "summary": "See findings.", + "findings": [finding], + } + base.update(overrides) + return base + + +# --------------------------------------------------------------------------- +# admits_missing_structural_review() +# --------------------------------------------------------------------------- + + +def test_admits_missing_phrase(): + """Test admits_missing_structural_review() matches a known structural-failure phrase.""" + assert admits_missing_structural_review( + "structural exploration was not possible", "" + ) + + +def test_admits_missing_pattern(): + """Test admits_missing_structural_review() matches a structural-failure regex pattern.""" + assert admits_missing_structural_review( + "Could not inspect the changed files", "" + ) + + +def test_admits_missing_no_match(): + """Test admits_missing_structural_review() returns False for clean text.""" + assert not admits_missing_structural_review( + "Reviewed all changes in scripts/ci/foo.py and they look good.", "" + ) + + +def test_admits_missing_phrase_in_summary(): + """Test admits_missing_structural_review() also checks the summary argument.""" + assert admits_missing_structural_review("", "no changed files") + + +# --------------------------------------------------------------------------- +# mentions_changed_file_evidence() +# --------------------------------------------------------------------------- + + +def test_mentions_file_path(): + """Test mentions_changed_file_evidence() detects a directory/file path.""" + assert mentions_changed_file_evidence("Changed scripts/ci/main.py", "") + + +def test_mentions_extension_file(): + """Test mentions_changed_file_evidence() detects a bare filename with extension.""" + assert mentions_changed_file_evidence("", "Updated requirements.txt") + + +def test_mentions_no_file(): + """Test mentions_changed_file_evidence() returns False when no file path is present.""" + assert not mentions_changed_file_evidence("looks good overall", "no files here") + + +# --------------------------------------------------------------------------- +# check_structural_approval() +# --------------------------------------------------------------------------- + + +def test_check_structural_approval_file_not_found(tmp_path: Path) -> None: + """Test check_structural_approval() returns 65 when the file does not exist.""" + result = check_structural_approval(tmp_path / "nonexistent.json") + assert result == 65 + + +def test_check_structural_approval_invalid_json(tmp_path: Path) -> None: + """Test check_structural_approval() returns 65 on a JSON parse error.""" + bad = tmp_path / "bad.json" + bad.write_text("not json{{{") + result = check_structural_approval(bad) + assert result == 65 + + +def test_check_structural_approval_not_dict(tmp_path: Path) -> None: + """Test check_structural_approval() returns 4 when the JSON value is not a dict.""" + f = tmp_path / "control.json" + f.write_text('"just a string"') + assert check_structural_approval(f) == 4 + + +def test_check_structural_approval_approve_admits_missing(tmp_path: Path) -> None: + """Test check_structural_approval() returns 4 when APPROVE admits missing structure.""" + f = tmp_path / "control.json" + f.write_text( + json.dumps( + { + "result": "APPROVE", + "reason": "structural exploration was not possible", + "summary": "ok", + } + ) + ) + assert check_structural_approval(f) == 4 + + +def test_check_structural_approval_approve_no_file_evidence(tmp_path: Path) -> None: + """Test check_structural_approval() returns 4 when APPROVE lacks file evidence.""" + f = tmp_path / "control.json" + f.write_text(json.dumps({"result": "APPROVE", "reason": "looks good", "summary": "ok"})) + assert check_structural_approval(f) == 4 + + +def test_check_structural_approval_approve_valid(tmp_path: Path) -> None: + """Test check_structural_approval() returns 0 for a properly evidenced APPROVE.""" + f = tmp_path / "control.json" + f.write_text( + json.dumps( + { + "result": "APPROVE", + "reason": "Reviewed scripts/ci/pr_review_merge_scheduler.py.", + "summary": "All changes look correct.", + } + ) + ) + assert check_structural_approval(f) == 0 + + +def test_check_structural_approval_request_changes(tmp_path: Path) -> None: + """Test check_structural_approval() returns 0 for a REQUEST_CHANGES block.""" + f = tmp_path / "control.json" + f.write_text( + json.dumps({"result": "REQUEST_CHANGES", "reason": "issues found", "summary": "fix it"}) + ) + assert check_structural_approval(f) == 0 + + +# --------------------------------------------------------------------------- +# valid_control() +# --------------------------------------------------------------------------- + + +def test_valid_control_not_dict() -> None: + """Test valid_control() returns None for non-dict input.""" + assert valid_control("string", **_VALID_KWARGS) is None + + +def test_valid_control_wrong_head_sha() -> None: + """Test valid_control() returns None when head_sha does not match.""" + v = _valid_approve_value(head_sha="wrong") + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_wrong_run_id() -> None: + """Test valid_control() returns None when run_id does not match.""" + v = _valid_approve_value(run_id="wrong") + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_wrong_run_attempt() -> None: + """Test valid_control() returns None when run_attempt does not match.""" + v = _valid_approve_value(run_attempt="99") + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_invalid_result() -> None: + """Test valid_control() returns None for an unrecognized result value.""" + v = _valid_approve_value(result="MAYBE") + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_reason_not_str() -> None: + """Test valid_control() returns None when reason is not a string.""" + v = _valid_approve_value(reason=123) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_reason_empty() -> None: + """Test valid_control() returns None when reason is an empty/whitespace string.""" + v = _valid_approve_value(reason=" ") + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_summary_not_str() -> None: + """Test valid_control() returns None when summary is not a string.""" + v = _valid_approve_value(summary=None) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_summary_empty() -> None: + """Test valid_control() returns None when summary is an empty string.""" + v = _valid_approve_value(summary="") + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_findings_none_approve() -> None: + """Test valid_control() sets findings to [] for APPROVE when findings is None.""" + v = _valid_approve_value() + del v["findings"] + result = valid_control(v, **_VALID_KWARGS) + assert result is not None + assert result["findings"] == [] + + +def test_valid_control_findings_not_list_request_changes() -> None: + """Test valid_control() returns None when findings is not a list for REQUEST_CHANGES.""" + v = _valid_request_changes_value(findings="not a list") + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_approve_with_non_empty_findings() -> None: + """Test valid_control() returns None when APPROVE has non-empty findings.""" + finding = _valid_request_changes_value()["findings"][0] + v = _valid_approve_value(findings=[finding]) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_request_changes_empty_findings() -> None: + """Test valid_control() returns None when REQUEST_CHANGES has no findings.""" + v = _valid_request_changes_value(findings=[]) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_approve_admits_missing() -> None: + """Test valid_control() returns None for APPROVE that admits missing structure.""" + v = _valid_approve_value( + reason="structural exploration was not possible", + summary="ok", + ) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_approve_no_file_evidence() -> None: + """Test valid_control() returns None for APPROVE with no changed file evidence.""" + v = _valid_approve_value(reason="everything looks fine", summary="all good") + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_approve_valid() -> None: + """Test valid_control() returns a normalized dict for a valid APPROVE.""" + v = _valid_approve_value() + result = valid_control(v, **_VALID_KWARGS) + assert result is not None + assert result["result"] == "APPROVE" + assert result["findings"] == [] + + +def test_valid_control_request_changes_valid() -> None: + """Test valid_control() returns a normalized dict for a valid REQUEST_CHANGES.""" + v = _valid_request_changes_value() + result = valid_control(v, **_VALID_KWARGS) + assert result is not None + assert result["result"] == "REQUEST_CHANGES" + assert len(result["findings"]) == 1 + + +def test_valid_control_finding_not_dict() -> None: + """Test valid_control() returns None when a finding entry is not a dict.""" + v = _valid_request_changes_value(findings=["not-a-dict"]) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_finding_line_is_bool() -> None: + """Test valid_control() returns None when finding line is a boolean (not an int).""" + finding = dict(_valid_request_changes_value()["findings"][0]) + finding["line"] = True + v = _valid_request_changes_value(findings=[finding]) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_finding_line_not_int() -> None: + """Test valid_control() returns None when finding line is not an integer.""" + finding = dict(_valid_request_changes_value()["findings"][0]) + finding["line"] = "ten" + v = _valid_request_changes_value(findings=[finding]) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_finding_line_zero() -> None: + """Test valid_control() returns None when finding line is <= 0.""" + finding = dict(_valid_request_changes_value()["findings"][0]) + finding["line"] = 0 + v = _valid_request_changes_value(findings=[finding]) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_finding_missing_required_field() -> None: + """Test valid_control() returns None when a required finding field is absent.""" + finding = dict(_valid_request_changes_value()["findings"][0]) + del finding["title"] + v = _valid_request_changes_value(findings=[finding]) + assert valid_control(v, **_VALID_KWARGS) is None + + +def test_valid_control_finding_empty_required_field() -> None: + """Test valid_control() returns None when a required finding field is empty.""" + finding = dict(_valid_request_changes_value()["findings"][0]) + finding["title"] = " " + v = _valid_request_changes_value(findings=[finding]) + assert valid_control(v, **_VALID_KWARGS) is None + + +# --------------------------------------------------------------------------- +# iter_json_objects() +# --------------------------------------------------------------------------- + + +def test_iter_json_objects_plain_json() -> None: + """Test iter_json_objects() parses a plain JSON string.""" + result = iter_json_objects('{"key": "value"}') + # Should contain at least the parsed object + assert {"key": "value"} in result + + +def test_iter_json_objects_prose_with_json() -> None: + """Test iter_json_objects() finds embedded JSON in surrounding prose.""" + text = 'Some text here {"result": "APPROVE"} more text' + result = iter_json_objects(text) + assert any(isinstance(v, dict) and v.get("result") == "APPROVE" for v in result) + + +def test_iter_json_objects_invalid_json() -> None: + """Test iter_json_objects() handles text that is not valid JSON gracefully.""" + result = iter_json_objects("not json at all") + # No valid JSON objects found + assert result == [] + + +def test_iter_json_objects_invalid_brace() -> None: + """Test iter_json_objects() skips braces that start invalid JSON.""" + result = iter_json_objects("{invalid}") + assert result == [] + + +# --------------------------------------------------------------------------- +# main() +# --------------------------------------------------------------------------- + + +def test_main_wrong_arg_count(capsys: pytest.CaptureFixture) -> None: + """Test main() returns 64 and prints usage when called with wrong number of args.""" + result = main(["prog"]) + assert result == 64 + assert "usage" in capsys.readouterr().err.lower() + + +def test_main_check_structural_approval_delegates(tmp_path: Path) -> None: + """Test main() delegates to check_structural_approval for --check-structural-approval.""" + f = tmp_path / "ctrl.json" + f.write_text( + json.dumps( + { + "result": "APPROVE", + "reason": "Reviewed scripts/ci/foo.py.", + "summary": "ok", + } + ) + ) + result = main(["prog", "--check-structural-approval", str(f)]) + assert result == 0 + + +def test_main_file_not_found(tmp_path: Path, capsys: pytest.CaptureFixture) -> None: + """Test main() returns 65 when the output file cannot be read.""" + result = main(["prog", "sha1", "run1", "1", str(tmp_path / "missing.txt")]) + assert result == 65 + assert "cannot read" in capsys.readouterr().err + + +def test_main_no_valid_control(tmp_path: Path, capsys: pytest.CaptureFixture) -> None: + """Test main() returns 4 when no valid control block is found in the file.""" + f = tmp_path / "output.txt" + f.write_text("just some prose, no valid control block here") + result = main(["prog", "sha1", "run1", "1", str(f)]) + assert result == 4 + assert "NO_CONCLUSION" in capsys.readouterr().err + + +def test_main_valid_control_written(tmp_path: Path) -> None: + """Test main() writes the normalized control block and returns 0.""" + control = _valid_approve_value() + f = tmp_path / "output.txt" + f.write_text(json.dumps(control)) + result = main(["prog", "sha1", "run1", "1", str(f)]) + assert result == 0 + written = f.read_text(encoding="utf-8") + assert "opencode-review-control-v1" in written + assert "APPROVE" in written + + +def test_main_first_invalid_second_valid(tmp_path: Path) -> None: + """Test main() skips invalid control blocks and uses the first valid one.""" + invalid = {"head_sha": "sha1", "run_id": "run1", "run_attempt": "1", "result": "MAYBE"} + valid = _valid_approve_value() + f = tmp_path / "output.txt" + # Put invalid first so the `continue` branch is exercised, then valid second + f.write_text(json.dumps(invalid) + "\n" + json.dumps(valid)) + result = main(["prog", "sha1", "run1", "1", str(f)]) + assert result == 0 + written = f.read_text(encoding="utf-8") + assert "APPROVE" in written diff --git a/tests/test_pr_review_merge_scheduler.py b/tests/test_pr_review_merge_scheduler.py new file mode 100644 index 00000000..88008b42 --- /dev/null +++ b/tests/test_pr_review_merge_scheduler.py @@ -0,0 +1,867 @@ +"""Tests for scripts/ci/pr_review_merge_scheduler.py.""" +from __future__ import annotations + +import json +from unittest.mock import MagicMock, call, patch + +import pytest + +import pr_review_merge_scheduler as sched +from pr_review_merge_scheduler import ( + Decision, + context_nodes, + current_head_review_state, + dispatch_opencode_review, + enable_auto_merge, + fetch_open_prs, + gh_graphql, + has_current_head_approval, + has_current_head_changes_requested, + inspect_pr, + is_opencode_context, + is_opencode_review, + main, + opencode_in_progress, + parse_args, + print_summary, + review_author_login, + run, + self_test, + split_repo, + unresolved_thread_count, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_BASE_PR: dict = { + "number": 42, + "headRefOid": "abc123", + "isDraft": False, + "baseRefName": "main", + "headRefName": "feature", + "baseRefOid": "base123", + "headRepository": {"nameWithOwner": "owner/repo"}, + "autoMergeRequest": None, + "reviewThreads": {"nodes": []}, + "reviews": {"nodes": []}, + "statusCheckRollup": {"contexts": {"nodes": []}}, +} + + +def _make_pr(**overrides: object) -> dict: + """Return a copy of the base PR dict with the given overrides applied.""" + pr = dict(_BASE_PR) + pr.update(overrides) + return pr + + +def _approved_pr(head: str = "abc123") -> dict: + """Return a PR dict that has a current-head approval from opencode-agent.""" + return _make_pr( + headRefOid=head, + reviews={ + "nodes": [ + { + "state": "APPROVED", + "author": {"login": "opencode-agent"}, + "body": "", + "commit": {"oid": head}, + } + ] + }, + ) + + +# --------------------------------------------------------------------------- +# run() +# --------------------------------------------------------------------------- + + +def test_run_success(): + """Test that run() returns stdout on a successful command.""" + with patch("subprocess.run") as mock_sub: + mock_sub.return_value = MagicMock(returncode=0, stdout="hello\n", stderr="") + assert run(["echo", "hello"]) == "hello\n" + + +def test_run_failure(): + """Test that run() raises RuntimeError when the subprocess exits non-zero.""" + with patch("subprocess.run") as mock_sub: + mock_sub.return_value = MagicMock(returncode=1, stdout="", stderr="oops") + with pytest.raises(RuntimeError, match="Command failed"): + run(["false"]) + + +def test_run_passes_stdin(): + """Test that run() forwards the stdin argument to subprocess.""" + with patch("subprocess.run") as mock_sub: + mock_sub.return_value = MagicMock(returncode=0, stdout="", stderr="") + run(["cat"], stdin="data") + mock_sub.assert_called_once() + call_kwargs = mock_sub.call_args[1] + assert call_kwargs.get("input") == "data" + + +# --------------------------------------------------------------------------- +# split_repo() +# --------------------------------------------------------------------------- + + +def test_split_repo_valid(): + """Test that split_repo() splits 'owner/name' into a tuple.""" + assert split_repo("owner/repo") == ("owner", "repo") + + +def test_split_repo_with_nested_slash(): + """Test that split_repo() only splits on the first slash.""" + assert split_repo("owner/some/deep") == ("owner", "some/deep") + + +def test_split_repo_no_slash(): + """Test that split_repo() raises ValueError when there is no slash.""" + with pytest.raises(ValueError, match="owner/name"): + split_repo("noslash") + + +def test_split_repo_empty_owner(): + """Test that split_repo() raises ValueError when owner is empty.""" + with pytest.raises(ValueError, match="owner/name"): + split_repo("/repo") + + +def test_split_repo_empty_name(): + """Test that split_repo() raises ValueError when name is empty.""" + with pytest.raises(ValueError, match="owner/name"): + split_repo("owner/") + + +# --------------------------------------------------------------------------- +# gh_graphql() +# --------------------------------------------------------------------------- + + +def test_gh_graphql_string_field(): + """Test that gh_graphql() uses -f for string-valued fields.""" + with patch("pr_review_merge_scheduler.run") as mock_run: + mock_run.return_value = '{"data": {}}' + result = gh_graphql("query {}", owner="myowner") + args_passed = mock_run.call_args[0][0] + assert "-f" in args_passed + assert any("owner=myowner" in a for a in args_passed) + assert result == {"data": {}} + + +def test_gh_graphql_int_field(): + """Test that gh_graphql() uses -F for integer-valued fields.""" + with patch("pr_review_merge_scheduler.run") as mock_run: + mock_run.return_value = '{"data": {}}' + gh_graphql("query {}", pageSize=50) + args_passed = mock_run.call_args[0][0] + assert "-F" in args_passed + assert any("pageSize=50" in a for a in args_passed) + + +def test_gh_graphql_passes_query_as_stdin(): + """Test that gh_graphql() passes the query string as stdin.""" + with patch("pr_review_merge_scheduler.run") as mock_run: + mock_run.return_value = '{"data": {}}' + gh_graphql("my query") + _, kwargs = mock_run.call_args + assert kwargs.get("stdin") == "my query" + + +# --------------------------------------------------------------------------- +# fetch_open_prs() +# --------------------------------------------------------------------------- + + +def _gql_page(nodes: list, *, has_next: bool, cursor: str | None = None) -> dict: + """Build a mock gh_graphql response for a page of pull requests.""" + return { + "data": { + "repository": { + "pullRequests": { + "pageInfo": {"hasNextPage": has_next, "endCursor": cursor}, + "nodes": nodes, + } + } + } + } + + +def test_fetch_open_prs_single_page(): + """Test fetch_open_prs() with a single page of results.""" + nodes = [{"number": 1}] + with patch("pr_review_merge_scheduler.gh_graphql") as mock_gql: + mock_gql.return_value = _gql_page(nodes, has_next=False) + prs = fetch_open_prs("owner/repo", max_prs=100) + assert prs == nodes + mock_gql.assert_called_once() + # No cursor in first call + first_kwargs = mock_gql.call_args[1] + assert "cursor" not in first_kwargs + + +def test_fetch_open_prs_pagination(): + """Test fetch_open_prs() follows pagination to collect all PRs.""" + page1 = _gql_page([{"number": 1}], has_next=True, cursor="cur1") + page2 = _gql_page([{"number": 2}], has_next=False) + with patch("pr_review_merge_scheduler.gh_graphql") as mock_gql: + mock_gql.side_effect = [page1, page2] + prs = fetch_open_prs("owner/repo", max_prs=100) + assert len(prs) == 2 + # Second call must include cursor + second_kwargs = mock_gql.call_args_list[1][1] + assert second_kwargs.get("cursor") == "cur1" + + +def test_fetch_open_prs_max_prs_stops_loop(): + """Test fetch_open_prs() stops collecting when max_prs is reached.""" + with patch("pr_review_merge_scheduler.gh_graphql") as mock_gql: + mock_gql.return_value = _gql_page([{"number": 1}], has_next=True, cursor="c") + prs = fetch_open_prs("owner/repo", max_prs=1) + assert len(prs) == 1 + mock_gql.assert_called_once() + + +# --------------------------------------------------------------------------- +# context_nodes() +# --------------------------------------------------------------------------- + + +def test_context_nodes_normal(): + """Test context_nodes() extracts nodes from a normal PR.""" + pr = _make_pr(statusCheckRollup={"contexts": {"nodes": [{"a": 1}]}}) + assert context_nodes(pr) == [{"a": 1}] + + +def test_context_nodes_missing(): + """Test context_nodes() returns an empty list when statusCheckRollup is absent.""" + assert context_nodes({}) == [] + + +def test_context_nodes_none_values(): + """Test context_nodes() handles None at every level gracefully.""" + assert context_nodes({"statusCheckRollup": None}) == [] + + +# --------------------------------------------------------------------------- +# is_opencode_context() +# --------------------------------------------------------------------------- + + +def test_is_opencode_context_checkrun_by_name(): + """Test is_opencode_context() matches a CheckRun named opencode-review.""" + node = { + "__typename": "CheckRun", + "name": "opencode-review", + "status": "COMPLETED", + "checkSuite": None, + } + assert is_opencode_context(node) + + +def test_is_opencode_context_checkrun_by_workflow_name(): + """Test is_opencode_context() matches a CheckRun with the OpenCode Review workflow name.""" + node = { + "__typename": "CheckRun", + "name": "some-job", + "checkSuite": { + "workflowRun": {"workflow": {"name": "OpenCode Review"}} + }, + } + assert is_opencode_context(node) + + +def test_is_opencode_context_checkrun_no_match(): + """Test is_opencode_context() returns False for an unrelated CheckRun.""" + node = { + "__typename": "CheckRun", + "name": "ci-tests", + "checkSuite": {"workflowRun": {"workflow": {"name": "CI"}}}, + } + assert not is_opencode_context(node) + + +def test_is_opencode_context_status_context_match(): + """Test is_opencode_context() matches a StatusContext with opencode-review context.""" + node = {"__typename": "StatusContext", "context": "opencode-review", "state": "success"} + assert is_opencode_context(node) + + +def test_is_opencode_context_status_context_no_match(): + """Test is_opencode_context() returns False for an unrelated StatusContext.""" + node = {"__typename": "StatusContext", "context": "other-check", "state": "success"} + assert not is_opencode_context(node) + + +# --------------------------------------------------------------------------- +# opencode_in_progress() +# --------------------------------------------------------------------------- + + +def _checkrun_node(name: str, status: str) -> dict: + """Build a minimal CheckRun node dict for testing.""" + return {"__typename": "CheckRun", "name": name, "status": status, "checkSuite": None} + + +def test_opencode_in_progress_queued(): + """Test opencode_in_progress() returns True when status is QUEUED.""" + pr = _make_pr( + statusCheckRollup={"contexts": {"nodes": [_checkrun_node("opencode-review", "QUEUED")]}} + ) + assert opencode_in_progress(pr) + + +def test_opencode_in_progress_in_progress(): + """Test opencode_in_progress() returns True when status is IN_PROGRESS.""" + pr = _make_pr( + statusCheckRollup={ + "contexts": {"nodes": [_checkrun_node("opencode-review", "IN_PROGRESS")]} + } + ) + assert opencode_in_progress(pr) + + +def test_opencode_in_progress_completed(): + """Test opencode_in_progress() returns False when status is COMPLETED.""" + pr = _make_pr( + statusCheckRollup={ + "contexts": {"nodes": [_checkrun_node("opencode-review", "COMPLETED")]} + } + ) + assert not opencode_in_progress(pr) + + +def test_opencode_in_progress_non_opencode_node(): + """Test opencode_in_progress() ignores non-opencode nodes.""" + pr = _make_pr( + statusCheckRollup={ + "contexts": {"nodes": [_checkrun_node("ci-tests", "IN_PROGRESS")]} + } + ) + assert not opencode_in_progress(pr) + + +def test_opencode_in_progress_empty_status(): + """Test opencode_in_progress() treats an empty status as not in progress.""" + pr = _make_pr( + statusCheckRollup={"contexts": {"nodes": [_checkrun_node("opencode-review", "")]}} + ) + assert not opencode_in_progress(pr) + + +def test_opencode_in_progress_no_nodes(): + """Test opencode_in_progress() returns False when there are no nodes.""" + assert not opencode_in_progress(_BASE_PR) + + +# --------------------------------------------------------------------------- +# unresolved_thread_count() +# --------------------------------------------------------------------------- + + +def test_unresolved_thread_count_none(): + """Test unresolved_thread_count() returns 0 when there are no threads.""" + assert unresolved_thread_count(_make_pr()) == 0 + + +def test_unresolved_thread_count_mixed(): + """Test unresolved_thread_count() counts only non-outdated unresolved threads.""" + pr = _make_pr( + reviewThreads={ + "nodes": [ + {"isResolved": False, "isOutdated": False}, # counts + {"isResolved": True, "isOutdated": False}, # resolved + {"isResolved": False, "isOutdated": True}, # outdated + ] + } + ) + assert unresolved_thread_count(pr) == 1 + + +# --------------------------------------------------------------------------- +# review_author_login() +# --------------------------------------------------------------------------- + + +def test_review_author_login_present(): + """Test review_author_login() returns the lowercased login.""" + assert review_author_login({"author": {"login": "OpenCode-Agent"}}) == "opencode-agent" + + +def test_review_author_login_missing_author(): + """Test review_author_login() returns an empty string when author is absent.""" + assert review_author_login({}) == "" + + +def test_review_author_login_null_login(): + """Test review_author_login() handles a None login value.""" + assert review_author_login({"author": {"login": None}}) == "" + + +# --------------------------------------------------------------------------- +# is_opencode_review() +# --------------------------------------------------------------------------- + + +def test_is_opencode_review_true(): + """Test is_opencode_review() returns True for opencode-agent.""" + assert is_opencode_review({"author": {"login": "opencode-agent"}}) + + +def test_is_opencode_review_false(): + """Test is_opencode_review() returns False for other authors.""" + assert not is_opencode_review({"author": {"login": "human-reviewer"}}) + + +# --------------------------------------------------------------------------- +# current_head_review_state() +# --------------------------------------------------------------------------- + + +def test_current_head_review_state_no_reviews(): + """Test current_head_review_state() returns False when there are no reviews.""" + assert not current_head_review_state(_make_pr(), "APPROVED") + + +def test_current_head_review_state_match(): + """Test current_head_review_state() finds an opencode-agent approval on the head.""" + pr = _approved_pr("sha1") + assert current_head_review_state(pr, "APPROVED") + + +def test_current_head_review_state_non_opencode(): + """Test current_head_review_state() ignores reviews from non-opencode authors.""" + pr = _make_pr( + headRefOid="sha1", + reviews={ + "nodes": [ + { + "state": "APPROVED", + "author": {"login": "human"}, + "commit": {"oid": "sha1"}, + } + ] + }, + ) + assert not current_head_review_state(pr, "APPROVED") + + +def test_current_head_review_state_wrong_state(): + """Test current_head_review_state() returns False when state does not match.""" + pr = _make_pr( + headRefOid="sha1", + reviews={ + "nodes": [ + { + "state": "CHANGES_REQUESTED", + "author": {"login": "opencode-agent"}, + "commit": {"oid": "sha1"}, + } + ] + }, + ) + assert not current_head_review_state(pr, "APPROVED") + + +def test_current_head_review_state_old_commit(): + """Test current_head_review_state() returns False when review is on an old commit.""" + pr = _make_pr( + headRefOid="new-sha", + reviews={ + "nodes": [ + { + "state": "APPROVED", + "author": {"login": "opencode-agent"}, + "commit": {"oid": "old-sha"}, + } + ] + }, + ) + assert not current_head_review_state(pr, "APPROVED") + + +# --------------------------------------------------------------------------- +# has_current_head_approval / has_current_head_changes_requested +# --------------------------------------------------------------------------- + + +def test_has_current_head_approval_true(): + """Test has_current_head_approval() returns True for a current-head approval.""" + assert has_current_head_approval(_approved_pr()) + + +def test_has_current_head_approval_false(): + """Test has_current_head_approval() returns False when no approval exists.""" + assert not has_current_head_approval(_make_pr()) + + +def test_has_current_head_changes_requested_true(): + """Test has_current_head_changes_requested() detects a current-head changes-requested review.""" + pr = _make_pr( + headRefOid="sha1", + reviews={ + "nodes": [ + { + "state": "CHANGES_REQUESTED", + "author": {"login": "opencode-agent"}, + "commit": {"oid": "sha1"}, + } + ] + }, + ) + assert has_current_head_changes_requested(pr) + + +def test_has_current_head_changes_requested_false(): + """Test has_current_head_changes_requested() returns False when no such review exists.""" + assert not has_current_head_changes_requested(_make_pr()) + + +# --------------------------------------------------------------------------- +# enable_auto_merge() +# --------------------------------------------------------------------------- + + +def test_enable_auto_merge_dry_run(): + """Test enable_auto_merge() does nothing when dry_run is True.""" + with patch("pr_review_merge_scheduler.run") as mock_run: + enable_auto_merge("owner/repo", {"number": 1, "headRefOid": "sha"}, dry_run=True) + mock_run.assert_not_called() + + +def test_enable_auto_merge_not_dry_run(): + """Test enable_auto_merge() calls gh pr merge when not in dry_run mode.""" + with patch("pr_review_merge_scheduler.run") as mock_run: + enable_auto_merge("owner/repo", {"number": 42, "headRefOid": "sha"}, dry_run=False) + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "gh" in cmd + assert "--auto" in cmd + assert "42" in cmd + + +# --------------------------------------------------------------------------- +# dispatch_opencode_review() +# --------------------------------------------------------------------------- + + +def test_dispatch_opencode_review_dry_run(): + """Test dispatch_opencode_review() does nothing when dry_run is True.""" + pr = { + "number": 1, + "baseRefName": "main", + "baseRefOid": "base", + "headRefName": "feat", + "headRefOid": "head", + } + with patch("pr_review_merge_scheduler.run") as mock_run: + dispatch_opencode_review("owner/repo", "wf", pr, dry_run=True) + mock_run.assert_not_called() + + +def test_dispatch_opencode_review_not_dry_run(): + """Test dispatch_opencode_review() calls gh workflow run when not in dry_run mode.""" + pr = { + "number": 7, + "baseRefName": "main", + "baseRefOid": "base", + "headRefName": "feat", + "headRefOid": "head", + } + with patch("pr_review_merge_scheduler.run") as mock_run: + dispatch_opencode_review("owner/repo", "OpenCode Review", pr, dry_run=False) + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "workflow" in cmd + assert "run" in cmd + + +# --------------------------------------------------------------------------- +# inspect_pr() +# --------------------------------------------------------------------------- + +_INSPECT_DEFAULTS: dict = { + "repo": "owner/repo", + "dry_run": True, + "trigger_reviews": True, + "enable_auto_merge_flag": True, + "workflow": "OpenCode Review", + "base_branch": "main", +} + + +def _inspect(pr: dict, **kwargs: object) -> Decision: + """Call inspect_pr() with default kwargs merged with caller overrides.""" + opts = {**_INSPECT_DEFAULTS, **kwargs} + return inspect_pr(opts.pop("repo"), pr, **opts) # type: ignore[arg-type] + + +def test_inspect_pr_draft(): + """Test inspect_pr() returns 'skip' for a draft PR.""" + d = _inspect(_make_pr(isDraft=True)) + assert d.action == "skip" + assert "draft" in d.reason + + +def test_inspect_pr_wrong_base_branch(): + """Test inspect_pr() returns 'skip' when the base branch is unexpected.""" + d = _inspect(_make_pr(baseRefName="develop")) + assert d.action == "skip" + assert "base branch" in d.reason + + +def test_inspect_pr_fork(): + """Test inspect_pr() returns 'skip' for fork or external head repo.""" + d = _inspect(_make_pr(**{"headRepository": {"nameWithOwner": "other/repo"}})) + assert d.action == "skip" + assert "fork" in d.reason + + +def test_inspect_pr_unresolved_thread(): + """Test inspect_pr() returns 'block' when there are unresolved review threads.""" + pr = _make_pr( + reviewThreads={"nodes": [{"isResolved": False, "isOutdated": False}]} + ) + d = _inspect(pr) + assert d.action == "block" + assert "unresolved" in d.reason + + +def test_inspect_pr_changes_requested(): + """Test inspect_pr() returns 'block' when current head has changes requested.""" + pr = _make_pr( + headRefOid="sha1", + reviews={ + "nodes": [ + { + "state": "CHANGES_REQUESTED", + "author": {"login": "opencode-agent"}, + "commit": {"oid": "sha1"}, + } + ] + }, + ) + d = _inspect(pr) + assert d.action == "block" + + +def test_inspect_pr_approved_auto_merge_already_enabled(): + """Test inspect_pr() returns 'wait' when current head is approved and auto-merge is already set.""" + pr = dict(_approved_pr()) + pr["autoMergeRequest"] = {"enabledAt": "2026-01-01"} + d = _inspect(pr) + assert d.action == "wait" + assert "already" in d.reason + + +def test_inspect_pr_approved_auto_merge_flag_disabled(): + """Test inspect_pr() returns 'wait' when approved but auto-merge flag is False.""" + d = _inspect(_approved_pr(), enable_auto_merge_flag=False) + assert d.action == "wait" + assert "disabled" in d.reason + + +def test_inspect_pr_approved_enables_auto_merge(): + """Test inspect_pr() enables auto-merge and returns 'auto_merge' when conditions are met.""" + with patch("pr_review_merge_scheduler.run"): + d = _inspect(_approved_pr(), dry_run=False) + assert d.action == "auto_merge" + + +def test_inspect_pr_opencode_in_progress(): + """Test inspect_pr() returns 'wait' when an OpenCode review is already running.""" + pr = _make_pr( + statusCheckRollup={ + "contexts": {"nodes": [_checkrun_node("opencode-review", "IN_PROGRESS")]} + } + ) + d = _inspect(pr) + assert d.action == "wait" + assert "in progress" in d.reason + + +def test_inspect_pr_triggers_review(): + """Test inspect_pr() dispatches a review and returns 'review_dispatch' when trigger is True.""" + with patch("pr_review_merge_scheduler.run"): + d = _inspect(_make_pr(), dry_run=False) + assert d.action == "review_dispatch" + + +def test_inspect_pr_block_no_trigger(): + """Test inspect_pr() returns 'block' when trigger_reviews is False and no approval.""" + d = _inspect(_make_pr(), trigger_reviews=False) + assert d.action == "block" + assert "no OpenCode approval" in d.reason + + +# --------------------------------------------------------------------------- +# print_summary() +# --------------------------------------------------------------------------- + + +def test_print_summary(capsys: pytest.CaptureFixture) -> None: + """Test print_summary() emits per-PR lines and a JSON summary.""" + decisions = [ + Decision(pr=1, action="auto_merge", reason="approved"), + Decision(pr=2, action="skip", reason="draft"), + Decision(pr=3, action="auto_merge", reason="approved"), + ] + print_summary(decisions, dry_run=True, base_branch="main", project_flow="default") + out = capsys.readouterr().out + assert "PR #1" in out + assert "PR #2" in out + summary = json.loads(out.splitlines()[-1]) + assert summary["inspected"] == 3 + assert summary["counts"]["auto_merge"] == 2 + assert summary["counts"]["skip"] == 1 + assert summary["base_branch"] == "main" + assert summary["dry_run"] is True + + +def test_print_summary_empty(capsys: pytest.CaptureFixture) -> None: + """Test print_summary() works with an empty decision list.""" + print_summary([], dry_run=False, base_branch="main", project_flow="flow") + out = capsys.readouterr().out + summary = json.loads(out.strip()) + assert summary["inspected"] == 0 + assert summary["counts"] == {} + + +# --------------------------------------------------------------------------- +# self_test() +# --------------------------------------------------------------------------- + + +def test_self_test(capsys: pytest.CaptureFixture) -> None: + """Test self_test() runs without raising and prints a success message.""" + self_test() + assert "self-test passed" in capsys.readouterr().out + + +# --------------------------------------------------------------------------- +# parse_args() +# --------------------------------------------------------------------------- + + +def test_parse_args_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + """Test parse_args() sets default values when no arguments are provided.""" + monkeypatch.delenv("GITHUB_REPOSITORY", raising=False) + monkeypatch.delenv("DEFAULT_BRANCH", raising=False) + monkeypatch.delenv("PROJECT_FLOW", raising=False) + args = parse_args([]) + assert args.max_prs == 100 + assert not args.dry_run + assert args.trigger_reviews is True + assert args.enable_auto_merge is True + assert not args.self_test + + +def test_parse_args_custom() -> None: + """Test parse_args() accepts and returns custom values.""" + args = parse_args( + [ + "--repo", + "owner/repo", + "--base-branch", + "main", + "--max-prs", + "25", + "--dry-run", + "--review-workflow", + "My Workflow", + "--self-test", + ] + ) + assert args.repo == "owner/repo" + assert args.base_branch == "main" + assert args.max_prs == 25 + assert args.dry_run + assert args.review_workflow == "My Workflow" + assert args.self_test + + +def test_parse_args_no_trigger_reviews() -> None: + """Test parse_args() handles --no-trigger-reviews flag.""" + args = parse_args(["--no-trigger-reviews"]) + assert args.trigger_reviews is False + + +def test_parse_args_no_auto_merge() -> None: + """Test parse_args() handles --no-enable-auto-merge flag.""" + args = parse_args(["--no-enable-auto-merge"]) + assert args.enable_auto_merge is False + + +# --------------------------------------------------------------------------- +# main() +# --------------------------------------------------------------------------- + + +def test_main_self_test() -> None: + """Test main() runs the self-test and returns 0 when --self-test is given.""" + assert main(["--self-test"]) == 0 + + +def test_main_missing_repo(monkeypatch: pytest.MonkeyPatch) -> None: + """Test main() raises SystemExit when --repo is not supplied.""" + monkeypatch.delenv("GITHUB_REPOSITORY", raising=False) + monkeypatch.delenv("DEFAULT_BRANCH", raising=False) + monkeypatch.delenv("PROJECT_FLOW", raising=False) + with pytest.raises(SystemExit): + main([]) + + +def test_main_missing_base_branch(monkeypatch: pytest.MonkeyPatch) -> None: + """Test main() raises SystemExit when --base-branch is not supplied.""" + monkeypatch.delenv("DEFAULT_BRANCH", raising=False) + monkeypatch.delenv("PROJECT_FLOW", raising=False) + with pytest.raises(SystemExit): + main(["--repo", "owner/repo"]) + + +def test_main_missing_project_flow(monkeypatch: pytest.MonkeyPatch) -> None: + """Test main() raises SystemExit when --project-flow is not supplied.""" + monkeypatch.delenv("PROJECT_FLOW", raising=False) + with pytest.raises(SystemExit): + main(["--repo", "owner/repo", "--base-branch", "main"]) + + +def test_main_normal_run() -> None: + """Test main() fetches PRs and prints summary when all required args are given.""" + with patch("pr_review_merge_scheduler.fetch_open_prs") as mock_fetch: + mock_fetch.return_value = [] + result = main( + [ + "--repo", + "owner/repo", + "--base-branch", + "main", + "--project-flow", + "default", + ] + ) + assert result == 0 + mock_fetch.assert_called_once_with("owner/repo", 100) + + +def test_main_normal_run_with_prs() -> None: + """Test main() processes multiple PRs and returns 0.""" + sample_pr = _make_pr() + with patch("pr_review_merge_scheduler.fetch_open_prs") as mock_fetch: + mock_fetch.return_value = [sample_pr] + result = main( + [ + "--repo", + "owner/repo", + "--base-branch", + "main", + "--project-flow", + "default", + "--dry-run", + ] + ) + assert result == 0