73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
"""paper/live 生命周期 API 单测（D1）：stop/resume/delete/update 直调路由函数。"""
|
|
import pytest
|
|
|
|
from sanguo_api import routes_paper as rp
|
|
|
|
|
|
@pytest.fixture()
|
|
def paper_db(tmp_path):
|
|
db = str(tmp_path / "paper.db")
|
|
rp.set_db_path(db)
|
|
return db
|
|
|
|
|
|
def _create(paper_db, mode="live", name="p1"):
|
|
req = rp.PaperCreateRequest(
|
|
mode=mode, symbols=["600000"],
|
|
strategies=[rp.StrategyCfg(name="DoubleMaStrategy", symbol="600000")],
|
|
start="2026-01-01", end="2026-12-31", name=name,
|
|
)
|
|
# 直接走 save_account（绕过 create 的后台线程/引擎依赖）
|
|
from sanguo_trader.persistence import save_account, update_account_status
|
|
aid = save_account(paper_db, req.model_dump())
|
|
if mode == "live":
|
|
update_account_status(paper_db, aid, "running")
|
|
return aid
|
|
|
|
|
|
def test_stop_resume_live(paper_db):
|
|
aid = _create(paper_db, mode="live")
|
|
assert rp.stop_paper(aid)["status"] == "stopped"
|
|
assert rp.resume_paper(aid)["status"] == "running"
|
|
|
|
|
|
def test_stop_replay_rejected(paper_db):
|
|
aid = _create(paper_db, mode="replay")
|
|
with pytest.raises(Exception):
|
|
rp.stop_paper(aid)
|
|
|
|
|
|
def test_stop_unknown_404(paper_db):
|
|
with pytest.raises(Exception):
|
|
rp.stop_paper(999)
|
|
|
|
|
|
def test_delete_removes_all_tables(paper_db):
|
|
aid = _create(paper_db, mode="live")
|
|
import sqlite3
|
|
from sanguo_trader.persistence import save_daily_balance, save_trade
|
|
save_daily_balance(paper_db, aid, "2026-08-13", 100.0, 0.0, 100.0)
|
|
save_trade(paper_db, aid, {"strategy_id": "s", "datetime": "2026-08-13",
|
|
"symbol": "600000", "direction": "long",
|
|
"offset": "open", "match_session": "next_open",
|
|
"price": 10.0, "volume": 100})
|
|
assert rp.delete_paper(aid)["deleted"] is True
|
|
with sqlite3.connect(paper_db) as conn:
|
|
for t in ("paper_accounts", "paper_daily_balance", "paper_trades"):
|
|
n = conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
|
|
assert n == 0, t
|
|
|
|
|
|
def test_update_fields(paper_db):
|
|
aid = _create(paper_db, mode="live")
|
|
req = rp.PaperUpdateRequest(name="renamed", initial_capital=2_000_000)
|
|
assert rp.update_paper(aid, req)["updated"] is True
|
|
acc = rp.get_paper(aid)
|
|
assert acc["name"] == "renamed"
|
|
assert acc["initial_capital"] == 2_000_000
|
|
|
|
|
|
def test_update_noop(paper_db):
|
|
aid = _create(paper_db, mode="live")
|
|
assert rp.update_paper(aid, rp.PaperUpdateRequest())["updated"] is False
|