68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
"""Offline replay tests for demo service adapters."""
|
|
|
|
import asyncio
|
|
|
|
from app.adapters.demo.cassette import load_cassette
|
|
from app.adapters.demo.catalog import DemoCatalog
|
|
from app.adapters.demo.recommender import DemoRecommender
|
|
from app.config import Settings
|
|
|
|
|
|
def test_cassette_decodes_bodies_and_preserves_rerank_chunks() -> None:
|
|
cassette = load_cassette("focus-coding")
|
|
|
|
assert cassette.intent_response_body.startswith(b'{"model"')
|
|
assert len(cassette.spotify_search_responses) > 20
|
|
assert len(cassette.spotify_taste_responses) == 6
|
|
assert len(cassette.rerank_response_chunks) > 1
|
|
assert b"event: content_block_delta" in b"".join(cassette.rerank_response_chunks)
|
|
|
|
|
|
def test_demo_catalog_replays_search_and_synthetic_taste_pages() -> None:
|
|
async def run() -> None:
|
|
catalog = DemoCatalog(load_cassette("focus-coding"))
|
|
|
|
tracks = await catalog.search_tracks('track:"Stay" artist:"Hybrid Minds"')
|
|
artists = await catalog.fetch_top_artists("short_term", 50)
|
|
top_tracks = await catalog.fetch_top_tracks("long_term", 50)
|
|
saved_tracks = await catalog.fetch_saved_tracks(100)
|
|
|
|
assert tracks[0].title == "Stay"
|
|
assert artists[0] == "Synthetic Focus Artist"
|
|
assert {track.title for track in top_tracks} >= {
|
|
"Synthetic Focus Track",
|
|
"Synthetic Jazz Track",
|
|
}
|
|
assert saved_tracks[0].title == "Synthetic Saved Track"
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_demo_recommender_parses_intent_and_streams_recorded_rerank() -> None:
|
|
async def run() -> None:
|
|
recommender = DemoRecommender(Settings(demo_chunk_delay_seconds=0))
|
|
intent = await recommender.create_intent(
|
|
"Focus while coding",
|
|
(),
|
|
(),
|
|
"Synthetic taste",
|
|
35,
|
|
)
|
|
selections = [
|
|
selection
|
|
async for selection in recommender.stream_rerank(
|
|
intent,
|
|
(),
|
|
"Synthetic taste",
|
|
(),
|
|
15,
|
|
)
|
|
]
|
|
|
|
assert intent.activity == "programming"
|
|
assert len(intent.candidates) == 35
|
|
assert len(selections) == 15
|
|
assert selections[0].track_id == "2nIixNuuV5eHCJydG5aYIB"
|
|
assert all(selection.justification for selection in selections)
|
|
|
|
asyncio.run(run())
|