from pathlib import Path ROOT = Path(__file__).resolve().parents[1] def test_competitor_intel_cache_reuses_memory_and_shared_file(tmp_path, monkeypatch): from services import competitor_intel_repository as repo monkeypatch.setattr(repo, "_CACHE_FILE", Path(tmp_path) / "competitor_intel_cache.pkl") repo._MEM_CACHE.clear() calls = {"count": 0} def producer(): calls["count"] += 1 return {"valid_matches": 7, "match_rate": 0.1} first = repo._cached_payload("coverage:test", producer, ttl_seconds=60) second = repo._cached_payload("coverage:test", producer, ttl_seconds=60) repo._MEM_CACHE.clear() third = repo._cached_payload("coverage:test", producer, ttl_seconds=60) assert first == second == third == {"valid_matches": 7, "match_rate": 0.1} assert calls["count"] == 1 def test_clear_competitor_intel_cache_removes_shared_file(tmp_path, monkeypatch): from services import competitor_intel_repository as repo cache_file = Path(tmp_path) / "competitor_intel_cache.pkl" monkeypatch.setattr(repo, "_CACHE_FILE", cache_file) repo._MEM_CACHE["x"] = {"time": 1, "value": {"ok": True}} cache_file.write_bytes(b"stale") repo.clear_competitor_intel_cache() assert repo._MEM_CACHE == {} assert not cache_file.exists() def test_competitor_ppt_results_keep_pending_diagnostics_in_export(): source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8") assert "LEFT JOIN valid_competitor" in source assert "\"found\": found" in source assert "\"match_status\"" in source assert "\"candidate_count\"" in source assert "\"unit_comparison\"" in source assert "build_unit_price_comparison" in source assert "(vc.pchome_price IS NULL)" in source assert "(lm.momo_price - vc.pchome_price) AS price_diff" in source assert "((lm.momo_price - vc.pchome_price) / vc.pchome_price * 100) AS price_diff_pct" in source def test_competitor_ppt_results_use_history_for_dated_reports(): source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8") route_source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8") assert "requested_historical_prices" in source assert "use_history_prices" in source assert "FROM competitor_price_history cph" in source assert "cph.crawled_at >= DATE(:start_date)" in source assert "cph.crawled_at < DATE(:end_date) + INTERVAL '1 day'" in source assert "pr.timestamp < DATE(:end_date) + INTERVAL '1 day'" in source assert "'competitor_price_history' AS competitor_source" in source assert "\"competitor_source\"" in source assert "\"pc_crawled_at\"" in source assert "指定期間 competitor_price_history" in route_source def test_competitor_coverage_counts_only_active_product_intersection(): source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8") coverage_source = source.split("def _fetch_competitor_coverage_uncached", 1)[1].split( "def _fetch_manual_review_summary", 1 )[0] assert "coverage:v7" in source assert "rescore_accepted_count" in coverage_source assert "(SELECT COUNT(*) FROM valid_competitor) AS valid_matches" not in coverage_source assert "identity_competitor AS" in coverage_source assert "fresh_competitor AS" in coverage_source assert "FROM latest_momo lm\n JOIN identity_competitor ic ON ic.sku = lm.sku" in coverage_source assert "LEFT JOIN fresh_competitor fc ON fc.sku = lm.sku" in coverage_source assert "WHERE fc.sku IS NULL" in coverage_source assert "\"fresh_matches\": fresh" in coverage_source assert "\"stale_matches\": stale" in coverage_source assert "FROM products p\n JOIN LATERAL" in coverage_source assert "WHERE p.status = 'ACTIVE'" in coverage_source def test_competitor_ppt_and_ai_use_momo_minus_pchome_gap_direction(): ppt_source = (ROOT / "services" / "ppt_generator.py").read_text(encoding="utf-8") route_source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8") assert 'pc_wins = [r for r in found if r.get("price_diff", 0) > 10]' in ppt_source assert 'mo_wins = [r for r in found if r.get("price_diff", 0) < -10]' in ppt_source assert "平均價差 {avg_pct:+.1f}%（momo - PChome）" in ppt_source assert "price_diff = MOMO售價 − PChome售價" in route_source assert "正值=MOMO較貴、PChome低價壓力" in route_source def test_competitor_review_queue_is_canonical_unit_price_handoff(): source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8") daily_template = (ROOT / "templates" / "daily_sales.html").read_text(encoding="utf-8") growth_template = (ROOT / "templates" / "growth_analysis.html").read_text(encoding="utf-8") assert "def fetch_competitor_review_queue" in source assert "review_queue = fetch_competitor_review_queue" in source assert "\"review_queue\": review_queue" in source assert "\"unit_comparable_count\"" in source assert "\"rescore_accepted_count\"" in source assert "manual_review_summary" in source assert "manual_accept_count" in source assert "manual_reject_count" in source assert "manual_unit_price_count" in source assert "manual_rejected" in source assert "manual_unit_price_required" in source assert "manual_needs_research" in source assert "manual_closed" in source assert "competitor_match_reviews" in source assert "\"status_label\"" in source assert "\"action_label\"" in source assert "build_unit_price_comparison" in source assert "需單位價覆核" in daily_template assert "人工採用" in daily_template assert "人工否決" in daily_template assert "人工單位價" in daily_template assert "competitor_intel.review_queue" in daily_template assert "coverage.fresh_matches" in growth_template assert "coverage.fresh_match_rate" in growth_template assert "coverage.stale_matches" in growth_template assert "coverage.unit_comparable_count" in growth_template assert "coverage.rescore_accepted_count" in growth_template assert "coverage.manual_accept_count" in growth_template assert "coverage.manual_reject_count" in growth_template assert "coverage.manual_unit_price_count" in growth_template assert "comp_coverage.rescore_accepted_count" in daily_template assert "comp_coverage.stale_matches" in daily_template def test_competitor_review_filters_split_low_score_operational_buckets(): from services.competitor_intel_repository import REVIEW_STATUS_FILTER_GROUPS assert REVIEW_STATUS_FILTER_GROUPS["recoverable_low_score"] == ("recoverable_low_score",) assert REVIEW_STATUS_FILTER_GROUPS["true_low_confidence"] == ("true_low_confidence",) assert REVIEW_STATUS_FILTER_GROUPS["legacy_low_score"] == ("low_score", "refresh_low_score") assert set(REVIEW_STATUS_FILTER_GROUPS["low_score"]) == { "low_score", "refresh_low_score", "recoverable_low_score", "true_low_confidence", } def test_competitor_review_reasons_prefer_json_payload_labels(): from services.competitor_intel_repository import _format_competitor_review_item item = _format_competitor_review_item({ "sku": "SKU-1", "name": "M.A.C Macximal 柔霧唇膏", "momo_price": 990, "attempt_status": "identity_veto", "candidate_count": 1, "best_competitor_product_id": "DABC123", "best_competitor_product_name": "MAC Macximal 緞光唇膏", "best_competitor_price": 880, "best_match_score": 0.32, "match_diagnostic_json": { "match_type": "no_match", "price_basis": "none", "alert_tier": "suppress", "reasons": [ "makeup_finish_conflict", "nail_tool_function_conflict", "schick_razor_line_conflict", ], }, "error_message": "", }) assert item["match_type_label"] == "非同款" assert item["price_basis_label"] == "不可比" assert item["alert_tier_label"] == "不告警" assert item["diagnostic_reason_text"] == "妝效質地不同、工具功能不同、除毛刀品線不同" assert [reason["code"] for reason in item["diagnostic_reasons"]] == [ "makeup_finish_conflict", "nail_tool_function_conflict", "schick_razor_line_conflict", ] envelope = item["decision_envelope"] assert envelope["decision_type"] == "pchome_match_review" assert envelope["subject"]["sku"] == "SKU-1" assert envelope["subject"]["competitor_product_id"] == "DABC123" assert envelope["guardrails"]["can_auto_execute"] is False assert envelope["guardrails"]["data_quality"] == "partial" assert envelope["guardrails"]["match_type"] == "no_match" assert envelope["recommended_action"]["requires_hitl"] is True assert envelope["recommended_action"]["action"] == "verify_or_reject_identity" assert any(evidence["metric"] == "reasons" for evidence in envelope["evidence"]) def test_rescore_accepted_review_item_has_actionable_decision_envelope(): from services.competitor_intel_repository import _format_competitor_review_item item = _format_competitor_review_item({ "sku": "10922465", "name": "【Herbacin 德國小甘菊】小甘菊1號護手霜20ml", "momo_price": 99, "attempt_status": "rescore_accepted_current", "candidate_count": 1, "best_competitor_product_id": "DDAO4C-A79050612", "best_competitor_product_name": "小甘菊經典護手霜20ml", "best_competitor_price": 89, "best_match_score": 0.872, "match_diagnostic_json": { "match_type": "exact", "price_basis": "total_price", "alert_tier": "identity_review", "reasons": ["focused_exact_identity_herbacin_classic_hand_cream_20ml_brandless"], }, }) envelope = item["decision_envelope"] assert envelope["severity"] in {"P1", "P2"} assert envelope["recommended_action"]["action"] == "review_accept_identity" assert envelope["guardrails"]["data_quality"] == "complete" assert envelope["expected_impact"]["gap_amount"] == 10 assert envelope["expected_impact"]["candidate_gap_pct"] == 11.2 assert any(evidence["metric"] == "candidate_gap_pct" for evidence in envelope["evidence"]) def test_protected_existing_match_envelope_explains_candidate_conflict(): from services.competitor_intel_repository import _format_competitor_review_item item = _format_competitor_review_item({ "sku": "14338675", "name": "【Relove】胺基酸私密潔淨精華凝露120ml", "momo_price": 399, "attempt_status": "protected_existing_match", "candidate_count": 2, "best_competitor_product_id": "QEAE1O-A900A6DRS", "best_competitor_product_name": "RELOVE胺基酸私密清潔凝露120ml", "best_competitor_price": 329, "best_match_score": 0.817, "match_diagnostic_json": { "match_type": "exact", "price_basis": "total_price", "alert_tier": "identity_review", "reasons": ["strong_exact_spec_match", "spec_name_alignment"], }, "error_message": ( "existing_match_conflict;existing_id=QEAE1O-A900A6DNN;" "incoming_id=QEAE1O-A900A6DRS;existing_score=0.766;incoming_score=0.817" ), }) envelope = item["decision_envelope"] conflict = envelope["expected_impact"]["existing_match_conflict"] assert item["existing_match_conflict"]["score_delta"] == 0.051 assert envelope["recommended_action"]["action"] == "compare_existing_identity" assert envelope["severity"] == "P2" assert envelope["guardrails"]["existing_match_protected"] is True assert conflict["existing_product_id"] == "QEAE1O-A900A6DNN" assert conflict["incoming_product_id"] == "QEAE1O-A900A6DRS" assert any(evidence["metric"] == "existing_match_conflict" for evidence in envelope["evidence"]) def test_review_decision_brief_is_shared_by_openclaw_and_ppt(): from services.competitor_intel_repository import ( _format_competitor_review_item, summarize_review_decision_envelopes, ) item = _format_competitor_review_item({ "sku": "10922465", "name": "【Herbacin 德國小甘菊】小甘菊1號護手霜20ml", "momo_price": 99, "attempt_status": "rescore_accepted_current", "candidate_count": 1, "best_competitor_product_id": "DDAO4C-A79050612", "best_competitor_product_name": "小甘菊經典護手霜20ml", "best_competitor_price": 89, "best_match_score": 0.872, "match_diagnostic_json": { "match_type": "exact", "price_basis": "total_price", "alert_tier": "identity_review", }, }) brief = summarize_review_decision_envelopes([item], limit=5) assert brief["hitl_count"] == 1 assert brief["auto_execute_blocked_count"] == 1 assert brief["severity_counts"] assert brief["data_quality_counts"] == {"complete": 1} assert "人工覆核後採用同款" in brief["text"] assert "HITL" in brief["text"] assert "DDAO4C-A79050612" in brief["text"] repo_source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8") openclaw_source = (ROOT / "services" / "openclaw_strategist_service.py").read_text(encoding="utf-8") ppt_source = (ROOT / "services" / "ppt_generator.py").read_text(encoding="utf-8") bot_source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8") assert "\"review_decision_brief\": summarize_review_decision_envelopes" in repo_source assert "summarize_review_decision_envelopes" in openclaw_source assert "review_decision_text" in openclaw_source assert "PChome 覆核決策信封（HITL，不可自動寫正式價差）" in openclaw_source assert "SUM(CASE WHEN attempt_status IN ('unit_comparable'" not in openclaw_source assert "review_decision_brief" in bot_source assert "覆核決策信封（HITL，不可自動寫正式價差）" in bot_source assert "覆核決策信封（HITL）" in ppt_source def test_competitor_ppt_prompt_uses_neutral_ewooc_viewpoint(): source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8") assert "EwoooC 商品營運視角" in source assert "待補資料不可當成成功配對" in source assert "高信心比對" in source assert "待補身份/價格" in source assert "需單位價比較" in source assert "單位價覆核樣本" in source assert "我方 = PChome" not in source assert "請以 PChome 視角" not in source def test_top_competitor_risks_reads_latest_momo_price_after_valid_competitor_filter(): source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8") risk_source = source.split("def _fetch_top_competitor_risks_uncached", 1)[1].split("def fetch_competitor_review_queue", 1)[0] assert "FROM valid_competitor vc" in risk_source assert "JOIN LATERAL" in risk_source assert "WHERE pr.product_id = p.id" in risk_source assert "ROW_NUMBER() OVER (PARTITION BY p.id" not in risk_source