133 lines
4.1 KiB
Python
133 lines
4.1 KiB
Python
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
|