FabledSteward/tests/plugins/prometheus/test_templates.py

"""Prometheus widget templates (#5690): the checker says what is wrong and
where, and the renderer turns a template and a target's newest values into a
verdict, then blocks with their problems first."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS

import pytest

from plugins.prometheus.template_format import TemplateError, check, duration, problems
from plugins.prometheus.template_view import render

NOW = datetime(2026, 10, 12, 19, 0, tzinfo=timezone.utc)
T = "Archives"


def before(**kw) -> float:
    return (NOW - timedelta(**kw)).timestamp()


def latest(series: dict) -> list:
    """PluginMetricLatest-like rows from {(labels, metric): value}, labels
    as stored after the target's name."""
    return [NS(resource_name=f"{T}:{labels}" if labels else T, metric_name=metric, value=value,
               recorded_at=NOW) for (labels, metric), value in series.items()]


def states(prefix: str, current: str, all_states: list[str]) -> dict:
    return {(f"{prefix},state={s}", "fa_target_state"): float(s == current) for s in all_states}


TARGET_STATES = ["disabled", "fresh", "late", "missing", "waiting"]

SERIES = {
    **states("target=Steward,target_id=1", "fresh", TARGET_STATES),
    **states("target=Photos,target_id=2", "late", TARGET_STATES),
    **states("target=Old,target_id=3", "disabled", TARGET_STATES),
    ("target=Steward,target_id=1", "fa_target_last_backup_timestamp_seconds"): before(hours=16),
    ("target=Photos,target_id=2", "fa_target_last_backup_timestamp_seconds"): before(hours=30),
    ("", "fa_disk_free_bytes"): 17.8e12,
    ("", "fa_disk_size_bytes"): 46e12,
    ("", "fa_files"): 94116.0,
    ("", "fa_files_checked"): 93310.0,
    ("", "fa_malware_held_files"): 2.0,
    ("person=Bryan,person_id=9,kind=files", "fa_person_stored_bytes"): 4.7e11,
    ("person=Bryan,person_id=9,kind=trash", "fa_person_stored_bytes"): 1.2e9,
    ("person=Ana,person_id=8,kind=files", "fa_person_stored_bytes"): 3.0e10,
    ("job=Nightly,job_id=5", "fa_job_next_run_timestamp_seconds"): (NOW + timedelta(hours=8)).timestamp(),
    ("version=2026.10.12.1717,channel=dev,commit=3605d0a", "fa_build_info"): 1.0,
}

TEMPLATE = """\
name: Test Archives
version: 1
expects: fa_
blocks:
  - states:
      title: Backup targets
      metric: fa_target_state
      name: target
      healthy: [fresh]
      quiet: [disabled, waiting]
      warning: [late]
      columns:
        - {label: Last backup, metric: fa_target_last_backup_timestamp_seconds}
  - values:
      title: Storage
      items:
        - {label: Disk free, metric: fa_disk_free_bytes, of: fa_disk_size_bytes, warn_below: 10%,
           sparkline: true}
        - {label: Files checked, metric: fa_files_checked, of: fa_files, unit: count}
        - {label: Malware held, metric: fa_malware_held_files, unit: count, warn_above: 0}
  - times:
      items:
        - {label: Next run, metric: fa_job_next_run_timestamp_seconds, name: job}
        - {label: Last backup, metric: fa_target_last_backup_timestamp_seconds, name: target,
           warn_after: 26h, crit_after: 3d}
  - table:
      title: Stored per person
      metric: fa_person_stored_bytes
      rows: person
      columns: kind
  - info:
      metric: fa_build_info
      labels: [version, commit]
"""


# ── The checker ──────────────────────────────────────────────────────────────

def test_a_template_reads_into_its_blocks():
    t = check(TEMPLATE)
    assert (t.name, t.version, t.expects) == ("Test Archives", 1, "fa_")
    assert [b.kind for b in t.blocks] == ["states", "values", "times", "table", "info"]
    assert t.blocks[2].title == "Times"                       # a block's default title
    disk = t.blocks[1].spec["items"][0]
    assert disk["of"] == "fa_disk_size_bytes" and disk["warn_below"].share and disk["sparkline"]
    assert t.blocks[2].spec["items"][1]["warn_after"] == 26 * 3600


def _said(text: str) -> list[str]:
    return [str(p) for p in problems(text)]


