56 lines
2.3 KiB
Python
56 lines
2.3 KiB
Python
"""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
|