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 time

import pytest

from services.search.browser_base import BrowserSearchAdapter, ENV_BLOCK_MARKER
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):
        # 회전하면 실제로 포트가 바뀐다 — 고정값이면 '같은 IP 를 계속 쓰는' 버그를 테스트가 못 본다.
        return 10001 + self.rotations


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):
        # 재기동 판단(_recycle_if_due)·IP 세션 경계(_begin_ip_session)는 **실물 코드를 그대로 탄다**
        # — 브라우저 기동만 흉내낸다. 그래야 이 하니스가 진짜 동작을 검증한다.
        if not await self._recycle_if_due():
            return
        self._ctx = _MockCtx(self._page)
        await self._begin_ip_session(self._proxy.current_port if self._proxy else None)
        self._launched_at = time.monotonic()
        self._force_recycle = False

    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(["<html>BOTBLOCK</html>", "<html>BOTBLOCK</html>"])
    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, 10002]   # 차단마다 **그때 쓰던** 포트를 쿨다운 격리


async def test_unknown_short_html_rotates_but_does_not_burn():
    """'0건인데 페이지가 짧다'는 정황일 뿐 — 진짜 검색결과 없음일 수 있다.
    확신 없이 태우면 정상 응답에 IP 를 30분 묶는다. 회전·재시도까지만 한다."""
    ad = _ad(["x", "x"])
    with pytest.raises(AdapterError) as ei:
        await ad.search("q")
    assert ei.value.blocked is True, "회복 시도는 해야 하므로 blocked 로 알린다"
    assert ad._proxy.rotations == 2, "IP 는 바꿔 재시도한다(비용이 짧고 되돌릴 수 있다)"
    assert ad._proxy.burned == [], "확신 없는 판정으로 30분 쿨다운을 걸지 않는다"


async def test_unknown_short_html_does_not_trip_the_env_breaker():
    """무결과 질의가 연달아 나와도 '환경 차단'으로 오판해 소스를 멈추면 안 된다."""
    ad = _MockAdapter(_MockPage(["x"] * 8), proxy=_MockProxy(), max_block_retries=1)
    for _ in range(3):
        with pytest.raises(AdapterError):
            await ad.search("q")
    assert ad._env_blocked is False and ad._fresh_ip_blocks == []


# ── IP 세션 ≠ 브라우저 수명 (2026-08-05 회귀) ────────────────────────────
# 유휴 정리(close_if_idle)는 브라우저만 닫고 IP 임대는 그대로 둔다. 예전엔 재기동 때마다
# 요청 카운터를 0 으로 되돌려, 검색이 유휴 임계보다 뜸하면 **한 IP 를 무한히 쓰면서 예산이
# 영영 발화하지 않았다**(실측: 6회 검색이 전부 같은 포트·ip_req#1). ip_request_no 도 항상 1 로
# 찍혀 '몇 번째 요청에서 막혔나' 진단까지 무너졌다.

async def _idle_cleanup(ad):
    """run_browser_reaper 가 하는 일 — 유휴 임계를 넘겼다고 보고 브라우저만 닫는다."""
    ad._last_used -= 10_000
    await ad.close_if_idle(120)


async def test_idle_cleanup_keeps_the_ip_session_counter():
    ad = _MockAdapter(_MockPage(["<PRODUCT>ok</PRODUCT>"] * 3), proxy=_MockProxy())
    seen = []
    for _ in range(3):
        await ad.search("q")
        seen.append((ad._current_port, ad._ip_requests))
        await _idle_cleanup(ad)
    assert seen == [(10001, 1), (10001, 2), (10001, 3)], \
        "같은 IP 로 돌아왔으면 요청 수가 이어져야 한다(브라우저 재기동은 IP 교체가 아니다)"


async def test_budget_still_fires_when_idle_cleanup_happens_between_searches():
    ad = _MockAdapter(_MockPage(["<PRODUCT>ok</PRODUCT>"] * 4), proxy=_MockProxy(), ip_request_budget=3)
    for _ in range(4):
        await ad.search("q")
        await _idle_cleanup(ad)
    assert ad._proxy.rotations == 1, "예산 3회를 채웠으면 4번째 검색 전에 선제 회전해야 한다"
    assert ad._proxy.kinds == ["budget"]
    assert (ad._current_port, ad._ip_requests) == (10002, 1), "회전 뒤에는 새 IP 로 카운터가 리셋된다"