def test_every_problem_is_said_with_its_line():
    said = _said("""\
name: X
version: 0
blocks:
  - states:
      metric: fa_state
      healthy: [ok]
      warning: [ok]
      colour: red
  - gauge: {metric: x}
  - values:
      items:
        - {label: Free, metric: fa free, warn_below: 10%}
""")
    assert said == [
        "Line 2: `version` is the template's own version: a whole number from 1. "
        "Raise it when you change a template you have shared.",
        "Line 8: Block 1 (states) has no `colour`. It takes: title, metric, where, name, healthy, "
        "quiet, warning, warn_past, crit_past, columns.",
        "Line 9: Block 2 is “gauge”; the kinds are: states, values, times, table, info.",
        "Line 12: `metric` in an item of block 3 (values) should be a metric name as Steward stores it, "
        "like fabledarchives_files (a counter is stored as <name>_per_sec).",
        "Line 12: `warn_below` in an item of block 3 (values) is a share (10%), which needs `of`: "
        "the metric it is a share of.",
    ]


def test_a_state_in_two_lists_and_a_yaml_boolean_are_caught():
    said = _said("name: X\nversion: 1\nblocks:\n  - states: {metric: s_state, healthy: [ok], warning: [ok]}\n")
    assert said == ["Line 4: “ok” is in both `healthy` and `warning` in block 1 (states)."]
    said = _said("name: X\nversion: 1\nblocks:\n  - states: {metric: s_state, healthy: [on]}\n")
    assert "put it in quotes, like 'off'" in said[0]


def test_what_isnt_a_template_says_so():
    broken = _said("name: [unclosed\n")
    assert broken[0].startswith("Line ") and "This isn't valid YAML" in broken[0]
    assert _said("just text") == ["Line 1: A template starts with its keys: name, version and blocks."]
    assert _said("name: X\nname: Y\n") == ["Line 2: `name` is given twice."]
    assert _said("name: X\n") == ["Line 1: A template needs `version`.", "Line 1: A template needs `blocks`."]
    assert "Anchors and aliases" in _said("a: &x 1\nb: *x\n")[0]
    assert "at most 64 KB" in _said("x" * (65 * 1024))[0]
    with pytest.raises(TemplateError):
        check("name: X\nversion: 1\nblocks: []\n")


def test_spans():
    assert [duration(v) for v in (90, "90s", "15m", "26h", "2d", "1.5h")] == [90, 90, 900, 93600, 172800, 5400]
    assert duration("soon") is None and duration(-1) is None and duration(True) is None


# ── The renderer ─────────────────────────────────────────────────────────────

def _view(series=SERIES, text=TEMPLATE, history=None):
    return render(check(text), T, latest(series), NOW, history=history)


def test_the_verdict_names_the_worst_and_each_block_leads_with_its_problems():
    view = _view()
    # Photos is late, its last backup is past 26h, and malware is held: three warnings.
    assert view.tone == "warn"
    assert view.verdict == "3 need a look. Backup targets: Photos is late."
    targets = view.blocks[0]
    assert [(r.name, r.tone, r.text) for r in targets.rows] == [
        ("Photos", "warn", "late"), ("Steward", "ok", "fresh"), ("Old", "dim", "disabled")]
    assert targets.rows[0].why == "is late" and targets.columns == ["Last backup"]
    assert targets.rows[1].cells[0]["moment"] == datetime.fromtimestamp(before(hours=16), timezone.utc)
    assert targets.tone == "warn"


def test_one_problem_is_the_verdict_itself():
    view = _view({**SERIES, ("", "fa_malware_held_files"): 0.0,
                  ("target=Photos,target_id=2", "fa_target_last_backup_timestamp_seconds"): before(hours=2)})
    assert view.verdict == "Backup targets: Photos is late."


def test_values_show_shares_bars_and_thresholds():
    storage = _view().blocks[1]
    rows = {r.name: r for r in storage.rows}
    assert rows["Disk free"].text == "16.2 TiB of 41.8 TiB" and round(rows["Disk free"].bar, 2) == 0.39
    assert rows["Disk free"].tone == "ok"
    assert rows["Files checked"].text == "93,310 of 94,116"
    assert (rows["Malware held"].tone, rows["Malware held"].why) == ("warn", "above 0")
    assert storage.rows[0].name == "Malware held"                       # problems first
    low = _view({**SERIES, ("", "fa_disk_free_bytes"): 2e12}).blocks[1]
    assert {r.name: (r.tone, r.why) for r in low.rows}["Disk free"] == ("warn", "below 10%")


