music-assistant-server

1"""
2Integration tests for dynamic radio stations on the RadioController.
3
4Uses a fully booted MusicAssistant instance (mirrors test_radio_export_import.py) with a
5small fake music provider that owns one dynamic and one non-dynamic radio station, to
6verify the guards and track-feed wiring introduced for dynamic radio stations (mirrors
7the existing dynamic-playlist wiring on PlaylistController/media_resolver).
8"""
9
10from __future__ import annotations
11
12from datetime import UTC, datetime
13from typing import TYPE_CHECKING, cast
14
15import pytest
16from music_assistant_models.config_entries import ProviderConfig
17from music_assistant_models.enums import MediaType, ProviderFeature, ProviderType
18from music_assistant_models.errors import UnsupportedFeaturedException
19from music_assistant_models.media_items import ProviderMapping, Radio, SearchResults, Track
20from music_assistant_models.provider import ProviderManifest
21
22from music_assistant.controllers.music.media.radio import RadioController
23from music_assistant.mass import MusicAssistant
24from music_assistant.models.music_provider import MusicProvider
25
26if TYPE_CHECKING:
27    from collections.abc import AsyncGenerator
28
29FAKE_DOMAIN = "fake_dynamic_radio"
30FAKE_INSTANCE = "fake_dynamic_radio--instance"
31DYNAMIC_STATION_ID = "dynamic-1"
32STATIC_STATION_ID = "static-1"
33TOGGLE_STATION_ID = "toggle-1"
34
35
36class FakeDynamicRadioProvider(MusicProvider):
37    """Streaming-style provider owning one dynamic and one static radio station."""
38
39    # controls the is_dynamic value get_library_radios reports for TOGGLE_STATION_ID
40    toggle_station_is_dynamic: bool = False
41    toggle_station_date_added: datetime | None = None
42
43    async def sync_library(self, media_type: MediaType) -> None:
44        """No-op sync implementation for tests."""
45
46    async def get_library_radios(self) -> AsyncGenerator[Radio]:
47        """Yield the single toggle station, honoring the current is_dynamic flag."""
48        yield Radio(
49            item_id=TOGGLE_STATION_ID,
50            provider=self.instance_id,
51            name="Toggle Station",
52            is_dynamic=self.toggle_station_is_dynamic,
53            date_added=self.toggle_station_date_added,
54            provider_mappings={
55                ProviderMapping(
56                    item_id=TOGGLE_STATION_ID,
57                    provider_domain=self.domain,
58                    provider_instance=self.instance_id,
59                )
60            },
61        )
62
63    async def get_radio(self, prov_radio_id: str) -> Radio:
64        """Return the requested fake station."""
65        is_dynamic = prov_radio_id == DYNAMIC_STATION_ID
66        return Radio(
67            item_id=prov_radio_id,
68            provider=self.instance_id,
69            name="Dynamic Station" if is_dynamic else "Static Station",
70            is_dynamic=is_dynamic,
71            provider_mappings={
72                ProviderMapping(
73                    item_id=prov_radio_id,
74                    provider_domain=self.domain,
75                    provider_instance=self.instance_id,
76                )
77            },
78        )
79
80    async def get_dynamic_radio_tracks(self, prov_radio_id: str) -> list[Track]:
81        """Return a fixed batch of fake tracks for the dynamic station."""
82        return [
83            Track(
84                item_id=f"{prov_radio_id}-t{i}",
85                provider=self.instance_id,
86                name=f"Track {i}",
87                provider_mappings={
88                    ProviderMapping(
89                        item_id=f"{prov_radio_id}-t{i}",
90                        provider_domain=self.domain,
91                        provider_instance=self.instance_id,
92                    )
93                },
94            )
95            for i in range(3)
96        ]
97
98    async def search(
99        self, search_query: str, media_types: list[MediaType], limit: int = 5
100    ) -> SearchResults:
101        """Return a same-named station, so a name-only match would find something to report."""
102        return SearchResults(
103            radio=[
104                Radio(
105                    item_id="same-name-elsewhere",
106                    provider=self.instance_id,
107                    name=search_query,
108                    provider_mappings={
109                        ProviderMapping(
110                            item_id="same-name-elsewhere",
111                            provider_domain=self.domain,
112                            provider_instance=self.instance_id,
113                        )
114                    },
115                )
116            ]
117        )
118
119
120@pytest.fixture(name="radio_mass")
121async def radio_mass_fixture(mass: MusicAssistant) -> AsyncGenerator[MusicAssistant]:
122    """Return a booted instance with the fake dynamic-radio provider registered."""
123    config = ProviderConfig(
124        values={},
125        type=ProviderType.MUSIC,
126        domain=FAKE_DOMAIN,
127        instance_id=FAKE_INSTANCE,
128        name="Fake Dynamic Radio",
129    )
130    provider = FakeDynamicRadioProvider(
131        mass,
132        manifest=ProviderManifest(
133            type=ProviderType.MUSIC,
134            domain=FAKE_DOMAIN,
135            name="Fake Dynamic Radio",
136            description="Fake dynamic radio provider",
137            codeowners=["@music-assistant"],
138        ),
139        config=config,
140        supported_features={ProviderFeature.LIBRARY_RADIOS, ProviderFeature.SEARCH},
141    )
142    provider.available = True
143    mass._providers[FAKE_INSTANCE] = provider
144    try:
145        yield mass
146    finally:
147        mass._providers.pop(FAKE_INSTANCE, None)
148
149
150@pytest.fixture(name="radio_ctrl")
151def radio_ctrl_fixture(radio_mass: MusicAssistant) -> RadioController:
152    """Get the radio controller from the booted Music Assistant instance."""
153    return radio_mass.music.radio
154
155
156class TestRadioTracks:
157    """Tests for RadioController.radio_tracks."""
158
159    async def test_raises_for_non_dynamic_station(self, radio_ctrl: RadioController) -> None:
160        """A non-dynamic (live-stream) station has no track feed."""
161        with pytest.raises(UnsupportedFeaturedException):
162            await radio_ctrl.radio_tracks(STATIC_STATION_ID, FAKE_INSTANCE)
163
164    async def test_returns_provider_tracks_for_provider_item(
165        self, radio_ctrl: RadioController
166    ) -> None:
167        """A dynamic station's tracks are fetched straight from its provider."""
168        tracks = await radio_ctrl.radio_tracks(DYNAMIC_STATION_ID, FAKE_INSTANCE)
169        assert [track.name for track in tracks] == ["Track 0", "Track 1", "Track 2"]
170
171    async def test_returns_provider_tracks_for_library_item(
172        self, radio_mass: MusicAssistant, radio_ctrl: RadioController
173    ) -> None:
174        """A library-resolved dynamic station still fetches its tracks from its own provider."""
175        provider = cast("MusicProvider", radio_mass.get_provider(FAKE_INSTANCE))
176        station = await provider.get_radio(DYNAMIC_STATION_ID)
177        library_item = await radio_ctrl.add_item_to_library(station)
178        tracks = await radio_ctrl.radio_tracks(str(library_item.item_id), "library")
179        assert len(tracks) == 3
180
181
182class TestAddToLibraryDynamicGuard:
183    """Tests that adding a dynamic station never folds it into an existing station."""
184
185    async def test_same_named_live_station_is_left_alone(
186        self, radio_mass: MusicAssistant, radio_ctrl: RadioController
187    ) -> None:
188        """A dynamic station gets its own library item instead of joining a same-named stream."""
189        live_station = Radio(
190            item_id="live-1",
191            provider="some_radio_directory--instance",
192            name="Chill Vibes",
193            provider_mappings={
194                ProviderMapping(
195                    item_id="live-1",
196                    provider_domain="some_radio_directory",
197                    provider_instance="some_radio_directory--instance",
198                )
199            },
200        )
201        live_item = await radio_ctrl.add_item_to_library(live_station)
202        station = Radio(
203            item_id=DYNAMIC_STATION_ID,
204            provider=FAKE_INSTANCE,
205            name="Chill Vibes",
206            is_dynamic=True,
207            provider_mappings={
208                ProviderMapping(
209                    item_id=DYNAMIC_STATION_ID,
210                    provider_domain=FAKE_DOMAIN,
211                    provider_instance=FAKE_INSTANCE,
212                )
213            },
214        )
215
216        dynamic_item = await radio_ctrl.add_item_to_library(station)
217
218        assert dynamic_item.item_id != live_item.item_id
219        assert {mapping.provider_domain for mapping in dynamic_item.provider_mappings} == {
220            FAKE_DOMAIN
221        }
222
223
224class TestVersionsDynamicGuard:
225    """Tests for the dynamic-station guard on RadioController.versions."""
226
227    async def test_dynamic_station_has_no_other_versions(self, radio_ctrl: RadioController) -> None:
228        """A dynamic station reports no other versions, and searches nothing to find that out."""
229        assert await radio_ctrl.versions(DYNAMIC_STATION_ID, FAKE_INSTANCE) == []
230
231
232class TestMatchProvidersDynamicGuard:
233    """Tests for the dynamic-station guard on RadioController.match_providers."""
234
235    async def test_dynamic_station_returns_before_searching(
236        self,
237        radio_mass: MusicAssistant,
238        radio_ctrl: RadioController,
239        monkeypatch: pytest.MonkeyPatch,
240    ) -> None:
241        """A dynamic station returns immediately, without inspecting any other provider."""
242        dynamic_radio = Radio(
243            item_id="1",
244            provider="library",
245            name="Dynamic",
246            is_dynamic=True,
247            provider_mappings={
248                ProviderMapping(
249                    item_id=DYNAMIC_STATION_ID,
250                    provider_domain=FAKE_DOMAIN,
251                    provider_instance=FAKE_INSTANCE,
252                )
253            },
254        )
255
256        def _boom(_self: object) -> list[MusicProvider]:
257            raise AssertionError("must not access other providers for a dynamic station")
258
259        monkeypatch.setattr(type(radio_mass.music), "providers", property(_boom))
260        # must not raise: the guard returns before the (patched-to-explode) providers property
261        await radio_ctrl.match_providers(dynamic_radio)
262
263    async def test_non_dynamic_station_still_searches(
264        self,
265        radio_mass: MusicAssistant,
266        radio_ctrl: RadioController,
267        monkeypatch: pytest.MonkeyPatch,
268    ) -> None:
269        """A non-dynamic station is unaffected by the guard and still triggers matching."""
270        static_radio = Radio(
271            item_id="2",
272            provider="library",
273            name="Static",
274            is_dynamic=False,
275            provider_mappings={
276                ProviderMapping(
277                    item_id=STATIC_STATION_ID,
278                    provider_domain=FAKE_DOMAIN,
279                    provider_instance=FAKE_INSTANCE,
280                )
281            },
282        )
283        accessed = False
284        real_providers = list(radio_mass.music.providers)
285
286        def _track_access(_self: object) -> list[MusicProvider]:
287            nonlocal accessed
288            accessed = True
289            return real_providers
290
291        monkeypatch.setattr(type(radio_mass.music), "providers", property(_track_access))
292        await radio_ctrl.match_providers(static_radio)
293        assert accessed is True
294
295
296class TestSyncLibraryRadiosDynamicFlag:
297    """Tests for the is_dynamic branch of MusicProvider._sync_library_radios."""
298
299    @staticmethod
300    def _toggle_mapping() -> ProviderMapping:
301        return ProviderMapping(
302            item_id=TOGGLE_STATION_ID,
303            provider_domain=FAKE_DOMAIN,
304            provider_instance=FAKE_INSTANCE,
305        )
306
307    async def test_switching_to_dynamic_updates_library_item(
308        self, radio_mass: MusicAssistant, radio_ctrl: RadioController
309    ) -> None:
310        """A provider reporting the same station as dynamic updates the stored flag."""
311        provider = cast("FakeDynamicRadioProvider", radio_mass.get_provider(FAKE_INSTANCE))
312        provider.toggle_station_is_dynamic = False
313        await provider._sync_library_radios()
314        library_item = await radio_ctrl.get_library_item_by_prov_mappings([self._toggle_mapping()])
315        assert library_item is not None
316        assert library_item.is_dynamic is False
317
318        provider.toggle_station_is_dynamic = True
319        await provider._sync_library_radios()
320        library_item = await radio_ctrl.get_library_item_by_prov_mappings([self._toggle_mapping()])
321        assert library_item is not None
322        assert library_item.is_dynamic is True
323
324    async def test_switching_to_dynamic_keeps_the_station_listed(
325        self, radio_mass: MusicAssistant, radio_ctrl: RadioController
326    ) -> None:
327        """A station that becomes dynamic stays in the library listing."""
328        provider = cast("FakeDynamicRadioProvider", radio_mass.get_provider(FAKE_INSTANCE))
329        provider.toggle_station_is_dynamic = False
330        await provider._sync_library_radios()
331        assert "Toggle Station" in [item.name for item in await radio_ctrl.library_items()]
332
333        provider.toggle_station_is_dynamic = True
334        await provider._sync_library_radios()
335
336        assert "Toggle Station" in [item.name for item in await radio_ctrl.library_items()]
337
338    async def test_switching_to_dynamic_drops_name_matched_mappings(
339        self, radio_mass: MusicAssistant, radio_ctrl: RadioController
340    ) -> None:
341        """Only the owning provider is left to serve a station's tracks once it becomes dynamic."""
342        provider = cast("FakeDynamicRadioProvider", radio_mass.get_provider(FAKE_INSTANCE))
343        provider.toggle_station_is_dynamic = False
344        await provider._sync_library_radios()
345        library_item = await radio_ctrl.get_library_item_by_prov_mappings([self._toggle_mapping()])
346        assert library_item is not None
347        # a same-named station on another provider, as cross-provider matching would have added
348        await radio_ctrl.add_provider_mappings(
349            library_item.item_id,
350            [
351                ProviderMapping(
352                    item_id="unrelated-stream",
353                    provider_domain="some_radio_directory",
354                    provider_instance="some_radio_directory--instance",
355                )
356            ],
357        )
358
359        provider.toggle_station_is_dynamic = True
360        await provider._sync_library_radios()
361
362        library_item = await radio_ctrl.get_library_item_by_prov_mappings([self._toggle_mapping()])
363        assert library_item is not None
364        assert {mapping.provider_domain for mapping in library_item.provider_mappings} == {
365            FAKE_DOMAIN
366        }
367
368    async def test_dynamic_switch_wins_over_a_coinciding_generic_update(
369        self, radio_mass: MusicAssistant, radio_ctrl: RadioController
370    ) -> None:
371        """A station going dynamic is overwritten even when the same sync has a generic update."""
372        provider = cast("FakeDynamicRadioProvider", radio_mass.get_provider(FAKE_INSTANCE))
373        provider.toggle_station_is_dynamic = False
374        provider.toggle_station_date_added = datetime(2026, 1, 1, tzinfo=UTC)
375        await provider._sync_library_radios()
376        library_item = await radio_ctrl.get_library_item_by_prov_mappings([self._toggle_mapping()])
377        assert library_item is not None
378        await radio_ctrl.add_provider_mappings(
379            library_item.item_id,
380            [
381                ProviderMapping(
382                    item_id="unrelated-stream",
383                    provider_domain="some_radio_directory",
384                    provider_instance="some_radio_directory--instance",
385                )
386            ],
387        )
388
389        # a changed date_added makes _library_item_needs_update true alongside the dynamic switch
390        provider.toggle_station_is_dynamic = True
391        provider.toggle_station_date_added = datetime(2026, 6, 1, tzinfo=UTC)
392        await provider._sync_library_radios()
393
394        library_item = await radio_ctrl.get_library_item_by_prov_mappings([self._toggle_mapping()])
395        assert library_item is not None
396        assert library_item.is_dynamic is True
397        assert {mapping.provider_domain for mapping in library_item.provider_mappings} == {
398            FAKE_DOMAIN
399        }
400
401    async def test_switching_to_non_dynamic_updates_library_item(
402        self, radio_mass: MusicAssistant, radio_ctrl: RadioController
403    ) -> None:
404        """A provider reporting the same station as no longer dynamic updates the stored flag."""
405        provider = cast("FakeDynamicRadioProvider", radio_mass.get_provider(FAKE_INSTANCE))
406        provider.toggle_station_is_dynamic = True
407        await provider._sync_library_radios()
408        library_item = await radio_ctrl.get_library_item_by_prov_mappings([self._toggle_mapping()])
409        assert library_item is not None
410        assert library_item.is_dynamic is True
411
412        provider.toggle_station_is_dynamic = False
413        await provider._sync_library_radios()
414        library_item = await radio_ctrl.get_library_item_by_prov_mappings([self._toggle_mapping()])
415        assert library_item is not None
416        assert library_item.is_dynamic is False
417