94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
"""End-to-end route tests for the Spotify login surface."""
|
|
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import httpx2
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import AppMode, Settings
|
|
from app.main import create_app
|
|
|
|
|
|
def test_login_callback_cookie_and_current_user_flow() -> None:
|
|
async def spotify_handler(request: httpx2.Request) -> httpx2.Response:
|
|
if request.url.host == "accounts.spotify.com":
|
|
return httpx2.Response(
|
|
200,
|
|
json={
|
|
"access_token": "access",
|
|
"refresh_token": "refresh",
|
|
"expires_in": 3600,
|
|
},
|
|
)
|
|
assert request.url == "https://api.spotify.com/v1/me"
|
|
return httpx2.Response(
|
|
200,
|
|
json={"account_id": "stable-account", "display_name": "Ada Listener"},
|
|
)
|
|
|
|
app = create_app(
|
|
application_settings=_live_settings(),
|
|
http_transport=httpx2.MockTransport(spotify_handler),
|
|
)
|
|
with TestClient(app, follow_redirects=False) as client:
|
|
unauthenticated_response = client.get("/api/auth/me")
|
|
login_response = client.get("/api/auth/login")
|
|
|
|
authorize_url = urlparse(login_response.headers["location"])
|
|
authorize_query = parse_qs(authorize_url.query)
|
|
state = authorize_query["state"][0]
|
|
|
|
assert unauthenticated_response.status_code == 401
|
|
assert login_response.status_code == 307
|
|
assert authorize_url.netloc == "accounts.spotify.com"
|
|
assert authorize_url.path == "/authorize"
|
|
assert authorize_query["code_challenge"][0]
|
|
assert authorize_query["code_challenge_method"] == ["S256"]
|
|
|
|
callback_response = client.get("/callback", params={"code": "code", "state": state})
|
|
|
|
assert callback_response.status_code == 307
|
|
assert callback_response.headers["location"] == "/"
|
|
assert "discovery_session=" in callback_response.headers["set-cookie"]
|
|
assert "HttpOnly" in callback_response.headers["set-cookie"]
|
|
assert "SameSite=lax" in callback_response.headers["set-cookie"]
|
|
assert client.get("/api/auth/me").json() == {"display_name": "Ada Listener"}
|
|
|
|
|
|
def test_unknown_callback_state_redirects_to_login_error() -> None:
|
|
async def spotify_handler(request: httpx2.Request) -> httpx2.Response:
|
|
raise AssertionError("Spotify must not be called for an unknown state")
|
|
|
|
app = create_app(
|
|
application_settings=_live_settings(),
|
|
http_transport=httpx2.MockTransport(spotify_handler),
|
|
)
|
|
with TestClient(app, follow_redirects=False) as client:
|
|
response = client.get("/callback", params={"code": "code", "state": "unknown"})
|
|
|
|
assert response.status_code == 307
|
|
assert response.headers["location"] == "/?login=error"
|
|
|
|
|
|
def test_spotify_failure_during_callback_redirects_to_login_error() -> None:
|
|
async def spotify_handler(request: httpx2.Request) -> httpx2.Response:
|
|
return httpx2.Response(400)
|
|
|
|
app = create_app(
|
|
application_settings=_live_settings(),
|
|
http_transport=httpx2.MockTransport(spotify_handler),
|
|
)
|
|
with TestClient(app, follow_redirects=False) as client:
|
|
login_response = client.get("/api/auth/login")
|
|
authorize_query = parse_qs(urlparse(login_response.headers["location"]).query)
|
|
response = client.get(
|
|
"/callback",
|
|
params={"code": "code", "state": authorize_query["state"][0]},
|
|
)
|
|
|
|
assert response.status_code == 307
|
|
assert response.headers["location"] == "/?login=error"
|
|
|
|
|
|
def _live_settings() -> Settings:
|
|
return Settings(app_mode=AppMode.LIVE, spotify_client_id="client-id")
|