#!/usr/bin/env python3 """Offline behavior checks for the production CLI; no GitHub or runner access.""" from contextlib import redirect_stdout from copy import deepcopy import importlib.util import io import json import os from pathlib import Path import subprocess import tempfile import unittest from unittest.mock import patch _spec = importlib.util.spec_from_file_location( "unit_reuse", Path(__file__).with_name("push-main-unit-reuse.py")) reuse = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(reuse) REPO = "BigscreenVR/spt-bs-core" SHA = "a" * 40 ROOT = f"repos/{REPO}/actions" RUN = f"{ROOT}/runs/73" RUNS = f"{ROOT}/workflows/42/runs?event=pull_request&head_sha={SHA}&per_page=100&page=1" JOBS = f"{RUN}/attempts/2/jobs?per_page=100&page=1" ENV = { "GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/heads/main", "GITHUB_REPOSITORY": REPO, "GITHUB_REPOSITORY_ID": "123", "GITHUB_SHA": SHA, "GITHUB_SERVER_URL": "https://github.com", "GH_TOKEN": "fixture-secret-not-for-logs", } def proof(): repository = {"id": 123, "full_name": REPO} run = { "id": 73, "run_attempt": 2, "workflow_id": 42, "path": ".github/workflows/ci.yml", "event": "pull_request", "head_sha": SHA, "repository": repository, "head_repository": repository, "status": "completed", "conclusion": "success", } jobs = [ {"id": 100 + index, "run_id": 73, "run_attempt": 2, "head_sha": SHA, "name": name, "status": "completed", "conclusion": "success"} for index, name in enumerate(reuse.UNIT_JOBS) ] return { f"{ROOT}/workflows/ci.yml": {"id": 42, "name": "ci", "path": reuse.WORKFLOW}, RUNS: {"total_count": 1, "workflow_runs": [deepcopy(run)]}, RUN: deepcopy(run), f"{RUN}/attempts/2": deepcopy(run), JOBS: {"total_count": 2, "jobs": jobs}, } # [unit->REQ-CI-PUSH-MAIN-UNIT-REUSE] class ReuseTests(unittest.TestCase): def invoke(self, responses, run_unit, reason, env=None): """Run the real main/get/pagination path; only gh's I/O is replaced.""" calls = [] missing = [] def gh(argv, **kwargs): endpoint = argv[-1] calls.append(endpoint) if endpoint not in responses: missing.append(endpoint) raise OSError("unprovided fixture") value = responses[endpoint] if isinstance(value, list): value = value.pop(0) if isinstance(value, Exception): raise value if isinstance(value, subprocess.CompletedProcess): return value return subprocess.CompletedProcess(argv, 0, json.dumps(value), "") with tempfile.TemporaryDirectory() as directory: output = Path(directory) / "output" environment = dict(ENV, GITHUB_OUTPUT=str(output)) environment.update(env or {}) stdout = io.StringIO() with patch.dict(os.environ, environment, clear=True), \ patch.object(reuse.subprocess, "run", side_effect=gh), redirect_stdout(stdout): self.assertEqual(reuse.main(), 0) self.assertEqual(missing, [], "fixture must cover every requested endpoint") self.assertEqual(output.read_text(), f"run-unit={str(run_unit).lower()}\n") self.assertIn(f"reason={reason}", stdout.getvalue()) self.assertNotIn(ENV["GH_TOKEN"], stdout.getvalue()) return calls def test_exact_pair_in_one_successful_attempt_reuses(self): self.invoke(proof(), False, f"exact-pr-proof run=73 attempt=2 sha={SHA}") def test_non_main_push_pr_and_dispatch_do_not_query(self): for env in ( {"GITHUB_EVENT_NAME": "pull_request"}, {"GITHUB_EVENT_NAME": "workflow_dispatch"}, {"GITHUB_REF": "refs/heads/dev-freeform"}, {"GITHUB_REF": "refs/tags/v1.2.3"}, ): with self.subTest(env=env): self.assertEqual(self.invoke({}, True, "not-push-main", env), []) def test_direct_or_release_shaped_sha_without_pr_proof_runs(self): data = proof() data[RUNS] = {"total_count": 0, "workflow_runs": []} self.invoke(data, True, "no-exact-successful-pr-run") def test_green_run_with_wrong_provenance_cannot_qualify(self): for field, value in ( ("event", "push"), ("head_sha", "b" * 40), ("workflow_id", 99), ("path", ".github/workflows/golden.yml"), ("repository", {"id": 999, "full_name": REPO}), ("repository", {"id": 123, "full_name": "other/repository"}), ("head_repository", {"id": 999, "full_name": "fork/repository"}), ("status", "in_progress"), ("conclusion", "failure"), ("conclusion", "cancelled"), ("conclusion", None), ("run_attempt", None), ): with self.subTest(field=field, value=value): data = proof() data[RUNS]["workflow_runs"][0][field] = value self.invoke(data, True, "no-exact-successful-pr-run") def test_workflow_name_alone_is_not_identity(self): data = proof() data[f"{ROOT}/workflows/ci.yml"]["path"] = ".github/workflows/other.yml" self.invoke(data, True, "workflow-identity-mismatch") def test_both_units_must_be_successful_not_skipped_or_neutral(self): for conclusion in ("skipped", "cancelled", "neutral", "failure", "timed_out", None): with self.subTest(conclusion=conclusion): data = proof() data[JOBS]["jobs"][1]["conclusion"] = conclusion self.invoke(data, True, "both-unit-jobs-not-successful") def test_absent_platform_and_duplicate_platform_do_not_form_pair(self): for duplicate in (False, True): data = proof() if duplicate: data[JOBS]["jobs"][1]["name"] = reuse.UNIT_JOBS[0] else: data[JOBS]["jobs"].pop() data[JOBS]["total_count"] = 1 self.invoke(data, True, "both-unit-jobs-not-successful") def test_partial_rerun_cannot_borrow_earlier_attempt_windows_success(self): data = proof() data[JOBS]["jobs"][1]["run_attempt"] = 1 self.invoke(data, True, "jobs-incomplete-or-mixed-attempt") def test_job_identity_and_completion_are_required(self): for field, value in (("run_id", 74), ("head_sha", "b" * 40), ("status", "in_progress"), ("run_attempt", None)): with self.subTest(field=field): data = proof() data[JOBS]["jobs"][0][field] = value self.invoke(data, True, "jobs-incomplete-or-mixed-attempt") def test_attempt_endpoint_must_itself_be_successful(self): data = proof() data[f"{RUN}/attempts/2"]["conclusion"] = "failure" self.invoke(data, True, "attempt-identity-or-result-mismatch") def test_rerun_started_during_proof_runs_units(self): data = proof() first = data[RUN] data[RUN] = [first, dict(first, run_attempt=3, status="in_progress", conclusion=None)] self.invoke(data, True, "run-changed-during-proof") def test_run_changed_since_listing_runs_units(self): data = proof() data[RUN]["status"] = "in_progress" self.invoke(data, True, "run-no-longer-qualifies") def test_runs_and_jobs_can_require_second_pages(self): data = proof() candidate = data[RUNS]["workflow_runs"][0] data[RUNS] = {"total_count": 101, "workflow_runs": [ dict(candidate, id=1000 + index, conclusion="failure") for index in range(100)]} data[RUNS.replace("&page=1", "&page=2")] = {"total_count": 101, "workflow_runs": [candidate]} linux, windows = data[JOBS]["jobs"] bystanders = [dict(linux, id=2000 + index, name=f"other-{index}") for index in range(99)] data[JOBS] = {"total_count": 101, "jobs": [linux] + bystanders} data[JOBS.replace("&page=1", "&page=2")] = {"total_count": 101, "jobs": [windows]} self.invoke(data, False, "exact-pr-proof") def test_partial_duplicate_and_capped_pages_are_not_proof(self): for change, reason in ( ({"total_count": 3}, "api-incomplete-page"), ({"total_count": 1000}, "api-count-unproven"), ({"total_count": 1}, "api-count-mismatch"), ({"jobs": []}, "api-incomplete-page"), ({"jobs": None}, "api-malformed-page"), ): with self.subTest(change=change): data = proof() data[JOBS].update(change) self.invoke(data, True, reason) data = proof() data[JOBS]["jobs"][1]["id"] = data[JOBS]["jobs"][0]["id"] self.invoke(data, True, "api-duplicate-row") def test_api_failures_and_malformed_responses_run_without_secret_logs(self): for response, reason in ( (subprocess.CompletedProcess([], 1, "", ENV["GH_TOKEN"]), "api-error"), (subprocess.CompletedProcess([], 0, "not json", ""), "api-malformed-json"), (subprocess.CompletedProcess([], 0, "[]", ""), "api-malformed-object"), (subprocess.TimeoutExpired("gh", 15), "api-timeout"), (FileNotFoundError("gh"), "gh-unavailable"), ({"total_count": 2, "jobs": [None, None]}, "api-malformed-row"), ): with self.subTest(reason=reason): data = proof() data[JOBS] = response self.invoke(data, True, reason) def test_missing_token_is_named_and_does_not_query(self): self.assertEqual(self.invoke({}, True, "missing-token", {"GH_TOKEN": ""}), []) if __name__ == "__main__": unittest.main()