72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
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
|
|
|
|
|
|
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 "我方 = 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")
|
|
|
|
assert "FROM valid_competitor vc" in source
|
|
assert "JOIN LATERAL" in source
|
|
assert "WHERE pr.product_id = p.id" in source
|
|
assert "ROW_NUMBER() OVER (PARTITION BY p.id" not in source.split("def _fetch_top_competitor_risks_uncached", 1)[1].split("def fetch_competitor_comparison_results", 1)[0]
|