"""tilderadio: parsing, caching and the offline fallback. The payloads below are trimmed copies of real ``/api/now/`` and AzuraCast ``/api/nowplaying/`` responses, so the tests keep working when the station is off air. """ from __future__ import annotations import httpx import pytest from beefy_player.providers.tilderadio import TilderadioProvider, api, fallback NOW_PAYLOAD = { "api_version": 1, "status": "ok", "updated_at": 1793741211, "available": True, "station": { "id": 1, "name": "tilderadio", "shortcode": "tilderadio", "listen_url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg", }, "is_live": False, "broadcast_start": None, "dj": None, "listeners": 5, "now_playing": { "artist": None, "title": "souldies-dramatics-feel it", "text": "souldies-dramatics-feel it", "art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/199f03", "played_at": 1793741079, "duration": 192, }, "history": [ { "artist": None, "title": "souldies-everytime i see a pretty lady-gemini", "text": "souldies-everytime i see a pretty lady-gemini", "art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/5e46a9", "played_at": 1793740639, }, { "artist": "TILDE PROMO (station)", "title": "owly (bots 3)", "text": "TILDE PROMO (station) - owly (bots 3)", "art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/54ac9f", "played_at": 1793740629, }, ], } MOUNTS_PAYLOAD = { "station": { "id": 1, "name": "tilderadio", "shortcode": "tilderadio", "listen_url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg", "mounts": [ { "name": "/radio.mp3 (192kbps MP3)", "url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.mp3", "bitrate": 192, "format": "mp3", "is_default": False, }, { "name": "/320k.mp3 (320kbps MP3)", "url": "https://azuracast.tilderadio.org/listen/tilderadio/320k.mp3", "bitrate": 320, "format": "mp3", "is_default": False, }, { "name": "192kbps OGG Vorbis", "url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg", "bitrate": 192, "format": "ogg", "is_default": True, }, ], } } def client_for(*responses: tuple[str, int, object]) -> httpx.AsyncClient: """An httpx client that answers from ``(url_prefix, status, payload)``.""" table = list(responses) def handler(request: httpx.Request) -> httpx.Response: for prefix, status, payload in table: if str(request.url).startswith(prefix): return httpx.Response(status, json=payload) return httpx.Response(404, json={}) return httpx.AsyncClient(transport=httpx.MockTransport(handler)) def test_provider_describes_its_catalogue() -> None: provider = TilderadioProvider() assert provider.id == "tilderadio" assert provider.brand == "TILDE RADIO" assert provider.station_source == "tilderadio.org/api/now/" assert "tilderadio.org" in provider.user_agent stations = provider.fallback_stations() assert len(stations) == 1 station = stations[0] assert station.uid == "tilderadio:tilderadio" assert all("tilderadio.org" in url for url in provider.fallback_urls(station)) # the bundled list still advertises a real bitrate, not a URL guess assert station.format_label == "mp3 · 320k" def test_station_from_now_reads_the_single_station() -> None: station = api.station_from_now(NOW_PAYLOAD, fallback.fallback_playlists()) assert station is not None assert station.id == "tilderadio" assert station.provider == "tilderadio" assert station.listeners == 5 assert station.dj == "" # nobody is on air assert station.genre_label == "community · eclectic" assert station.now_playing is not None assert station.now_playing.title == "souldies-dramatics-feel it" assert station.now_playing.art_url.endswith("199f03") assert station.last_playing == "souldies-dramatics-feel it" def test_live_dj_is_only_named_when_someone_is_on_air() -> None: live = dict(NOW_PAYLOAD, is_live=True, dj="ffog") station = api.station_from_now(live) assert station is not None and station.dj == "ffog" # the AzuraCast flavour of the same field wins when present nested = dict(NOW_PAYLOAD, live={"is_live": True, "name": "controlfreak"}) station = api.station_from_now(nested) assert station is not None and station.dj == "controlfreak" def test_artist_is_split_out_of_the_text_blob() -> None: song = api.song_from_track( { "artist": "TILDE PROMO (station)", "title": "owly (bots 3)", "text": "TILDE PROMO (station) - owly (bots 3)", "played_at": 1793740629, } ) assert song is not None assert (song.artist, song.title) == ("TILDE PROMO (station)", "owly (bots 3)") # no artist field: the "Artist - Title" text has to carry it song = api.song_from_track( {"title": "Kenny Ken", "text": "Kenny Ken - Amazon Jungle Collection 3"} ) assert song is not None assert (song.artist, song.title) == ("Kenny Ken", "Amazon Jungle Collection 3") def test_songs_from_now_puts_the_current_track_first_and_dedupes() -> None: payload = dict(NOW_PAYLOAD) # a feed that repeats the current track in its own history payload["history"] = [NOW_PAYLOAD["now_playing"], *NOW_PAYLOAD["history"]] songs = api.songs_from_now(payload) assert len(songs) == 3 assert songs[0].title == "souldies-dramatics-feel it" assert songs[1].title == "souldies-everytime i see a pretty lady-gemini" played = [song.played_at for song in songs] assert played == sorted(played, reverse=True) def test_mounts_become_playlists_best_first() -> None: playlists = api.playlists_from_mounts(MOUNTS_PAYLOAD) assert [entry["bitrate"] for entry in playlists] == ["192", "320", "192"] station = api.station_from_now( {**NOW_PAYLOAD, "now_playing": None}, playlists ) assert station is not None assert station.format_label == "mp3 · 320k" assert station.sorted_playlists()[0]["url"].endswith("320k.mp3") assert [entry["quality"] for entry in playlists] == ["high", "highest", "high"] def test_station_without_mounts_falls_back_to_the_listen_url() -> None: station = api.station_from_now(NOW_PAYLOAD) assert station is not None assert len(station.playlists) == 1 assert station.playlists[0]["url"].endswith("radio.ogg") assert station.format_label == "ogg · 128k" # guessed, the feed didn't say @pytest.mark.asyncio async def test_fetch_station_caches_what_it_learned() -> None: async with client_for( ("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD), ("https://azuracast.tilderadio.org/api/nowplaying/", 200, MOUNTS_PAYLOAD), ) as client: station = await api.fetch_station(client) assert station.listeners == 5 assert station.format_label == "mp3 · 320k" cached = api.load_cached_station() assert len(cached) == 1 assert cached[0].id == "tilderadio" assert cached[0].now_playing is not None assert cached[0].format_label == "mp3 · 320k" @pytest.mark.asyncio async def test_fetch_station_survives_an_unreachable_azuracast() -> None: async with client_for(("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD)) as client: station = await api.fetch_station(client) # no mounts -> the documented listen_url is still a usable stream assert station.playlists[0]["url"].endswith("radio.ogg") @pytest.mark.asyncio async def test_history_falls_back_to_the_cache_when_the_feed_is_down() -> None: async with client_for( ("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD) ) as client: songs = await api.fetch_songs(client, "tilderadio") assert len(songs) == 3 async with client_for( ("https://tilderadio.org/api/now/", 500, {}) ) as client: cached = await api.fetch_songs(client, "tilderadio") assert [song.title for song in cached] == [song.title for song in songs] def test_station_cache_is_per_provider() -> None: from beefy_player import config assert config.channels_cache_path("tilderadio").name == "channels-tilderadio.json" # SomaFM keeps its historical file name, so upgrades don't lose the cache assert config.channels_cache_path("somafm") == config.CHANNELS_CACHE def test_fallback_station_is_usable_offline() -> None: provider = TilderadioProvider() stations = provider.fallback_stations() assert [station.id for station in stations] == ["tilderadio"] station = stations[0] assert station.listeners == 0 assert station.now_playing is None assert {entry["format"] for entry in station.playlists} == {"mp3", "ogg"}