57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
import importlib
|
|
import os
|
|
|
|
from sqlalchemy import create_engine, text
|
|
|
|
|
|
def _load_scheduler(monkeypatch, database_url):
|
|
os.environ.setdefault("MOMO_ALLOW_INSECURE_CONFIG_FOR_TESTS", "true")
|
|
import config
|
|
|
|
monkeypatch.setattr(config, "DATABASE_PATH", database_url)
|
|
import scheduler
|
|
|
|
return importlib.reload(scheduler)
|
|
|
|
|
|
def test_verify_import_data_sync_scopes_to_imported_date(monkeypatch, tmp_path):
|
|
db_path = tmp_path / "sales.db"
|
|
database_url = f"sqlite:///{db_path}"
|
|
scheduler = _load_scheduler(monkeypatch, database_url)
|
|
|
|
engine = create_engine(database_url)
|
|
with engine.begin() as conn:
|
|
conn.execute(text('CREATE TABLE daily_sales_snapshot ("日期" TEXT, snapshot_date TEXT, "商品ID" TEXT)'))
|
|
conn.execute(text('CREATE TABLE realtime_sales_monthly ("日期" TEXT, "商品ID" TEXT)'))
|
|
conn.execute(text("""
|
|
INSERT INTO daily_sales_snapshot ("日期", snapshot_date, "商品ID") VALUES
|
|
('2026-05-05', '2026-05-05', 'A001'),
|
|
('2026-05-05', '2026-05-05', 'A002')
|
|
"""))
|
|
conn.execute(text("""
|
|
INSERT INTO realtime_sales_monthly ("日期", "商品ID") VALUES
|
|
('2026-05-01', 'OLD1'),
|
|
('2026-05-01', 'OLD2'),
|
|
('2026-05-05', 'A001'),
|
|
('2026-05-05', 'A002')
|
|
"""))
|
|
|
|
result = scheduler.verify_import_data_sync(
|
|
expected_rows=2,
|
|
date_range={"min": "2026-05-05", "max": "2026-05-05"},
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["daily_sales_snapshot"]["rows"] == 2
|
|
assert result["realtime_sales_monthly"]["rows"] == 2
|
|
|
|
|
|
def test_verify_import_data_sync_rejects_missing_import_scope(monkeypatch, tmp_path):
|
|
scheduler = _load_scheduler(monkeypatch, f"sqlite:///{tmp_path / 'sales.db'}")
|
|
|
|
result = scheduler.verify_import_data_sync(expected_rows=10, date_range=None)
|
|
|
|
assert result["success"] is False
|
|
assert "缺少本次匯入日期範圍" in result["errors"][0]
|
|
|