57 lines
2.3 KiB
Python
57 lines
2.3 KiB
Python
|
"""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.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"
|