78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""Tests for strict live stream property evaluation."""
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.responses import StreamingResponse
|
|
from live import LiveLimits, run_live_scenario
|
|
from scenario import Scenario
|
|
from wire import validate_event
|
|
|
|
|
|
def test_live_scenario_accepts_a_valid_property_stream() -> None:
|
|
async def run() -> None:
|
|
app = FastAPI()
|
|
|
|
@app.post("/api/recommendations")
|
|
async def recommendations() -> StreamingResponse:
|
|
events = [
|
|
{
|
|
"type": "metadata",
|
|
"request_id": "request",
|
|
"intent_summary": "Calm music for coding with a familiar discovery mix.",
|
|
"candidate_count": 10,
|
|
},
|
|
{
|
|
"type": "track",
|
|
"rank": 1,
|
|
"track": {
|
|
"id": "track",
|
|
"uri": "spotify:track:track",
|
|
"title": "Track",
|
|
"artists": ["Artist"],
|
|
"album_name": "Album",
|
|
"album_art_url": None,
|
|
"external_url": None,
|
|
},
|
|
"justification": "A focused fit.",
|
|
},
|
|
{"type": "done", "track_count": 1, "total_ms": 1},
|
|
]
|
|
body = "".join(json.dumps(event) + "\n" for event in events)
|
|
return StreamingResponse(iter((body,)), media_type="application/x-ndjson")
|
|
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
result = await run_live_scenario(
|
|
client,
|
|
Scenario(
|
|
"focus-coding",
|
|
"Focus while coding",
|
|
"calm coding music",
|
|
{"activity": "coding", "familiarity": "mix"},
|
|
),
|
|
LiveLimits(1, 1, 1, 1_000, 1_000),
|
|
)
|
|
|
|
assert result.passed
|
|
assert result.checks["event_order"]
|
|
assert result.checks["done_track_count"]
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_wire_validation_rejects_extra_fields() -> None:
|
|
line = json.dumps(
|
|
{
|
|
"type": "done",
|
|
"track_count": 1,
|
|
"total_ms": 1,
|
|
"unexpected": True,
|
|
}
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="fields do not match"):
|
|
validate_event(line)
|