From 8e0fa266e1058b277dc1ca95f8133dc5109ab95f Mon Sep 17 00:00:00 2001 From: yanyishuai <1093994647@qq.com> Date: Fri, 3 Jul 2026 22:43:11 +0800 Subject: [PATCH] fix: overlay + ruff format/lint (#1162) --- scripts/gh_cli.py | 95 +++++++++++++++++++++++++++++++++ scripts/public_json_fetch.py | 39 ++++++++++++++ scripts/source_args.py | 28 ++++++++++ tests/test_gh_cli.py | 43 +++++++++++++++ tests/test_public_json_fetch.py | 93 ++++++++++++++++++++++++++++++++ 5 files changed, 298 insertions(+) create mode 100644 scripts/gh_cli.py create mode 100644 scripts/public_json_fetch.py create mode 100644 scripts/source_args.py create mode 100644 tests/test_gh_cli.py create mode 100644 tests/test_public_json_fetch.py diff --git a/scripts/gh_cli.py b/scripts/gh_cli.py new file mode 100644 index 00000000..74e38999 --- /dev/null +++ b/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) diff --git a/scripts/public_json_fetch.py b/scripts/public_json_fetch.py new file mode 100644 index 00000000..29f06dc4 --- /dev/null +++ b/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 diff --git a/scripts/source_args.py b/scripts/source_args.py new file mode 100644 index 00000000..e82e7793 --- /dev/null +++ b/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") diff --git a/tests/test_gh_cli.py b/tests/test_gh_cli.py new file mode 100644 index 00000000..c9d12035 --- /dev/null +++ b/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"]) diff --git a/tests/test_public_json_fetch.py b/tests/test_public_json_fetch.py new file mode 100644 index 00000000..288f7eac --- /dev/null +++ b/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"