Chris Amow
e3aad01baf
The callback is registered with the provider and has to answer an unauthenticated browser redirect. It is exempt by construction — a separate router without the token dependency — but nothing held that in place, and the failure would only appear in production, where the token is the one setting that differs from local, at the last step of a login flow. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
104 lines
3.4 KiB
Python
104 lines
3.4 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.schwab_auth import router as schwab_auth_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(schwab_auth_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_oauth_callback_stays_open_when_a_token_is_set():
|
|
# The provider redirects a browser here and cannot attach the chart token.
|
|
# A 401 would break the login flow at its last step, on production only,
|
|
# where CHART_AUTH_TOKEN is the one thing that differs from local.
|
|
from fastapi import FastAPI as _FastAPI
|
|
from app.config import Settings as _Settings
|
|
from app.runtime import Runtime as _Runtime
|
|
import tempfile, pathlib as _pathlib
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
app = _FastAPI()
|
|
app.include_router(meta_router)
|
|
app.include_router(schwab_auth_router)
|
|
app.include_router(api_router)
|
|
app.state.runtime = _Runtime(
|
|
_Settings(chart_auth_token="s3cret",
|
|
manual_lines_path=_pathlib.Path(tmp) / "manual_lines.json")
|
|
)
|
|
probe = TestClient(app)
|
|
assert probe.get("/api/qt").status_code == 200
|
|
assert probe.get("/api/status").status_code == 401
|
|
|
|
|
|
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"
|