test(ariadne): cover cluster anomaly edges

This commit is contained in:
codex 2026-04-21 04:16:43 -03:00
parent 13594eba57
commit 71c1b9b7bf
2 changed files with 8 additions and 10 deletions

View File

@ -34,13 +34,7 @@ def _pvc_pressure_signals(metrics: dict[str, Any]) -> list[dict[str, Any]]:
return output return output
def _build_anomalies( def _build_anomalies(metrics: dict[str, Any], nodes_summary: dict[str, Any], workloads_health: dict[str, Any], kustomizations: dict[str, Any], events: dict[str, Any]) -> list[dict[str, Any]]:
metrics: dict[str, Any],
nodes_summary: dict[str, Any],
workloads_health: dict[str, Any],
kustomizations: dict[str, Any],
events: dict[str, Any],
) -> list[dict[str, Any]]:
anomalies: list[dict[str, Any]] = [] anomalies: list[dict[str, Any]] = []
_append_pod_anomalies(anomalies, metrics) _append_pod_anomalies(anomalies, metrics)
_append_workload_anomalies(anomalies, workloads_health) _append_workload_anomalies(anomalies, workloads_health)
@ -73,9 +67,7 @@ def _append_pod_anomalies(anomalies: list[dict[str, Any]], metrics: dict[str, An
) )
def _append_workload_anomalies( def _append_workload_anomalies(anomalies: list[dict[str, Any]], workloads_health: dict[str, Any]) -> None:
anomalies: list[dict[str, Any]], workloads_health: dict[str, Any]
) -> None:
for key in ("deployments", "statefulsets", "daemonsets"): for key in ("deployments", "statefulsets", "daemonsets"):
entry = workloads_health.get(key) if isinstance(workloads_health.get(key), dict) else {} entry = workloads_health.get(key) if isinstance(workloads_health.get(key), dict) else {}
not_ready = entry.get("not_ready") or 0 not_ready = entry.get("not_ready") or 0

View File

@ -151,7 +151,13 @@ def test_health_anomaly_signal_profile_and_attention_domains() -> None:
anomaly_rows = anomalies._build_anomalies(metrics, nodes_summary, workloads_health, kustomizations, events) anomaly_rows = anomalies._build_anomalies(metrics, nodes_summary, workloads_health, kustomizations, events)
assert {row["kind"] for row in anomaly_rows} >= {"pods_pending", "pvc_pressure", "flux_not_ready"} assert {row["kind"] for row in anomaly_rows} >= {"pods_pending", "pvc_pressure", "flux_not_ready"}
assert anomalies._severity_rank("critical") == 0 assert anomalies._severity_rank("critical") == 0
assert anomalies._severity_rank("unknown") == 2
assert anomalies._pvc_pressure_signals(metrics)[0]["target"] == "apps/data" assert anomalies._pvc_pressure_signals(metrics)[0]["target"] == "apps/data"
assert anomalies._pvc_pressure_signals({"pvc_usage_top": [{"metric": {"namespace": "apps", "persistentvolumeclaim": "cold"}, "value": 79.0}]}) == []
without_optional = []
anomalies._append_node_anomalies(without_optional, {})
anomalies._append_event_anomalies(without_optional, {})
assert without_optional == []
bullets = health._health_bullets(metrics, nodes_summary, workloads_health, anomaly_rows) bullets = health._health_bullets(metrics, nodes_summary, workloads_health, anomaly_rows)
assert bullets[0] == "Nodes ready: 1/2" assert bullets[0] == "Nodes ready: 1/2"