from pathlib import Path import logging ROOT = Path(__file__).resolve().parents[1] def test_competitor_feeder_persists_all_match_attempt_outcomes(): source = (ROOT / "services/competitor_price_feeder.py").read_text(encoding="utf-8") migration = (ROOT / "migrations/023_competitor_match_attempts.sql").read_text(encoding="utf-8") assert "attempts_written" in source assert "_ensure_competitor_match_attempts_table" in source assert "_record_match_attempt" in source assert "INSERT INTO competitor_match_attempts" in source assert "CAST(:search_terms AS jsonb)" in source assert 'attempt_status="matched"' in source assert 'attempt_status="low_score"' in source assert 'attempt_status="no_result"' in source assert 'attempt_status="no_match"' in source assert 'attempt_status="error"' in source assert "_search_pchome_candidates(crawler, momo_name, search_terms, momo_price=momo_price)" in source assert 'attempt_status="needs_review"' in source assert "_should_upsert_competitor_price" in source assert "replace_legacy_unverified" in source assert "identity_v2" in source assert "CREATE TABLE IF NOT EXISTS competitor_match_attempts" in migration assert "attempt_status" in migration assert "search_terms" in migration assert "best_match_score" in migration assert "error_message" in migration assert "idx_comp_match_attempts_sku_source_time" in migration def test_competitor_feeder_logs_keyword_parser_fallback(monkeypatch, caplog): from services import competitor_price_feeder from services import marketplace_product_matcher def broken_build_search_terms(*_args, **_kwargs): raise RuntimeError("matcher unavailable") monkeypatch.setattr(marketplace_product_matcher, "build_search_terms", broken_build_search_terms) caplog.set_level(logging.DEBUG, logger="services.competitor_price_feeder") terms = competitor_price_feeder._build_search_keywords("理膚寶水 B5 修復霜 40ml") assert terms assert "fallback to cleaned product name" in caplog.text