67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
|
"""
|
||
|
|
Tests for brain_watchdog.main.
|
||
|
|
|
||
|
|
Module-level env vars are required at import time; set them before the first
|
||
|
|
import of the module so tests can run without a real control-plane.
|
||
|
|
"""
|
||
|
|
import importlib.util
|
||
|
|
import os
|
||
|
|
import time
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
os.environ.setdefault("CONTROL_PLANE_URL", "http://test-cp:8080")
|
||
|
|
os.environ.setdefault("TG_TOKEN", "test_token")
|
||
|
|
os.environ.setdefault("TG_CHAT_ID", "12345")
|
||
|
|
|
||
|
|
import brain_watchdog.main as bwm
|
||
|
|
|
||
|
|
|
||
|
|
def test_package_importable():
|
||
|
|
spec = importlib.util.find_spec("brain_watchdog")
|
||
|
|
assert spec is not None
|
||
|
|
|
||
|
|
|
||
|
|
def test_check_ok_fresh():
|
||
|
|
now = time.time()
|
||
|
|
with patch.object(bwm, "http_get", return_value=(200, {"last_update": now - 10})):
|
||
|
|
ok, reason = bwm.check()
|
||
|
|
assert ok
|
||
|
|
assert "ok" in reason
|
||
|
|
|
||
|
|
|
||
|
|
def test_check_fail_stale():
|
||
|
|
now = time.time()
|
||
|
|
stale_ts = now - (bwm.STALE_THRESHOLD + 120)
|
||
|
|
with patch.object(bwm, "http_get", return_value=(200, {"last_update": stale_ts})):
|
||
|
|
ok, reason = bwm.check()
|
||
|
|
assert not ok
|
||
|
|
assert "stale" in reason
|
||
|
|
|
||
|
|
|
||
|
|
def test_check_fail_unreachable():
|
||
|
|
with patch.object(bwm, "http_get", return_value=(None, None)):
|
||
|
|
ok, reason = bwm.check()
|
||
|
|
assert not ok
|
||
|
|
assert "unreachable" in reason
|
||
|
|
|
||
|
|
|
||
|
|
def test_check_fail_http_error():
|
||
|
|
with patch.object(bwm, "http_get", return_value=(503, None)):
|
||
|
|
ok, reason = bwm.check()
|
||
|
|
assert not ok
|
||
|
|
assert "503" in reason
|
||
|
|
|
||
|
|
|
||
|
|
def test_check_fail_missing_last_update():
|
||
|
|
with patch.object(bwm, "http_get", return_value=(200, {"other": "data"})):
|
||
|
|
ok, reason = bwm.check()
|
||
|
|
assert not ok
|
||
|
|
assert "last_update" in reason
|
||
|
|
|
||
|
|
|
||
|
|
def test_check_fail_unparseable_timestamp():
|
||
|
|
with patch.object(bwm, "http_get", return_value=(200, {"last_update": "not-a-number"})):
|
||
|
|
ok, reason = bwm.check()
|
||
|
|
assert not ok
|
||
|
|
assert "parseable" in reason
|