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