FabledSteward/tests/plugins/prometheus/test_service.py

"""The Prometheus plugin's pure parts: which series are stored and under what
names, the form, and what a page says about the newest scrape (#5668)."""
from datetime import datetime, timedelta, timezone

import httpx
import pytest

from plugins.prometheus.models import PrometheusTarget
from plugins.prometheus.scheduler import is_due
from plugins.prometheus.scraper import ScrapeError, scrape
from plugins.prometheus.service import (FormError, map_families, parse_form, patterns,
                                        rate_name, resource_for, target_state, wanted)
from steward.core import prometheus
from steward.core.counter_rates import CounterRates

TEXT = """\
# TYPE fa_backup_target_state gauge
fa_backup_target_state{target="Steward",state="fresh"} 1
fa_backup_target_state{target="Steward",state="late"} 0
# TYPE fa_http_requests_total counter
fa_http_requests_total{route="GET /",status="200"} 100
# TYPE fa_job_seconds histogram
fa_job_seconds_bucket{le="1"} 2
fa_job_seconds_bucket{le="+Inf"} 3
fa_job_seconds_sum 4.5
fa_job_seconds_count 3
# TYPE fa_rpc summary
fa_rpc{quantile="0.5"} 0.2
fa_rpc_count 9
fa_disk_free_bytes 1e9
fa_broken NaN
other_thing 7
"""


def _mapped(keep=(), drop=(), cap=1000, target="Archives"):
    return map_families(target, prometheus.parse(TEXT), list(keep), list(drop), cap)


def test_gauges_stay_counters_become_rates_and_distributions_keep_sum_and_count():
    got = {(s.resource, s.metric): (s.value, s.counter) for s in _mapped().series}
    assert got == {
        ("Archives:target=Steward,state=fresh", "fa_backup_target_state"): (1.0, False),
        ("Archives:target=Steward,state=late", "fa_backup_target_state"): (0.0, False),
        ("Archives:route=GET /,status=200", "fa_http_requests_per_sec"): (100.0, True),
        ("Archives", "fa_job_seconds_sum_per_sec"): (4.5, True),
        ("Archives", "fa_job_seconds_count_per_sec"): (3.0, True),
        ("Archives", "fa_rpc_count_per_sec"): (9.0, True),
        ("Archives", "fa_disk_free_bytes"): (1e9, False),
        ("Archives", "other_thing"): (7.0, False),
    }


def test_keep_and_drop_patterns_choose_families_and_drop_wins():
    m = _mapped(keep=["fa_*"], drop=["fa_http_*"])
    assert {s.metric for s in m.series} == {
        "fa_backup_target_state", "fa_job_seconds_sum_per_sec", "fa_job_seconds_count_per_sec",
        "fa_rpc_count_per_sec", "fa_disk_free_bytes"}
    assert m.dropped == 2          # the counter's one sample, other_thing's one
    assert wanted("x", [], []) and not wanted("x", ["y*"], []) and not wanted("y1", ["y*"], ["y1"])


def test_the_series_cap_and_overlong_names_are_counted_not_stored():
    m = _mapped(cap=3)
    assert len(m.series) == 3 and m.capped == 5
    long = map_families("t" * 250, prometheus.parse('m{a="bcdefgh"} 1\n'), [], [], 10)
    assert long.series == [] and long.capped == 1


def test_names():
    assert rate_name("jobs_total") == "jobs_per_sec"
    assert rate_name("job_seconds_sum") == "job_seconds_sum_per_sec"
    assert resource_for("T", ()) == "T"
    assert resource_for("T", (("a", "1"), ("b", "x"))) == "T:a=1,b=x"
    assert patterns("a_*, b_*\n\n a_* \nc") == ["a_*", "b_*", "c"]


def test_a_counter_becomes_a_rate_and_a_reset_is_skipped():
    rates = CounterRates()
    t0 = datetime(2026, 10, 12, 12, 0, tzinfo=timezone.utc)
    key = ("t1", "Archives", "x_per_sec")
    assert rates.rate(key, 100, t0) is None
    assert rates.rate(key, 160, t0 + timedelta(seconds=60)) == 1.0
    assert rates.rate(key, 5, t0 + timedelta(seconds=120)) is None        # the service restarted
    assert rates.rate(key, 35, t0 + timedelta(seconds=150)) == 1.0
    rates.forget("t1")
    assert rates.rate(key, 99, t0 + timedelta(seconds=180)) is None


class _Form(dict):
    def getlist(self, key):
        v = self.get(key)
        return v if isinstance(v, list) else ([v] if v else [])


def test_the_form_checks_what_it_stores_and_keeps_a_blank_token():
    values = parse_form(_Form(name="Archives", url="http://archives:8081/metrics", token="s3cret",
                              interval_seconds="30", series_cap="500", keep="fabledarchives_*",
                              enabled="on", verify_tls="on"))
    assert values["token"].startswith("enc:") or values["token"] == "s3cret"
    assert (values["interval_seconds"], values["series_cap"], values["keep"], values["enabled"]) == (
        30, 500, ["fabledarchives_*"], True)
    existing = PrometheusTarget(name="Archives", url="http://a/metrics", token="enc:v1:kept")
    again = parse_form(_Form(name="Archives", url="http://a/metrics"), existing)
    assert again["token"] == "enc:v1:kept" and again["enabled"] is False
    cleared = parse_form(_Form(name="Archives", url="http://a/metrics", clear_token="on"), existing)
    assert cleared["token"] == ""
    for bad in (dict(name="", url="http://a"), dict(name="a:b", url="http://a"),
                dict(name="a", url="ftp://a"), dict(name="a", url="http://a", interval_seconds="5"),
                dict(name="a", url="http://a", series_cap="0")):
        with pytest.raises(FormError):
            parse_form(_Form(bad))


def test_state_and_due():
    now = datetime(2026, 10, 12, 12, 0, tzinfo=timezone.utc)
    t = PrometheusTarget(name="a", url="http://a", enabled=True, interval_seconds=60)
    assert target_state(t, now).reach == "pending" and is_due(t, now)
    t.last_scraped_at = now - timedelta(seconds=30)
    assert target_state(t, now).reach == "up" and not is_due(t, now)
    t.last_scraped_at = now - timedelta(seconds=58)
    assert is_due(t, now)                          # half a tick of slack
    t.last_error = "Couldn't connect: refused."
    t.last_scraped_at = now - timedelta(seconds=30)
    assert (target_state(t, now).reach, target_state(t, now).says) == ("down", "Couldn't connect: refused.")
    t.last_scraped_at = now - timedelta(minutes=10)
    assert target_state(t, now).reach == "stale"
    t.enabled = False
    assert target_state(t, now).reach == "off" and not is_due(t, now)


@pytest.mark.asyncio
async def test_a_scrape_sends_the_token_and_says_why_it_failed():
    seen = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["auth"] = request.headers.get("Authorization")
        if request.url.path == "/denied":
            return httpx.Response(401)
        if request.url.path == "/html":
            return httpx.Response(200, text="<html>hello</html>")
        return httpx.Response(200, text="up_thing 1\n")

    transport = httpx.MockTransport(handler)
    fams = await scrape("http://svc/metrics", "tok", transport=transport)
    assert list(fams) == ["up_thing"] and seen["auth"] == "Bearer tok"
    with pytest.raises(ScrapeError, match="token was refused"):
        await scrape("http://svc/denied", "bad", transport=transport)
    with pytest.raises(ScrapeError, match="no metrics"):
        await scrape("http://svc/html", transport=transport)