100 lines
3.9 KiB
Python
100 lines
3.9 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
|
|
|
|
|
|
def test_file_read_returns_code(monkeypatch, tmp_path):
|
|
c = _client(monkeypatch, tmp_path)
|
|
files = c.get("/api/v1/strategy/files").json()["files"]
|
|
if not files:
|
|
return
|
|
name = files[0]["name"]
|
|
r = c.get(f"/api/v1/strategy/file/{name}")
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert isinstance(body["code"], str) and body["code"] # 编辑器空白 bug 的回归测试
|
|
assert body["name"] == name
|
|
assert c.get("/api/v1/strategy/file/__no_such__.py").status_code == 404
|
|
|
|
|
|
# ===== 语法检查端点(编辑器「语法检查」按钮真接线)=====
|
|
|
|
def test_check_syntax_ok(monkeypatch, tmp_path):
|
|
c = _client(monkeypatch, tmp_path)
|
|
r = c.post("/api/v1/strategy/file/foo.py/check",
|
|
json={"code": "x = [1, 2, 3]\nprint(x)\n"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"ok": True}
|
|
|
|
|
|
def test_check_syntax_catches_missing_bracket(monkeypatch, tmp_path):
|
|
"""删掉一个中括号必须报错并给行号(2026-08-15 用户实况:删括号仍报通过)。"""
|
|
c = _client(monkeypatch, tmp_path)
|
|
r = c.post("/api/v1/strategy/file/foo.py/check",
|
|
json={"code": "x = [1, 2, 3\nprint(x)\n"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is False
|
|
assert body["line"] is not None and body["line"] >= 1
|
|
assert body["error"]
|
|
|
|
|
|
def test_check_syntax_empty_code_ok(monkeypatch, tmp_path):
|
|
"""空代码/纯注释可编译 → ok(编辑器允许半成品里写注释)。"""
|
|
c = _client(monkeypatch, tmp_path)
|
|
r = c.post("/api/v1/strategy/file/foo.py/check", json={"code": "# 只有注释\n"})
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True
|