From 5f29d23b2b72b2cbb99fd5a0c0a9d5a729fcc503 Mon Sep 17 00:00:00 2001 From: lkoerber Date: Thu, 6 Aug 2026 12:24:36 +0200 Subject: [PATCH 1/2] fix: catch IndexError in read_uptime_s and read_meminfo like read_load already does An empty /proc/uptime (f.read().split() -> []) or a MemTotal:/ MemAvailable: line with nothing after the colon raised an uncaught IndexError and 500ed the node agent's /metrics endpoint, while the identical parsing pattern in read_load() already degrades gracefully by listing IndexError in its except clause. Both readers now do the same. Found by the new unit test suite; regression-tested there. --- backend/app/node_agent.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/backend/app/node_agent.py b/backend/app/node_agent.py index 04678e0..95112ec 100644 --- a/backend/app/node_agent.py +++ b/backend/app/node_agent.py @@ -51,7 +51,9 @@ def read_meminfo() -> dict[str, int]: key, _, rest = line.partition(":") if key in ("MemTotal", "MemAvailable"): out[key] = int(rest.strip().split()[0]) # kB - except (OSError, ValueError): + # IndexError included like in read_load(): a malformed line with nothing after the + # colon must degrade to a partial/empty dict, not 500 the /metrics endpoint. + except (OSError, ValueError, IndexError): pass return out @@ -60,7 +62,9 @@ def read_uptime_s() -> int | None: try: with open(f"{PROC}/uptime") as f: return int(float(f.read().split()[0])) - except (OSError, ValueError): + # IndexError included like in read_load(): an empty /proc/uptime must degrade to + # None, not 500 the /metrics endpoint. + except (OSError, ValueError, IndexError): return None From 3e3638bb49cedddcb0023ea9dc62007531b871bb Mon Sep 17 00:00:00 2001 From: lkoerber Date: Thu, 6 Aug 2026 12:24:36 +0200 Subject: [PATCH 2/2] test: raise backend coverage from 50% to ~100% 85 new tests in two files, no production changes beyond the preceding node-agent fix: - test_units.py (48): node_agent (was 0% - all /proc//sys readers via tmp_path-redirected paths, platform-independent), hardware and healthcheck collector loops (broken out of while True by making asyncio.sleep raise a sentinel; real loopback servers for HTTP/TCP success and refused paths, patched timeouts for the timeout paths), auth token issue/verify edges, state's QueueFull drop-oldest and broken-queue swallow paths, demo's fake_logs generator. - test_framework.py (37): k8s_watch mapping functions against SimpleNamespace-shaped fakes plus the full watch loop with scripted fake watches (retry/backoff, unknown-kind ValueError, CancelledError passthrough, backoff growth), ws connect/full-state/4401/heartbeat/ demo-log-streaming/send-failure paths via TestClient websockets and a direct fake-WebSocket invocation for the disconnect branch, metrics-server payload parsing incl. malformed-usage and API-error resilience, and main.py's cluster-detection/lifespan/readyz/SPA- fallback branches with per-test fresh ClusterState instances. Backend suite: 18 -> 103 tests, coverage 49.6% -> ~99.7% (a single timing-dependent healthcheck branch occasionally reports uncovered). Verified in the CI-equivalent python:3.13-slim container; ruff clean. --- backend/tests/test_framework.py | 941 ++++++++++++++++++++++++++++++++ backend/tests/test_units.py | 768 ++++++++++++++++++++++++++ 2 files changed, 1709 insertions(+) create mode 100644 backend/tests/test_framework.py create mode 100644 backend/tests/test_units.py diff --git a/backend/tests/test_framework.py b/backend/tests/test_framework.py new file mode 100644 index 0000000..b9460cf --- /dev/null +++ b/backend/tests/test_framework.py @@ -0,0 +1,941 @@ +"""Coverage-focused tests for k8s_watch, ws, metrics collectors and main's +startup-mode branches. + +Follows the house style of test_backend.py: plain pytest functions, +`asyncio.run(scenario())` for ad-hoc async scenarios, `importlib.reload` +plus `monkeypatch.setenv` for modules that read environment variables at +import time, and `TestClient` for HTTP/WebSocket behavior. + +Kubernetes interactions are always faked -- these tests never talk to a +real cluster or apiserver. +""" +from __future__ import annotations + +import asyncio +import importlib +import os +import sys +import types +from datetime import datetime, timezone +from typing import ClassVar + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + + +# ================================================================== +# app.collectors.k8s_watch +# ================================================================== + + +def _node_obj(name="pi-1", role_label=True, with_optional=True): + """Build a fake kubernetes_asyncio V1Node-shaped object.""" + conditions = [types.SimpleNamespace(type="Ready", status="True")] + labels = {"node-role.kubernetes.io/control-plane": ""} if role_label else {} + addresses = [types.SimpleNamespace(type="InternalIP", address="192.168.1.10")] + node_info = ( + types.SimpleNamespace( + architecture="arm64", kubelet_version="v1.29.4+k3s1", os_image="Debian 12" + ) + if with_optional + else None + ) + status = types.SimpleNamespace( + conditions=conditions, + addresses=addresses, + node_info=node_info, + capacity={"cpu": "4", "memory": "8Gi"} if with_optional else None, + ) + metadata = types.SimpleNamespace( + name=name, + labels=labels, + creation_timestamp=datetime.now(timezone.utc) if with_optional else None, + ) + spec = types.SimpleNamespace(unschedulable=False) + return types.SimpleNamespace(status=status, metadata=metadata, spec=spec) + + +def _pod_obj(name="p1", namespace="default", waiting=False, no_statuses=False): + if no_statuses: + statuses = [] + else: + state = types.SimpleNamespace( + waiting=types.SimpleNamespace(reason="CrashLoopBackOff") if waiting else None + ) + statuses = [ + types.SimpleNamespace(restart_count=2, ready=not waiting, state=state) + ] + metadata = types.SimpleNamespace( + name=name, namespace=namespace, creation_timestamp=datetime.now(timezone.utc) + ) + spec = types.SimpleNamespace( + node_name="pi-1", containers=[types.SimpleNamespace(name="app")] + ) + status = types.SimpleNamespace( + container_statuses=statuses, phase="Running", reason=None + ) + return types.SimpleNamespace(metadata=metadata, spec=spec, status=status) + + +def _deployment_obj(name="d1", namespace="default"): + metadata = types.SimpleNamespace(name=name, namespace=namespace) + containers = [types.SimpleNamespace(image="registry.local/app:latest")] + template = types.SimpleNamespace(spec=types.SimpleNamespace(containers=containers)) + spec = types.SimpleNamespace(replicas=2, template=template) + status = types.SimpleNamespace(ready_replicas=2, available_replicas=2, updated_replicas=2) + return types.SimpleNamespace(metadata=metadata, spec=spec, status=status) + + +def _event_obj(uid="evt-1"): + metadata = types.SimpleNamespace(uid=uid, creation_timestamp=None) + involved = types.SimpleNamespace(kind="Pod", name="p1", namespace="default") + return types.SimpleNamespace( + metadata=metadata, + last_timestamp=datetime.now(timezone.utc), + event_time=None, + type="Normal", + reason="Scheduled", + message="assigned", + involved_object=involved, + count=3, + ) + + +def test_map_node_with_role_label_and_optionals(): + from app.collectors.k8s_watch import map_node + + d = map_node(_node_obj(role_label=True, with_optional=True)) + assert d["ready"] is True + assert d["roles"] == ["control-plane"] + assert d["arch"] == "arm64" + assert d["internal_ip"] == "192.168.1.10" + assert d["cpu_capacity"] == "4" + assert d["created"] is not None + + +def test_map_node_defaults_worker_role_no_optionals(): + from app.collectors.k8s_watch import map_node + + d = map_node(_node_obj(role_label=False, with_optional=False)) + assert d["roles"] == ["worker"] # no node-role.* label -> default fallback + assert d["arch"] is None + assert d["kubelet"] is None + assert d["os_image"] is None + assert d["cpu_capacity"] is None + assert d["created"] is None + + +def test_map_pod_waiting_reason_and_ready_ratio(): + from app.collectors.k8s_watch import map_pod + + d = map_pod(_pod_obj(waiting=True)) + assert d["key"] == "default/p1" + assert d["reason"] == "CrashLoopBackOff" + assert d["ready"] == "0/1" + assert d["restarts"] == 2 + assert d["containers"] == ["app"] + + +def test_map_pod_no_container_statuses(): + from app.collectors.k8s_watch import map_pod + + d = map_pod(_pod_obj(no_statuses=True)) + assert d["ready"] == "0/0" + assert d["restarts"] == 0 + assert d["reason"] is None + + +def test_map_deployment(): + from app.collectors.k8s_watch import map_deployment + + d = map_deployment(_deployment_obj()) + assert d["key"] == "default/d1" + assert d["replicas"] == 2 + assert d["images"] == ["registry.local/app:latest"] + + +def test_map_event_uses_last_timestamp(): + from app.collectors.k8s_watch import map_event + + d = map_event(_event_obj()) + assert d["object"] == "Pod/p1" + assert d["count"] == 3 + assert d["t"] is not None + + +def test_apply_all_kinds_upsert_and_delete(): + """Directly exercises _apply's dispatch table for every resource kind.""" + from app.collectors.k8s_watch import _apply + from app.state import ClusterState + + st = ClusterState() + + _apply(st, "nodes", "ADDED", {"name": "pi-1"}) + assert "pi-1" in st.nodes + _apply(st, "nodes", "DELETED", {"name": "pi-1"}) + assert "pi-1" not in st.nodes + + _apply(st, "pods", "ADDED", {"key": "ns/p1"}) + assert "ns/p1" in st.pods + _apply(st, "pods", "DELETED", {"key": "ns/p1"}) + assert "ns/p1" not in st.pods + + _apply(st, "deployments", "ADDED", {"key": "ns/d1"}) + assert "ns/d1" in st.deployments + _apply(st, "deployments", "DELETED", {"key": "ns/d1"}) + assert "ns/d1" not in st.deployments + + _apply(st, "events", "ADDED", {"uid": "e1"}) + assert len(st.events) == 1 + # a DELETED event is a no-op: events have no delete semantics + _apply(st, "events", "DELETED", {"uid": "e2"}) + assert len(st.events) == 1 + + +def test_load_config_incluster(monkeypatch): + from kubernetes_asyncio import config as kconfig + + from app.collectors import k8s_watch + + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.0.0.1") + calls = [] + monkeypatch.setattr(kconfig, "load_incluster_config", lambda: calls.append("incluster")) + asyncio.run(k8s_watch.load_config()) + assert calls == ["incluster"] + + +def test_load_config_kubeconfig(monkeypatch): + from kubernetes_asyncio import config as kconfig + + from app.collectors import k8s_watch + + monkeypatch.delenv("KUBERNETES_SERVICE_HOST", raising=False) + + calls = [] + + async def fake_load_kube_config(): + calls.append("kubeconfig") + + monkeypatch.setattr(kconfig, "load_kube_config", fake_load_kube_config) + asyncio.run(k8s_watch.load_config()) + assert calls == ["kubeconfig"] + + +class _FakeApiClient: + """Fakes kubernetes_asyncio.client.ApiClient as an async context manager.""" + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +class _FakeList: + def __init__(self, items, resource_version="1"): + self.items = items + self.metadata = types.SimpleNamespace(resource_version=resource_version) + + +class _FakeStream: + """Async context manager + async iterator over scripted watch events, + then raises to simulate a 410 Gone / connection drop.""" + + def __init__(self, events, error): + self._events = list(events) + self._error = error + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + def __aiter__(self): + return self + + async def __anext__(self): + if self._events: + return self._events.pop(0) + raise self._error + + +class _FakeWatch: + """Fakes kubernetes_asyncio.watch.Watch. + + Subclasses override `events`/`error` via `type(...)` per test; declared + as ClassVar so ruff doesn't flag them as a mutable-default footgun. + """ + + events: ClassVar[list] = [] + error: ClassVar[Exception] = RuntimeError("410 Gone") + + def stream(self, lister, resource_version=None, timeout_seconds=None): + return _FakeStream(self.__class__.events, self.__class__.error) + + +def _patch_k8s_client(monkeypatch, nodes=None, pods=None, deployments=None, events=None): + """Patch kubernetes_asyncio.client's Api classes used by _watch_loop.""" + from kubernetes_asyncio import client as kclient + + class _FakeCoreV1Api: + def __init__(self, api_client): + pass + + async def list_node(self): + return _FakeList(nodes or []) + + async def list_pod_for_all_namespaces(self): + return _FakeList(pods or []) + + async def list_event_for_all_namespaces(self): + return _FakeList(events or []) + + class _FakeAppsV1Api: + def __init__(self, api_client): + pass + + async def list_deployment_for_all_namespaces(self): + return _FakeList(deployments or []) + + monkeypatch.setattr(kclient, "ApiClient", _FakeApiClient) + monkeypatch.setattr(kclient, "CoreV1Api", _FakeCoreV1Api) + monkeypatch.setattr(kclient, "AppsV1Api", _FakeAppsV1Api) + + +@pytest.mark.parametrize( + "kind,seed_kw,initial_obj,event_obj", + [ + ("nodes", "nodes", _node_obj(), _node_obj(name="pi-2")), + ("pods", "pods", _pod_obj(), _pod_obj(name="p2")), + ("deployments", "deployments", _deployment_obj(), _deployment_obj(name="d2")), + ("events", "events", _event_obj(), _event_obj(uid="e2")), + ], +) +def test_watch_loop_seeds_state_then_reconnects_on_drop( + monkeypatch, kind, seed_kw, initial_obj, event_obj +): + """Exercises the initial list+seed, the live ADDED watch event, and the + reconnect/backoff path (simulated 410 Gone -> log -> sleep -> propagate).""" + from kubernetes_asyncio import watch as kwatch + + from app.collectors import k8s_watch + from app.state import ClusterState + + _patch_k8s_client(monkeypatch, **{seed_kw: [initial_obj]}) + + added = {"type": "ADDED", "object": event_obj} + fake_watch_cls = type( + "FakeWatch", + (_FakeWatch,), + {"events": [added], "error": RuntimeError("410 Gone")}, + ) + monkeypatch.setattr(kwatch, "Watch", fake_watch_cls) + + sleep_calls = [] + + async def fake_sleep(seconds): + sleep_calls.append(seconds) + raise asyncio.CancelledError() # stop the infinite loop deterministically + + monkeypatch.setattr(k8s_watch.asyncio, "sleep", fake_sleep) + + st = ClusterState() + with pytest.raises(asyncio.CancelledError): + asyncio.run(k8s_watch._watch_loop(st, kind)) + + collection = getattr(st, kind) + # initial list seeds 1 item, the watch stream ADDs a second (different key) + # before the simulated 410 Gone triggers the reconnect/backoff path + assert len(collection) == 2 + assert sleep_calls == [1] # backoff starts at 1s after the simulated 410 + + +def test_watch_loop_unknown_kind_raises_value_error_and_backs_off(monkeypatch): + """kind not in {nodes,pods,deployments,events} -> ValueError, caught by the + generic except, triggering the same backoff/reconnect path.""" + from app.collectors import k8s_watch + from app.state import ClusterState + + _patch_k8s_client(monkeypatch) + + sleep_calls = [] + + async def fake_sleep(seconds): + sleep_calls.append(seconds) + raise asyncio.CancelledError() + + monkeypatch.setattr(k8s_watch.asyncio, "sleep", fake_sleep) + + st = ClusterState() + with pytest.raises(asyncio.CancelledError): + asyncio.run(k8s_watch._watch_loop(st, "bogus-kind")) + assert sleep_calls == [1] + + +def test_watch_loop_propagates_cancellation_from_inside_try(monkeypatch): + """A CancelledError raised while listing/watching must be re-raised as-is + (not swallowed as a generic reconnect-worthy Exception).""" + from kubernetes_asyncio import client as kclient + + from app.collectors import k8s_watch + from app.state import ClusterState + + class _CancellingCoreV1Api: + def __init__(self, api_client): + pass + + async def list_node(self): + raise asyncio.CancelledError() + + monkeypatch.setattr(kclient, "ApiClient", _FakeApiClient) + monkeypatch.setattr(kclient, "CoreV1Api", _CancellingCoreV1Api) + monkeypatch.setattr(kclient, "AppsV1Api", lambda api_client: None) + + st = ClusterState() + with pytest.raises(asyncio.CancelledError): + asyncio.run(k8s_watch._watch_loop(st, "nodes")) + + +def test_watch_loop_backoff_grows_across_repeated_failures(monkeypatch): + """First reconnect sleeps 1s and doubles the backoff; second reconnect + must then sleep 2s -- covers the `backoff = min(backoff * 2, 30)` line. + + The initial list() must keep failing on every attempt here: a successful + list resets backoff to 1 (line 135), so growth is only observable when + every reconnect attempt fails before ever reaching that reset. + """ + from kubernetes_asyncio import client as kclient + + from app.collectors import k8s_watch + from app.state import ClusterState + + class _AlwaysFailingCoreV1Api: + def __init__(self, api_client): + pass + + async def list_node(self): + raise RuntimeError("apiserver unreachable") + + monkeypatch.setattr(kclient, "ApiClient", _FakeApiClient) + monkeypatch.setattr(kclient, "CoreV1Api", _AlwaysFailingCoreV1Api) + monkeypatch.setattr(kclient, "AppsV1Api", lambda api_client: None) + + sleep_calls = [] + + async def fake_sleep(seconds): + sleep_calls.append(seconds) + if len(sleep_calls) >= 2: + raise asyncio.CancelledError() # stop after observing backoff growth + + monkeypatch.setattr(k8s_watch.asyncio, "sleep", fake_sleep) + + st = ClusterState() + with pytest.raises(asyncio.CancelledError): + asyncio.run(k8s_watch._watch_loop(st, "nodes")) + assert sleep_calls == [1, 2] + + +def test_run_starts_all_four_watch_loops(monkeypatch): + from app.collectors import k8s_watch + from app.state import ClusterState + + calls = [] + + async def fake_load_config(): + calls.append("config") + + async def fake_watch_loop(state, kind): + calls.append(kind) + + monkeypatch.setattr(k8s_watch, "load_config", fake_load_config) + monkeypatch.setattr(k8s_watch, "_watch_loop", fake_watch_loop) + + st = ClusterState() + asyncio.run(k8s_watch.run(st)) + assert calls[0] == "config" + assert set(calls[1:]) == {"nodes", "pods", "deployments", "events"} + + +# ================================================================== +# app.collectors.metrics +# ================================================================== + + +def test_parse_cpu_micro_suffix(): + from app.collectors.metrics import parse_cpu + + assert parse_cpu("500u") == pytest.approx(0.0005) + + +class _FakeCustomObjectsApi: + def __init__(self, items_sequence, error=None): + self._items_sequence = items_sequence + self._error = error + self.calls = 0 + + async def list_cluster_custom_object(self, group, version, plural): + if self._error: + raise self._error + self.calls += 1 + return {"items": self._items_sequence} + + +def test_metrics_run_records_valid_and_skips_unparsable(monkeypatch): + """Happy path: one node with a known capacity, one node falling back to + defaults, and one malformed entry hitting the KeyError/ValueError branch.""" + from kubernetes_asyncio import client as kclient + + from app.collectors import metrics + from app.state import ClusterState + + st = ClusterState() + st.nodes["pi-1"] = {"cpu_capacity": "4", "mem_capacity": "8Gi"} + # pi-2 intentionally absent from st.nodes -> exercises the default fallback + + items = [ + {"metadata": {"name": "pi-1"}, "usage": {"cpu": "200m", "memory": "512Mi"}}, + {"metadata": {"name": "pi-2"}, "usage": {"cpu": "100m", "memory": "256Mi"}}, + {"metadata": {"name": "pi-3"}, "usage": {"cpu": "not-a-number", "memory": "256Mi"}}, + ] + fake_api = _FakeCustomObjectsApi(items) + monkeypatch.setattr(kclient, "ApiClient", _FakeApiClient) + monkeypatch.setattr(kclient, "CustomObjectsApi", lambda api_client: fake_api) + + sleep_calls = [] + + async def fake_sleep(seconds): + sleep_calls.append(seconds) + raise asyncio.CancelledError() + + monkeypatch.setattr(metrics.asyncio, "sleep", fake_sleep) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(metrics.run(st)) + + assert fake_api.calls == 1 + assert "pi-1" in st.node_metrics + assert st.node_metrics["pi-1"]["cpu_pct"] == pytest.approx(5.0) # 0.2 cores / 4 + assert "pi-2" in st.node_metrics # used default 4-core / 8Gi fallback + assert "pi-3" not in st.node_metrics # unparsable usage.cpu -> skipped + assert sleep_calls == [metrics.POLL_INTERVAL] + + +def test_metrics_run_retries_after_apiserver_error(monkeypatch): + """metrics-server unreachable -> outer except -> warn + 30s backoff.""" + from kubernetes_asyncio import client as kclient + + from app.collectors import metrics + from app.state import ClusterState + + st = ClusterState() + fake_api = _FakeCustomObjectsApi([], error=RuntimeError("connection refused")) + monkeypatch.setattr(kclient, "ApiClient", _FakeApiClient) + monkeypatch.setattr(kclient, "CustomObjectsApi", lambda api_client: fake_api) + + sleep_calls = [] + + async def fake_sleep(seconds): + sleep_calls.append(seconds) + raise asyncio.CancelledError() + + monkeypatch.setattr(metrics.asyncio, "sleep", fake_sleep) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(metrics.run(st)) + assert sleep_calls == [30] + + +# ================================================================== +# app.ws +# ================================================================== + + +def _fresh_ws_app(monkeypatch, password=None): + """Reload app.auth (to pick up the password) and app.ws (to pick up the + reloaded verify_token), then wire ws.router into a minimal FastAPI app + with a brand-new ClusterState so tests never see cross-test pollution.""" + if password is None: + monkeypatch.delenv("PIWATCH_PASSWORD", raising=False) + else: + monkeypatch.setenv("PIWATCH_PASSWORD", password) + import app.auth as auth_mod + import app.ws as ws_mod + + auth_mod = importlib.reload(auth_mod) + ws_mod = importlib.reload(ws_mod) + + from app.state import ClusterState + + fresh_state = ClusterState() + monkeypatch.setattr(ws_mod, "state", fresh_state) + + from fastapi import FastAPI + + app = FastAPI() + app.include_router(ws_mod.router) + return app, ws_mod, auth_mod, fresh_state + + +def test_ws_state_unauthorized_closes_with_4401(monkeypatch): + from fastapi.testclient import TestClient + from starlette.websockets import WebSocketDisconnect + + app, _ws_mod, _auth_mod, _fresh_state = _fresh_ws_app(monkeypatch, password="secret123") + client = TestClient(app) + with client.websocket_connect("/ws") as websocket: + with pytest.raises(WebSocketDisconnect) as excinfo: + websocket.receive_text() + assert excinfo.value.code == 4401 + + +def test_ws_state_full_snapshot_then_broadcast(monkeypatch): + from fastapi.testclient import TestClient + + app, _ws_mod, _auth_mod, fresh_state = _fresh_ws_app(monkeypatch, password=None) + fresh_state.upsert_node("seed", {"name": "seed", "ready": True}) + + client = TestClient(app) + with client.websocket_connect("/ws") as websocket: + first = websocket.receive_json() + assert first["type"] == "full_state" + assert "seed" in first["data"]["nodes"] + + # Publish a delta from the server's own event loop thread (via the + # test session's portal) -- state.subscribe()'s Queue is not + # cross-thread safe, so mutations must run on the portal's loop. + websocket.portal.call( + fresh_state.upsert_node, "pi-1", {"name": "pi-1", "ready": True} + ) + second = websocket.receive_json() + assert second["type"] == "node" + assert second["data"]["name"] == "pi-1" + + assert len(fresh_state._subscribers) == 0 # unsubscribed in the finally block + + +def test_ws_state_heartbeat_ping_on_idle(monkeypatch): + from fastapi.testclient import TestClient + + app, ws_mod, _auth_mod, _fresh_state = _fresh_ws_app(monkeypatch, password=None) + monkeypatch.setattr(ws_mod, "HEARTBEAT_S", 0.05) + + client = TestClient(app) + with client.websocket_connect("/ws") as websocket: + first = websocket.receive_json() + assert first["type"] == "full_state" + second = websocket.receive_json() + assert second["type"] == "ping" + + +def test_ws_logs_unauthorized_closes_with_4401(monkeypatch): + from fastapi.testclient import TestClient + from starlette.websockets import WebSocketDisconnect + + app, _ws_mod, _auth_mod, _fresh_state = _fresh_ws_app(monkeypatch, password="secret123") + client = TestClient(app) + with client.websocket_connect("/ws/logs/default/mypod") as websocket: + with pytest.raises(WebSocketDisconnect) as excinfo: + websocket.receive_text() + assert excinfo.value.code == 4401 + + +def test_ws_logs_demo_mode_streams_fake_lines(monkeypatch): + from fastapi.testclient import TestClient + + app, _ws_mod, _auth_mod, fresh_state = _fresh_ws_app(monkeypatch, password=None) + fresh_state.demo_mode = True + + client = TestClient(app) + with client.websocket_connect("/ws/logs/monitoring/piwatch-7c9d4-a") as websocket: + msg = websocket.receive_json() + assert msg["type"] == "log" + assert "line" in msg + + +def test_ws_logs_non_demo_source_error_becomes_log_error(monkeypatch): + """Real-cluster log source failures must surface as a 'log_error' message + (covers the except branch in ws_logs' sender()).""" + from fastapi.testclient import TestClient + + app, ws_mod, _auth_mod, fresh_state = _fresh_ws_app(monkeypatch, password=None) + fresh_state.demo_mode = False + + async def broken_source(namespace, pod, container): + raise RuntimeError("pod not found") + yield "unreachable" # pragma: no cover -- makes this an async generator + + monkeypatch.setattr(ws_mod, "_k8s_log_lines", broken_source) + + client = TestClient(app) + with client.websocket_connect("/ws/logs/default/mypod") as websocket: + msg = websocket.receive_json() + assert msg["type"] == "log_error" + assert "pod not found" in msg["error"] + + +class _FailingWebSocket: + """Minimal fake satisfying the subset of the Starlette WebSocket API that + ws_state() touches -- lets us drive the disconnect-during-send except + branch directly and deterministically, without racing TestClient's + threaded portal teardown.""" + + def __init__(self, fail_exc): + self.sent = [] + self.closed = None + self._fail_exc = fail_exc + + async def accept(self): + pass + + async def close(self, code=1000, reason=""): + self.closed = (code, reason) + + async def send_text(self, text): + self.sent.append(text) + raise self._fail_exc + + +def test_ws_state_send_disconnect_is_swallowed(monkeypatch): + """A send() failure (peer went away) must be caught by the + `except (WebSocketDisconnect, RuntimeError): pass` and still unsubscribe.""" + from starlette.websockets import WebSocketDisconnect + + import app.ws as ws_mod + from app.state import ClusterState + + fresh_state = ClusterState() + monkeypatch.setattr(ws_mod, "state", fresh_state) + monkeypatch.delenv("PIWATCH_PASSWORD", raising=False) + import app.auth as auth_mod + + importlib.reload(auth_mod) + ws_mod = importlib.reload(ws_mod) + monkeypatch.setattr(ws_mod, "state", fresh_state) + + fake_ws = _FailingWebSocket(WebSocketDisconnect(code=1001)) + asyncio.run(ws_mod.ws_state(fake_ws, token=None)) # must not raise + assert len(fresh_state._subscribers) == 0 # finally still unsubscribed + + +def test_ws_state_send_runtime_error_is_swallowed(monkeypatch): + import app.ws as ws_mod + from app.state import ClusterState + + fresh_state = ClusterState() + monkeypatch.setattr(ws_mod, "state", fresh_state) + monkeypatch.delenv("PIWATCH_PASSWORD", raising=False) + import app.auth as auth_mod + + importlib.reload(auth_mod) + ws_mod = importlib.reload(ws_mod) + monkeypatch.setattr(ws_mod, "state", fresh_state) + + fake_ws = _FailingWebSocket(RuntimeError("connection already closed")) + asyncio.run(ws_mod.ws_state(fake_ws, token=None)) # must not raise + assert len(fresh_state._subscribers) == 0 + + +class _FakeLogContent: + """Fakes the httpx-style streaming body (`resp.content`) that + _k8s_log_lines iterates over.""" + + def __init__(self, lines: list[str]): + self._chunks = [line.encode() for line in lines] + + def __aiter__(self): + return self + + async def __anext__(self): + if self._chunks: + return self._chunks.pop(0) + raise StopAsyncIteration + + +class _FakeLogResponse: + def __init__(self, lines: list[str]): + self.content = _FakeLogContent(lines) + self.closed = False + + def close(self): + self.closed = True + + +def test_k8s_log_lines_streams_and_closes_response(monkeypatch): + """Direct test of the real (non-demo) log source generator: decodes raw + chunks from the k8s API response and always closes it afterwards.""" + from kubernetes_asyncio import client as kclient + + import app.ws as ws_mod + + fake_resp = _FakeLogResponse(["hello world\n", "second line\n"]) + + class _FakeCoreV1Api: + def __init__(self, api_client): + pass + + async def read_namespaced_pod_log(self, **kwargs): + assert kwargs["name"] == "mypod" + assert kwargs["namespace"] == "default" + return fake_resp + + monkeypatch.setattr(kclient, "ApiClient", _FakeApiClient) + monkeypatch.setattr(kclient, "CoreV1Api", _FakeCoreV1Api) + + async def scenario(): + return [ + line + async for line in ws_mod._k8s_log_lines("default", "mypod", None) + ] + + lines = asyncio.run(scenario()) + assert lines == ["hello world", "second line"] + assert fake_resp.closed is True + + +# ================================================================== +# app.main -- startup-mode branches +# ================================================================== + + +def _reload_main(monkeypatch, **env): + """Set/clear env vars then reload app.main so its module-level reads + (STATIC_DIR, etc.) pick up the new values. Also swaps in a brand-new + ClusterState: app.state.state is a process-wide singleton, and earlier + tests (in this file or test_backend.py) may have already flipped + demo_mode / seeded nodes on it, which would silently poison readyz/ + get_state assertions here.""" + for key, value in env.items(): + if value is None: + monkeypatch.delenv(key, raising=False) + else: + monkeypatch.setenv(key, value) + import app.main as main_mod + + main_mod = importlib.reload(main_mod) + + from app.state import ClusterState + + monkeypatch.setattr(main_mod, "state", ClusterState()) + return main_mod + + +def test_cluster_reachable_true_when_in_cluster_env(monkeypatch): + main_mod = _reload_main(monkeypatch, KUBERNETES_SERVICE_HOST="10.0.0.1", PIWATCH_DEMO=None) + assert asyncio.run(main_mod._cluster_reachable()) is True + + +def test_cluster_reachable_true_when_kubeconfig_loads(monkeypatch): + main_mod = _reload_main(monkeypatch, KUBERNETES_SERVICE_HOST=None, PIWATCH_DEMO=None) + + async def ok_load_kube_config(): + return None + + from kubernetes_asyncio import config as kconfig + + monkeypatch.setattr(kconfig, "load_kube_config", ok_load_kube_config) + assert asyncio.run(main_mod._cluster_reachable()) is True + + +def test_cluster_reachable_false_when_no_kubeconfig(monkeypatch): + main_mod = _reload_main(monkeypatch, KUBERNETES_SERVICE_HOST=None, PIWATCH_DEMO=None) + + async def failing_load_kube_config(): + raise FileNotFoundError("no kubeconfig") + + from kubernetes_asyncio import config as kconfig + + monkeypatch.setattr(kconfig, "load_kube_config", failing_load_kube_config) + assert asyncio.run(main_mod._cluster_reachable()) is False + + +def test_lifespan_real_cluster_mode_starts_k8s_collectors(monkeypatch, tmp_path): + """PIWATCH_DEMO unset + reachable cluster -> lifespan starts the + k8s_watch/metrics/hardware collectors (not demo.run).""" + nostatic = tmp_path / "nostatic" # deliberately not created -> else branch + main_mod = _reload_main( + monkeypatch, + PIWATCH_DEMO=None, + KUBERNETES_SERVICE_HOST="10.0.0.1", + PIWATCH_STATIC_DIR=str(nostatic), + ) + + started = [] + + async def fake_collector(state): + started.append(state) + await asyncio.sleep(3600) # stays alive until lifespan cancels it + + monkeypatch.setattr(main_mod.k8s_watch, "run", fake_collector) + monkeypatch.setattr(main_mod.metrics, "run", fake_collector) + monkeypatch.setattr(main_mod.hardware, "run", fake_collector) + + from fastapi.testclient import TestClient + + with TestClient(main_mod.app) as client: + assert client.get("/healthz").json() == {"ok": True} + # not demo mode and no nodes seeded yet -> not ready + r = client.get("/readyz") + assert r.status_code == 503 + assert r.json() == {"ready": False} + assert len(started) == 3 # k8s_watch, metrics, hardware all launched + + +def test_get_state_endpoint_returns_snapshot(monkeypatch, tmp_path): + nostatic = tmp_path / "nostatic" + main_mod = _reload_main( + monkeypatch, PIWATCH_DEMO="1", PIWATCH_STATIC_DIR=str(nostatic), PIWATCH_PASSWORD=None + ) + from fastapi.testclient import TestClient + + with TestClient(main_mod.app) as client: + r = client.get("/api/state") + assert r.status_code == 200 + body = r.json() + assert "nodes" in body and "demo_mode" in body + + +def test_readyz_true_in_demo_mode(monkeypatch, tmp_path): + nostatic = tmp_path / "nostatic" + main_mod = _reload_main(monkeypatch, PIWATCH_DEMO="1", PIWATCH_STATIC_DIR=str(nostatic)) + from fastapi.testclient import TestClient + + with TestClient(main_mod.app) as client: + r = client.get("/readyz") + assert r.status_code == 200 + assert r.json() == {"ready": True} + + +def test_spa_serves_existing_file_directly(monkeypatch, tmp_path): + """A real file at the STATIC_DIR root (outside /assets) is served as-is + via FileResponse(candidate), not the index.html fallback.""" + static = tmp_path / "static" + (static / "assets").mkdir(parents=True) + (static / "index.html").write_text("index") + (static / "favicon.ico").write_text("ICO-BYTES") + + main_mod = _reload_main( + monkeypatch, PIWATCH_DEMO="1", PIWATCH_STATIC_DIR=str(static) + ) + from fastapi.testclient import TestClient + + with TestClient(main_mod.app) as client: + r = client.get("/favicon.ico") + assert r.text == "ICO-BYTES" + + +def test_static_dir_missing_logs_warning_and_skips_mount(monkeypatch, tmp_path, caplog): + nostatic = tmp_path / "does-not-exist" + with caplog.at_level("WARNING", logger="piwatch"): + main_mod = _reload_main( + monkeypatch, PIWATCH_DEMO="1", PIWATCH_STATIC_DIR=str(nostatic) + ) + assert any("missing" in rec.message for rec in caplog.records) + # no catch-all route registered -> unknown paths 404 instead of falling + # back to a (non-existent) index.html + from fastapi.testclient import TestClient + + with TestClient(main_mod.app) as client: + r = client.get("/some/random/path") + assert r.status_code == 404 diff --git a/backend/tests/test_units.py b/backend/tests/test_units.py new file mode 100644 index 0000000..0b4522c --- /dev/null +++ b/backend/tests/test_units.py @@ -0,0 +1,768 @@ +"""Unit tests that push coverage on the modules test_backend.py barely +touches: the Pi node-agent, the hardware/healthcheck collectors, remaining +auth edge branches, state.py's pub/sub error handling, and demo.py's +fake_logs() generator. + +Style matches test_backend.py: plain functions, `asyncio.run(scenario())` +for async code (no pytest-asyncio dependency), monkeypatch for env/attrs, +direct calls into module internals rather than spinning up full apps where +that is enough to exercise the target lines. +""" +from __future__ import annotations + +import asyncio +import base64 +import importlib +import os +import socket +import sys +import types + +import httpx +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +# -------------------------------------------------------------------------- +# app.node_agent imports `os.uname().nodename` as the *default argument* to +# os.environ.get(), so it is evaluated unconditionally at import time -- +# even when NODE_NAME is set. os.uname() does not exist on Windows, so a +# bare `import app.node_agent` would crash dev machines while working fine +# on Linux CI. Stub it explicitly, once, before the only import of that +# module, so behaviour is identical (and deterministic) on both platforms. +# -------------------------------------------------------------------------- +if not hasattr(os, "uname"): + _stub_uname = types.SimpleNamespace(nodename="stub-host") + os.uname = lambda: _stub_uname # type: ignore[attr-defined] + +from app import node_agent +from app.collectors import demo, hardware, healthcheck +from app.state import ClusterState + + +class _StopLoop(Exception): + """Sentinel used to break out of collectors' `while True` loops.""" + + +async def _raise_stop(*_a, **_kw): + raise _StopLoop() + + +async def _instant_sleep(*_a, **_kw): + """Fast stand-in for asyncio.sleep() that does not actually wait.""" + return + + +# =========================== app.node_agent =============================== + +def test_read_temp_c_returns_millidegrees_as_celsius(tmp_path, monkeypatch): + zone = tmp_path / "sys" / "class" / "thermal" / "thermal_zone0" + zone.mkdir(parents=True) + (zone / "temp").write_text("48250") + monkeypatch.setattr(node_agent, "SYS", str(tmp_path / "sys")) + assert node_agent.read_temp_c() == 48.2 + + +def test_read_temp_c_skips_unreadable_zone_and_uses_next(tmp_path, monkeypatch): + thermal = tmp_path / "sys" / "class" / "thermal" + zone0 = thermal / "thermal_zone0" + zone1 = thermal / "thermal_zone1" + zone0.mkdir(parents=True) + zone1.mkdir(parents=True) + (zone0 / "temp").write_text("not-a-number") # triggers ValueError, skip + (zone1 / "temp").write_text("52340") + monkeypatch.setattr(node_agent, "SYS", str(tmp_path / "sys")) + assert node_agent.read_temp_c() == 52.3 + + +def test_read_temp_c_returns_none_when_no_thermal_zones(tmp_path, monkeypatch): + monkeypatch.setattr(node_agent, "SYS", str(tmp_path / "no-such-sys")) + assert node_agent.read_temp_c() is None + + +def test_read_load_parses_loadavg(tmp_path, monkeypatch): + proc = tmp_path / "proc" + proc.mkdir() + (proc / "loadavg").write_text("0.11 0.22 0.33 1/222 3456\n") + monkeypatch.setattr(node_agent, "PROC", str(proc)) + assert node_agent.read_load() == (0.11, 0.22, 0.33) + + +def test_read_load_returns_none_when_file_missing(tmp_path, monkeypatch): + monkeypatch.setattr(node_agent, "PROC", str(tmp_path / "no-proc")) + assert node_agent.read_load() is None + + +def test_read_load_returns_none_on_malformed_content(tmp_path, monkeypatch): + proc = tmp_path / "proc" + proc.mkdir() + (proc / "loadavg").write_text("a b c\n") + monkeypatch.setattr(node_agent, "PROC", str(proc)) + assert node_agent.read_load() is None + + +def test_read_meminfo_extracts_total_and_available(tmp_path, monkeypatch): + proc = tmp_path / "proc" + proc.mkdir() + (proc / "meminfo").write_text( + "MemTotal: 16384000 kB\n" + "MemFree: 1000000 kB\n" + "MemAvailable: 8000000 kB\n" + ) + monkeypatch.setattr(node_agent, "PROC", str(proc)) + assert node_agent.read_meminfo() == {"MemTotal": 16384000, "MemAvailable": 8000000} + + +def test_read_meminfo_returns_empty_dict_when_file_missing(tmp_path, monkeypatch): + monkeypatch.setattr(node_agent, "PROC", str(tmp_path / "no-proc")) + assert node_agent.read_meminfo() == {} + + +def test_read_meminfo_malformed_line_degrades_to_empty_dict(tmp_path, monkeypatch): + """A MemTotal/MemAvailable line with nothing after the colon makes + `rest.strip().split()[0]` raise IndexError - which the except clause + must swallow (like read_load() does for the same parsing pattern), so + malformed host /proc/meminfo content degrades the /metrics payload + instead of 500ing it. Regression test for a bug found by this suite. + """ + proc = tmp_path / "proc" + proc.mkdir() + (proc / "meminfo").write_text("MemTotal:\n") + monkeypatch.setattr(node_agent, "PROC", str(proc)) + assert node_agent.read_meminfo() == {} + + +def test_read_uptime_s_parses_seconds(tmp_path, monkeypatch): + proc = tmp_path / "proc" + proc.mkdir() + (proc / "uptime").write_text("54321.9 12345.6\n") + monkeypatch.setattr(node_agent, "PROC", str(proc)) + assert node_agent.read_uptime_s() == 54321 + + +def test_read_uptime_s_returns_none_when_file_missing(tmp_path, monkeypatch): + monkeypatch.setattr(node_agent, "PROC", str(tmp_path / "no-proc")) + assert node_agent.read_uptime_s() is None + + +def test_read_uptime_s_empty_file_degrades_to_none(tmp_path, monkeypatch): + """An empty /proc/uptime makes `f.read().split()[0]` raise IndexError - + which must be swallowed to None (like read_load() does for the same + parsing pattern) instead of 500ing the /metrics endpoint. Regression + test for a bug found by this suite. + """ + proc = tmp_path / "proc" + proc.mkdir() + (proc / "uptime").write_text("") + monkeypatch.setattr(node_agent, "PROC", str(proc)) + assert node_agent.read_uptime_s() is None + + +def _seed_full_node_agent_fs(tmp_path, monkeypatch): + sys_dir = tmp_path / "sys" + proc_dir = tmp_path / "proc" + zone = sys_dir / "class" / "thermal" / "thermal_zone0" + zone.mkdir(parents=True) + (zone / "temp").write_text("48250") + proc_dir.mkdir() + (proc_dir / "loadavg").write_text("0.11 0.22 0.33 1/222 3456\n") + (proc_dir / "meminfo").write_text( + "MemTotal: 1000000 kB\nMemAvailable: 500000 kB\nOther: 1 kB\n" + ) + (proc_dir / "uptime").write_text("54321.9 12345.6\n") + monkeypatch.setattr(node_agent, "SYS", str(sys_dir)) + monkeypatch.setattr(node_agent, "PROC", str(proc_dir)) + monkeypatch.setattr(node_agent, "DISK_PATH", str(tmp_path)) + monkeypatch.setattr(node_agent, "NODE_NAME", "pi-test") + + +def test_metrics_endpoint_full_data(tmp_path, monkeypatch): + from fastapi.testclient import TestClient + + _seed_full_node_agent_fs(tmp_path, monkeypatch) + with TestClient(node_agent.app) as client: + body = client.get("/metrics").json() + + assert body["node"] == "pi-test" + assert body["temp_c"] == 48.2 + assert body["load1"] == 0.11 + assert body["load5"] == 0.22 + assert body["mem_total_kb"] == 1000000 + assert body["mem_available_kb"] == 500000 + assert body["uptime_s"] == 54321 + assert 0.0 <= body["disk_used_pct"] <= 100.0 + + +def test_metrics_endpoint_degrades_gracefully_when_everything_missing( + tmp_path, monkeypatch +): + from fastapi.testclient import TestClient + + monkeypatch.setattr(node_agent, "SYS", str(tmp_path / "no-sys")) + monkeypatch.setattr(node_agent, "PROC", str(tmp_path / "no-proc")) + monkeypatch.setattr(node_agent, "DISK_PATH", str(tmp_path / "no-disk")) + monkeypatch.setattr(node_agent, "NODE_NAME", "pi-empty") + + with TestClient(node_agent.app) as client: + body = client.get("/metrics").json() + + assert body == { + "node": "pi-empty", + "temp_c": None, + "load1": None, + "load5": None, + "mem_total_kb": None, + "mem_available_kb": None, + "disk_used_pct": None, + "uptime_s": None, + } + + +def test_node_agent_healthz(): + from fastapi.testclient import TestClient + + with TestClient(node_agent.app) as client: + assert client.get("/healthz").json() == {"ok": True} + + +# ========================= collectors.hardware ============================= + +def test_resolve_agents_returns_sorted_unique_ips(monkeypatch): + async def scenario(): + loop = asyncio.get_running_loop() + + async def fake_getaddrinfo(host, port, type=None): + assert host == hardware.AGENT_SERVICE + assert port == hardware.AGENT_PORT + return [ + (2, 1, 6, "", ("10.0.0.6", port)), + (2, 1, 6, "", ("10.0.0.5", port)), + (2, 1, 6, "", ("10.0.0.5", port)), # duplicate must be deduped + ] + + monkeypatch.setattr(loop, "getaddrinfo", fake_getaddrinfo) + assert await hardware._resolve_agents() == ["10.0.0.5", "10.0.0.6"] + + asyncio.run(scenario()) + + +def test_resolve_agents_returns_empty_list_on_dns_failure(monkeypatch): + async def scenario(): + loop = asyncio.get_running_loop() + + async def fake_getaddrinfo(*_a, **_kw): + raise socket.gaierror("name or service not known") + + monkeypatch.setattr(loop, "getaddrinfo", fake_getaddrinfo) + assert await hardware._resolve_agents() == [] + + asyncio.run(scenario()) + + +class _FakeAgentResponse: + def __init__(self, data): + self._data = data + + def json(self): + return self._data + + +def test_hardware_run_records_a_sample_and_computes_mem_pct(monkeypatch): + async def scenario(): + st = ClusterState() + + async def fake_resolve(): + return ["10.0.0.5"] + + async def fake_get(_self, _url, *_a, **_kw): + return _FakeAgentResponse( + { + "node": "pi-1", + "temp_c": 55.5, + "mem_total_kb": 8_000_000, + "mem_available_kb": 2_000_000, + } + ) + + monkeypatch.setattr(hardware, "_resolve_agents", fake_resolve) + monkeypatch.setattr(httpx.AsyncClient, "get", fake_get) + monkeypatch.setattr(asyncio, "sleep", _raise_stop) + + with pytest.raises(_StopLoop): + await hardware.run(st) + + recorded = st.hardware["pi-1"] + assert recorded["temp_c"] == 55.5 + assert recorded["mem_pct"] == pytest.approx(75.0) + + asyncio.run(scenario()) + + +def test_hardware_run_swallows_unreachable_agent_errors(monkeypatch): + async def scenario(): + st = ClusterState() + + async def fake_resolve(): + return ["10.0.0.9"] + + async def fake_get(_self, _url, *_a, **_kw): + raise httpx.ConnectError("connection refused") + + monkeypatch.setattr(hardware, "_resolve_agents", fake_resolve) + monkeypatch.setattr(httpx.AsyncClient, "get", fake_get) + monkeypatch.setattr(asyncio, "sleep", _raise_stop) + + with pytest.raises(_StopLoop): + await hardware.run(st) + + assert st.hardware == {} + + asyncio.run(scenario()) + + +def test_hardware_run_handles_no_agents_resolved(monkeypatch): + async def scenario(): + st = ClusterState() + + async def fake_resolve(): + return [] + + monkeypatch.setattr(hardware, "_resolve_agents", fake_resolve) + monkeypatch.setattr(asyncio, "sleep", _raise_stop) + + with pytest.raises(_StopLoop): + await hardware.run(st) + + assert st.hardware == {} + + asyncio.run(scenario()) + + +# ======================== collectors.healthcheck ============================ +# test_backend.py's SPA test indirectly runs healthcheck.run() in DEMO mode +# (via app.main's lifespan), which already covers load_checks(demo=True), +# _run_demo() and the demo branch of _check_loop(). What's still untested is +# the *real* (non-demo) path: reading the YAML config file, _run_http(), +# _run_tcp(), the tcp/http dispatch in _check_loop(), and run()'s early +# return when there is nothing to check. + +def test_load_checks_demo_mode_returns_demo_checks(): + assert healthcheck.load_checks(True) == healthcheck.DEMO_CHECKS + + +def test_load_checks_reads_and_parses_yaml_file(tmp_path, monkeypatch): + cfg = tmp_path / "healthchecks.yaml" + cfg.write_text( + "checks:\n" + " - name: svc\n" + " type: http\n" + " url: http://x\n" + ) + monkeypatch.setattr(healthcheck, "CHECKS_FILE", str(cfg)) + assert healthcheck.load_checks(False) == [ + {"name": "svc", "type": "http", "url": "http://x"} + ] + + +def test_load_checks_missing_file_returns_empty_list(tmp_path, monkeypatch): + monkeypatch.setattr(healthcheck, "CHECKS_FILE", str(tmp_path / "nope.yaml")) + assert healthcheck.load_checks(False) == [] + + +def test_load_checks_invalid_yaml_returns_empty_list(tmp_path, monkeypatch): + cfg = tmp_path / "bad.yaml" + cfg.write_text("checks: [unterminated") # triggers a yaml.YAMLError + monkeypatch.setattr(healthcheck, "CHECKS_FILE", str(cfg)) + assert healthcheck.load_checks(False) == [] + + +async def _start_raw_http_server(status: int = 200): + """Minimal loopback HTTP/1.1 server -- no real network access needed.""" + + async def handler(reader, writer): + try: + await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), timeout=2) + except Exception: + pass + writer.write( + f"HTTP/1.1 {status} X\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".encode() + ) + await writer.drain() + writer.close() + await writer.wait_closed() + + server = await asyncio.start_server(handler, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + return server, port + + +def test_run_http_success_without_expected_status(monkeypatch): + async def scenario(): + server, port = await _start_raw_http_server(200) + try: + async with httpx.AsyncClient() as client: + ok, ms, detail = await healthcheck._run_http( + {"url": f"http://127.0.0.1:{port}/"}, client + ) + assert ok is True + assert detail == "HTTP 200" + assert ms is not None + finally: + server.close() + await server.wait_closed() + + asyncio.run(scenario()) + + +def test_run_http_status_mismatch_is_not_ok(monkeypatch): + async def scenario(): + server, port = await _start_raw_http_server(500) + try: + async with httpx.AsyncClient() as client: + ok, ms, detail = await healthcheck._run_http( + {"url": f"http://127.0.0.1:{port}/", "expected_status": 200}, client + ) + assert ok is False + assert detail == "HTTP 500" + assert ms is not None # request succeeded, just the wrong status + finally: + server.close() + await server.wait_closed() + + asyncio.run(scenario()) + + +def test_run_http_connection_refused(monkeypatch): + async def scenario(): + # Bind then immediately close, so the port is guaranteed to refuse. + server, port = await _start_raw_http_server(200) + server.close() + await server.wait_closed() + + async with httpx.AsyncClient() as client: + ok, ms, detail = await healthcheck._run_http( + {"url": f"http://127.0.0.1:{port}/"}, client + ) + assert ok is False + assert ms is None + assert detail # exact exception class name varies by platform + + asyncio.run(scenario()) + + +def test_run_http_timeout(monkeypatch): + # A genuine network timeout needs an unroutable target, which is flaky + # in CI sandboxes. Simulate the same code path deterministically instead. + async def scenario(): + async def fake_get(_self, _url, *_a, **_kw): + raise httpx.ConnectTimeout("timed out") + + monkeypatch.setattr(httpx.AsyncClient, "get", fake_get) + async with httpx.AsyncClient() as client: + ok, ms, detail = await healthcheck._run_http({"url": "http://x"}, client) + assert ok is False + assert ms is None + assert detail == "ConnectTimeout" + + asyncio.run(scenario()) + + +async def _start_raw_tcp_server(): + def handler(_reader, _writer): + return None + + server = await asyncio.start_server(handler, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + return server, port + + +def test_run_tcp_success(): + async def scenario(): + server, port = await _start_raw_tcp_server() + try: + ok, ms, detail = await healthcheck._run_tcp( + {"host": "127.0.0.1", "port": port} + ) + assert ok is True + assert detail == "TCP open" + assert ms is not None + finally: + server.close() + await server.wait_closed() + + asyncio.run(scenario()) + + +def test_run_tcp_connection_refused(): + async def scenario(): + server, port = await _start_raw_tcp_server() + server.close() + await server.wait_closed() + + ok, ms, detail = await healthcheck._run_tcp({"host": "127.0.0.1", "port": port}) + assert ok is False + assert ms is None + assert detail + + asyncio.run(scenario()) + + +def test_run_tcp_timeout(monkeypatch): + async def scenario(): + async def hanging_open_connection(*_a, **_kw): + await asyncio.sleep(10) + + monkeypatch.setattr(asyncio, "open_connection", hanging_open_connection) + ok, ms, detail = await healthcheck._run_tcp( + {"host": "127.0.0.1", "port": 1, "timeout": 0.05} + ) + assert ok is False + assert ms is None + assert detail == "TimeoutError" + + asyncio.run(scenario()) + + +def test_check_loop_dispatches_tcp_checks(monkeypatch): + async def scenario(): + server, port = await _start_raw_tcp_server() + try: + st = ClusterState() + check = { + "name": "mqtt", + "type": "tcp", + "host": "127.0.0.1", + "port": port, + "interval": 0, + } + monkeypatch.setattr(asyncio, "sleep", _raise_stop) + with pytest.raises(_StopLoop): + await healthcheck._check_loop(st, check, demo=False) + assert st.healthchecks["mqtt"]["last"]["ok"] is True + finally: + server.close() + await server.wait_closed() + + asyncio.run(scenario()) + + +def test_check_loop_dispatches_http_checks(monkeypatch): + async def scenario(): + server, port = await _start_raw_http_server(200) + try: + st = ClusterState() + check = { + "name": "web", + "type": "http", + "url": f"http://127.0.0.1:{port}/", + "interval": 0, + } + monkeypatch.setattr(asyncio, "sleep", _raise_stop) + with pytest.raises(_StopLoop): + await healthcheck._check_loop(st, check, demo=False) + assert st.healthchecks["web"]["last"]["ok"] is True + assert st.healthchecks["web"]["last"]["detail"] == "HTTP 200" + finally: + server.close() + await server.wait_closed() + + asyncio.run(scenario()) + + +def test_run_returns_immediately_when_no_checks_configured(monkeypatch): + async def scenario(): + st = ClusterState() + st.demo_mode = False + monkeypatch.setattr(healthcheck, "load_checks", lambda demo: []) + # Safety net: if run() ever stopped early-returning, this would hang. + await asyncio.wait_for(healthcheck.run(st), timeout=2) + + asyncio.run(scenario()) + + +# =============================== app.auth =================================== + +def _reload_auth(monkeypatch, password: str | None): + if password is None: + monkeypatch.delenv("PIWATCH_PASSWORD", raising=False) + else: + monkeypatch.setenv("PIWATCH_PASSWORD", password) + monkeypatch.delenv("PIWATCH_SECRET", raising=False) + import app.auth as auth_mod + + return importlib.reload(auth_mod) + + +def test_verify_token_rejects_payload_that_is_not_a_valid_expiry(monkeypatch): + """Valid signature, but the signed payload does not decode to an int + expiry -- exercises verify_token()'s `except Exception: return False`. + """ + auth = _reload_auth(monkeypatch, "pw123") + bad_payload = base64.urlsafe_b64encode(b"not-a-number").decode().rstrip("=") + token = f"{bad_payload}.{auth._sign(bad_payload)}" + assert not auth.verify_token(token) + + +def test_login_endpoint_returns_no_token_when_auth_disabled(monkeypatch): + auth = _reload_auth(monkeypatch, None) + result = auth.login(auth.LoginRequest(password="whatever")) + assert result == {"token": "", "auth": False} + + +def test_login_endpoint_rejects_wrong_password(monkeypatch): + auth = _reload_auth(monkeypatch, "correct-horse") + with pytest.raises(auth.HTTPException) as exc_info: + auth.login(auth.LoginRequest(password="wrong")) + assert exc_info.value.status_code == 401 + + +def test_login_endpoint_issues_token_for_correct_password(monkeypatch): + auth = _reload_auth(monkeypatch, "correct-horse") + result = auth.login(auth.LoginRequest(password="correct-horse")) + assert result["auth"] is True + assert result["ttl"] == auth.TOKEN_TTL + assert auth.verify_token(result["token"]) + + +def test_auth_info_reflects_whether_a_password_is_configured(monkeypatch): + auth_off = _reload_auth(monkeypatch, None) + assert auth_off.auth_info() == {"auth": False} + + auth_on = _reload_auth(monkeypatch, "pw") + assert auth_on.auth_info() == {"auth": True} + + +def test_require_auth_passes_silently_when_auth_disabled(monkeypatch): + auth = _reload_auth(monkeypatch, None) + assert auth.require_auth(authorization=None) is None + + +def test_require_auth_rejects_missing_header_when_enabled(monkeypatch): + auth = _reload_auth(monkeypatch, "pw123") + with pytest.raises(auth.HTTPException) as exc_info: + auth.require_auth(authorization=None) + assert exc_info.value.status_code == 401 + + +def test_require_auth_rejects_non_bearer_scheme(monkeypatch): + auth = _reload_auth(monkeypatch, "pw123") + with pytest.raises(auth.HTTPException): + auth.require_auth(authorization="Basic dXNlcjpwYXNz") + + +def test_require_auth_accepts_valid_bearer_token(monkeypatch): + auth = _reload_auth(monkeypatch, "pw123") + token = auth.create_token() + assert auth.require_auth(authorization=f"Bearer {token}") is None + + +def test_ws_token_ok(monkeypatch): + auth = _reload_auth(monkeypatch, "pw123") + token = auth.create_token() + assert auth.ws_token_ok(token) is True + assert auth.ws_token_ok("garbage") is False + + +# =============================== app.state =================================== + +def test_publish_drops_oldest_message_when_subscriber_queue_is_full(): + async def scenario(): + st = ClusterState() + q = asyncio.Queue(maxsize=1) + q.put_nowait({"stale": True}) + st._subscribers.add(q) + + st.publish("node", {"name": "pi-1"}) + + assert q.qsize() == 1 + msg = q.get_nowait() + assert msg["type"] == "node" + assert msg["data"] == {"name": "pi-1"} + + asyncio.run(scenario()) + + +def test_publish_swallows_errors_from_a_misbehaving_subscriber(): + """A subscriber whose queue can neither accept the new message nor be + drained must not stop other subscribers from getting theirs. + """ + + class BrokenQueue: + def put_nowait(self, _item): + raise asyncio.QueueFull() + + def get_nowait(self): + raise RuntimeError("nothing to drop") + + async def scenario(): + st = ClusterState() + st._subscribers.add(BrokenQueue()) + good_q = st.subscribe() + + st.publish("node", {"name": "pi-2"}) # must not raise + + msg = good_q.get_nowait() + assert msg["data"]["name"] == "pi-2" + + asyncio.run(scenario()) + + +def test_remove_node_pod_deployment_update_state_and_publish(): + async def scenario(): + st = ClusterState() + st.upsert_node("pi-1", {"name": "pi-1"}) + st.upsert_pod("ns/pod-1", {"name": "pod-1"}) + st.upsert_deployment("ns/dep-1", {"name": "dep-1"}) + q = st.subscribe() # subscribed after the upserts above + + st.remove_node("pi-1") + st.remove_pod("ns/pod-1") + st.remove_deployment("ns/dep-1") + st.remove_node("does-not-exist") # pop(..., None): must not raise + + assert "pi-1" not in st.nodes + assert "ns/pod-1" not in st.pods + assert "ns/dep-1" not in st.deployments + + msgs = [q.get_nowait() for _ in range(4)] + assert [m["type"] for m in msgs] == [ + "node_deleted", + "pod_deleted", + "deployment_deleted", + "node_deleted", + ] + assert msgs[0]["data"] == {"name": "pi-1"} + assert msgs[1]["data"] == {"key": "ns/pod-1"} + assert msgs[2]["data"] == {"key": "ns/dep-1"} + + asyncio.run(scenario()) + + +# ============================ collectors.demo ================================ + +def test_fake_logs_yields_formatted_lines_and_handles_both_msg_shapes(monkeypatch): + async def scenario(): + # Force deterministic level/message picks: first iteration uses the + # "%d"-templated message (exercises the formatting branch), second + # iteration uses a plain message (exercises the pass-through branch). + choices = iter(["INFO", "request handled in %dms", "DEBUG", "heartbeat ok"]) + monkeypatch.setattr(demo.random, "choice", lambda _seq: next(choices)) + monkeypatch.setattr(demo.random, "randint", lambda _a, _b: 77) + monkeypatch.setattr(demo.asyncio, "sleep", _instant_sleep) + + gen = demo.fake_logs("home", "mosquitto-6b8d2") + try: + line1 = await gen.__anext__() + line2 = await gen.__anext__() + finally: + await gen.aclose() + + assert "[mosquitto-6b8d2]" in line1 + assert "INFO" in line1 + assert "request handled in 77ms" in line1 + + assert "[mosquitto-6b8d2]" in line2 + assert "DEBUG" in line2 + assert "heartbeat ok" in line2 + + asyncio.run(scenario())