matitos
3e19612cc0
Browsers don't reliably replay Basic-auth credentials on a WebSocket upgrade, so the dashboard's live feed was stuck on "reconnecting" (only the 5s REST poll worked). Give /api/ws a higher-priority Traefik router that skips Basic auth; the backend still requires API_TOKEN, which nginx injects on the handshake. The deploy now fails closed unless CRYPTO_API_TOKEN is set. Also: split the nginx /api/ws block so upgrade headers are only set there, and add reconnect backoff (2s -> 30s) so a failing socket stops hammering.
188 lines
6.6 KiB
Python
188 lines
6.6 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_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_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_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
|