172 lines
7.6 KiB
Python
172 lines
7.6 KiB
Python
"""ORCH-058 TC-01..05: staging-image provenance helpers (src/image_freshness.py).
|
|
|
|
Covers the INV-FRESH building blocks in isolation:
|
|
* TC-01/02/03 — the PURE provenance verdict (match / mismatch / fail-closed).
|
|
* TC-04 — never-raise: docker/ssh/git errors -> safe verdict, no exception.
|
|
* TC-05 — conditionality: non-self repo = no-op (N/A); self repo = real.
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
|
|
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
|
|
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
|
|
|
|
from src import image_freshness as imf # noqa: E402
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TC-01: matching revisions -> fresh (PASS)
|
|
# ---------------------------------------------------------------------------
|
|
def test_tc01_matching_revisions_are_fresh():
|
|
ok, reason = imf.provenance_verdict("abc123def456", "abc123def456")
|
|
assert ok is True
|
|
assert "match" in reason.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TC-02: differing revisions -> NOT fresh (input for fail-fast)
|
|
# ---------------------------------------------------------------------------
|
|
def test_tc02_differing_revisions_are_not_fresh():
|
|
ok, reason = imf.provenance_verdict("aaaaaaaaaaaa", "bbbbbbbbbbbb")
|
|
assert ok is False
|
|
assert "mismatch" in reason.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TC-03: fail-closed — empty label OR empty expected -> never "fresh by default"
|
|
# ---------------------------------------------------------------------------
|
|
def test_tc03_empty_image_label_fails_closed():
|
|
ok, reason = imf.provenance_verdict("abc123", "")
|
|
assert ok is False
|
|
assert "fail-closed" in reason.lower()
|
|
|
|
|
|
def test_tc03_empty_expected_revision_fails_closed():
|
|
ok, reason = imf.provenance_verdict("", "abc123")
|
|
assert ok is False
|
|
assert "fail-closed" in reason.lower()
|
|
|
|
|
|
def test_tc03_both_empty_fails_closed():
|
|
ok, _ = imf.provenance_verdict("", "")
|
|
assert ok is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TC-04: never-raise on docker/ssh/inspect/git errors -> safe verdict
|
|
# ---------------------------------------------------------------------------
|
|
def test_tc04_image_revision_inspect_error_returns_empty(monkeypatch):
|
|
def _boom(*a, **k):
|
|
raise OSError("docker not found")
|
|
monkeypatch.setattr(imf.subprocess, "run", _boom)
|
|
# Never raises; fail-closed empty -> downstream provenance mismatch.
|
|
assert imf.image_revision("orchestrator-orchestrator-staging") == ""
|
|
|
|
|
|
def test_tc04_image_revision_nonzero_rc_returns_empty(monkeypatch):
|
|
monkeypatch.setattr(
|
|
imf.subprocess, "run",
|
|
lambda *a, **k: subprocess.CompletedProcess(a, 1, stdout="", stderr="no such image"),
|
|
)
|
|
assert imf.image_revision("missing-image") == ""
|
|
|
|
|
|
def test_tc04_image_revision_no_value_label_returns_empty(monkeypatch):
|
|
# `docker inspect` prints "<no value>" when the label key is absent.
|
|
monkeypatch.setattr(
|
|
imf.subprocess, "run",
|
|
lambda *a, **k: subprocess.CompletedProcess(a, 0, stdout="<no value>\n", stderr=""),
|
|
)
|
|
assert imf.image_revision("unlabelled-image") == ""
|
|
|
|
|
|
def test_tc04_validated_revision_missing_worktree_returns_empty(monkeypatch, tmp_path):
|
|
# No worktree on disk -> fail-closed empty SHA, never raises.
|
|
monkeypatch.setattr(imf.settings, "worktrees_dir", str(tmp_path / "nope"))
|
|
monkeypatch.setattr(imf.settings, "repos_dir", str(tmp_path / "nope"))
|
|
assert imf.validated_revision("orchestrator", "feature/ORCH-058-x") == ""
|
|
|
|
|
|
def test_tc04_check_staging_image_fresh_never_raises(monkeypatch):
|
|
# Self repo + enabled, but rebuild blows up -> caught -> safe (False) verdict.
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", True)
|
|
monkeypatch.setattr(imf.settings, "image_freshness_repos", "")
|
|
monkeypatch.setattr(imf, "validated_revision", lambda r, b: "deadbeef")
|
|
|
|
def _boom(*a, **k):
|
|
raise RuntimeError("ssh exploded")
|
|
monkeypatch.setattr(imf, "rebuild_staging_image", _boom)
|
|
ok, reason = imf.check_staging_image_fresh("orchestrator", "ORCH-058", "feature/ORCH-058-x")
|
|
assert ok is False
|
|
assert "error" in reason.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TC-05: conditionality (self-hosting only)
|
|
# ---------------------------------------------------------------------------
|
|
def test_tc05_applies_only_to_self_hosting_by_default(monkeypatch):
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", True)
|
|
monkeypatch.setattr(imf.settings, "image_freshness_repos", "")
|
|
assert imf.image_freshness_applies("orchestrator") is True
|
|
assert imf.image_freshness_applies("enduro-trails") is False
|
|
|
|
|
|
def test_tc05_applies_respects_repos_csv(monkeypatch):
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", True)
|
|
monkeypatch.setattr(imf.settings, "image_freshness_repos", "enduro-trails")
|
|
assert imf.image_freshness_applies("enduro-trails") is True
|
|
# CSV is authoritative: orchestrator not listed -> not real.
|
|
assert imf.image_freshness_applies("orchestrator") is False
|
|
|
|
|
|
def test_tc05_kill_switch_disables_for_everyone(monkeypatch):
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", False)
|
|
monkeypatch.setattr(imf.settings, "image_freshness_repos", "")
|
|
assert imf.image_freshness_applies("orchestrator") is False
|
|
|
|
|
|
def test_tc05_check_is_noop_for_non_self_repo(monkeypatch):
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", True)
|
|
monkeypatch.setattr(imf.settings, "image_freshness_repos", "")
|
|
ok, reason = imf.check_staging_image_fresh("enduro-trails", "ET-001", "feature/ET-001-x")
|
|
assert ok is True
|
|
assert "N/A" in reason
|
|
|
|
|
|
def test_tc05_check_disabled_is_pass(monkeypatch):
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", False)
|
|
ok, reason = imf.check_staging_image_fresh("orchestrator", "ORCH-058", "feature/ORCH-058-x")
|
|
assert ok is True
|
|
assert "disabled" in reason.lower()
|
|
|
|
|
|
def test_tc05_check_real_for_self_repo_rebuilds(monkeypatch):
|
|
# Self repo + enabled: validated commit resolved + rebuild OK -> fresh PASS.
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", True)
|
|
monkeypatch.setattr(imf.settings, "image_freshness_repos", "")
|
|
monkeypatch.setattr(imf, "validated_revision", lambda r, b: "abc123def456")
|
|
monkeypatch.setattr(imf, "rebuild_staging_image", lambda r, b, s: (True, "healthy"))
|
|
ok, reason = imf.check_staging_image_fresh("orchestrator", "ORCH-058", "feature/ORCH-058-x")
|
|
assert ok is True
|
|
assert "abc123def456"[:12] in reason
|
|
|
|
|
|
def test_tc05_check_fail_closed_when_no_validated_revision(monkeypatch):
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", True)
|
|
monkeypatch.setattr(imf.settings, "image_freshness_repos", "")
|
|
monkeypatch.setattr(imf, "validated_revision", lambda r, b: "")
|
|
ok, reason = imf.check_staging_image_fresh("orchestrator", "ORCH-058", "feature/ORCH-058-x")
|
|
assert ok is False
|
|
assert "fail-closed" in reason.lower()
|
|
|
|
|
|
def test_tc05_check_fails_when_rebuild_fails(monkeypatch):
|
|
monkeypatch.setattr(imf.settings, "image_freshness_enabled", True)
|
|
monkeypatch.setattr(imf.settings, "image_freshness_repos", "")
|
|
monkeypatch.setattr(imf, "validated_revision", lambda r, b: "abc123def456")
|
|
monkeypatch.setattr(imf, "rebuild_staging_image", lambda r, b, s: (False, "build error"))
|
|
ok, reason = imf.check_staging_image_fresh("orchestrator", "ORCH-058", "feature/ORCH-058-x")
|
|
assert ok is False
|
|
assert "rebuild failed" in reason.lower()
|