o2o-negosium-original/lps/tests/test_openmarket_parser.py
민헌 8a594b667d perf(lps): Step2/3 — Turnstile 웜업·동적 차단 + 유휴 브라우저 정리
G마켓/옥션 Cloudflare Turnstile 을 실제로 통과시키고, 쌓이는 브라우저를 정리한다.

Turnstile 규명(실측): patchright 가 콜드 ~12초에 자동 통과, 통과 후 cf_clearance 쿠키로
이후 요청은 ~5초(웜). 단 **리소스를 조금이라도 차단하면 Turnstile 지문검사에 걸려 실패**
(이미지만 막아도 실패). 하지만 **cf_clearance 확보 후엔 차단해도 재챌린지 없음**.
- 동적 차단: _block_active + _blocking_now() 훅. ESM 은 cf_clearance 있을 때만 차단
  (콜드=허용→Turnstile 통과, 웜=이미지/미디어/폰트 차단→대역폭↓). ready_timeout 25s.
  best-effort Turnstile 체크박스 클릭(인터랙티브/나쁜IP 대비).
- 웜업: 기동 시 gmarket/auction/coupang 를 1회 풀어 쿠키 선점(백그라운드) → 실 작업은 웜.
- 유휴 브라우저 정리: _last_used + close_if_idle + run_browser_reaper(유휴 120s) — 메모리 회수.
  쿠키는 user_data_dir 에 남아 재기동해도 (같은 IP면) 웜 유지.
- 테스트: 유휴 정리 2종.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 20:20:58 +09:00

104 lines
3.8 KiB
Python

# 오픈마켓(G마켓·옥션·11번가) 크롤 파서 결정론적 단위 테스트(네트워크/브라우저 불필요).
# fixture 는 실제 렌더된 검색결과에서 카드 3개씩 추출한 것.
from pathlib import Path
import pytest
import time
from services.search.card_parser import parse_cards
from services.search.browser_base import is_proxy_error, BrowserSearchAdapter
from services.search.esm.selectors import GMARKET, AUCTION
from services.search.st11.selectors import CARDS as ST11
FIX = Path(__file__).parent / "fixtures"
CASES = [
("gmarket_search.html", GMARKET.cards, "gmarket", "G마켓"),
("auction_search.html", AUCTION.cards, "auction", "옥션"),
("st11_search.html", ST11, "st11", "11번가"),
]
@pytest.mark.parametrize("fixture,cfg,source,mall", CASES)
def test_parse_extracts_valid_products(fixture, cfg, source, mall):
items = parse_cards((FIX / fixture).read_text(), cfg)
assert len(items) >= 2, f"{source}: 카드 파싱 실패"
for p in items:
assert p.source == source
assert p.mall_name == mall
assert p.name and len(p.name) > 2
assert "상품명" not in p.name and "브랜드명" not in p.name # a11y 라벨 제거 확인
assert p.price >= 100, f"이상 저가: {p.price} ({p.name})"
@pytest.mark.parametrize("fixture,cfg,source,mall", CASES)
def test_shipping_type_valid(fixture, cfg, source, mall):
items = parse_cards((FIX / fixture).read_text(), cfg)
for p in items:
assert p.shipping_type in (None, "free", "paid")
if p.shipping_type == "paid":
assert p.shipping_fee and p.shipping_fee > 0
if p.shipping_type == "free":
assert p.shipping_fee == 0
@pytest.mark.parametrize("msg", [
"Page.goto: net::ERR_TUNNEL_CONNECTION_FAILED at https://...",
"net::ERR_HTTP_RESPONSE_CODE_FAILURE at https://www.coupang.com/...",
"HTTP ERROR 407 Proxy Authentication Required",
"net::ERR_PROXY_CONNECTION_FAILED",
])
def test_is_proxy_error_true(msg):
# 프록시 전송 실패(포트/IP 사망·407) → IP 회전 대상
assert is_proxy_error(msg) is True
@pytest.mark.parametrize("msg", [
"쿠팡 결과 없음/차단 (blocked=True)",
"net::ERR_NAME_NOT_RESOLVED", # DNS — 프록시 문제 아님
"Timeout 40000ms exceeded", # 단순 타임아웃(사이트 지연)
"",
])
def test_is_proxy_error_false(msg):
assert is_proxy_error(msg) is False
class _IdleAdapter(BrowserSearchAdapter):
source = "test"
def _search_url(self, q, l): return ""
def _parse(self, h): return []
class _FakeCtx:
def __init__(self): self.closed = False
async def close(self): self.closed = True
async def test_close_if_idle_keeps_recent_closes_idle():
ad = _IdleAdapter()
ad._ctx = _FakeCtx()
ad._last_used = time.monotonic() # 방금 사용
await ad.close_if_idle(60)
assert ad._ctx is not None # 최근 사용 → 유지
ctx = ad._ctx
ad._last_used = time.monotonic() - 100 # 100s 전(유휴)
await ad.close_if_idle(60)
assert ad._ctx is None and ctx.closed # 유휴 초과 → 브라우저 정리
async def test_close_if_idle_skips_when_no_ctx():
ad = _IdleAdapter() # 브라우저 미기동
await ad.close_if_idle(0) # 예외 없이 no-op
assert ad._ctx is None
def test_dedup_by_link():
# 동일 링크 카드가 반복돼도 1건으로 축약
card = ('<div class="box__item-container"><a href="/x"><span class="text__item-title">상품명 물병</span></a>'
'<div class="box__price-seller">판매가5,000원</div></div>')
html = f"<ul>{card}{card}{card}</ul>"
items = parse_cards(html, GMARKET.cards)
assert len(items) == 1 and items[0].price == 5000