🧪 [Add unit tests and implementation for get_pr_status.py]

.gitignore

@@ -34,7 +34,7 @@ router/free_models_roster.json
 .jules/
 workdirs/
 pr_description.txt
-scripts/get_pr_status.py
+/get_pr_status.py
 
 # Dataset work in progress
 data/

scripts/README.md

@@ -4,7 +4,7 @@ This directory and the repository root contain various scripts used for stack or
 
 ---
 
-## 1. Stack Orchestration & Backups
+## 1. Stack Orchestration, Backups, and PR Utilities
 
 ### `start-stack.sh` (Root Directory)
 Unified startup and credential extraction script for the Podman Kubernetes container stack.
@@ -16,6 +16,13 @@ Unified startup and credential extraction script for the Podman Kubernetes conta
 ### `scripts/backup.sh`
 Automated database backup script that runs before every stack deployment. Uses `pg_isready` to safely wait for database connections and manages timestamped backups under `backups/`.
 
+### `scripts/get_pr_status.py`
+A CLI utility to fetch and print the status of a GitHub PR (its state, review decisions, and checks status rollup) using the `gh` CLI.
+- **Usage**:
+  - `python3 scripts/get_pr_status.py` (Fetches status of the current branch's PR)
+  - `python3 scripts/get_pr_status.py <pr_id>` (Fetches status for the specified PR ID)
+
+
 ---
 
 ## 2. Routing & Cooldown Verification Scripts
@@ -82,6 +89,9 @@ The integration test suite is located in the root directory. Tests are categoriz
 ### Dashboard & Annotations Tests
 - **`tests/test_read_annotations_async.py`**: Unit tests for the asynchronous dashboard annotation reading and caching mechanism.
 
+### Utility Script Tests
+- **`tests/test_get_pr_status.py`**: Unit tests validating command execution, status fetching, and error/timeout handling in `scripts/get_pr_status.py`.
+
 ### Simulation Tests
 - **`test_agy_behavior.py`**: Asserts the behavior of the `agy` CLI client under quota limits.
 - **`test_quota_reset.sh`**: Simulates/triggers quota reset conditions.

scripts/get_pr_status.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+import subprocess
+import json
+import sys
+from typing import Sequence
+
+
+def run_cmd(argv: Sequence[str]) -> str:
+    """Runs a command and returns stripped stdout."""
+    result = subprocess.run(argv, shell=False, capture_output=True, text=True, check=True, timeout=30)
+    return result.stdout.strip()
+
+
+def get_pr_status(pr_id: str = "") -> None:
+    """Fetches and prints the status of a PR using gh CLI."""
+    cmd = ["gh", "pr", "view"]
+    if pr_id:
+        cmd.append(pr_id)
+    cmd.extend(["--json", "state,reviewDecision,statusCheckRollup"])
+
+    try:
+        output = run_cmd(cmd)
+        data = json.loads(output)
+
+        state = data.get("state")
+        review = data.get("reviewDecision") or "NONE"
+        checks = data.get("statusCheckRollup", [])
+
+        # Summarize checks
+        success_count = 0
+        total_count = len(checks)
+        for check in checks:
+            # gh CLI returns conclusion for CheckRun and state for StatusContext
+            conclusion = check.get("conclusion") or check.get("state")
+            if conclusion == "SUCCESS":
+                success_count += 1
+
+        print(f"PR Status: {state}")
+        print(f"Review Decision: {review}")
+        print(f"Checks: {success_count}/{total_count} passed")
+
+    except subprocess.CalledProcessError as e:
+        print(f"Error: Failed to fetch PR status: {e.stderr.strip()}", file=sys.stderr)
+        sys.exit(1)
+    except json.JSONDecodeError:
+        print("Error: Failed to parse gh CLI output", file=sys.stderr)
+        sys.exit(1)
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+def main():
+    pr_id = sys.argv[1] if len(sys.argv) > 1 else ""
+    get_pr_status(pr_id)
+
+
+if __name__ == "__main__":
+    main()

tests/test_get_pr_status.py

@@ -0,0 +1,84 @@
+import pytest
+import subprocess
+import json
+from unittest.mock import patch, MagicMock
+from scripts.get_pr_status import run_cmd, get_pr_status
+
+def test_run_cmd_success():
+    output = run_cmd(["echo", "hello"])
+    assert output == "hello"
+
+def test_run_cmd_strips_whitespace():
+    output = run_cmd(["echo", "  hello  "])
+    assert output == "hello"
+
+def test_run_cmd_error():
+    with pytest.raises(subprocess.CalledProcessError):
+        run_cmd(["false"])
+
+@patch("scripts.get_pr_status.subprocess.run")
+def test_run_cmd_timeout(mock_run):
+    # run_cmd has a 30s timeout. We mock subprocess.run to raise it immediately.
+    mock_run.side_effect = subprocess.TimeoutExpired(["sleep", "0.1"], 30)
+    with pytest.raises(subprocess.TimeoutExpired):
+        run_cmd(["sleep", "0.1"])
+
+@patch("scripts.get_pr_status.run_cmd")
+def test_get_pr_status_success(mock_run_cmd, capsys):
+    mock_data = {
+        "state": "OPEN",
+        "reviewDecision": "APPROVED",
+        "statusCheckRollup": [
+            {"conclusion": "SUCCESS", "name": "test1"},
+            {"state": "SUCCESS", "name": "test2"},
+            {"conclusion": "FAILURE", "name": "test3"}
+        ]
+    }
+    mock_run_cmd.return_value = json.dumps(mock_data)
+
+    get_pr_status("123")
+
+    captured = capsys.readouterr()
+    assert "PR Status: OPEN" in captured.out
+    assert "Review Decision: APPROVED" in captured.out
+    assert "Checks: 2/3 passed" in captured.out
+    mock_run_cmd.assert_called_once_with(["gh", "pr", "view", "123", "--json", "state,reviewDecision,statusCheckRollup"])
+
+@patch("scripts.get_pr_status.run_cmd")
+def test_get_pr_status_no_id(mock_run_cmd, capsys):
+    mock_data = {
+        "state": "MERGED",
+        "reviewDecision": None,
+        "statusCheckRollup": []
+    }
+    mock_run_cmd.return_value = json.dumps(mock_data)
+
+    get_pr_status()
+
+    captured = capsys.readouterr()
+    assert "PR Status: MERGED" in captured.out
+    assert "Review Decision: NONE" in captured.out
+    assert "Checks: 0/0 passed" in captured.out
+    mock_run_cmd.assert_called_once_with(["gh", "pr", "view", "--json", "state,reviewDecision,statusCheckRollup"])
+
+@patch("scripts.get_pr_status.run_cmd")
+def test_get_pr_status_error(mock_run_cmd, capsys):
+    mock_run_cmd.side_effect = subprocess.CalledProcessError(1, ["gh"], stderr="gh not found")
+
+    with pytest.raises(SystemExit) as e:
+        get_pr_status("123")
+
+    assert e.value.code == 1
+    captured = capsys.readouterr()
+    assert "Error: Failed to fetch PR status: gh not found" in captured.err
+
+@patch("scripts.get_pr_status.run_cmd")
+def test_get_pr_status_invalid_json(mock_run_cmd, capsys):
+    mock_run_cmd.return_value = "invalid json"
+
+    with pytest.raises(SystemExit) as e:
+        get_pr_status("123")
+
+    assert e.value.code == 1
+    captured = capsys.readouterr()
+    assert "Error: Failed to parse gh CLI output" in captured.err