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