"""Tests for the 'spotify' plugin""" import json from urllib.parse import parse_qs, urlparse import responses from beets.library import Item from beets.test import _common from beets.test.helper import PluginTestCase from beetsplug import spotify class ArgumentsMock: def __init__(self, mode, show_failures): self.mode = mode self.show_failures = show_failures self.verbose = 1 def _params(url): """Get the query parameters from a URL.""" return parse_qs(urlparse(url).query) class SpotifyPluginTest(PluginTestCase): plugin = "spotify" db_on_disk = True @responses.activate def setUp(self): responses.add( responses.POST, spotify.SpotifyPlugin.oauth_token_url, status=200, json={ "access_token": "3XyiC3raJySbIAV5LVYj1DaWbcocNi3LAJTNXRnYY" "GVUl6mbbqXNhW3YcZnQgYXNWHFkVGSMlc0tMuvq8CF", "token_type": "Bearer", "expires_in": 3600, "scope": "", }, ) super().setUp() self.spotify = spotify.SpotifyPlugin() opts = ArgumentsMock("list", False) self.spotify._parse_opts(opts) def test_args(self): opts = ArgumentsMock("fail", True) assert not self.spotify._parse_opts(opts) opts = ArgumentsMock("list", False) assert self.spotify._parse_opts(opts) def test_empty_query(self): assert self.spotify._match_library_tracks(self.lib, "1=2") is None @responses.activate def test_missing_request(self): json_file = _common.RSRC / "spotify" / "missing_request.json" response_body = json_file.read_bytes() responses.add( responses.GET, spotify.SpotifyPlugin.search_url, body=response_body, status=200, content_type="application/json", ) item = Item( mb_trackid="01234", album="lkajsdflakjsd", albumartist="ujydfsuihse", title="duifhjslkef", length=10, ) item.add(self.lib) assert [] == self.spotify._match_library_tracks(self.lib, "") params = _params(responses.calls[0].request.url) query = params["q"][0] assert "duifhjslkef" in query assert "artist:'ujydfsuihse'" in query assert "album:'lkajsdflakjsd'" in query assert params["type"] == ["track"] @responses.activate def test_track_request(self): json_file = _common.RSRC / "spotify" / "track_request.json" response_body = json_file.read_bytes() responses.add( responses.GET, spotify.SpotifyPlugin.search_url, body=response_body, status=200, content_type="application/json", ) item = Item( mb_trackid="01234", album="Despicable Me 2", albumartist="Pharrell Williams", title="Happy", length=10, ) item.add(self.lib) results = self.spotify._match_library_tracks(self.lib, "Happy") assert 1 == len(results) assert "6NPVjNh8Jhru9xOmyQigds" == results[0]["id"] self.spotify._output_match_results(results) params = _params(responses.calls[0].request.url) query = params["q"][0] assert "Happy" in query assert "artist:'Pharrell Williams'" in query assert "album:'Despicable Me 2'" in query assert params["type"] == ["track"] @responses.activate def test_track_for_id(self): """Tests if plugin is able to fetch a track by its Spotify ID""" # Mock the Spotify 'Get Track' call json_file = _common.RSRC / "spotify" / "track_info.json" response_body = json_file.read_bytes() responses.add( responses.GET, f"{spotify.SpotifyPlugin.track_url}/6NPVjNh8Jhru9xOmyQigds", body=response_body, status=200, content_type="application/json", ) # Mock the Spotify 'Get Album' call json_file = _common.RSRC / "spotify" / "album_info.json" response_body = json_file.read_bytes() responses.add( responses.GET, f"{spotify.SpotifyPlugin.album_url}5l3zEmMrOhOzG8d8s83GOL", body=response_body, status=200, content_type="application/json", ) # Mock the Spotify 'Search' call json_file = _common.RSRC / "spotify" / "track_request.json" response_body = json_file.read_bytes() responses.add( responses.GET, spotify.SpotifyPlugin.search_url, body=response_body, status=200, content_type="application/json", ) track_info = self.spotify.track_for_id("6NPVjNh8Jhru9xOmyQigds") item = Item( mb_trackid=track_info.track_id, albumartist=track_info.artist, title=track_info.title, length=track_info.length, ) item.add(self.lib) results = self.spotify._match_library_tracks(self.lib, "Happy") assert 1 == len(results) assert "6NPVjNh8Jhru9xOmyQigds" == results[0]["id"] @responses.activate def test_japanese_track(self): """Ensure non-ASCII characters remain unchanged in search queries""" # Path to the mock JSON file for the Japanese track json_file = _common.RSRC / "spotify" / "japanese_track_request.json" # Load the mock JSON response response_body = json_file.read_bytes() # Mock Spotify Search API response responses.add( responses.GET, spotify.SpotifyPlugin.search_url, body=response_body, status=200, content_type="application/json", ) # Create a mock item with Japanese metadata item = Item( mb_trackid="56789", album="盗作", albumartist="ヨルシカ", title="思想犯", length=10, ) item.add(self.lib) # Search without ascii encoding with self.configure_plugin({"search_query_ascii": False}): assert self.spotify.config["search_query_ascii"].get() is False # Call the method to match library tracks results = self.spotify._match_library_tracks(self.lib, item.title) # Assertions to verify results assert results is not None assert 1 == len(results) assert results[0]["name"] == item.title assert results[0]["artists"][0]["name"] == item.albumartist assert results[0]["album"]["name"] == item.album # Verify search query parameters params = _params(responses.calls[0].request.url) query = params["q"][0] assert item.title in query assert f"artist:'{item.albumartist}'" in query assert f"album:'{item.album}'" in query assert not query.isascii() # Is not found in the library if ascii encoding is enabled with self.configure_plugin({"search_query_ascii": True}): assert self.spotify.config["search_query_ascii"].get() is True results = self.spotify._match_library_tracks(self.lib, item.title) params = _params(responses.calls[1].request.url) query = params["q"][0] assert query.isascii() @responses.activate def test_multiartist_album_and_track(self): """Tests if plugin is able to map multiple artists in an album and track info correctly""" # Mock the Spotify 'Get Album' call json_file = _common.RSRC / "spotify" / "multiartist_album.json" album_response_body = json_file.read_bytes() responses.add( responses.GET, f"{spotify.SpotifyPlugin.album_url}0yhKyyjyKXWUieJ4w1IAEa", body=album_response_body, status=200, content_type="application/json", ) # Mock the Spotify 'Get Track' call json_file = _common.RSRC / "spotify" / "multiartist_track.json" track_response_body = json_file.read_bytes() responses.add( responses.GET, f"{spotify.SpotifyPlugin.track_url}/6sjZfVJworBX6TqyjkxIJ1", body=track_response_body, status=200, content_type="application/json", ) album_info = self.spotify.album_for_id("0yhKyyjyKXWUieJ4w1IAEa") assert album_info is not None assert album_info.artist == "Project Skylate, Sugar Shrill" assert album_info.artists == ["Project Skylate", "Sugar Shrill"] assert album_info.artist_id == "6m8MRXIVKb6wQaPlBIDMr1" assert album_info.artists_ids == [ "6m8MRXIVKb6wQaPlBIDMr1", "4kkAIoQmNT5xEoNH5BuQLe", ] assert len(album_info.tracks) == 1 assert album_info.tracks[0].artist == "Foo, Bar" assert album_info.tracks[0].artists == ["Foo", "Bar"] assert album_info.tracks[0].artist_id == "12345" assert album_info.tracks[0].artists_ids == ["12345", "67890"] track_info = self.spotify.track_for_id("6sjZfVJworBX6TqyjkxIJ1") assert track_info is not None assert track_info.artist == "Foo, Bar" assert track_info.artists == ["Foo", "Bar"] assert track_info.artist_id == "12345" assert track_info.artists_ids == ["12345", "67890"] @responses.activate def test_get_track_details_by_id_chunks_requests(self): ids_per_request = [] def callback(request): ids = _params(request.url)["ids"][0].split(",") ids_per_request.append(len(ids)) return ( 200, {"Content-Type": "application/json"}, json.dumps( { "tracks": [ { "id": track_id, "popularity": 50, "external_ids": {}, } for track_id in ids ] } ), ) responses.add_callback( responses.GET, spotify.SpotifyPlugin.track_url, callback=callback, content_type="application/json", ) track_ids = [f"track-{idx}" for idx in range(51)] track_info = self.spotify.get_track_details_by_id(track_ids) assert len(track_info) == 51 assert ids_per_request == [50, 1] @responses.activate def test_fetch_info_uses_batch_endpoints(self): responses.add( responses.GET, spotify.SpotifyPlugin.track_url, status=200, json={ "tracks": [ { "id": "id-1", "popularity": 10, "external_ids": { "isrc": "isrc-1", "ean": "ean-1", "upc": "upc-1", }, }, { "id": "id-2", "popularity": 20, "external_ids": { "isrc": "isrc-2", "ean": "ean-2", "upc": "upc-2", }, }, { "id": "id-3", "popularity": 30, "external_ids": { "isrc": "isrc-3", "ean": "ean-3", "upc": "upc-3", }, }, ] }, content_type="application/json", ) responses.add( responses.GET, spotify.SpotifyPlugin.audio_features_url, status=200, json={ "audio_features": [ {"id": "id-1", "tempo": 100.1, "energy": 0.4}, {"id": "id-2", "tempo": 110.2, "energy": 0.5}, {"id": "id-3", "tempo": 120.3, "energy": 0.6}, ] }, content_type="application/json", ) items = [] for idx in range(1, 4): item = Item(title=f"Track {idx}", artist="Artist", length=10) item.add(self.lib) item["spotify_track_id"] = f"id-{idx}" items.append(item) self.spotify._fetch_info(self.lib, items, write=False, force=True) get_calls = [ call for call in responses.calls if call.request.method == "GET" ] batch_track_calls = [ call for call in get_calls if urlparse(call.request.url).path == "/v1/tracks" ] single_track_calls = [ call for call in get_calls if urlparse(call.request.url).path.startswith("/v1/tracks/") ] batch_audio_calls = [ call for call in get_calls if urlparse(call.request.url).path == "/v1/audio-features" ] single_audio_calls = [ call for call in get_calls if urlparse(call.request.url).path.startswith("/v1/audio-features/") ] assert len(batch_track_calls) == 1 assert len(single_track_calls) == 0 assert len(batch_audio_calls) == 1 assert len(single_audio_calls) == 0 assert items[0]["spotify_track_popularity"] == 10 assert items[1]["spotify_track_popularity"] == 20 assert items[2]["spotify_track_popularity"] == 30 assert items[0]["spotify_tempo"] == 100.1 assert items[1]["spotify_tempo"] == 110.2 assert items[2]["spotify_tempo"] == 120.3 @responses.activate def test_fetch_info_deduplicates_batch_ids(self): seen_track_ids = [] seen_audio_ids = [] def track_callback(request): ids = _params(request.url)["ids"][0].split(",") seen_track_ids.append(ids) return ( 200, {"Content-Type": "application/json"}, json.dumps( { "tracks": [ { "id": track_id, "popularity": 50, "external_ids": {}, } for track_id in ids ] } ), ) def audio_callback(request): ids = _params(request.url)["ids"][0].split(",") seen_audio_ids.append(ids) return ( 200, {"Content-Type": "application/json"}, json.dumps( { "audio_features": [ {"id": track_id, "tempo": 100.0} for track_id in ids ] } ), ) responses.add_callback( responses.GET, spotify.SpotifyPlugin.track_url, callback=track_callback, content_type="application/json", ) responses.add_callback( responses.GET, spotify.SpotifyPlugin.audio_features_url, callback=audio_callback, content_type="application/json", ) items = [] for idx in range(2): item = Item(title=f"Track {idx}", artist="Artist", length=10) item.add(self.lib) item["spotify_track_id"] = "shared-id" items.append(item) self.spotify._fetch_info(self.lib, items, write=False, force=True) assert seen_track_ids == [["shared-id"]] assert seen_audio_ids == [["shared-id"]] assert items[0]["spotify_track_popularity"] == 50 assert items[1]["spotify_track_popularity"] == 50 @responses.activate def test_fetch_info_writes_before_store(self): responses.add( responses.GET, spotify.SpotifyPlugin.track_url, status=200, json={ "tracks": [{"id": "id-1", "popularity": 10, "external_ids": {}}] }, content_type="application/json", ) responses.add( responses.GET, spotify.SpotifyPlugin.audio_features_url, status=200, json={"audio_features": [{"id": "id-1", "tempo": 100.1}]}, content_type="application/json", ) item = self.add_item_fixture(title="Track 1", artist="Artist") item["spotify_track_id"] = "id-1" with self.assertLogs("beets", level="DEBUG") as captured_logs: self.spotify._fetch_info(self.lib, [item], write=True, force=True) logs = captured_logs.output write_event_index = next( idx for idx, message in enumerate(logs) if message.endswith("Sending event: write") ) database_change_index = next( idx for idx, message in enumerate(logs) if message.endswith("Sending event: database_change") ) assert write_event_index < database_change_index @responses.activate def test_track_audio_features_batch_disables_on_403(self): responses.add( responses.GET, spotify.SpotifyPlugin.audio_features_url, status=403, json={"error": {"status": 403}}, content_type="application/json", ) assert self.spotify.track_audio_features_batch(["id-1"]) == {} assert self.spotify.audio_features_available is False assert self.spotify.track_audio_features_batch(["id-2"]) == {} assert len(responses.calls) == 1 @responses.activate def test_track_audio_features_batch_keeps_partial_results_on_api_error( self, ): def callback(request): ids = _params(request.url)["ids"][0].split(",") if "track-100" in ids: return ( 502, {"Content-Type": "application/json"}, json.dumps({"error": {"status": 502}}), ) return ( 200, {"Content-Type": "application/json"}, json.dumps( { "audio_features": [ {"id": track_id, "tempo": 100.0} for track_id in ids ] } ), ) responses.add_callback( responses.GET, spotify.SpotifyPlugin.audio_features_url, callback=callback, content_type="application/json", ) track_ids = [f"track-{idx}" for idx in range(201)] features = self.spotify.track_audio_features_batch(track_ids) assert "track-0" in features assert "track-99" in features assert "track-100" not in features assert "track-199" not in features assert "track-200" in features