thermograph/tests/core/test_dashboard.py

"""Tests for the ops dashboard's daemon health rendering (scripts/dashboard.py).

The single-leader subscription notifier must be judged by heartbeat freshness, not by
whether its thread happens to live on the worker the dashboard polled — otherwise ~2/3
of polls (the non-leader workers) would cry DOWN for a perfectly healthy notifier.
"""
import importlib.util
import os

REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
_spec = importlib.util.spec_from_file_location(
    "dashboard", os.path.join(REPO, "scripts", "dashboard.py"))
dashboard = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(dashboard)

C = dashboard.C(on=False)  # no ANSI colors, so we can assert on plain text


def _mark(name, tnames, heartbeats):
    return dashboard._daemon_mark(name, tnames, heartbeats, C)


def test_notifier_fresh_heartbeat_is_up_even_when_thread_absent():
    # The polled worker is a non-leader (no notifier thread), but a fresh shared heartbeat
    # proves the leader is running it.
    mark, detail = _mark("subscription-notifier", ["MainThread", "neighbor-warmer"],
                         {"subscription-notifier": {"age_s": 120.0, "interval_s": 900}})
    assert mark == "up"
    assert "2m ago" in detail


def test_notifier_stale_heartbeat_is_down():
    # Beat older than two intervals + slack => genuinely stuck/dead, flag it.
    mark, detail = _mark("subscription-notifier", ["MainThread"],
                         {"subscription-notifier": {"age_s": 4_000.0, "interval_s": 900}})
    assert mark == "DOWN"
    assert "stale" in detail


def test_notifier_missing_heartbeat_is_down():
    mark, detail = _mark("subscription-notifier", ["MainThread"], {})
    assert mark == "DOWN"
    assert "no heartbeat" in detail


def test_neighbor_warmer_still_uses_per_worker_thread_check():
    # The warmer runs in every worker, so thread presence is the right signal and a
    # heartbeat is neither written nor consulted.
    assert _mark("neighbor-warmer", ["MainThread", "neighbor-warmer"], {})[0] == "up"
    assert _mark("neighbor-warmer", ["MainThread"], {})[0] == "DOWN"


def test_short_dur_formats():
    assert dashboard._short_dur(45) == "45s"
    assert dashboard._short_dur(120) == "2m"
    assert dashboard._short_dur(3 * 3600 + 5 * 60) == "3h05m"