from __future__ import annotations import json import unittest from datetime import datetime, timedelta, timezone from ai_triage.github import ( GitHubClient, GitHubLogTooLargeError, GitHubPageLimitError, GitHubUnsafeRedirectError, HTTPRequest, HTTPResponse, ) NOW = datetime(2026, 7, 18, 2, 0, tzinfo=timezone.utc) def response(status: int, payload=None, headers=None) -> HTTPResponse: body = b"" if payload is None else json.dumps(payload).encode() return HTTPResponse(status=status, headers=headers or {}, body=body) def run_payload(run_id: int, **overrides: object) -> dict[str, object]: payload: dict[str, object] = { "id": run_id, "status": "completed", "conclusion": "failure", "run_attempt": 2, "html_url": f"https://github.test/runs/{run_id}", "name": "Scenario", "display_title": "Nightly scenario", "created_at": "2026-07-18T01:00:00Z", "updated_at": "2026-07-18T02:00:00Z", } payload.update(overrides) return payload class FakeTransport: def __init__(self, responses: list[HTTPResponse]) -> None: self.responses = list(responses) self.requests: list[HTTPRequest] = [] def request(self, request: HTTPRequest) -> HTTPResponse: self.requests.append(request) if not self.responses: raise AssertionError(f"unexpected request: {request.method} {request.url}") return self.responses.pop(0) class FakeClock: def __init__(self) -> None: self.wall = NOW self.sleeps: list[float] = [] def now(self) -> datetime: return self.wall def sleep(self, seconds: float) -> None: self.sleeps.append(seconds) self.wall += timedelta(seconds=seconds) class GitHubClientTests(unittest.TestCase): def _client(self, responses: list[HTTPResponse], **kwargs): transport = FakeTransport(responses) clock = FakeClock() client = GitHubClient( token="test-token", owner="tidbcloud", repo="tidb-management-service", transport=transport, clock=clock, **kwargs, ) return client, transport, clock def test_lists_completed_runs_with_branch_created_filter_and_timestamps(self) -> None: client, transport, _ = self._client( [response(200, {"workflow_runs": [run_payload(101)]})] ) runs = client.list_completed_runs( "scenario.yaml", branch="main", created_since=datetime(2026, 7, 17, tzinfo=timezone.utc), max_pages=3, ) self.assertEqual(101, runs[0].id) self.assertEqual(NOW, runs[0].updated_at) self.assertEqual(datetime(2026, 7, 18, 1, 0, tzinfo=timezone.utc), runs[0].created_at) self.assertIn("status=completed", transport.requests[0].url) self.assertIn("branch=main", transport.requests[0].url) self.assertIn("created=%3E%3D2026-07-17T00%3A00%3A00Z", transport.requests[0].url) self.assertEqual("GET", transport.requests[0].method) self.assertEqual("2026-03-10", transport.requests[0].headers["X-GitHub-Api-Version"]) def test_omits_branch_filter_when_target_has_no_branch(self) -> None: client, transport, _ = self._client( [response(200, {"workflow_runs": []})] ) client.list_completed_runs( "scenario.yaml", branch=None, created_since=NOW, max_pages=2 ) self.assertNotIn("branch=", transport.requests[0].url) def test_paginates_completed_runs(self) -> None: first_page = [run_payload(run_id) for run_id in range(100)] client, transport, _ = self._client( [ response( 200, {"workflow_runs": first_page}, {"Link": '; rel="next"'}, ), response(200, {"workflow_runs": [run_payload(101)]}), ] ) runs = client.list_completed_runs( "scenario.yaml", branch="main", created_since=NOW, max_pages=2 ) self.assertEqual(101, len(runs)) self.assertIn("page=2", transport.requests[1].url) def test_raises_when_page_cap_cannot_prove_exhaustion(self) -> None: client, _, _ = self._client( [ response( 200, {"workflow_runs": [run_payload(i) for i in range(100)]}, {"Link": '; rel="next"'}, ) ] ) with self.assertRaises(GitHubPageLimitError): client.list_completed_runs( "scenario.yaml", branch=None, created_since=NOW, max_pages=1 ) def test_exact_full_page_without_next_link_is_complete(self) -> None: client, transport, _ = self._client( [response(200, {"workflow_runs": [run_payload(i) for i in range(100)]})] ) runs = client.list_completed_runs( "scenario.yaml", branch=None, created_since=NOW, max_pages=1 ) self.assertEqual(100, len(runs)) self.assertEqual(1, len(transport.requests)) def test_rejects_run_without_valid_timestamps(self) -> None: client, _, _ = self._client( [response(200, {"workflow_runs": [run_payload(1, updated_at="bad")]})] ) with self.assertRaisesRegex(Exception, "updated_at"): client.list_completed_runs( "scenario.yaml", branch=None, created_since=NOW, max_pages=1 ) def test_read_only_request_retries_transient_response(self) -> None: client, transport, clock = self._client( [ response(503, {"message": "down"}), response(200, {"workflow_runs": [run_payload(22)]}), ], read_retry_attempts=2, read_retry_base_seconds=3, ) runs = client.list_completed_runs( "scenario.yaml", branch=None, created_since=NOW, max_pages=1 ) self.assertEqual(22, runs[0].id) self.assertEqual(2, len(transport.requests)) self.assertEqual([3], clock.sleeps) def test_retry_after_is_bounded(self) -> None: client, _, clock = self._client( [ response(429, {"message": "slow down"}, {"Retry-After": "999"}), response(200, {"workflow_runs": []}), ], read_retry_attempts=2, max_retry_delay_seconds=30, ) client.list_completed_runs( "scenario.yaml", branch=None, created_since=NOW, max_pages=1 ) self.assertEqual([30], clock.sleeps) def test_lists_jobs_for_exact_attempt_with_pagination(self) -> None: first_page = [ {"id": value, "name": f"job-{value}", "status": "completed", "conclusion": "success", "steps": []} for value in range(100) ] failed = { "id": 101, "name": "nonprod (case.yaml)", "status": "completed", "conclusion": "failure", "html_url": "https://github.test/jobs/101", "steps": [ {"number": 13, "name": "Run scenario cases", "status": "completed", "conclusion": "failure"} ], } client, transport, _ = self._client( [response(200, {"jobs": first_page}), response(200, {"jobs": [failed]})] ) jobs = client.list_jobs(22, attempt_number=3, max_pages=3) self.assertEqual(101, len(jobs)) self.assertEqual("Run scenario cases", jobs[-1].steps[0].name) self.assertIn("/runs/22/attempts/3/jobs", transport.requests[0].url) self.assertIn("page=2", transport.requests[1].url) def test_job_pagination_stops_at_the_configured_page_cap(self) -> None: full_page = [ { "id": value, "name": f"job-{value}", "status": "completed", "conclusion": "success", "steps": [], } for value in range(100) ] client, transport, _ = self._client( [ response(200, {"jobs": full_page}), response(200, {"jobs": full_page}), ] ) with self.assertRaises(GitHubPageLimitError): client.list_jobs(22, attempt_number=3, max_pages=2) self.assertEqual(2, len(transport.requests)) def test_downloads_bounded_job_log_as_text(self) -> None: client, transport, _ = self._client( [HTTPResponse(200, {}, "log \N{SNOWMAN}".encode())] ) self.assertEqual("log \N{SNOWMAN}", client.download_job_log(101, max_bytes=100)) self.assertEqual(101, transport.requests[0].max_response_bytes) def test_downloads_only_the_requested_job_log_tail_after_redirect(self) -> None: tail = b"x" * 100 client, transport, _ = self._client( [ HTTPResponse( 302, { "Location": ( "https://pipelines.actions.githubusercontent.com/logs/101" ) }, b"", ), HTTPResponse( 200, { "Content-Length": "250", "Accept-Ranges": "bytes", }, b"", ), HTTPResponse( 206, { "Content-Length": "100", "Content-Range": "bytes 150-249/250", }, tail, ), ] ) self.assertTrue(hasattr(client, "download_job_log_tail")) self.assertEqual( tail.decode(), client.download_job_log_tail(101, max_bytes=100), ) self.assertEqual("HEAD", transport.requests[1].method) self.assertEqual("bytes=150-249", transport.requests[2].headers["Range"]) self.assertEqual("identity", transport.requests[2].headers["Accept-Encoding"]) self.assertNotIn("Authorization", transport.requests[1].headers) self.assertNotIn("Authorization", transport.requests[2].headers) def test_job_log_tail_rejects_a_server_that_ignores_the_range(self) -> None: client, _, _ = self._client( [ HTTPResponse( 302, { "Location": ( "https://pipelines.actions.githubusercontent.com/logs/101" ) }, b"", ), HTTPResponse(200, {"Content-Length": "250"}, b""), HTTPResponse(200, {"Content-Length": "250"}, b"x" * 101), ] ) self.assertTrue(hasattr(client, "download_job_log_tail")) with self.assertRaisesRegex(Exception, "range"): client.download_job_log_tail(101, max_bytes=100) def test_job_log_redirect_drops_github_authorization_header(self) -> None: client, transport, _ = self._client( [ HTTPResponse( 302, {"Location": "https://pipelines.actions.githubusercontent.com/logs/101"}, b"", ), HTTPResponse(200, {}, b"bounded log"), ] ) self.assertEqual("bounded log", client.download_job_log(101, max_bytes=100)) self.assertIn("Authorization", transport.requests[0].headers) self.assertNotIn("Authorization", transport.requests[1].headers) def test_job_log_redirect_rejects_non_https_location(self) -> None: client, transport, _ = self._client( [HTTPResponse(302, {"Location": "http://logs.invalid/101"}, b"")] ) with self.assertRaises(GitHubUnsafeRedirectError): client.download_job_log(101, max_bytes=100) self.assertEqual(1, len(transport.requests)) def test_rejects_oversized_job_log(self) -> None: client, _, _ = self._client([HTTPResponse(200, {}, b"12345")]) with self.assertRaises(GitHubLogTooLargeError): client.download_job_log(101, max_bytes=4) if __name__ == "__main__": unittest.main()