/
/
/
1"""Test Bandcamp Provider integration."""
2
3from unittest.mock import AsyncMock, Mock, patch
4
5import pytest
6from bandcamp_async_api import BandcampAPIError, BandcampNotFoundError
7from music_assistant_models.enums import MediaType, StreamType
8from music_assistant_models.errors import InvalidDataError, MediaNotFoundError
9from music_assistant_models.streamdetails import StreamDetails
10
11from music_assistant.providers.bandcamp import DEFAULT_TOP_TRACKS_LIMIT, BandcampProvider
12
13
14@pytest.fixture
15def mass_mock() -> Mock:
16 """Return a mock MusicAssistant instance."""
17 mass = Mock()
18 mass.http_session = AsyncMock()
19 mass.metadata.locale = "en_US"
20 mass.cache.get = AsyncMock(return_value=None)
21 mass.cache.set = AsyncMock()
22 mass.cache.delete = AsyncMock()
23 return mass
24
25
26@pytest.fixture
27def manifest_mock() -> Mock:
28 """Return a mock provider manifest."""
29 manifest = Mock()
30 manifest.domain = "bandcamp"
31 return manifest
32
33
34@pytest.fixture
35def config_mock() -> Mock:
36 """Return a mock provider config."""
37 config = Mock()
38 config.name = "Bandcamp Test"
39 config.instance_id = "bandcamp_test"
40 config.enabled = True
41 config.get_value.side_effect = lambda key, default=None: {
42 "identity": "mock_identity_token",
43 "search_limit": 10,
44 "top_tracks_limit": 50,
45 "log_level": "INFO",
46 }.get(
47 key,
48 default
49 if default is not None
50 else (10 if key == "search_limit" else (50 if key == "top_tracks_limit" else "INFO")),
51 )
52 return config
53
54
55@pytest.fixture
56async def provider(mass_mock: Mock, manifest_mock: Mock, config_mock: Mock) -> BandcampProvider:
57 """Return a BandcampProvider instance."""
58 provider = BandcampProvider(mass_mock, manifest_mock, config_mock)
59
60 # Initialize the provider
61 with patch("music_assistant.providers.bandcamp.BandcampAPIClient") as mock_client_class:
62 mock_client = AsyncMock()
63 mock_client_class.return_value = mock_client
64 await provider.handle_async_init()
65
66 return provider
67
68
69async def test_provider_initialization(
70 mass_mock: Mock, manifest_mock: Mock, config_mock: Mock
71) -> None:
72 """Test provider initialization."""
73 provider = BandcampProvider(mass_mock, manifest_mock, config_mock)
74
75 assert provider.domain == "bandcamp"
76 assert provider.instance_id == "bandcamp_test"
77
78 # Test that initialization sets the correct values
79 with patch("music_assistant.providers.bandcamp.BandcampAPIClient") as mock_client_class:
80 mock_client = AsyncMock()
81 mock_client_class.return_value = mock_client
82
83 await provider.handle_async_init()
84
85 assert provider.top_tracks_limit == DEFAULT_TOP_TRACKS_LIMIT
86
87
88async def test_handle_async_init_with_identity(provider: BandcampProvider) -> None:
89 """Test successful async initialization with identity token."""
90 with patch("music_assistant.providers.bandcamp.BandcampAPIClient") as mock_client_class:
91 mock_client = AsyncMock()
92 mock_client_class.return_value = mock_client
93
94 await provider.handle_async_init()
95
96 mock_client_class.assert_called_once_with(
97 session=provider.mass.http_session,
98 identity_token="mock_identity_token",
99 default_retry_after=3,
100 )
101 assert provider._client == mock_client
102 assert provider._converters is not None
103
104
105async def test_handle_async_init_without_identity(mass_mock: Mock, manifest_mock: Mock) -> None:
106 """Test async initialization without identity token."""
107 config = Mock()
108 config.get_value.side_effect = (
109 lambda key, default=None: default
110 if default is not None
111 else ("INFO" if key == "log_level" else None)
112 )
113 provider = BandcampProvider(mass_mock, manifest_mock, config)
114
115 with patch("music_assistant.providers.bandcamp.BandcampAPIClient") as mock_client_class:
116 mock_client = AsyncMock()
117 mock_client_class.return_value = mock_client
118
119 await provider.handle_async_init()
120
121 mock_client_class.assert_called_once_with(
122 session=provider.mass.http_session,
123 identity_token=None,
124 default_retry_after=3,
125 )
126
127
128async def test_is_streaming_provider(provider: BandcampProvider) -> None:
129 """Test that Bandcamp is not a streaming provider."""
130 assert provider.is_streaming_provider is True
131
132
133async def test_search_with_identity(provider: BandcampProvider) -> None:
134 """Test search functionality with identity token."""
135
136 # Create mock objects with proper class names
137 class MockSearchResultTrack:
138 def __init__(self) -> None:
139 self.__class__.__name__ = "SearchResultTrack"
140
141 class MockSearchResultAlbum:
142 def __init__(self) -> None:
143 self.__class__.__name__ = "SearchResultAlbum"
144
145 class MockSearchResultArtist:
146 def __init__(self) -> None:
147 self.__class__.__name__ = "SearchResultArtist"
148
149 mock_search_results = [
150 MockSearchResultTrack(),
151 MockSearchResultAlbum(),
152 MockSearchResultArtist(),
153 ]
154
155 with (
156 patch.object(provider._client, "search", new_callable=AsyncMock) as mock_search,
157 patch.object(provider._converters, "track_from_search") as mock_track_converter,
158 patch.object(provider._converters, "album_from_search") as mock_album_converter,
159 patch.object(provider._converters, "artist_from_search") as mock_artist_converter,
160 ):
161 mock_search.return_value = mock_search_results
162
163 mock_track_converter.return_value = Mock()
164 mock_album_converter.return_value = Mock()
165 mock_artist_converter.return_value = Mock()
166
167 results = await provider.search(
168 "test query", [MediaType.TRACK, MediaType.ALBUM, MediaType.ARTIST], limit=5
169 )
170
171 mock_search.assert_called_once_with("test query")
172 assert results.tracks is not None
173 assert results.albums is not None
174 assert results.artists is not None
175
176
177async def test_search_without_identity(provider: BandcampProvider) -> None:
178 """Test search returns empty results without identity token."""
179 provider._client.identity = None
180
181 results = await provider.search("test query", [MediaType.TRACK])
182
183 assert len(results.tracks) == 0
184 assert len(results.albums) == 0
185 assert len(results.artists) == 0
186
187
188async def test_search_api_error(provider: BandcampProvider) -> None:
189 """Test search handles API errors gracefully."""
190 with (
191 patch.object(provider._client, "search", side_effect=BandcampAPIError("API Error")),
192 pytest.raises(InvalidDataError, match="Unexpected error during Bandcamp search"),
193 ):
194 await provider.search("test query", [MediaType.TRACK])
195
196
197async def test_get_artist_success(provider: BandcampProvider) -> None:
198 """Test successful artist retrieval."""
199 mock_artist = Mock()
200
201 with (
202 patch.object(provider._client, "get_artist", new_callable=AsyncMock) as mock_get_artist,
203 patch.object(provider._converters, "artist_from_api") as mock_converter,
204 ):
205 mock_get_artist.return_value = mock_artist
206 mock_converter.return_value = Mock()
207
208 result = await provider.get_artist("123")
209
210 mock_get_artist.assert_called_once_with("123")
211 mock_converter.assert_called_once_with(mock_artist)
212 assert result is not None
213
214
215async def test_get_artist_not_found(provider: BandcampProvider) -> None:
216 """Test artist retrieval when not found."""
217 with (
218 patch.object(
219 provider._client, "get_artist", side_effect=BandcampNotFoundError("Not found")
220 ),
221 pytest.raises(MediaNotFoundError, match=r"Bandcamp artist 123 search returned no results"),
222 ):
223 await provider.get_artist("123")
224
225
226async def test_get_album_success(provider: BandcampProvider) -> None:
227 """Test successful album retrieval."""
228 mock_album = Mock()
229
230 with (
231 patch.object(provider._client, "get_album", new_callable=AsyncMock) as mock_get_album,
232 patch.object(provider._converters, "album_from_api") as mock_converter,
233 ):
234 mock_get_album.return_value = mock_album
235 mock_converter.return_value = Mock()
236
237 result = await provider.get_album("123-456")
238
239 mock_get_album.assert_called_once_with(123, 456)
240 assert result is not None
241
242
243async def test_get_track_success(provider: BandcampProvider) -> None:
244 """Test successful track retrieval."""
245 mock_album = Mock()
246 mock_track = Mock()
247 mock_album.tracks = [mock_track]
248 mock_track.id = 789
249
250 with (
251 patch.object(provider._client, "get_album", new_callable=AsyncMock) as mock_get_album,
252 patch.object(provider._converters, "track_from_api") as mock_converter,
253 ):
254 mock_get_album.return_value = mock_album
255 mock_converter.return_value = Mock()
256
257 result = await provider.get_track("123-456-789")
258
259 mock_get_album.assert_called_once_with(123, 456)
260 assert result is not None
261
262
263async def test_get_track_not_found(provider: BandcampProvider) -> None:
264 """Test track retrieval when not found."""
265 with (
266 patch.object(provider._client, "get_album", side_effect=BandcampNotFoundError("Not found")),
267 pytest.raises(
268 MediaNotFoundError, match=r"Bandcamp track 123-456-789 search returned no results"
269 ),
270 ):
271 await provider.get_track("123-456-789")
272
273
274async def test_get_album_tracks_success(provider: BandcampProvider) -> None:
275 """Test successful album tracks retrieval."""
276 mock_album = Mock()
277 mock_track = Mock()
278 mock_track.streaming_url = {"mp3-128": "http://example.com/track.mp3"}
279 mock_album.tracks = [mock_track]
280 mock_album.title = "Test Album"
281 mock_album.art_url = "http://example.com/art.jpg"
282
283 with (
284 patch.object(provider._client, "get_album", new_callable=AsyncMock) as mock_get_album,
285 patch.object(provider._converters, "track_from_api") as mock_converter,
286 ):
287 mock_get_album.return_value = mock_album
288 mock_converter.return_value = Mock()
289
290 result = await provider.get_album_tracks("123-456")
291
292 assert len(result) == 1
293 mock_converter.assert_called_once()
294
295
296async def test_get_artist_albums_success(provider: BandcampProvider) -> None:
297 """Test successful artist albums retrieval."""
298 mock_discography = [{"item_type": "album", "band_id": 123, "item_id": 456}]
299
300 with (
301 patch.object(
302 provider._client, "get_artist_discography", new_callable=AsyncMock
303 ) as mock_get_discography,
304 patch.object(provider, "get_album", new_callable=AsyncMock) as mock_get_album,
305 ):
306 mock_get_discography.return_value = mock_discography
307 mock_get_album.return_value = Mock()
308
309 result = await provider.get_artist_albums("123")
310
311 mock_get_discography.assert_called_once_with("123")
312 assert len(result) == 1
313
314
315async def test_get_stream_details_success(provider: BandcampProvider) -> None:
316 """Test successful stream details retrieval."""
317 # Create mock album and track with proper attributes
318 mock_artist = Mock()
319 mock_artist.id = 123
320 mock_artist.name = "Test Artist"
321
322 mock_track = Mock()
323 mock_track.id = 789
324 mock_track.artist = mock_artist
325 mock_track.title = "Test Track"
326 mock_track.duration = 180
327 mock_track.track_number = 1
328 mock_track.streaming_url = {"mp3-320": "http://example.com/track.mp3"}
329 mock_track.url = "http://example.com/track"
330 mock_track.lyrics = None
331
332 mock_album = Mock()
333 mock_album.id = 456
334 mock_album.title = "Test Album"
335 mock_album.art_url = "http://example.com/art.jpg"
336 mock_album.artist = mock_artist
337 mock_album.tracks = [mock_track]
338
339 with (
340 patch.object(provider._client, "get_album", new_callable=AsyncMock) as mock_get_album,
341 patch.object(provider._converters, "track_from_api") as mock_converter,
342 ):
343 mock_get_album.return_value = mock_album
344
345 # Create a mock track with metadata.links containing the streaming URL
346 mock_ma_track = Mock()
347 mock_link = Mock()
348 mock_link.url = "http://example.com/track.mp3"
349 mock_ma_track.metadata.links = {mock_link}
350 mock_converter.return_value = mock_ma_track
351
352 result = await provider.get_stream_details("123-456-789", MediaType.TRACK)
353
354 assert isinstance(result, StreamDetails)
355 assert result.stream_type == StreamType.HTTP
356 assert result.path == "http://example.com/track.mp3"
357
358
359async def test_get_stream_details_no_streaming_url(provider: BandcampProvider) -> None:
360 """Test stream details when no streaming URL is available."""
361 # Mock the get_track method directly to return a track with no streaming URLs
362 mock_track = Mock()
363 mock_track.metadata.links = [] # Empty links list means no streaming URL
364
365 with patch.object(provider, "get_track", new_callable=AsyncMock) as mock_get_track:
366 mock_get_track.return_value = mock_track
367
368 with pytest.raises(
369 MediaNotFoundError,
370 match=r"No streaming links found for track 123-456-789. Please report this",
371 ):
372 await provider.get_stream_details("123-456-789", MediaType.TRACK)
373
374
375async def test_get_artist_toptracks_success(provider: BandcampProvider) -> None:
376 """Test successful artist top tracks retrieval."""
377 mock_album = Mock()
378 mock_track = Mock()
379
380 with (
381 patch.object(provider, "get_artist_albums", new_callable=AsyncMock) as mock_get_albums,
382 patch.object(provider, "get_album_tracks", new_callable=AsyncMock) as mock_get_tracks,
383 ):
384 mock_get_albums.return_value = [mock_album]
385 mock_get_tracks.return_value = [mock_track]
386
387 result = await provider.get_artist_toptracks("123")
388
389 assert len(result) == 1
390 mock_get_albums.assert_called_once_with("123")
391
392
393async def test_get_library_artists_success(provider: BandcampProvider) -> None:
394 """Test successful library artists retrieval."""
395 # Test that the method exists and doesn't raise an exception
396 # This is a complex async generator method, so we just test it can be called
397 assert hasattr(provider, "get_library_artists")
398 assert callable(provider.get_library_artists)
399
400
401async def test_get_library_albums_success(provider: BandcampProvider) -> None:
402 """Test successful library albums retrieval."""
403 # Test that the method exists and doesn't raise an exception
404 # This is a complex async generator method, so we just test it can be called
405 assert hasattr(provider, "get_library_albums")
406 assert callable(provider.get_library_albums)
407
408
409async def test_get_library_tracks_success(provider: BandcampProvider) -> None:
410 """Test successful library tracks retrieval."""
411 # Test that the method exists and doesn't raise an exception
412 # This is a complex async generator method, so we just test it can be called
413 assert hasattr(provider, "get_library_tracks")
414 assert callable(provider.get_library_tracks)
415