test(bstein-home): cover lab status routes

backend/tests/test_lab_routes.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+"""Tests for lab health query helpers and route payloads."""
+
+import json
+from urllib.error import URLError
+
+from atlas_portal.app_factory import create_app
+from atlas_portal.routes import lab
+
+
+class DummyUrlResponse:
+    """Small context-manager response for urlopen tests."""
+
+    def __init__(self, payload: dict | str, *, status: int = 200) -> None:
+        self.payload = payload
+        self.status = status
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def read(self, size: int | None = None) -> bytes:
+        """Return JSON or string content as response bytes."""
+
+        if isinstance(self.payload, str):
+            return self.payload.encode("utf-8")
+        return json.dumps(self.payload).encode("utf-8")
+
+
+def test_vm_query_success_and_empty_paths(monkeypatch) -> None:
+    payloads = [
+        {"status": "success", "data": {"result": [{"value": [0, "2"]}, {"value": [0, "5"]}]}},
+        {"status": "error"},
+        {"status": "success", "data": {"result": []}},
+        {"status": "success", "data": {"result": [{"bad": []}]}},
+    ]
+
+    def fake_urlopen(url, timeout):
+        return DummyUrlResponse(payloads.pop(0))
+
+    monkeypatch.setattr(lab, "urlopen", fake_urlopen)
+
+    assert lab._vm_query("up") == 5.0
+    assert lab._vm_query("up") is None
+    assert lab._vm_query("up") is None
+    assert lab._vm_query("up") is None
+
+
+def test_http_ok_status_substring_and_errors(monkeypatch) -> None:
+    responses = [
+        DummyUrlResponse("service ok"),
+        DummyUrlResponse("wrong body"),
+        DummyUrlResponse("bad", status=503),
+    ]
+
+    def fake_urlopen(url, timeout):
+        item = responses.pop(0)
+        if item == "raise":
+            raise URLError("offline")
+        return item
+
+    monkeypatch.setattr(lab, "urlopen", fake_urlopen)
+
+    assert lab._http_ok("https://grafana", expect_substring="ok")
+    assert not lab._http_ok("https://grafana", expect_substring="ok")
+    assert not lab._http_ok("https://grafana")
+    responses.append("raise")
+    assert not lab._http_ok("https://grafana")
+
+
+def test_lab_status_uses_cache_and_probe_fallbacks(monkeypatch) -> None:
+    app = create_app()
+    client = app.test_client()
+    lab._LAB_STATUS_CACHE["ts"] = 0.0
+    lab._LAB_STATUS_CACHE["value"] = None
+    monkeypatch.setattr(lab.settings, "OCEANUS_NODE_EXPORTER_URL", "https://oceanus.example.dev/metrics")
+
+    calls: list[str] = []
+
+    def fake_http_ok(url, expect_substring=None):
+        calls.append(url)
+        return "grafana" in url or "oceanus" in url
+
+    monkeypatch.setattr(lab, "_http_ok", fake_http_ok)
+    monkeypatch.setattr(lab, "_vm_query", lambda expr: 1.0)
+
+    response = client.get("/api/lab/status")
+    payload = response.get_json()
+
+    assert response.status_code == 200
+    assert payload["connected"] is True
+    assert payload["atlas"]["source"] == "grafana"
+    assert payload["oceanus"]["source"] == "node-exporter"
+
+    second = client.get("/api/lab/status")
+    assert second.get_json() == payload
+
+    lab._LAB_STATUS_CACHE["ts"] = 0.0
+    lab._LAB_STATUS_CACHE["value"] = None
+    monkeypatch.setattr(lab, "_http_ok", lambda *a, **k: False)
+    monkeypatch.setattr(lab, "_vm_query", lambda expr: 0.0)
+
+    response = client.get("/api/lab/status")
+    payload = response.get_json()
+
+    assert payload["atlas"]["source"] == "victoria-metrics"
+    assert payload["atlas"]["up"] is False
+    assert payload["oceanus"]["known"] is False
+
+
+def test_lab_status_handles_probe_exceptions(monkeypatch) -> None:
+    app = create_app()
+    client = app.test_client()
+    lab._LAB_STATUS_CACHE["ts"] = 0.0
+    lab._LAB_STATUS_CACHE["value"] = None
+
+    def boom(*args, **kwargs):
+        raise RuntimeError("offline")
+
+    monkeypatch.setattr(lab, "_http_ok", boom)
+    monkeypatch.setattr(lab, "_vm_query", boom)
+
+    response = client.get("/api/lab/status")
+    payload = response.get_json()
+
+    assert response.status_code == 200
+    assert payload["connected"] is False
+    assert payload["atlas"]["known"] is False
+    assert payload["oceanus"]["known"] is False