36 lines
1.2 KiB
Python
36 lines
1.2 KiB
Python
"""scheduler 测试（mock APScheduler，spec §9.2）。"""
|
||
from unittest.mock import MagicMock
|
||
|
||
from sanguo_trader import scheduler
|
||
from sanguo_trader.persistence import (
|
||
init_db, save_account, update_account_status,
|
||
)
|
||
|
||
|
||
def test_register_paper_job(monkeypatch):
|
||
fake = MagicMock()
|
||
monkeypatch.setattr(scheduler, "start_scheduler", lambda: fake)
|
||
job_id = scheduler.register_paper_job(123, job_func=lambda a: None)
|
||
assert job_id == "paper_123"
|
||
fake.add_job.assert_called_once()
|
||
|
||
|
||
def test_remove_paper_job(monkeypatch):
|
||
fake = MagicMock()
|
||
monkeypatch.setattr(scheduler, "get_scheduler", lambda: fake)
|
||
scheduler.remove_paper_job(123)
|
||
fake.remove_job.assert_called_once_with("paper_123")
|
||
|
||
|
||
def test_restore_live_jobs_filters_running_live(monkeypatch, tmp_path):
|
||
db = str(tmp_path / "s.db")
|
||
init_db(db)
|
||
a_live = save_account(db, {"name": "t1", "mode": "live"})
|
||
a_replay = save_account(db, {"name": "t2", "mode": "replay"})
|
||
update_account_status(db, a_live, "running")
|
||
update_account_status(db, a_replay, "running")
|
||
fake = MagicMock()
|
||
monkeypatch.setattr(scheduler, "start_scheduler", lambda: fake)
|
||
n = scheduler.restore_live_jobs(db, job_func=lambda a: None)
|
||
assert n == 1 # 只 live & running
|