"""Exercise the real workflow's receipt publisher without Docker or network.""" from __future__ import annotations import copy import json import os import subprocess from pathlib import Path import pytest import yaml ROOT = Path(__file__).resolve().parents[2] WORKFLOW = ROOT / ".gitea/workflows/runner-build-acceptance.yml" REVISION = "a" * 40 TREE = "b" * 40 MARKER = "ITWORX_CI_JOB_DOCKER_PROBE_RESULT=" @pytest.fixture def workflow(): return yaml.safe_load(WORKFLOW.read_text(encoding="utf-8")) def test_only_manual_fixed_build_profile_and_pinned_actions(workflow): # PyYAML's YAML 1.1 parser treats the Actions `on` key as boolean true. events = workflow.get("on", workflow.get(True)) assert set(events) == {"workflow_dispatch"} assert set(events["workflow_dispatch"]["inputs"]) == {"source_sha"} assert workflow["permissions"] == {"contents": "read"} assert workflow["concurrency"]["cancel-in-progress"] is False assert set(workflow["jobs"]) == {"build-runner-acceptance"} job = workflow["jobs"]["build-runner-acceptance"] assert job["runs-on"] == "itworx-ci-build" assert "services" not in job and "container" not in job checkout, probe, upload = job["steps"] assert checkout["uses"] == "actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683" assert checkout["with"]["persist-credentials"] is False assert checkout["with"]["github-server-url"] == "https://gitea.example.com" assert checkout["with"]["ref"] == "${{ gitea.sha }}" assert "secrets." not in WORKFLOW.read_text(encoding="utf-8") assert probe["shell"] == "bash" assert upload["uses"] == "ChristopherHX/gitea-upload-artifact@62ac910c5d3dfa85c7cb2df15afe2e342b2407c2" assert upload.get("if") is None # Actions default success; never always(). assert upload["with"]["path"] == "${{ runner.temp }}/runner-build-acceptance/receipt.json" assert upload["with"]["if-no-files-found"] == "error" assert upload["with"]["overwrite"] is False @pytest.fixture def publisher(workflow, monkeypatch, tmp_path): script = workflow["jobs"]["build-runner-acceptance"]["steps"][1]["run"] script = script.split("python3 - <<'PY'\n", 1)[1].rsplit("\nPY", 1)[0] environment = { "EXPECTED_SOURCE_SHA": REVISION, "DISPATCH_SHA": REVISION, "DISPATCH_REF": "refs/heads/main", "DISPATCH_EVENT": "workflow_dispatch", "RUN_REPOSITORY": "Jens/Unraid-AppOps-Gateway", "RUN_ID": "22840", "RUN_NUMBER": "1261", "RUN_ATTEMPT": "1", "RUN_JOB": "build-runner-acceptance", "RUNNER_TEMP": str(tmp_path), } probe = { "schema": "itworx.ci.job-docker-probe/v1", "status": "passed", "sourceRevision": REVISION, "sourceTree": TREE, "engineVersion": "29.8.2", "engineRootless": True, "engineStorageDriver": "overlay2", "engineCgroupDriver": "none", "engineSocket": "unix:///run/itworx-job-docker/docker.sock", "engineSocketUid": 1000, "composeVersion": "5.5.1", "buildRun": True, "composeHealthy": True, "composeDnsResolved": True, "archiveNormalized": True, "archiveLoaded": True, "holdSeconds": 30, "imageLabels": { "org.opencontainers.image.source": "https://gitea.example.com/Jens/Unraid-AppOps-Gateway", "org.opencontainers.image.revision": REVISION, "tech.itworx.source-tree": TREE, "tech.itworx.ci.profile": "build", }, "imageId": "sha256:" + "c" * 64, "archiveSha256": "sha256:" + "d" * 64, } state = {"calls": [], "probe": probe, "exit": 0, "dirty": "", "tree": TREE} def run(args, **kwargs): state["calls"].append(args) assert kwargs["stdin"] == subprocess.DEVNULL if args[0] == "git": assert kwargs["stderr"] == subprocess.DEVNULL command = args[1:] values = { ("rev-parse", "HEAD"): REVISION, ("cat-file", "-t", REVISION): "commit", ("rev-parse", "HEAD^{tree}"): state["tree"], ("cat-file", "-t", TREE): "tree", ("rev-parse", REVISION + "^{tree}"): TREE, ("status", "--porcelain", "--untracked-files=no"): state["dirty"], } return subprocess.CompletedProcess(args, 0, values[tuple(command)] + "\n") assert args == ["/usr/local/bin/itworx-ci-job-docker-probe", REVISION, TREE, "--hold-seconds", "30"] assert kwargs["stderr"] == subprocess.DEVNULL assert kwargs["timeout"] == 1000 if state.get("timeout"): raise subprocess.TimeoutExpired(args, 1000, stderr="PRIVATE_TEST_SECRET") output = state.get("raw", MARKER + json.dumps(state["probe"]) + "\n") kwargs["stdout"].write(output.encode()) return subprocess.CompletedProcess(args, state["exit"]) def execute(): with monkeypatch.context() as patch: for key, value in environment.items(): patch.setenv(key, value) patch.setattr(subprocess, "run", run) # Execute repository-owned workflow code; every subprocess is intercepted. previous_umask = os.umask(0o077) try: exec(compile(script, str(WORKFLOW), "exec"), {}) # noqa: S102 finally: os.umask(previous_umask) state["environment"] = environment state["execute"] = execute state["receipt"] = tmp_path / "runner-build-acceptance/receipt.json" return state def test_success_publishes_only_finite_post_cleanup_receipt(publisher): publisher["execute"]() receipt = json.loads(publisher["receipt"].read_text()) assert receipt["cleanupVerified"] is True assert receipt["probeExitCode"] == 0 assert receipt["probe"] == publisher["probe"] assert receipt["gitea"] == { "repository": "Jens/Unraid-AppOps-Gateway", "workflow": "runner-build-acceptance.yml", "jobKey": "build-runner-acceptance", "runId": 22840, "runNumber": 1261, "runAttempt": 1, "nativeCorrelationRequired": True, } def test_missing_attempt_is_honest_null_not_invented_one(publisher): publisher["environment"]["RUN_ATTEMPT"] = "" publisher["execute"]() run = json.loads(publisher["receipt"].read_text())["gitea"] assert run["runAttempt"] is None assert run["nativeCorrelationRequired"] is True @pytest.mark.parametrize("failed_probe", [False, True]) def test_publisher_fixture_restores_process_umask(publisher, failed_probe, monkeypatch): mask = [0o022] def umask(value): previous, mask[0] = mask[0], value return previous monkeypatch.setattr(os, "umask", umask) publisher["exit"] = 70 if failed_probe else 0 if failed_probe: with pytest.raises(SystemExit): publisher["execute"]() else: publisher["execute"]() assert mask[0] == 0o022 @pytest.mark.parametrize(("key", "value"), [ ("EXPECTED_SOURCE_SHA", "$(echo PRIVATE_TEST_SECRET)"), ("EXPECTED_SOURCE_SHA", "e" * 40), ("DISPATCH_SHA", "e" * 40), ("DISPATCH_REF", "refs/heads/other"), ("DISPATCH_EVENT", "push"), ("RUN_REPOSITORY", "Jens/Other"), ("RUN_ID", ""), ("RUN_ID", "0"), ("RUN_ATTEMPT", "garbage"), ("RUN_JOB", "other"), ]) def test_invalid_identity_refuses_before_probe(publisher, key, value): publisher["environment"][key] = value with pytest.raises(SystemExit, match="no success receipt published"): publisher["execute"]() assert all(args[0] == "git" for args in publisher["calls"]) assert not publisher["receipt"].exists() @pytest.mark.parametrize(("key", "value"), [("dirty", " M tracked.txt"), ("tree", "invalid")]) def test_bad_checkout_tree_refuses_before_probe(publisher, key, value): publisher[key] = value with pytest.raises(SystemExit, match="no success receipt published"): publisher["execute"]() assert all(args[0] == "git" for args in publisher["calls"]) assert not publisher["receipt"].exists() @pytest.mark.parametrize("mode", ["cleanup_failure", "hold_failure", "timeout", "secret_output"]) def test_marker_does_not_authorize_success_before_exit_zero(publisher, mode): if mode in {"cleanup_failure", "hold_failure"}: publisher["exit"] = 70 elif mode == "timeout": publisher["timeout"] = True else: publisher["raw"] = "PRIVATE_TEST_SECRET\n" + MARKER + json.dumps(publisher["probe"]) with pytest.raises(SystemExit) as failure: publisher["execute"]() assert "PRIVATE_TEST_SECRET" not in str(failure.value) assert not publisher["receipt"].exists() @pytest.mark.parametrize(("key", "value"), [ ("engineRootless", False), ("engineRootless", 1), ("archiveLoaded", False), ("composeHealthy", False), ("sourceRevision", "e" * 40), ("sourceTree", "e" * 40), ("imageId", "invalid"), ("unexpected", "PRIVATE_TEST_SECRET"), ]) def test_false_or_foreign_probe_receipt_is_never_published(publisher, key, value): publisher["probe"] = copy.deepcopy(publisher["probe"]) publisher["probe"][key] = value with pytest.raises(SystemExit) as failure: publisher["execute"]() assert "PRIVATE_TEST_SECRET" not in str(failure.value) assert not publisher["receipt"].exists()