/
/
/
1"""Tests for the Radio Paradise music provider."""
2
3from __future__ import annotations
4
5from typing import Any, cast
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.enums import ContentType, MediaType, StreamType
10from music_assistant_models.media_items import AudioFormat
11from music_assistant_models.streamdetails import StreamDetails
12
13from music_assistant.controllers.streams.audio import StreamsAudio
14from music_assistant.controllers.streams.constants import (
15 STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY,
16 STREAMDETAILS_INBAND_TITLE_KEY,
17)
18from music_assistant.providers.radioparadise import SUPPORTED_FEATURES
19from music_assistant.providers.radioparadise.helpers import find_song_by_stream_title
20from music_assistant.providers.radioparadise.provider import RadioParadiseProvider
21
22
23def _block() -> dict[str, Any]:
24 """Return a play API block payload with two songs."""
25 return {
26 "sched_time_millis": 1_000_000,
27 "song": {
28 "0": {
29 "artist": "Mary Gauthier",
30 "title": "Mercy Now",
31 "event": "100",
32 "elapsed": 0,
33 "duration": 200_000,
34 "album": "Mercy Now",
35 "year": 2005,
36 "cover": "covers/l/1.jpg",
37 },
38 "1": {
39 "artist": "Mark Ambor",
40 "title": "Good to Be",
41 "event": "101",
42 "elapsed": 200_000,
43 "duration": 180_000,
44 "album": "Good to Be",
45 "year": 2024,
46 "cover": "covers/l/2.jpg",
47 },
48 },
49 }
50
51
52def _stream_details(*, opt_in: bool = False) -> StreamDetails:
53 """Return minimal StreamDetails for the Mellow Mix channel."""
54 return StreamDetails(
55 item_id="1",
56 provider="radioparadise_test",
57 audio_format=AudioFormat(content_type=ContentType.FLAC, channels=2),
58 media_type=MediaType.RADIO,
59 stream_type=StreamType.HTTP,
60 path="http://example.invalid/stream",
61 allow_seek=False,
62 can_seek=False,
63 duration=0,
64 data={STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY: True} if opt_in else None,
65 )
66
67
68@pytest.fixture
69def provider() -> RadioParadiseProvider:
70 """Return a RadioParadiseProvider wired to mocks for unit testing."""
71 manifest = MagicMock()
72 manifest.domain = "radioparadise"
73 manifest.name = "Radio Paradise"
74
75 config = MagicMock()
76 config.instance_id = "radioparadise_test"
77 config.values = {}
78
79 def _get_value(key: str, default: Any = None) -> Any:
80 if key == "log_level":
81 return "GLOBAL"
82 return default
83
84 config.get_value.side_effect = _get_value
85 return RadioParadiseProvider(MagicMock(), manifest, config, SUPPORTED_FEATURES)
86
87
88# ---------------------------------------------------------------------------
89# find_song_by_stream_title
90
91
92def test_find_song_matches_normalized() -> None:
93 """Case and whitespace differences do not prevent a match."""
94 songs = _block()["song"]
95 assert find_song_by_stream_title(songs, "mary gauthier - MERCY now") == songs["0"]
96
97
98def test_find_song_joins_instead_of_splitting_dashed_artist() -> None:
99 """A dash inside the artist name cannot break matching (the ICY string is never split)."""
100 songs = {"0": {"artist": "Jay - Z", "title": "Song A", "event": "1"}}
101 assert find_song_by_stream_title(songs, "Jay - Z - Song A") == songs["0"]
102
103
104def test_find_song_no_match_returns_none() -> None:
105 """Unknown or blank titles resolve to None."""
106 songs = _block()["song"]
107 assert find_song_by_stream_title(songs, "Commercial-free - Listener-supported") is None
108 assert find_song_by_stream_title(songs, " ") is None
109
110
111# ---------------------------------------------------------------------------
112# _update_stream_metadata
113
114
115async def test_icy_title_matched_from_cached_block(provider: RadioParadiseProvider) -> None:
116 """A cached-block hit publishes enriched metadata without any API call."""
117 sd = _stream_details()
118 sd.data = {
119 "block_data": _block(),
120 STREAMDETAILS_INBAND_TITLE_KEY: "Mary Gauthier - Mercy Now",
121 }
122 provider._fetch_json = AsyncMock() # type: ignore[method-assign]
123
124 await provider._update_stream_metadata(sd, 0)
125
126 provider._fetch_json.assert_not_awaited()
127 assert sd.stream_metadata is not None
128 assert sd.stream_metadata.title == "Mercy Now"
129 assert sd.stream_metadata.artist == "Mary Gauthier"
130
131 # The second tick alternates the artist slot to the upcoming-track info.
132 await provider._update_stream_metadata(sd, 10)
133 assert sd.stream_metadata.artist == "Up Next: Mark Ambor - Good to Be"
134
135
136async def test_block_refetch_is_validated_and_cached(provider: RadioParadiseProvider) -> None:
137 """On cache miss the block is fetched, accepted when it contains the title, and cached."""
138 sd = _stream_details()
139 sd.data = {STREAMDETAILS_INBAND_TITLE_KEY: "Mark Ambor - Good to Be"}
140 block = _block()
141 provider._fetch_json = AsyncMock(return_value=block) # type: ignore[method-assign]
142
143 await provider._update_stream_metadata(sd, 0)
144
145 provider._fetch_json.assert_awaited_once()
146 assert sd.data is not None
147 assert sd.data["block_data"] is block
148 assert sd.stream_metadata is not None
149 assert sd.stream_metadata.title == "Good to Be"
150
151
152async def test_unmatched_title_shows_verbatim_and_retries(
153 provider: RadioParadiseProvider,
154) -> None:
155 """A title missing from a (stale/wobbling) block shows verbatim; lookups keep retrying."""
156 sd = _stream_details()
157 sd.data = {
158 "block_data": _block(),
159 STREAMDETAILS_INBAND_TITLE_KEY: "Madrugada - This Old House",
160 }
161 stale_block = _block() # does not contain the title
162 provider._fetch_json = AsyncMock(return_value=stale_block) # type: ignore[method-assign]
163
164 await provider._update_stream_metadata(sd, 0)
165
166 assert sd.stream_metadata is not None
167 assert sd.stream_metadata.title == "Madrugada - This Old House"
168 assert sd.stream_metadata.artist is None
169
170 # Second tick with the same unmatched title: re-fetch, but no re-publish.
171 sd.stream_metadata = None
172 await provider._update_stream_metadata(sd, 10)
173 assert provider._fetch_json.await_count == 2
174 assert sd.stream_metadata is None
175
176
177async def test_no_icy_title_falls_back_to_schedule_guess(
178 provider: RadioParadiseProvider,
179) -> None:
180 """Without an in-band title the legacy schedule-derived lookup is used."""
181 sd = _stream_details()
182 block = _block()
183 provider._get_channel_metadata = AsyncMock( # type: ignore[method-assign]
184 return_value={
185 "current": block["song"]["0"],
186 "next": block["song"]["1"],
187 "block_data": block,
188 }
189 )
190
191 await provider._update_stream_metadata(sd, 0)
192
193 provider._get_channel_metadata.assert_awaited_once()
194 assert sd.stream_metadata is not None
195 assert sd.stream_metadata.title == "Mercy Now"
196
197
198async def test_get_stream_details_opts_in_and_seeds_block_cache(
199 provider: RadioParadiseProvider,
200) -> None:
201 """Initial enriched metadata opts in and keeps its block for ICY matching."""
202 block = _block()
203 provider._get_channel_metadata = AsyncMock( # type: ignore[method-assign]
204 return_value={
205 "current": block["song"]["0"],
206 "next": block["song"]["1"],
207 "block_data": block,
208 }
209 )
210
211 details = await provider.get_stream_details("1", MediaType.RADIO)
212
213 assert isinstance(details.path, str)
214 assert details.path.endswith("mellow-flacm")
215 assert details.stream_metadata_update_callback == provider._update_stream_metadata
216 assert details.stream_metadata_update_interval == 10
217 assert details.data is not None
218 assert details.data[STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY] is True
219 assert details.data["block_data"] is block
220 assert details.stream_metadata is not None
221 assert details.stream_metadata.title == "Mercy Now"
222
223
224async def test_controller_to_provider_inband_handoff(
225 provider: RadioParadiseProvider,
226) -> None:
227 """The controller's captured ICY title is consumed by Radio Paradise enrichment."""
228 sd = _stream_details(opt_in=True)
229 audio = StreamsAudio(MagicMock())
230 audio_mass = cast("Any", audio.mass)
231 audio_mass.metadata.normalize_radio_artist_name.side_effect = lambda artist: artist
232 audio_mass.metadata.get_radio_stream_station_image.return_value = None
233 audio._parse_icy_metadata(b"StreamTitle='Mary Gauthier - Mercy Now';", sd)
234 assert sd.data is not None
235 sd.data["block_data"] = _block()
236
237 await provider._update_stream_metadata(sd, 0)
238
239 assert sd.stream_metadata is not None
240 assert sd.stream_metadata.title == "Mercy Now"
241 assert sd.stream_metadata.artist == "Mary Gauthier"
242
243
244async def test_unmatched_title_then_match_restarts_artist_view(
245 provider: RadioParadiseProvider,
246) -> None:
247 """A station-break title followed by a match resets the upcoming alternation."""
248 sd = _stream_details(opt_in=True)
249 assert sd.data is not None
250 sd.data["block_data"] = _block()
251 provider._fetch_json = AsyncMock(return_value=_block()) # type: ignore[method-assign]
252
253 sd.data[STREAMDETAILS_INBAND_TITLE_KEY] = "Station break"
254 await provider._update_stream_metadata(sd, 0)
255 assert sd.stream_metadata is not None
256 assert sd.stream_metadata.title == "Station break"
257
258 sd.data[STREAMDETAILS_INBAND_TITLE_KEY] = "Mark Ambor - Good to Be"
259 await provider._update_stream_metadata(sd, 10)
260
261 assert sd.stream_metadata is not None
262 assert sd.stream_metadata.title == "Good to Be"
263 assert sd.stream_metadata.artist == "Mark Ambor"
264 assert sd.data is not None
265 assert "last_verbatim_title" not in sd.data
266 assert sd.data["show_upcoming"] is True
267