/
/
/
1"""Unit tests for Apple Music get_similar_artists."""
2
3from unittest.mock import AsyncMock, MagicMock
4
5import pytest
6from aiohttp.client_exceptions import ClientResponseError
7from aiohttp.client_reqrep import RequestInfo
8from multidict import CIMultiDict, CIMultiDictProxy
9from music_assistant_models.enums import MediaType
10from music_assistant_models.media_items import Artist
11from yarl import URL
12
13from music_assistant.providers.apple_music.recommendations import AppleMusicRecommendationManager
14from tests.common import use_real_create_task
15
16
17def _make_artist_obj(artist_id: str, name: str) -> dict[str, object]:
18 return {
19 "id": artist_id,
20 "type": "artists",
21 "attributes": {
22 "name": name,
23 "url": f"https://music.apple.com/artist/{artist_id}",
24 },
25 "relationships": {},
26 }
27
28
29def _make_api_response(artist_objects: list[dict[str, object]]) -> dict[str, object]:
30 return {
31 "data": [
32 {
33 "id": "123",
34 "views": {
35 "similar-artists": {
36 "data": artist_objects,
37 }
38 },
39 }
40 ]
41 }
42
43
44def _make_track_obj(track_id: str) -> dict[str, object]:
45 return {
46 "id": track_id,
47 "type": "songs",
48 "attributes": {
49 "name": f"Track {track_id}",
50 "artistName": "Artist",
51 "durationInMillis": 180000,
52 "playParams": {"id": track_id},
53 },
54 "relationships": {},
55 }
56
57
58@pytest.fixture
59def mock_api() -> MagicMock:
60 """Return a MagicMock representing the Apple Music API client."""
61 api_client = MagicMock()
62 api_client.get_data = AsyncMock()
63 api_client.post_data = AsyncMock()
64 api_client.get_ratings = AsyncMock(return_value={})
65 return api_client
66
67
68@pytest.fixture
69def manager(mock_api: MagicMock) -> AppleMusicRecommendationManager:
70 """Return an AppleMusicRecommendationManager wired to a mock API client."""
71 provider = MagicMock()
72 provider.instance_id = "apple_music_test"
73 provider.domain = "apple_music"
74 provider._storefront = "us"
75 provider.logger = MagicMock()
76 provider.mass.cache.get = AsyncMock(return_value=None)
77 provider.mass.cache.get_with_freshness = AsyncMock(return_value=(None, False, False))
78 provider.mass.cache.set = AsyncMock()
79 use_real_create_task(provider.mass)
80 provider.api_client = mock_api
81
82 return AppleMusicRecommendationManager(provider)
83
84
85@pytest.mark.asyncio
86async def test_get_similar_artists_returns_artists(
87 manager: AppleMusicRecommendationManager,
88 mock_api: MagicMock,
89) -> None:
90 """get_similar_artists parses artists from the views.similar-artists response."""
91 mock_api.get_data.return_value = _make_api_response(
92 [
93 _make_artist_obj("456", "Radiohead"),
94 _make_artist_obj("789", "Portishead"),
95 ]
96 )
97
98 result = await manager.get_similar_artists("123", limit=25)
99
100 mock_api.get_data.assert_called_once_with(
101 "catalog/us/artists/123",
102 views="similar-artists",
103 )
104 assert len(result) == 2
105 assert all(isinstance(a, Artist) for a in result)
106 names = {a.name for a in result}
107 assert "Radiohead" in names
108 assert "Portishead" in names
109
110
111@pytest.mark.asyncio
112async def test_get_similar_artists_respects_limit(
113 manager: AppleMusicRecommendationManager,
114 mock_api: MagicMock,
115) -> None:
116 """get_similar_artists truncates results to the requested limit."""
117 many_artists = [_make_artist_obj(str(i), f"Artist {i}") for i in range(10)]
118 mock_api.get_data.return_value = _make_api_response(many_artists)
119
120 result = await manager.get_similar_artists("123", limit=3)
121
122 assert len(result) == 3
123
124
125@pytest.mark.asyncio
126async def test_get_similar_artists_empty_data(
127 manager: AppleMusicRecommendationManager,
128 mock_api: MagicMock,
129) -> None:
130 """get_similar_artists returns empty list when API returns no data."""
131 mock_api.get_data.return_value = {"data": []}
132
133 result = await manager.get_similar_artists("123")
134
135 assert result == []
136
137
138@pytest.mark.asyncio
139async def test_get_similar_artists_missing_view(
140 manager: AppleMusicRecommendationManager,
141 mock_api: MagicMock,
142) -> None:
143 """get_similar_artists returns empty list when similar-artists view is absent."""
144 mock_api.get_data.return_value = {"data": [{"id": "123", "views": {}}]}
145
146 result = await manager.get_similar_artists("123")
147
148 assert result == []
149
150
151@pytest.mark.asyncio
152async def test_get_similar_artists_api_error(
153 manager: AppleMusicRecommendationManager,
154 mock_api: MagicMock,
155) -> None:
156 """get_similar_artists propagates exceptions so they are not cached."""
157 mock_api.get_data.side_effect = Exception("API error")
158
159 with pytest.raises(Exception, match="API error"):
160 await manager.get_similar_artists("123")
161
162
163@pytest.mark.asyncio
164async def test_get_similar_tracks_returns_tracks(
165 manager: AppleMusicRecommendationManager,
166 mock_api: MagicMock,
167) -> None:
168 """get_similar_tracks returns one limited response from the Apple API."""
169 track_ids = [str(index) for index in range(1, 9)]
170 mock_api.post_data.return_value = {"data": [_make_track_obj(item_id) for item_id in track_ids]}
171 mock_api.get_ratings.return_value = dict.fromkeys(track_ids, False)
172
173 result = await manager.get_similar_tracks("123", limit=7)
174
175 mock_api.post_data.assert_awaited_once_with("me/stations/next-tracks/ra.123", include="artists")
176 mock_api.get_ratings.assert_awaited_once_with(track_ids[:7], MediaType.TRACK)
177 assert [track.item_id for track in result] == track_ids[:7]
178
179
180@pytest.mark.asyncio
181async def test_get_similar_tracks_apple_500_returns_empty(
182 manager: AppleMusicRecommendationManager,
183 mock_api: MagicMock,
184) -> None:
185 """Apple 500 on similar tracks should not bubble up to websocket handlers."""
186 request_info = RequestInfo(
187 url=URL("https://api.music.apple.com/v1/me/stations/next-tracks/ra.i.test"),
188 method="POST",
189 headers=CIMultiDictProxy(CIMultiDict()),
190 real_url=URL("https://api.music.apple.com/v1/me/stations/next-tracks/ra.i.test"),
191 )
192 mock_api.post_data.side_effect = ClientResponseError(
193 request_info=request_info,
194 history=(),
195 status=500,
196 message="Internal Server Error",
197 headers=None,
198 )
199
200 result = await manager.get_similar_tracks("i.test", limit=5)
201
202 assert result == []
203