88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
from datetime import date
|
|
|
|
from sqlalchemy import create_engine
|
|
|
|
from routes.sales_routes import (
|
|
_fetch_growth_latest_detail_month,
|
|
_fetch_growth_payload_sql,
|
|
_fetch_growth_payload_summary,
|
|
_get_growth_source_fingerprint,
|
|
)
|
|
|
|
|
|
def _seed_growth_tables(engine):
|
|
with engine.begin() as conn:
|
|
conn.exec_driver_sql(
|
|
"""
|
|
CREATE TABLE monthly_summary_analysis (
|
|
"year" INTEGER,
|
|
"month" INTEGER,
|
|
sales_amt_curr REAL,
|
|
profit_amt_curr REAL,
|
|
sales_vol_curr REAL
|
|
)
|
|
"""
|
|
)
|
|
conn.exec_driver_sql(
|
|
"""
|
|
CREATE TABLE realtime_sales_monthly (
|
|
"日期" TEXT,
|
|
"總業績" TEXT,
|
|
"總成本" TEXT,
|
|
"數量" TEXT,
|
|
"訂單編號" TEXT
|
|
)
|
|
"""
|
|
)
|
|
conn.exec_driver_sql(
|
|
"""
|
|
INSERT INTO monthly_summary_analysis
|
|
("year", "month", sales_amt_curr, profit_amt_curr, sales_vol_curr)
|
|
VALUES (2025, 12, 1000, 250, 10)
|
|
"""
|
|
)
|
|
conn.exec_driver_sql(
|
|
"""
|
|
INSERT INTO realtime_sales_monthly
|
|
("日期", "總業績", "總成本", "數量", "訂單編號")
|
|
VALUES
|
|
('2026-05-01', '1000', '700', '10', 'A001'),
|
|
('2026-05-02', '2000', '1200', '20', 'A002')
|
|
"""
|
|
)
|
|
|
|
|
|
def test_growth_summary_is_skipped_when_realtime_detail_is_newer():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
_seed_growth_tables(engine)
|
|
|
|
assert _fetch_growth_latest_detail_month(engine) == date(2026, 5, 1)
|
|
assert _fetch_growth_payload_summary(engine) is None
|
|
|
|
chart_data, kpi = _fetch_growth_payload_sql(engine, "realtime_sales_monthly")
|
|
assert chart_data["labels"] == ["2026-05"]
|
|
assert chart_data["revenue"] == [3000.0]
|
|
assert chart_data["orders"] == [30]
|
|
assert chart_data["aov"] == [100.0]
|
|
assert kpi["current_year"] == 2026
|
|
assert kpi["ytd_revenue"] == 3000.0
|
|
assert kpi["total_orders"] == 30
|
|
|
|
|
|
def test_growth_source_fingerprint_changes_after_import_rows_change():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
_seed_growth_tables(engine)
|
|
|
|
before = _get_growth_source_fingerprint(engine)
|
|
with engine.begin() as conn:
|
|
conn.exec_driver_sql(
|
|
"""
|
|
INSERT INTO realtime_sales_monthly
|
|
("日期", "總業績", "總成本", "數量", "訂單編號")
|
|
VALUES ('2026-05-03', '500', '300', '5', 'A003')
|
|
"""
|
|
)
|
|
after = _get_growth_source_fingerprint(engine)
|
|
|
|
assert before != after
|