85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from govoplan_core.core.operations import (
|
|
OperationalCheck,
|
|
OperationalCheckProviderRegistration,
|
|
)
|
|
from govoplan_ops.backend.api.v1 import routes
|
|
|
|
|
|
@dataclass
|
|
class _Manifest:
|
|
operational_check_providers: tuple[OperationalCheckProviderRegistration, ...]
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, *registrations: OperationalCheckProviderRegistration):
|
|
self._manifest = _Manifest(tuple(registrations))
|
|
|
|
def manifests(self):
|
|
return (self._manifest,)
|
|
|
|
|
|
def test_module_operational_checks_cache_and_force() -> None:
|
|
routes._module_check_cache.clear()
|
|
calls = 0
|
|
|
|
def provider() -> OperationalCheck:
|
|
nonlocal calls
|
|
calls += 1
|
|
return OperationalCheck("example.roundtrip", "Example", "ok", "Passed")
|
|
|
|
registration = OperationalCheckProviderRegistration(
|
|
module_id="example",
|
|
check_id="example.roundtrip",
|
|
provider=provider,
|
|
)
|
|
registry = _Registry(registration)
|
|
|
|
assert routes._module_operational_checks(registry, force=False)[0]["state"] == "ok"
|
|
assert routes._module_operational_checks(registry, force=False)[0]["state"] == "ok"
|
|
assert calls == 1
|
|
routes._module_operational_checks(registry, force=True)
|
|
assert calls == 2
|
|
|
|
|
|
def test_module_operational_check_failure_is_isolated() -> None:
|
|
routes._module_check_cache.clear()
|
|
|
|
def provider() -> OperationalCheck:
|
|
raise RuntimeError("secret detail")
|
|
|
|
result = routes._module_operational_checks(
|
|
_Registry(OperationalCheckProviderRegistration(
|
|
module_id="example",
|
|
check_id="example.failed",
|
|
provider=provider,
|
|
)),
|
|
force=True,
|
|
)[0]
|
|
|
|
assert result["state"] == "error"
|
|
assert "secret detail" not in result["detail"]
|
|
|
|
|
|
def test_shared_runtime_cluster_missing_replicas_blocks_readiness() -> None:
|
|
check = routes._runtime_cluster_check(
|
|
{
|
|
"available": True,
|
|
"state_profile": "shared",
|
|
"nodes": [
|
|
{"role": "api", "state": "active", "stale": False},
|
|
{"role": "worker", "state": "active", "stale": True},
|
|
],
|
|
"expected": {"api": 2, "worker": 1},
|
|
"active": {"api": 1, "worker": 0},
|
|
"recovery": {"requires_attention": 1},
|
|
}
|
|
)
|
|
|
|
assert check["state"] == "error"
|
|
assert check["readiness_critical"] is True
|
|
assert check["metrics"]["recovery_required"] == 1
|