114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
"""Radio Browser: a directory of stations, not a station."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from beefy_player.providers.radiobrowser import RadioBrowserProvider, api, fallback
|
|
|
|
ENTRY = {
|
|
"stationuuid": "96062a7b-0601-11e8-ae97-52543be04c81",
|
|
"name": "France Info",
|
|
"url_resolved": "http://direct.franceinfo.fr/live/franceinfo-midfi.mp3",
|
|
"url": "http://direct.franceinfo.fr/live/franceinfo-midfi.mp3",
|
|
"country": "France",
|
|
"language": "french",
|
|
"tags": "news,talk,politic",
|
|
"codec": "MP3",
|
|
"bitrate": 128,
|
|
"favicon": "https://www.francetvinfo.fr/icon.png",
|
|
}
|
|
|
|
|
|
def test_an_entry_becomes_a_station() -> None:
|
|
station = api.station_from_entry(ENTRY)
|
|
assert station is not None
|
|
assert station.id == ENTRY["stationuuid"]
|
|
assert station.title == "France Info"
|
|
assert station.description == "France · french"
|
|
assert station.genre_label == "news · talk · politic"
|
|
assert station.image.endswith("icon.png")
|
|
assert station.format_label == "mp3 · 128k"
|
|
assert station.provider == "radiobrowser"
|
|
|
|
|
|
def test_entries_without_a_stream_or_a_name_are_skipped() -> None:
|
|
assert api.station_from_entry({**ENTRY, "url_resolved": "", "url": ""}) is None
|
|
assert api.station_from_entry({**ENTRY, "name": " "}) is None
|
|
assert api.station_from_entry({**ENTRY, "stationuuid": ""}) is None
|
|
assert api.station_from_entry("not a dict") is None
|
|
|
|
|
|
def test_parse_dedupes_and_sorts_by_name() -> None:
|
|
payload = [
|
|
{**ENTRY, "stationuuid": "b", "name": "Zed FM"},
|
|
{**ENTRY, "stationuuid": "a", "name": "Alpha Radio"},
|
|
{**ENTRY, "stationuuid": "a", "name": "Alpha Radio"},
|
|
]
|
|
stations = api.parse_stations(payload)
|
|
assert [station.title for station in stations] == ["Alpha Radio", "Zed FM"]
|
|
assert api.parse_stations(None) == []
|
|
assert api.parse_stations({}) == []
|
|
|
|
|
|
def test_hls_streams_survive_as_direct_urls() -> None:
|
|
hls = {
|
|
**ENTRY,
|
|
"stationuuid": "hls",
|
|
"url_resolved": "https://example.test/live/index.m3u8?token=1",
|
|
"codec": "AAC",
|
|
"bitrate": 64,
|
|
}
|
|
station = api.station_from_entry(hls)
|
|
assert station is not None
|
|
assert station.playlists[0]["format"] == "aac"
|
|
assert station.format_label == "aac · 64k"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_stations_caches_the_directory() -> None:
|
|
payload = [ENTRY, {**ENTRY, "stationuuid": "x", "name": "Another"}]
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert "hidebroken=true" in str(request.url)
|
|
assert "limit=300" in str(request.url)
|
|
return httpx.Response(200, json=payload)
|
|
|
|
provider = RadioBrowserProvider()
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
stations = await provider.fetch_stations(client)
|
|
assert [station.title for station in stations] == ["Another", "France Info"]
|
|
assert [station.title for station in provider.load_cached_stations()] == [
|
|
"Another",
|
|
"France Info",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_failed_fetch_does_not_kill_the_catalogue() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(500, json={})
|
|
|
|
provider = RadioBrowserProvider()
|
|
with pytest.raises(RuntimeError):
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
await provider.fetch_stations(client)
|
|
|
|
|
|
def test_the_bundled_directory_works_offline() -> None:
|
|
providers = fallback.fallback_stations()
|
|
assert len(providers) == 10
|
|
for station in providers:
|
|
assert station.playlists and station.provider == "radiobrowser"
|
|
assert station.playlists[0]["url"].startswith("https://")
|
|
|
|
|
|
def test_the_directory_offers_no_history_and_no_schedule() -> None:
|
|
provider = RadioBrowserProvider()
|
|
assert provider.has_schedule is False
|
|
assert provider.bandcamp is False
|
|
assert len(provider.hero) == 3
|
|
assert provider.hero_lines[0] == "RADIO BROWSER"
|
|
assert provider.load_cached_history("anything") == []
|