137 lines
5.7 KiB
Python
137 lines
5.7 KiB
Python
from sqlalchemy import create_engine, text
|
|
|
|
|
|
def test_connector_contract_keeps_shopee_and_coupang_paused_with_manual_csv_path():
|
|
from services.external_market_offer_service import build_connector_contracts
|
|
|
|
payload = build_connector_contracts()
|
|
|
|
assert payload["success"] is True
|
|
assert payload["plain_summary"] == "所有外部市場資料都先轉成同一份商品報價格式，再進作戰清單。"
|
|
sources = {source["code"]: source for source in payload["sources"]}
|
|
assert sources["momo_reference"]["status_label"] == "正在使用"
|
|
assert sources["shopee"]["status_label"] == "先暫停"
|
|
assert sources["coupang"]["status_label"] == "先暫停"
|
|
assert "手動 CSV" in sources["shopee"]["input_methods"]
|
|
assert "官方 API" in sources["coupang"]["input_methods"]
|
|
assert "price" in payload["manual_csv"]["required_headers"]
|
|
|
|
|
|
def test_normalized_offer_payload_validates_plain_required_fields():
|
|
from services.external_market_offer_service import normalize_external_offer_payload
|
|
|
|
record, errors = normalize_external_offer_payload({
|
|
"source_code": "momo_reference",
|
|
"source_product_id": "MOMO-1",
|
|
"title": "外部參考商品",
|
|
"price": "899",
|
|
"observed_at": "2026-06-15T10:00:00",
|
|
"ingestion_method": "manual_csv",
|
|
"quality_score": 82,
|
|
"quality_note": "人工確認同款",
|
|
})
|
|
|
|
assert errors == []
|
|
assert record is not None
|
|
assert record.to_record()["source_offer_key"] == "momo_reference:MOMO-1"
|
|
assert record.to_record()["data_quality_status"] == "needs_review"
|
|
|
|
missing_record, missing_errors = normalize_external_offer_payload({
|
|
"source_code": "shopee",
|
|
"price": 0,
|
|
})
|
|
|
|
assert missing_record is None
|
|
assert "缺少外部商品 ID" in missing_errors
|
|
assert "售價必須大於 0" in missing_errors
|
|
|
|
|
|
def test_external_offer_csv_dry_run_accepts_chinese_headers_and_classifies_rows():
|
|
from services.external_market_offer_service import dry_run_external_offer_csv
|
|
|
|
csv_text = "\n".join([
|
|
"資料來源,外部商品ID,商品名稱,售價,資料時間,取得方式,PChome商品ID,同款狀態,資料可信度",
|
|
"momo_reference,MOMO-1,可用商品,899,2026-06-15T10:00:00,manual_csv,PCH-1,verified,88",
|
|
"shopee,SHP-1,暫停來源商品,799,2026-06-15T10:01:00,manual_csv,PCH-2,verified,90",
|
|
"momo_reference,MOMO-2,缺價格商品,,2026-06-15T10:02:00,manual_csv,PCH-3,verified,90",
|
|
])
|
|
|
|
payload = dry_run_external_offer_csv(csv_text)
|
|
|
|
assert payload["success"] is True
|
|
assert payload["message"] == "CSV 預檢完成，尚未寫入資料。"
|
|
assert payload["summary"] == {
|
|
"total_rows": 3,
|
|
"ready_count": 1,
|
|
"review_count": 1,
|
|
"blocked_count": 1,
|
|
}
|
|
rows = payload["rows"]
|
|
assert rows[0]["status_label"] == "可使用"
|
|
assert rows[1]["status_label"] == "需人工確認"
|
|
assert "這個來源目前先暫停，不進告警" in rows[1]["reasons"]
|
|
assert rows[2]["status_label"] == "不能使用"
|
|
assert "缺少售價" in rows[2]["reasons"]
|
|
|
|
|
|
def test_external_offer_csv_dry_run_reports_empty_file_plainly():
|
|
from services.external_market_offer_service import dry_run_external_offer_csv
|
|
|
|
payload = dry_run_external_offer_csv("")
|
|
|
|
assert payload["success"] is False
|
|
assert payload["errors"] == ["CSV 內容是空的"]
|
|
assert payload["summary"]["blocked_count"] == 0
|
|
|
|
|
|
def test_external_source_readiness_uses_legacy_momo_reference_cache():
|
|
from services.external_market_offer_service import build_external_source_readiness
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
with engine.begin() as conn:
|
|
conn.execute(text(
|
|
"CREATE TABLE competitor_prices ("
|
|
"source TEXT, competitor_product_id TEXT, price REAL, match_score REAL, "
|
|
"tags TEXT, crawled_at TEXT, expires_at TEXT)"
|
|
))
|
|
conn.execute(text("""
|
|
INSERT INTO competitor_prices
|
|
(source, competitor_product_id, price, match_score, tags, crawled_at, expires_at)
|
|
VALUES
|
|
('pchome', 'PCH-1', 1000, 0.91, '["identity_v2"]', '2026-06-15 10:00:00', NULL),
|
|
('pchome', 'PCH-2', 500, 0.50, '["identity_v2"]', '2026-06-15 10:00:00', NULL)
|
|
"""))
|
|
|
|
payload = build_external_source_readiness(engine)
|
|
|
|
assert payload["success"] is True
|
|
assert payload["schema_ready"] is False
|
|
sources = {source["code"]: source for source in payload["sources"]}
|
|
assert sources["momo_reference"]["usable_offer_count"] == 1
|
|
assert sources["momo_reference"]["plain_state"] == "已接入，可進作戰清單"
|
|
assert sources["shopee"]["plain_state"] == "先保留接口，不進告警"
|
|
assert payload["plain_summary"] == "MOMO 先用；蝦皮與酷澎先保留接口，暫不進告警。"
|
|
|
|
|
|
def test_external_market_migration_creates_source_and_offer_tables():
|
|
from pathlib import Path
|
|
|
|
migration = Path("migrations/044_external_market_offer_normalization.sql").read_text(encoding="utf-8")
|
|
|
|
assert "CREATE TABLE IF NOT EXISTS external_market_sources" in migration
|
|
assert "CREATE TABLE IF NOT EXISTS external_offers" in migration
|
|
assert "momo_reference" in migration
|
|
assert "shopee" in migration
|
|
assert "coupang" in migration
|
|
assert "DROP " not in migration.upper()
|
|
assert "TRUNCATE " not in migration.upper()
|
|
|
|
|
|
def test_external_offer_csv_dry_run_route_is_registered_as_post_only():
|
|
from pathlib import Path
|
|
|
|
route_source = Path("routes/ai_routes.py").read_text(encoding="utf-8")
|
|
|
|
assert "@ai_bp.route('/api/ai/pchome-growth/external-offers/csv-dry-run', methods=['POST'])" in route_source
|
|
assert "dry_run_external_offer_csv" in route_source
|