from __future__ import annotations from ariadne.services import cluster_state_fetchers as fetchers from ariadne.services import cluster_state_flux_events as flux_events from ariadne.services import cluster_state_nodes as nodes from ariadne.services import cluster_state_pods as pods from ariadne.services import cluster_state_workloads as workloads def test_node_summary_inventory_and_hardware_usage() -> None: payload = { "items": [ { "metadata": { "name": "titan-1", "labels": {"kubernetes.io/arch": "arm64", "hardware": "rpi5"}, "creationTimestamp": "2026-01-01T00:00:00Z", }, "spec": {"unschedulable": True, "taints": [{"key": "dedicated", "effect": "NoSchedule"}]}, "status": { "conditions": [ {"type": "Ready", "status": "True"}, {"type": "DiskPressure", "status": "True", "reason": "Full", "message": "disk full"}, ], "capacity": {"cpu": "4", "memory": "8Gi", "pods": "110"}, "allocatable": {"cpu": "3", "memory": "7Gi"}, "nodeInfo": {"architecture": "arm64"}, "addresses": [{"type": "InternalIP", "address": "10.0.0.1"}], }, }, {"metadata": {"name": "titan-2", "labels": {}}, "status": {"conditions": []}}, ] } summary = nodes._summarize_nodes(payload) assert summary["total"] == 2 assert summary["not_ready"] == 1 details = nodes._node_details(payload) assert details[0]["pressure"]["DiskPressure"] is True assert details[0]["taints"][0]["key"] == "dedicated" assert nodes._node_labels({"node-role.kubernetes.io/control-plane": "", "other": "skip"}) assert nodes._node_addresses({"addresses": [{"type": "Hostname", "address": "titan-1"}]}) == {"Hostname": "titan-1"} assert nodes._node_capacity({"cpu": "4", "unknown": "skip"}) == {"cpu": "4"} assert nodes._node_pressure_conditions([{"type": "PIDPressure", "status": "True"}])["PIDPressure"] is True assert nodes._node_roles({"node-role.kubernetes.io/control-plane": ""}) == ["control-plane"] assert nodes._node_is_worker({"node-role.kubernetes.io/control-plane": ""}) is False assert nodes._hardware_hint({"jetson": "true"}, {"architecture": "arm64"}) == "jetson" assert nodes._condition_status([], "Ready") == (None, "", "") assert nodes._age_hours("not-a-date") is None assert nodes._node_age_stats(details)["oldest"] assert nodes._node_flagged(details, "unschedulable") == ["titan-1"] assert nodes._summarize_inventory(details)["unschedulable_nodes"] == ["titan-1"] assert nodes._hardware_groups(details)[0]["hardware"] assert nodes._pressure_summary(nodes._summarize_inventory(details))["total"] == 1 usage = nodes._node_usage_by_hardware([{"node": "titan-1", "cpu": 80.0, "load_index": 0.5}], details) assert usage[0]["hardware"] == "rpi5" def test_node_summary_edge_filters_and_hardware_hints() -> None: usage = nodes._node_usage_by_hardware( [None, {"node": ""}, {"node": "titan-1", "cpu": 50.0, "load_index": 0.5}], [None, {"name": "titan-1", "hardware": ""}], ) assert usage[0]["hardware"] == "unknown" assert nodes._hardware_map([None, {"name": "titan-1", "hardware": "rpi5"}]) == {"titan-1": "rpi5"} assert nodes._node_ready("bad") is False assert nodes._node_ready([None]) is False summary = nodes._summarize_nodes({"items": [{"metadata": {}}, {"metadata": {"name": "titan-1"}, "status": {"conditions": "bad"}}]}) assert summary["not_ready_names"] == ["titan-1"] assert nodes._node_labels("bad") == {} assert nodes._node_addresses({"addresses": [None, {"type": "InternalIP", "address": "10.0.0.1"}]}) == {"InternalIP": "10.0.0.1"} assert nodes._node_details({"items": [{"metadata": {}}, {"metadata": {"name": "titan-1"}, "status": {}, "spec": {}}]})[0]["name"] == "titan-1" assert nodes._node_flagged([None, {"name": ""}, {"name": "titan-1", "taints": [{"key": "dedicated"}]}], "taints") == ["titan-1"] assert nodes._node_taints([None, {"key": "dedicated", "effect": "NoSchedule", "value": 5}]) == [ {"key": "dedicated", "value": "", "effect": "NoSchedule"} ] assert nodes._hardware_groups([None, {"name": ""}, {"name": "titan-1", "hardware": "rpi5"}]) == [ {"hardware": "rpi5", "count": 1, "nodes": ["titan-1"]} ] inventory = nodes._summarize_inventory( [ {"name": "", "ready": True}, {"name": "worker", "ready": True, "is_worker": True, "roles": ["worker"], "pressure": "bad"}, { "name": "master", "ready": False, "is_worker": False, "roles": ["control-plane"], "pressure": {"DiskPressure": True}, "taints": [{"key": "dedicated"}], "unschedulable": True, }, ] ) assert inventory["workers"] == {"total": 1, "ready": 1} assert inventory["by_role"]["worker"] == 1 assert inventory["pressure_nodes"]["DiskPressure"] == ["master"] assert nodes._node_pressure_conditions("bad") == {} assert nodes._node_pressure_conditions([None, {"type": "DiskPressure", "status": "True"}]) == {"DiskPressure": True} assert nodes._node_is_worker({"node-role.kubernetes.io/master": ""}) is False assert nodes._node_is_worker({"node-role.kubernetes.io/worker": ""}) is True assert nodes._hardware_hint({}, {"kernelVersion": "tegra", "osImage": ""}) == "jetson" assert nodes._hardware_hint({}, {"kernelVersion": "raspi", "osImage": ""}) == "rpi" assert nodes._condition_status("bad", "Ready") == (None, "", "") assert nodes._condition_status([None, {"type": "DiskPressure"}], "Ready") == (None, "", "") assert nodes._condition_status([{"type": "Ready", "status": "Unknown", "reason": "Checking", "message": "wait"}], "Ready") == ( None, "Checking", "wait", ) def test_flux_and_event_summaries() -> None: flux_payload = { "items": [ { "metadata": {"name": "apps", "namespace": "flux-system"}, "spec": {"suspend": False}, "status": {"conditions": [{"type": "Ready", "status": "True"}]}, }, { "metadata": {"name": "broken", "namespace": "flux-system"}, "spec": {"suspend": True}, "status": {"conditions": [{"type": "Ready", "status": "False", "reason": "Bad", "message": "no"}]}, }, ] } assert flux_events._summarize_kustomizations(flux_payload)["not_ready"] == 1 assert flux_events._namespace_allowed("apps") is True assert flux_events._namespace_allowed("kube-system") is False assert flux_events._event_sort_key("bad") == 0.0 events = flux_events._summarize_events( { "items": [ { "metadata": {"namespace": "apps"}, "type": "Warning", "reason": "BackOff", "message": "retry", "count": 2, "lastTimestamp": "2026-01-01T00:00:00Z", "involvedObject": {"kind": "Pod", "name": "api"}, }, {"metadata": {"namespace": "kube-system"}, "type": "Warning", "reason": "Ignored"}, {"metadata": {"namespace": "apps"}, "type": "Normal"}, ] } ) assert events["warnings_total"] == 1 assert events["warnings_top_reason"]["reason"] == "BackOff" def test_pod_summaries_and_issue_detection() -> None: payload = { "items": [ { "metadata": { "name": "api-1", "namespace": "apps", "labels": {"app": "api"}, "creationTimestamp": "2026-01-01T00:00:00Z", }, "spec": {"nodeName": "titan-1"}, "status": { "phase": "Pending", "reason": "Unschedulable", "containerStatuses": [ {"restartCount": 2, "state": {"waiting": {"reason": "CrashLoopBackOff"}}} ], }, }, { "metadata": { "name": "worker-1", "namespace": "apps", "ownerReferences": [{"name": "worker", "kind": "ReplicaSet"}], }, "spec": {"nodeName": "titan-2"}, "status": {"phase": "Running"}, }, ] } assert pods._workload_from_labels({"app": "api"}) == ("api", "label:app") assert pods._owner_reference({"ownerReferences": [{"name": "rs", "kind": "ReplicaSet"}]}) == ("rs", "owner:ReplicaSet") assert pods._pod_workload({"labels": {}, "ownerReferences": [{"name": "rs"}]})[0] == "rs" assert pods._summarize_workloads(payload)[0]["workload"] == "api" assert pods._summarize_namespace_pods(payload)[0]["pods_total"] == 2 assert pods._summarize_namespace_nodes(payload)[0]["primary_node"] node_pods = pods._summarize_node_pods(payload) assert pods._node_pods_top(node_pods)[0]["node"] == "titan-1" issues = pods._summarize_pod_issues(payload) assert issues["counts"]["Pending"] == 1 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( { "items": [ { "metadata": {"name": "backup", "namespace": "apps", "creationTimestamp": "2026-01-01T00:00:00Z"}, "status": {"failed": 1, "succeeded": 0, "active": 1}, } ] } ) assert jobs["totals"]["failed"] == 1 deployments = workloads._summarize_deployments( {"items": [{"metadata": {"name": "api", "namespace": "apps"}, "spec": {"replicas": 2}, "status": {"readyReplicas": 1}}]} ) statefulsets = workloads._summarize_statefulsets( {"items": [{"metadata": {"name": "db", "namespace": "apps"}, "spec": {"replicas": 2}, "status": {"readyReplicas": 1}}]} ) daemonsets = workloads._summarize_daemonsets( {"items": [{"metadata": {"name": "agent", "namespace": "apps"}, "status": {"desiredNumberScheduled": 2, "numberReady": 1}}]} ) health = workloads._summarize_workload_health(deployments, statefulsets, daemonsets) assert health["deployments"]["not_ready"] == 1 longhorn = workloads._summarize_longhorn_volumes( { "items": [ { "metadata": {"name": "pvc-data"}, "spec": {"size": "1Gi"}, "status": {"state": "attached", "robustness": "degraded", "actualSize": "500Mi"}, } ] } ) assert longhorn["degraded_count"] == 1 def fake_get_json(path: str): if path.endswith("/nodes"): return {"items": []} if path.startswith("/api/v1/pods"): return {"items": []} if path.startswith("/apis/batch"): return {"items": []} if "longhorn" in path: return {"items": []} if "deployments" in path or "statefulsets" in path or "daemonsets" in path: return {"items": []} if path.startswith("/api/v1/events"): return {"items": []} return {"items": []} monkeypatch.setattr(fetchers, "_get_json", fake_get_json) errors: list[str] = [] assert fetchers._fetch_nodes(errors)[0]["total"] == 0 assert fetchers._fetch_flux(errors)["not_ready"] == 0 assert fetchers._fetch_pods(errors)[0] == [] assert fetchers._fetch_jobs(errors)["totals"]["total"] == 0 assert fetchers._fetch_longhorn(errors) == {} assert fetchers._fetch_workload_health(errors)["deployments"]["total"] == 0 assert fetchers._fetch_events(errors)["warnings_total"] == 0 assert errors == [] monkeypatch.setattr(fetchers, "_get_json", lambda _path: (_ for _ in ()).throw(RuntimeError("boom"))) errors = [] assert fetchers._fetch_jobs(errors) == {} assert errors == ["jobs: boom"]