225 lines
8.2 KiB
Python
225 lines
8.2 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_search_stations_queries_all_searchable_fields() -> None:
|
|
entries = {
|
|
"name": {**ENTRY, "stationuuid": "name", "name": "Jazz Name"},
|
|
"tag": {**ENTRY, "stationuuid": "tag", "name": "Zed Jazz Tag"},
|
|
"country": {**ENTRY, "stationuuid": "country", "name": "Country Jazz"},
|
|
"language": {**ENTRY, "stationuuid": "language", "name": "Language Jazz"},
|
|
}
|
|
seen: list[dict[str, str]] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
params = dict(request.url.params)
|
|
seen.append(params)
|
|
field = next(name for name in api._SEARCH_FIELDS if name in params)
|
|
payload = [entries[field]]
|
|
if field in ("name", "tag"):
|
|
payload.append(ENTRY) # the same station may match twice
|
|
return httpx.Response(200, json=payload)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
stations = await api.search_stations(client, "jazz")
|
|
|
|
assert {station.id for station in stations} == {
|
|
ENTRY["stationuuid"],
|
|
"name",
|
|
"tag",
|
|
"country",
|
|
"language",
|
|
}
|
|
assert [params["limit"] for params in seen] == [
|
|
str(api.SEARCH_LIMIT)
|
|
] * len(api._SEARCH_FIELDS)
|
|
for params in seen:
|
|
assert params["hidebroken"] == "true"
|
|
assert params["order"] == "clickcount"
|
|
assert params["reverse"] == "true"
|
|
assert {next(name for name in api._SEARCH_FIELDS if name in params) for params in seen} == set(
|
|
api._SEARCH_FIELDS
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_stations_ignores_a_failed_field(monkeypatch) -> None:
|
|
async def fake_fetch_json(client, url):
|
|
if "name=jazz" in url:
|
|
raise RuntimeError("name search is down")
|
|
return [ENTRY]
|
|
|
|
monkeypatch.setattr(api, "fetch_json", fake_fetch_json)
|
|
async with httpx.AsyncClient() as client:
|
|
stations = await api.search_stations(client, "jazz")
|
|
assert [station.id for station in stations] == [ENTRY["stationuuid"]]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_empty_search_does_not_hit_the_api() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
raise AssertionError("empty queries must not make a request")
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
assert await api.search_stations(client, " ") == []
|
|
|
|
|
|
@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:
|
|
params = dict(request.url.params)
|
|
assert params["hidebroken"] == "true"
|
|
assert params["limit"] == str(api.PAGE_SIZE)
|
|
assert params["offset"] == "0"
|
|
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] == ["France Info", "Another"]
|
|
assert [station.title for station in provider.load_cached_stations()] == [
|
|
"France Info",
|
|
"Another",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_station_page_continues_in_catalogue_order() -> None:
|
|
first = [
|
|
{**ENTRY, "stationuuid": f"id-{index}", "name": f"Station {index}"}
|
|
for index in range(api.PAGE_SIZE)
|
|
]
|
|
last = [{**ENTRY, "stationuuid": "last", "name": "Last Station"}]
|
|
seen: list[dict[str, str]] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
params = dict(request.url.params)
|
|
seen.append(params)
|
|
return httpx.Response(200, json=first if params["offset"] == "100" else last)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
page = await api.fetch_station_page(client, 100)
|
|
end = await api.fetch_station_page(client, 200)
|
|
|
|
assert [station.id for station in page.stations] == [
|
|
f"id-{index}" for index in range(api.PAGE_SIZE)
|
|
]
|
|
assert page.next_offset == 200
|
|
assert end.next_offset is None
|
|
assert [params["offset"] for params in seen] == ["100", "200"]
|
|
assert [params["limit"] for params in seen] == [str(api.PAGE_SIZE)] * 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_stations_by_ids_uses_the_uuid_endpoint() -> None:
|
|
seen: list[dict[str, str]] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen.append(dict(request.url.params))
|
|
return httpx.Response(200, json=[ENTRY])
|
|
|
|
station_id = ENTRY["stationuuid"]
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
stations = await api.fetch_stations_by_ids(
|
|
client, ["defcon", station_id, station_id]
|
|
)
|
|
|
|
assert [station.id for station in stations] == [station_id]
|
|
assert len(seen) == 1
|
|
assert seen[0]["uuids"] == station_id
|
|
|
|
|
|
@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") == []
|