Chris Amow
cc25871032
The live socket never made a REST call, so the seven-day refresh token expired while the chart still looked fine. A deploy then could not log in. Ping user preferences every six hours, and when the grant is already dead offer a one-click reconnect that writes the token on the existing callback.
64 lines
2 KiB
Python
64 lines
2 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.schwab_auth import router
|
|
|
|
|
|
def client() -> TestClient:
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_callback_needs_no_chart_token():
|
|
# Schwab redirects a browser here and cannot attach the token, so this
|
|
# endpoint has to stay open the way /health and /version do.
|
|
assert client().get("/api/qt").status_code == 200
|
|
|
|
|
|
def test_page_does_not_name_the_brokerage():
|
|
# The path is neutral so the host does not advertise who it trades with;
|
|
# the page saying it anyway would defeat that.
|
|
assert "chwab" not in client().get("/api/qt").text
|
|
|
|
|
|
def test_landing_here_directly_explains_itself():
|
|
body = client().get("/api/qt").text
|
|
assert "Register this exact URL" in body
|
|
assert "code" not in body.split("<style>")[0]
|
|
|
|
|
|
def test_authorisation_code_is_echoed_for_the_manual_flow():
|
|
response = client().get("/api/qt", params={"code": "abc123", "session": "s"})
|
|
assert "abc123" in response.text
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
def test_the_code_is_not_retained_for_a_later_visitor():
|
|
session = client()
|
|
session.get("/api/qt", params={"code": "secret-code"})
|
|
# A second, code-less request must not replay the first one's code.
|
|
assert "secret-code" not in session.get("/api/qt").text
|
|
|
|
|
|
def test_a_pending_login_exchanges_the_code_and_returns_home(monkeypatch):
|
|
from app.config import Settings
|
|
from app.runtime import Runtime
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
runtime = Runtime(Settings())
|
|
runtime.schwab_login = object()
|
|
finished = []
|
|
|
|
def finish(url):
|
|
finished.append(url)
|
|
|
|
runtime.finish_schwab_login = finish
|
|
app.state.runtime = runtime
|
|
response = TestClient(app, follow_redirects=False).get(
|
|
"/api/qt", params={"code": "abc", "state": "s"}
|
|
)
|
|
assert response.status_code in (302, 303, 307)
|
|
assert response.headers["location"] == "/"
|
|
assert finished and "code=abc" in finished[0]
|