state-hub/tests/test_release_headroom_preflight.py
tegwick 261a48772c
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s
Build and Publish Multi-Context Image / build-and-push (push) Successful in 42s
Add a read-only headroom preflight before State Hub promotion.
STATE-WP-0091: refuse Helm when remaining CPU cannot cover the API
surge, migrate hook, or unrelated pending demand. 65m fails, 105m is
narrowly sufficient, not factory admission.

Assistant: grok
Assistant-Session: 01a09dc1-b21e-77e1-919e-fcad2f82b267
2026-09-14 10:13:48 +02:00

66 lines
2.2 KiB
Python

from __future__ import annotations
import json
from pathlib import Path
from scripts.release_headroom_preflight import evaluate, main, pod_effective_requests
FIXTURES = Path(__file__).parent / "fixtures" / "release_headroom"
def _load(name: str) -> dict:
return json.loads((FIXTURES / name).read_text())
def test_revision_61_65m_refuses_api_surge() -> None:
report = evaluate(_load("rev61-65m.json"))
assert report["ok"] is False
assert report["remaining_cpu_m"] == 65
assert any("65m remaining" in r or "needs 100m" in r for r in report["reasons"])
assert "not a kube-scheduler guarantee" in " ".join(report["notes"]).lower() or any(
"guarantee" in n.lower() for n in report["notes"]
)
def test_revision_63_105m_is_narrowly_sufficient() -> None:
report = evaluate(_load("rev63-105m.json"))
assert report["ok"] is True
assert report["remaining_cpu_m"] == 105
assert report["reasons"] == []
assert any("105m" in n or "Not factory" in n for n in report["notes"])
def test_unrelated_pending_demand_refuses_105m() -> None:
report = evaluate(_load("rev63-105m-concurrent-pending.json"))
assert report["ok"] is False
assert any("unrelated pending" in r for r in report["reasons"])
def test_stale_observation_refuses() -> None:
observation = _load("rev63-105m.json")
observation["freshness_seconds"] = 3600
report = evaluate(observation)
assert report["ok"] is False
assert any("old" in r for r in report["reasons"])
def test_non_atomic_refuses() -> None:
report = evaluate(_load("rev63-105m.json"), {"atomic": False})
assert report["ok"] is False
assert any("atomic" in r for r in report["reasons"])
def test_init_overhead_uses_max_not_sum() -> None:
pod = {
"spec": {
"initContainers": [{"resources": {"requests": {"cpu": "80m", "memory": "64Mi"}}}],
"containers": [{"resources": {"requests": {"cpu": "100m", "memory": "512Mi"}}}],
}
}
cpu, _mem = pod_effective_requests(pod)
assert cpu == 100
def test_cli_fixture_exit_codes() -> None:
assert main(["--fixture", str(FIXTURES / "rev63-105m.json")]) == 0
assert main(["--fixture", str(FIXTURES / "rev61-65m.json")]) == 1