21f7ef4d19
Move the flat backend/tests/*.py into domain subfolders so the suite mirrors the code's concerns: data/ climate, grading, scoring, grid, places, store web/ api, content, homepage, views notifications/ notify, digest, discord (+ dm/interactions/link) accounts/ api_accounts core/ metrics, singleton, dashboard conftest.py stays at the tests/ root, so its sys.path setup and shared fixtures still apply to every subfolder. The four tests that derive repo paths from __file__ get their depth bumped one level to match their new location. Pytest discovers the subfolders recursively; the CI command (python -m pytest backend/tests) is unchanged. Claude-Session: https://claude.ai/code/session_01XXxmNFy9cZ6Gh8Y9thZn62
56 lines
2.3 KiB
Python
56 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.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"
|