| #!/usr/bin/env python3 | |
| # Licensed to the Apache Software Foundation (ASF) under one | |
| # or more contributor license agreements. See the NOTICE file | |
| # distributed with this work for additional information | |
| # regarding copyright ownership. The ASF licenses this file | |
| # to you under the Apache License, Version 2.0 (the | |
| # "License"); you may not use this file except in compliance | |
| # with the License. You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, | |
| # software distributed under the License is distributed on an | |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| # KIND, either express or implied. See the License for the | |
| # specific language governing permissions and limitations | |
| # under the License. | |
| import http.server | |
| import io | |
| import json | |
| import os | |
| import re | |
| import shutil | |
| import subprocess | |
| import sys | |
| import threading | |
| import tempfile | |
| import unittest | |
| from contextlib import redirect_stderr | |
| from pathlib import Path | |
| from unittest import mock | |
| import review_github_api as diagnostic | |
| def trace(status=502, body='{"message":', path="/repos/apache/doris/statuses/abc"): | |
| return ( | |
| "* Request at 2026-10-21\n" | |
| "* Request to https://api.github.com" + path + "\n" | |
| "> POST " + path + " HTTP/2.0\n" | |
| "> Authorization: token request-secret\n" | |
| "> Cookie: request-cookie\n\n" | |
| '{"private_request":"DO_NOT_PERSIST_REQUEST_PAYLOAD"}\n\n' | |
| f"< HTTP/2.0 {status} Example\n" | |
| "< Content-Type: application/json\n" | |
| "< X-Github-Request-Id: REQUEST-123\n" | |
| "< X-Ratelimit-Remaining: 0\n" | |
| "< Retry-After: 10\n" | |
| "< Set-Cookie: response-cookie\n" | |
| "< Authorization: Bearer response-secret\n\n" | |
| + body + "\n* Request took 1ms\n" | |
| ) | |
| class GitHubDiagnosticsTest(unittest.TestCase): | |
| def setUp(self): | |
| temp = tempfile.TemporaryDirectory() | |
| self.addCleanup(temp.cleanup) | |
| self.directory = Path(temp.name) | |
| self.env = self.enterContext(mock.patch.dict(os.environ, { | |
| "REVIEW_GITHUB_DIAGNOSTICS_DIR": temp.name, | |
| "GH_TOKEN": "env-github-secret", | |
| "LANGFUSE_SECRET_KEY": "env-telemetry-secret", | |
| })) | |
| self.logs = self.enterContext(redirect_stderr(io.StringIO())) | |
| self.command = ["gh", "api", "repos/apache/doris/statuses/abc", | |
| "-X", "POST", "-f", "sensitive=COMMAND_SECRET"] | |
| def records(self): | |
| return [json.loads(p.read_text()) for p in sorted(self.directory.glob("*.json"))] | |
| def test_malformed_http_error_retains_status_headers_body_and_exit(self): | |
| error = subprocess.CalledProcessError(1, self.command, "", trace() + "unexpected end of JSON input\n") | |
| with mock.patch.object(diagnostic.subprocess, "run", side_effect=error) as run: | |
| with self.assertRaises(subprocess.CalledProcessError) as result: | |
| diagnostic.run_gh(self.command, capture_output=True, text=True, check=True, timeout=10) | |
| self.assertIs(error, result.exception) | |
| self.assertEqual(1, run.call_count) # Never replay a write for diagnostics. | |
| self.assertEqual("api", run.call_args.kwargs["env"]["GH_DEBUG"]) | |
| record, = self.records() | |
| self.assertEqual(502, record["http"]["status"]) | |
| self.assertEqual('{"message":', record["http"]["body_excerpt"]) | |
| self.assertEqual("REQUEST-123", record["http"]["headers"]["x-github-request-id"]) | |
| self.assertEqual("10", record["http"]["headers"]["retry-after"]) | |
| self.assertEqual(1, record["returncode"]) | |
| self.assertIn("unexpected end of JSON input", error.stderr) | |
| data = json.dumps(record) | |
| for secret in ("COMMAND_SECRET", "request-secret", "request-cookie", | |
| "response-cookie", "response-secret", "DO_NOT_PERSIST_REQUEST_PAYLOAD"): | |
| self.assertNotIn(secret, data) | |
| self.assertEqual(0o600, next(self.directory.glob("*.json")).stat().st_mode & 0o777) | |
| def test_success_stdout_and_pagination_are_unchanged_and_not_archived(self): | |
| output = '[[{"id":1}],[{"id":2}]]\n' | |
| response = subprocess.CompletedProcess(self.command, 0, output, trace(200, '{"id":1}')+trace(200, '{"id":2}')) | |
| with mock.patch.object(diagnostic.subprocess, "run", return_value=response): | |
| actual = diagnostic.run_gh(self.command, check=True, capture_output=True, text=True) | |
| self.assertEqual(output, actual.stdout) | |
| self.assertEqual("", actual.stderr) | |
| self.assertEqual(2, actual.github_response["request_count"]) | |
| self.assertEqual([], self.records()) | |
| self.assertEqual("", self.logs.getvalue()) | |
| def test_later_page_failure_has_last_request_not_previous_body(self): | |
| stderr = trace(200, '{"previous":"large successful body"}') + trace(503, "") | |
| error = subprocess.CalledProcessError(1, self.command, "partial stdout", stderr+"unexpected end of JSON input") | |
| with mock.patch.object(diagnostic.subprocess, "run", side_effect=error): | |
| with self.assertRaises(subprocess.CalledProcessError): | |
| diagnostic.run_gh(self.command, check=True) | |
| http = self.records()[0]["http"] | |
| self.assertEqual((503, 2, ""), (http["status"], http["request_count"], http["body_excerpt"])) | |
| def test_timeout_keeps_partial_response_and_does_not_retry(self): | |
| error = subprocess.TimeoutExpired(self.command, 10, output=b"", stderr=trace(502).encode()) | |
| with mock.patch.object(diagnostic.subprocess, "run", side_effect=error) as run: | |
| with self.assertRaises(subprocess.TimeoutExpired): | |
| diagnostic.run_gh(self.command, check=True, timeout=10) | |
| record, = self.records() | |
| self.assertEqual(10, record["timeout_seconds"]) | |
| self.assertEqual(502, record["http"]["status"]) | |
| self.assertEqual(1, run.call_count) | |
| def test_transport_failure_without_response_does_not_claim_http_status(self): | |
| stderr = trace().split("< HTTP/",1)[0] + "* Request took 1ms\nPost: connection reset by peer\n" | |
| error = subprocess.CalledProcessError(1, self.command, "", stderr) | |
| with mock.patch.object(diagnostic.subprocess, "run", side_effect=error): | |
| with self.assertRaises(subprocess.CalledProcessError): | |
| diagnostic.run_gh(self.command, check=True) | |
| record, = self.records() | |
| self.assertIsNone(record["http"]["status"]) | |
| self.assertEqual("", record["http"]["body_excerpt"]) | |
| self.assertIn("connection reset", record["stderr"]) | |
| def test_json_decode_failure_keeps_success_status_and_raw_output_excerpt(self): | |
| response = subprocess.CompletedProcess(self.command, 0, "{", trace(201, "{")) | |
| with mock.patch.object(diagnostic.subprocess, "run", return_value=response): | |
| result = diagnostic.run_gh(self.command, check=True) | |
| with self.assertRaises(json.JSONDecodeError): | |
| diagnostic.decode_json(result, self.command) | |
| record, = self.records() | |
| self.assertEqual(201, record["http"]["status"]) | |
| self.assertEqual("JSONDecodeError", record["error_type"]) | |
| self.assertEqual("{", record["stdout_excerpt"]) | |
| def test_credential_echoes_are_redacted_in_artifacts_and_summary(self): | |
| body = '{"access_token":"not-in-env","password":"another-secret","message":"env-github-secret env-telemetry-secret ghp_ExampleToken"}' | |
| error = subprocess.CalledProcessError(1, self.command, body, trace(403,body)+"env-github-secret") | |
| with mock.patch.object(diagnostic.subprocess, "run", side_effect=error): | |
| with self.assertRaises(subprocess.CalledProcessError): | |
| diagnostic.run_gh(self.command, check=True) | |
| data = json.dumps(self.records()) + self.logs.getvalue() | |
| for secret in ("not-in-env","another-secret","env-github-secret","env-telemetry-secret","ghp_ExampleToken"): | |
| self.assertNotIn(secret, data) | |
| def test_excerpt_is_bounded_and_paths_drop_sensitive_queries(self): | |
| http, _ = diagnostic.split_trace(trace(500, "x"*100000, "/repos/a/b?token=secret")) | |
| self.assertEqual("/repos/a/b", http["path"]) | |
| self.assertLessEqual(len(http["body_excerpt"]), diagnostic.LIMIT + 20) | |
| def test_later_failure_cannot_overwrite_earlier_evidence(self): | |
| for status in (502,403): | |
| error = subprocess.CalledProcessError(1, self.command, "", trace(status)) | |
| with mock.patch.object(diagnostic.subprocess, "run", side_effect=error): | |
| with self.assertRaises(subprocess.CalledProcessError): | |
| diagnostic.run_gh(self.command, check=True) | |
| self.assertEqual([502,403], [r["http"]["status"] for r in self.records()]) | |
| def test_artifact_write_failure_does_not_replace_original_error(self): | |
| error = subprocess.CalledProcessError(7, self.command, "", trace()) | |
| with mock.patch.object(diagnostic.subprocess, "run", side_effect=error), \ | |
| mock.patch.object(Path, "mkdir", side_effect=PermissionError): | |
| with self.assertRaises(subprocess.CalledProcessError) as result: | |
| diagnostic.run_gh(self.command, check=True) | |
| self.assertEqual(7, result.exception.returncode) | |
| self.assertIn("Could not persist", self.logs.getvalue()) | |
| def test_unknown_trace_format_is_not_dumped_as_request_data(self): | |
| http, ordinary = diagnostic.split_trace("* Request at now\nSECRET REQUEST CONTENT\n") | |
| self.assertEqual("", http["body_excerpt"]) | |
| self.assertEqual("", ordinary) | |
| def test_redaction_precedes_body_truncation(self): | |
| prefix = "x" * (diagnostic.LIMIT - 4) | |
| http, _ = diagnostic.split_trace(trace(500, prefix + "env-github-secret")) | |
| self.assertNotIn("env-", http["body_excerpt"]) | |
| def test_json_escaped_credentials_are_fully_redacted(self): | |
| raw = json.dumps({"password": 'a"b,secret', "id_token": "ID_SECRET", "api_key": "KEY_SECRET"}) | |
| output = diagnostic.redact(raw) | |
| for secret in ("secret", "ID_SECRET", "KEY_SECRET"): | |
| self.assertNotIn(secret, output) | |
| json.loads(output) | |
| def test_cli_signal_exit_uses_shell_convention(self): | |
| error = subprocess.CalledProcessError(-15, self.command, "", "terminated") | |
| with mock.patch.object(diagnostic, "run_gh", side_effect=error): | |
| self.assertEqual(143, diagnostic.main()) | |
| @unittest.skipUnless(os.environ.get("RUN_GH_HTTP_TESTS") == "1" and shutil.which("gh"), | |
| "opt-in real gh loopback test") | |
| class GitHubHTTPLoopbackTest(unittest.TestCase): | |
| def test_real_cli_failures_keep_http_evidence_without_reposting(self): | |
| requests = [] | |
| class Handler(http.server.BaseHTTPRequestHandler): | |
| def log_message(self, *_): | |
| pass | |
| def do_POST(self): | |
| requests.append(json.loads(self.rfile.read(int(self.headers["Content-Length"])))) | |
| status = int(self.path.rsplit("/", 1)[1]) | |
| body = b'{"id":1}' if status == 201 else b'' | |
| self.send_response(status) | |
| self.send_header("Content-Type", "application/json") | |
| self.send_header("Content-Length", str(len(body))) | |
| self.send_header("X-GitHub-Request-Id", "loopback-id") | |
| self.send_header("Retry-After", "12") | |
| self.end_headers() | |
| self.wfile.write(body) | |
| server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) | |
| thread = threading.Thread(target=server.serve_forever, daemon=True) | |
| thread.start() | |
| try: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| env = {"PATH": os.environ["PATH"], "GH_TOKEN": "fake-only", | |
| "GH_ENTERPRISE_TOKEN": "fake-only", "GH_CONFIG_DIR": tmp, | |
| "GH_NO_UPDATE_NOTIFIER": "1", "GH_PROMPT_DISABLED": "1", | |
| "REVIEW_GITHUB_DIAGNOSTICS_DIR": tmp} | |
| helper = Path(__file__).with_name("review_github_api.py") | |
| for status in (201,422,502,503): | |
| result = subprocess.run([ | |
| sys.executable, str(helper), "api", | |
| f"http://127.0.0.1:{server.server_port}/statuses/{status}", | |
| "-X", "POST", "-f", "state=pending", | |
| ], env=env, capture_output=True, text=True, timeout=10) | |
| self.assertEqual(0 if status == 201 else 1, result.returncode) | |
| if status == 201: | |
| self.assertEqual('{"id":1}', result.stdout) | |
| self.assertEqual("", result.stderr) | |
| else: | |
| self.assertEqual("", result.stdout) | |
| self.assertIn("unexpected end of JSON input", result.stderr) | |
| self.assertIn(f"HTTP={status}", result.stderr) | |
| records = [json.loads(p.read_text()) for p in sorted(Path(tmp).glob("*.json"))] | |
| self.assertEqual([422,502,503], [r["http"]["status"] for r in records]) | |
| self.assertTrue(all(r["http"]["headers"]["x-github-request-id"] == "loopback-id" for r in records)) | |
| self.assertEqual([{"state":"pending"}]*4, requests) | |
| self.assertNotIn("fake-only", json.dumps(records)) | |
| finally: | |
| server.shutdown() | |
| server.server_close() | |
| thread.join() | |
| class WorkflowDiagnosticsWiringTest(unittest.TestCase): | |
| def test_every_job_installs_before_api_and_uploads_after_cleanup(self): | |
| workflows = Path(__file__).resolve().parents[1] / "workflows" | |
| for filename, jobs in ( | |
| ("code-review-runner.yml", ["code-review"]), | |
| ("code-review-aggregate-status.yml", ["aggregate"]), | |
| ("code-review-sync-result.yml", ["accept-skill-review", "skip-on-comment"]), | |
| ): | |
| text = (workflows / filename).read_text() | |
| for job in jobs: | |
| block = text.split("\n " + job + ":", 1)[1] | |
| block = re.split(r"\n [\w-]+:", block, maxsplit=1)[0] | |
| self.assertLess(block.index("Install GitHub API diagnostics"), | |
| block.index('python3 "$REVIEW_GH_HELPER"')) | |
| upload = block.split(" - name: Preserve GitHub API failure diagnostics",1)[1] | |
| self.assertIn("if: always()", upload) | |
| self.assertIn("continue-on-error: true", upload) | |
| self.assertIn("review-github-diagnostics/*.json", upload) | |
| self.assertNotIn(" - name:", upload) | |
| self.assertNotIn("gh api ", block) | |
| aggregate = (workflows / "code-review-aggregate-status.yml").read_text() | |
| # Two reusable aggregate jobs can run for current and previous heads. | |
| self.assertIn("github-api-errors-aggregate-${{ inputs.head_sha }}-", aggregate) | |
| if __name__ == "__main__": | |
| unittest.main() |