# agent-office/tests/test_node_monitor.py import json, pytest from app import node_monitor class FakeRedis: """worker heartbeat + queue llen + scan_iter 흉내.""" def __init__(self, kv=None, lists=None): self._kv = kv or {} # key(str) -> bytes self._lists = lists or {} # key(str) -> length(int) async def get(self, key): return self._kv.get(key) async def llen(self, key): return self._lists.get(key, 0) async def scan_iter(self, match=None): prefix = match.rstrip("*") for k in list(self._lists): if k.startswith(prefix): yield k def _hb(name, kind, state, **extra): return json.dumps({"name": name, "kind": kind, "state": state, "ts": "2026-06-29T00:00:00Z", "last_job_at": None, "jobs_done": 0, "jobs_failed": 0, **extra}).encode() @pytest.mark.asyncio async def test_alive_worker_healthy_link(): r = FakeRedis(kv={"worker:image-render:heartbeat": _hb("image-render","render","idle")}) st = await node_monitor.collect_status(redis=r) img = next(w for w in st["workers"] if w["name"] == "image-render") assert img["alive"] is True and img["state"] == "idle" link = next(l for l in st["links"] if l["to"] == "image-render") assert link["status"] == "healthy" and link["type"] == "redis-queue" @pytest.mark.asyncio async def test_missing_heartbeat_is_dead_and_down(): r = FakeRedis() # heartbeat 없음 st = await node_monitor.collect_status(redis=r) img = next(w for w in st["workers"] if w["name"] == "image-render") assert img["alive"] is False link = next(l for l in st["links"] if l["to"] == "image-render") assert link["status"] == "down" @pytest.mark.asyncio async def test_dead_letter_makes_degraded(): r = FakeRedis(kv={"worker:video-render:heartbeat": _hb("video-render","render","idle")}, lists={"dead_letter:queue:video-render": 2}) st = await node_monitor.collect_status(redis=r) vid = next(w for w in st["workers"] if w["name"] == "video-render") assert vid["dead_letter"] == 2 link = next(l for l in st["links"] if l["to"] == "video-render") assert link["status"] == "degraded" @pytest.mark.asyncio async def test_paused_reason_from_watcher(): r = FakeRedis(kv={"queue:paused": b"1", "worker:task-watcher:heartbeat": _hb("task-watcher","watcher","trading",mode="trading")}) st = await node_monitor.collect_status(redis=r) assert st["paused"] is True and st["paused_reason"] == "trading" @pytest.mark.asyncio async def test_trader_http_pull_link(): r = FakeRedis(kv={"worker:ai_trade:heartbeat": _hb("ai_trade","trader","market_open")}) st = await node_monitor.collect_status(redis=r) link = next(l for l in st["links"] if l["from"] == "ai_trade") assert link["type"] == "http-pull" and link["status"] == "healthy" @pytest.mark.asyncio async def test_paused_no_watcher_heartbeat_fallback_reason(): """paused=True인데 watcher heartbeat 없으면 paused_reason == 'trading' 폴백.""" r = FakeRedis(kv={"queue:paused": b"1"}) # watcher heartbeat 없음 st = await node_monitor.collect_status(redis=r) assert st["paused"] is True assert st["paused_reason"] == "trading" @pytest.mark.asyncio async def test_processing_count_image_render(): """processing:: 리스트가 있으면 processing 필드에 합산된다.""" worker_id = "abc123" proc_key = f"processing:queue:image-render:{worker_id}" r = FakeRedis( kv={"worker:image-render:heartbeat": _hb("image-render", "render", "busy")}, lists={proc_key: 3}, ) st = await node_monitor.collect_status(redis=r) img = next(w for w in st["workers"] if w["name"] == "image-render") assert img["processing"] == 3 @pytest.mark.asyncio async def test_llen_exception_returns_redis_ok_false(): """워커 루프 중 llen 예외 발생 시 예외를 전파하지 않고 redis_ok=False 반환 (Blocker 회귀).""" class BrokenLlenRedis(FakeRedis): async def llen(self, key): raise ConnectionError("Redis 연결 끊김") r = BrokenLlenRedis( kv={"worker:music-render:heartbeat": _hb("music-render", "render", "idle")} ) st = await node_monitor.collect_status(redis=r) assert st["redis_ok"] is False import app.node_monitor as nm @pytest.mark.asyncio async def test_alert_on_alive_to_dead(monkeypatch): sent = [] async def fake_send_raw(text, **kw): sent.append(text); return {"ok": True} monkeypatch.setattr("app.telegram.messaging.send_raw", fake_send_raw) monkeypatch.setattr("app.db.add_log", lambda *a, **k: None) nm._node_state.clear(); nm._dl_notified.clear() alive = {"workers": [{"name":"image-render","alive":True,"dead_letter":0}], "links": []} dead = {"workers": [{"name":"image-render","alive":False,"dead_letter":0}], "links": []} await nm.check_and_alert(status=alive) # 첫 관측 — 경보 없음 assert sent == [] await nm.check_and_alert(status=dead) # alive→dead 전이 assert any("다운" in t for t in sent) @pytest.mark.asyncio async def test_alert_on_dead_letter_growth(monkeypatch): sent = [] async def fake_send_raw(text, **kw): sent.append(text); return {"ok": True} monkeypatch.setattr("app.telegram.messaging.send_raw", fake_send_raw) monkeypatch.setattr("app.db.add_log", lambda *a, **k: None) nm._node_state.clear(); nm._dl_notified.clear() s = {"workers": [{"name":"video-render","alive":True,"dead_letter":2}], "links": []} await nm.check_and_alert(status=s) assert any("dead-letter" in t for t in sent)