/
/
/
1"""Unit tests for the KION Music API client."""
2
3from __future__ import annotations
4
5from unittest import mock
6
7import pytest
8from music_assistant_models.errors import ResourceTemporarilyUnavailable
9from yandex_music.exceptions import NetworkError
10
11from music_assistant.providers.kion_music.api_client import KionMusicClient
12from music_assistant.providers.kion_music.constants import DEFAULT_BASE_URL
13
14
15@pytest.fixture
16def client() -> KionMusicClient:
17 """Return a KionMusicClient with a fake token."""
18 return KionMusicClient("fake_token")
19
20
21async def test_connect_sets_base_url(client: KionMusicClient) -> None:
22 """Verify connect() passes DEFAULT_BASE_URL to ClientAsync."""
23 with mock.patch("music_assistant.providers.kion_music.api_client.ClientAsync") as mock_cls:
24 mock_instance = mock.AsyncMock()
25 mock_instance.me = type("Me", (), {"account": type("Account", (), {"uid": 42})()})()
26 mock_instance.init = mock.AsyncMock(return_value=mock_instance)
27 mock_cls.return_value = mock_instance
28
29 result = await client.connect()
30
31 assert result is True
32 mock_cls.assert_called_once_with("fake_token", base_url=DEFAULT_BASE_URL)
33
34
35async def test_get_liked_albums_batching(client: KionMusicClient) -> None:
36 """Test that liked albums are fetched in batches of 50."""
37 mock_client = mock.AsyncMock()
38 client._client = mock_client
39 client._user_id = 1
40
41 # Create 60 likes so we get 2 batches
42 likes = []
43 for i in range(60):
44 like = type("Like", (), {"album": type("Album", (), {"id": i + 1})()})()
45 likes.append(like)
46
47 mock_client.users_likes_albums = mock.AsyncMock(return_value=likes)
48
49 batch1 = [type("Album", (), {"id": i + 1})() for i in range(50)]
50 batch2 = [type("Album", (), {"id": i + 51})() for i in range(10)]
51 mock_client.albums = mock.AsyncMock(side_effect=[batch1, batch2])
52
53 result = await client.get_liked_albums()
54
55 assert len(result) == 60
56 assert mock_client.albums.call_count == 2
57
58
59async def test_get_liked_albums_batch_fallback_on_network_error(
60 client: KionMusicClient,
61) -> None:
62 """Test fallback to minimal data when batch fetch fails."""
63 mock_client = mock.AsyncMock()
64 client._client = mock_client
65 client._user_id = 1
66
67 album_obj = type("Album", (), {"id": 1})()
68 likes = [type("Like", (), {"album": album_obj})()]
69
70 mock_client.users_likes_albums = mock.AsyncMock(return_value=likes)
71 mock_client.albums = mock.AsyncMock(side_effect=NetworkError("timeout"))
72
73 result = await client.get_liked_albums()
74
75 assert len(result) == 1
76 assert result[0].id == 1
77
78
79async def test_get_tracks_retry_on_network_error_then_success(
80 client: KionMusicClient,
81) -> None:
82 """Test that get_tracks retries once on NetworkError and succeeds."""
83 mock_client = mock.AsyncMock()
84 client._client = mock_client
85 client._user_id = 1
86
87 track = type("Track", (), {"id": 1})()
88 mock_client.tracks = mock.AsyncMock(side_effect=[NetworkError("timeout"), [track]])
89
90 result = await client.get_tracks(["1"])
91
92 assert len(result) == 1
93 assert mock_client.tracks.call_count == 2
94
95
96async def test_get_tracks_retry_on_network_error_both_fail(
97 client: KionMusicClient,
98) -> None:
99 """Test that get_tracks raises ResourceTemporarilyUnavailable when retry fails."""
100 mock_client = mock.AsyncMock()
101 client._client = mock_client
102 client._user_id = 1
103
104 mock_client.tracks = mock.AsyncMock(side_effect=NetworkError("timeout"))
105
106 with pytest.raises(ResourceTemporarilyUnavailable):
107 await client.get_tracks(["1"])
108
109 assert mock_client.tracks.call_count == 2
110