music-assistant-server

1"""
2Integration tests for the RadioController's M3U export/import round trip.
3
4Uses a booted Music Assistant so the real builtin provider performs the library
5writes; that is what makes the stored ``item_id`` assertions meaningful, since
6builtin treats a stored radio item_id as a plain stream URL.
7"""
8
9from __future__ import annotations
10
11import asyncio
12from typing import TYPE_CHECKING
13from unittest.mock import AsyncMock
14
15import pytest
16from music_assistant_models.config_entries import ProviderConfig
17from music_assistant_models.enums import (
18    ImageType,
19    MediaType,
20    ProviderFeature,
21    ProviderType,
22    TaskStatus,
23)
24from music_assistant_models.errors import InvalidDataError
25from music_assistant_models.media_items import (
26    ItemMapping,
27    MediaItemImage,
28    MediaItemType,
29    ProviderMapping,
30    Radio,
31)
32from music_assistant_models.provider import ProviderManifest
33
34from music_assistant.models.music_provider import MusicProvider
35from music_assistant.providers.builtin.constants import CONF_KEY_RADIOS
36
37if TYPE_CHECKING:
38    from collections.abc import AsyncGenerator
39
40    from music_assistant.controllers.tasks import TasksController
41    from music_assistant.mass import MusicAssistant
42
43# the instance sorts before "builtin", so a station mapped to both resolves against this
44# provider and exercises the refetch that add_item_to_library performs for other providers
45FAKE_DOMAIN = "beatstream"
46FAKE_INSTANCE = "beatstream--instance"
47BUILTIN_STREAM_URL = "http://stream.example.com/jazz"
48BUILTIN_IMAGE_URL = "http://img.example.com/jazz.jpg"
49
50
51class FakeRadioProvider(MusicProvider):
52    """Streaming-style provider that owns a single radio station."""
53
54    async def sync_library(self, media_type: MediaType) -> None:
55        """No-op sync implementation for tests."""
56
57    async def get_radio(self, prov_radio_id: str) -> Radio:
58        """Return the station this provider owns."""
59        return _make_radio(
60            item_id=prov_radio_id,
61            provider=FAKE_INSTANCE,
62            domain=FAKE_DOMAIN,
63            name="Provider Owned Radio",
64        )
65
66
67def _make_radio(
68    item_id: str,
69    provider: str,
70    domain: str,
71    name: str,
72    favorite: bool = False,
73    image_url: str | None = None,
74) -> Radio:
75    """Build a Radio as a music provider would hand it over."""
76    radio = Radio(
77        item_id=item_id,
78        provider=provider,
79        name=name,
80        favorite=favorite,
81        provider_mappings={
82            ProviderMapping(
83                item_id=item_id,
84                provider_domain=domain,
85                provider_instance=provider,
86            )
87        },
88    )
89    if image_url:
90        radio.metadata.add_image(
91            MediaItemImage(
92                type=ImageType.THUMB,
93                path=image_url,
94                provider=domain,
95                remotely_accessible=True,
96            )
97        )
98    return radio
99
100
101async def _wait_for_task_status(
102    controller: TasksController,
103    task_id: str,
104    *statuses: TaskStatus,
105    timeout: float = 5.0,
106) -> None:
107    """Wait until a managed task reaches one of the expected statuses."""
108    deadline = asyncio.get_running_loop().time() + timeout
109    while asyncio.get_running_loop().time() < deadline:
110        if controller.get_task(task_id).status in statuses:
111            return
112        await asyncio.sleep(0.01)
113    msg = f"Task {task_id} did not reach {[status.value for status in statuses]} before timeout"
114    raise AssertionError(msg)
115
116
117async def _clear_radio_library(mass: MusicAssistant) -> None:
118    """Empty the radio library and builtin's stored stations, mimicking a clean instance."""
119    for item in await mass.music.radio.library_items(limit=500, summary=False):
120        await mass.music.remove_item_from_library(MediaType.RADIO, item.item_id)
121    mass.config.set(CONF_KEY_RADIOS, [])
122
123
124@pytest.fixture(name="radio_mass")
125async def radio_mass_fixture(
126    mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
127) -> AsyncGenerator[MusicAssistant]:
128    """Return a booted instance with a fake radio provider and no metadata scanning."""
129    config = ProviderConfig(
130        values={},
131        type=ProviderType.MUSIC,
132        domain=FAKE_DOMAIN,
133        instance_id=FAKE_INSTANCE,
134        name="Fake Radio",
135    )
136    monkeypatch.setattr(config, "get_value", lambda *_args, **_kwargs: "GLOBAL")
137    provider = FakeRadioProvider(
138        mass,
139        manifest=ProviderManifest(
140            type=ProviderType.MUSIC,
141            domain=FAKE_DOMAIN,
142            name="Fake Radio",
143            description="Fake radio provider",
144            codeowners=["@music-assistant"],
145        ),
146        config=config,
147        supported_features={ProviderFeature.LIBRARY_RADIOS},
148    )
149    provider.available = True
150    mass._providers[FAKE_INSTANCE] = provider
151    # a full metadata scan is irrelevant here and would reach out to the network
152    mass.metadata.update_metadata = AsyncMock()  # type: ignore[method-assign]
153    try:
154        yield mass
155    finally:
156        mass._providers.pop(FAKE_INSTANCE, None)
157
158
159async def test_export_import_round_trip_restores_stations(radio_mass: MusicAssistant) -> None:
160    """An untouched export must restore names, favourites, artwork and provider ownership."""
161    mass = radio_mass
162    jazz_item = await mass.music.add_item_to_library(
163        _make_radio(
164            item_id=BUILTIN_STREAM_URL,
165            provider="builtin",
166            domain="builtin",
167            name="Jazz FM",
168            image_url=BUILTIN_IMAGE_URL,
169        )
170    )
171    owned_item = await mass.music.add_item_to_library(
172        _make_radio(
173            item_id="station-123",
174            provider=FAKE_INSTANCE,
175            domain=FAKE_DOMAIN,
176            name="Provider Owned Radio",
177        )
178    )
179    # favourite both through the library, the way a user does
180    await mass.music.radio.set_favorite(jazz_item.item_id, True)
181    await mass.music.radio.set_favorite(owned_item.item_id, True)
182
183    m3u_data = await mass.music.radio.export_radios()
184    # the name must travel in an #EXTINF line even though a station has no duration
185    assert "#EXTINF:-1,Jazz FM" in m3u_data
186    assert f"#EXTIMG:thumb||{BUILTIN_IMAGE_URL}||builtin||true" in m3u_data
187    assert f"#EXTPROV:builtin||{BUILTIN_STREAM_URL}||builtin" in m3u_data
188    assert m3u_data.count("favorite=true") == 2
189
190    await _clear_radio_library(mass)
191    assert await mass.music.radio.library_count() == 0
192
193    task = await mass.music.radio.import_radios(m3u_data)
194    await _wait_for_task_status(mass.tasks, task.id, TaskStatus.SUCCESS)
195    assert mass.tasks.get_task(task.id).failure_messages == []
196
197    library = {item.name: item for item in await mass.music.radio.library_items(summary=False)}
198    assert set(library) == {"Jazz FM", "Provider Owned Radio"}
199
200    jazz = library["Jazz FM"]
201    assert jazz.favorite is True
202    assert jazz.image is not None
203    assert jazz.image.path == BUILTIN_IMAGE_URL
204    jazz_mapping = next(iter(jazz.provider_mappings))
205    assert jazz_mapping.provider_domain == "builtin"
206    # builtin's item_id is the raw stream URL, never the builtin://radio/... MA URI
207    assert jazz_mapping.item_id == BUILTIN_STREAM_URL
208
209    # the provider-owned station is restored against its own provider, not as a builtin row
210    owned = library["Provider Owned Radio"]
211    assert {mapping.provider_domain for mapping in owned.provider_mappings} == {FAKE_DOMAIN}
212    assert next(iter(owned.provider_mappings)).item_id == "station-123"
213    # a provider-owned item is refetched from its provider, so the favorite flag has to be
214    # reapplied to the library item rather than set on the object handed to the library
215    assert owned.favorite is True
216
217    stored_radios = mass.config.get(CONF_KEY_RADIOS, [])
218    assert [item["item_id"] for item in stored_radios] == [BUILTIN_STREAM_URL]
219    assert stored_radios[0]["name"] == "Jazz FM"
220    assert stored_radios[0]["image_url"] == BUILTIN_IMAGE_URL
221
222
223async def test_export_import_keeps_every_provider_mapping(radio_mass: MusicAssistant) -> None:
224    """A station linked to several providers keeps all of those links across a round trip."""
225    mass = radio_mass
226    await _clear_radio_library(mass)
227    await mass.music.add_item_to_library(
228        Radio(
229            item_id=BUILTIN_STREAM_URL,
230            provider="builtin",
231            name="Dual Mapped Station",
232            provider_mappings={
233                ProviderMapping(
234                    item_id=BUILTIN_STREAM_URL,
235                    provider_domain="builtin",
236                    provider_instance="builtin",
237                ),
238                ProviderMapping(
239                    item_id="station-123",
240                    provider_domain=FAKE_DOMAIN,
241                    provider_instance=FAKE_INSTANCE,
242                ),
243            },
244        )
245    )
246
247    m3u_data = await mass.music.radio.export_radios()
248    await _clear_radio_library(mass)
249    task = await mass.music.radio.import_radios(m3u_data)
250    await _wait_for_task_status(mass.tasks, task.id, TaskStatus.SUCCESS)
251
252    items = await mass.music.radio.library_items(summary=False)
253    assert len(items) == 1
254    station = items[0]
255    # the name proves the station was refetched from its provider rather than taken from
256    # the file, which is the path that used to return only that provider's mapping
257    assert station.name == "Provider Owned Radio"
258    assert {(pm.provider_domain, pm.item_id) for pm in station.provider_mappings} == {
259        ("builtin", BUILTIN_STREAM_URL),
260        (FAKE_DOMAIN, "station-123"),
261    }
262
263
264async def test_import_plain_url_m3u_yields_playable_radios(radio_mass: MusicAssistant) -> None:
265    """A third-party M3U of bare stream URLs imports as radio with a usable mapping."""
266    mass = radio_mass
267    await _clear_radio_library(mass)
268    m3u_data = "#EXTM3U\n#EXTINF:-1,Plain Station\nhttp://stream.example.com/plain\n"
269
270    task = await mass.music.radio.import_radios(m3u_data)
271    await _wait_for_task_status(mass.tasks, task.id, TaskStatus.SUCCESS)
272
273    items = await mass.music.radio.library_items(summary=False)
274    assert len(items) == 1
275    station = items[0]
276    # a bare url carries no #EXTMA, so import_radios supplies the media type itself
277    assert isinstance(station, Radio)
278    assert station.name == "Plain Station"
279    # an item with no provider mappings never reaches library_add and stays unplayable
280    assert len(station.provider_mappings) == 1
281    mapping = next(iter(station.provider_mappings))
282    assert mapping.provider_domain == "builtin"
283    assert mapping.item_id == "http://stream.example.com/plain"
284    assert [item["item_id"] for item in mass.config.get(CONF_KEY_RADIOS, [])] == [
285        "http://stream.example.com/plain"
286    ]
287
288
289async def test_import_reports_unresolvable_station(radio_mass: MusicAssistant) -> None:
290    """A station whose provider is not loaded is reported, not silently skipped."""
291    mass = radio_mass
292    await _clear_radio_library(mass)
293    m3u_data = (
294        "#EXTM3U\n"
295        "#EXTMA:media_type=radio||name=Gone Radio\n"
296        "#EXTPROV:absent_provider||station-9||absent_provider--instance\n"
297        "#EXTINF:-1,Gone Radio\n"
298        "absent_provider://radio/station-9\n"
299    )
300
301    task = await mass.music.radio.import_radios(m3u_data)
302    await _wait_for_task_status(mass.tasks, task.id, TaskStatus.PARTIAL_SUCCESS)
303
304    failures = mass.tasks.get_task(task.id).failure_messages
305    assert len(failures) == 1
306    assert "Gone Radio" in failures[0]
307    assert await mass.music.radio.library_count() == 0
308
309
310@pytest.mark.parametrize(
311    ("path", "expected"),
312    [
313        # a radio station is a stream URL, so a bare local path is not one
314        ("some/file.mp3", "not a stream URL"),
315        # an entry that resolves to another media type must not be stored as a station
316        ("spotify://track/abc123", "is a track, not a radio station"),
317    ],
318)
319async def test_import_reports_entry_that_is_not_a_station(
320    radio_mass: MusicAssistant, path: str, expected: str
321) -> None:
322    """An entry that is not a radio station is reported, and nothing is stored for it."""
323    mass = radio_mass
324    await _clear_radio_library(mass)
325
326    task = await mass.music.radio.import_radios(f"#EXTM3U\n#EXTINF:-1,Some Entry\n{path}\n")
327    await _wait_for_task_status(mass.tasks, task.id, TaskStatus.PARTIAL_SUCCESS)
328
329    failures = mass.tasks.get_task(task.id).failure_messages
330    assert len(failures) == 1
331    assert expected in failures[0]
332    assert await mass.music.radio.library_count() == 0
333
334
335async def test_import_continues_after_unexpected_error(radio_mass: MusicAssistant) -> None:
336    """A transient fault on one station is reported and the queue behind it still imports."""
337    mass = radio_mass
338    await _clear_radio_library(mass)
339    m3u_data = "#EXTM3U\n" + "".join(
340        f"#EXTMA:media_type=radio||name=Station {name}\n"
341        f"#EXTINF:-1,Station {name}\nhttp://stream.example.com/{name.lower()}\n"
342        for name in ("One", "Two", "Three")
343    )
344    add_item_to_library = mass.music.add_item_to_library
345
346    async def flaky(
347        item: str | MediaItemType | ItemMapping, overwrite_existing: bool = False
348    ) -> MediaItemType:
349        if isinstance(item, Radio) and item.name == "Station Two":
350            raise TimeoutError("connection timed out")
351        return await add_item_to_library(item, overwrite_existing)
352
353    mass.music.add_item_to_library = flaky  # type: ignore[method-assign]
354
355    task = await mass.music.radio.import_radios(m3u_data)
356    await _wait_for_task_status(mass.tasks, task.id, TaskStatus.PARTIAL_SUCCESS)
357
358    failures = mass.tasks.get_task(task.id).failure_messages
359    assert len(failures) == 1
360    assert "Station Two" in failures[0]
361    # the stations queued behind the failure must still be imported
362    imported = {item.name for item in await mass.music.radio.library_items(summary=False)}
363    assert imported == {"Station One", "Station Three"}
364
365
366async def test_import_radios_rejects_empty_m3u(radio_mass: MusicAssistant) -> None:
367    """An M3U without entries is an error, not an empty background task."""
368    with pytest.raises(InvalidDataError):
369        await radio_mass.music.radio.import_radios("#EXTM3U\n")
370