70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import create_engine
|
|
|
|
from routes.daily_sales_routes import (
|
|
_read_daily_sales_window,
|
|
_snapshot_date_expr,
|
|
_snapshot_date_window_clause,
|
|
)
|
|
|
|
|
|
class _FakePostgresDialect:
|
|
name = "postgresql"
|
|
|
|
|
|
class _FakePostgresEngine:
|
|
dialect = _FakePostgresDialect()
|
|
|
|
|
|
def test_daily_sales_window_uses_date_cast_for_postgres_text_snapshot_date():
|
|
engine = _FakePostgresEngine()
|
|
|
|
assert _snapshot_date_expr(engine) == '"snapshot_date"::date'
|
|
assert _snapshot_date_window_clause(engine) == (
|
|
'"snapshot_date"::date >= CAST(:start_date AS date) '
|
|
'AND "snapshot_date"::date <= CAST(:end_date AS date)'
|
|
)
|
|
|
|
|
|
def test_read_daily_sales_window_filters_text_snapshot_date_on_sqlite():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
with engine.begin() as conn:
|
|
conn.exec_driver_sql(
|
|
"""
|
|
CREATE TABLE daily_sales_snapshot (
|
|
snapshot_date TEXT,
|
|
"商品ID" TEXT,
|
|
"總業績" TEXT
|
|
)
|
|
"""
|
|
)
|
|
conn.exec_driver_sql(
|
|
"""
|
|
INSERT INTO daily_sales_snapshot (snapshot_date, "商品ID", "總業績")
|
|
VALUES
|
|
('2026-05-01', 'A', '100'),
|
|
('2026-05-02', 'B', '200'),
|
|
('2026-05-03', 'C', '300')
|
|
"""
|
|
)
|
|
|
|
df = _read_daily_sales_window(
|
|
engine,
|
|
"daily_sales_snapshot",
|
|
date(2026, 5, 2),
|
|
date(2026, 5, 3),
|
|
)
|
|
|
|
assert df["商品ID"].tolist() == ["B", "C"]
|
|
|
|
|
|
def test_chart_generator_monthly_revenue_casts_snapshot_date_before_date_math():
|
|
source = (Path(__file__).resolve().parents[1] / "services" / "chart_generator_service.py").read_text(encoding="utf-8")
|
|
|
|
assert "DATE_TRUNC('month', snapshot_date::date)::date" in source
|
|
assert "WHERE snapshot_date::date >= CURRENT_DATE" in source
|
|
assert "DATE_TRUNC('month', snapshot_date)::date" not in source
|
|
assert "WHERE snapshot_date >= NOW()" not in source
|