ca7f057e41
협의로 선정한 조기 신호 4종을 AlertManager 에 추가한다. - deadline: 최근 1h JobDeadlineExceeded 수 ≥ LPS_ALERT_DEADLINE_1H(5). 재시도로 살아나면 dead 룰엔 안 잡히는 크롤 행 반복 신호를 별도 집계. - cost: 최근 1h 완료 잡 검색원가 합 ≥ LPS_ALERT_COST_1H_USD(1.0). 비용의 87%가 프록시 대역폭 — 리소스차단 풀림·재시도 루프의 조용한 비용 폭주를 감시. job.result 의 metrics.cost.total_usd JSONB 합산. - proxy_ports_low: 가용 포트 비율 ≤ LPS_ALERT_PORTS_LOW_PCT(30%). 쿨다운 격리 누적 — blocks_1h(80건)보다 먼저 우는 대규모 차단 조기 신호. 워커별 프록시 중 가장 소진된 것 기준(min). - budget_leak: 최근 6h end_reason=block 세션 ≥ LPS_ALERT_BLOCK_SESSIONS_6H(1). 요청 예산(3회)을 지켰는데도 차단됨 = 예산 하향 검토 신호. - deadline_1h·cost_1h_usd 는 queue.ops() 에 편입 → /v1/lps/ops 로도 노출. 포트·세션 지표는 워커 웹훅 스냅샷에 포함(프록시 상태는 워커에만 있음). - 테스트 4건 추가(ops 집계 2·포트 스냅샷 2), 전체 145 passed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
"""AlertManager 테스트 — 발화·쿨다운(스팸 방지)·회복 알림 (sender/clock 주입, 네트워크·대기 없음)."""
|
||
|
||
from common.alerts import AlertManager
|
||
from common.database.db_session_manager import DB_SESSION_MNG
|
||
from services.search.naver.adapter import NaverAdapter
|
||
|
||
|
||
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 = NaverAdapter(keys=[("id", "sec")])
|
||
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 = NaverAdapter(keys=[("id", "sec")])
|
||
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
|
||
|
||
|
||
# ---- 가용 프록시 포트 현황(포트 고갈 룰의 데이터) --------------------------
|
||
|
||
def test_proxy_ports_snapshot_min_across_workers():
|
||
from config.config_models import DecodoConfig
|
||
from services.search.proxy import DecodoProxy
|
||
from worker_main import _proxy_ports_snapshot
|
||
|
||
def _proxy():
|
||
return DecodoProxy(DecodoConfig(host="h", username="u", password="p",
|
||
port_start=10001, port_end=10010, session_minutes=10))
|
||
|
||
class _Ad:
|
||
def __init__(self, proxy):
|
||
self._proxy = proxy
|
||
|
||
p1, p2 = _proxy(), _proxy()
|
||
p2.mark_burned(10001)
|
||
p2.mark_burned(10002)
|
||
avail, total = _proxy_ports_snapshot([_Ad(p1), _Ad(p2), object()]) # 프록시 없는 어댑터 혼재 OK
|
||
assert (avail, total) == (8, 10) # 가장 소진된 워커(p2) 기준
|
||
|
||
|
||
def test_proxy_ports_snapshot_none_without_proxy():
|
||
from worker_main import _proxy_ports_snapshot
|
||
assert _proxy_ports_snapshot([object()]) is None
|
||
assert _proxy_ports_snapshot(None) is None
|