74 lines
2.9 KiB
Python
74 lines
2.9 KiB
Python
from sqlalchemy import create_engine, text
|
|
|
|
|
|
def test_daily_sales_columns_support_sqlite():
|
|
from services.ai_product_pick_agent import _daily_sales_columns, _has_daily_sales_snapshot
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
with engine.begin() as conn:
|
|
conn.execute(text(
|
|
'CREATE TABLE daily_sales_snapshot ('
|
|
'"商品ID" TEXT, snapshot_date TEXT, "總業績" REAL, "數量" INTEGER, "毛利" REAL)'
|
|
))
|
|
|
|
assert _has_daily_sales_snapshot(conn) is True
|
|
assert _daily_sales_columns(conn) == {
|
|
"sku": "商品ID",
|
|
"date": "snapshot_date",
|
|
"revenue": "總業績",
|
|
"qty": "數量",
|
|
"profit": "毛利",
|
|
"cost": None,
|
|
}
|
|
|
|
|
|
def test_fetch_candidates_falls_back_without_sales_on_sqlite_snapshot():
|
|
from services.ai_product_pick_agent import _fetch_candidates
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
with engine.begin() as conn:
|
|
conn.execute(text(
|
|
"CREATE TABLE products ("
|
|
"id INTEGER PRIMARY KEY, i_code TEXT, name TEXT, url TEXT, category TEXT, status TEXT)"
|
|
))
|
|
conn.execute(text(
|
|
"CREATE TABLE price_records ("
|
|
"id INTEGER PRIMARY KEY, product_id INTEGER, price REAL, timestamp TEXT)"
|
|
))
|
|
conn.execute(text(
|
|
"CREATE TABLE competitor_prices ("
|
|
"sku TEXT, source TEXT, expires_at TEXT, match_score REAL, price REAL, original_price REAL, "
|
|
"discount_pct REAL, competitor_product_id TEXT, competitor_product_name TEXT, tags TEXT, crawled_at TEXT)"
|
|
))
|
|
conn.execute(text(
|
|
"CREATE TABLE competitor_price_history ("
|
|
"sku TEXT, source TEXT, price REAL, crawled_at TEXT)"
|
|
))
|
|
conn.execute(text(
|
|
'CREATE TABLE daily_sales_snapshot ('
|
|
'"商品ID" TEXT, snapshot_date TEXT, "總業績" REAL, "數量" INTEGER)'
|
|
))
|
|
conn.execute(text(
|
|
"INSERT INTO products (id, i_code, name, url, category, status) "
|
|
"VALUES (1, 'SKU-1', '測試商品', 'https://example.test/p/1', '美妝', 'ACTIVE')"
|
|
))
|
|
conn.execute(text(
|
|
"INSERT INTO price_records (product_id, price, timestamp) "
|
|
"VALUES (1, 1200, '2026-05-13 10:00:00')"
|
|
))
|
|
conn.execute(text(
|
|
"INSERT INTO competitor_prices ("
|
|
"sku, source, expires_at, match_score, price, original_price, discount_pct, "
|
|
"competitor_product_id, competitor_product_name, tags, crawled_at"
|
|
") VALUES ("
|
|
"'SKU-1', 'pchome', NULL, 0.9, 999, 1200, 16.75, "
|
|
"'PCH-1', '測試商品 PChome', '[\"identity_v2\", \"on_sale\"]', '2026-05-13 10:00:00'"
|
|
")"
|
|
))
|
|
|
|
rows = _fetch_candidates(conn, limit=1)
|
|
|
|
assert len(rows) == 1
|
|
assert rows[0]["sku"] == "SKU-1"
|
|
assert rows[0]["sales_7d"] == 0
|