From eb1ff8fee14b9bda3fee8b20747111e4eaaca8e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 9 Jul 2026 23:25:42 +0900 Subject: [PATCH] =?UTF-8?q?test(lps):=20P5=20=EB=B8=8C=EB=9D=BC=EC=9A=B0?= =?UTF-8?q?=EC=A0=80=20=EA=B2=BD=EB=A1=9C=20=EC=8A=A4=EB=AA=A8=ED=81=AC=20?= =?UTF-8?q?=E2=80=94=20search=20=EC=98=A4=EC=BC=80=EC=8A=A4=ED=8A=B8?= =?UTF-8?q?=EB=A0=88=EC=9D=B4=EC=85=98=20+=20=EB=9D=BC=EC=9D=B4=EB=B8=8C?= =?UTF-8?q?=20=EB=93=9C=EB=A6=AC=ED=94=84=ED=8A=B8=20=EA=B0=90=EC=A7=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 가장 취약·복잡한 경로(회전·재시도·차단감지)의 자동 테스트 공백을 메운다. - mock 하니스(_MockPage/_MockCtx/_MockProxy)로 search() 를 결정론 테스트(브라우저 없이): 정상 반환 / 차단→IP회전→복구 / 프록시전송오류→회전→복구 / 비프록시오류→실패(무회전) / 차단 소진→blocked 실패. 5종. - 라이브 스모크(각 어댑터 실제 사이트 검색→파싱): 셀렉터·안티봇 드리프트 감지. IP 의존·느려서 기본 skip, LPS_LIVE=1 로 명시 실행(수동/야간). 네이버 스모크 통과 확인. 전체 96 passed, 5 skipped. Co-Authored-By: Claude Fable 5 --- lps/tests/test_browser_base.py | 146 +++++++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) create mode 100644 lps/tests/test_browser_base.py diff --git a/lps/tests/test_browser_base.py b/lps/tests/test_browser_base.py new file mode 100644 index 0000000..f370c96 --- /dev/null +++ b/lps/tests/test_browser_base.py @@ -0,0 +1,146 @@ +"""브라우저 어댑터 search() 오케스트레이션 테스트. + +가장 취약·복잡한 경로(차단 감지→IP 회전 재시도, 프록시 전송오류→회전, 소진→실패)를 +mock 페이지로 **결정론적**으로 검증한다(실제 브라우저/네트워크 없이). 파서 자체는 +test_coupang_parser / test_openmarket_parser 가 저장 HTML 로 커버. + ++ 라이브 스모크: 실제 사이트에 붙어 각 어댑터가 결과를 파싱하는지(셀렉터·안티봇 드리프트 감지). + 느리고 IP 의존적이라 기본 skip — LPS_LIVE=1 로 명시 실행(수동/야간용). +""" + +import os + +import pytest + +from services.search.browser_base import BrowserSearchAdapter +from services.search.contract import NormalizedProduct, AdapterError +from services.search.rate_limiter import RateLimiter + + +# ── mock 하니스: page.goto/content 를 스크립트로 제어 ────────────────── +class _MockPage: + """steps: 시도별 결과. Exception→goto 가 raise, str→content 가 그 HTML 반환.""" + def __init__(self, steps): + self._steps, self._i, self._cur = steps, 0, None + + async def goto(self, url, **kw): + self._cur = self._steps[self._i] + self._i += 1 + if isinstance(self._cur, BaseException): + raise self._cur + + async def content(self): + return self._cur + + async def wait_for_selector(self, *a, **k): pass + async def query_selector(self, *a, **k): return None + async def wait_for_timeout(self, *a, **k): pass + async def evaluate(self, *a, **k): return None + + +class _MockCtx: + def __init__(self, page): self.pages = [page] + async def new_page(self): return self.pages[0] + async def cookies(self, *a): return [] + async def new_cdp_session(self, *a): raise RuntimeError("no cdp") # → DOM 바이트 폴백 + async def close(self): pass + + +class _MockProxy: + enabled = True + session_minutes = 10 + def __init__(self): self.rotations = 0 + def rotate(self): self.rotations += 1 + def playwright_proxy(self): return None + @property + def current_port(self): return 10001 + + +class _MockAdapter(BrowserSearchAdapter): + source = "mock" + block_markers = ("BOTBLOCK",) + min_result_html = 50 # 이보다 짧은 0건 응답 = short_html 차단 + + def __init__(self, page, **kw): + self._page = page + super().__init__(rate_limiter=RateLimiter(0, 0), **kw) + + async def _ensure_browser(self): + if self._ctx is None or self._force_recycle: # 회전(force_recycle) 시 재기동 흉내 + self._ctx = _MockCtx(self._page) + self._force_recycle = False + self._ip_requests = 0 + + async def _ensure_net_meter(self, page): pass # CDP 없음 → last_bytes=DOM 크기 + async def _wait_ready(self, page): pass + async def _blocking_now(self): return False + def _search_url(self, q, l): return "http://mock" + def _parse(self, html): + return [NormalizedProduct(source="mock", name="p", price=100)] if "PRODUCT" in html else [] + + +def _ad(steps): + return _MockAdapter(_MockPage(steps), proxy=_MockProxy()) + + +async def test_search_returns_products_no_rotation(): + ad = _ad(["ok"]) + r = await ad.search("q", limit=5) + assert len(r) == 1 and ad._proxy.rotations == 0 + + +async def test_block_rotates_ip_then_recovers(): + # 시도1: 짧은 HTML(차단) → IP 회전 / 시도2: 상품 + ad = _ad(["x", "ok"]) + r = await ad.search("q") + assert len(r) == 1 and ad._proxy.rotations == 1 + + +async def test_proxy_transport_error_rotates_then_recovers(): + ad = _ad([RuntimeError("Page.goto: net::ERR_TUNNEL_CONNECTION_FAILED"), "ok"]) + r = await ad.search("q") + assert len(r) == 1 and ad._proxy.rotations == 1 + + +async def test_non_proxy_goto_error_raises_no_rotation(): + ad = _ad([RuntimeError("net::ERR_NAME_NOT_RESOLVED")]) # DNS — 프록시 문제 아님 + with pytest.raises(AdapterError): + await ad.search("q") + assert ad._proxy.rotations == 0 + + +async def test_persistent_block_exhausts_and_raises_blocked(): + # 차단 2회 연속(max_block_retries=1) → 회전 1회 후 소진 → blocked=True 로 실패 + ad = _ad(["x", "x"]) + with pytest.raises(AdapterError) as ei: + await ad.search("q") + assert ei.value.blocked is True and ad._proxy.rotations == 1 + + +# ── 라이브 스모크(옵트인): 실제 사이트 셀렉터·안티봇 드리프트 감지 ────── +def _live_adapters(): + from services.search.proxy import DecodoProxy + from services.search.coupang.adapter import CoupangAdapter + from services.search.esm.adapter import EsmAdapter + from services.search.st11.adapter import ElevenStAdapter + from services.search.naver.adapter import NaverAdapter + p = DecodoProxy() + return [ + ("naver", NaverAdapter()), + ("coupang", CoupangAdapter(headless=True, proxy=p)), + ("gmarket", EsmAdapter("gmarket", headless=True, proxy=p)), + ("auction", EsmAdapter("auction", headless=True, proxy=p)), + ("st11", ElevenStAdapter(headless=True, proxy=p)), + ] + + +@pytest.mark.skipif(not os.environ.get("LPS_LIVE"), reason="라이브 스모크 — LPS_LIVE=1 로 실행") +@pytest.mark.parametrize("name", ["naver", "coupang", "gmarket", "auction", "st11"]) +async def test_live_smoke(name): + ad = dict(_live_adapters())[name] + try: + ps = await ad.search("생수", limit=5) + assert len(ps) > 0, f"{name}: 0건 — 셀렉터/안티봇 드리프트 의심" + assert all(p.price > 0 and p.name for p in ps) + finally: + await ad.close()