async def test_session_is_recorded_once_per_ip_not_per_browser_restart():
    events = []

    async def on_end(e):
        events.append(e)

    ad = _MockAdapter(_MockPage(["<PRODUCT>ok</PRODUCT>"] * 4), proxy=_MockProxy(),
                      ip_request_budget=3, on_session_end=on_end)
    for _ in range(4):
        await ad.search("q")
        await _idle_cleanup(ad)
    assert len(events) == 1, "유휴 정리는 IP 세션을 끝내지 않는다 — 포트가 바뀔 때만 1건"
    assert events[0]["requests"] == 3 and events[0]["end_reason"] == "budget"

    await ad.close()   # 종료 시 진행 중이던 IP 세션도 마감된다
    assert [e["end_reason"] for e in events] == ["budget", "shutdown"]

    await ad.close()   # 종료 경로가 겹쳐 두 번 불려도 같은 세션을 또 기록하지 않는다
    assert [e["end_reason"] for e in events] == ["budget", "shutdown"]


async def test_sticky_window_expiry_actually_rotates_the_ip():
    """시간창 만료 = 제공자 sticky 세션도 끝났다 → 같은 포트를 계속 붙잡으면 안 된다.
    예전엔 브라우저만 재기동하고 임대를 renew 해 IP 가 그대로였다(로그만 'IP 회전')."""
    ad = _MockAdapter(_MockPage(["<PRODUCT>ok</PRODUCT>"] * 2), proxy=_MockProxy())
    await ad.search("q")
    assert (ad._current_port, ad._proxy.rotations) == (10001, 0)

    ad._launched_at -= ad._proxy.session_minutes * 60 + 1      # sticky 수명 경과
    await ad.search("q")
    assert ad._proxy.kinds == ["window"], "시간창 만료는 window 사유로 회전한다"
    assert (ad._current_port, ad._ip_requests) == (10002, 1), "새 IP 로 바뀌고 카운터도 새로 센다"


async def test_forced_rotation_does_not_double_rotate_on_window():
    """이미 회전이 예약된 상태(예산·차단)에서 시간창까지 걸려도 회전은 1번이어야 한다."""
    ad = _MockAdapter(_MockPage(["<PRODUCT>ok</PRODUCT>"] * 2), proxy=_MockProxy(), ip_request_budget=1)
    await ad.search("q")                     # 예산 1회 → 다음 검색 전에 선제 회전 예약
    ad._launched_at -= ad._proxy.session_minutes * 60 + 1
    await ad.search("q")
    assert ad._proxy.kinds == ["budget"], "예약된 회전이 우선 — window 로 덧회전하지 않는다"
    assert ad._proxy.rotations == 1


# ── 환경 차단 서킷브레이커 (2026-08-05) ─────────────────────────────────
# 서로 다른 IP 가 연속으로 첫 요청부터 막히면 IP 문제가 아니다 → 태우기를 멈춘다.
# 없을 때는 잡 1건이 포트 6개를 30분 쿨다운에 묶어, 16건이면 풀 100개가 고갈됐다(실측).

async def test_env_block_trips_and_stops_burning_ports():
    ad = _MockAdapter(_MockPage(["<html>BOTBLOCK</html>"] * 8), proxy=_MockProxy(), max_block_retries=1)
    # 검색 1: 서로 다른 IP 2개가 첫 요청부터 차단(임계 3 미달) → 기존대로 태우고 포기
    with pytest.raises(AdapterError):
        await ad.search("q")
    assert ad._proxy.burned == [10001, 10002] and ad._env_blocked is False

    # 검색 2: 3번째 IP 도 첫 요청부터 차단 → 트립. 이 포트는 태우지 않는다
    with pytest.raises(AdapterError) as ei:
        await ad.search("q")
    assert ad._env_blocked is True
    assert ei.value.fatal is True, "회전으로 회복 불가 → 호출부가 설정을 의심하도록 fatal"
    assert ad._proxy.burned == [10001, 10002], "트립한 IP 는 태우지 않는다"

    # 검색 3 이후: 계속 막혀도 포트를 더 태우지 않는다
    with pytest.raises(AdapterError):
        await ad.search("q")
    assert ad._proxy.burned == [10001, 10002]


async def test_env_block_records_marker_every_time_it_blocks():
    """알림은 '최근 1h fatal 마커 수'로 판정한다 — 트립 1회만 기록하면 문제가 그대로인데
    1시간 뒤 '해소' 알림이 나간다. 그래서 막힌 검색마다 남긴다."""
    events = []

    async def on_detect(ev):
        events.append(ev)

    ad = _MockAdapter(_MockPage(["<html>BOTBLOCK</html>"] * 6), proxy=_MockProxy(),
                      max_block_retries=1, on_detect=on_detect)
    for _ in range(3):                          # 검색1=미트립, 검색2=트립, 검색3=트립 유지
        with pytest.raises(AdapterError):
            await ad.search("q")
    assert [e["marker"] for e in events].count(ENV_BLOCK_MARKER) == 2


