From e2bacd932c474a0b1fa0de6717ed82610814ba5d Mon Sep 17 00:00:00 2001 From: yanyishuai <1093994647@qq.com> Date: Thu, 2 Jul 2026 23:48:01 +0800 Subject: [PATCH] fix: normalize LF line endings in source validation helpers (#1145) --- scripts/check_bounty_issue_states.py | 67 ++++---------- scripts/check_live_bounty_closing_refs.py | 62 +++---------- scripts/gh_cli.py | 95 +++++++++++++++++++ scripts/public_api_json.py | 96 ++++++++++++++++++++ scripts/source_args.py | 28 ++++++ tests/test_check_bounty_issue_states.py | 25 +++++ tests/test_check_live_bounty_closing_refs.py | 25 +++++ tests/test_gh_cli.py | 43 +++++++++ 8 files changed, 342 insertions(+), 99 deletions(-) create mode 100644 scripts/gh_cli.py create mode 100644 scripts/public_api_json.py create mode 100644 scripts/source_args.py create mode 100644 tests/test_gh_cli.py diff --git a/scripts/check_bounty_issue_states.py b/scripts/check_bounty_issue_states.py index 23a0a80d..a8d77365 100644 --- a/scripts/check_bounty_issue_states.py +++ b/scripts/check_bounty_issue_states.py @@ -4,8 +4,6 @@ import json import subprocess import sys -import urllib.error -import urllib.request from pathlib import Path from typing import Any @@ -13,9 +11,12 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts.api_host_args import public_api_host +from scripts.gh_cli import DEFAULT_GH_TIMEOUT_SECONDS as GH_TIMEOUT_SECONDS +from scripts.gh_cli import run_gh_json as _run_gh_json +from scripts.public_api_json import load_public_bounty_list +from scripts.source_args import validate_source_args DEFAULT_API_HOST = "https://api.mrwk.online" -GH_TIMEOUT_SECONDS = 30 def _int_or_none(value: Any) -> int | None: @@ -119,31 +120,8 @@ def format_text_report(report: dict[str, Any]) -> str: return "\n".join(lines) -def _run_gh_json(args: list[str]) -> Any: - command = " ".join(args) - try: - completed = subprocess.run( - args, - check=True, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=GH_TIMEOUT_SECONDS, - ) - except subprocess.TimeoutExpired as exc: - raise RuntimeError(f"gh command timed out after {GH_TIMEOUT_SECONDS}s: {command}") 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 json.loads(completed.stdout) - - def _run_gh(args: list[str]) -> None: + """Run gh for maintainer --fix paths (may mutate GitHub state).""" command = " ".join(args) try: subprocess.run( @@ -166,23 +144,6 @@ def _run_gh(args: list[str]) -> None: ) from exc -def _fetch_json(url: str) -> Any: - request = urllib.request.Request(url, headers={"Accept": "application/json"}) - try: - with urllib.request.urlopen(request, timeout=GH_TIMEOUT_SECONDS) as response: - return json.loads(response.read().decode("utf-8")) - except (TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc: - raise RuntimeError(f"failed to fetch JSON from {url}: {exc}") from exc - - -def _load_public_bounties(api_host: str) -> list[dict[str, Any]]: - url = f"{api_host.rstrip('/')}/api/v1/bounties?status=open&limit=200" - data = _fetch_json(url) - if not isinstance(data, list): - raise RuntimeError(f"expected a JSON list from {url}") - return [item for item in data if isinstance(item, dict)] - - def _load_issue(repo: str, issue_number: int) -> dict[str, Any]: return _run_gh_json( [ @@ -199,7 +160,7 @@ def _load_issue(repo: str, issue_number: int) -> dict[str, Any]: def load_live_data(repo: str, api_host: str) -> dict[str, Any]: - bounties = _load_public_bounties(api_host) + bounties = load_public_bounty_list(api_host, query="status=open&limit=200") issue_numbers = sorted( number for number in (_issue_number(bounty) for bounty in bounties) if number is not None ) @@ -240,13 +201,19 @@ def main(argv: list[str] | None = None) -> int: ) args = parser.parse_args(argv) - data = _load_input(args.input) if args.input else load_live_data(args.repo, args.api_host) + input_arg, repo_arg = validate_source_args( + parser, + input_value=args.input, + repo_value=args.repo, + fix=args.fix, + ) + data = ( + _load_input(input_arg) if input_arg is not None else load_live_data(repo_arg, args.api_host) + ) report = analyze_issue_states(data) if args.fix: - if args.input: - raise SystemExit("--fix requires --repo, not --input") - reopen_violations(args.repo, report["violations"]) - data = load_live_data(args.repo, args.api_host) + reopen_violations(repo_arg, report["violations"]) + data = load_live_data(repo_arg, args.api_host) report = analyze_issue_states(data) if args.format == "json": print(json.dumps(report, indent=2, sort_keys=True)) diff --git a/scripts/check_live_bounty_closing_refs.py b/scripts/check_live_bounty_closing_refs.py index 227ded65..bb0b90ec 100644 --- a/scripts/check_live_bounty_closing_refs.py +++ b/scripts/check_live_bounty_closing_refs.py @@ -2,10 +2,8 @@ import argparse import json -import subprocess +import subprocess # noqa: F401 - monkeypatched in tests import sys -import urllib.error -import urllib.request from pathlib import Path from typing import Any @@ -14,9 +12,11 @@ from scripts.api_host_args import public_api_host from scripts.bounty_refs import GITHUB_CLOSING_ISSUE_RE +from scripts.gh_cli import run_gh_json as _run_gh_json +from scripts.public_api_json import load_public_bounty_list +from scripts.source_args import validate_source_args DEFAULT_API_HOST = "https://api.mrwk.online" -GH_TIMEOUT_SECONDS = 30 GH_PR_SAFETY_CAP = 200 MAX_BOUNTY_REF = 2**63 - 1 @@ -122,47 +122,6 @@ def format_text_report(report: dict[str, Any]) -> str: return "\n".join(lines) -def _run_gh_json(args: list[str]) -> Any: - command = " ".join(args) - try: - completed = subprocess.run( - args, - check=True, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=GH_TIMEOUT_SECONDS, - ) - except subprocess.TimeoutExpired as exc: - raise RuntimeError(f"gh command timed out after {GH_TIMEOUT_SECONDS}s: {command}") 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 json.loads(completed.stdout) - - -def _fetch_json(url: str) -> Any: - request = urllib.request.Request(url, headers={"Accept": "application/json"}) - try: - with urllib.request.urlopen(request, timeout=GH_TIMEOUT_SECONDS) as response: - return json.loads(response.read().decode("utf-8")) - except (TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc: - raise RuntimeError(f"failed to fetch JSON from {url}: {exc}") from exc - - -def _load_public_bounties(api_host: str) -> list[dict[str, Any]]: - url = f"{api_host.rstrip('/')}/api/v1/bounties?status=open&limit=200" - data = _fetch_json(url) - if not isinstance(data, list): - raise RuntimeError(f"expected a JSON list from {url}") - return [item for item in data if isinstance(item, dict)] - - def _load_pull_requests(repo: str, state: str, pr_numbers: list[int]) -> list[dict[str, Any]]: if pr_numbers: return [ @@ -205,7 +164,7 @@ def _load_pull_requests(repo: str, state: str, pr_numbers: list[int]) -> list[di def load_live_data(repo: str, api_host: str, state: str, pr_numbers: list[int]) -> dict[str, Any]: return { - "bounties": _load_public_bounties(api_host), + "bounties": load_public_bounty_list(api_host, query="status=open&limit=200"), "pull_requests": _load_pull_requests(repo, state, pr_numbers), } @@ -235,10 +194,15 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--fail-on-issues", action="store_true") args = parser.parse_args(argv) + input_arg, repo_arg = validate_source_args( + parser, + input_value=args.input, + repo_value=args.repo, + ) data = ( - _load_input(args.input) - if args.input - else load_live_data(args.repo, args.api_host, args.state, args.pr) + _load_input(input_arg) + if input_arg is not None + else load_live_data(repo_arg, args.api_host, args.state, args.pr) ) report = analyze_closing_refs(data) if args.format == "json": 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_api_json.py b/scripts/public_api_json.py new file mode 100644 index 00000000..b546e215 --- /dev/null +++ b/scripts/public_api_json.py @@ -0,0 +1,96 @@ +"""Shared JSON fetch + shape validation for MergeWork public API reads.""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.request +from typing import Any + +DEFAULT_TIMEOUT_SECONDS = 30 + + +def fetch_public_json(url: str, *, timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS) -> Any: + request = urllib.request.Request(url, headers={"Accept": "application/json"}) + try: + with urllib.request.urlopen(request, timeout=timeout_seconds) as response: + return json.loads(response.read().decode("utf-8")) + except (TimeoutError, urllib.error.URLError) as exc: + raise RuntimeError(f"public API request failed: {url}") from exc + except json.JSONDecodeError as exc: + raise RuntimeError(f"public API returned invalid JSON from {url}") from exc + + +def ensure_json_list(data: Any, *, url: str, label: str = "response") -> list[Any]: + if not isinstance(data, list): + raise RuntimeError( + f"expected a JSON list for {label} from {url}, got {type(data).__name__}" + ) + return data + + +def ensure_json_object(data: Any, *, url: str, label: str = "response") -> dict[str, Any]: + if not isinstance(data, dict): + raise RuntimeError( + f"expected a JSON object for {label} from {url}, got {type(data).__name__}" + ) + return data + + +def dict_rows(data: list[Any], *, url: str) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for item in ensure_json_list(data, url=url): + if isinstance(item, dict): + rows.append(item) + return rows + + +def load_public_bounty_list( + api_host: str, *, query: str = "status=open&limit=200" +) -> list[dict[str, Any]]: + url = f"{api_host.rstrip('/')}/api/v1/bounties?{query}" + return dict_rows(fetch_public_json(url), url=url) + + +def validate_public_activity(activity: dict[str, Any], *, url: str) -> dict[str, Any]: + contributors = activity.get("contributors") + recent = activity.get("recent") + if contributors is not None and not isinstance(contributors, list): + raise RuntimeError( + f"expected contributors list from {url}, got {type(contributors).__name__}" + ) + if recent is not None and not isinstance(recent, list): + raise RuntimeError(f"expected recent list from {url}, got {type(recent).__name__}") + return activity + + +def load_public_activity(api_host: str, *, limit: int = 200) -> dict[str, Any]: + url = f"{api_host.rstrip('/')}/api/v1/activity?limit={limit}" + activity = ensure_json_object(fetch_public_json(url), url=url, label="activity") + return validate_public_activity(activity, url=url) + + +def extract_public_api_state(bounties: Any, activity: Any) -> dict[str, Any]: + """Best-effort shape extraction used by claim inventory live loads.""" + data: dict[str, Any] = {} + if isinstance(bounties, list): + data["bounties"] = bounties + if isinstance(activity, dict): + contributors = activity.get("contributors") + if isinstance(contributors, list): + data["contributors"] = contributors + recent = activity.get("recent") + if isinstance(recent, list): + data["recent"] = recent + return data + + +def load_public_api_state(api_host: str, *, limit: int = 200) -> dict[str, Any]: + host = api_host.rstrip("/") + bounties_url = f"{host}/api/v1/bounties?limit={limit}" + activity_url = f"{host}/api/v1/activity?limit={limit}" + bounties = fetch_public_json(bounties_url) + activity = fetch_public_json(activity_url) + if isinstance(activity, dict): + validate_public_activity(activity, url=activity_url) + return extract_public_api_state(bounties, activity) 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_check_bounty_issue_states.py b/tests/test_check_bounty_issue_states.py index 09afe52b..af6f66fb 100644 --- a/tests/test_check_bounty_issue_states.py +++ b/tests/test_check_bounty_issue_states.py @@ -5,6 +5,8 @@ import sys from pathlib import Path +import pytest + from scripts import check_bounty_issue_states from scripts.check_bounty_issue_states import analyze_issue_states, main @@ -109,3 +111,26 @@ def fake_run(args, **kwargs): ) assert calls == [["gh", "issue", "reopen", "936", "--repo", "ramimbo/mergework"]] + + +@pytest.mark.parametrize( + ("source_args", "expected_message"), + ( + (["--input", ""], "must not be empty or whitespace-only"), + (["--input", " "], "must not be empty or whitespace-only"), + (["--repo", ""], "must not be empty or whitespace-only"), + (["--repo", " "], "must not be empty or whitespace-only"), + ), +) +def test_issue_state_check_rejects_empty_source_args( + source_args: list[str], + expected_message: str, + capsys, +) -> None: + with pytest.raises(SystemExit) as exc_info: + main([*source_args, "--format", "json"]) + + assert exc_info.value.code == 2 + captured = capsys.readouterr() + assert expected_message in captured.err + assert captured.out == "" diff --git a/tests/test_check_live_bounty_closing_refs.py b/tests/test_check_live_bounty_closing_refs.py index dfcce5af..92fe6101 100644 --- a/tests/test_check_live_bounty_closing_refs.py +++ b/tests/test_check_live_bounty_closing_refs.py @@ -5,6 +5,8 @@ import sys from pathlib import Path +import pytest + from scripts import check_live_bounty_closing_refs from scripts.check_live_bounty_closing_refs import analyze_closing_refs, main @@ -113,3 +115,26 @@ def fake_run(args, **kwargs): assert prs[0]["number"] == 1015 assert calls[0][:4] == ["gh", "pr", "view", "1015"] + + +@pytest.mark.parametrize( + ("source_args", "expected_message"), + ( + (["--input", ""], "must not be empty or whitespace-only"), + (["--input", " "], "must not be empty or whitespace-only"), + (["--repo", ""], "must not be empty or whitespace-only"), + (["--repo", " "], "must not be empty or whitespace-only"), + ), +) +def test_closing_ref_check_rejects_empty_source_args( + source_args: list[str], + expected_message: str, + capsys, +) -> None: + with pytest.raises(SystemExit) as exc_info: + main([*source_args, "--format", "json"]) + + assert exc_info.value.code == 2 + captured = capsys.readouterr() + assert expected_message in captured.err + assert captured.out == "" 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"])