FabledSteward/tests/plugins/prometheus/test_pages.py

"""The Prometheus pages (#5669): targets not answering first and saying why,
what was left out, the series a target stored, and the editor."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS

from plugins.prometheus.routes import group_series, shown_value
from tests._templates import render_plugin_template
from tests.test_style_tokens import hex_colours


def _view(name, reach, says="Answering.", ok_minutes=None, capped=0):
    now = datetime.now(timezone.utc)
    return {"id": f"id-{name}", "name": name, "url": f"http://{name}:8081/metrics", "reach": reach,
            "enabled": reach != "paused",
            "says": says, "scraped_at": now - timedelta(seconds=20),
            "ok_at": now - timedelta(minutes=ok_minutes) if ok_minutes is not None else None,
            "interval": 60, "kept": 120, "dropped": 30, "capped": capped, "cap": 1000,
            "keep": ["fabledarchives_*"], "drop": [], "scrape_ms": 42.4}


async def test_a_target_not_answering_says_why_and_the_cap_is_shown(app):
    from quart import session
    targets = [_view("archives", "down", "Answered 401: the token was refused.", ok_minutes=180),
               _view("busy", "up", capped=37), _view("new", "pending"),
               _view("resting", "paused", "Paused: not scraped until it's resumed.")]
    async with app.test_request_context("/plugins/prometheus/"):
        session["user_role"] = "admin"
        page = await render_plugin_template(app, "prometheus", "prometheus/index.html", targets=targets)
    page = " ".join(page.split())
    assert '<span class="badge badge-crit">Not answering</span>' in page
    assert "Answered 401: the token was refused." in page and "Last answered" in page
    assert "37 series weren't stored" in page and "the cap of 1000" in page
    assert '<span class="stat-tile-value">42 ms</span>' in page
    assert "the first scrape runs within seconds of saving" in page
    # Paused and resumed from the list (#6419), each row saying which it is.
    assert '<span class="badge badge-dim">Paused</span>' in page and "not scraped until it&#39;s resumed" in page
    assert 'action="/plugins/prometheus/targets/id-resting/toggle"' in page
    assert page.count(">Pause</button>") == 3 and page.count(">Resume</button>") == 1
    assert 'href="/plugins/prometheus/targets/new"' in page and 'href="/plugins/prometheus/targets/id-busy/edit"' in page
    main = page[page.index("<main"):]
    assert not hex_colours(main), hex_colours(main)


async def test_a_viewer_sees_no_editing_and_an_empty_page_offers_to_add(app):
    from quart import session
    async with app.test_request_context("/plugins/prometheus/"):
        session["user_role"] = "viewer"
        page = await render_plugin_template(app, "prometheus", "prometheus/index.html",
                                            targets=[_view("archives", "up")])
        session["user_role"] = "admin"
        empty = await render_plugin_template(app, "prometheus", "prometheus/index.html", targets=[])
    assert "/edit" not in page and "targets/new" not in page and "/toggle" not in page
    assert "No targets yet." in empty and 'href="/plugins/prometheus/targets/new"' in empty


def test_series_are_grouped_by_metric_with_their_labels_and_units():
    at = datetime(2026, 10, 12, 12, 0, tzinfo=timezone.utc)
    stamp = (at - timedelta(hours=2)).timestamp()
    rows = [NS(resource_name="A", metric_name="is_up", value=1.0, recorded_at=at),
            NS(resource_name="A", metric_name="fa_disk_free_bytes", value=2 * 1024 ** 3, recorded_at=at),
            NS(resource_name="A:target=S,state=late", metric_name="fa_state", value=0.0, recorded_at=at),
            NS(resource_name="A:target=S,state=fresh", metric_name="fa_state", value=1.0, recorded_at=at),
            NS(resource_name="A:target=S", metric_name="fa_last_backup_timestamp_seconds", value=stamp,
               recorded_at=at)]
    groups = group_series("A", rows)
    assert [g["metric"] for g in groups] == ["fa_disk_free_bytes", "fa_last_backup_timestamp_seconds", "fa_state"]
    assert groups[0]["rows"][0]["labels"] == "" and groups[0]["rows"][0]["text"] == "2.0 GiB"
    assert groups[1]["rows"][0]["moment"] == datetime.fromtimestamp(stamp, timezone.utc)
    # A state set reads as one row naming the current state.
    assert groups[2]["states"] and not groups[0]["states"]
    assert [(r["labels"], r["text"], r["states"]) for r in groups[2]["rows"]] == [
        ("target=S", "fresh", ["fresh", "late"])]
    assert groups[2]["rows"][0]["resource"] == "A:target=S,state=fresh"
    assert shown_value("x_timestamp_seconds", 0) == {"text": "never", "moment": None}
    # Home Assistant names its Unix times *_time_seconds (#6423): a moment, not "20751d 21h".
    assert shown_value("homeassistant_last_updated_time_seconds", 1792964340.0) == {
        "text": "", "moment": datetime.fromtimestamp(1792964340.0, timezone.utc)}
    assert shown_value("jobs_per_sec", 0.25)["text"] == "0.25/s"
    assert shown_value("job_seconds", 90)["text"] == "1m 30s"
    assert shown_value("anything", None)["text"] == "—"


async def test_the_target_page_leads_with_its_state_then_its_series(app):
    from quart import session
    groups = [{"metric": "fa_state", "states": True, "rows": [
        {"labels": "target=S", "resource": "archives:target=S,state=late", "moment": None,
         "text": "late", "states": ["fresh", "late"]}]}]
    async with app.test_request_context("/plugins/prometheus/targets/id-archives"):
        session["user_role"] = "viewer"
        page = await render_plugin_template(app, "prometheus", "prometheus/target.html",
                                            target=_view("archives", "up", capped=3), groups=groups,
                                            total=1, q="")
    page = " ".join(page.split())
    assert page.index("Answering") < page.index("Series stored") < page.index("fa_state")
    assert "3 series weren't stored" in page and "Keeping fabledarchives_*" in page
    assert "resource=archives%3Atarget%3DS%2Cstate%3Dlate" in page and "metric=fa_state" in page
    assert 'title="One of: fresh, late">late</a>' in page and "a state set" in page
    assert 'href="/plugins/prometheus/">Prometheus</a>' in page          # its breadcrumb
    assert "/edit" not in page


async def test_the_form_never_echoes_the_token_and_offers_test(app):
    from quart import session
    target = NS(id="t1", name="archives")
    v = {"name": "archives", "url": "http://archives:8081/metrics", "interval_seconds": 60,
         "series_cap": 1000, "keep": "fabledarchives_*", "drop": "", "verify_tls": True,
         "enabled": True, "has_token": True}
    async with app.test_request_context("/plugins/prometheus/targets/t1/edit"):
        session["user_role"] = "admin"
        page = await render_plugin_template(app, "prometheus", "prometheus/form.html",
                                            target=target, v=v, error=None)
    assert 'name="token" type="password"' in page and "kept; type to change" in page
    assert 'name="clear_token"' in page and 'hx-post="/plugins/prometheus/targets/test"' in page
    assert "fabledarchives_*" in page and "Delete target" in page
    assert 'name="enabled"' not in page                # paused from the target's page, not here (#6419)