music-assistant-server

1"""Tests for OpenSonicProvider business logic."""
2
3from __future__ import annotations
4
5import asyncio
6from datetime import datetime
7from typing import Any
8from unittest.mock import AsyncMock, Mock
9
10import pytest
11from libopensonic.errors import DataNotFoundError
12from music_assistant_models.enums import ContentType, ImageType, MediaType, StreamType
13from music_assistant_models.errors import MediaNotFoundError
14
15from music_assistant.providers.opensubsonic.parsers import EP_CHAN_SEP
16from music_assistant.providers.opensubsonic.sonic_provider import OpenSonicProvider
17
18
19def _make_task_capturer() -> tuple[list[asyncio.Future[Any]], Mock]:
20    """Return (tasks, mock) that captures coroutines passed to mass.create_task."""
21    tasks: list[asyncio.Future[Any]] = []
22
23    # accept create_task's keyword-only options (task_id, abort_existing, ...) so the stub
24    # keeps matching its signature
25    def _schedule(coro: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
26        task: asyncio.Future[Any] = asyncio.ensure_future(coro)
27        tasks.append(task)
28        return task
29
30    return tasks, Mock(side_effect=_schedule)
31
32
33def _force_cache_miss(provider: OpenSonicProvider) -> list[asyncio.Future[Any]]:
34    """Make a @use_cache method run its body, capturing the background store task."""
35    tasks, task_mock = _make_task_capturer()
36    provider.mass.cache.get_with_freshness = AsyncMock(  # type: ignore[method-assign]
37        return_value=(None, False, False)
38    )
39    provider.mass.cache.set = AsyncMock()  # type: ignore[method-assign]
40    provider.mass.create_task = task_mock  # type: ignore[method-assign]
41    return tasks
42
43
44def _make_sonic_item(
45    *,
46    item_id: str = "tr-1",
47    title: str = "Test Track",
48    content_type: str | None = "audio/flac",
49    transcoded_content_type: str | None = None,
50    sampling_rate: int | None = None,
51    bit_depth: int | None = None,
52    channel_count: int | None = None,
53    duration: int = 120,
54    replay_gain: object = None,
55    cover_art: str | None = None,
56) -> Mock:
57    item = Mock()
58    item.id = item_id
59    item.title = title
60    item.content_type = content_type
61    item.transcoded_content_type = transcoded_content_type
62    item.sampling_rate = sampling_rate
63    item.bit_depth = bit_depth
64    item.channel_count = channel_count
65    item.duration = duration
66    item.replay_gain = replay_gain
67    item.cover_art = cover_art
68    # remaining fields consulted by parse_track when it isn't stubbed out; keep them falsy
69    # so tests exercising the real parser don't need to fake a full album/artist context
70    item.album_id = None
71    item.album = None
72    item.parent = None
73    item.explicit_status = None
74    item.genre = None
75    item.genres = None
76    item.moods = None
77    item.contributors = None
78    item.disc_number = None
79    item.starred = False
80    item.bit_rate = None
81    item.track = None
82    item.music_brainz_id = None
83    item.path = None
84    item.sort_name = None
85    item.artist_id = None
86    item.artists = None
87    item.artist = None
88    return item
89
90
91def _stub_conn(provider: OpenSonicProvider, item: Mock) -> None:
92    provider.conn = Mock()
93    provider.conn.get_song = AsyncMock(return_value=item)
94    provider.conn.get_stream_url = Mock(return_value=("http://server/stream", {}))
95
96
97# ---------------------------------------------------------------------------
98# _set_loudness
99# ---------------------------------------------------------------------------
100
101
102@pytest.mark.asyncio
103async def test_set_loudness_computes_lufs(provider: OpenSonicProvider) -> None:
104    """track_gain and album_gain are each converted to LUFS via -18 - gain."""
105    rg = Mock()
106    rg.track_gain = 2.0
107    rg.album_gain = -1.0
108    item = _make_sonic_item(item_id="tr-1", replay_gain=rg)
109
110    provider.mass.streams.audio_analysis.set_track_loudness = AsyncMock()  # type: ignore[method-assign]
111    tasks, task_mock = _make_task_capturer()
112    provider.mass.create_task = task_mock  # type: ignore[method-assign]
113
114    provider._set_loudness(item)
115    await asyncio.gather(*tasks)
116
117    provider.mass.streams.audio_analysis.set_track_loudness.assert_awaited_once_with(
118        "tr-1", provider.instance_id, -20.0, -17.0
119    )
120
121
122@pytest.mark.asyncio
123async def test_set_loudness_no_album_gain_passes_none(provider: OpenSonicProvider) -> None:
124    """album_loudness is None when the item carries no album_gain."""
125    rg = Mock()
126    rg.track_gain = 0.0
127    rg.album_gain = None
128    item = _make_sonic_item(item_id="tr-2", replay_gain=rg)
129
130    provider.mass.streams.audio_analysis.set_track_loudness = AsyncMock()  # type: ignore[method-assign]
131    tasks, task_mock = _make_task_capturer()
132    provider.mass.create_task = task_mock  # type: ignore[method-assign]
133
134    provider._set_loudness(item)
135    await asyncio.gather(*tasks)
136
137    provider.mass.streams.audio_analysis.set_track_loudness.assert_awaited_once_with(
138        "tr-2", provider.instance_id, -18.0, None
139    )
140
141
142def test_set_loudness_no_track_gain_skips_task(provider: OpenSonicProvider) -> None:
143    """No task is scheduled when track_gain is None."""
144    rg = Mock()
145    rg.track_gain = None
146    item = _make_sonic_item(replay_gain=rg)
147
148    provider.mass.create_task = Mock()  # type: ignore[method-assign]
149    provider._set_loudness(item)
150
151    provider.mass.create_task.assert_not_called()
152
153
154def test_set_loudness_no_replay_gain_skips_task(provider: OpenSonicProvider) -> None:
155    """No task is scheduled when the item has no replay_gain at all."""
156    item = _make_sonic_item(replay_gain=None)
157
158    provider.mass.create_task = Mock()  # type: ignore[method-assign]
159    provider._set_loudness(item)
160
161    provider.mass.create_task.assert_not_called()
162
163
164# ---------------------------------------------------------------------------
165# get_stream_details
166# ---------------------------------------------------------------------------
167
168
169@pytest.mark.asyncio
170async def test_stream_details_audio_format_fallbacks(provider: OpenSonicProvider) -> None:
171    """Missing sampling_rate, bit_depth, channel_count fall back to safe defaults."""
172    item = _make_sonic_item(sampling_rate=None, bit_depth=None, channel_count=None)
173    _stub_conn(provider, item)
174
175    sd = await provider.get_stream_details("tr-1", MediaType.TRACK)
176
177    assert sd.audio_format.sample_rate == 44100
178    assert sd.audio_format.bit_depth == 16
179    assert sd.audio_format.channels == 2
180
181
182@pytest.mark.asyncio
183async def test_stream_details_prefers_transcoded_content_type(
184    provider: OpenSonicProvider,
185) -> None:
186    """transcoded_content_type takes precedence over content_type for the mime type."""
187    item = _make_sonic_item(content_type="audio/flac", transcoded_content_type="audio/ogg")
188    _stub_conn(provider, item)
189
190    sd = await provider.get_stream_details("tr-1", MediaType.TRACK)
191
192    assert sd.audio_format.content_type == ContentType.try_parse("audio/ogg")
193
194
195@pytest.mark.asyncio
196async def test_stream_details_falls_back_to_content_type(provider: OpenSonicProvider) -> None:
197    """content_type is used when transcoded_content_type is absent."""
198    item = _make_sonic_item(content_type="audio/flac", transcoded_content_type=None)
199    _stub_conn(provider, item)
200
201    sd = await provider.get_stream_details("tr-1", MediaType.TRACK)
202
203    assert sd.audio_format.content_type == ContentType.try_parse("audio/flac")
204
205
206@pytest.mark.asyncio
207async def test_stream_details_stream_type_and_seek_flags(provider: OpenSonicProvider) -> None:
208    """StreamDetails must expose HTTP stream type with seeking enabled."""
209    item = _make_sonic_item()
210    _stub_conn(provider, item)
211
212    sd = await provider.get_stream_details("tr-1", MediaType.TRACK)
213
214    assert sd.stream_type == StreamType.HTTP
215    assert sd.can_seek is True
216    assert sd.allow_seek is True
217
218
219@pytest.mark.asyncio
220async def test_stream_details_raises_media_not_found(provider: OpenSonicProvider) -> None:
221    """DataNotFoundError from the server is translated to MediaNotFoundError."""
222    provider.conn = Mock()
223    provider.conn.get_song = AsyncMock(side_effect=DataNotFoundError("not found"))
224
225    with pytest.raises(MediaNotFoundError):
226        await provider.get_stream_details("tr-missing", MediaType.TRACK)
227
228
229@pytest.mark.asyncio
230async def test_get_library_radios_returns_parsed_radios(provider: OpenSonicProvider) -> None:
231    """Internet radio stations are pulled from the server and parsed into MA radios."""
232    radio_station = Mock()
233    radio_station.id = "radio-1"
234    radio_station.name = "Sample Radio"
235    radio_station.stream_url = "https://example.com/stream"
236    radio_station.home_page_url = "https://example.com"
237    radio_station.cover_art = "/cover.jpg"
238
239    provider.conn = Mock()
240    provider.conn.get_internet_radio_stations = AsyncMock(return_value=[radio_station])
241
242    radios = [radio async for radio in provider.get_library_radios()]
243
244    assert len(radios) == 1
245    assert radios[0].item_id == "radio-1"
246    assert radios[0].name == "Sample Radio"
247    assert radios[0].uri == "https://example.com/stream"
248
249
250@pytest.mark.asyncio
251async def test_get_radio_returns_matching_station(provider: OpenSonicProvider) -> None:
252    """Lookup by provider id resolves the requested radio station."""
253    radio_station = Mock()
254    radio_station.id = "radio-1"
255    radio_station.name = "Sample Radio"
256    radio_station.stream_url = "https://example.com/stream"
257    radio_station.home_page_url = "https://example.com"
258    radio_station.cover_art = None
259
260    provider.conn = Mock()
261    provider.conn.get_internet_radio_stations = AsyncMock(return_value=[radio_station])
262
263    radio = await provider.get_radio("radio-1")
264
265    assert radio.item_id == "radio-1"
266    assert radio.name == "Sample Radio"
267
268
269@pytest.mark.asyncio
270async def test_stream_details_for_radio(provider: OpenSonicProvider) -> None:
271    """Radio stream details point to the station stream URL and disable seeking."""
272    radio_station = Mock()
273    radio_station.id = "radio-1"
274    radio_station.name = "Sample Radio"
275    radio_station.stream_url = "https://example.com/stream"
276    radio_station.home_page_url = "https://example.com"
277    radio_station.cover_art = None
278
279    provider.conn = Mock()
280    provider.conn.get_internet_radio_stations = AsyncMock(return_value=[radio_station])
281
282    sd = await provider.get_stream_details("radio-1", MediaType.RADIO)
283
284    assert sd.stream_type == StreamType.HTTP
285    assert sd.path == "https://example.com/stream"
286    assert sd.can_seek is False
287    assert sd.allow_seek is False
288
289
290# ---------------------------------------------------------------------------
291# on_played
292# ---------------------------------------------------------------------------
293
294
295@pytest.mark.asyncio
296async def test_on_played_non_podcast_makes_no_calls(provider: OpenSonicProvider) -> None:
297    """on_played is a no-op for non-podcast media types."""
298    provider.conn = Mock()
299    provider.conn.delete_bookmark = AsyncMock()
300    provider.conn.create_bookmark = AsyncMock()
301
302    await provider.on_played(MediaType.TRACK, "tr-1", True, 60, Mock())
303
304    provider.conn.delete_bookmark.assert_not_awaited()
305    provider.conn.create_bookmark.assert_not_awaited()
306
307
308@pytest.mark.asyncio
309async def test_on_played_fully_played_deletes_bookmark(provider: OpenSonicProvider) -> None:
310    """Fully played podcast episode removes the server-side bookmark."""
311    provider.conn = Mock()
312    provider.conn.delete_bookmark = AsyncMock()
313
314    await provider.on_played(MediaType.PODCAST_EPISODE, f"pd-1{EP_CHAN_SEP}ep-1", True, 0, Mock())
315
316    provider.conn.delete_bookmark.assert_awaited_once_with(mid="ep-1")
317
318
319@pytest.mark.asyncio
320async def test_on_played_in_progress_stores_position_in_milliseconds(
321    provider: OpenSonicProvider,
322) -> None:
323    """Resume position is stored as milliseconds (MA seconds * 1000)."""
324    provider.conn = Mock()
325    provider.conn.create_bookmark = AsyncMock()
326
327    await provider.on_played(MediaType.PODCAST_EPISODE, f"pd-1{EP_CHAN_SEP}ep-1", False, 45, Mock())
328
329    provider.conn.create_bookmark.assert_awaited_once_with(
330        mid="ep-1",
331        position=45000,
332        comment="Music Assistant Bookmark",
333    )
334
335
336# ---------------------------------------------------------------------------
337# get_resume_position
338# ---------------------------------------------------------------------------
339
340
341@pytest.mark.asyncio
342async def test_get_resume_position_non_podcast_raises(provider: OpenSonicProvider) -> None:
343    """Non-podcast media type raises NotImplementedError."""
344    with pytest.raises(NotImplementedError):
345        await provider.get_resume_position("tr-1", MediaType.TRACK)
346
347
348@pytest.mark.asyncio
349async def test_get_resume_position_returns_matching_bookmark(
350    provider: OpenSonicProvider,
351) -> None:
352    """When a bookmark exists for the episode, its position and timestamp are returned."""
353    bookmark = Mock()
354    bookmark.entry.id = "ep-1"
355    bookmark.position = 90000
356    bookmark.created = "2026-07-25T10:00:00"
357
358    provider.conn = Mock()
359    provider.conn.get_bookmarks = AsyncMock(return_value=[bookmark])
360
361    fully_played, position, created = await provider.get_resume_position(
362        f"pd-1{EP_CHAN_SEP}ep-1", MediaType.PODCAST_EPISODE
363    )
364
365    assert not fully_played
366    assert position == 90000
367    assert isinstance(created, datetime)
368
369
370@pytest.mark.asyncio
371async def test_get_resume_position_no_bookmark_returns_zero(
372    provider: OpenSonicProvider,
373) -> None:
374    """When no bookmark exists, position is 0 and created is None."""
375    provider.conn = Mock()
376    provider.conn.get_bookmarks = AsyncMock(return_value=[])
377
378    fully_played, position, created = await provider.get_resume_position(
379        f"pd-1{EP_CHAN_SEP}ep-1", MediaType.PODCAST_EPISODE
380    )
381
382    assert not fully_played
383    assert position == 0
384    assert created is None
385
386
387# ---------------------------------------------------------------------------
388# get_playlist_tracks
389# ---------------------------------------------------------------------------
390
391
392def _parse_track_stub(*_args: Any, **_kwargs: Any) -> Mock:
393    """Return a lightweight stand-in Track (parse_track is covered by test_parsers)."""
394    return Mock(position=0)
395
396
397@pytest.mark.asyncio
398async def test_get_playlist_tracks_avoids_per_track_metadata_fetch(
399    provider: OpenSonicProvider, monkeypatch: pytest.MonkeyPatch
400) -> None:
401    """Enqueuing a playlist must not fetch album or lyrics per track (avoids N+1 requests)."""
402    sonic_playlist = Mock()
403    sonic_playlist.entry = [_make_sonic_item(item_id=f"tr-{i}") for i in range(3)]
404
405    provider.conn = Mock()
406    provider.conn.get_playlist = AsyncMock(return_value=sonic_playlist)
407    provider.conn.get_album = AsyncMock()
408    provider.conn.get_album_info2 = AsyncMock()
409    provider.conn.get_lyrics = AsyncMock()
410    provider.conn.get_lyrics_by_song_id = AsyncMock()
411
412    tasks = _force_cache_miss(provider)
413
414    # parse_track has its own coverage in test_parsers; isolate the fetch behaviour here
415    monkeypatch.setattr(
416        "music_assistant.providers.opensubsonic.sonic_provider.parse_track",
417        _parse_track_stub,
418    )
419
420    result = await provider.get_playlist_tracks("pl-1")
421    await asyncio.gather(*tasks)
422
423    assert len(result) == 3
424    assert [track.position for track in result] == [1, 2, 3]
425    provider.conn.get_playlist.assert_awaited_once_with("pl-1")
426    provider.conn.get_album.assert_not_awaited()
427    provider.conn.get_album_info2.assert_not_awaited()
428    provider.conn.get_lyrics.assert_not_awaited()
429    provider.conn.get_lyrics_by_song_id.assert_not_awaited()
430
431
432@pytest.mark.asyncio
433async def test_get_playlist_tracks_sets_track_image_from_cover_art(
434    provider: OpenSonicProvider,
435) -> None:
436    """Playlist tracks must carry their own cover art, since album is no longer fetched."""
437    sonic_playlist = Mock()
438    sonic_playlist.entry = [
439        _make_sonic_item(item_id=f"tr-{i}", cover_art=f"cover-{i}") for i in range(3)
440    ]
441
442    provider.conn = Mock()
443    provider.conn.get_playlist = AsyncMock(return_value=sonic_playlist)
444    provider.conn.get_album = AsyncMock()
445    provider.conn.get_album_info2 = AsyncMock()
446    provider.conn.get_lyrics = AsyncMock()
447    provider.conn.get_lyrics_by_song_id = AsyncMock()
448
449    tasks = _force_cache_miss(provider)
450
451    result = await provider.get_playlist_tracks("pl-1")
452    await asyncio.gather(*tasks)
453
454    assert len(result) == 3
455    for i, track in enumerate(result):
456        images = track.metadata.images
457        assert images is not None
458        assert len(images) == 1
459        assert images[0].type == ImageType.THUMB
460        assert images[0].path == f"cover-{i}"
461        assert images[0].provider == provider.instance_id
462        assert images[0].remotely_accessible is False
463    provider.conn.get_album.assert_not_awaited()
464    provider.conn.get_album_info2.assert_not_awaited()
465    provider.conn.get_lyrics.assert_not_awaited()
466    provider.conn.get_lyrics_by_song_id.assert_not_awaited()
467
468
469# ---------------------------------------------------------------------------
470# search / get_artist_toptracks
471# ---------------------------------------------------------------------------
472
473
474@pytest.mark.asyncio
475async def test_search_avoids_per_track_lyrics_fetch(
476    provider: OpenSonicProvider, monkeypatch: pytest.MonkeyPatch
477) -> None:
478    """Searching must not fetch lyrics per hit (avoids N+1 requests)."""
479    answer = Mock()
480    answer.artist = None
481    answer.album = None
482    answer.song = [_make_sonic_item(item_id=f"tr-{i}") for i in range(3)]
483
484    provider.conn = Mock()
485    provider.conn.search3 = AsyncMock(return_value=answer)
486    provider.conn.get_lyrics = AsyncMock()
487    provider.conn.get_lyrics_by_song_id = AsyncMock()
488
489    tasks = _force_cache_miss(provider)
490    monkeypatch.setattr(
491        "music_assistant.providers.opensubsonic.sonic_provider.parse_track",
492        _parse_track_stub,
493    )
494
495    result = await provider.search("query", [MediaType.TRACK], limit=3)
496    await asyncio.gather(*tasks)
497
498    assert len(result.tracks) == 3
499    provider.conn.search3.assert_awaited_once()
500    provider.conn.get_lyrics.assert_not_awaited()
501    provider.conn.get_lyrics_by_song_id.assert_not_awaited()
502
503
504@pytest.mark.asyncio
505async def test_get_artist_toptracks_avoids_per_track_lyrics_fetch(
506    provider: OpenSonicProvider, monkeypatch: pytest.MonkeyPatch
507) -> None:
508    """Opening an artist must not fetch lyrics per top track."""
509    provider.conn = Mock()
510    provider.conn.get_artist = AsyncMock(return_value=Mock(name="an-artist"))
511    provider.conn.get_top_songs = AsyncMock(
512        return_value=[_make_sonic_item(item_id=f"tr-{i}") for i in range(3)]
513    )
514    provider.conn.get_lyrics = AsyncMock()
515    provider.conn.get_lyrics_by_song_id = AsyncMock()
516
517    tasks = _force_cache_miss(provider)
518    monkeypatch.setattr(
519        "music_assistant.providers.opensubsonic.sonic_provider.parse_track",
520        _parse_track_stub,
521    )
522
523    result = await provider.get_artist_toptracks("ar-1")
524    await asyncio.gather(*tasks)
525
526    assert len(result) == 3
527    provider.conn.get_top_songs.assert_awaited_once()
528    provider.conn.get_lyrics.assert_not_awaited()
529    provider.conn.get_lyrics_by_song_id.assert_not_awaited()
530