music-assistant-server

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