test(ariadne): cover cluster pod summary edges

ariadne/services/cluster_state_pods.py

@@ -239,10 +239,7 @@ def _node_pods_top(node_pods: list[dict[str, Any]], limit: int = 5) -> list[dict
     return output
 
 
-def _record_pending_pod(
-    pending_oldest: list[dict[str, Any]],
-    info: dict[str, Any],
-) -> bool:
+def _record_pending_pod(pending_oldest: list[dict[str, Any]], info: dict[str, Any]) -> bool:
     age_hours = info.get("age_hours")
     if age_hours is None:
         return False
@@ -250,10 +247,7 @@ def _record_pending_pod(
     return age_hours >= _PENDING_15M_HOURS
 
 
-def _update_pod_issue(
-    pod: dict[str, Any],
-    acc: dict[str, Any],
-) -> None:
+def _update_pod_issue(pod: dict[str, Any], acc: dict[str, Any]) -> None:
     metadata = pod.get("metadata") if isinstance(pod.get("metadata"), dict) else {}
     status = pod.get("status") if isinstance(pod.get("status"), dict) else {}
     spec = pod.get("spec") if isinstance(pod.get("spec"), dict) else {}

tests/unit/services/test_cluster_state_kubernetes_domains.py

@@ -140,6 +140,62 @@ def test_pod_summaries_and_issue_detection() -> None:
     assert issues["waiting_reasons"]["CrashLoopBackOff"] == 1
 
 
+def test_pod_summary_edge_filters() -> None:
+    assert pods._owner_reference({"ownerReferences": [None, {}]}) == ("", "")
+    payload = {
+        "items": [
+            {
+                "metadata": {"name": "ignored", "namespace": "kube-system", "labels": {"app": "system"}},
+                "spec": {"nodeName": "titan-1"},
+                "status": {"phase": "Running"},
+            },
+            {
+                "metadata": {"name": "anonymous", "namespace": "apps"},
+                "spec": {"nodeName": "titan-1"},
+                "status": {"phase": "Running"},
+            },
+            {
+                "metadata": {
+                    "name": "failed",
+                    "namespace": "apps",
+                    "ownerReferences": [{"name": "batch", "kind": "Job"}],
+                },
+                "spec": {},
+                "status": {"phase": "Failed", "containerStatuses": [None]},
+            },
+            {
+                "metadata": {"name": "done", "namespace": "apps", "labels": {"app": "done"}},
+                "spec": {"nodeName": ""},
+                "status": {"phase": "Succeeded"},
+            },
+            {
+                "metadata": {"name": "waiting", "namespace": "apps", "labels": {"app": "waiting"}, "creationTimestamp": ""},
+                "spec": {"nodeName": "titan-2"},
+                "status": {"phase": "Pending", "containerStatuses": [None, {"restartCount": 0, "state": {}}]},
+            },
+            {"metadata": {"namespace": "apps"}, "status": {"phase": "Pending"}},
+        ]
+    }
+
+    workloads_summary = pods._summarize_workloads(payload)
+    assert {entry["workload"] for entry in workloads_summary} == {"batch", "done", "waiting"}
+
+    namespace_pods = pods._summarize_namespace_pods(payload)[0]
+    assert namespace_pods["pods_failed"] == 1
+    assert namespace_pods["pods_succeeded"] == 1
+
+    namespace_nodes = pods._summarize_namespace_nodes(payload)[0]
+    assert namespace_nodes["primary_node"] == "titan-1"
+
+    node_pods = pods._summarize_node_pods(payload)
+    assert [entry["node"] for entry in node_pods] == ["titan-1", "titan-2"]
+    assert pods._node_pods_top([None, node_pods[0]])[0]["node"] == "titan-1"
+
+    issues = pods._summarize_pod_issues(payload)
+    assert issues["counts"]["Failed"] == 1
+    assert issues["pending_over_15m"] == 0
+
+
 def test_workload_job_longhorn_and_fetch_summaries(monkeypatch) -> None:
     jobs = workloads._summarize_jobs(
         {