79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
from app.api.meta import router as meta_router
|
|
from app.api.routes import router as api_router
|
|
from app.api.ws import router as ws_router
|
|
from app.config import Settings
|
|
from app.runtime import Runtime
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path):
|
|
def build(token: str) -> TestClient:
|
|
settings = Settings(
|
|
chart_auth_token=token,
|
|
manual_lines_path=tmp_path / "manual_lines.json",
|
|
)
|
|
app = FastAPI()
|
|
app.include_router(meta_router)
|
|
app.include_router(api_router)
|
|
app.include_router(ws_router)
|
|
app.state.runtime = Runtime(settings)
|
|
return TestClient(app)
|
|
|
|
return build
|
|
|
|
|
|
def test_open_when_no_token_configured(client):
|
|
assert client("").get("/api/bars").status_code == 200
|
|
|
|
|
|
def test_rejects_missing_token(client):
|
|
assert client("s3cret").get("/api/bars").status_code == 401
|
|
|
|
|
|
def test_rejects_wrong_token(client):
|
|
response = client("s3cret").get("/api/bars", headers={"X-Chart-Token": "nope"})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_accepts_header_token(client):
|
|
response = client("s3cret").get("/api/bars", headers={"X-Chart-Token": "s3cret"})
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_accepts_query_token(client):
|
|
assert client("s3cret").get("/api/bars?token=s3cret").status_code == 200
|
|
|
|
|
|
def test_writes_are_protected(client):
|
|
payload = {
|
|
"tf": "1m",
|
|
"side": "support",
|
|
"anchor_t": 1,
|
|
"anchor_p": 1.0,
|
|
"end_t": 2,
|
|
"end_p": 2.0,
|
|
}
|
|
assert client("s3cret").post("/api/lines", json=payload).status_code == 401
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["/api/health", "/api/version"])
|
|
def test_meta_endpoints_stay_open(client, path):
|
|
"""bin/wait-deploy polls /api/version without carrying the token."""
|
|
assert client("s3cret").get(path).status_code == 200
|
|
|
|
|
|
def test_websocket_rejects_missing_token(client):
|
|
with pytest.raises(WebSocketDisconnect) as excinfo:
|
|
with client("s3cret").websocket_connect("/ws"):
|
|
pass
|
|
assert excinfo.value.code == 1008
|
|
|
|
|
|
def test_websocket_accepts_query_token(client):
|
|
with client("s3cret").websocket_connect("/ws?token=s3cret") as socket:
|
|
assert socket.receive_json()["type"] == "snapshot"
|