from __future__ import annotations import httpx import pytest from beefy_player import config from beefy_player.models import parse_last_playing from beefy_player.providers import base, get_provider from beefy_player.providers.somafm import api SOMAFM = get_provider("somafm") CHANNELS_PAYLOAD = { "channels": [ { "id": "groovesalad", "title": "Groove Salad", "description": "A nicely chilled plate.", "dj": "Rusty Hodge", "genre": "ambient|electronic", "image": "https://api.somafm.com/logos/120/groovesalad120.png", "listeners": "1,873", "lastPlaying": "Jens Buchert - Rauschebart", "playlists": [ {"url": "https://api.somafm.com/groovesalad32.pls", "format": "aacp", "quality": "low"}, {"url": "https://api.somafm.com/groovesalad256.pls", "format": "mp3", "quality": "highest"}, {"url": "https://api.somafm.com/groovesalad130.pls", "format": "aac", "quality": "highest"}, ], }, { "id": "bad", "title": "No playlists", "genre": "test", }, {"title": "missing id"}, ] } def test_parse_channels() -> None: stations = api.parse_channels(CHANNELS_PAYLOAD) assert len(stations) == 2 station = stations[0] assert station.id == "groovesalad" assert station.title == "Groove Salad" assert station.listeners == 1873 assert station.genre_label == "ambient · electronic" assert station.favorite is False assert station.now_playing is not None assert station.now_playing.artist == "Jens Buchert" assert station.now_playing.title == "Rauschebart" assert station.format_label == "mp3 · 256k" assert next(pl["url"] for pl in station.sorted_playlists()).endswith("256.pls") assert stations[1].id == "bad" def test_parse_last_playing() -> None: song = parse_last_playing("Jens Buchert - Rauschebart") assert song is not None and song.artist == "Jens Buchert" and song.title == "Rauschebart" en_dash = parse_last_playing("Tosca – Amber November") assert en_dash is not None and en_dash.artist == "Tosca" plain = parse_last_playing("Just A Title") assert plain is not None and plain.title == "Just A Title" and plain.artist == "" assert parse_last_playing("") is None assert parse_last_playing(None) is None def test_parse_pls() -> None: text = """[playlist] NumberOfEntries=2 File1=https://ice2.somafm.com/groovesalad-256-mp3 File2=https://ice6.somafm.com/groovesalad-256-mp3 Version=2 """ assert base.parse_pls(text) == [ "https://ice2.somafm.com/groovesalad-256-mp3", "https://ice6.somafm.com/groovesalad-256-mp3", ] assert base.parse_pls("https://ice1.somafm.com/x-128-mp3\n# comment\n") == [ "https://ice1.somafm.com/x-128-mp3" ] assert base.is_playlist("https://x/y.pls?foo=1") is True assert base.is_direct_stream("https://ice1.somafm.com/x-128-mp3") is True assert base.is_direct_stream("https://x/y.m3u") is False # HLS is the player's job, not ours: mpv reads the playlist itself assert base.is_playlist("https://x/live/index.m3u8") is False assert base.is_direct_stream("https://x/live/index.m3u8") is True def test_parse_songs_and_dates() -> None: payload = { "id": "groovesalad", "songs": [ { "title": "Rauschebart", "artist": "Jens Buchert", "album": "Gravity", "albumArt": "https://example.com/art.jpg", "date": "1793312765", }, { "title": "Room 505", "artist": "Alex Cortiz", "album": "The Chill Out Room", "albumArt": "", "date": 1793312438, }, ], } songs = api.parse_songs(payload) assert len(songs) == 2 assert songs[0].title == "Rauschebart" assert songs[0].art_url == "https://example.com/art.jpg" assert songs[0].played_at is not None and songs[0].played_at.year == 2026 # newest first assert songs[0].played_at > songs[1].played_at # type: ignore[operator] assert api.parse_songs({"songs": [{"title": ""}]}) == [] def test_history_cache_roundtrip() -> None: path = api._songs_cache_path("groovesalad") config.write_json( path, { "fetched": 1, "songs": [ { "title": "Eva", "artist": "Heights Of Abraham", "album": "Pork", "albumArt": "", "date": "2026-10-29T22:00:00+00:00", } ], }, ) songs = api.load_cached_songs("groovesalad") assert len(songs) == 1 assert songs[0].artist == "Heights Of Abraham" assert api.load_cached_songs("nope") == [] @pytest.mark.asyncio async def test_stream_candidates_with_mock_transport() -> None: pls = "[playlist]\nFile1=https://ice2.somafm.com/x-256-mp3\nFile2=https://ice6.somafm.com/x-256-mp3\n" requests: list[str] = [] def handler(request: httpx.Request) -> httpx.Response: requests.append(str(request.url)) return httpx.Response(200, text=pls) station = api.parse_channels(CHANNELS_PAYLOAD)[0] async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: urls = await SOMAFM.stream_candidates(client, station) assert urls[:2] == [ "https://ice2.somafm.com/x-256-mp3", "https://ice6.somafm.com/x-256-mp3", ] assert len(requests) == 3 # three playlist URLs, cached by URL # second run hits the playlist cache async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: await SOMAFM.stream_candidates(client, station) assert len(requests) == 3 @pytest.mark.asyncio async def test_fetch_channels_and_history_cache() -> None: def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/channels.json": return httpx.Response(200, json=CHANNELS_PAYLOAD) if request.url.path.endswith(".json"): return httpx.Response( 200, json={ "songs": [ {"title": "T", "artist": "A", "album": "B", "date": "1793312765"} ] }, ) return httpx.Response(404) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: stations = await api.fetch_channels(client) assert len(stations) == 2 assert config.CHANNELS_CACHE.exists() assert len(api.load_cached_channels()) == 2 songs = await api.fetch_history(client, "groovesalad") assert len(songs) == 1 assert api.load_cached_songs("groovesalad")[0].title == "T" # a failing request still returns the cached history def failing(request: httpx.Request) -> httpx.Response: return httpx.Response(500) async with httpx.AsyncClient(transport=httpx.MockTransport(failing)) as client: songs = await api.fetch_history(client, "groovesalad") assert songs[0].title == "T" def test_config_roundtrip() -> None: config.Config(favorites=["a"], volume=42, visualizer="wave", autoplay=True).save() loaded = config.Config.load() assert loaded.favorites == ["a"] assert loaded.volume == 42 assert loaded.visualizer == "wave" assert loaded.autoplay is True assert loaded.is_favorite("a") is True assert loaded.toggle_favorite("a") is False assert loaded.toggle_favorite("b") is True assert loaded.favorites == ["b"] def test_channel_cache_missing() -> None: assert api.load_cached_channels() == []