Nick
36ee604646
ROOT CAUSE (red-CI promotions 0.66.28/29): the required-checks 'Publish protected-branch gate' step read /tmp/need_* files that no step ever wrote and echoed 'All required CI checks passed' UNCONDITIONALLY. With if: always() the needs gate is overridden, so the job ran on red needs and published success — deploy_qualify correctly trusted it and promoted red commits (24413e4,fd6e9b4, de30cd9; status timelines in the closure evidence). FIX: publish step now runs only when every needs.*.result == 'success'; companion fail-closed step exits 1 otherwise, so the required-checks context reports failure instead of lying. DEFENSE-IN-DEPTH: deploy_qualify verdict is now the AND of two independent legs — (1) newest required-checks observation for the exact SHA, (2) the latest COMPLETED ci.yml@refs/heads/master run for that exact SHA with every job concluded success (fan-out resolved by path, run resolved by head_sha, jobs envelope unwrapped — no aggregate-status fallback). Any pending/missing/disagreement = not deployable. CANARY: gitea_pipeline now asserts the verdict evaluates needs results and fails closed — the class locked shut. 20 qualifier tests with fixtures from the real incident shapes; 1402 suite green.
179 lines
8.0 KiB
Python
179 lines
8.0 KiB
Python
"""test_deploy_qualify.py — deploy qualification verdict tests.
|
|
|
|
Fixtures modeled on the REAL red-promotion incidents (24413e4, fd6e9b4,
|
|
de30cd9 — 0.66.28/29 promoted with failing type-check because the
|
|
required-checks aggregator published success unconditionally). The
|
|
qualifier must fail closed against every shape seen live.
|
|
"""
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
import deploy_qualify as dq # noqa: E402
|
|
|
|
SHA = "a" * 40
|
|
OTHER_SHA = "b" * 40
|
|
|
|
|
|
def st(context, status, created, description=""):
|
|
return {"context": context, "status": status, "created_at": created,
|
|
"description": description}
|
|
|
|
|
|
def run(path, sha, status, created, run_id, conclusions=None):
|
|
return {"path": path, "head_sha": sha, "status": status, "created_at": created,
|
|
"id": run_id, "_conclusions": conclusions or []}
|
|
|
|
|
|
RC = "CI Pipeline / required-checks"
|
|
|
|
|
|
def jobs_fetch_for(runs):
|
|
def fetch(run_id):
|
|
for r in runs:
|
|
if r.get("id") == run_id:
|
|
return [{"name": f"job{i}", "conclusion": c}
|
|
for i, c in enumerate(r["_conclusions"])]
|
|
return None
|
|
return fetch
|
|
|
|
|
|
# ── leg 1: required-checks context ────────────────────────────────────────
|
|
|
|
class TestRequiredCheckVerdict:
|
|
def test_failed_required_check_is_failure(self):
|
|
v = dq.required_check_verdict([st(RC + " (push)", "failure", "T2")])
|
|
assert v == "failure"
|
|
|
|
def test_pending_required_check_is_pending(self):
|
|
v = dq.required_check_verdict([st(RC + " (push)", "pending", "T2")])
|
|
assert v == "pending"
|
|
|
|
def test_missing_required_check_is_pending(self):
|
|
assert dq.required_check_verdict([]) == "pending"
|
|
assert dq.required_check_verdict(
|
|
[st("CI Pipeline / lint (push)", "success", "T1")]) == "pending"
|
|
|
|
def test_stale_success_newer_failure_is_failure(self):
|
|
v = dq.required_check_verdict([
|
|
st(RC + " (push)", "success", "T1"),
|
|
st(RC + " (push)", "failure", "T2"),
|
|
])
|
|
assert v == "failure"
|
|
|
|
def test_newer_success_after_older_failure_is_success(self):
|
|
v = dq.required_check_verdict([
|
|
st(RC + " (push)", "failure", "T1"),
|
|
st(RC + " (push)", "success", "T2"),
|
|
])
|
|
assert v == "success"
|
|
|
|
def test_all_skipped_is_pending_not_success(self):
|
|
v = dq.required_check_verdict([st(RC + " (push)", "skipped", "T1")])
|
|
assert v == "pending"
|
|
|
|
|
|
# ── leg 2: independent jobs-level verification ────────────────────────────
|
|
|
|
class TestIndependentJobsVerdict:
|
|
def test_all_jobs_success_is_success(self):
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T2", 1,
|
|
["success", "success", "success"])]
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=SHA) == "success"
|
|
|
|
def test_any_job_failure_is_failure(self):
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T2", 1,
|
|
["success", "failure", "success"])]
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=SHA) == "failure"
|
|
|
|
def test_in_progress_run_is_pending(self):
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "in_progress", "T2", 1,
|
|
["success"])]
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=SHA) == "pending"
|
|
|
|
def test_no_run_for_this_sha_is_pending(self):
|
|
runs = [run("ci.yml@refs/heads/master", OTHER_SHA, "completed", "T2", 1,
|
|
["success"])]
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=SHA) == "pending"
|
|
|
|
def test_source_and_successor_sha_distinct(self):
|
|
"""A green run on the successor SHA must not qualify the red source
|
|
SHA (and vice versa) — the exact candidate governs."""
|
|
runs = [run("ci.yml@refs/heads/master", OTHER_SHA, "completed", "T1", 1,
|
|
["success", "success"]),
|
|
run("ci.yml@refs/heads/master", SHA, "completed", "T2", 2,
|
|
["success", "failure"])]
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=SHA) == "failure"
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=OTHER_SHA) == "success"
|
|
|
|
def test_tag_fanout_does_not_poison_verdict(self):
|
|
"""A failing tag-run must not fail a green master run, and a failing
|
|
master run must not be rescued by a green tag run."""
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T1", 1,
|
|
["success", "success"]),
|
|
run("ci.yml@refs/tags/v9.9.9", SHA, "completed", "T2", 2,
|
|
["failure"])]
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=SHA) == "success"
|
|
runs2 = [run("ci.yml@refs/heads/master", SHA, "completed", "T1", 1,
|
|
["failure"]),
|
|
run("ci.yml@refs/tags/v9.9.9", SHA, "completed", "T2", 2,
|
|
["success", "success"])]
|
|
assert dq.independent_jobs_verdict(runs2, jobs_fetch_for(runs2), sha=SHA) == "failure"
|
|
|
|
def test_stale_success_newer_completed_failure_wins(self):
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T1", 1,
|
|
["success", "success"]),
|
|
run("ci.yml@refs/heads/master", SHA, "completed", "T2", 2,
|
|
["success", "failure"])]
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=SHA) == "failure"
|
|
|
|
def test_jobs_probe_failure_is_pending(self):
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T1", 99)]
|
|
assert dq.independent_jobs_verdict(runs, lambda rid: None, sha=SHA) == "pending"
|
|
|
|
def test_empty_jobs_visible_is_pending(self):
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T1", 1, [])]
|
|
assert dq.independent_jobs_verdict(runs, jobs_fetch_for(runs), sha=SHA) == "pending"
|
|
|
|
|
|
# ── combined verdict: fail-closed conjunction ─────────────────────────────
|
|
|
|
class TestCombinedQualify:
|
|
def test_the_real_incident_shape_is_refused(self):
|
|
"""THE 24413e4 shape: type-check + unit-tests failed, yet the
|
|
aggregator published success. Even with the lying aggregator leg,
|
|
the independent jobs leg must refuse."""
|
|
statuses = [
|
|
st(RC + " (push)", "success", "2026-09-12T19:13:53"), # the lie
|
|
st("CI Pipeline / unit-tests (push)", "failure", "2026-09-12T19:13:51"),
|
|
st("CI Pipeline / type-check (push)", "failure", "2026-09-12T19:13:34"),
|
|
]
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T1", 1,
|
|
["success", "failure", "failure", "success"])]
|
|
v = dq.qualify(statuses, runs, jobs_fetch_for(runs), sha=SHA)
|
|
assert v == "failure", "red candidate must never qualify"
|
|
|
|
def test_pending_aggregator_with_green_jobs_still_pending(self):
|
|
statuses = [st(RC + " (push)", "pending", "T2")]
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T1", 1,
|
|
["success", "success"])]
|
|
assert dq.qualify(statuses, runs, jobs_fetch_for(runs), sha=SHA) == "pending"
|
|
|
|
def test_both_green_qualifies(self):
|
|
statuses = [st(RC + " (push)", "success", "T2")]
|
|
runs = [run("ci.yml@refs/heads/master", SHA, "completed", "T1", 1,
|
|
["success", "success"])]
|
|
assert dq.qualify(statuses, runs, jobs_fetch_for(runs), sha=SHA) == "success"
|
|
|
|
def test_missing_everything_is_pending(self):
|
|
assert dq.qualify([], [], jobs_fetch_for([]), sha=SHA) == "pending"
|
|
|
|
def test_combine_is_fail_closed(self):
|
|
assert dq.combine("success", "failure") == "failure"
|
|
assert dq.combine("failure", "pending") == "failure"
|
|
assert dq.combine("success", "pending") == "pending"
|
|
assert dq.combine("success", "success") == "success"
|