126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
"""HTTP contract tests for recommendation and playlist routes."""
|
|
|
|
import time
|
|
from collections.abc import AsyncGenerator
|
|
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import TypeAdapter
|
|
|
|
from app.adapters.spotify.auth import TokenSet
|
|
from app.adapters.spotify.session import SessionStore, SpotifySession
|
|
from app.api.routes import SESSION_COOKIE_NAME
|
|
from app.api.schemas import StreamEvent
|
|
from app.domain.models import ConversationTurn, CreatedPlaylist, PreviousRecommendation, Track
|
|
from app.main import create_app
|
|
from app.pipeline.event import PipelineDoneEvent, PipelineMetadataEvent, PipelineTrackEvent
|
|
from app.ports.protocols import MusicCatalog
|
|
|
|
|
|
class FakePipeline:
|
|
"""Emit one complete deterministic stream."""
|
|
|
|
async def stream(
|
|
self,
|
|
session_id: str,
|
|
request_id: str,
|
|
catalog: MusicCatalog,
|
|
query: str,
|
|
history: tuple[ConversationTurn, ...],
|
|
previous_recommendations: tuple[PreviousRecommendation, ...],
|
|
) -> AsyncGenerator[PipelineMetadataEvent | PipelineTrackEvent | PipelineDoneEvent]:
|
|
"""Yield metadata, one track, and completion."""
|
|
yield PipelineMetadataEvent(request_id, "A focused test request.", 35)
|
|
yield PipelineTrackEvent(1, _track(), "It fits the requested focus.")
|
|
yield PipelineDoneEvent(1, 4)
|
|
|
|
|
|
class FakePlaylistWriter:
|
|
"""Capture playlist writes without external calls."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Create an empty write trace."""
|
|
self.name: str | None = None
|
|
self.track_uris: list[str] = []
|
|
|
|
async def create_playlist(self, name: str, description: str) -> CreatedPlaylist:
|
|
"""Record the prefixed name and return a stable playlist."""
|
|
self.name = name
|
|
return CreatedPlaylist("playlist", "https://open.spotify.com/playlist/playlist")
|
|
|
|
async def add_tracks_to_playlist(self, playlist_id: str, track_uris: list[str]) -> None:
|
|
"""Record the ordered track URIs."""
|
|
self.track_uris = track_uris
|
|
|
|
|
|
def test_recommendations_stream_lines_validate_against_frozen_schemas() -> None:
|
|
app = create_app()
|
|
with TestClient(app) as client:
|
|
_authenticate(client, session_store=app.state.session_store)
|
|
app.state.recommendation_pipeline = FakePipeline()
|
|
|
|
response = client.post(
|
|
"/api/recommendations",
|
|
json={"schema_version": 1, "query": "focused electronic music"},
|
|
)
|
|
|
|
adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent)
|
|
events = [adapter.validate_json(line) for line in response.text.splitlines()]
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/x-ndjson"
|
|
assert [event.type for event in events] == ["metadata", "track", "done"]
|
|
|
|
|
|
def test_recommendations_require_a_valid_session_without_seed() -> None:
|
|
with TestClient(create_app()) as client:
|
|
response = client.post(
|
|
"/api/recommendations",
|
|
json={"schema_version": 1, "query": "focused electronic music"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert response.json() == {"detail": "Not authenticated"}
|
|
|
|
|
|
def test_playlist_endpoint_prefixes_name_and_adds_tracks() -> None:
|
|
app = create_app()
|
|
writer = FakePlaylistWriter()
|
|
with TestClient(app) as client:
|
|
_authenticate(client, session_store=app.state.session_store)
|
|
app.state.spotify_client_factory = lambda session: writer
|
|
|
|
response = client.post(
|
|
"/api/playlists",
|
|
json={
|
|
"schema_version": 1,
|
|
"name": "Night drive",
|
|
"track_uris": ["spotify:track:track"],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"url": "https://open.spotify.com/playlist/playlist"}
|
|
assert writer.name == "discovery-by-llm Night drive"
|
|
assert writer.track_uris == ["spotify:track:track"]
|
|
|
|
|
|
def _authenticate(client: TestClient, session_store: SessionStore) -> None:
|
|
session_id = session_store.create(
|
|
SpotifySession(
|
|
tokens=TokenSet("access", "refresh", time.monotonic() + 3600),
|
|
account_id="account",
|
|
display_name="Listener",
|
|
)
|
|
)
|
|
client.cookies.set(SESSION_COOKIE_NAME, session_id)
|
|
|
|
|
|
def _track() -> Track:
|
|
return Track(
|
|
id="track",
|
|
uri="spotify:track:track",
|
|
title="Test Track",
|
|
artists=("Test Artist",),
|
|
album_name="Test Album",
|
|
album_art_url=None,
|
|
external_url="https://open.spotify.com/track/track",
|
|
)
|