"""AlertManager 테스트 — 발화·쿨다운(스팸 방지)·회복 알림 (sender/clock 주입, 네트워크·대기 없음).""" from common.alerts import AlertManager from common.database.db_session_manager import DB_SESSION_MNG from services.search.contract import NormalizedProduct, SearchAdapter class _StubAdapter(SearchAdapter): """윈도우 카운터(_note_result/recent_stats)는 베이스 계약이라 아무 어댑터로도 검증된다.""" source = "stub" async def search(self, query, limit=40) -> list[NormalizedProduct]: return [] class _Clock: def __init__(self): self.t = 1000.0 def __call__(self): return self.t def _mgr(cooldown=1800): sent = [] async def sender(text): sent.append(text) clock = _Clock() return AlertManager(origin="test", cooldown_sec=cooldown, sender=sender, clock=clock), sent, clock async def test_fires_once_on_activation(): mgr, sent, _ = _mgr() await mgr.check("dead", True, "DEAD 1h=25") await mgr.check("dead", True, "DEAD 1h=26") # 조건 지속 — 쿨다운 안이라 재발송 없음 assert len(sent) == 1 and "DEAD 1h=25" in sent[0] assert mgr.is_active("dead") async def test_refires_after_cooldown(): mgr, sent, clock = _mgr(cooldown=1800) await mgr.check("dead", True, "m1") clock.t += 1801 # 쿨다운 경과 — 리마인드 1회 await mgr.check("dead", True, "m2") assert len(sent) == 2 and "m2" in sent[1] async def test_recovery_notice_once(): mgr, sent, _ = _mgr() await mgr.check("blocks", True, "차단 1h=90") await mgr.check("blocks", False, "차단 1h=3") # 해소 알림 1회 await mgr.check("blocks", False, "차단 1h=0") # 이미 비활성 — 무발송 assert len(sent) == 2 assert "해소" in sent[1] assert not mgr.is_active("blocks") async def test_inactive_rule_never_fires(): mgr, sent, _ = _mgr() await mgr.check("queue_lag", False, "큐지연=3s") assert sent == [] async def test_rules_are_independent(): mgr, sent, _ = _mgr() await mgr.check("dead", True, "a") await mgr.check("db_pool", True, "b") # 다른 키 — 각자 발화 assert len(sent) == 2 async def test_sender_failure_does_not_raise(): async def boom(text): raise RuntimeError("webhook down") mgr = AlertManager(origin="test", cooldown_sec=10, sender=boom) try: await mgr.check("dead", True, "m") except RuntimeError: # sender 주입 시 예외는 호출부(모니터 루프의 try)가 처리 — 여기서는 전파돼도 루프가 삼킨다 pass # ---- 윈도우 성공/실패 카운터(소스별 장기 실패 룰의 데이터) ---------------- def test_recent_stats_counts_within_window(): ad = _StubAdapter() ad._note_result(True) ad._note_result(False) ad._note_result(False) assert ad.recent_stats(1800) == (3, 1) assert ad.recent_stats(0) == (0, 0) # 창 밖이면 미집계 def test_recent_stats_empty_adapter(): ad = _StubAdapter() assert ad.recent_stats(1800) == (0, 0) # lazy init — 기록 전에도 안전 # ---- DB 풀 사용 현황 ------------------------------------------------------ def test_pool_status_shape(): st = DB_SESSION_MNG.pool_status() assert set(st) == {"checked_out", "capacity", "pct"} assert st["capacity"] > 0 # R/W 2엔진 × (pool+overflow) assert 0 <= st["pct"] <= 100 # ---- 가용 프록시 포트 현황(포트 고갈 룰의 데이터) -------------------------- async def test_port_pool_status_picks_worst_gateway(): """게이트웨이가 둘이라 **가장 마른 쪽** 기준으로 알림을 걸어야 한다 — 한쪽만 고갈돼도 그 소스(네이버/쿠팡)는 검색을 못 하므로 평균으로 덮으면 안 된다.""" from worker_main import _port_pool_status class _Store: async def snapshot(self, host=None): return {"gate.decodo.com": {"held": 2, "resting": 0, "cooling": 0, "available": 8, "total": 10}, "kr.decodo.com": {"held": 1, "resting": 2, "cooling": 4, "available": 3, "total": 10}} class _Proxy: _store = _Store() class _Ad: _proxy = _Proxy() detail, ports = await _port_pool_status([object(), _Ad()]) # 프록시 없는 어댑터 혼재 OK assert ports == (3, 10) # kr(=3) 기준 assert detail["kr.decodo.com"]["cooling"] == 4 # 원인 파악용 상세도 함께 실린다 async def test_port_pool_status_none_without_store(): from worker_main import _port_pool_status assert await _port_pool_status([object()]) == ({}, None) assert await _port_pool_status(None) == ({}, None) async def test_port_pool_status_survives_db_error(): """장부 조회가 실패해도 ops 모니터는 계속 돌아야 한다(알림만 건너뛴다).""" from worker_main import _port_pool_status class _Broken: async def snapshot(self, host=None): raise RuntimeError("db down") class _Ad: class _proxy: _store = _Broken() assert await _port_pool_status([_Ad()]) == ({}, None) # ---- 크롤 프리플라이트(웜업) 알림 ------------------------------------------ class _WarmAdapter(SearchAdapter): """웜업 검증용 — 항상 실패하거나 항상 성공하는 어댑터.""" def __init__(self, source, fail): self.source, self._fail = source, fail self.calls = 0 async def search(self, query, limit=40): self.calls += 1 if self._fail: raise RuntimeError("blocked") return [NormalizedProduct(source=self.source, name="p", price=1)] def _rotate_ip(self, *a, **k): pass async def test_warmup_alerts_when_source_cannot_crawl(): """기동 직후 크롤이 막혔으면 잡이 DEAD 로 쌓일 때까지 기다리지 않고 바로 알린다. (컨테이너에서만 막히는 사례 — 프로세스는 healthy 라 조용히 0건이 된다)""" from worker_main import _warmup_worker mgr, sent, _ = _mgr() ad = _WarmAdapter("naver", fail=True) await _warmup_worker([ad], tries=2, attempt_timeout=5, alerts=mgr) assert ad.calls == 2 # 재시도까지 소진 assert len(sent) == 1 and "naver" in sent[0] assert mgr.is_active("warmup_naver") async def test_warmup_success_clears_the_alert(): from worker_main import _warmup_worker mgr, sent, _ = _mgr() await _warmup_worker([_WarmAdapter("coupang", fail=True)], tries=1, attempt_timeout=5, alerts=mgr) assert mgr.is_active("warmup_coupang") await _warmup_worker([_WarmAdapter("coupang", fail=False)], tries=1, attempt_timeout=5, alerts=mgr) assert not mgr.is_active("warmup_coupang") # 회복 알림 후 해소 assert "해소" in sent[-1] async def test_warmup_skips_non_crawl_sources(): from worker_main import _warmup_worker mgr, sent, _ = _mgr() ad = _WarmAdapter("someapi", fail=True) # 브라우저 소스가 아님 → 웜업 대상 아님 await _warmup_worker([ad], tries=2, attempt_timeout=5, alerts=mgr) assert ad.calls == 0 and sent == []