matitos
7b97f010c9
Phones kept running old JS because their cached index.html pointed at an old hashed asset. Both images now bake the commit SHA as BUILD_VERSION; the backend exposes it at GET /api/version and the client reloads once with a cache-busting query when its own baked version differs. Guarded by sessionStorage so it cannot loop. /api/version is unauthenticated like /api/health.
244 lines
8.8 KiB
Python
244 lines
8.8 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
from app.main import create_app
|
|
from app.market.fake import FakeMarketData
|
|
from app.runtime import Runtime
|
|
|
|
|
|
def test_version_endpoint(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert client.get("/api/version").status_code == 200
|
|
assert "version" in client.get("/api/version").json()
|
|
|
|
|
|
def test_health_and_strategies(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert client.get("/api/health").json() == {"status": "ok"}
|
|
keys = {s["key"] for s in client.get("/api/strategies").json()}
|
|
assert {"sma_cross", "rsi_reversion"} <= keys
|
|
|
|
|
|
def test_start_status_stop_flow(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
status = client.get("/api/status").json()
|
|
assert status["running"] is False
|
|
|
|
started = client.post(
|
|
"/api/bot/start",
|
|
json={"strategy_key": "sma_cross", "params": {"fast": 2, "slow": 4}},
|
|
)
|
|
assert started.status_code == 200
|
|
assert started.json()["running"] is True
|
|
|
|
assert client.get("/api/positions").status_code == 200
|
|
assert client.get("/api/orders").status_code == 200
|
|
assert client.get("/api/events").status_code == 200
|
|
|
|
stopped = client.post("/api/bot/stop").json()
|
|
assert stopped["running"] is False
|
|
|
|
|
|
def test_unknown_strategy_returns_404(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post("/api/bot/start", json={"strategy_key": "does_not_exist"})
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_backtest_endpoint(runtime: Runtime, market: FakeMarketData) -> None:
|
|
market.candles = FakeMarketData.sine_wave(200).candles
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/backtest",
|
|
json={"strategy_key": "sma_cross", "params": {"fast": 5, "slow": 20}, "limit": 200},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["strategy_key"] == "sma_cross"
|
|
assert data["error"] is None
|
|
assert len(data["equity_curve"]) == 200
|
|
assert "total_return_pct" in data["metrics"]
|
|
assert data["metrics"]["num_trades"] >= 1
|
|
|
|
|
|
def test_sweep_endpoint(runtime: Runtime, market: FakeMarketData) -> None:
|
|
market.candles = FakeMarketData.sine_wave(200).candles
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/sweep",
|
|
json={
|
|
"strategy_key": "sma_cross",
|
|
"grid": {"fast": [3, 5], "slow": [10, 20]},
|
|
"limit": 200,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["combinations"] == 4
|
|
assert 1 <= len(data["rows"]) <= 4
|
|
assert "total_return_pct" in data["rows"][0]["metrics"]
|
|
|
|
|
|
def test_sweep_unknown_strategy_404(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post("/api/sweep", json={"strategy_key": "nope"})
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_backtest_unknown_strategy_404(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post("/api/backtest", json={"strategy_key": "nope"})
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_backtest_limit_validation(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert (
|
|
client.post("/api/backtest", json={"strategy_key": "sma_cross", "limit": 5}).status_code
|
|
== 422
|
|
)
|
|
|
|
|
|
def test_kill_endpoint(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
client.post("/api/bot/start", json={"strategy_key": "sma_cross"})
|
|
killed = client.post("/api/bot/kill", json={"reason": "test kill"}).json()
|
|
assert killed["running"] is False
|
|
assert killed["halted"] is True
|
|
assert killed["halt_reason"] == "test kill"
|
|
|
|
|
|
def test_status_exposes_risk_fields(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
status = client.get("/api/status").json()
|
|
assert {"halted", "halt_reason", "trades_today", "day_pnl", "drawdown_pct"} <= set(status)
|
|
|
|
|
|
def test_preset_crud(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
created = client.post(
|
|
"/api/presets",
|
|
json={
|
|
"name": "My SMA",
|
|
"strategy_key": "sma_cross",
|
|
"symbol": "BTC/USDT",
|
|
"timeframe": "1h",
|
|
"params": {"fast": 5, "slow": 20},
|
|
},
|
|
)
|
|
assert created.status_code == 200
|
|
preset = created.json()
|
|
assert preset["name"] == "My SMA"
|
|
|
|
listed = client.get("/api/presets").json()
|
|
assert any(p["id"] == preset["id"] for p in listed)
|
|
|
|
assert client.delete(f"/api/presets/{preset['id']}").status_code == 200
|
|
assert client.get("/api/presets").json() == []
|
|
|
|
|
|
def test_preset_unknown_strategy_404(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/presets",
|
|
json={
|
|
"strategy_key": "nope",
|
|
"symbol": "BTC/USDT",
|
|
"timeframe": "1m",
|
|
"params": {},
|
|
},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_api_token_enforced(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = "secret"
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert client.get("/api/status").status_code == 401
|
|
ok = client.get("/api/status", headers={"X-API-Token": "secret"})
|
|
assert ok.status_code == 200
|
|
# Health stays open.
|
|
assert client.get("/api/health").status_code == 200
|
|
|
|
|
|
def test_websocket_connects_without_any_token(runtime: Runtime) -> None:
|
|
# API_TOKEN is unset by default: the API and the live feed must work with no
|
|
# token at all, even when (as nginx does) an empty token header is sent.
|
|
runtime.settings.api_token = ""
|
|
app = create_app(runtime.settings, runtime)
|
|
with (
|
|
TestClient(app) as client,
|
|
client.websocket_connect("/api/ws", headers={"X-API-Token": ""}) as websocket,
|
|
):
|
|
message = websocket.receive_json()
|
|
assert message["type"] == "status"
|
|
assert message["data"]["running"] is False
|
|
|
|
|
|
def test_api_works_without_token(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = ""
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client:
|
|
assert client.get("/api/status").status_code == 200
|
|
assert client.get("/api/strategies").status_code == 200
|
|
assert client.post("/api/backtest", json={"strategy_key": "sma_cross"}).status_code != 401
|
|
|
|
|
|
def test_websocket_rejects_missing_token(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = "secret"
|
|
app = create_app(runtime.settings, runtime)
|
|
with (
|
|
TestClient(app) as client,
|
|
pytest.raises(WebSocketDisconnect) as excinfo,
|
|
client.websocket_connect("/api/ws"),
|
|
):
|
|
pass
|
|
assert excinfo.value.code == 1008
|
|
|
|
|
|
def test_websocket_accepts_header_token(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = "secret"
|
|
app = create_app(runtime.settings, runtime)
|
|
with (
|
|
TestClient(app) as client,
|
|
client.websocket_connect("/api/ws", headers={"X-API-Token": "secret"}) as websocket,
|
|
):
|
|
message = websocket.receive_json()
|
|
assert message["type"] == "status"
|
|
|
|
|
|
def test_websocket_accepts_query_token(runtime: Runtime) -> None:
|
|
runtime.settings.api_token = "secret"
|
|
app = create_app(runtime.settings, runtime)
|
|
with (
|
|
TestClient(app) as client,
|
|
client.websocket_connect("/api/ws?token=secret") as websocket,
|
|
):
|
|
message = websocket.receive_json()
|
|
assert message["type"] == "status"
|
|
|
|
|
|
def test_websocket_sends_initial_status(runtime: Runtime) -> None:
|
|
app = create_app(runtime.settings, runtime)
|
|
with TestClient(app) as client, client.websocket_connect("/api/ws") as websocket:
|
|
message = websocket.receive_json()
|
|
assert message["type"] == "status"
|
|
assert message["data"]["running"] is False
|