bstein-dev-home/backend/tests/test_lab_routes.py

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_vm_query_vector_filters_bad_samples(monkeypatch) -> None:
    payload = {
        "status": "success",
        "data": {
            "result": [
                {"metric": {"service": "nextcloud"}, "value": [0, "12"]},
                {"metric": {"service": "broken"}, "value": []},
            ],
        },
    }

    monkeypatch.setattr(lab, "urlopen", lambda url, timeout: DummyUrlResponse(payload))

    assert lab._vm_query_vector("topk(1, up)") == [{"metric": {"service": "nextcloud"}, "value": 12.0}]


def test_public_active_service_uses_external_ai_chat_url(monkeypatch) -> None:
    monkeypatch.setattr(lab.settings, "LAB_ACTIVE_SERVICE_QUERY", "topk(1, fake)")
    monkeypatch.setattr(
        lab,
        "_vm_query_vector",
        lambda expr: [{"metric": {"service": "atlas-ai-chat"}, "value": 9.0}],
    )

    assert lab._public_active_service() == {
        "known": True,
        "label": "AI Chat",
        "url": "https://bstein.dev/ai/chat",
        "window": lab.settings.LAB_ACTIVE_SERVICE_WINDOW,
    }


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, "ATLAS_DB_HOST_HEALTH_URL", "https://db.example.dev/health")
    monkeypatch.setattr(lab.settings, "ATLAS_JUMPHOST_HEALTH_URL", "https://jump.example.dev/health")

    calls: list[str] = []

    def fake_http_ok(url, expect_substring=None):
        calls.append(url)
        return "grafana" in url or "db" in url or "jump" in url

    monkeypatch.setattr(lab, "_http_ok", fake_http_ok)
    monkeypatch.setattr(lab, "_vm_query", lambda expr: 1.0)
    monkeypatch.setattr(
        lab,
        "_vm_query_vector",
        lambda expr: [{"metric": {"service": "nextcloud-web"}, "value": 8.0}],
    )

    response = client.get("/api/lab/status")
    payload = response.get_json()

    assert response.status_code == 200
    assert response.headers["Cache-Control"] == "no-store, max-age=0"
    assert response.headers["Pragma"] == "no-cache"
    assert payload["connected"] is True
    assert payload["atlas"]["source"] == "grafana"
    assert payload["dedicated_hosts"]["up"] is True
    assert payload["dedicated_hosts"]["up_count"] == 2
    assert payload["dedicated_hosts"]["hosts"][0]["label"] == "Database node"
    assert payload["active_service"] == {
        "known": True,
        "label": "Nextcloud",
        "url": "https://cloud.bstein.dev",
        "window": lab.settings.LAB_ACTIVE_SERVICE_WINDOW,
    }
    assert payload["oceanus"]["source"] == "atlas-member"

    second = client.get("/api/lab/status")
    assert second.get_json() == payload
    assert second.headers["Cache-Control"] == "no-store, max-age=0"

    lab._LAB_STATUS_CACHE["ts"] = 0.0
    lab._LAB_STATUS_CACHE["value"] = None
    monkeypatch.setattr(lab.settings, "ATLAS_DB_HOST_HEALTH_URL", "")
    monkeypatch.setattr(lab.settings, "ATLAS_JUMPHOST_HEALTH_URL", "")
    monkeypatch.setattr(lab, "_http_ok", lambda *a, **k: False)
    monkeypatch.setattr(lab, "_vm_query", lambda expr: 0.0)
    monkeypatch.setattr(lab, "_vm_query_vector", lambda expr: [])

    response = client.get("/api/lab/status")
    payload = response.get_json()

    assert payload["atlas"]["source"] == "victoria-metrics"
    assert payload["atlas"]["up"] is False
    assert payload["dedicated_hosts"]["known"] is True
    assert payload["dedicated_hosts"]["up"] is False
    assert payload["active_service"]["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)
    monkeypatch.setattr(lab, "_vm_query_vector", 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["dedicated_hosts"]["known"] is False
    assert payload["active_service"]["known"] is False