From dc183af98eaeb0cdb5546072caf8f04a36e5bb55 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kim=20Harjam=C3=A4ki?= Date: Fri, 3 Jul 2026 14:02:59 +0300 Subject: [PATCH 1/2] fix(agent): remove unused import and fix gh api POST bug --- agent/poll_once.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/agent/poll_once.py b/agent/poll_once.py index 1b9af89..be0f8d4 100644 --- a/agent/poll_once.py +++ b/agent/poll_once.py @@ -2,7 +2,6 @@ import os import re import subprocess -import sys import urllib.error import urllib.parse import urllib.request @@ -66,7 +65,7 @@ def _gh_api_json(path: str, fields: dict[str, str] | None = None, timeout: int = print("Using GITHUB_TOKEN for GitHub API requests.") return _gh_api_http(path, fields=fields, timeout=timeout) print("GITHUB_TOKEN not found, falling back to gh CLI.") - cmd = ["gh", "api", path, "-H", "Accept: application/vnd.github+json"] + cmd = ["gh", "api", path, "-X", "GET", "-H", "Accept: application/vnd.github+json"] if fields: for key, value in fields.items(): cmd.extend(["-f", f"{key}={value}"]) From 8573c82844112ced363acbec1633d0ffb6c01631 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kim=20Harjam=C3=A4ki?= Date: Fri, 3 Jul 2026 17:56:54 +0300 Subject: [PATCH 2/2] test(agent): add gh-api and intake unit tests Regression coverage for the gh api POST bug and intake/fixer branches. Co-Authored-By: Claude Opus 4.8 --- tests/test_gh_api.py | 235 +++++++++++++++++++++++++++++++++++ tests/test_intake_and_run.py | 187 ++++++++++++++++++++++++++++ 2 files changed, 422 insertions(+) create mode 100644 tests/test_gh_api.py create mode 100644 tests/test_intake_and_run.py diff --git a/tests/test_gh_api.py b/tests/test_gh_api.py new file mode 100644 index 0000000..480f900 --- /dev/null +++ b/tests/test_gh_api.py @@ -0,0 +1,235 @@ +"""Unit tests for gh API request construction in agent.poll_once. + +These tests lock down the exact HTTP method, endpoint, query, headers and +gh-CLI command shape so regressions like the "gh api POST bug" +(missing ``-X GET`` on the CLI fallback, which made ``gh api`` default to a +POST for any request carrying ``-f`` fields) are caught before shipping. + +All network / subprocess boundaries are mocked. No real gh or HTTP calls. +""" + +import json +import os +import unittest +import urllib.error +from unittest.mock import MagicMock, patch + +from agent import poll_once + + +class GhCliFallbackCommandTests(unittest.TestCase): + """The gh-CLI fallback path (no token) must build a safe GET command. + + This is the exact regression surface of the prior fix: + ``gh api -f k=v`` implicitly POSTs, so ``-X GET`` must be present. + """ + + def _run_fallback(self, path, fields=None): + """Drive _gh_api_json through the CLI fallback and return the argv used.""" + captured = {} + + def fake_run(cmd, timeout=20): + captured["cmd"] = cmd + captured["timeout"] = timeout + return "[]" + + # No token in env -> forces the gh CLI fallback branch. + with patch.dict(os.environ, {}, clear=True): + with patch("agent.poll_once._run", side_effect=fake_run): + poll_once._gh_api_json(path, fields=fields) + return captured + + def test_fallback_forces_get_method(self) -> None: + # Regression guard: without -X GET, gh treats -f fields as a POST body. + cmd = self._run_fallback("/repos/acme/worker/issues", fields={"state": "open"})["cmd"] + self.assertIn("-X", cmd) + method_value = cmd[cmd.index("-X") + 1] + self.assertEqual(method_value, "GET") + + def test_fallback_never_uses_a_mutating_method(self) -> None: + cmd = self._run_fallback("/repos/acme/worker/issues", fields={"state": "open"})["cmd"] + for mutating in ("POST", "PUT", "PATCH", "DELETE"): + self.assertNotIn(mutating, cmd) + + def test_fallback_command_prefix_and_endpoint(self) -> None: + cmd = self._run_fallback("/repos/acme/worker/issues")["cmd"] + self.assertEqual(cmd[0], "gh") + self.assertEqual(cmd[1], "api") + self.assertEqual(cmd[2], "/repos/acme/worker/issues") + + def test_fallback_sets_accept_header(self) -> None: + cmd = self._run_fallback("/repos/acme/worker/issues")["cmd"] + self.assertIn("-H", cmd) + header_value = cmd[cmd.index("-H") + 1] + self.assertEqual(header_value, "Accept: application/vnd.github+json") + + def test_fallback_serializes_fields_as_f_pairs(self) -> None: + cmd = self._run_fallback( + "/repos/acme/worker/issues", + fields={"state": "open", "labels": "queued", "per_page": "50"}, + )["cmd"] + # Each field becomes a "-f key=value" pair, in insertion order. + self.assertIn("-f", cmd) + pairs = [cmd[i + 1] for i, tok in enumerate(cmd) if tok == "-f"] + self.assertEqual(pairs, ["state=open", "labels=queued", "per_page=50"]) + self.assertEqual(cmd.count("-f"), 3) + + def test_fallback_without_fields_has_no_f_flag(self) -> None: + cmd = self._run_fallback("/repos/acme/worker/issues")["cmd"] + self.assertNotIn("-f", cmd) + + def test_fallback_parses_json_output(self) -> None: + with patch.dict(os.environ, {}, clear=True): + with patch("agent.poll_once._run", return_value=json.dumps([{"number": 7}])): + result = poll_once._gh_api_json("/x") + self.assertEqual(result, [{"number": 7}]) + + def test_fallback_empty_output_is_none(self) -> None: + with patch.dict(os.environ, {}, clear=True): + with patch("agent.poll_once._run", return_value=""): + self.assertIsNone(poll_once._gh_api_json("/x")) + + def test_fallback_invalid_json_raises_runtime_error(self) -> None: + with patch.dict(os.environ, {}, clear=True): + with patch("agent.poll_once._run", return_value="{not json"): + with self.assertRaisesRegex(RuntimeError, "invalid JSON"): + poll_once._gh_api_json("/x") + + +class TokenRoutingTests(unittest.TestCase): + """_gh_api_json must route to HTTP when a token exists, CLI otherwise.""" + + def test_token_present_uses_http_not_cli(self) -> None: + with patch.dict(os.environ, {"GITHUB_TOKEN": "t"}, clear=True): + with patch("agent.poll_once._gh_api_http", return_value=[1]) as http, \ + patch("agent.poll_once._run") as run: + self.assertEqual(poll_once._gh_api_json("/x", fields={"a": "b"}), [1]) + http.assert_called_once_with("/x", fields={"a": "b"}, timeout=20) + run.assert_not_called() + + def test_gh_token_env_also_selects_http(self) -> None: + with patch.dict(os.environ, {"GH_TOKEN": "t"}, clear=True): + with patch("agent.poll_once._gh_api_http", return_value=[]) as http, \ + patch("agent.poll_once._run") as run: + poll_once._gh_api_json("/x") + http.assert_called_once() + run.assert_not_called() + + def test_no_token_uses_cli_not_http(self) -> None: + with patch.dict(os.environ, {}, clear=True): + with patch("agent.poll_once._gh_api_http") as http, \ + patch("agent.poll_once._run", return_value="[]") as run: + poll_once._gh_api_json("/x") + run.assert_called_once() + http.assert_not_called() + + +class HttpRequestConstructionTests(unittest.TestCase): + """The urllib HTTP path must build a correct GET request with auth headers.""" + + def _capture_request(self, path, fields=None): + captured = {} + + def fake_urlopen(req, timeout=20): + captured["req"] = req + captured["timeout"] = timeout + resp = MagicMock() + resp.read.return_value = b"[]" + ctx = MagicMock() + ctx.__enter__.return_value = resp + return ctx + + with patch.dict(os.environ, {"GITHUB_TOKEN": "secret-token"}, clear=True): + with patch("agent.poll_once.urllib.request.urlopen", side_effect=fake_urlopen): + poll_once._gh_api_http(path, fields=fields) + return captured["req"] + + def test_builds_expected_base_url(self) -> None: + req = self._capture_request("/repos/acme/worker/issues") + self.assertEqual(req.full_url, "https://api.github.com/repos/acme/worker/issues") + + def test_encodes_fields_into_query_string(self) -> None: + req = self._capture_request( + "/repos/acme/worker/issues", + fields={"state": "open", "labels": "queued", "per_page": "50"}, + ) + self.assertEqual( + req.full_url, + "https://api.github.com/repos/acme/worker/issues" + "?state=open&labels=queued&per_page=50", + ) + + def test_no_fields_produces_no_query(self) -> None: + req = self._capture_request("/repos/acme/worker/issues") + self.assertNotIn("?", req.full_url) + + def test_uses_get_method(self) -> None: + req = self._capture_request("/x", fields={"a": "b"}) + # urllib infers GET when no data payload is attached; assert we never + # attach a body (which would flip the method to POST). + self.assertIsNone(req.data) + self.assertEqual(req.get_method(), "GET") + + def test_sets_required_headers_including_bearer_auth(self) -> None: + req = self._capture_request("/x") + # urllib capitalizes header keys. + self.assertEqual(req.get_header("Accept"), "application/vnd.github+json") + self.assertEqual(req.get_header("X-github-api-version"), "2022-11-28") + self.assertEqual(req.get_header("Authorization"), "Bearer secret-token") + + def test_missing_token_raises_before_any_request(self) -> None: + with patch.dict(os.environ, {}, clear=True): + with patch("agent.poll_once.urllib.request.urlopen") as urlopen: + with self.assertRaisesRegex(RuntimeError, "not available"): + poll_once._gh_api_http("/x") + urlopen.assert_not_called() + + +class HttpErrorHandlingTests(unittest.TestCase): + """HTTP failures must be surfaced as RuntimeError with useful context.""" + + def test_http_error_includes_status_and_body(self) -> None: + err = urllib.error.HTTPError( + url="https://api.github.com/x", + code=403, + msg="Forbidden", + hdrs=None, + fp=None, + ) + err.read = MagicMock(return_value=b'{"message":"rate limited"}') + with patch.dict(os.environ, {"GITHUB_TOKEN": "t"}, clear=True): + with patch("agent.poll_once.urllib.request.urlopen", side_effect=err): + with self.assertRaisesRegex(RuntimeError, "GitHub API error 403"): + poll_once._gh_api_http("/x") + + def test_url_error_is_wrapped(self) -> None: + with patch.dict(os.environ, {"GITHUB_TOKEN": "t"}, clear=True): + with patch( + "agent.poll_once.urllib.request.urlopen", + side_effect=urllib.error.URLError("dns down"), + ): + with self.assertRaisesRegex(RuntimeError, "request failed: dns down"): + poll_once._gh_api_http("/x") + + def test_invalid_json_body_is_wrapped(self) -> None: + resp = MagicMock() + resp.read.return_value = b"not json" + ctx = MagicMock() + ctx.__enter__.return_value = resp + with patch.dict(os.environ, {"GITHUB_TOKEN": "t"}, clear=True): + with patch("agent.poll_once.urllib.request.urlopen", return_value=ctx): + with self.assertRaisesRegex(RuntimeError, "invalid JSON"): + poll_once._gh_api_http("/x") + + def test_empty_body_returns_none(self) -> None: + resp = MagicMock() + resp.read.return_value = b"" + ctx = MagicMock() + ctx.__enter__.return_value = resp + with patch.dict(os.environ, {"GITHUB_TOKEN": "t"}, clear=True): + with patch("agent.poll_once.urllib.request.urlopen", return_value=ctx): + self.assertIsNone(poll_once._gh_api_http("/x")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_intake_and_run.py b/tests/test_intake_and_run.py new file mode 100644 index 0000000..b707151 --- /dev/null +++ b/tests/test_intake_and_run.py @@ -0,0 +1,187 @@ +"""Unit tests for the failure-intake / fixer decision logic and the +subprocess wrapper in agent.poll_once. + +`main()` is the intake path: it resolves the target repo, pulls the queued +issue backlog from GitHub, and decides which items are actionable failures +(open issues labelled "queued", excluding pull requests). These tests cover +the branch decisions and error propagation. `_run` is the subprocess boundary +every gh CLI call flows through; its failure translation is asserted here. + +All subprocess / API boundaries are mocked. No real gh or HTTP calls. +""" + +import io +import os +import subprocess +import unittest +from unittest.mock import MagicMock, patch + +from agent import poll_once + + +def _issue(number, title="t", url="u", **extra): + data = {"number": number, "title": title, "html_url": url} + data.update(extra) + return data + + +class MainIntakeQueryTests(unittest.TestCase): + """main() must request the correct queued-issue backlog for the target repo.""" + + def test_requests_open_queued_issues_for_resolved_repo(self) -> None: + with patch.dict(os.environ, {"GITHUB_REPOSITORY": "acme/worker"}, clear=True): + with patch("agent.poll_once._gh_api_json", return_value=[]) as api, \ + patch("sys.stdout", io.StringIO()): + self.assertEqual(poll_once.main(), 0) + api.assert_called_once_with( + "/repos/acme/worker/issues", + fields={"state": "open", "labels": "queued", "per_page": "50"}, + ) + + def test_falls_back_to_default_owner_repo(self) -> None: + with patch.dict(os.environ, {}, clear=True): + with patch("agent.poll_once._gh_api_json", return_value=[]) as api, \ + patch("sys.stdout", io.StringIO()): + poll_once.main() + called_path = api.call_args.args[0] + self.assertEqual(called_path, "/repos/Coding-Autopilot-System/ci-autopilot/issues") + + +class MainIntakeFilteringTests(unittest.TestCase): + """The fixer only acts on real issues; PRs and malformed items are excluded.""" + + def _run_main(self, payload): + out = io.StringIO() + with patch.dict(os.environ, {"GITHUB_REPOSITORY": "acme/worker"}, clear=True): + with patch("agent.poll_once._gh_api_json", return_value=payload), \ + patch("sys.stdout", out): + rc = poll_once.main() + return rc, out.getvalue() + + def test_excludes_pull_requests_from_queue(self) -> None: + payload = [ + _issue(1), + _issue(2, pull_request={"url": "x"}), # a PR, must be filtered out + _issue(3), + ] + rc, out = self._run_main(payload) + self.assertEqual(rc, 0) + self.assertIn("Found 2 queued issues", out) + + def test_ignores_non_dict_entries(self) -> None: + payload = [_issue(1), "junk", None, 42, _issue(2)] + rc, out = self._run_main(payload) + self.assertEqual(rc, 0) + self.assertIn("Found 2 queued issues", out) + + def test_empty_list_reports_zero(self) -> None: + rc, out = self._run_main([]) + self.assertEqual(rc, 0) + self.assertIn("Found 0 queued issues", out) + + def test_none_response_treated_as_empty(self) -> None: + rc, out = self._run_main(None) + self.assertEqual(rc, 0) + self.assertIn("Found 0 queued issues", out) + + def test_only_first_five_issues_are_detailed(self) -> None: + payload = [_issue(n, title=f"issue-{n}") for n in range(1, 9)] + rc, out = self._run_main(payload) + self.assertEqual(rc, 0) + self.assertIn("Found 8 queued issues", out) + self.assertIn("issue-5", out) + self.assertNotIn("issue-6", out) + + def test_issue_detail_lines_include_number_and_url(self) -> None: + rc, out = self._run_main([_issue(42, title="broken build", url="http://gh/42")]) + self.assertEqual(rc, 0) + self.assertIn("#42 broken build", out) + self.assertIn("http://gh/42", out) + + +class MainErrorHandlingTests(unittest.TestCase): + """main() must fail closed (return 1) on bad config, API errors, or bad shapes.""" + + def test_invalid_repo_env_returns_one(self) -> None: + with patch.dict(os.environ, {"GITHUB_REPOSITORY": "acme/worker/extra"}, clear=True): + with patch("agent.poll_once._gh_api_json") as api, \ + patch("sys.stdout", io.StringIO()): + self.assertEqual(poll_once.main(), 1) + api.assert_not_called() # never queries GitHub with a bad target + + def test_api_runtime_error_returns_one(self) -> None: + with patch.dict(os.environ, {"GITHUB_REPOSITORY": "acme/worker"}, clear=True): + with patch( + "agent.poll_once._gh_api_json", + side_effect=RuntimeError("boom"), + ), patch("sys.stdout", io.StringIO()) as out: + self.assertEqual(poll_once.main(), 1) + self.assertIn("boom", out.getvalue()) + + def test_unexpected_dict_response_returns_one(self) -> None: + with patch.dict(os.environ, {"GITHUB_REPOSITORY": "acme/worker"}, clear=True): + with patch( + "agent.poll_once._gh_api_json", + return_value={"message": "Not Found"}, + ), patch("sys.stdout", io.StringIO()): + self.assertEqual(poll_once.main(), 1) + + +class RepoResolutionTests(unittest.TestCase): + """Repo-segment validation guards against path injection into API URLs.""" + + def test_owner_repo_split_from_env(self) -> None: + with patch.dict(os.environ, {"GITHUB_OWNER": "o", "GITHUB_REPO": "r"}, clear=True): + self.assertEqual(poll_once._repo_from_env(), ("o", "r")) + + def test_rejects_slash_in_segment(self) -> None: + with self.assertRaisesRegex(RuntimeError, "Invalid GitHub owner"): + poll_once._validate_repo_segment("a/b", "owner") + + def test_rejects_empty_segment(self) -> None: + with self.assertRaisesRegex(RuntimeError, "Invalid GitHub repository"): + poll_once._validate_repo_segment(" ", "repository") + + def test_accepts_dotted_and_dashed_names(self) -> None: + self.assertEqual(poll_once._validate_repo_segment("my-repo.v2_x", "repository"), "my-repo.v2_x") + + +class RunSubprocessTests(unittest.TestCase): + """_run wraps subprocess and translates failures into RuntimeError.""" + + def test_returns_stripped_stdout_on_success(self) -> None: + proc = MagicMock(returncode=0, stdout=" hello \n", stderr="") + with patch("agent.poll_once.subprocess.run", return_value=proc): + self.assertEqual(poll_once._run(["gh", "api", "/x"]), "hello") + + def test_nonzero_exit_raises_with_stderr(self) -> None: + proc = MagicMock(returncode=1, stdout="", stderr="gh: HTTP 404") + with patch("agent.poll_once.subprocess.run", return_value=proc): + with self.assertRaisesRegex(RuntimeError, "Command failed"): + poll_once._run(["gh", "api", "/missing"]) + + def test_missing_binary_raises_command_not_found(self) -> None: + with patch("agent.poll_once.subprocess.run", side_effect=FileNotFoundError()): + with self.assertRaisesRegex(RuntimeError, "Command not found: gh"): + poll_once._run(["gh", "api", "/x"]) + + def test_timeout_is_translated(self) -> None: + with patch( + "agent.poll_once.subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="gh", timeout=20), + ): + with self.assertRaisesRegex(RuntimeError, "timed out after 20s"): + poll_once._run(["gh", "api", "/x"], timeout=20) + + def test_passes_capture_and_text_flags(self) -> None: + proc = MagicMock(returncode=0, stdout="ok", stderr="") + with patch("agent.poll_once.subprocess.run", return_value=proc) as run: + poll_once._run(["gh", "--version"], timeout=5) + kwargs = run.call_args.kwargs + self.assertTrue(kwargs["capture_output"]) + self.assertTrue(kwargs["text"]) + self.assertEqual(kwargs["timeout"], 5) + + +if __name__ == "__main__": + unittest.main()