223 lines
9.1 KiB
Python
223 lines
9.1 KiB
Python
"""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()
|