154 lines
5.7 KiB
Python
154 lines
5.7 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",
|
|
"can_logout": True,
|
|
}
|
|
|
|
|
|
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 test_seed_session_authenticates_requests_without_a_cookie() -> None:
|
|
async def spotify_handler(request: httpx2.Request) -> httpx2.Response:
|
|
if request.url.host == "accounts.spotify.com":
|
|
return httpx2.Response(200, json={"access_token": "seed-access", "expires_in": 3600})
|
|
assert request.url.path == "/v1/me"
|
|
return httpx2.Response(200, json={"id": "seed-account", "display_name": "Seed Listener"})
|
|
|
|
app = create_app(
|
|
application_settings=Settings(
|
|
app_mode=AppMode.LIVE,
|
|
spotify_client_id="client-id",
|
|
anthropic_api_key="test-key",
|
|
spotify_seed_refresh_token="seed-refresh",
|
|
),
|
|
http_transport=httpx2.MockTransport(spotify_handler),
|
|
)
|
|
with TestClient(app) as client:
|
|
response = client.get("/api/auth/me")
|
|
logout_response = client.post("/api/auth/logout")
|
|
after_logout_response = client.get("/api/auth/me")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"display_name": "Seed Listener", "can_logout": False}
|
|
assert logout_response.status_code == 204
|
|
assert after_logout_response.json() == {
|
|
"display_name": "Seed Listener",
|
|
"can_logout": False,
|
|
}
|
|
|
|
|
|
def test_seed_session_failure_keeps_application_serving() -> None:
|
|
async def spotify_handler(request: httpx2.Request) -> httpx2.Response:
|
|
return httpx2.Response(400)
|
|
|
|
app = create_app(
|
|
application_settings=Settings(
|
|
app_mode=AppMode.LIVE,
|
|
spotify_client_id="client-id",
|
|
anthropic_api_key="test-key",
|
|
spotify_seed_refresh_token="seed-refresh",
|
|
),
|
|
http_transport=httpx2.MockTransport(spotify_handler),
|
|
)
|
|
with TestClient(app, follow_redirects=False) as client:
|
|
health_response = client.get("/api/health")
|
|
login_response = client.get("/api/auth/login")
|
|
|
|
assert app.state.seed_session_id is None
|
|
|
|
assert health_response.status_code == 200
|
|
assert login_response.status_code == 307
|
|
|
|
|
|
def _live_settings() -> Settings:
|
|
return Settings(
|
|
app_mode=AppMode.LIVE,
|
|
spotify_client_id="client-id",
|
|
anthropic_api_key="test-key",
|
|
)
|