def test_a_sparkline_is_asked_for_then_drawn():
    view = _view()
    assert view.wants == [(T, "fa_disk_free_bytes")]
    drawn = _view(history={(T, "fa_disk_free_bytes"): [1.0, 2.0]})
    assert {r.name: r.history for r in drawn.blocks[1].rows}["Disk free"] == [1.0, 2.0]


def test_times_read_past_and_future_and_flag_the_overdue():
    times = _view().blocks[2]
    rows = {(r.name, r.metric): r for r in times.rows}
    photos = rows[("Photos", "fa_target_last_backup_timestamp_seconds")]
    assert (photos.tone, photos.why) == ("warn", "over 1d 2h ago")
    nightly = rows[("Nightly", "fa_job_next_run_timestamp_seconds")]
    assert nightly.tone == "ok" and nightly.moment > NOW                # "in 8h"
    stale = _view({**SERIES, ("target=Photos,target_id=2", "fa_target_last_backup_timestamp_seconds"):
                   before(days=4)}).blocks[2]
    assert stale.rows[0].tone == "crit" and stale.rows[0].why == "over 3d ago"


def test_a_table_pivots_two_labels_and_info_shows_its_labels():
    table, info = _view().blocks[3:]
    assert table.columns == ["files", "trash"]
    assert [(r.name, [c["text"] for c in r.cells]) for r in table.rows] == [
        ("Ana", ["27.9 GiB", "—"]), ("Bryan", ["437.7 GiB", "1.1 GiB"])]
    assert [r.name for r in info.rows] == ["version 2026.10.12.1717 · commit 3605d0a"]


def test_missing_series_are_said_not_hidden():
    bare = {k: v for k, v in SERIES.items() if k[1] != "fa_malware_held_files" and k[1] != "fa_build_info"}
    view = _view(bare)
    malware = {r.name: r for r in view.blocks[1].rows}["Malware held"]
    assert (malware.text, malware.tone, malware.why) == ("—", "warn", "not reported")
    assert view.blocks[4].note == "Nothing reported for fa_build_info." and view.blocks[4].tone == "warn"


def test_a_template_for_another_target_says_so():
    view = _view({("", "node_load1"): 0.5})
    assert view.tone == "warn" and "reports no fa_… metrics" in view.verdict


def test_all_well():
    healthy = {**SERIES, **states("target=Photos,target_id=2", "fresh", TARGET_STATES),
               ("target=Photos,target_id=2", "fa_target_last_backup_timestamp_seconds"): before(hours=2),
               ("", "fa_malware_held_files"): 0.0}
    view = _view(healthy)
    assert (view.tone, view.verdict) == ("ok", "All well.")


def test_a_healthy_state_past_its_deadline_is_overdue():
    """warn_past / crit_past (#5693): a row whose state is healthy but whose
    deadline has passed reads as a warning or a problem; a quiet or already
    failing row keeps its own reading."""
    text = """\
name: Jobs
version: 1
blocks:
  - states:
      title: Jobs
      metric: j_state
      name: job
      healthy: [ok, running]
      quiet: [paused]
      warn_past: j_late_at_timestamp_seconds
      crit_past: j_missing_at_timestamp_seconds
"""
    series = {}
    for job, state, late, missing in (("A", "running", before(hours=2), before(hours=1) + 7200),
                                      ("B", "running", before(hours=5), before(hours=3)),
                                      ("C", "ok", before(hours=-1), before(hours=-2)),
                                      ("D", "paused", before(hours=9), before(hours=8))):
        for s in ("ok", "running", "paused"):
            series[(f"job={job},state={s}", "j_state")] = float(s == state)
        series[(f"job={job}", "j_late_at_timestamp_seconds")] = late
        series[(f"job={job}", "j_missing_at_timestamp_seconds")] = missing
    rows = {r.name: r for r in render(check(text), T, latest(series), NOW).blocks[0].rows}
    assert (rows["A"].tone, rows["A"].why) == ("warn", "is running, 2h overdue")
    assert (rows["B"].tone, rows["B"].why) == ("crit", "is running, 3h overdue")
    assert (rows["C"].tone, rows["C"].why) == ("ok", "")
    assert rows["D"].tone == "dim"