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