""" tests/test_nodes.py Unit tests for fetch_nodes(). No real cluster needed — all Kubernetes API calls are mocked. Run: pytest tests/test_nodes.py -v """ from datetime import datetime, timezone, timedelta from unittest.mock import MagicMock, patch import pytest from app.k8s_client import fetch_nodes, _parse_millicores, _parse_mebibytes # ── Fixtures ─────────────────────────────────────────────────────────────────── def _make_condition(type_: str, status: str): c = MagicMock() c.type = type_ c.status = status return c def _make_node( name="node-1", ready_status="True", cpu="4", memory="8058752Ki", alloc_memory="7542528Ki", os_image="Ubuntu 22.04.3 LTS", kubelet_version="v1.28.5+k3s1", labels=None, created_days_ago=10, mem_pressure="False", disk_pressure="False", pid_pressure="False", ): node = MagicMock() node.metadata.name = name node.metadata.labels = labels if labels is not None else {"node-role.kubernetes.io/control-plane": "true"} node.metadata.creation_timestamp = datetime.now(timezone.utc) - timedelta( days=created_days_ago ) node.status.conditions = [ _make_condition("Ready", ready_status), _make_condition("MemoryPressure", mem_pressure), _make_condition("DiskPressure", disk_pressure), _make_condition("PIDPressure", pid_pressure), ] node.status.capacity = {"cpu": cpu, "memory": memory} node.status.allocatable = {"cpu": cpu, "memory": alloc_memory} node.status.node_info.os_image = os_image node.status.node_info.kubelet_version = kubelet_version return node @pytest.fixture def mock_core(monkeypatch): """Patch get_core() so no real kubeconfig is needed.""" mock = MagicMock() monkeypatch.setattr("app.k8s_client.get_core", lambda: mock) return mock # ── Parser unit tests (no mock needed) ──────────────────────────────────────── class TestParsers: def test_millicores_with_suffix(self): assert _parse_millicores("250m") == pytest.approx(0.25) def test_millicores_without_suffix(self): assert _parse_millicores("4") == pytest.approx(4.0) def test_mebibytes_mi(self): assert _parse_mebibytes("512Mi") == pytest.approx(512.0) def test_mebibytes_gi(self): assert _parse_mebibytes("1Gi") == pytest.approx(1024.0) def test_mebibytes_ki(self): assert _parse_mebibytes("1024Ki") == pytest.approx(1.0) def test_mebibytes_raw_bytes(self): assert _parse_mebibytes("1048576") == pytest.approx(1.0) # ── fetch_nodes() tests ──────────────────────────────────────────────────────── class TestFetchNodes: def test_returns_list_of_dicts(self, mock_core): mock_core.list_node.return_value.items = [_make_node()] result = fetch_nodes() assert isinstance(result, list) assert isinstance(result[0], dict) def test_ready_node(self, mock_core): mock_core.list_node.return_value.items = [_make_node(ready_status="True")] node = fetch_nodes()[0] assert node["ready"] is True def test_not_ready_node(self, mock_core): mock_core.list_node.return_value.items = [_make_node(ready_status="False")] node = fetch_nodes()[0] assert node["ready"] is False def test_ready_is_bool_not_string(self, mock_core): """Regression: status comes back as string 'True' — must be cast to bool.""" mock_core.list_node.return_value.items = [_make_node(ready_status="True")] node = fetch_nodes()[0] assert node["ready"] is True assert not isinstance(node["ready"], str) def test_node_name(self, mock_core): mock_core.list_node.return_value.items = [_make_node(name="k3s-worker-1")] assert fetch_nodes()[0]["name"] == "k3s-worker-1" def test_control_plane_role(self, mock_core): mock_core.list_node.return_value.items = [ _make_node(labels={"node-role.kubernetes.io/control-plane": "true"}) ] assert "control-plane" in fetch_nodes()[0]["roles"] def test_worker_role_fallback(self, mock_core): """Nodes with no role label should fall back to 'worker'.""" mock_core.list_node.return_value.items = [_make_node(labels={})] assert fetch_nodes()[0]["roles"] == "worker" def test_memory_parsed_to_float(self, mock_core): mock_core.list_node.return_value.items = [_make_node(memory="8058752Ki")] node = fetch_nodes()[0] assert isinstance(node["mem_cap_mi"], float) assert node["mem_cap_mi"] > 0 def test_cpu_parsed_to_float(self, mock_core): mock_core.list_node.return_value.items = [_make_node(cpu="4")] assert fetch_nodes()[0]["cpu_cap"] == pytest.approx(4.0) def test_age_days(self, mock_core): mock_core.list_node.return_value.items = [_make_node(created_days_ago=5)] assert fetch_nodes()[0]["age"] == "5d" def test_age_hours(self, mock_core): node = _make_node() node.metadata.creation_timestamp = datetime.now(timezone.utc) - timedelta( hours=3 ) mock_core.list_node.return_value.items = [node] assert fetch_nodes()[0]["age"] == "3h" def test_pressure_fields_present(self, mock_core): mock_core.list_node.return_value.items = [_make_node()] node = fetch_nodes()[0] assert "mem_pressure" in node assert "disk_pressure" in node assert "pid_pressure" in node def test_memory_pressure_true(self, mock_core): mock_core.list_node.return_value.items = [_make_node(mem_pressure="True")] assert fetch_nodes()[0]["mem_pressure"] == "True" def test_multiple_nodes(self, mock_core): mock_core.list_node.return_value.items = [ _make_node(name="node-1"), _make_node(name="node-2"), ] result = fetch_nodes() assert len(result) == 2 assert {r["name"] for r in result} == {"node-1", "node-2"} def test_empty_cluster(self, mock_core): mock_core.list_node.return_value.items = [] assert fetch_nodes() == [] def test_null_conditions_handled(self, mock_core): """Nodes mid-join can have conditions=None — must not raise.""" node = _make_node() node.status.conditions = None mock_core.list_node.return_value.items = [node] result = fetch_nodes() assert result[0]["ready"] is False def test_null_labels_handled(self, mock_core): node = _make_node() node.metadata.labels = None mock_core.list_node.return_value.items = [node] assert fetch_nodes()[0]["roles"] == "worker" def test_all_keys_present(self, mock_core): """Every dict must have the same stable set of keys.""" mock_core.list_node.return_value.items = [_make_node()] node = fetch_nodes()[0] expected_keys = { "name", "ready", "roles", "age", "cpu_cap", "mem_cap_mi", "mem_alloc_mi", "os", "kubelet_version", "mem_pressure", "disk_pressure", "pid_pressure", } assert expected_keys == set(node.keys())