b7fc327779
**낡은 테스트**: test_handler_skips_record_on_negative_cache_hit 는 '캐시 히트면 이력을 남기지 않는다'를 검증했는데, 그 동작은 실측 버그였다 — 잡은 DONE 인데 price_history 에 새 행이 없어 이를 폴링하는 소비자(negodata 최저가 모달)가 결과를 영영 못 받고 로딩만 돌았다. 코드는 이미 '캐시 히트도 이 잡의 결과이므로 기록한다'로 고쳐져 있었고 테스트만 남아 있었다. → 현재 계약(not_found 스냅샷 1건 기록, 가격은 null)을 검증하도록 다시 씀. 전체 220 passed·0 failed. **오픈API 어댑터 제거**: shop.json 이 2026-07-31 종료돼 404 SE05 만 반환하고, 파이프라인은 naver_shop(크롤)로 옮겨 갔다. 되살릴 수 없는 코드를 남겨두면 다음 사람이 "키를 넣으면 되나" 하고 시간을 쓴다. - services/search/naver/ (adapter·transform) 삭제 - NaverConfig 모델·로더·설정 섹션 3개 파일에서 제거(죽은 키) - test_naver_transform 삭제, test_alerts 는 NaverAdapter 대신 스텁 사용 (검증 대상인 recent_stats/_note_result 는 베이스 SearchAdapter 계약이라 무관) **문서 정합화**: architecture(네이버 안티봇=WTM, 통과 3조건) · api(배송비가 이제 채워짐, 가격은 즉시판매가·쿠폰가 제외) · operations(kr_host·naver_ip_request_budget) · README 트리. source 이름 "naver" 는 그대로다 — price_history·by_mall·프론트 계약은 구현 교체와 무관하다.
152 lines
5.2 KiB
Python
152 lines
5.2 KiB
Python
"""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)
|