/
/
/
1"""Tests for the Pins and Listening History browse handlers."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, Mock, patch
6
7import pytest
8from music_assistant_models.errors import InvalidDataError, MediaNotFoundError
9from music_assistant_models.media_items import Album, Artist, Playlist, Track
10
11from music_assistant.providers.yandex_music.provider import YandexMusicProvider
12
13
14@pytest.fixture
15def provider_mock() -> Mock:
16 """Return a mock Yandex Music provider with cache + client stubs."""
17 provider = Mock(spec=YandexMusicProvider)
18 provider.domain = "yandex_music"
19 provider.instance_id = "yandex_music_instance"
20 provider.logger = Mock()
21 provider.client = AsyncMock()
22 provider.client.user_id = 12345
23 provider.mass = Mock()
24 provider.mass.cache = AsyncMock()
25 provider.mass.cache.get = AsyncMock(return_value=None)
26 provider.mass.cache.set = AsyncMock()
27 return provider
28
29
30@pytest.mark.asyncio
31async def test_browse_pins_returns_empty_when_no_pins(provider_mock: Mock) -> None:
32 """_browse_pins returns [] when client returns None."""
33 provider_mock.client.get_pins = AsyncMock(return_value=None)
34
35 result = await YandexMusicProvider._browse_pins(provider_mock)
36
37 assert result == []
38
39
40@pytest.mark.asyncio
41async def test_browse_pins_returns_empty_when_pins_field_missing(
42 provider_mock: Mock,
43) -> None:
44 """_browse_pins returns [] when PinsList.pins is None."""
45 provider_mock.client.get_pins = AsyncMock(return_value=type("PinsList", (), {"pins": None})())
46
47 result = await YandexMusicProvider._browse_pins(provider_mock)
48
49 assert result == []
50
51
52@pytest.mark.asyncio
53async def test_browse_pins_resolves_artist_album_playlist(provider_mock: Mock) -> None:
54 """_browse_pins routes each pin type to the corresponding lookup."""
55 artist_pin = type(
56 "Pin",
57 (),
58 {"type": "artist_item", "data": type("D", (), {"id": 11})()},
59 )()
60 album_pin = type(
61 "Pin",
62 (),
63 {"type": "album_item", "data": type("D", (), {"id": 22})()},
64 )()
65 playlist_pin = type(
66 "Pin",
67 (),
68 {"type": "playlist_item", "data": type("D", (), {"uid": 33, "kind": 44})()},
69 )()
70 pins = type("PinsList", (), {"pins": [artist_pin, album_pin, playlist_pin]})()
71 provider_mock.client.get_pins = AsyncMock(return_value=pins)
72
73 artist = Mock(spec=Artist)
74 album = Mock(spec=Album)
75 playlist = Mock(spec=Playlist)
76 provider_mock.get_artist = AsyncMock(return_value=artist)
77 provider_mock.get_album = AsyncMock(return_value=album)
78 provider_mock.get_playlist = AsyncMock(return_value=playlist)
79
80 result = await YandexMusicProvider._browse_pins(provider_mock)
81
82 provider_mock.get_artist.assert_awaited_once_with("11")
83 provider_mock.get_album.assert_awaited_once_with("22")
84 provider_mock.get_playlist.assert_awaited_once_with("33:44")
85 assert result == [artist, album, playlist]
86
87
88@pytest.mark.asyncio
89async def test_browse_pins_skips_wave_and_missing_data(provider_mock: Mock) -> None:
90 """_browse_pins ignores wave pins and pins with missing data."""
91 wave_pin = type(
92 "Pin",
93 (),
94 {"type": "wave_item", "data": type("D", (), {})()},
95 )()
96 bad_pin = type("Pin", (), {"type": "album_item", "data": None})()
97 pins = type("PinsList", (), {"pins": [wave_pin, bad_pin]})()
98 provider_mock.client.get_pins = AsyncMock(return_value=pins)
99 provider_mock.get_album = AsyncMock()
100
101 result = await YandexMusicProvider._browse_pins(provider_mock)
102
103 assert result == []
104 provider_mock.get_album.assert_not_called()
105
106
107@pytest.mark.asyncio
108async def test_browse_pins_skips_lookup_errors(provider_mock: Mock) -> None:
109 """_browse_pins survives MediaNotFoundError during single-item lookups."""
110 album_pin = type(
111 "Pin",
112 (),
113 {"type": "album_item", "data": type("D", (), {"id": 22})()},
114 )()
115 pins = type("PinsList", (), {"pins": [album_pin]})()
116 provider_mock.client.get_pins = AsyncMock(return_value=pins)
117 provider_mock.get_album = AsyncMock(side_effect=MediaNotFoundError("gone"))
118
119 result = await YandexMusicProvider._browse_pins(provider_mock)
120
121 assert result == []
122
123
124@pytest.mark.asyncio
125async def test_browse_history_returns_empty_when_no_history(
126 provider_mock: Mock,
127) -> None:
128 """_browse_history returns [] when client returns None."""
129 provider_mock.client.get_music_history = AsyncMock(return_value=None)
130
131 result = await YandexMusicProvider._browse_history(provider_mock)
132
133 assert result == []
134
135
136def _hist_item(track_id: int) -> object:
137 """
138 Build a history entry the way MarshalX actually returns it.
139
140 `data.item_id` is a dict containing track_id, album_id, etc.; `full_model`
141 is not populated by the live API. Callers batch-resolve via get_tracks.
142 """
143 data = type("D", (), {"item_id": {"track_id": str(track_id)}, "full_model": None})()
144 return type("HistItem", (), {"type": "track", "data": data})()
145
146
147@pytest.mark.asyncio
148async def test_browse_history_flattens_and_deduplicates(provider_mock: Mock) -> None:
149 """_browse_history flattens daysâgroupsâtracks, de-dupes by track id, preserves order."""
150 group1 = type("Group", (), {"tracks": [_hist_item(1), _hist_item(2)]})()
151 group2 = type("Group", (), {"tracks": [_hist_item(2), _hist_item(3)]})() # dup id=2
152 tab1 = type("Tab", (), {"items": [group1]})()
153 tab2 = type("Tab", (), {"items": [group2]})()
154 history = type("MusicHistory", (), {"history_tabs": [tab1, tab2]})()
155 provider_mock.client.get_music_history = AsyncMock(return_value=history)
156
157 # Batch-hydrate returns the yandex tracks in their own order; the provider
158 # re-orders them to match the de-duplicated id list.
159 yt1 = type("Yt", (), {"id": 1})()
160 yt2 = type("Yt", (), {"id": 2})()
161 yt3 = type("Yt", (), {"id": 3})()
162 provider_mock.client.get_tracks = AsyncMock(return_value=[yt3, yt1, yt2])
163
164 parsed = [Mock(spec=Track, name="p1"), Mock(spec=Track, name="p2"), Mock(spec=Track, name="p3")]
165 with patch(
166 "music_assistant.providers.yandex_music.provider.parse_track",
167 side_effect=parsed,
168 ):
169 result = await YandexMusicProvider._browse_history(provider_mock)
170
171 provider_mock.client.get_tracks.assert_awaited_once_with(["1", "2", "3"])
172 assert result == parsed
173
174
175@pytest.mark.asyncio
176async def test_browse_history_skips_non_track_items(provider_mock: Mock) -> None:
177 """_browse_history ignores items with type != 'track'."""
178 album_item = type(
179 "HistItem",
180 (),
181 {
182 "type": "album",
183 "data": type("D", (), {"item_id": {"track_id": "99"}})(),
184 },
185 )()
186 group = type("Group", (), {"tracks": [album_item]})()
187 tab = type("Tab", (), {"items": [group]})()
188 history = type("MusicHistory", (), {"history_tabs": [tab]})()
189 provider_mock.client.get_music_history = AsyncMock(return_value=history)
190 provider_mock.client.get_tracks = AsyncMock()
191
192 with patch("music_assistant.providers.yandex_music.provider.parse_track") as parse_track:
193 result = await YandexMusicProvider._browse_history(provider_mock)
194 parse_track.assert_not_called()
195
196 assert result == []
197 # No IDs collected â no hydration round-trip at all
198 provider_mock.client.get_tracks.assert_not_awaited()
199
200
201@pytest.mark.asyncio
202async def test_browse_history_skips_invalid_track(provider_mock: Mock) -> None:
203 """_browse_history drops tracks where parse_track raises InvalidDataError."""
204 group = type("Group", (), {"tracks": [_hist_item(1)]})()
205 tab = type("Tab", (), {"items": [group]})()
206 history = type("MusicHistory", (), {"history_tabs": [tab]})()
207 provider_mock.client.get_music_history = AsyncMock(return_value=history)
208
209 yt1 = type("Yt", (), {"id": 1})()
210 provider_mock.client.get_tracks = AsyncMock(return_value=[yt1])
211
212 with patch(
213 "music_assistant.providers.yandex_music.provider.parse_track",
214 side_effect=InvalidDataError("nope"),
215 ):
216 result = await YandexMusicProvider._browse_history(provider_mock)
217
218 assert result == []
219