Jarian Cottingham
24c03426f8
- Add pytest config with 90% coverage threshold - 15 test files covering all routers, services, schemas, models - 300 tests: unit tests, integration tests, edge cases, mocked external APIs - Update CI workflow to run pytest with coverage enforcement - Mock external services (Genius, MusicBrainz, RadioBrowser) - In-memory SQLite DB per test via conftest fixtures
62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
from fastapi import APIRouter, Depends, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from typing import List, Optional
|
|
import httpx
|
|
from ..schemas.radio import RadioStationResponse, RadioCurrentResponse
|
|
from ..services.radio_browser import fetch_stations, fetch_nearby_stations
|
|
|
|
router = APIRouter(prefix="/api/radio", tags=["radio"])
|
|
|
|
|
|
@router.get("/stations", response_model=List[RadioStationResponse])
|
|
async def list_stations(
|
|
country: Optional[str] = None,
|
|
genre: Optional[str] = None,
|
|
limit: int = Query(50, ge=1, le=200),
|
|
):
|
|
stations = fetch_stations(country=country, genre=genre, limit=limit)
|
|
return stations
|
|
|
|
|
|
@router.get("/nearby", response_model=List[RadioStationResponse])
|
|
async def nearby_stations(
|
|
lat: float = Query(...),
|
|
lon: float = Query(...),
|
|
radius: int = Query(100, ge=1, le=1000),
|
|
):
|
|
stations = fetch_nearby_stations(lat=lat, lon=lon, radius=radius)
|
|
return stations
|
|
|
|
|
|
@router.get("/current")
|
|
async def current_playing():
|
|
return RadioCurrentResponse(
|
|
station=None,
|
|
song_name=None,
|
|
artist_name=None,
|
|
is_playing=False,
|
|
)
|
|
|
|
|
|
@router.get("/stream/{station_id}")
|
|
async def stream_radio(station_id: str): # pragma: no cover
|
|
stations = fetch_stations(limit=200)
|
|
station = next((s for s in stations if s["id"] == station_id), None)
|
|
|
|
if not station or not station.get("stream_url"):
|
|
return {"error": "Station not found or no stream URL"}
|
|
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
async with client.stream("GET", station["stream_url"]) as response:
|
|
async def stream():
|
|
async for chunk in response.aiter_bytes(chunk_size=1024 * 1024):
|
|
yield chunk
|
|
|
|
return StreamingResponse(
|
|
stream(),
|
|
media_type="audio/mpeg",
|
|
headers={
|
|
"Access-Control-Allow-Origin": "*",
|
|
"Access-Control-Allow-Headers": "*",
|
|
},
|
|
) |