98 lines
4.8 KiB
Python
98 lines
4.8 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
MODULE_PATH = Path(__file__).parents[2] / "autoinstall/offline/activation-health-window.py"
|
|
SPEC = importlib.util.spec_from_file_location("activation_health_window", MODULE_PATH)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
def profile(required=True):
|
|
return {"schema": "com.cezen.nexus-one.activation-health-profile.v1", "components": [
|
|
{"name": "api", "url": "http://127.0.0.1:18080/health", "required": required,
|
|
"expected_service": "api", "systemd_unit": None, "max_restarts": 0}
|
|
]}
|
|
|
|
|
|
class ActivationHealthWindowTests(unittest.TestCase):
|
|
def test_ready_requires_all_consecutive_samples(self):
|
|
evidence = MODULE.evaluate(profile(), "op-1", 3, 1, 2, 10, False,
|
|
lambda component, timeout: {"ok": True, "code": "READY", "status": "ready"},
|
|
lambda component: {"ok": True, "code": "RESTARTS_OK", "restarts": 0},
|
|
lambda interval: None)
|
|
self.assertEqual(evidence["outcome"], "ready")
|
|
self.assertEqual(evidence["samples_completed"], 3)
|
|
self.assertEqual(len(evidence["checks"]), 3)
|
|
|
|
def test_optional_degraded_requires_explicit_policy(self):
|
|
degraded = lambda component, timeout: {"ok": False, "code": "DEGRADED", "status": "degraded"}
|
|
accepted = MODULE.evaluate(profile(False), "op-1", 1, 0, 2, 10, True, degraded)
|
|
rejected = MODULE.evaluate(profile(False), "op-1", 1, 0, 2, 10, False, degraded)
|
|
self.assertEqual(accepted["outcome"], "degraded")
|
|
self.assertEqual(rejected["outcome"], "rejected")
|
|
|
|
def test_timeout_and_malformed_response_reject(self):
|
|
for code in ("TIMEOUT", "MALFORMED_JSON"):
|
|
with self.subTest(code=code):
|
|
evidence = MODULE.evaluate(profile(), "op-1", 3, 0, 2, 10, False,
|
|
lambda component, timeout, code=code: {"ok": False, "code": code})
|
|
self.assertEqual(evidence["outcome"], "rejected")
|
|
self.assertEqual(evidence["samples_completed"], 1)
|
|
self.assertEqual(evidence["checks"][-1]["code"], code)
|
|
|
|
def test_rejected_window_returns_23_for_systemd_onfailure(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
config = root / "profile.json"
|
|
config.write_text(json.dumps(profile()))
|
|
evidence = root / "evidence.json"
|
|
arguments = ["--operation-id", "op-1", "--profile-config", str(config),
|
|
"--samples", "3", "--interval-seconds", "0", "--timeout-seconds", "2",
|
|
"--window-timeout-seconds", "10",
|
|
"--evidence", str(evidence)]
|
|
result = MODULE.main(arguments,
|
|
lambda component, timeout: {"ok": False, "code": "HTTP_503"})
|
|
self.assertEqual(result, 23)
|
|
self.assertEqual(json.loads(evidence.read_text())["outcome"], "rejected")
|
|
self.assertEqual(os.stat(evidence).st_mode & 0o777, 0o600)
|
|
|
|
def test_non_loopback_and_hostname_urls_fail_closed(self):
|
|
for url in ("http://172.16.10.144:8080/health", "http://localhost:8080/health"):
|
|
candidate = profile()
|
|
candidate["components"][0]["url"] = url
|
|
with self.assertRaises(MODULE.HealthConfigError):
|
|
MODULE.validate_config(candidate)
|
|
|
|
def test_total_window_deadline_rejects_before_next_sample(self):
|
|
now = [0.0]
|
|
evidence = MODULE.evaluate(
|
|
profile(), "op-1", 3, 2, 1, 1.5, False,
|
|
lambda component, timeout: {"ok": True, "code": "READY", "status": "ready"},
|
|
lambda component: {"ok": True, "code": "RESTARTS_OK", "restarts": 0},
|
|
lambda interval: now.__setitem__(0, now[0] + interval), lambda: now[0])
|
|
self.assertEqual(evidence["outcome"], "rejected")
|
|
self.assertEqual(evidence["checks"][-1]["code"], "WINDOW_TIMEOUT")
|
|
|
|
def test_restart_storm_rejects_before_http_probe(self):
|
|
called = []
|
|
candidate = profile()
|
|
candidate["components"][0].update(systemd_unit="cezen-api.service", max_restarts=2)
|
|
evidence = MODULE.evaluate(
|
|
candidate, "op-1", 3, 0, 1, 10, False,
|
|
lambda component, timeout: called.append(component),
|
|
lambda component: {"ok": False, "code": "RESTART_STORM", "restarts": 3})
|
|
self.assertEqual(evidence["outcome"], "rejected")
|
|
self.assertEqual(evidence["checks"][-1]["code"], "RESTART_STORM")
|
|
self.assertEqual(evidence["checks"][-1]["restart_count"], 3)
|
|
self.assertEqual(called, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|