music-assistant-server

1"""Unit tests for the KION Music API client."""
2
3from __future__ import annotations
4
5import base64
6import hashlib
7import hmac
8from unittest import mock
9
10import pytest
11from yandex_music.exceptions import NetworkError
12from yandex_music.utils.sign_request import DEFAULT_SIGN_KEY
13
14from music_assistant.providers.kion_music.api_client import KionMusicClient
15from music_assistant.providers.kion_music.constants import DEFAULT_BASE_URL
16
17
18@pytest.fixture
19def client() -> KionMusicClient:
20    """Return a KionMusicClient with a fake token and explicit base URL."""
21    return KionMusicClient("fake_token", base_url=DEFAULT_BASE_URL)
22
23
24async def test_connect_sets_base_url(client: KionMusicClient) -> None:
25    """Verify connect() passes DEFAULT_BASE_URL to ClientAsync."""
26    with mock.patch("music_assistant.providers.kion_music.api_client.ClientAsync") as mock_cls:
27        mock_instance = mock.AsyncMock()
28        mock_instance.me = type("Me", (), {"account": type("Account", (), {"uid": 42})()})()
29        mock_instance.init = mock.AsyncMock(return_value=mock_instance)
30        mock_cls.return_value = mock_instance
31
32        result = await client.connect()
33
34        assert result is True
35        mock_cls.assert_called_once_with("fake_token", base_url=DEFAULT_BASE_URL)
36
37
38async def test_get_liked_albums_batching(client: KionMusicClient) -> None:
39    """Test that liked albums are fetched in batches of 50."""
40    mock_client = mock.AsyncMock()
41    client._client = mock_client
42    client._user_id = 1
43
44    # Create 60 likes so we get 2 batches
45    likes = []
46    for i in range(60):
47        like = type("Like", (), {"album": type("Album", (), {"id": i + 1})()})()
48        likes.append(like)
49
50    mock_client.users_likes_albums = mock.AsyncMock(return_value=likes)
51
52    batch1 = [type("Album", (), {"id": i + 1})() for i in range(50)]
53    batch2 = [type("Album", (), {"id": i + 51})() for i in range(10)]
54    mock_client.albums = mock.AsyncMock(side_effect=[batch1, batch2])
55
56    result = await client.get_liked_albums()
57
58    assert len(result) == 60
59    assert mock_client.albums.call_count == 2
60
61
62async def test_get_liked_albums_batch_fallback_on_network_error(
63    client: KionMusicClient,
64) -> None:
65    """Test fallback to minimal data when batch fetch fails."""
66    mock_client = mock.AsyncMock()
67    client._client = mock_client
68    client._user_id = 1
69
70    album_obj = type("Album", (), {"id": 1})()
71    likes = [type("Like", (), {"album": album_obj})()]
72
73    mock_client.users_likes_albums = mock.AsyncMock(return_value=likes)
74    mock_client.albums = mock.AsyncMock(side_effect=NetworkError("timeout"))
75
76    result = await client.get_liked_albums()
77
78    assert len(result) == 1
79    assert result[0].id == 1
80
81
82# ─── send_rotor_station_feedback: dispatch to typed v3 helpers ────────────────
83
84
85async def test_rotor_feedback_radio_started_calls_typed_helper(
86    client: KionMusicClient,
87) -> None:
88    """RadioStarted dispatches to rotor_station_feedback_radio_started."""
89    mock_client = mock.AsyncMock()
90    mock_client.rotor_station_feedback_radio_started = mock.AsyncMock(return_value=True)
91    mock_client.rotor_station_feedback = mock.AsyncMock(return_value=True)
92    client._client = mock_client
93
94    ok = await client.send_rotor_station_feedback("user:onyourwave", "radioStarted")
95
96    assert ok is True
97    mock_client.rotor_station_feedback_radio_started.assert_awaited_once()
98    mock_client.rotor_station_feedback.assert_not_called()
99
100
101async def test_rotor_feedback_track_started_requires_track_id(
102    client: KionMusicClient,
103) -> None:
104    """TrackStarted without track_id returns False and skips the API call."""
105    mock_client = mock.AsyncMock()
106    mock_client.rotor_station_feedback_track_started = mock.AsyncMock(return_value=True)
107    client._client = mock_client
108
109    ok = await client.send_rotor_station_feedback("user:onyourwave", "trackStarted", track_id=None)
110
111    assert ok is False
112    mock_client.rotor_station_feedback_track_started.assert_not_called()
113
114
115async def test_rotor_feedback_track_started_calls_typed_helper(
116    client: KionMusicClient,
117) -> None:
118    """TrackStarted with track_id dispatches to rotor_station_feedback_track_started."""
119    mock_client = mock.AsyncMock()
120    mock_client.rotor_station_feedback_track_started = mock.AsyncMock(return_value=True)
121    client._client = mock_client
122
123    ok = await client.send_rotor_station_feedback(
124        "user:onyourwave", "trackStarted", track_id="42", batch_id="batch-1"
125    )
126
127    assert ok is True
128    mock_client.rotor_station_feedback_track_started.assert_awaited_once()
129    args, kwargs = mock_client.rotor_station_feedback_track_started.call_args
130    assert args[0] == "user:onyourwave"
131    assert kwargs["track_id"] == "42"
132    assert kwargs["batch_id"] == "batch-1"
133
134
135async def test_rotor_feedback_track_finished_calls_typed_helper(
136    client: KionMusicClient,
137) -> None:
138    """TrackFinished dispatches to rotor_station_feedback_track_finished with seconds."""
139    mock_client = mock.AsyncMock()
140    mock_client.rotor_station_feedback_track_finished = mock.AsyncMock(return_value=True)
141    client._client = mock_client
142
143    ok = await client.send_rotor_station_feedback(
144        "user:onyourwave",
145        "trackFinished",
146        track_id="42",
147        total_played_seconds=123,
148    )
149
150    assert ok is True
151    mock_client.rotor_station_feedback_track_finished.assert_awaited_once()
152    _, kwargs = mock_client.rotor_station_feedback_track_finished.call_args
153    assert kwargs["track_id"] == "42"
154    assert kwargs["total_played_seconds"] == 123.0
155
156
157async def test_rotor_feedback_skip_calls_typed_helper(
158    client: KionMusicClient,
159) -> None:
160    """Skip dispatches to rotor_station_feedback_skip with played seconds."""
161    mock_client = mock.AsyncMock()
162    mock_client.rotor_station_feedback_skip = mock.AsyncMock(return_value=True)
163    client._client = mock_client
164
165    ok = await client.send_rotor_station_feedback(
166        "user:onyourwave", "skip", track_id="42", total_played_seconds=10
167    )
168
169    assert ok is True
170    mock_client.rotor_station_feedback_skip.assert_awaited_once()
171    _, kwargs = mock_client.rotor_station_feedback_skip.call_args
172    assert kwargs["track_id"] == "42"
173    assert kwargs["total_played_seconds"] == 10.0
174
175
176async def test_rotor_feedback_unknown_type_falls_back(
177    client: KionMusicClient,
178) -> None:
179    """Unknown feedback types fall back to the generic rotor_station_feedback helper."""
180    mock_client = mock.AsyncMock()
181    mock_client.rotor_station_feedback = mock.AsyncMock(return_value=True)
182    # Typed helpers must NOT be called for an unknown feedback type.
183    mock_client.rotor_station_feedback_radio_started = mock.AsyncMock(return_value=True)
184    mock_client.rotor_station_feedback_track_started = mock.AsyncMock(return_value=True)
185    mock_client.rotor_station_feedback_track_finished = mock.AsyncMock(return_value=True)
186    mock_client.rotor_station_feedback_skip = mock.AsyncMock(return_value=True)
187    client._client = mock_client
188
189    ok = await client.send_rotor_station_feedback("user:onyourwave", "like", track_id="42")
190
191    assert ok is True
192    mock_client.rotor_station_feedback.assert_awaited_once()
193    mock_client.rotor_station_feedback_radio_started.assert_not_called()
194    mock_client.rotor_station_feedback_track_started.assert_not_called()
195    mock_client.rotor_station_feedback_track_finished.assert_not_called()
196    mock_client.rotor_station_feedback_skip.assert_not_called()
197
198
199# ─── get_track_file_info: params + sign construction ─────────────────────────
200
201
202async def test_get_track_file_info_normalizes_codec_whitespace(
203    client: KionMusicClient,
204) -> None:
205    """Whitespace around codec tokens is stripped from both params and sign string."""
206    mock_client = mock.AsyncMock()
207    mock_client.base_url = DEFAULT_BASE_URL
208    mock_request = mock.AsyncMock()
209    mock_request.get = mock.AsyncMock(return_value={"downloadInfo": None})
210    mock_client._request = mock_request
211    client._client = mock_client
212
213    await client.get_track_file_info(
214        "42",
215        quality="lossless",
216        codecs=" flac-mp4 , flac , aac-mp4 ",
217        transport="raw",
218    )
219
220    mock_request.get.assert_awaited_once()
221    _, kwargs = mock_request.get.call_args
222    params = kwargs["params"]
223    assert params["codecs"] == "flac-mp4,flac,aac-mp4"
224
225
226async def test_get_track_file_info_builds_signed_params(
227    client: KionMusicClient,
228) -> None:
229    """Sign string is ts+trackId+quality+codecs_no_commas+transport, b64(HMAC-SHA256)[:-1]."""
230    mock_client = mock.AsyncMock()
231    mock_client.base_url = DEFAULT_BASE_URL
232    mock_request = mock.AsyncMock()
233    mock_request.get = mock.AsyncMock(return_value={"downloadInfo": None})
234    mock_client._request = mock_request
235    client._client = mock_client
236
237    await client.get_track_file_info(
238        "42",
239        quality="lossless",
240        codecs="flac-mp4,flac",
241        transport="encraw",
242    )
243
244    mock_request.get.assert_awaited_once()
245    args, kwargs = mock_request.get.call_args
246    assert args[0] == f"{DEFAULT_BASE_URL}/get-file-info"
247    params = kwargs["params"]
248    assert params["trackId"] == "42"
249    assert params["quality"] == "lossless"
250    assert params["codecs"] == "flac-mp4,flac"
251    assert params["transports"] == "encraw"
252    # Recompute expected sign from the emitted ts to verify the formula.
253    ts = params["ts"]
254    expected_sign_input = f"{ts}42losslessflac-mp4flacencraw".encode()
255    expected_sign = (
256        base64.b64encode(
257            hmac.new(DEFAULT_SIGN_KEY.encode(), expected_sign_input, hashlib.sha256).digest()
258        )
259        .decode()
260        .rstrip("=")
261    )
262    assert params["sign"] == expected_sign
263    # Kion API expects 43 chars (base64 of SHA-256 with trailing '=' padding stripped).
264    assert len(params["sign"]) == 43
265
266
267# ─── get_track_lyrics / get_track_lyrics_from_track ───────────────────────────
268
269
270def _make_track_with_lyrics(
271    track_id: str = "42",
272    lyrics_available: bool = True,
273    lyrics_text: str | None = "Just a line of plain lyrics",
274) -> mock.AsyncMock:
275    """Build a KionTrack-like mock whose get_lyrics_async() yields lyrics_text."""
276    track = mock.AsyncMock()
277    track.id = track_id
278    track.lyrics_available = lyrics_available
279    if lyrics_text is None:
280        track.get_lyrics_async = mock.AsyncMock(return_value=None)
281    else:
282        track_lyrics = mock.AsyncMock()
283        track_lyrics.fetch_lyrics_async = mock.AsyncMock(return_value=lyrics_text)
284        track.get_lyrics_async = mock.AsyncMock(return_value=track_lyrics)
285    return track
286
287
288async def test_get_track_lyrics_from_track_synced_detects_lrc(
289    client: KionMusicClient,
290) -> None:
291    """LRC timestamps anywhere in the payload flag lyrics as synced."""
292    track = _make_track_with_lyrics(
293        lyrics_text="[ar:Artist]\n[00:12.34]First line\n[01:03.5]Second line",
294    )
295
296    text, synced = await client.get_track_lyrics_from_track(track)
297
298    assert text is not None
299    assert synced is True
300
301
302async def test_get_track_lyrics_from_track_plain_detects_unsynced(
303    client: KionMusicClient,
304) -> None:
305    """Text without LRC timestamps is reported as non-synced."""
306    track = _make_track_with_lyrics(lyrics_text="Line one\nLine two\nLine three")
307
308    text, synced = await client.get_track_lyrics_from_track(track)
309
310    assert text == "Line one\nLine two\nLine three"
311    assert synced is False
312
313
314async def test_get_track_lyrics_from_track_unavailable_skips_fetch(
315    client: KionMusicClient,
316) -> None:
317    """lyrics_available=False returns (None, False) without calling get_lyrics_async."""
318    track = _make_track_with_lyrics(lyrics_available=False)
319
320    result = await client.get_track_lyrics_from_track(track)
321
322    assert result == (None, False)
323    track.get_lyrics_async.assert_not_called()
324
325
326async def test_get_track_lyrics_from_track_empty_metadata_returns_none(
327    client: KionMusicClient,
328) -> None:
329    """When get_lyrics_async() yields None, return (None, False)."""
330    track = _make_track_with_lyrics(lyrics_text=None)
331
332    assert await client.get_track_lyrics_from_track(track) == (None, False)
333
334
335async def test_get_track_lyrics_from_track_empty_text_returns_none(
336    client: KionMusicClient,
337) -> None:
338    """Empty lyrics text is treated as unavailable."""
339    track = _make_track_with_lyrics(lyrics_text="")
340
341    assert await client.get_track_lyrics_from_track(track) == (None, False)
342
343
344async def test_get_track_lyrics_from_track_swallows_network_error(
345    client: KionMusicClient,
346) -> None:
347    """Network errors during lyric fetch don't bubble — return (None, False)."""
348    track = mock.AsyncMock()
349    track.id = "42"
350    track.lyrics_available = True
351    track.get_lyrics_async = mock.AsyncMock(side_effect=NetworkError("boom"))
352
353    assert await client.get_track_lyrics_from_track(track) == (None, False)
354
355
356async def test_get_track_lyrics_delegates_to_from_track(
357    client: KionMusicClient,
358) -> None:
359    """get_track_lyrics(track_id) fetches the track, then delegates."""
360    mock_client = mock.AsyncMock()
361    track = _make_track_with_lyrics(lyrics_text="[00:01.00]hi")
362    mock_client.tracks = mock.AsyncMock(return_value=[track])
363    client._client = mock_client
364
365    text, synced = await client.get_track_lyrics("42")
366
367    assert text is not None
368    assert synced is True
369    mock_client.tracks.assert_awaited_once()
370
371
372async def test_get_track_lyrics_missing_track_returns_none(
373    client: KionMusicClient,
374) -> None:
375    """Empty result from tracks([id]) returns (None, False)."""
376    mock_client = mock.AsyncMock()
377    mock_client.tracks = mock.AsyncMock(return_value=[])
378    client._client = mock_client
379
380    assert await client.get_track_lyrics("42") == (None, False)
381