dashboards/tests/test_prom.py

"""The hand-rolled Prometheus parser, tested against real engine expositions.

This parser exists so the app has no Prometheus dependency for the two upstreams that
already speak the format. That makes it load-bearing on text we do not control: llama.cpp
emits names with colons, vLLM emits label sets and histograms, and both are free to change
them. Fixtures are captured verbatim (scripts/capture-fixtures.sh --with-llm).
"""
from __future__ import annotations

import math

from dashboards.prom import Prom


def test_parses_counters_and_gauges_from_a_real_capture(metrics_text) -> None:
    prom = Prom(metrics_text("gufo"))
    # the capture's counters and gauges are live numbers (volatile between
    # re-captures): the pin is that they parse at all, in a sane range
    generated = prom.scalar(["llamacpp:tokens_predicted_total"])
    assert generated is not None and generated >= 0
    usage = prom.scalar(["llamacpp:kv_cache_usage_ratio"])
    assert usage is not None and 0.0 <= usage <= 1.0
    assert prom.parse_errors == 0


def test_help_and_type_comments_are_not_samples(metrics_text) -> None:
    prom = Prom(metrics_text("gufo"))
    assert not any(name.startswith("#") for name in prom.series)
    assert "# HELP llamacpp:prompt_tokens_total" not in prom.series


def test_labels_are_parsed_and_filterable(metrics_text) -> None:
    prom = Prom(metrics_text("vllm"))
    running = prom.values(["vllm:num_requests_running"])
    assert running, "vLLM exposes num_requests_running"
    by_model = prom.by_label(["vllm:generation_tokens_total"], "model_name")
    assert by_model and all(model for model in by_model)


def test_values_by_series_keeps_the_label_sets_apart() -> None:
    """Counter deltas diff per label set; the summed view would let a joining series
    lend its carried value to one interval (phase 13). This is the accessor that view."""
    prom = Prom('x{a="1"} 5.0\nx{a="2"} 7.0\ny 1.0\n')
    assert prom.values_by_series(["x"]) == [({"a": "1"}, 5.0), ({"a": "2"}, 7.0)]
    assert prom.values_by_series(["y"]) == [({}, 1.0)]
    assert prom.values_by_series(["x"], {"a": "2"}) == [({"a": "2"}, 7.0)]
    assert prom.values_by_series(["nope"]) == []


def test_values_by_series_on_a_real_labelled_capture(metrics_text) -> None:
    prom = Prom(metrics_text("vllm"))
    series = prom.values_by_series(["vllm:generation_tokens_total"])
    assert series, "vLLM exposes generation_tokens_total"
    assert all(labels.get("model_name") for labels, _v in series), "the capture lost its labels"


def test_aggregations(metrics_text) -> None:
    prom = Prom(metrics_text("vllm"))
    total = prom.scalar(["vllm:prompt_tokens_total"], agg="sum")
    assert total is not None and total > 1e6
    # ratios must never be summed across label sets — a 3-way sum would read as 300% used
    usage = prom.scalar(["vllm:kv_cache_usage_perc"], agg="max")
    assert usage is not None and 0.0 <= usage <= 1.0


def test_missing_series_is_none_not_zero(metrics_text) -> None:
    prom = Prom(metrics_text("gufo"))
    assert prom.scalar(["vllm:does_not_exist"]) is None
    assert prom.values(["vllm:does_not_exist"]) == []
    assert prom.by_label(["nope"], "model_name") == {}


def test_histogram_mean_and_p95(metrics_text) -> None:
    hist = Prom(metrics_text("vllm")).histogram(["vllm:time_to_first_token_seconds"])
    assert hist is not None
    assert hist["count"] and hist["count"] > 0
    assert hist["mean"] and hist["mean"] > 0
    assert hist["p95"] is not None
    assert hist["p95"] >= hist["mean"], "p95 below the mean means the interpolation is wrong"
    assert math.isfinite(hist["p95"])


def test_histogram_resolves_through_suffixes(metrics_text) -> None:
    prom = Prom(metrics_text("vllm"))
    # the bare name never appears in the exposition; only _bucket/_sum/_count do
    assert "vllm:time_to_first_token_seconds" not in prom.series
    assert prom.resolve_hist(["vllm:time_to_first_token_seconds"]) == "vllm:time_to_first_token_seconds"
    assert prom.resolve(["vllm:time_to_first_token_seconds"]) is None


def test_resolve_returns_the_first_present_name() -> None:
    prom = Prom('a_total 5\nb_total 7\n')
    assert prom.resolve(["missing_total", "b_total", "a_total"]) == "b_total"
    assert prom.resolve(["nope"]) is None


def test_label_match_narrows_the_series() -> None:
    prom = Prom('x{engine="0"} 1\nx{engine="1"} 2\n')
    assert prom.scalar(["x"], agg="sum") == 3
    assert prom.scalar(["x"], label_match={"engine": "1"}) == 2
    assert prom.scalar(["x"], label_match={"engine": "9"}) is None


def test_malformed_lines_are_counted_not_fatal() -> None:
    prom = Prom("good_metric 1\nthis is not prometheus\nalso_bad{unclosed 2\n")
    assert prom.scalar(["good_metric"]) == 1.0
    assert prom.parse_errors >= 1


def test_empty_exposition_is_harmless() -> None:
    prom = Prom("")
    assert prom.series == {}
    assert prom.scalar(["anything"]) is None
    assert prom.histogram(["anything"]) is None


def test_quantile_ignores_non_finite_values() -> None:
    prom = Prom('m_sum 10\ncount 2\nm{le="1"} NaN\n')
    assert prom.scalar(["m"], label_match={"le": "1"}) is None


def test_multiple_histogram_label_sets_do_not_invent_a_p95(metrics_text) -> None:
    """Mixing per-DP-rank buckets into one cumulative curve would fabricate a percentile."""
    two = (
        'h_sum{dp="0"} 10\nh_sum{dp="1"} 10\n'
        'h_count{dp="0"} 2\nh_count{dp="1"} 2\n'
        'h_bucket{dp="0",le="1"} 1\nh_bucket{dp="0",le="+Inf"} 2\n'
        'h_bucket{dp="1",le="1"} 1\nh_bucket{dp="1",le="+Inf"} 2\n'
    )
    hist = Prom(two).histogram(["h"])
    assert hist is not None
    assert hist["mean"] == 5.0
    assert hist["p95"] is None, "p95 across mixed label sets is a made-up number"