mirror of
https://github.com/apache/doris.git
synced 2026-11-04 07:31:17 +00:00
shuke
a2638f38ea
Problem Summary: An empty or malformed JSON error response can make `gh api` fail with only `unexpected end of JSON input`, hiding the HTTP status and GitHub request ID. This prevents distinguishing transient server failures from rejected requests in the review pipeline. Add a shared diagnostic helper for review startup, status synchronization and aggregation, helper downloads, submission, and head polling. Each failure records HTTP status, allowlisted response headers, bounded redacted response/output excerpts, exit code or timeout, and run context. Failure artifacts are uploaded even before PR checkout or when the caller handles the error. Load the helper from the immutable workflow commit before the first API request; distinguish aggregate artifacts by head SHA. Outgoing request headers, bodies, and complete command arguments are excluded. Credential values and fields are redacted before truncation. Response excerpts depend on gh's debug format and are not complete wire captures. Normal stdout, existing request arguments, permissions, review verdicts, submission deduplication, and retry behavior are preserved.
285 lines
15 KiB
Python
285 lines
15 KiB
Python
#!/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()
|