import ast import io import json import os import threading import unittest from contextlib import redirect_stderr, redirect_stdout from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path from unittest.mock import patch from scripts import fetch_star_count TOKEN_SENTINEL = "TOKEN_FETCH_ONLY_DO_NOT_LEAK_7z9" class RecordingHandler(BaseHTTPRequestHandler): def do_GET(self): self.server.requests.append( {"path": self.path, "headers": dict(self.headers.items())} ) response = self.server.response self.send_response(response["status"]) for name, value in response.get("headers", {}).items(): self.send_header(name, value) self.end_headers() try: self.wfile.write(response.get("body", b"")) except BrokenPipeError: pass def log_message(self, *_args): return class LocalHttpServer: def __init__(self, *, status=200, headers=None, body=b""): self.response = { "status": status, "headers": headers or {}, "body": body, } def __enter__(self): self.server = HTTPServer(("127.0.0.1", 0), RecordingHandler) self.server.requests = [] self.server.response = self.response self.thread = threading.Thread( target=self.server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True, ) self.thread.start() host, port = self.server.server_address self.url = f"http://{host}:{port}/repository" return self def __exit__(self, _exc_type, _exc, _traceback): self.server.shutdown() self.server.server_close() self.thread.join(timeout=2) class FetchStarCountTests(unittest.TestCase): def run_main(self, api_url): stdout = io.StringIO() stderr = io.StringIO() with ( patch.object(fetch_star_count, "API_URL", api_url), patch.dict(os.environ, {"GITHUB_TOKEN": TOKEN_SENTINEL}, clear=False), redirect_stdout(stdout), redirect_stderr(stderr), ): exit_code = fetch_star_count.main([]) return exit_code, stdout.getvalue(), stderr.getvalue() def test_fetcher_is_standalone_stdlib_only_and_has_fixed_api(self): source_path = Path(fetch_star_count.__file__) tree = ast.parse(source_path.read_text(encoding="utf-8")) imported_roots = set() for node in ast.walk(tree): if isinstance(node, ast.Import): imported_roots.update(alias.name.split(".", 1)[0] for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: imported_roots.add(node.module.split(".", 1)[0]) self.assertNotIn("scripts", imported_roots) self.assertNotIn("star_history", imported_roots) self.assertEqual( fetch_star_count.API_URL, "https://api.github.com/repos/666ghj/MiroFish", ) def test_success_stdout_is_only_one_decimal_count(self): body = json.dumps({"stargazers_count": 41782}).encode() with LocalHttpServer(body=body) as server: exit_code, stdout, stderr = self.run_main(server.url) self.assertEqual(exit_code, 0) self.assertEqual(stdout, "41782\n") self.assertEqual(stderr, "") self.assertEqual(len(server.server.requests), 1) request = server.server.requests[0] self.assertEqual(request["path"], "/repository") self.assertEqual( request["headers"]["Authorization"], f"Bearer {TOKEN_SENTINEL}" ) def test_redirect_is_refused_without_forwarding_token(self): with LocalHttpServer(body=b'{"stargazers_count": 99}') as target: with LocalHttpServer( status=302, headers={"Location": target.url}, body=f"unsafe-body {TOKEN_SENTINEL}".encode(), ) as source: exit_code, stdout, stderr = self.run_main(source.url) self.assertEqual(exit_code, 1) self.assertEqual(stdout, "") self.assertEqual(target.server.requests, []) self.assertIn("redirect was refused", stderr) self.assertNotIn(TOKEN_SENTINEL, stderr) self.assertNotIn("unsafe-body", stderr) def test_malformed_oversized_and_invalid_counts_are_sanitized(self): bodies = [ b"not-json " + TOKEN_SENTINEL.encode(), b"x" * (fetch_star_count.MAX_HTTP_BYTES + 1), json.dumps([]).encode(), json.dumps({}).encode(), json.dumps({"stargazers_count": True}).encode(), json.dumps({"stargazers_count": -1}).encode(), json.dumps({"stargazers_count": 1.5}).encode(), json.dumps({"stargazers_count": "1"}).encode(), ] for body in bodies: with self.subTest(body_prefix=body[:32]): with LocalHttpServer(body=body) as server: exit_code, stdout, stderr = self.run_main(server.url) self.assertEqual(exit_code, 1) self.assertEqual(stdout, "") self.assertTrue(stderr.startswith("error: ")) self.assertNotIn(TOKEN_SENTINEL, stderr) self.assertNotIn("not-json", stderr) def test_status_and_network_errors_do_not_echo_exception_data(self): with LocalHttpServer( status=500, body=f"unsafe-body {TOKEN_SENTINEL}".encode(), ) as server: exit_code, stdout, stderr = self.run_main(server.url) self.assertEqual(exit_code, 1) self.assertEqual(stdout, "") self.assertIn("unavailable", stderr) self.assertNotIn(TOKEN_SENTINEL, stderr) self.assertNotIn("unsafe-body", stderr) class FailingOpener: def open(self, *_args, **_kwargs): raise OSError(TOKEN_SENTINEL) with self.assertRaises(fetch_star_count.FetchError) as captured: fetch_star_count.fetch_star_count(TOKEN_SENTINEL, FailingOpener()) self.assertNotIn(TOKEN_SENTINEL, str(captured.exception)) def test_missing_or_newline_token_is_rejected_without_stdout(self): for token in ("", "bad\ntoken", "bad\rtoken"): with self.subTest(token=repr(token)): stdout = io.StringIO() stderr = io.StringIO() with ( patch.dict(os.environ, {"GITHUB_TOKEN": token}, clear=False), redirect_stdout(stdout), redirect_stderr(stderr), ): exit_code = fetch_star_count.main([]) self.assertEqual(exit_code, 1) self.assertEqual(stdout.getvalue(), "") if token: self.assertNotIn(token, stderr.getvalue()) def test_workflow_keeps_credentials_out_of_record_and_render_steps(self): repository = Path(__file__).resolve().parents[1] workflow = ( repository / ".github/workflows/update-star-history.yml" ).read_text(encoding="utf-8") renderer = (repository / "scripts/star_history.py").read_text( encoding="utf-8" ) self.assertIn("cron: '17 3 1,16 * *'", workflow) self.assertIn("timezone: 'UTC'", workflow) self.assertNotIn("due-check:", workflow) self.assertNotIn("star_history.py due", workflow) self.assertNotIn("inputs.force", workflow) self.assertNotIn("actions/checkout", workflow) self.assertNotIn("uses:", workflow) self.assertNotIn("pull_request:", workflow) self.assertNotIn("pull_request_target:", workflow) self.assertNotIn("workflow_run:", workflow) self.assertNotIn("secrets.", workflow) self.assertIn("sha256sum --check --strict", workflow) self.assertIn("-c core.hooksPath=/dev/null", workflow) self.assertNotIn("star_history.py sample", workflow) self.assertNotIn('os.environ.get("GITHUB_TOKEN"', renderer) self.assertNotIn('subparsers.add_parser("sample"', renderer) token_steps = [ section for section in workflow.split("\n - name: ") if "GITHUB_TOKEN: ${{ github.token }}" in section ] self.assertGreaterEqual(len(token_steps), 2) for section in token_steps: step_name = section.splitlines()[0] self.assertTrue( step_name.startswith("Fetch aggregate Star count only") or step_name.startswith( "Push one allowlisted commit with an ephemeral credential" ) ) record_steps = [ section for section in workflow.split("\n - name: ") if "star_history.py record" in section ] self.assertEqual(len(record_steps), 1) for section in record_steps: self.assertIn("GITHUB_TOKEN: ''", section) self.assertIn("GH_TOKEN: ''", section) self.assertNotIn("${{ github.token }}", section) self.assertIn("--force", section) if __name__ == "__main__": unittest.main()