Files
sanguo_vnpy_v2/tests/api/test_instance_store.py
T

44 lines
1.7 KiB
Python

"""Tests for sanguo_api.instance_store (策略实例 JSON 存储)."""
from sanguo_api import instance_store
def _fresh_store(tmp_path, monkeypatch):
db = tmp_path / "instances.json"
monkeypatch.setattr(instance_store, "_STORE_PATH", str(db))
return str(db)
def test_create_get_list_delete(monkeypatch, tmp_path):
_fresh_store(tmp_path, monkeypatch)
new_id = instance_store.create_instance({
"code_file": "double_ma.py", "name": "测试实例", "type": "cta",
"params": {"fast_window": 10}, "symbol_or_pool": "600519.SH",
"interval": "d", "match_session": "next_open",
})
assert isinstance(new_id, int) and new_id > 0
got = instance_store.get_instance(new_id)["instance"]
assert got["name"] == "测试实例" and got["params"]["fast_window"] == 10
items = instance_store.list_instances()["instances"]
assert any(i["id"] == new_id for i in items)
assert instance_store.delete_instance(new_id) is True
assert instance_store.get_instance(new_id)["instance"] is None
def test_update_instance(monkeypatch, tmp_path):
_fresh_store(tmp_path, monkeypatch)
nid = instance_store.create_instance({
"code_file": "double_ma.py", "name": "u", "type": "cta",
"params": {}, "symbol_or_pool": "000001.SZ", "interval": "d", "match_session": "next_open",
})
instance_store.update_instance(nid, {"name": "改名", "params": {"fast_window": 20}})
got = instance_store.get_instance(nid)["instance"]
assert got["name"] == "改名" and got["params"]["fast_window"] == 20
def test_delete_missing_returns_false(monkeypatch, tmp_path):
_fresh_store(tmp_path, monkeypatch)
assert instance_store.delete_instance(99999) is False