feat(strategy): 策略代码+实例路由组(routes_strategy)并注册 app [nas]
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
"""Tests for sanguo_api.routes_strategy(策略代码 + 实例 路由组)。
|
||||
|
||||
需 fastapi/httpx → 在 NAS docker 容器或装齐后端的 venv 跑(Mac venv310 无 fastapi)。
|
||||
"""
|
||||
from fastapi.testclient import TestClient
|
||||
from sanguo_api.app import create_app
|
||||
from sanguo_api import instance_store
|
||||
|
||||
|
||||
def _client(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(instance_store, "_STORE_PATH", str(tmp_path / "i.json"))
|
||||
app = create_app(db_path=str(tmp_path / "test.db"))
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_files_endpoint_shape(monkeypatch, tmp_path):
|
||||
c = _client(monkeypatch, tmp_path)
|
||||
r = c.get("/api/v1/strategy/files")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert "files" in body and isinstance(body["files"], list)
|
||||
# 自研目录至少出现一个真实文件
|
||||
if body["files"]:
|
||||
dirs = {f["dir"] for f in body["files"]}
|
||||
assert any("sanguo_portfolio/strategies" in d or "sanguo_trader/strategy" in d for d in dirs)
|
||||
|
||||
|
||||
def test_instance_crud_via_http(monkeypatch, tmp_path):
|
||||
c = _client(monkeypatch, tmp_path)
|
||||
r = c.post("/api/v1/strategy/instances", json={
|
||||
"code_file": "double_ma.py", "name": "HTTP实例", "type": "cta",
|
||||
"params": {"fast_window": 5}, "symbol_or_pool": "600519.SH",
|
||||
"interval": "d", "match_session": "next_open",
|
||||
})
|
||||
assert r.status_code == 200, r.text
|
||||
new_id = r.json()["id"]
|
||||
|
||||
assert any(i["id"] == new_id for i in c.get("/api/v1/strategy/instances").json()["instances"])
|
||||
assert c.get(f"/api/v1/strategy/instances/{new_id}").json()["instance"]["name"] == "HTTP实例"
|
||||
|
||||
# update
|
||||
upd = c.put(f"/api/v1/strategy/instances/{new_id}", json={"name": "改名", "params": {"fast_window": 20}})
|
||||
assert upd.status_code == 200
|
||||
assert c.get(f"/api/v1/strategy/instances/{new_id}").json()["instance"]["name"] == "改名"
|
||||
|
||||
# delete
|
||||
assert c.delete(f"/api/v1/strategy/instances/{new_id}").status_code == 200
|
||||
assert c.get(f"/api/v1/strategy/instances/{new_id}").json()["instance"] is None
|
||||
|
||||
|
||||
def test_file_write_rejects_syntax_error(monkeypatch, tmp_path):
|
||||
c = _client(monkeypatch, tmp_path)
|
||||
# 故意写语法错误的代码 → 400(py_compile 门禁)
|
||||
r = c.post("/api/v1/strategy/file/__probe_bad.py", json={"code": "def (:"})
|
||||
assert r.status_code == 400
|
||||
Reference in New Issue
Block a user