This commit is contained in:
56
tests/test_auto_import_data_sync.py
Normal file
56
tests/test_auto_import_data_sync.py
Normal file
@@ -0,0 +1,56 @@
|
||||
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]
|
||||
|
||||
@@ -7,9 +7,12 @@ GUIDE = ROOT / "docs/guides/modularization_governance.md"
|
||||
|
||||
|
||||
def _python_line_counts():
|
||||
ignored_parts = {".git", "venv", "__pycache__", ".pytest_cache"}
|
||||
ignored_parts = {".git", "venv", "backups", "__pycache__", ".pytest_cache"}
|
||||
for path in ROOT.rglob("*.py"):
|
||||
if any(part in ignored_parts for part in path.relative_to(ROOT).parts):
|
||||
parts = path.relative_to(ROOT).parts
|
||||
if any(part in ignored_parts for part in parts):
|
||||
continue
|
||||
if len(parts) >= 2 and parts[0] == ".claude" and parts[1] == "worktrees":
|
||||
continue
|
||||
with path.open(encoding="utf-8", errors="ignore") as handle:
|
||||
yield path.relative_to(ROOT).as_posix(), sum(1 for _ in handle)
|
||||
|
||||
Reference in New Issue
Block a user