mirror of
https://github.com/Rainyy21/k3s-health-dashboard.git
synced 2026-10-10 23:00:19 -04:00
Rainy21
249 lines
8.9 KiB
Python
249 lines
8.9 KiB
Python
"""
|
|
tests/test_pods.py
|
|
|
|
Unit tests for fetch_pods().
|
|
No real cluster needed — all Kubernetes API calls are mocked.
|
|
|
|
Run:
|
|
pytest tests/test_pods.py -v
|
|
"""
|
|
|
|
from datetime import datetime, timezone, timedelta
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from app.k8s_client import fetch_pods
|
|
|
|
# ── Fixtures ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_container_status(restart_count=0, ready=True):
|
|
cs = MagicMock()
|
|
cs.restart_count = restart_count
|
|
cs.ready = ready
|
|
return cs
|
|
|
|
|
|
def _make_pod(
|
|
name="pod-1",
|
|
namespace="default",
|
|
phase="Running",
|
|
ready_status="True",
|
|
restarts=0,
|
|
node_name="k3s-worker-1",
|
|
created_minutes_ago=60,
|
|
container_statuses=None,
|
|
):
|
|
pod = MagicMock()
|
|
pod.metadata.name = name
|
|
pod.metadata.namespace = namespace
|
|
pod.metadata.creation_timestamp = datetime.now(timezone.utc) - timedelta(
|
|
minutes=created_minutes_ago
|
|
)
|
|
pod.status.phase = phase
|
|
pod.status.conditions = [
|
|
_cond("Ready", ready_status),
|
|
_cond("ContainersReady", ready_status),
|
|
_cond("PodScheduled", "True"),
|
|
_cond("Initialized", "True"),
|
|
]
|
|
pod.status.container_statuses = container_statuses or [
|
|
_make_container_status(restart_count=restarts)
|
|
]
|
|
pod.spec.node_name = node_name
|
|
return pod
|
|
|
|
|
|
def _cond(type_: str, status: str):
|
|
c = MagicMock()
|
|
c.type = type_
|
|
c.status = status
|
|
return c
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_core(monkeypatch):
|
|
mock = MagicMock()
|
|
monkeypatch.setattr("app.k8s_client.get_core", lambda: mock)
|
|
return mock
|
|
|
|
|
|
# ── fetch_pods() — return shape ────────────────────────────────────────────────
|
|
|
|
|
|
class TestFetchPodsShape:
|
|
def test_returns_list_of_dicts(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [_make_pod()]
|
|
result = fetch_pods("all")
|
|
assert isinstance(result, list)
|
|
assert isinstance(result[0], dict)
|
|
|
|
def test_all_keys_present(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [_make_pod()]
|
|
pod = fetch_pods("all")[0]
|
|
assert set(pod.keys()) == {
|
|
"namespace",
|
|
"name",
|
|
"ready",
|
|
"phase",
|
|
"restarts",
|
|
"node",
|
|
"age",
|
|
}
|
|
|
|
def test_empty_result(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = []
|
|
assert fetch_pods("all") == []
|
|
|
|
|
|
# ── fetch_pods() — namespace routing ──────────────────────────────────────────
|
|
|
|
|
|
class TestNamespaceRouting:
|
|
def test_all_namespace_uses_cluster_wide_call(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = []
|
|
fetch_pods("all")
|
|
mock_core.list_pod_for_all_namespaces.assert_called_once()
|
|
mock_core.list_namespaced_pod.assert_not_called()
|
|
|
|
def test_specific_namespace_uses_namespaced_call(self, mock_core):
|
|
mock_core.list_namespaced_pod.return_value.items = []
|
|
fetch_pods("kube-system")
|
|
mock_core.list_namespaced_pod.assert_called_once_with(namespace="kube-system")
|
|
mock_core.list_pod_for_all_namespaces.assert_not_called()
|
|
|
|
def test_namespace_preserved_in_result(self, mock_core):
|
|
mock_core.list_namespaced_pod.return_value.items = [
|
|
_make_pod(namespace="kube-system")
|
|
]
|
|
pod = fetch_pods("kube-system")[0]
|
|
assert pod["namespace"] == "kube-system"
|
|
|
|
|
|
# ── fetch_pods() — readiness ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestPodReadiness:
|
|
def test_ready_pod(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(ready_status="True")
|
|
]
|
|
assert fetch_pods("all")[0]["ready"] is True
|
|
|
|
def test_not_ready_pod(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(ready_status="False")
|
|
]
|
|
assert fetch_pods("all")[0]["ready"] is False
|
|
|
|
def test_ready_is_bool_not_string(self, mock_core):
|
|
"""Regression: condition.status is the string 'True', must be cast to bool."""
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(ready_status="True")
|
|
]
|
|
assert fetch_pods("all")[0]["ready"] is True
|
|
assert not isinstance(fetch_pods("all")[0]["ready"], str)
|
|
|
|
def test_running_but_not_ready(self, mock_core):
|
|
"""phase=Running does NOT mean ready. This is the key gotcha."""
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(phase="Running", ready_status="False")
|
|
]
|
|
pod = fetch_pods("all")[0]
|
|
assert pod["phase"] == "Running"
|
|
assert pod["ready"] is False # not the same thing
|
|
|
|
|
|
# ── fetch_pods() — phase ───────────────────────────────────────────────────────
|
|
|
|
|
|
class TestPodPhase:
|
|
@pytest.mark.parametrize("phase", ["Running", "Pending", "Failed", "Succeeded"])
|
|
def test_phase_preserved(self, mock_core, phase):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(phase=phase)
|
|
]
|
|
assert fetch_pods("all")[0]["phase"] == phase
|
|
|
|
def test_none_phase_becomes_unknown(self, mock_core):
|
|
pod = _make_pod()
|
|
pod.status.phase = None
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [pod]
|
|
assert fetch_pods("all")[0]["phase"] == "Unknown"
|
|
|
|
|
|
# ── fetch_pods() — restarts ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestPodRestarts:
|
|
def test_zero_restarts(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(restarts=0)
|
|
]
|
|
assert fetch_pods("all")[0]["restarts"] == 0
|
|
|
|
def test_nonzero_restarts(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(restarts=7)
|
|
]
|
|
assert fetch_pods("all")[0]["restarts"] == 7
|
|
|
|
def test_restarts_summed_across_containers(self, mock_core):
|
|
"""Multi-container pods sum restarts from all containers."""
|
|
pod = _make_pod(
|
|
container_statuses=[
|
|
_make_container_status(restart_count=3),
|
|
_make_container_status(restart_count=5),
|
|
]
|
|
)
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [pod]
|
|
assert fetch_pods("all")[0]["restarts"] == 8
|
|
|
|
def test_null_container_statuses_handled(self, mock_core):
|
|
"""Pending pods have no container_statuses yet — must not raise."""
|
|
pod = _make_pod()
|
|
pod.status.container_statuses = None
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [pod]
|
|
assert fetch_pods("all")[0]["restarts"] == 0
|
|
|
|
|
|
# ── fetch_pods() — edge cases ──────────────────────────────────────────────────
|
|
|
|
|
|
class TestPodEdgeCases:
|
|
def test_null_conditions_handled(self, mock_core):
|
|
pod = _make_pod()
|
|
pod.status.conditions = None
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [pod]
|
|
result = fetch_pods("all")
|
|
assert result[0]["ready"] is False
|
|
|
|
def test_null_node_name(self, mock_core):
|
|
"""Pending pods are not yet scheduled to a node."""
|
|
pod = _make_pod()
|
|
pod.spec.node_name = None
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [pod]
|
|
assert fetch_pods("all")[0]["node"] == "—"
|
|
|
|
def test_multiple_pods(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(name="pod-a", namespace="default"),
|
|
_make_pod(name="pod-b", namespace="kube-system"),
|
|
]
|
|
result = fetch_pods("all")
|
|
assert len(result) == 2
|
|
assert {r["name"] for r in result} == {"pod-a", "pod-b"}
|
|
|
|
def test_age_minutes(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(created_minutes_ago=25)
|
|
]
|
|
assert fetch_pods("all")[0]["age"] == "25m"
|
|
|
|
def test_age_hours(self, mock_core):
|
|
mock_core.list_pod_for_all_namespaces.return_value.items = [
|
|
_make_pod(created_minutes_ago=180)
|
|
]
|
|
assert fetch_pods("all")[0]["age"] == "3h"
|