90 lines
3.8 KiB
Python
90 lines
3.8 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_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()
|