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, 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) -> Mock:
56    item = Mock()
57    item.id = item_id
58    item.title = title
59    item.content_type = content_type
60    item.transcoded_content_type = transcoded_content_type
61    item.sampling_rate = sampling_rate
62    item.bit_depth = bit_depth
63    item.channel_count = channel_count
64    item.duration = duration
65    item.replay_gain = replay_gain
66    return item
67
68
69def _stub_conn(provider: OpenSonicProvider, item: Mock) -> None:
70    provider.conn = Mock()
71    provider.conn.get_song = AsyncMock(return_value=item)
72    provider.conn.get_stream_url = Mock(return_value=("http://server/stream", {}))
73
74
75# ---------------------------------------------------------------------------
76# _set_loudness
77# ---------------------------------------------------------------------------
78
79
80@pytest.mark.asyncio
81async def test_set_loudness_computes_lufs(provider: OpenSonicProvider) -> None:
82    """track_gain and album_gain are each converted to LUFS via -18 - gain."""
83    rg = Mock()
84    rg.track_gain = 2.0
85    rg.album_gain = -1.0
86    item = _make_sonic_item(item_id="tr-1", replay_gain=rg)
87
88    provider.mass.streams.audio_analysis.set_track_loudness = AsyncMock()  # type: ignore[method-assign]
89    tasks, task_mock = _make_task_capturer()
90    provider.mass.create_task = task_mock  # type: ignore[method-assign]
91
92    provider._set_loudness(item)
93    await asyncio.gather(*tasks)
94
95    provider.mass.streams.audio_analysis.set_track_loudness.assert_awaited_once_with(
96        "tr-1", provider.instance_id, -20.0, -17.0
97    )
98
99
100@pytest.mark.asyncio
101async def test_set_loudness_no_album_gain_passes_none(provider: OpenSonicProvider) -> None:
102    """album_loudness is None when the item carries no album_gain."""
103    rg = Mock()
104    rg.track_gain = 0.0
105    rg.album_gain = None
106    item = _make_sonic_item(item_id="tr-2", replay_gain=rg)
107
108    provider.mass.streams.audio_analysis.set_track_loudness = AsyncMock()  # type: ignore[method-assign]
109    tasks, task_mock = _make_task_capturer()
110    provider.mass.create_task = task_mock  # type: ignore[method-assign]
111
112    provider._set_loudness(item)
113    await asyncio.gather(*tasks)
114
115    provider.mass.streams.audio_analysis.set_track_loudness.assert_awaited_once_with(
116        "tr-2", provider.instance_id, -18.0, None
117    )
118
119
120def test_set_loudness_no_track_gain_skips_task(provider: OpenSonicProvider) -> None:
121    """No task is scheduled when track_gain is None."""
122    rg = Mock()
123    rg.track_gain = None
124    item = _make_sonic_item(replay_gain=rg)
125
126    provider.mass.create_task = Mock()  # type: ignore[method-assign]
127    provider._set_loudness(item)
128
129    provider.mass.create_task.assert_not_called()
130
131
132def test_set_loudness_no_replay_gain_skips_task(provider: OpenSonicProvider) -> None:
133    """No task is scheduled when the item has no replay_gain at all."""
134    item = _make_sonic_item(replay_gain=None)
135
136    provider.mass.create_task = Mock()  # type: ignore[method-assign]
137    provider._set_loudness(item)
138
139    provider.mass.create_task.assert_not_called()
140
141
142# ---------------------------------------------------------------------------
143# get_stream_details
144# ---------------------------------------------------------------------------
145
146
147@pytest.mark.asyncio
148async def test_stream_details_audio_format_fallbacks(provider: OpenSonicProvider) -> None:
149    """Missing sampling_rate, bit_depth, channel_count fall back to safe defaults."""
150    item = _make_sonic_item(sampling_rate=None, bit_depth=None, channel_count=None)
151    _stub_conn(provider, item)
152
153    sd = await provider.get_stream_details("tr-1", MediaType.TRACK)
154
155    assert sd.audio_format.sample_rate == 44100
156    assert sd.audio_format.bit_depth == 16
157    assert sd.audio_format.channels == 2
158
159
160@pytest.mark.asyncio
161async def test_stream_details_prefers_transcoded_content_type(
162    provider: OpenSonicProvider,
163) -> None:
164    """transcoded_content_type takes precedence over content_type for the mime type."""
165    item = _make_sonic_item(content_type="audio/flac", transcoded_content_type="audio/ogg")
166    _stub_conn(provider, item)
167
168    sd = await provider.get_stream_details("tr-1", MediaType.TRACK)
169
170    assert sd.audio_format.content_type == ContentType.try_parse("audio/ogg")
171
172
173@pytest.mark.asyncio
174async def test_stream_details_falls_back_to_content_type(provider: OpenSonicProvider) -> None:
175    """content_type is used when transcoded_content_type is absent."""
176    item = _make_sonic_item(content_type="audio/flac", transcoded_content_type=None)
177    _stub_conn(provider, item)
178
179    sd = await provider.get_stream_details("tr-1", MediaType.TRACK)
180
181    assert sd.audio_format.content_type == ContentType.try_parse("audio/flac")
182
183
184@pytest.mark.asyncio
185async def test_stream_details_stream_type_and_seek_flags(provider: OpenSonicProvider) -> None:
186    """StreamDetails must expose HTTP stream type with seeking enabled."""
187    item = _make_sonic_item()
188    _stub_conn(provider, item)
189
190    sd = await provider.get_stream_details("tr-1", MediaType.TRACK)
191
192    assert sd.stream_type == StreamType.HTTP
193    assert sd.can_seek is True
194    assert sd.allow_seek is True
195
196
197@pytest.mark.asyncio
198async def test_stream_details_raises_media_not_found(provider: OpenSonicProvider) -> None:
199    """DataNotFoundError from the server is translated to MediaNotFoundError."""
200    provider.conn = Mock()
201    provider.conn.get_song = AsyncMock(side_effect=DataNotFoundError("not found"))
202
203    with pytest.raises(MediaNotFoundError):
204        await provider.get_stream_details("tr-missing", MediaType.TRACK)
205
206
207@pytest.mark.asyncio
208async def test_get_library_radios_returns_parsed_radios(provider: OpenSonicProvider) -> None:
209    """Internet radio stations are pulled from the server and parsed into MA radios."""
210    radio_station = Mock()
211    radio_station.id = "radio-1"
212    radio_station.name = "Sample Radio"
213    radio_station.stream_url = "https://example.com/stream"
214    radio_station.home_page_url = "https://example.com"
215    radio_station.cover_art = "/cover.jpg"
216
217    provider.conn = Mock()
218    provider.conn.get_internet_radio_stations = AsyncMock(return_value=[radio_station])
219
220    radios = [radio async for radio in provider.get_library_radios()]
221
222    assert len(radios) == 1
223    assert radios[0].item_id == "radio-1"
224    assert radios[0].name == "Sample Radio"
225    assert radios[0].uri == "https://example.com/stream"
226
227
228@pytest.mark.asyncio
229async def test_get_radio_returns_matching_station(provider: OpenSonicProvider) -> None:
230    """Lookup by provider id resolves the requested radio station."""
231    radio_station = Mock()
232    radio_station.id = "radio-1"
233    radio_station.name = "Sample Radio"
234    radio_station.stream_url = "https://example.com/stream"
235    radio_station.home_page_url = "https://example.com"
236    radio_station.cover_art = None
237
238    provider.conn = Mock()
239    provider.conn.get_internet_radio_stations = AsyncMock(return_value=[radio_station])
240
241    radio = await provider.get_radio("radio-1")
242
243    assert radio.item_id == "radio-1"
244    assert radio.name == "Sample Radio"
245
246
247@pytest.mark.asyncio
248async def test_stream_details_for_radio(provider: OpenSonicProvider) -> None:
249    """Radio stream details point to the station stream URL and disable seeking."""
250    radio_station = Mock()
251    radio_station.id = "radio-1"
252    radio_station.name = "Sample Radio"
253    radio_station.stream_url = "https://example.com/stream"
254    radio_station.home_page_url = "https://example.com"
255    radio_station.cover_art = None
256
257    provider.conn = Mock()
258    provider.conn.get_internet_radio_stations = AsyncMock(return_value=[radio_station])
259
260    sd = await provider.get_stream_details("radio-1", MediaType.RADIO)
261
262    assert sd.stream_type == StreamType.HTTP
263    assert sd.path == "https://example.com/stream"
264    assert sd.can_seek is False
265    assert sd.allow_seek is False
266
267
268# ---------------------------------------------------------------------------
269# on_played
270# ---------------------------------------------------------------------------
271
272
273@pytest.mark.asyncio
274async def test_on_played_non_podcast_makes_no_calls(provider: OpenSonicProvider) -> None:
275    """on_played is a no-op for non-podcast media types."""
276    provider.conn = Mock()
277    provider.conn.delete_bookmark = AsyncMock()
278    provider.conn.create_bookmark = AsyncMock()
279
280    await provider.on_played(MediaType.TRACK, "tr-1", True, 60, Mock())
281
282    provider.conn.delete_bookmark.assert_not_awaited()
283    provider.conn.create_bookmark.assert_not_awaited()
284
285
286@pytest.mark.asyncio
287async def test_on_played_fully_played_deletes_bookmark(provider: OpenSonicProvider) -> None:
288    """Fully played podcast episode removes the server-side bookmark."""
289    provider.conn = Mock()
290    provider.conn.delete_bookmark = AsyncMock()
291
292    await provider.on_played(MediaType.PODCAST_EPISODE, f"pd-1{EP_CHAN_SEP}ep-1", True, 0, Mock())
293
294    provider.conn.delete_bookmark.assert_awaited_once_with(mid="ep-1")
295
296
297@pytest.mark.asyncio
298async def test_on_played_in_progress_stores_position_in_milliseconds(
299    provider: OpenSonicProvider,
300) -> None:
301    """Resume position is stored as milliseconds (MA seconds * 1000)."""
302    provider.conn = Mock()
303    provider.conn.create_bookmark = AsyncMock()
304
305    await provider.on_played(MediaType.PODCAST_EPISODE, f"pd-1{EP_CHAN_SEP}ep-1", False, 45, Mock())
306
307    provider.conn.create_bookmark.assert_awaited_once_with(
308        mid="ep-1",
309        position=45000,
310        comment="Music Assistant Bookmark",
311    )
312
313
314# ---------------------------------------------------------------------------
315# get_resume_position
316# ---------------------------------------------------------------------------
317
318
319@pytest.mark.asyncio
320async def test_get_resume_position_non_podcast_raises(provider: OpenSonicProvider) -> None:
321    """Non-podcast media type raises NotImplementedError."""
322    with pytest.raises(NotImplementedError):
323        await provider.get_resume_position("tr-1", MediaType.TRACK)
324
325
326@pytest.mark.asyncio
327async def test_get_resume_position_returns_matching_bookmark(
328    provider: OpenSonicProvider,
329) -> None:
330    """When a bookmark exists for the episode, its position and timestamp are returned."""
331    bookmark = Mock()
332    bookmark.entry.id = "ep-1"
333    bookmark.position = 90000
334    bookmark.created = "2026-07-25T10:00:00"
335
336    provider.conn = Mock()
337    provider.conn.get_bookmarks = AsyncMock(return_value=[bookmark])
338
339    fully_played, position, created = await provider.get_resume_position(
340        f"pd-1{EP_CHAN_SEP}ep-1", MediaType.PODCAST_EPISODE
341    )
342
343    assert not fully_played
344    assert position == 90000
345    assert isinstance(created, datetime)
346
347
348@pytest.mark.asyncio
349async def test_get_resume_position_no_bookmark_returns_zero(
350    provider: OpenSonicProvider,
351) -> None:
352    """When no bookmark exists, position is 0 and created is None."""
353    provider.conn = Mock()
354    provider.conn.get_bookmarks = AsyncMock(return_value=[])
355
356    fully_played, position, created = await provider.get_resume_position(
357        f"pd-1{EP_CHAN_SEP}ep-1", MediaType.PODCAST_EPISODE
358    )
359
360    assert not fully_played
361    assert position == 0
362    assert created is None
363
364
365# ---------------------------------------------------------------------------
366# get_playlist_tracks
367# ---------------------------------------------------------------------------
368
369
370def _parse_track_stub(*_args: Any, **_kwargs: Any) -> Mock:
371    """Return a lightweight stand-in Track (parse_track is covered by test_parsers)."""
372    return Mock(position=0)
373
374
375@pytest.mark.asyncio
376async def test_get_playlist_tracks_avoids_per_track_metadata_fetch(
377    provider: OpenSonicProvider, monkeypatch: pytest.MonkeyPatch
378) -> None:
379    """Enqueuing a playlist must not fetch album or lyrics per track (avoids N+1 requests)."""
380    sonic_playlist = Mock()
381    sonic_playlist.entry = [_make_sonic_item(item_id=f"tr-{i}") for i in range(3)]
382
383    provider.conn = Mock()
384    provider.conn.get_playlist = AsyncMock(return_value=sonic_playlist)
385    provider.conn.get_album = AsyncMock()
386    provider.conn.get_album_info2 = AsyncMock()
387    provider.conn.get_lyrics = AsyncMock()
388    provider.conn.get_lyrics_by_song_id = AsyncMock()
389
390    tasks = _force_cache_miss(provider)
391
392    # parse_track has its own coverage in test_parsers; isolate the fetch behaviour here
393    monkeypatch.setattr(
394        "music_assistant.providers.opensubsonic.sonic_provider.parse_track",
395        _parse_track_stub,
396    )
397
398    result = await provider.get_playlist_tracks("pl-1")
399    await asyncio.gather(*tasks)
400
401    assert len(result) == 3
402    assert [track.position for track in result] == [1, 2, 3]
403    provider.conn.get_playlist.assert_awaited_once_with("pl-1")
404    provider.conn.get_album.assert_not_awaited()
405    provider.conn.get_album_info2.assert_not_awaited()
406    provider.conn.get_lyrics.assert_not_awaited()
407    provider.conn.get_lyrics_by_song_id.assert_not_awaited()
408
409
410# ---------------------------------------------------------------------------
411# search / get_artist_toptracks
412# ---------------------------------------------------------------------------
413
414
415@pytest.mark.asyncio
416async def test_search_avoids_per_track_lyrics_fetch(
417    provider: OpenSonicProvider, monkeypatch: pytest.MonkeyPatch
418) -> None:
419    """Searching must not fetch lyrics per hit (avoids N+1 requests)."""
420    answer = Mock()
421    answer.artist = None
422    answer.album = None
423    answer.song = [_make_sonic_item(item_id=f"tr-{i}") for i in range(3)]
424
425    provider.conn = Mock()
426    provider.conn.search3 = AsyncMock(return_value=answer)
427    provider.conn.get_lyrics = AsyncMock()
428    provider.conn.get_lyrics_by_song_id = AsyncMock()
429
430    tasks = _force_cache_miss(provider)
431    monkeypatch.setattr(
432        "music_assistant.providers.opensubsonic.sonic_provider.parse_track",
433        _parse_track_stub,
434    )
435
436    result = await provider.search("query", [MediaType.TRACK], limit=3)
437    await asyncio.gather(*tasks)
438
439    assert len(result.tracks) == 3
440    provider.conn.search3.assert_awaited_once()
441    provider.conn.get_lyrics.assert_not_awaited()
442    provider.conn.get_lyrics_by_song_id.assert_not_awaited()
443
444
445@pytest.mark.asyncio
446async def test_get_artist_toptracks_avoids_per_track_lyrics_fetch(
447    provider: OpenSonicProvider, monkeypatch: pytest.MonkeyPatch
448) -> None:
449    """Opening an artist must not fetch lyrics per top track."""
450    provider.conn = Mock()
451    provider.conn.get_artist = AsyncMock(return_value=Mock(name="an-artist"))
452    provider.conn.get_top_songs = AsyncMock(
453        return_value=[_make_sonic_item(item_id=f"tr-{i}") for i in range(3)]
454    )
455    provider.conn.get_lyrics = AsyncMock()
456    provider.conn.get_lyrics_by_song_id = AsyncMock()
457
458    tasks = _force_cache_miss(provider)
459    monkeypatch.setattr(
460        "music_assistant.providers.opensubsonic.sonic_provider.parse_track",
461        _parse_track_stub,
462    )
463
464    result = await provider.get_artist_toptracks("ar-1")
465    await asyncio.gather(*tasks)
466
467    assert len(result) == 3
468    provider.conn.get_top_songs.assert_awaited_once()
469    provider.conn.get_lyrics.assert_not_awaited()
470    provider.conn.get_lyrics_by_song_id.assert_not_awaited()
471