Serve /api/ws without Basic auth so the live feed connects
CI/CD / backend (pull_request) Successful in 5m58s
CI/CD / frontend (pull_request) Successful in 5m3s
CI/CD / build (pull_request) Skipped
CI/CD / deploy (pull_request) Skipped
CI/CD / cleanup (pull_request) Skipped

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.
This commit is contained in:
matitos committed 2026-09-19 12:52:06 +00:00
1 parent 97fbd5901e
commit 3e19612cc0
7 files changed
+101 -4

No files matched your search

+36
View File
@@ -1,6 +1,8 @@
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
@@ -143,6 +145,40 @@ def test_api_token_enforced(runtime: Runtime) -> None:
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: