matitos
2a88f55689
The deploy job hard-failed without CRYPTO_API_TOKEN, and /api/ws (which skips Basic auth) relied on that token. Drop the requirement: the app now works with no token at all, matching the backend guard that was always a no-op when unset. When a token is configured it still protects /api and /api/ws; when it is not, /api/ws is as open as the rest of the API, which the Traefik Basic auth still guards for everything else. Adds tests for the tokenless API and WebSocket.
211 lines
7.6 KiB
Python
211 lines
7.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_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
|