25 lines
1.1 KiB
Python
25 lines
1.1 KiB
Python
# tests/data/test_scheduler.py
|
|
import json
|
|
import pandas as pd
|
|
from unittest.mock import patch
|
|
from sanguo_data.config import DataConfig
|
|
from sanguo_data.scheduler import run_daily_update, UpdateReport
|
|
|
|
def test_run_daily_update_skips_completed_on_resume(tmp_path):
|
|
progress_file = tmp_path / "progress.json"
|
|
progress_file.write_text('{"600000": "done"}')
|
|
cfg = DataConfig(
|
|
data_paths={"daily_dir": str(tmp_path), "vnpy_db": str(tmp_path / "q.db"),
|
|
"progress_file": str(progress_file)},
|
|
data_sources={"daily": [{"name": "eastmoney", "enabled": True}]},
|
|
validation={}, performance={},
|
|
)
|
|
with patch("sanguo_data.scheduler.fetch_daily", return_value=pd.DataFrame({
|
|
"date": ["2026-01-01"], "open": [10.0], "high": [10.0],
|
|
"low": [10.0], "close": [10.0], "volume": [100]})) as m_fetch, \
|
|
patch("sanguo_data.scheduler.write_daily") as m_write:
|
|
report = run_daily_update(cfg, symbols=["600000"])
|
|
assert m_fetch.call_count == 0 # 已 done，跳过
|
|
assert isinstance(report, UpdateReport)
|
|
assert report.skipped == 1
|