music-assistant-server

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