dashboards/tests/test_collectors.py

"""Collectors: the fetch half of every poll, exercised with a fake transport.

`httpx.MockTransport` means these run at the speed of the parser and still cover what the
live hosts do badly: a 401 from a rotated agent token, a half-degraded agent snapshot, a
gufo server that only answers `/v1/slots`, an engine that counts nothing yet.

NUT is covered here for the *monitor* bookkeeping (payload shape, command gating, status
transitions); the socket protocol itself is in test_nut.py.
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import Any, ClassVar

import httpx
import pytest

from dashboards.collectors import (
    NutMonitor,
    UpstreamError,
    fetch_engine,
    fetch_host_snapshot,
)
from dashboards.config import Target
from dashboards.rates import RateEstimator
from dashboards.sources import nut
from dashboards.sources.host import HostCollector

FIXTURES = Path(__file__).parent / "fixtures"


def client_for(handler: Any) -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=httpx.MockTransport(handler),
                             base_url="http://agent.test")


# --------------------------------------------------------------------------- #
# host agent
# --------------------------------------------------------------------------- #
def agent_body() -> dict[str, Any]:
    return json.loads((FIXTURES / "agent_driveripper.json").read_text())


def advance_cpu(body: dict[str, Any], *, total: float, idle: float) -> None:
    """Move the jiffy counters the way a real /proc/stat does between polls."""
    modes = body["cpu"]["modes"]
    for mode, value in modes.items():
        modes[mode] = value + (total if mode == "total" else idle if mode == "idle" else 0)


async def test_a_real_agent_snapshot_parses() -> None:
    body = agent_body()
    calls: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/snapshot"
        calls.append(request.url.path)
        return httpx.Response(200, json=body)

    target = Target("driveripper", "http://agent.test")
    collector = HostCollector(RateEstimator())
    async with client_for(handler) as client:
        first = (await fetch_host_snapshot(client, target, collector)).payload
        # CPU usage is Δjiffies/Δt: one scrape has no window, so it is n/a and not 0 %
        assert first["cpu"]["usage_pct"] is None
        # the counters have to actually move — total 0 delta means "cannot divide", not idle
        same = (await fetch_host_snapshot(client, target, collector)).payload
        assert same["cpu"]["usage_pct"] is None, "no counter movement must read as n/a"
        advance_cpu(body, total=1000, idle=700)
        result = await fetch_host_snapshot(client, target, collector)
    payload = result.payload
    assert payload["cpu"]["usage_pct"] == pytest.approx(30.0, abs=1.0)
    assert payload["name"] == "driveripper"
    assert payload["cpu"]["usage_pct"] is not None
    assert payload["memory"]["total_bytes"] > 0
    # real captures have real GPUs; the legend wants the short name, the tooltip the full one
    assert payload["gpus"], "driveripper's capture must yield at least one GPU"
    assert payload["gpus"][0]["model"] and "model_full" in payload["gpus"][0]
    assert any(s.metric == "cpu.usage_pct" for s in result.samples)


async def test_target_headers_are_sent_to_the_agent() -> None:
    seen: dict[str, str] = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["authorization"] = request.headers.get("authorization", "")
        return httpx.Response(200, json=agent_body())

    target = Target("gateway", "http://agent.test", headers={"Authorization": "Bearer tok"})
    async with client_for(handler) as client:
        await fetch_host_snapshot(client, target, HostCollector(RateEstimator()))
    assert seen["authorization"] == "Bearer tok"


@pytest.mark.parametrize("status,fragment", [
    (401, "token"),
    (403, "HTTP 403"),
    (500, "HTTP 500"),
])
async def test_agent_rejections_say_which_one_it_was(status, fragment) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status, text="nope")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))
    assert fragment in excinfo.value.message
    assert excinfo.value.status == 502


async def test_a_refused_connection_is_an_upstream_error_not_a_traceback() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("connection refused")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="unreachable"):
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))


@pytest.mark.parametrize("failure", ["refused", "timeout", "http-500"])
async def test_a_failing_host_never_echoes_its_token_in_the_message(failure: str) -> None:
    """An `UpstreamError` message is not just a log line: the poller keeps it in the health
    record (`/api/health`) and writes it to the `event` table (`/api/events`), both of which
    the browser reads — so the URL it names has to be the redacted one, on every failure path
    (AGENTS.md invariant 6). `parse_targets` normally moves a userinfo token into a header, so
    this is the backstop that keeps the guarantee true for any `Target` that still carries one.
    """
    def handler(request: httpx.Request) -> httpx.Response:
        if failure == "refused":
            raise httpx.ConnectError("connection refused")
        if failure == "timeout":
            raise httpx.ReadTimeout("too slow")
        return httpx.Response(500, text="boom")

    target = Target("gw", "http://SEKRIT-AGENT-TOKEN@agent.test")
    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_host_snapshot(client, target, HostCollector(RateEstimator()))
    message = excinfo.value.message
    assert "SEKRIT-AGENT-TOKEN" not in message, f"the token leaked through {failure}"
    # …and the message still names the machine it failed to reach
    assert "http://agent.test/snapshot" in message


async def test_a_failing_engine_never_echoes_credentials_in_its_url() -> None:
    """Same rule for the engines: the URL in the message is the redacted one."""
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("connection refused")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="unreachable") as excinfo:
            await fetch_engine(client, "gufo", "http://SEKRIT-ENGINE-TOKEN@agent.test", "",
                               RateEstimator())
    assert "SEKRIT-ENGINE-TOKEN" not in excinfo.value.message
    assert "http://agent.test/metrics" in excinfo.value.message


async def test_a_slow_agent_is_a_gateway_timeout() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("too slow")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))
    assert excinfo.value.status == 504


async def test_html_instead_of_json_is_reported_as_such() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, text="<html><body>502 Bad Gateway</body></html>")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="non-JSON"):
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))


async def test_an_agent_answering_not_ok_is_not_silently_empty() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"ok": False, "detail": "procfs unreadable"})

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="procfs unreadable"):
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))


async def test_a_degraded_agent_still_renders_and_says_what_broke() -> None:
    body = agent_body()
    body["errors"] = {"temperatures": "no hwmon sensors"}

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=body)

    async with client_for(handler) as client:
        result = await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                           HostCollector(RateEstimator()))
    assert result.payload["degraded"] == {"temperatures": "no hwmon sensors"}
    assert result.payload["memory"]["total_bytes"] > 0


# --------------------------------------------------------------------------- #
# LLM engines
# --------------------------------------------------------------------------- #
GUFO = (FIXTURES / "metrics_gufo.txt").read_text()
VLLM = (FIXTURES / "metrics_vllm.txt").read_text()


def bump(text: str, series: str, amount: float) -> str:
    lines = []
    for line in text.splitlines():
        if line.startswith(series + " "):
            head, _, value = line.rpartition(" ")
            line = f"{head} {float(value) + amount}"
        lines.append(line)
    return "\n".join(lines) + "\n"


async def test_first_gufo_poll_has_no_measured_rate_yet() -> None:
    """One scrape cannot produce Δ/Δt. The cell is n/a, not a stale single-request gauge."""
    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/metrics":
            return httpx.Response(200, text=GUFO)
        return httpx.Response(404, text="nope")

    async with client_for(handler) as client:
        result = await fetch_engine(client, "gufo", "http://engine.test", "", RateEstimator(),
                                    with_slots=True)
    decode = result.payload["metrics"]["decode_tokens_per_second"]
    assert decode["value"] is None
    assert decode["source"] is None
    assert result.payload["sessions"]["active"] == 0.0
    assert "no /slots answer" in result.payload["sessions"]["detail"]


async def test_second_gufo_poll_measures_throughput_over_a_window() -> None:
    state = {"text": GUFO}

    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/metrics":
            return httpx.Response(200, text=state["text"])
        if request.url.path == "/slots":
            return httpx.Response(200, json=[{"index": 0, "n_ctx": 8192, "predicted": 12}])
        return httpx.Response(404)

    rates = RateEstimator(span=10.0)
    async with client_for(handler) as client:
        await fetch_engine(client, "gufo", "http://engine.test", "", rates)
        state["text"] = bump(GUFO, "llamacpp:tokens_predicted_total", 40.0)
        result = await fetch_engine(client, "gufo", "http://engine.test", "", rates,
                                    with_slots=True)
    decode = result.payload["metrics"]["decode_tokens_per_second"]
    assert decode["source"].startswith("computed:Δllamacpp:")
    assert decode["derived"] is True
    assert decode["window"] > 0
    assert decode["value"] == pytest.approx(40.0 / decode["window"], rel=0.05)
    assert result.payload["sessions"]["source"] == "gufo /slots"
    assert result.payload["sessions"]["list"][0]["index"] == 0


async def test_gufo_slots_alias_is_used_when_plain_slots_is_missing() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/metrics":
            return httpx.Response(200, text=GUFO)
        if request.url.path == "/v1/slots":
            return httpx.Response(200, json={"slots": [{"index": 1}]})
        return httpx.Response(404)

    async with client_for(handler) as client:
        result = await fetch_engine(client, "gufo", "http://engine.test", "", RateEstimator(),
                                    with_slots=True)
    assert result.payload["sessions"]["list"] == [{"index": 1}]


async def test_vllm_metrics_parse_and_carry_the_api_key() -> None:
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request.headers.get("authorization", ""))
        return httpx.Response(200, text=VLLM)

    async with client_for(handler) as client:
        result = await fetch_engine(client, "vllm", "http://engine.test", "SECRET",
                                    RateEstimator())
    assert seen == ["Bearer SECRET"]
    metrics = result.payload["metrics"]
    assert metrics["num_requests_running"]["value"] is not None
    # vLLM publishes no per-session list: the page is told, rather than shown an empty table
    assert "counts, not sessions" in result.payload["sessions"]["detail"]
    assert any(s.metric.startswith("llm.") for s in result.samples)


@pytest.mark.parametrize("status", [404, 500, 503])
async def test_engine_http_errors_name_the_status(status) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status, text="")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match=f"HTTP {status}"):
            await fetch_engine(client, "vllm", "http://engine.test", "", RateEstimator())


async def test_engine_timeout_is_504() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("engine busy")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_engine(client, "gufo", "http://engine.test", "", RateEstimator())
    assert excinfo.value.status == 504


# --------------------------------------------------------------------------- #
# NUT monitor
# --------------------------------------------------------------------------- #
class FakeSimulator:
    """Stands in for NutSimulator: hands back readings we control."""

    def __init__(self, readings: list[nut.UpsReading]) -> None:
        self.readings = readings

    async def read(self) -> list[nut.UpsReading]:
        return list(self.readings)


def reading(status: str, **vars: str) -> nut.UpsReading:
    item = nut.UpsReading(name="cyberpower", description="CP1350PFCLCD", ts=1_700_000_000.0)
    item.vars = {"ups.status": status, "battery.charge": "92", "ups.load": "35",
                 "battery.runtime": "1800", **vars}
    item.commands = ["beeper.mute", "test.panel.start", "test.battery.start.deep"]
    return item


class SettingsStub:
    nut_host, nut_port, nut_timeout = "ups.test", 3493, 5.0
    nut_user = nut_password = ""
    nut_interval = 5.0
    nut_enabled = True
    actions_allowed = True
    read_only = False


async def test_monitor_payload_labels_the_simulator() -> None:
    monitor = NutMonitor(SettingsStub(), simulator=FakeSimulator([reading("OL")]))
    result = await monitor.collect()
    payload = result.payload
    assert payload["simulated"] is True
    assert payload["server"] is None, "a simulator must not claim a server address"
    assert payload["actions_enabled"] is True
    card = payload["ups"][0]
    assert card["name"] == "cyberpower"
    assert card["status"] == "OL"
    assert card["numbers"]["charge_pct"] == 92.0
    # a driver that reports no battery date, temperature, or self-test result is `unknown`,
    # never a silent "healthy" — and never a failure either
    assert card["health"]["overall"] == "unknown", card["health"]
    assert card["health"]["counts"]["fail"] == 0
    assert card["health"]["counts"]["warn"] == 0
    # commands are offered with their risk, and the dangerous one is present but marked
    risks = {item["command"]: item for item in card["commands"]}
    assert risks["beeper.mute"]["enabled"] is True
    assert risks["test.battery.start.deep"]["risk"] == "dangerous"


async def test_a_fully_report_ups_is_called_healthy() -> None:
    full = reading("OL", **{"battery.temperature": "27.4", "battery.mfr.date": "2024-06-01",
                            "battery.test.result": "OK", "input.voltage": "230"})
    full.vars["battery.charge"] = "100"
    card = (await NutMonitor(SettingsStub(), simulator=FakeSimulator([full])).collect()).payload["ups"][0]
    assert card["health"]["overall"] == "healthy", card["health"]
    assert card["health"]["counts"]["unknown"] == 0


async def test_monitor_sends_history_samples_for_every_ups_metric() -> None:
    monitor = NutMonitor(SettingsStub(), simulator=FakeSimulator([reading("OB")]))
    result = await monitor.collect()
    names = {sample.metric for sample in result.samples}
    assert "ups.charge_pct" in names
    assert "ups.load_pct" in names
    assert all(sample.source == "ups" for sample in result.samples)


async def test_status_changes_become_one_event_each() -> None:
    monitor = NutMonitor(SettingsStub(), simulator=FakeSimulator([reading("OL")]))
    assert (await monitor.collect()).events == []
    monitor.simulator = FakeSimulator([reading("OL")])
    assert (await monitor.collect()).events == [], "an unchanged status is not news"
    monitor.simulator = FakeSimulator([reading("OB")])
    events = (await monitor.collect()).events
    assert len(events) == 1
    assert events[0].kind == "status"
    assert "OL → OB" in events[0].message
    assert events[0].severity == "warn", "on battery is a warning, not a failure"
    monitor.simulator = FakeSimulator([reading("OL FUB")])
    events = (await monitor.collect()).events
    assert events[0].severity == "error", "FUB (the UPS decided to power off) is a failure"


async def test_a_self_test_result_is_reported_when_it_changes() -> None:
    monitor = NutMonitor(SettingsStub(), simulator=FakeSimulator([reading("OL")]))
    await monitor.collect()
    # the variable is `ups.test.result` (drivers also use battery.test.result) — see VAR_MAP
    monitor.simulator = FakeSimulator([reading("OL", **{"ups.test.result": "INPROGRESS"})])
    assert (await monitor.collect()).events == [], "first sighting is not news"
    monitor.simulator = FakeSimulator([reading("OL", **{"ups.test.result": "FAIL"})])
    events = (await monitor.collect()).events
    assert [event.kind for event in events] == ["test"]
    assert events[0].severity == "warn"
    monitor.simulator = FakeSimulator([reading("OL", **{"ups.test.result": "OK"})])
    assert (await monitor.collect()).events[0].severity == "info"


async def test_gating_is_applied_where_the_commands_are_built() -> None:
    class ReadOnly(SettingsStub):
        actions_allowed = False

    monitor = NutMonitor(ReadOnly(), simulator=FakeSimulator([reading("OL")]))
    card = (await monitor.collect()).payload["ups"][0]
    assert all(item["enabled"] is False for item in card["commands"])
    assert all(item["reason"] for item in card["commands"]), "every refusal must explain itself"
    assert set(card["advertised_commands"]) == {"beeper.mute", "test.panel.start",
                                                "test.battery.start.deep"}, (
        "the driver still advertises them — the refusal is ours, and must say so")


class FakeNutClient:
    """The real upsd branch without a socket: vars work, the optional lists do not."""

    instances: ClassVar[list[FakeNutClient]] = []

    def __init__(self, host: str, port: int, **kwargs: Any) -> None:
        self.host, self.port, self.kwargs = host, port, kwargs
        self.entered = False
        FakeNutClient.instances.append(self)

    async def __aenter__(self) -> FakeNutClient:
        self.entered = True
        return self

    async def __aexit__(self, *exc: Any) -> None:
        self.entered = False

    async def list_ups(self) -> list[dict[str, str]]:
        return [{"name": "cyberpower", "description": "CP1350PFCLCD"}]

    async def list_vars(self, ups: str) -> dict[str, str]:
        return {"ups.status": "OL", "battery.charge": "100", "battery.temperature": "301.1"}

    async def list_cmds(self, ups: str) -> list[str]:
        raise nut.NutError("this driver exports no commands")

    async def list_rw_vars(self, ups: str) -> list[str]:
        raise nut.NutError("LIST RW denied")


async def test_real_upsd_branch_survives_drivers_that_refuse_the_optional_lists(
        monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(nut, "NutClient", FakeNutClient)
    monitor = NutMonitor(SettingsStub())
    result = await monitor.collect()
    client = FakeNutClient.instances[-1]
    assert client.entered is False, "the connection must not outlive one poll"
    assert client.kwargs["timeout"] == 5.0
    card = result.payload["ups"][0]
    assert card["commands"] == [] and card["advertised_commands"] == []
    assert card["numbers"]["charge_pct"] == 100.0
    # a driver reporting Kelvin is normalised, not charted as 301 °C
    assert card["numbers"]["battery_temperature"] == pytest.approx(301.1 - 273.15, abs=0.01)
    samples = {sample.metric: sample.value for sample in result.samples}
    assert samples["ups.battery_temperature_c"] == pytest.approx(27.95, abs=0.01)


async def test_a_real_server_names_itself_in_the_payload(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(nut, "NutClient", FakeNutClient)
    payload = (await NutMonitor(SettingsStub()).collect()).payload
    assert payload["simulated"] is False
    assert payload["server"] == "ups.test:3493"