Add shared read-only gh CLI helper for maintenance scripts

scripts/gh_cli.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+import json
+import subprocess
+from typing import Any
+
+DEFAULT_GH_TIMEOUT_SECONDS = 30
+
+_NON_READ_ONLY_GH_API = frozenset({"POST", "PUT", "PATCH", "DELETE"})
+_NON_READ_ONLY_GH_SUBCOMMANDS = frozenset(
+    {"comment", "edit", "close", "reopen", "merge", "review", "delete"}
+)
+
+
+def _command_text(args: list[str]) -> str:
+    return " ".join(args)
+
+
+def assert_read_only_gh_command(args: list[str]) -> None:
+    """Reject gh invocations that could mutate GitHub state."""
+    if args[:2] == ["gh", "api"]:
+        for flag in ("--method", "-X"):
+            if flag not in args:
+                continue
+            index = args.index(flag)
+            if index + 1 >= len(args):
+                continue
+            if args[index + 1].upper() in _NON_READ_ONLY_GH_API:
+                raise RuntimeError(f"refusing non-read-only gh api command: {_command_text(args)}")
+    if any(arg in {"issue", "pr"} for arg in args) and any(
+        arg in _NON_READ_ONLY_GH_SUBCOMMANDS for arg in args
+    ):
+        raise RuntimeError(f"refusing non-read-only gh command: {_command_text(args)}")
+
+
+def run_gh(args: list[str], *, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS) -> str:
+    """Run a read-only gh command and return stdout text."""
+    assert_read_only_gh_command(args)
+    command = _command_text(args)
+    try:
+        completed = subprocess.run(
+            args,
+            check=True,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=timeout_seconds,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise RuntimeError(f"gh command timed out after {timeout_seconds}s: {command}") from exc
+    except FileNotFoundError as exc:
+        raise RuntimeError(
+            "GitHub CLI executable 'gh' was not found; install gh and ensure it is on PATH "
+            "before using live --repo mode"
+        ) from exc
+    except subprocess.CalledProcessError as exc:
+        raise RuntimeError(
+            "gh command failed "
+            f"(exit {exc.returncode}): {command}\n"
+            f"stdout:\n{exc.stdout or exc.output or ''}\n"
+            f"stderr:\n{exc.stderr or ''}"
+        ) from exc
+    return completed.stdout
+
+
+def run_gh_json(args: list[str], *, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS) -> Any:
+    """Run a read-only gh command and parse JSON stdout."""
+    return json.loads(run_gh(args, timeout_seconds=timeout_seconds))
+
+
+def ensure_json_list(data: Any, *, label: str) -> list[Any]:
+    if not isinstance(data, list):
+        raise RuntimeError(f"{label} returned non-list JSON ({type(data).__name__})")
+    return data
+
+
+def ensure_json_object(data: Any, *, label: str) -> dict[str, Any]:
+    if not isinstance(data, dict):
+        raise RuntimeError(f"{label} returned non-object JSON ({type(data).__name__})")
+    return data
+
+
+def run_gh_json_list(
+    args: list[str], *, label: str | None = None, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS
+) -> list[Any]:
+    context = label or _command_text(args)
+    return ensure_json_list(run_gh_json(args, timeout_seconds=timeout_seconds), label=context)
+
+
+def run_gh_json_object(
+    args: list[str], *, label: str | None = None, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS
+) -> dict[str, Any]:
+    context = label or _command_text(args)
+    return ensure_json_object(run_gh_json(args, timeout_seconds=timeout_seconds), label=context)

scripts/public_json_fetch.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.request
+from typing import Any
+
+
+class PublicJsonError(RuntimeError):
+    """Raised when a public JSON endpoint cannot be fetched or decoded."""
+
+
+def load_public_json(
+    url: str,
+    *,
+    description: str | None = None,
+    timeout: float = 30,
+    user_agent: str = "mergework-maintenance-script",
+    accept: str = "application/json",
+) -> Any:
+    label = description or "public JSON"
+    request = urllib.request.Request(
+        url,
+        headers={
+            "Accept": accept,
+            "User-Agent": user_agent,
+        },
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            payload = response.read().decode("utf-8")
+    except urllib.error.HTTPError as exc:
+        raise PublicJsonError(f"{label} unavailable: HTTP {exc.code}") from exc
+    except (urllib.error.URLError, TimeoutError) as exc:
+        raise PublicJsonError(f"{label} unavailable: {exc}") from exc
+    try:
+        return json.loads(payload)
+    except json.JSONDecodeError as exc:
+        raise PublicJsonError(f"{label} unavailable: invalid JSON") from exc

scripts/source_args.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import argparse
+
+
+def _require_non_empty(parser: argparse.ArgumentParser, value: str | None, *, label: str) -> str:
+    if value is None or not value.strip():
+        parser.error(f"{label} must not be empty or whitespace-only")
+    return value.strip()
+
+
+def validate_source_args(
+    parser: argparse.ArgumentParser,
+    *,
+    input_value: str | None,
+    repo_value: str | None,
+    fix: bool = False,
+) -> tuple[str | None, str]:
+    if input_value is not None and repo_value is not None:
+        parser.error("argument --repo: not allowed with argument --input")
+    if input_value is None and repo_value is None:
+        parser.error("one of the arguments --input --repo is required")
+    if input_value is not None:
+        input_arg = _require_non_empty(parser, input_value, label="--input")
+        if fix:
+            parser.error("--fix requires --repo, not --input")
+        return input_arg, ""
+    return None, _require_non_empty(parser, repo_value, label="--repo")

tests/test_gh_cli.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import json
+from unittest.mock import patch
+
+import pytest
+
+from scripts.gh_cli import assert_read_only_gh_command, run_gh, run_gh_json
+
+
+def test_assert_read_only_gh_command_rejects_mutating_api_method() -> None:
+    with pytest.raises(RuntimeError, match="refusing non-read-only gh api command"):
+        assert_read_only_gh_command(["gh", "api", "repos/x/y/issues/1", "--method", "POST"])
+
+
+def test_assert_read_only_gh_command_rejects_issue_comment() -> None:
+    with pytest.raises(RuntimeError, match="refusing non-read-only gh command"):
+        assert_read_only_gh_command(
+            ["gh", "issue", "comment", "1", "--repo", "x/y", "--body", "hi"]
+        )
+
+
+def test_run_gh_returns_stdout() -> None:
+    completed = type("Completed", (), {"stdout": '{"ok": true}'})()
+
+    with patch("scripts.gh_cli.subprocess.run", return_value=completed):
+        assert run_gh(["gh", "api", "repos/x/y"]) == '{"ok": true}'
+
+
+def test_run_gh_json_parses_payload() -> None:
+    payload = {"items": [1, 2]}
+    completed = type("Completed", (), {"stdout": json.dumps(payload)})()
+
+    with patch("scripts.gh_cli.subprocess.run", return_value=completed):
+        assert run_gh_json(["gh", "api", "repos/x/y"]) == payload
+
+
+def test_run_gh_wraps_missing_executable() -> None:
+    with (
+        patch("scripts.gh_cli.subprocess.run", side_effect=FileNotFoundError("gh")),
+        pytest.raises(RuntimeError, match="GitHub CLI executable 'gh' was not found"),
+    ):
+        run_gh(["gh", "api", "repos/x/y"])

tests/test_public_json_fetch.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import io
+import json
+import urllib.error
+from unittest.mock import patch
+
+import pytest
+
+from scripts.public_json_fetch import PublicJsonError, load_public_json
+
+
+def test_load_public_json_returns_decoded_payload() -> None:
+    payload = {"status": "ok"}
+    response = io.BytesIO(json.dumps(payload).encode("utf-8"))
+    response.status = 200  # type: ignore[attr-defined]
+
+    with patch("urllib.request.urlopen", return_value=response):
+        assert load_public_json("https://example.test/api/v1/status") == payload
+
+
+def test_load_public_json_sets_default_headers() -> None:
+    captured: dict[str, str] = {}
+
+    class FakeResponse(io.BytesIO):
+        def __init__(self) -> None:
+            super().__init__(b"[]")
+
+    def fake_urlopen(request, timeout=30):  # noqa: ANN001
+        captured["accept"] = request.get_header("Accept")
+        captured["user_agent"] = request.get_header("User-agent")
+        captured["timeout"] = str(timeout)
+        return FakeResponse()
+
+    with patch("urllib.request.urlopen", side_effect=fake_urlopen):
+        load_public_json("https://example.test/api/v1/bounties")
+
+    assert captured["accept"] == "application/json"
+    assert captured["user_agent"] == "mergework-maintenance-script"
+    assert captured["timeout"] == "30"
+
+
+def test_load_public_json_wraps_http_error() -> None:
+    error = urllib.error.HTTPError(
+        url="https://example.test/api/v1/bounties",
+        code=503,
+        msg="service unavailable",
+        hdrs=None,
+        fp=None,
+    )
+    with (
+        patch("urllib.request.urlopen", side_effect=error),
+        pytest.raises(PublicJsonError, match="MergeWork API bounty data unavailable: HTTP 503"),
+    ):
+        load_public_json(
+            "https://example.test/api/v1/bounties",
+            description="MergeWork API bounty data",
+        )
+
+
+def test_load_public_json_wraps_timeout() -> None:
+    with (
+        patch("urllib.request.urlopen", side_effect=TimeoutError("timed out")),
+        pytest.raises(PublicJsonError, match="public JSON unavailable:"),
+    ):
+        load_public_json("https://example.test/api/v1/status")
+
+
+def test_load_public_json_wraps_invalid_json() -> None:
+    response = io.BytesIO(b"not-json")
+
+    with (
+        patch("urllib.request.urlopen", return_value=response),
+        pytest.raises(PublicJsonError, match="public JSON unavailable: invalid JSON"),
+    ):
+        load_public_json("https://example.test/api/v1/status")
+
+
+def test_load_public_json_honors_custom_timeout() -> None:
+    captured: dict[str, str] = {}
+
+    class FakeResponse(io.BytesIO):
+        def __init__(self) -> None:
+            super().__init__(b"{}")
+
+    def fake_urlopen(request, timeout=30):  # noqa: ANN001
+        captured["timeout"] = str(timeout)
+        return FakeResponse()
+
+    with patch("urllib.request.urlopen", side_effect=fake_urlopen):
+        load_public_json("https://example.test/api/v1/status", timeout=12)
+
+    assert captured["timeout"] == "12"