async def test_repeated_block_on_one_ip_is_not_an_env_block():
    """같은 IP 가 반복해 막히는 건 그냥 나쁜 IP 다 — 환경 차단으로 오판하면 안 된다."""
    class _StickyProxy(_MockProxy):
        @property
        def current_port(self): return 10001        # 회전해도 같은 포트(=같은 IP)

    ad = _MockAdapter(_MockPage(["<html>BOTBLOCK</html>"] * 8), proxy=_StickyProxy(), max_block_retries=1)
    for _ in range(3):
        with pytest.raises(AdapterError):
            await ad.search("q")
    assert ad._env_blocked is False and ad._fresh_ip_blocks == [10001]


async def test_block_after_several_requests_is_not_an_env_block():
    """뒤쪽 요청(ip_req#2+)에서 막힌 건 '이 IP 를 많이 썼다'는 뜻 — 예산 문제지 환경 문제가 아니다."""
    ad = _MockAdapter(_MockPage(["<PRODUCT>ok</PRODUCT>", "<html>BOTBLOCK</html>", "<html>BOTBLOCK</html>"]),
                      proxy=_MockProxy(), max_block_retries=0)
    await ad.search("q")                       # ip_req#1 성공
    with pytest.raises(AdapterError):
        await ad.search("q")                   # ip_req#2 에서 차단
    assert ad._env_blocked is False and ad._fresh_ip_blocks == []


async def test_success_resets_the_breaker():
    # 검색1 = 차단 2회(재시도 포함), 검색2 = 차단 1회로 트립 → 검색3 에서 회복
    ad = _MockAdapter(_MockPage(["<html>BOTBLOCK</html>"] * 3 + ["<PRODUCT>ok</PRODUCT>"]),
                      proxy=_MockProxy(), max_block_retries=1)
    for _ in range(2):
        with pytest.raises(AdapterError):
            await ad.search("q")
    assert ad._env_blocked is True
    await ad.search("q")                       # 환경 회복
    assert ad._env_blocked is False and ad._fresh_ip_blocks == []


# ── 라이브 스모크(옵트인): 실제 사이트 셀렉터·안티봇 드리프트 감지 ──────
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()


# ── 구조적 차단(회전 무효) ───────────────────────────────────────────────
class _FatalAdapter(_MockAdapter):
    """해외 IP 하드차단처럼 '회전해도 같은 결과'인 마커를 가진 어댑터."""
    block_markers = ("BOTBLOCK", "HARDBLOCK")
    fatal_block_markers = ("HARDBLOCK",)


async def test_fatal_block_does_not_rotate_or_burn():
    """구조적 차단은 IP 문제가 아니다 — 태우면 원인도 못 고친 채 풀만 말린다
    (같은 게이트웨이를 쓰는 다른 소스의 IP 까지 쿨다운에 묶인다)."""
    page = _MockPage(["<html>HARDBLOCK</html>"])
    proxy = _MockProxy()
    ad = _FatalAdapter(page, proxy=proxy, max_block_retries=3)
    with pytest.raises(AdapterError) as e:
        await ad.search("q")
    assert e.value.fatal is True and e.value.blocked is True
    assert proxy.burned == [], "구조적 차단은 포트를 태우지 않는다"
    assert proxy.rotations == 0, "회전해도 소용없으므로 회전하지 않는다"


async def test_fatal_block_is_recorded_for_alerting():
    """감지 기록은 남겨야 한다 — 알림이 이 마커를 세서 '설정 고치라'고 알린다."""
    events = []

    async def on_detect(ev):
        events.append(ev)

    page = _MockPage(["<html>HARDBLOCK</html>"])
    ad = _FatalAdapter(page, proxy=_MockProxy(), on_detect=on_detect)
    with pytest.raises(AdapterError):
        await ad.search("q")
    assert len(events) == 1 and events[0]["marker"] == "HARDBLOCK"


async def test_ordinary_block_still_rotates_and_burns():
    """일반 차단(IP 평판)은 기존대로 태우고 회전해 회복을 시도한다 — 두 경로가 갈렸는지 확인."""
    page = _MockPage(["<html>BOTBLOCK</html>", "<PRODUCT>ok</PRODUCT>"])
    proxy = _MockProxy()
    ad = _FatalAdapter(page, proxy=proxy, max_block_retries=1)
    ps = await ad.search("q")
    assert len(ps) > 0
    assert proxy.burned == [10001] and proxy.rotations == 1