c955a8f367
프록시(DECODO) 포트/IP 사망(407/ERR_TUNNEL/ERR_HTTP_RESPONSE_CODE_FAILURE)이
봇차단과 구분 없이 예외로 튕겨 같은 죽은 포트로 재시도만 하다 DEAD 되던 문제를 고친다.
- browser_base: is_proxy_error(순수함수) + search 루프에서 프록시 전송오류 시 IP 회전 재시도
(max_proxy_retries=2). 봇감지 회전과 통합. uses_proxy 프로퍼티.
- 쿠팡 어댑터를 BrowserSearchAdapter 로 통합 — 중복 machinery 제거, 회전 로직 한 곳에서 공유
(detect_block 순수함수는 유지, 테스트 호환).
- proxy.healthcheck(): 시작 프리플라이트 — 살아있는 포트 선점 + egress IP 로그(빠른 실패·가시성).
worker_main 기동 시 호출.
- 비용: DecodoConfig.cost_per_gb 추가. metrics 에 proxy_bytes(네이버 직접 제외) + 컴포넌트별
cost{ai_usd, proxy_usd, total_usd}. FE 원가 타일에 AI/DECODO 분해·프록시 바이트.
- 테스트: is_proxy_error 8종 + 비용 분해 1종.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
# 오픈마켓(G마켓·옥션·11번가) 크롤 파서 결정론적 단위 테스트(네트워크/브라우저 불필요).
|
|
# fixture 는 실제 렌더된 검색결과에서 카드 3개씩 추출한 것.
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from services.search.card_parser import parse_cards
|
|
from services.search.browser_base import is_proxy_error
|
|
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
|
|
|
|
|
|
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
|