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