313d882df4
네이버 오픈API 결과엔 G마켓·옥션·11번가 등이 이미 mallName 으로 들어오는데 최저가 1건만 쓰고 몰 정보를 버리고 있었다. summarize_by_mall() 로 매칭 후보를 (소스, 판매몰)별 최저가로 축약해 result.by_mall(가격 오름차순)에 노출한다. 추가 요청 0건으로 오픈마켓 몰별 최저가를 확보 — 별도 크롤러 불필요. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
"""코어 파이프라인 테스트 — 필터·IQR 이상치·top-N 최저가 (결정론적, 네트워크 불필요)."""
|
|
|
|
from pathlib import Path
|
|
|
|
from services.search.contract import NormalizedProduct
|
|
from services.search.coupang.parser import parse_search_html
|
|
from services.pipeline.filters import filter_by_price_band, filter_out_malls
|
|
from services.pipeline.outliers import remove_price_outliers
|
|
from services.pipeline.core import run_price_pipeline, summarize_by_mall
|
|
|
|
FIXTURE = Path(__file__).parent / "fixtures" / "coupang_search.html"
|
|
|
|
|
|
def _p(price, name="p", mall="쿠팡"):
|
|
return NormalizedProduct(source="coupang", name=name, price=price, mall_name=mall)
|
|
|
|
|
|
def test_top_n_is_lowest_price_sorted():
|
|
prods = [_p(x) for x in [3000, 1000, 2000, 5000, 4000]]
|
|
r = run_price_pipeline(prods, remove_outliers=False, top_n=3)
|
|
assert [p["price"] for p in r["top"]] == [1000, 2000, 3000]
|
|
assert r["lowest"]["price"] == 1000
|
|
|
|
|
|
def test_outlier_removes_extreme_low_and_high():
|
|
normal = [_p(x) for x in [1000, 1010, 1020, 1030, 1040, 1050, 1060, 1070, 1080, 1090]]
|
|
kept, removed = remove_price_outliers(normal + [_p(5), _p(500000)])
|
|
prices_removed = {p.price for p in removed}
|
|
assert 5 in prices_removed and 500000 in prices_removed
|
|
assert all(1000 <= p.price <= 1090 for p in kept)
|
|
|
|
|
|
def test_price_band_filter():
|
|
prods = [_p(x) for x in [8000, 10000, 12000, 30000, 1000]]
|
|
# 기준가 10000, ±70% → [3000, 17000] 만 통과
|
|
out = filter_by_price_band(prods, base_price=10000, tolerance=0.7)
|
|
assert {p.price for p in out} == {8000, 10000, 12000}
|
|
|
|
|
|
def test_filter_out_malls():
|
|
prods = [_p(1000, mall="쿠팡"), _p(2000, mall="G마켓"), _p(3000, mall="옥션")]
|
|
out = filter_out_malls(prods, ["G마켓", "옥션"])
|
|
assert {p.mall_name for p in out} == {"쿠팡"}
|
|
|
|
|
|
def test_empty_input_is_safe():
|
|
r = run_price_pipeline([], top_n=5)
|
|
assert r["lowest"] is None and r["top"] == [] and r["total_found"] == 0
|
|
|
|
|
|
def test_stages_recorded():
|
|
prods = [_p(x) for x in [1000, 2000, 3000, 4000, 5000]]
|
|
r = run_price_pipeline(prods, base_price=3000, top_n=2)
|
|
names = [s["stage"] for s in r["stages"]]
|
|
assert "price_band" in names and "outlier" in names and names[-1] == "top_n"
|
|
assert r["stages"][-1]["out"] == 2 # top_n 결과 수
|
|
|
|
|
|
def _pm(source, mall, price):
|
|
return NormalizedProduct(source=source, name=f"{mall}상품", price=price, mall_name=mall)
|
|
|
|
|
|
def test_summarize_by_mall_lowest_per_mall_sorted():
|
|
# 같은 몰 여러 건 → 몰별 최저가만, 전체 가격 오름차순
|
|
prods = [
|
|
_pm("naver", "G마켓", 12000), _pm("naver", "G마켓", 11000),
|
|
_pm("naver", "11번가", 10500), _pm("naver", "네이버", 9800),
|
|
_pm("coupang", "쿠팡", 10200),
|
|
]
|
|
rows = summarize_by_mall(prods)
|
|
# 몰별 최저가 1건씩, 전체 가격 오름차순
|
|
assert [(r["mall_name"], r["price"]) for r in rows] == [
|
|
("네이버", 9800), ("쿠팡", 10200), ("11번가", 10500), ("G마켓", 11000),
|
|
]
|
|
|
|
|
|
def test_summarize_by_mall_in_result():
|
|
r = run_price_pipeline([_pm("naver", "11번가", 5000), _pm("naver", "G마켓", 6000)], remove_outliers=False)
|
|
malls = {row["mall_name"] for row in r["by_mall"]}
|
|
assert malls == {"11번가", "G마켓"}
|
|
|
|
|
|
def test_pipeline_on_real_fixture():
|
|
products = parse_search_html(FIXTURE.read_text())
|
|
r = run_price_pipeline(products, remove_outliers=False, top_n=3)
|
|
prices = [p["price"] for p in r["top"]]
|
|
assert prices == sorted(prices) # 최저가순
|
|
assert r["lowest"]["price"] == min(p.price for p in products)
|