287 lines
12 KiB
Python
287 lines
12 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 _seed_external_offer_sync_tables(engine):
|
|
with engine.begin() as conn:
|
|
conn.execute(text("""
|
|
CREATE TABLE external_market_sources (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
code TEXT UNIQUE NOT NULL,
|
|
display_name TEXT NOT NULL,
|
|
platform_code TEXT NOT NULL,
|
|
source_kind TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'paused',
|
|
enabled BOOLEAN NOT NULL DEFAULT 0,
|
|
write_enabled BOOLEAN NOT NULL DEFAULT 0,
|
|
allowed_input_methods_json TEXT,
|
|
quality_policy_json TEXT,
|
|
plain_note TEXT,
|
|
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TEXT DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE TABLE external_offers (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
source_code TEXT NOT NULL,
|
|
platform_code TEXT NOT NULL,
|
|
source_product_id TEXT NOT NULL,
|
|
source_offer_key TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
brand TEXT,
|
|
category_text TEXT,
|
|
product_url TEXT,
|
|
image_url TEXT,
|
|
price REAL,
|
|
original_price REAL,
|
|
currency TEXT NOT NULL DEFAULT 'TWD',
|
|
stock_status TEXT,
|
|
sold_count INTEGER,
|
|
rating REAL,
|
|
review_count INTEGER,
|
|
observed_at TEXT NOT NULL,
|
|
expires_at TEXT,
|
|
ingestion_method TEXT NOT NULL,
|
|
connector_key TEXT,
|
|
pchome_product_id TEXT,
|
|
momo_sku TEXT,
|
|
match_status TEXT NOT NULL DEFAULT 'unmatched',
|
|
quality_score REAL NOT NULL DEFAULT 0,
|
|
data_quality_status TEXT NOT NULL DEFAULT 'needs_review',
|
|
quality_notes_json TEXT,
|
|
raw_payload_json TEXT,
|
|
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE (source_code, source_product_id, observed_at, ingestion_method)
|
|
)
|
|
"""))
|
|
conn.execute(text(
|
|
"CREATE TABLE competitor_prices ("
|
|
"sku TEXT, source TEXT, competitor_product_id TEXT, competitor_product_name TEXT, "
|
|
"price REAL, match_score REAL, tags TEXT, crawled_at TEXT, expires_at TEXT)"
|
|
))
|
|
conn.execute(text(
|
|
"CREATE TABLE products ("
|
|
"id INTEGER PRIMARY KEY, i_code TEXT, name TEXT, url TEXT, image_url TEXT, status TEXT)"
|
|
))
|
|
conn.execute(text(
|
|
"CREATE TABLE price_records ("
|
|
"id INTEGER PRIMARY KEY, product_id INTEGER, price REAL, timestamp TEXT)"
|
|
))
|
|
conn.execute(text("""
|
|
INSERT INTO competitor_prices
|
|
(sku, source, competitor_product_id, competitor_product_name,
|
|
price, match_score, tags, crawled_at, expires_at)
|
|
VALUES
|
|
('MOMO-1', 'pchome', 'PCH-1', 'PChome 商品',
|
|
1000, 0.91, '["identity_v2"]', '2026-06-15 09:30:00', NULL),
|
|
('MOMO-2', 'pchome', 'PCH-2', '低信心商品',
|
|
600, 0.60, '["identity_v2"]', '2026-06-15 09:30:00', NULL)
|
|
"""))
|
|
conn.execute(text(
|
|
"INSERT INTO products (id, i_code, name, url, image_url, status) "
|
|
"VALUES (1, 'MOMO-1', 'MOMO 同款商品', 'https://momo.test/1', 'https://img.test/1.jpg', 'ACTIVE')"
|
|
))
|
|
conn.execute(text(
|
|
"INSERT INTO price_records (id, product_id, price, timestamp) "
|
|
"VALUES (1, 1, 900, '2026-06-15 10:00:00')"
|
|
))
|
|
|
|
|
|
def test_sync_legacy_momo_reference_offers_writes_verified_cache_to_external_offers():
|
|
from services.external_market_offer_service import sync_legacy_momo_reference_offers
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
_seed_external_offer_sync_tables(engine)
|
|
|
|
payload = sync_legacy_momo_reference_offers(engine, limit=20)
|
|
|
|
assert payload["success"] is True
|
|
assert payload["status"] == "synced"
|
|
assert payload["candidate_count"] == 1
|
|
assert payload["written_count"] == 1
|
|
with engine.connect() as conn:
|
|
rows = conn.execute(text("""
|
|
SELECT source_code, platform_code, source_product_id, title, price,
|
|
pchome_product_id, momo_sku, match_status, quality_score,
|
|
data_quality_status, ingestion_method
|
|
FROM external_offers
|
|
""")).mappings().all()
|
|
|
|
assert len(rows) == 1
|
|
row = dict(rows[0])
|
|
assert row["source_code"] == "momo_reference"
|
|
assert row["platform_code"] == "momo"
|
|
assert row["source_product_id"] == "MOMO-1"
|
|
assert row["pchome_product_id"] == "PCH-1"
|
|
assert row["title"] == "MOMO 同款商品"
|
|
assert row["price"] == 900
|
|
assert row["match_status"] == "verified"
|
|
assert row["quality_score"] == 91
|
|
assert row["data_quality_status"] == "verified"
|
|
assert row["ingestion_method"] == "legacy_competitor_cache"
|
|
|
|
|
|
def test_sync_legacy_momo_reference_offers_dry_run_does_not_write():
|
|
from services.external_market_offer_service import sync_legacy_momo_reference_offers
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
_seed_external_offer_sync_tables(engine)
|
|
|
|
payload = sync_legacy_momo_reference_offers(engine, limit=20, dry_run=True)
|
|
|
|
assert payload["success"] is True
|
|
assert payload["status"] == "dry_run"
|
|
assert payload["candidate_count"] == 1
|
|
assert payload["written_count"] == 0
|
|
with engine.connect() as conn:
|
|
count = conn.execute(text("SELECT COUNT(*) FROM external_offers")).scalar()
|
|
assert 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
|
|
|
|
|
|
def test_external_offer_sync_is_registered_in_scheduler():
|
|
from pathlib import Path
|
|
|
|
scheduler_source = Path("scheduler.py").read_text(encoding="utf-8")
|
|
run_scheduler_source = Path("run_scheduler.py").read_text(encoding="utf-8")
|
|
|
|
assert "def run_external_offer_sync_task" in scheduler_source
|
|
assert "sync_legacy_momo_reference_offers" in scheduler_source
|
|
assert "run_external_offer_sync_task" in run_scheduler_source
|
|
assert "external_offer_sync" in run_scheduler_source
|