o2o-negosium-original/lps/tests/test_browser_base.py

"""브라우저 어댑터 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, self.burned, self.kinds = 0, [], []
    # kind 는 회전 종류(budget=선제/block=차단/proxy_error/window) — 실물은 budget 일 때 휴식을 준다
    def rotate(self, kind: str = "rotate"): self.rotations += 1; self.kinds.append(kind)
    def mark_burned(self, port, cooldown_sec=None): self.burned.append(port)
    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
            self._current_port = self._proxy.current_port if self._proxy else None  # 실제 _ensure_browser 와 동일

    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(["<PRODUCT>ok</PRODUCT>"])
    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", "<PRODUCT>ok</PRODUCT>"])
    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"), "<PRODUCT>ok</PRODUCT>"])
    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회 + 소진 후 '다음 검색용' 회전 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 == 2
    assert ad._proxy.burned == [10001, 10001]   # 차단마다 해당 포트 쿨다운 격리


# ── 라이브 스모크(옵트인): 실제 사이트 셀렉터·안티봇 드리프트 감지 ──────
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_shop.adapter import NaverShopAdapter
    p = DecodoProxy()
    return [
        # 네이버는 오픈API 종료(2026-07-31)로 크롤 전환. headless 는 WTM 에 걸리므로 headful.
        ("naver", NaverShopAdapter(headless=False)),
        ("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()