diff --git a/musify/local/library/library.py b/musify/local/library/library.py index 9a893912..19a17ef3 100644 --- a/musify/local/library/library.py +++ b/musify/local/library/library.py @@ -60,12 +60,12 @@ class LocalLibrary(LocalCollection[LocalTrack], Library[LocalTrack]): ) __attributes_classes__ = (Library, LocalCollection) - # noinspection PyTypeChecker,PyPropertyDefinition + # noinspection PyPropertyDefinition @classmethod @property def name(cls) -> str: """The type of library loaded""" - return cls.source + return str(cls.source) # noinspection PyPropertyDefinition @classmethod diff --git a/musify/local/track/mp3.py b/musify/local/track/mp3.py index f2ce6f69..2284dd6d 100644 --- a/musify/local/track/mp3.py +++ b/musify/local/track/mp3.py @@ -88,8 +88,6 @@ def _write_genres(self, track: LocalTrack, dry_run: bool = True) -> bool: def _write_images(self, track: LocalTrack, dry_run: bool = True) -> bool: tag_id_prefix = next(iter(self.tag_map.images), None) - print(self.file.tags.keys()) - updated = False for image_kind, image_link in track.image_links.items(): image = open_image(image_link) @@ -112,7 +110,6 @@ def _write_images(self, track: LocalTrack, dry_run: bool = True) -> bool: track.has_image = tag_id_prefix is not None or track.has_image updated = tag_id_prefix is not None - print(self.file.tags.keys()) return updated def _write_comments(self, track: LocalTrack, dry_run: bool = True) -> bool: diff --git a/musify/shared/remote/base.py b/musify/shared/remote/base.py index 7108bd9f..0294b0c1 100644 --- a/musify/shared/remote/base.py +++ b/musify/shared/remote/base.py @@ -114,7 +114,7 @@ def load( def reload(self, use_cache: bool = True, *args, **kwargs) -> None: """ Reload this object from the API, calling all required endpoints - to get a complete set of data for this item type + to get a complete set of data for this item type. :param use_cache: Use the cache when calling the API endpoint. Set as False to refresh the cached response. """ diff --git a/musify/shared/remote/object.py b/musify/shared/remote/object.py index 104cd194..a23ce4ef 100644 --- a/musify/shared/remote/object.py +++ b/musify/shared/remote/object.py @@ -29,8 +29,10 @@ class RemoteTrack(RemoteItem, Track, metaclass=ABCMeta): __attributes_classes__ = (Track, RemoteItem) + # noinspection PyPropertyDefinition + @classmethod @property - def kind(self): + def kind(cls): return RemoteObjectType.TRACK @@ -78,6 +80,9 @@ def load( * An object of the same type as this collection. The remote API JSON response will be used to load a new object. + You may also provide a set of kwargs relating that will extend aspects of the response + before using it to initialise a new object. See :py:meth:`reload` for possible extensions. + :param value: The value representing some remote collection. See description for allowed value types. :param api: An authorised API object to load the object from. :param use_cache: Use the cache when calling the API endpoint. Set as False to refresh the cached response. @@ -125,8 +130,10 @@ class RemotePlaylist[T: RemoteTrack](Playlist[T], RemoteCollectionLoader[T], met __attributes_classes__ = (Playlist, RemoteCollectionLoader) + # noinspection PyPropertyDefinition + @classmethod @property - def kind(self): + def kind(cls): return RemoteObjectType.PLAYLIST @property @@ -241,7 +248,7 @@ def sync( if not dry_run: added = self.api.add_to_playlist(self.url, items=uri_add, skip_dupes=kind != "refresh") if reload: # reload the current playlist object from remote - self.reload(use_cache=False) + self.reload(use_cache=False, extend_tracks=True) return SyncResultRemotePlaylist( start=len(uri_remote), @@ -269,8 +276,10 @@ class RemoteAlbum[T: RemoteTrack](RemoteCollectionLoader[T], Album[T], metaclass __attributes_classes__ = (Album, RemoteCollectionLoader) + # noinspection PyPropertyDefinition + @classmethod @property - def kind(self): + def kind(cls): return RemoteObjectType.ALBUM @property @@ -288,8 +297,10 @@ class RemoteArtist[T: RemoteTrack](Artist[T], RemoteCollectionLoader[T], metacla __attributes_classes__ = (Artist, RemoteCollectionLoader) + # noinspection PyPropertyDefinition + @classmethod @property - def kind(self): + def kind(cls): return RemoteObjectType.ARTIST @property diff --git a/musify/shared/remote/response.py b/musify/shared/remote/response.py index 396a31f8..999130b8 100644 --- a/musify/shared/remote/response.py +++ b/musify/shared/remote/response.py @@ -24,9 +24,10 @@ def id(self) -> str: """The ID of this item/collection.""" raise NotImplementedError + # noinspection PyPropertyDefinition,PyMethodParameters @property @abstractmethod - def kind(self) -> RemoteObjectType: + def kind(cls) -> RemoteObjectType: """The type of remote object this python object represents""" raise NotImplementedError diff --git a/musify/spotify/api/api.py b/musify/spotify/api/api.py index b9cbd291..ea300941 100644 --- a/musify/spotify/api/api.py +++ b/musify/spotify/api/api.py @@ -71,8 +71,6 @@ class SpotifyAPI(SpotifyAPIMisc, SpotifyAPIItems, SpotifyAPIPlaylists): :param scopes: The scopes to request access to. """ - items_key = "items" - @property def user_id(self) -> str | None: """ID of the currently authenticated user""" diff --git a/musify/spotify/api/item.py b/musify/spotify/api/item.py index b7a50a48..4f072ff3 100644 --- a/musify/spotify/api/item.py +++ b/musify/spotify/api/item.py @@ -176,6 +176,11 @@ def extend_items( if self.items_key not in response: response[self.items_key] = [] + if len(response[self.items_key]) == response["total"]: # skip on fully extended response + url = response["href"].split("?")[0] + self.logger.debug(f"{'SKIP':<7}: {url:<43} | Response already extended") + return response[self.items_key] + # this usually happens on the items block of a current user's playlist if "next" not in response: response["next"] = response["href"] diff --git a/musify/spotify/api/playlist.py b/musify/spotify/api/playlist.py index f371111d..2ceaa6f1 100644 --- a/musify/spotify/api/playlist.py +++ b/musify/spotify/api/playlist.py @@ -48,6 +48,7 @@ def get_playlist_url(self, playlist: str | Mapping[str, Any] | RemoteResponse, u return self.wrangler.convert( playlist["uri"], kind=RemoteObjectType.PLAYLIST, type_in=RemoteIDType.URI, type_out=RemoteIDType.URL ) + try: return self.wrangler.convert(playlist, kind=RemoteObjectType.PLAYLIST, type_out=RemoteIDType.URL) except RemoteIDTypeError: diff --git a/musify/spotify/base.py b/musify/spotify/base.py index 074a9d37..f8dce479 100644 --- a/musify/spotify/base.py +++ b/musify/spotify/base.py @@ -13,7 +13,7 @@ from musify.spotify.api import SpotifyAPI -class SpotifyObject(RemoteObject, metaclass=ABCMeta): +class SpotifyObject(RemoteObject[SpotifyAPI], metaclass=ABCMeta): """Generic base class for Spotify-stored objects. Extracts key data from a Spotify API JSON response.""" _url_pad = 71 diff --git a/musify/spotify/object.py b/musify/spotify/object.py index cb4c4a04..3983260b 100644 --- a/musify/spotify/object.py +++ b/musify/spotify/object.py @@ -4,8 +4,8 @@ from __future__ import annotations -from abc import ABCMeta -from collections.abc import Iterable, MutableMapping, Mapping +from abc import ABCMeta, abstractmethod +from collections.abc import Iterable, MutableMapping, Mapping, Collection from copy import copy, deepcopy from datetime import datetime from typing import Any, Self @@ -237,11 +237,11 @@ def load( def reload( self, + use_cache: bool = True, features: bool = False, analysis: bool = False, extend_album: bool = False, extend_artists: bool = False, - use_cache: bool = True, *_, **__ ) -> None: @@ -259,97 +259,138 @@ def reload( self.__init__(response=response, api=self.api) -class SpotifyCollectionLoader[T: SpotifyItem](RemoteCollectionLoader[T], SpotifyObject, metaclass=ABCMeta): +class SpotifyCollectionLoader[T: SpotifyObject](RemoteCollectionLoader[T], SpotifyObject, metaclass=ABCMeta): """Generic class for storing a collection of Spotify objects that can be loaded from an API response.""" + @classmethod + def _get_item_kind(cls, api: SpotifyAPI) -> RemoteObjectType: + """Returns the :py:class:`RemoteObjectType` for the items in this collection""" + return api.collection_item_map[cls.kind] + + @classmethod + @abstractmethod + def _get_items( + cls, items: Collection[str] | MutableMapping[str, Any], api: SpotifyAPI, use_cache: bool = True + ) -> list[dict[str, Any]]: + """Call the ``api`` to get values for the given ``items`` URIs""" + raise NotImplementedError + + @classmethod + def _filter_items(cls, items: Iterable[T], response: Mapping[str, Any]) -> Iterable[T]: + """ + Filter down and return only ``items`` associated with the given ``response``. + The default implementation of this will just return the given ``items``. + Override for object-specific filtering. + """ + return items + + @classmethod + def _extend_response( + cls, response: MutableMapping[str, Any], api: SpotifyAPI, use_cache: bool = True, *_, **__ + ) -> bool: + """ + Apply extensions to specific aspects of the given ``response``. + Does nothing by default. Override to implement object-specific extensions. + + :return: True if checks should be skipped when initialising the object. + """ + pass + + @classmethod + def _merge_items_to_response( + cls, + items: Iterable[T | Mapping[str, Any]], + response: Iterable[MutableMapping[str, Any]], + skip: Collection[str] = () + ) -> tuple[set[str], set[str]]: + """ + Find items in the ``response`` that match from the given ``items`` and replace them in the response. + Optionally, provide a list of URIs to ``skip``. + + :return: Two sets of URIs for items that could and could not be found in response. + """ + items_mapped: dict[str, T | Mapping[str, Any]] = { + item.uri if isinstance(item, SpotifyObject) else item["uri"]: item for item in items + } + uri_matched: set[str] = set() + uri_missing: set[str] = set() + + # find items in the response that match from the given items + for source_item in response: + if cls.kind == RemoteObjectType.PLAYLIST: + source_item = source_item["track"] + + if source_item["uri"] in skip: + continue + elif source_item["uri"] in items_mapped: + # replace the skeleton response with the response from the given items + replacement_item = items_mapped.pop(source_item["uri"]) + if isinstance(replacement_item, SpotifyObject): + replacement_item = replacement_item.response + + source_item.clear() + source_item |= replacement_item + uri_matched.add(source_item["uri"]) + elif not source_item.get("is_local"): # add to missing list + uri_missing.add(source_item["uri"]) + + return uri_matched, uri_missing + + @classmethod + def _load_new(cls, value: str | Mapping[str, Any] | RemoteResponse, api: SpotifyAPI, *args, **kwargs) -> Self: + """ + Sets up a new object of the current class for the given ``value`` by calling ``__new__`` + and adding just enough attributes to the object to get :py:meth:`reload` to run. + """ + # noinspection PyTypeChecker + id_ = next(iter(api.wrangler.extract_ids(values=value, kind=cls.kind))) + # noinspection PyTypeChecker + url = api.wrangler.convert(id_, kind=cls.kind, type_in=RemoteIDType.ID, type_out=RemoteIDType.URL) + + self = cls.__new__(cls) + self.api = api + self._response = {"href": url} + + self.reload(*args, **kwargs) + return self + @classmethod def load( cls, value: str | Mapping[str, Any] | RemoteResponse, api: SpotifyAPI, - items: Iterable[SpotifyTrack] = (), - extend_tracks: bool = False, use_cache: bool = True, + items: Iterable[T] = (), leave_bar: bool = True, *args, **kwargs ) -> Self: - unit = cls.__name__.removeprefix("Spotify").lower() - kind = RemoteObjectType.from_name(unit)[0] - key = api.collection_item_map[kind].name.lower() + "s" + item_kind = cls._get_item_kind(api=api) + item_key = item_kind.name.lower() + "s" + + if isinstance(value, RemoteResponse): + value = value.response # no items given, regenerate API response from the URL - if any({ - not items, - isinstance(value, Mapping | RemoteResponse) and (key not in value or api.items_key not in value[key]) - }): - if kind == RemoteObjectType.PLAYLIST: - value = api.get_playlist_url(value) - - self = cls.__new__(cls) - self.api = api - self._response = {"href": value} - self.reload(*args, **kwargs, extend_tracks=extend_tracks, use_cache=use_cache) - return self - - # get response - if isinstance(value, MutableMapping) and api.wrangler.get_item_type(value) == kind: + if any({not items, isinstance(value, Mapping) and api.items_key not in value.get(item_key, [])}): + return cls._load_new(value=value, api=api, use_cache=use_cache, *args, **kwargs) + + if isinstance(value, MutableMapping) and api.wrangler.get_item_type(value) == cls.kind: # input is response response = deepcopy(value) - else: # reload response from the API - response = cls.api.get_items(value, kind=kind, use_cache=use_cache)[0] - - # attempt to find items for this track collection in the given items - if kind == RemoteObjectType.ALBUM: - uri_tracks_input: dict[str, SpotifyTrack] = { - track.uri: track for track in items if track.response.get("album", {}).get("id") == response["id"] - } - else: - uri_tracks_input: dict[str, SpotifyTrack] = {track.uri: track for track in items} - uri_get: list[str] = [] - - # loop through the skeleton response for this album, find items that match from the given items - for track_response in response[key][api.items_key]: - if kind == RemoteObjectType.PLAYLIST: - track_response = track_response["track"] - - if track_response["uri"] in uri_tracks_input: - # replace the skeleton response with the response from the track - track = uri_tracks_input.pop(track_response["uri"]) - track_response.clear() - track_response |= track.response - elif not track_response["is_local"]: # add to get from API list - uri_get.append(track_response["uri"]) - - if len(uri_get) > 0: # get remaining items from API - uri_tracks_get = {r["uri"]: r for r in api.get_tracks(uri_get, features=True, use_cache=use_cache)} - - for track_response in response[key][api.items_key]: - if kind == RemoteObjectType.PLAYLIST: - track_response = track_response["track"] - - track_new: dict[str, Any] = uri_tracks_get.pop(track_response["uri"], None) - if track_new: # replace the skeleton response with the new response - track_response.clear() - track_response |= track_new - - if kind == RemoteObjectType.ALBUM: - if len(response[key][api.items_key]) < response["total_tracks"]: - response[key][api.items_key].extend([track.response for track in uri_tracks_input.values()]) - response[key][api.items_key].sort(key=lambda x: x["track_number"]) - - extend_response = api.extend_items( - response[key], - kind=kind, - key=api.collection_item_map[kind], - use_cache=use_cache, - leave_bar=leave_bar - ) - if extend_tracks: - if kind == RemoteObjectType.PLAYLIST: - extend_response = [item["track"] for item in extend_response] - api.extend_tracks(extend_response, limit=response[key]["limit"], features=True, use_cache=use_cache) + else: # load fresh response from the API + response = cls.api.get_items(value, kind=cls.kind, use_cache=use_cache)[0] + + # filter down input items to those that match the response + items = cls._filter_items(items=items, response=response) + matched, missing = cls._merge_items_to_response(items=items, response=response[item_key][api.items_key]) - return cls(response=response, api=api, skip_checks=False) + if missing: + print(missing) + items_missing = cls._get_items(items=missing, api=api, use_cache=use_cache) + cls._merge_items_to_response(items=items_missing, response=response[item_key][api.items_key], skip=matched) + + skip_checks = cls._extend_response(response=response, api=api, use_cache=use_cache, *args, **kwargs) + return cls(response=response, api=api, skip_checks=skip_checks) class SpotifyPlaylist(SpotifyCollectionLoader[SpotifyTrack], RemotePlaylist[SpotifyTrack]): @@ -464,15 +505,57 @@ def refresh(self, skip_checks: bool = False) -> None: if not skip_checks: self._check_total() - def reload(self, extend_tracks: bool = False, use_cache: bool = True, *_, **__) -> None: - self._check_for_api() + @classmethod + def _get_items( + cls, items: Collection[str] | MutableMapping[str, Any], api: SpotifyAPI, use_cache: bool = True + ) -> list[dict[str, Any]]: + return api.get_tracks(items, use_cache=use_cache) + + @classmethod + def _extend_response( + cls, + response: MutableMapping[str, Any], + api: SpotifyAPI, + use_cache: bool = True, + leave_bar: bool = True, + extend_tracks: bool = False, + extend_features: bool = False, + *_, + **__ + ) -> bool: + item_kind = api.collection_item_map[cls.kind] - response = self.api.get_items(self.url, kind=RemoteObjectType.PLAYLIST, use_cache=use_cache)[0] if extend_tracks: - tracks = [track["track"] for track in response["tracks"]["items"]] - self.api.extend_tracks(tracks, limit=response["tracks"]["limit"], features=True, use_cache=use_cache) + # noinspection PyTypeChecker + api.extend_items(response, kind=cls.kind, key=item_kind, use_cache=use_cache, leave_bar=leave_bar) - self.__init__(response=response, api=self.api, skip_checks=not extend_tracks) + item_key = item_kind.name.lower() + "s" + tracks = [item["track"] for item in response.get(item_key, {}).get(api.items_key, [])] + if tracks and extend_features: + api.extend_tracks(tracks, limit=response[item_key]["limit"], features=True, use_cache=use_cache) + + return not extend_tracks + + def reload( + self, + use_cache: bool = True, + extend_tracks: bool = False, + extend_features: bool = False, + *_, + **__ + ) -> None: + self._check_for_api() + response = self.api.get_items(self.url, kind=RemoteObjectType.PLAYLIST, extend=False, use_cache=use_cache)[0] + + skip_checks = self._extend_response( + response=response, + api=self.api, + use_cache=use_cache, + extend_tracks=extend_tracks, + extend_features=extend_features + ) + + self.__init__(response=response, api=self.api, skip_checks=skip_checks) def _get_track_uris_from_api_response(self) -> list[str]: return [track["track"]["uri"] for track in self.response["tracks"]["items"]] @@ -593,19 +676,68 @@ def refresh(self, skip_checks: bool = False) -> None: for track in self.tracks: track.disc_total = self.disc_total + @classmethod + def _get_items( + cls, items: Collection[str] | MutableMapping[str, Any], api: SpotifyAPI, use_cache: bool = True + ) -> list[dict[str, Any]]: + return api.get_tracks(items, use_cache=use_cache) + + @classmethod + def _filter_items(cls, items: Iterable[SpotifyTrack], response: Mapping[str, Any]) -> Iterable[SpotifyTrack]: + """Filter down and return only ``items`` the items associated with the given ``response``""" + return [item for item in items if item.response.get("album", {}).get("id") == response["id"]] + + @classmethod + def _extend_response( + cls, + response: MutableMapping[str, Any], + api: SpotifyAPI, + use_cache: bool = True, + leave_bar: bool = True, + extend_tracks: bool = False, + extend_artists: bool = False, + extend_features: bool = False, + *_, + **__ + ) -> bool: + item_kind = api.collection_item_map[cls.kind] + + if extend_artists: + api.get_items(response["artists"], kind=RemoteObjectType.ARTIST, use_cache=use_cache) + + if extend_tracks: + # noinspection PyTypeChecker + api.extend_items(response, kind=cls.kind, key=item_kind, use_cache=use_cache, leave_bar=leave_bar) + + item_key = item_kind.name.lower() + "s" + tracks = response.get(item_key, {}).get(api.items_key) + if tracks and extend_features: + api.extend_tracks(tracks, limit=response[item_key]["limit"], features=True, use_cache=use_cache) + + return not extend_tracks + def reload( - self, extend_artists: bool = True, extend_tracks: bool = True, use_cache: bool = True, *_, **__ + self, + use_cache: bool = True, + extend_artists: bool = False, + extend_tracks: bool = False, + extend_features: bool = False, + *_, + **__ ) -> None: self._check_for_api() + response = self.api.get_items(self.url, kind=RemoteObjectType.ALBUM, extend=False, use_cache=use_cache)[0] - response = self.api.get_items(self.url, kind=RemoteObjectType.ALBUM, use_cache=use_cache)[0] - if extend_artists: - self.api.get_items(response["artists"], kind=RemoteObjectType.ARTIST, use_cache=use_cache) - if extend_tracks: - tracks = response["tracks"] - self.api.extend_tracks(tracks["items"], limit=tracks["limit"], features=True, use_cache=use_cache) + skip_checks = self._extend_response( + response=response, + api=self.api, + use_cache=use_cache, + extend_tracks=extend_tracks, + extend_artists=extend_artists, + extend_features=extend_features, + ) - self.__init__(response=response, api=self.api, skip_checks=not extend_tracks) + self.__init__(response=response, api=self.api, skip_checks=skip_checks) class SpotifyArtist(RemoteArtist[SpotifyAlbum], SpotifyCollectionLoader[SpotifyAlbum]): @@ -670,44 +802,78 @@ def refresh(self, skip_checks: bool = False) -> None: for album in self.response.get("albums", {}).get("items", []) ] - # TODO: this currently overrides parent loader because parent loader is too 'tracks' specific - # see if this can be modified to take advantage of the item filtering logic in the parent loader @classmethod - def load( + def _get_item_kind(cls, api: SpotifyAPI) -> RemoteObjectType: + return RemoteObjectType.ALBUM + + @classmethod + def _get_items( + cls, items: Collection[str] | MutableMapping[str, Any], api: SpotifyAPI, use_cache: bool = True + ) -> list[dict[str, Any]]: + return api.get_items(items, extend=False, use_cache=use_cache) + + @classmethod + def _filter_items(cls, items: Iterable[SpotifyAlbum], response: dict[str, Any]) -> Iterable[SpotifyAlbum]: + """Filter down and return only ``items`` the items associated with the given ``response``""" + return [ + item for item in items + if any(artist["id"] == response["id"] for artist in item.response.get("artists", [])) + ] + + @classmethod + def _extend_response( cls, - value: str | Mapping[str, Any] | RemoteResponse, + response: MutableMapping[str, Any], api: SpotifyAPI, + use_cache: bool = True, + leave_bar: bool = True, extend_albums: bool = False, extend_tracks: bool = False, - use_cache: bool = True, + extend_features: bool = False, *_, **__ - ) -> Self: - self = cls.__new__(cls) - self.api = api + ) -> bool: + item_kind = RemoteObjectType.ALBUM + item_key = item_kind.name.lower() + "s" - # set a mock response with URL to load from - id_ = api.wrangler.extract_ids(value)[0] - self._response = { - "href": api.wrangler.convert( - id_, kind=RemoteObjectType.ARTIST, type_in=RemoteIDType.ID, type_out=RemoteIDType.URL - ) - } - self.reload(extend_albums=extend_albums, extend_tracks=extend_tracks, use_cache=use_cache) - return self + response_items = response.get(item_key, {}) + has_all_albums = item_key in response and len(response_items[api.items_key]) == response_items["total"] + if extend_albums and not has_all_albums: + api.get_artist_albums(response, limit=response.get(item_key, {}).get("limit", 50), use_cache=use_cache) + + album_item_kind = api.collection_item_map[item_kind] + album_item_key = album_item_kind.name.lower() + "s" + albums = response.get(item_key, {}).get(api.items_key) + + if albums and extend_tracks: + for album in albums: + api.extend_items(album[album_item_key], kind=item_kind, key=album_item_kind, use_cache=use_cache) + + if albums and extend_features: + tracks = [track for album in albums for track in album[album_item_key]["items"]] + api.extend_tracks(tracks, limit=response[item_key]["limit"], features=True, use_cache=use_cache) + + return not extend_albums or not extend_tracks def reload( - self, extend_albums: bool = False, extend_tracks: bool = False, use_cache: bool = True, *_, **__ + self, + use_cache: bool = True, + extend_albums: bool = False, + extend_tracks: bool = False, + extend_features: bool = False, + *_, + **__ ) -> None: self._check_for_api() - response = self.api.handler.get(url=self.url, use_cache=use_cache, log_pad=self._url_pad) - if extend_albums: - self.api.get_artist_albums(response, use_cache=use_cache) - if extend_albums and extend_tracks: - kind = RemoteObjectType.ALBUM - key = self.api.collection_item_map[kind] - for album in response["albums"]["items"]: - self.api.extend_items(album["tracks"], kind=kind, key=key, use_cache=use_cache) - - self.__init__(response=response, api=self.api, skip_checks=extend_albums and not extend_tracks) + + skip_checks = self._extend_response( + response=response, + api=self.api, + use_cache=use_cache, + extend_albums=extend_albums, + extend_tracks=extend_tracks, + extend_features=extend_features, + ) + + self.__init__(response=response, api=self.api, skip_checks=skip_checks) diff --git a/tests/shared/remote/utils.py b/tests/shared/remote/utils.py index bb89fd6d..b1e72981 100644 --- a/tests/shared/remote/utils.py +++ b/tests/shared/remote/utils.py @@ -43,7 +43,9 @@ def calculate_pages(limit: int, total: int) -> int: Calculates the numbers of a pages that need to be called from a given ``total`` and ``limit`` per page to get all items related to this response. """ - return total // limit + (total % limit > 0) # round up + if limit > 0 and total > 0: + return total // limit + (total % limit > 0) # round up + return 0 @abstractmethod def calculate_pages_from_response(self, response: Mapping[str, Any]) -> int: diff --git a/tests/spotify/api/mock.py b/tests/spotify/api/mock.py index caf1c866..d4c44fb0 100644 --- a/tests/spotify/api/mock.py +++ b/tests/spotify/api/mock.py @@ -66,11 +66,13 @@ def item_type_map_user(self): ObjectType.AUDIOBOOK: self.user_audiobooks, } - def calculate_pages_from_response(self, response: Mapping[str, Any]) -> int: - kind = ObjectType.from_name(response["type"])[0] - key = SpotifyAPI.collection_item_map[kind].name.lower() + "s" + def calculate_pages_from_response(self, response: Mapping[str, Any], item_key: str | None = None) -> int: + item_kind = ObjectType.from_name(response["type"])[0] + if not item_key: + item_key = SpotifyAPI.collection_item_map[item_kind].name.lower() + "s" - return self.calculate_pages(limit=response[key]["limit"], total=response[key]["total"]) + items_block = response.get(item_key, {}) + return self.calculate_pages(limit=items_block.get("limit", 0), total=items_block.get("total", 0)) def __init__(self, **kwargs): super().__init__(case_sensitive=True, **{k: v for k, v in kwargs.items() if k != "case_sensitive"}) diff --git a/tests/spotify/api/test_item.py b/tests/spotify/api/test_item.py index 1b208d58..e46f1d18 100644 --- a/tests/spotify/api/test_item.py +++ b/tests/spotify/api/test_item.py @@ -1,7 +1,7 @@ from collections.abc import Collection from copy import deepcopy from itertools import batched -from random import sample, randrange +from random import sample, randrange, choice from typing import Any from urllib.parse import parse_qs @@ -192,6 +192,22 @@ def test_get_items_batches_limited(self, api: SpotifyAPI, api_mock: SpotifyMock) ########################################################################### ## Input validation ########################################################################### + @pytest.mark.parametrize("object_type", [RemoteObjectType.ALBUM], ids=idfn) + def test_extend_items_input_validation( + self, + object_type: RemoteObjectType, + response: dict[str, Any], + key: str, + api: SpotifyAPI, + api_mock: SpotifyMock + ): + # function should skip on fully extended response + while len(response[key][api.items_key]) < response[key]["total"]: + response[key][api.items_key].append(choice(response[key][api.items_key])) + + api.extend_items(response, kind=object_type, key=api.collection_item_map[object_type]) + assert not api_mock.request_history + def test_get_user_items_input_validation(self, api: SpotifyAPI): # raises error when invalid item type given for kind in set(ALL_ITEM_TYPES) - api.user_item_types: diff --git a/tests/spotify/object/test_album.py b/tests/spotify/object/test_album.py index bb7937a4..53488dc9 100644 --- a/tests/spotify/object/test_album.py +++ b/tests/spotify/object/test_album.py @@ -1,3 +1,4 @@ +import re from collections.abc import Iterable from copy import deepcopy from datetime import date @@ -11,10 +12,7 @@ from musify.shared.remote.exception import RemoteObjectTypeError, RemoteError from musify.shared.types import Number from musify.spotify.api import SpotifyAPI -from musify.spotify.object import SpotifyAlbum, SpotifyArtist -from musify.spotify.object import SpotifyTrack -from musify.spotify.processors import SpotifyDataWrangler -from tests.shared.remote.object import RemoteCollectionTester +from musify.spotify.object import SpotifyAlbum, SpotifyTrack from tests.spotify.api.mock import SpotifyMock from tests.spotify.object.testers import SpotifyCollectionLoaderTester from tests.spotify.utils import assert_id_attributes @@ -32,6 +30,10 @@ def collection(self, response_random: dict[str, Any], api: SpotifyAPI) -> Spotif def collection_merge_items(self, api_mock: SpotifyMock) -> Iterable[SpotifyTrack]: return [SpotifyTrack(api_mock.generate_track()) for _ in range(randrange(5, 10))] + @pytest.fixture + def item_kind(self, api: SpotifyAPI) -> RemoteObjectType: + return api.collection_item_map[RemoteObjectType.ALBUM] + @pytest.fixture def response_random(self, api_mock: SpotifyMock) -> dict[str, Any]: """Yield a randomly generated response from the Spotify API for a track item type""" @@ -46,6 +48,7 @@ def _response_valid(self, api: SpotifyAPI, api_mock: SpotifyMock) -> dict[str, A response = next( deepcopy(album) for album in api_mock.albums if album["tracks"]["total"] > len(album["tracks"]["items"]) > 5 and album["genres"] + and album["artists"] ) api.extend_items(response=response, key=RemoteObjectType.TRACK) @@ -200,204 +203,130 @@ def test_reload(self, response_valid: dict[str, Any], api: SpotifyAPI): assert album.compilation != is_compilation album.api = api - album.reload(extend_artists=True, extend_tracks=False) + album.reload(extend_artists=True, extend_features=False) assert album.genres assert album.rating is not None assert album.compilation == is_compilation - def test_load_with_items(self, response_valid: dict[str, Any], api: SpotifyAPI, api_mock: SpotifyMock): - api_mock.reset_mock() # test checks the number of requests made - key = api.collection_item_map[RemoteObjectType.ALBUM].name.lower() + "s" + ########################################################################### + ## Load method tests + ########################################################################### + @staticmethod + def get_load_without_items( + loader: SpotifyAlbum, + response_valid: dict[str, Any], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + return loader.load(response_valid["href"], api=api, extend_tracks=True) - # ensure extension can be made by reducing available items and adding next page URL - response_valid[key]["items"] = response_valid[key]["items"][:response_valid[key]["limit"]] - response_valid[key]["next"] = SpotifyAPI.format_next_url( - url=response_valid[key]["href"], offset=response_valid[key]["limit"], limit=response_valid[key]["limit"] - ) + @pytest.fixture + def load_items( + self, response_valid: dict[str, Any], item_key: str, api: SpotifyAPI, api_mock: SpotifyMock + ) -> list[SpotifyTrack]: + """ + Extract some item responses from the given ``response_valid`` and remove them from the response. + This fixture manipulates the ``response_valid`` by removing these items + and reformatting the values in the items block to ensure 'extend_items' calls can still be run successfully. + + :return: The extracted response as SpotifyTracks. + """ + api_mock.reset_mock() # tests check the number of requests made + + # ensure extension of items can be made by reducing available items + limit = response_valid[item_key]["limit"] + response_valid[item_key]["items"] = response_valid[item_key][api.items_key][:limit] + response_items = response_valid[item_key]["items"] + assert len(response_items) < response_valid[item_key]["total"] # produce a list of items for input and ensure all items have this album assigned - available_id_list = {item["id"] for item in response_valid[key]["items"]} - limit = len(available_id_list) // 2 + available_ids = {item["id"] for item in response_items} + limit = len(available_ids) // 2 items = [] - response_without_items = {k: v for k, v in response_valid.items() if k != key} - for response in response_valid[key]["items"][:limit]: + response_without_items = {k: v for k, v in response_valid.items() if k != item_key} + for response in response_items[:limit]: response = deepcopy(response) response["album"] = response_without_items items.append(SpotifyTrack(response)) - # limit the list of items in the response so that some are in the input items list and some are not - items_ids_limited = [item["id"] for item in items][:len(available_id_list) // 3] - response_items = [item for item in response_valid[key]["items"] if item["id"] not in items_ids_limited] - response_valid[key]["items"] = response_items + # ensure all initially available items are covered by the response items and input items + assert {item["id"] for item in response_items} | {item.id for item in items} == available_ids - # ensure extension will happen and all initially available items are covered by the response and input items - assert len(response_valid[key]["items"]) < response_valid[key]["total"] - ids = {item["id"] for item in response_valid[key]["items"]} | {item.id for item in items} - assert ids == available_id_list - - self.assert_load_with_tracks( - cls=SpotifyAlbum, items=items, response=response_valid, api=api, api_mock=api_mock + # fix the items block to ensure extension doesn't over/under extend + response_valid[item_key] = api_mock.format_items_block( + url=response_valid[item_key]["href"], + items=response_valid[item_key][api.items_key], + limit=len(response_valid[item_key][api.items_key]), + total=response_valid[item_key]["total"], ) + return items -class TestSpotifyArtist(RemoteCollectionTester): + def test_load_with_all_items( + self, response_valid: dict[str, Any], item_key: str, api: SpotifyAPI, api_mock: SpotifyMock + ): + api_mock.reset_mock() # test checks the number of requests made - @pytest.fixture - def collection(self, response_random: dict[str, Any]) -> SpotifyArtist: - artist = SpotifyArtist(response_random) - artist._albums = [item for item in artist.items if artist.items.count(item) == 1] - return artist + load_items = [SpotifyTrack(response) for response in response_valid[item_key][api.items_key]] + SpotifyAlbum.load( + response_valid, api=api, items=load_items, extend_albums=True, extend_tracks=False, extend_features=False + ) - @pytest.fixture - def collection_merge_items(self, api_mock: SpotifyMock) -> Iterable[SpotifyAlbum]: - albums = [api_mock.generate_album() for _ in range(randrange(5, 10))] - for album in albums: - album["total_tracks"] = len(album["tracks"]["items"]) + assert not api_mock.request_history - return list(map(SpotifyAlbum, albums)) + def test_load_with_some_items( + self, + response_valid: dict[str, Any], + item_key: str, + load_items: list[SpotifyTrack], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + kind = RemoteObjectType.ALBUM - @pytest.fixture - def response_random(self, api_mock: SpotifyMock) -> dict[str, Any]: - """Yield a randomly generated response from the Spotify API for an artist item type""" - artist = api_mock.generate_artist() - albums = [ - api_mock.generate_album(tracks=False, artists=False, use_stored=False) - for _ in range(randrange(5, 10)) - ] - for album in albums: - album["artists"] = [deepcopy(artist)] - album["total_tracks"] = 0 - - items_block = api_mock.format_items_block( - url=f"{SpotifyDataWrangler.url_api}/artists/{artist["id"]}/albums", items=albums, total=len(albums) + result: SpotifyAlbum = SpotifyAlbum.load( + response_valid, api=api, items=load_items, extend_tracks=True, extend_features=True ) - return artist | {"albums": items_block} - @pytest.fixture - def response_valid(self, api_mock: SpotifyMock) -> dict[str, Any]: - """Yield a valid enriched response from the Spotify API for an artist item type.""" - artist_album_map = { - artist["id"]: [ - deepcopy(album) for album in api_mock.artist_albums - if any(art["id"] == artist["id"] for art in album["artists"]) - ] - for artist in api_mock.artists - } - id_, albums = next((id_, albums) for id_, albums in artist_album_map.items() if len(albums) >= 10) - artist = next(deepcopy(artist) for artist in api_mock.artists if artist["id"] == id_) - - for album in albums: - tracks = [deepcopy(track) for track in api_mock.tracks if track["album"]["id"] == album["id"]] - [track.pop("popularity", None) for track in tracks] - tracks = [track | {"track_number": i} for i, track in enumerate(tracks, 1)] - - album["tracks"] = api_mock.format_items_block(url=album["href"], items=tracks, total=len(tracks)) - - items_block = api_mock.format_items_block( - url=f"{SpotifyDataWrangler.url_api}/artists/{artist["id"]}/albums", items=albums, total=len(albums) + self.assert_load_with_items_requests( + response=response_valid, result=result, items=load_items, key=item_key, api_mock=api_mock + ) + self.assert_load_with_items_extended( + response=response_valid, result=result, items=load_items, kind=kind, key=item_key, api_mock=api_mock ) - return artist | {"albums": items_block} - - def test_input_validation(self, response_random: dict[str, Any], api_mock: SpotifyMock): - with pytest.raises(RemoteObjectTypeError): - SpotifyArtist(api_mock.generate_track(artists=False, album=False)) - with pytest.raises(APIError): - SpotifyArtist(response_random).reload() - - def test_attributes(self, response_random: dict[str, Any]): - artist = SpotifyArtist(response_random) - original_response = deepcopy(response_random) - - assert_id_attributes(item=artist, response=original_response) - assert len(artist.albums) == len(original_response["albums"]["items"]) - assert len(artist.artists) == len({art.name for album in artist.albums for art in album.artists}) - assert len(artist.tracks) == artist.track_total == sum(len(album) for album in artist.albums) - - assert artist.name == artist.artist - assert artist.artist == original_response["name"] - new_name = "new name" - artist.response["name"] = new_name - assert artist.artist == new_name - - assert artist.genres == original_response["genres"] - new_genres = ["electronic", "dance"] - artist.response["genres"] = new_genres - assert artist.genres == new_genres - - if not artist.has_image: - artist.response["images"] = [{"height": 200, "url": "old url"}] - images = {image["height"]: image["url"] for image in artist.response["images"]} - assert len(artist.image_links) == 1 - assert artist.image_links["cover_front"] == next(url for height, url in images.items() if height == max(images)) - new_image_link = "new url" - artist.response["images"].append({"height": max(images) * 2, "url": new_image_link}) - assert artist.image_links["cover_front"] == new_image_link - - assert artist.rating == original_response["popularity"] - new_rating = artist.rating + 20 - artist.response["popularity"] = new_rating - assert artist.rating == new_rating - - assert artist.followers == original_response["followers"]["total"] - new_followers = artist.followers + 20 - artist.response["followers"]["total"] = new_followers - assert artist.followers == new_followers - def test_refresh(self, response_valid: dict[str, Any]): - artist = SpotifyArtist(response_valid, skip_checks=True) - original_album_count = len(artist.albums) - artist.response["albums"]["items"] = artist.response["albums"]["items"][:original_album_count // 2] + # requests for extension data + expected = api_mock.calculate_pages_from_response(response_valid) + # -1 for not calling initial page + assert len(api_mock.get_requests(re.compile(f"{result.url}/{item_key}"))) == expected - 1 + assert len(api_mock.get_requests(re.compile(f"{api.url}/audio-features"))) == expected + assert not api_mock.get_requests(re.compile(f"{api.url}/artists")) # did not extend artists + + def test_load_with_some_items_and_no_extension( + self, + response_valid: dict[str, Any], + item_kind: RemoteObjectType, + item_key: str, + load_items: list[SpotifyTrack], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + api.extend_items(response_valid, kind=RemoteObjectType.ALBUM, key=item_kind) + api_mock.reset_mock() + + assert len(response_valid[item_key][api.items_key]) == response_valid[item_key]["total"] + assert not api_mock.get_requests(response_valid[item_key]["href"]) + + result: SpotifyAlbum = SpotifyAlbum.load( + response_valid, api=api, items=load_items, extend_artists=True, extend_tracks=True, extend_features=True + ) - artist.refresh(skip_checks=True) - assert len(artist.albums) == original_album_count // 2 + self.assert_load_with_items_requests( + response=response_valid, result=result, items=load_items, key=item_key, api_mock=api_mock + ) - def test_reload(self, response_valid: dict[str, Any], api: SpotifyAPI): - genres = response_valid.pop("genres", None) - response_valid.pop("popularity", None) - response_valid.pop("followers", None) - - albums = response_valid.pop("albums")["items"] - album_ids = {album["id"] for album in albums} - artist_names = {artist["name"] for album in albums for artist in album["artists"]} - - artist = SpotifyArtist(response_valid) - assert not artist.genres - assert artist.rating is None - assert artist.followers is None - assert not artist.albums - assert not artist.artists - assert not artist.tracks - - artist.api = api - artist.reload(extend_albums=False, extend_tracks=True) - if genres: - assert artist.genres - assert artist.rating is not None - assert artist.followers is not None - assert not artist.albums - assert not artist.artists - assert not artist.tracks - - artist.reload(extend_albums=True, extend_tracks=False) - assert {album.id for album in artist._albums} == album_ids - assert len(artist.artists) == len(artist_names) - assert set(artist.artists) == artist_names - assert not artist.tracks - - artist.reload(extend_albums=True, extend_tracks=True) - assert artist.tracks - - def test_load(self, response_valid: dict[str, Any], api: SpotifyAPI): - artist = SpotifyArtist.load(response_valid["href"], api=api) - - assert artist.name == response_valid["name"] - assert artist.id == response_valid["id"] - assert artist.url == response_valid["href"] - assert not artist.albums - assert not artist.artists - assert not artist.tracks - - artist = SpotifyArtist.load(response_valid["href"], api=api, extend_albums=True, extend_tracks=True) - assert artist.albums - assert artist.artists - assert artist.tracks + # requests for extension data + expected = api_mock.calculate_pages_from_response(response_valid) + assert not api_mock.get_requests(re.compile(f"{result.url}/{item_key}")) # already extended on input + assert len(api_mock.get_requests(re.compile(f"{api.url}/audio-features"))) == expected + assert api_mock.get_requests(re.compile(f"{api.url}/artists")) # called the artists endpoint at least once diff --git a/tests/spotify/object/test_artist.py b/tests/spotify/object/test_artist.py index d03b3060..7c77786f 100644 --- a/tests/spotify/object/test_artist.py +++ b/tests/spotify/object/test_artist.py @@ -1,3 +1,4 @@ +import re from collections.abc import Iterable from copy import deepcopy from random import randrange @@ -6,16 +7,17 @@ import pytest from musify.shared.api.exception import APIError +from musify.shared.remote.enum import RemoteObjectType from musify.shared.remote.exception import RemoteObjectTypeError from musify.spotify.api import SpotifyAPI from musify.spotify.object import SpotifyAlbum, SpotifyArtist from musify.spotify.processors import SpotifyDataWrangler -from tests.shared.remote.object import RemoteCollectionTester +from tests.spotify.object.testers import SpotifyCollectionLoaderTester from tests.spotify.api.mock import SpotifyMock from tests.spotify.utils import assert_id_attributes -class TestSpotifyArtist(RemoteCollectionTester): +class TestSpotifyArtist(SpotifyCollectionLoaderTester): @pytest.fixture def collection(self, response_random: dict[str, Any]) -> SpotifyArtist: @@ -31,6 +33,10 @@ def collection_merge_items(self, api_mock: SpotifyMock) -> Iterable[SpotifyAlbum return list(map(SpotifyAlbum, albums)) + @pytest.fixture + def item_kind(self, *_) -> RemoteObjectType: + return RemoteObjectType.ALBUM + @pytest.fixture def response_random(self, api_mock: SpotifyMock) -> dict[str, Any]: """Yield a randomly generated response from the Spotify API for an artist item type""" @@ -162,17 +168,123 @@ def test_reload(self, response_valid: dict[str, Any], api: SpotifyAPI): artist.reload(extend_albums=True, extend_tracks=True) assert artist.tracks - def test_load(self, response_valid: dict[str, Any], api: SpotifyAPI): - artist = SpotifyArtist.load(response_valid["href"], api=api) + ########################################################################### + ## Load method tests + ########################################################################### + @staticmethod + def get_load_without_items( + loader: SpotifyArtist, + response_valid: dict[str, Any], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + return loader.load(response_valid["href"], api=api, extend_albums=True, extend_tracks=True) - assert artist.name == response_valid["name"] - assert artist.id == response_valid["id"] - assert artist.url == response_valid["href"] - assert not artist.albums - assert not artist.artists - assert not artist.tracks + @pytest.fixture + def load_items( + self, response_valid: dict[str, Any], item_key: str, api: SpotifyAPI, api_mock: SpotifyMock + ) -> list[SpotifyAlbum]: + """ + Extract some item responses from the given ``response_valid`` and remove them from the response. + This fixture manipulates the ``response_valid`` by removing these items + and reformatting the values in the items block to ensure 'extend_items' calls can still be run successfully. + + :return: The extracted response as SpotifyTracks. + """ + api_mock.reset_mock() # tests check the number of requests made + + # ensure extension of items can be made by reducing available items + limit = response_valid[item_key]["limit"] + response_valid[item_key]["items"] = response_valid[item_key][api.items_key][:limit] + response_items = response_valid[item_key]["items"] + assert len(response_items) < response_valid[item_key]["total"] + + # produce a list of items for input + available_ids = {item["id"] for item in response_items} + limit = len(available_ids) // 2 + items = [SpotifyAlbum(response, skip_checks=True) for response in deepcopy(response_items[:limit])] + + # ensure all initially available items are covered by the response items and input items + assert {item["id"] for item in response_items} | {item.id for item in items} == available_ids + + # fix the items block to ensure extension doesn't over/under extend + response_valid[item_key] = api_mock.format_items_block( + url=response_valid[item_key]["href"], + items=response_valid[item_key][api.items_key], + limit=len(response_valid[item_key][api.items_key]), + total=response_valid[item_key]["total"], + ) - artist = SpotifyArtist.load(response_valid["href"], api=api, extend_albums=True, extend_tracks=True) - assert artist.albums - assert artist.artists - assert artist.tracks + return items + + def test_load_with_all_items( + self, response_valid: dict[str, Any], item_key: str, api: SpotifyAPI, api_mock: SpotifyMock + ): + api_mock.reset_mock() # test checks the number of requests made + + load_items = [SpotifyAlbum(response, skip_checks=True) for response in response_valid[item_key][api.items_key]] + SpotifyArtist.load( + response_valid, api=api, items=load_items, extend_albums=True, extend_tracks=False, extend_features=False + ) + + assert not api_mock.request_history + + def test_load_with_some_items( + self, + response_valid: dict[str, Any], + item_key: str, + load_items: list[SpotifyAlbum], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + kind = RemoteObjectType.ARTIST + + result: SpotifyArtist = SpotifyArtist.load( + response_valid, api=api, items=load_items, extend_albums=True, extend_tracks=True, extend_features=True + ) + + self.assert_load_with_items_requests( + response=response_valid, result=result, items=load_items, key=item_key, api_mock=api_mock + ) + self.assert_load_with_items_extended( + response=response_valid, result=result, items=load_items, kind=kind, key=item_key, api_mock=api_mock + ) + + # requests for extension data + expected = api_mock.calculate_pages_from_response(response_valid, item_key=item_key) + assert len(api_mock.get_requests(re.compile(f"{result.url}/{item_key}"))) == expected + + for album in result.response[item_key][api.items_key]: + url = album["tracks"]["href"].split("?")[0] + expected = api_mock.calculate_pages_from_response(album) + assert len(api_mock.get_requests(re.compile(url))) == expected + + assert result.tracks + expected_features = api_mock.calculate_pages(limit=response_valid[item_key]["limit"], total=len(result.tracks)) + assert len(api_mock.get_requests(re.compile(f"{api.url}/audio-features"))) == expected_features + + def test_load_with_some_items_and_no_extension( + self, + response_valid: dict[str, Any], + item_kind: RemoteObjectType, + item_key: str, + load_items: list[SpotifyAlbum], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + api.extend_items(response_valid, kind=RemoteObjectType.ARTIST, key=item_kind) + api_mock.reset_mock() + + assert len(response_valid[item_key][api.items_key]) == response_valid[item_key]["total"] + assert not api_mock.get_requests(response_valid[item_key]["href"]) + + result: SpotifyArtist = SpotifyArtist.load(response_valid, api=api, items=load_items, extend_albums=True) + + self.assert_load_with_items_requests( + response=response_valid, result=result, items=load_items, key=item_key, api_mock=api_mock + ) + + # requests for extension data + assert not api_mock.get_requests(re.compile(f"{result.url}/{item_key}")) + assert not api_mock.get_requests(re.compile(f"{api.url}/audio-features")) + assert not api_mock.get_requests(re.compile(f"{api.url}/artists")) diff --git a/tests/spotify/object/test_playlist.py b/tests/spotify/object/test_playlist.py index 80a186a0..75d80e1b 100644 --- a/tests/spotify/object/test_playlist.py +++ b/tests/spotify/object/test_playlist.py @@ -1,3 +1,4 @@ +import re from collections.abc import Iterable from copy import deepcopy from datetime import datetime @@ -12,8 +13,7 @@ from musify.shared.remote.exception import RemoteObjectTypeError, RemoteError from musify.spotify.api import SpotifyAPI from musify.spotify.exception import SpotifyCollectionError -from musify.spotify.object import SpotifyPlaylist -from musify.spotify.object import SpotifyTrack +from musify.spotify.object import SpotifyPlaylist, SpotifyTrack from tests.shared.remote.object import RemotePlaylistTester from tests.spotify.api.mock import SpotifyMock from tests.spotify.object.testers import SpotifyCollectionLoaderTester @@ -26,6 +26,10 @@ class TestSpotifyPlaylist(SpotifyCollectionLoaderTester, RemotePlaylistTester): def collection_merge_items(self, api_mock: SpotifyMock) -> Iterable[SpotifyTrack]: return [SpotifyTrack(api_mock.generate_track()) for _ in range(randrange(5, 10))] + @pytest.fixture + def item_kind(self, api: SpotifyAPI) -> RemoteObjectType: + return api.collection_item_map[RemoteObjectType.PLAYLIST] + @pytest.fixture def playlist(self, response_valid: dict[str, Any], api: SpotifyAPI) -> SpotifyPlaylist: pl = SpotifyPlaylist(response=response_valid, api=api) @@ -189,32 +193,125 @@ def test_reload(self, response_valid: dict[str, Any], api: SpotifyAPI): assert pl.public is not response_valid["public"] assert pl.collaborative is not response_valid["collaborative"] - def test_load_with_items(self, response_valid: dict[str, Any], api: SpotifyAPI, api_mock: SpotifyMock): + ########################################################################### + ## Load method tests + ########################################################################### + @staticmethod + def get_load_without_items( + loader: SpotifyPlaylist, + response_valid: dict[str, Any], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + return loader.load(response_valid["href"], api=api, extend_tracks=True) + + @pytest.fixture + def load_items( + self, response_valid: dict[str, Any], item_key: str, api: SpotifyAPI, api_mock: SpotifyMock + ) -> list[SpotifyTrack]: + """ + Extract some item responses from the given ``response_valid`` and remove them from the response. + This fixture manipulates the ``response_valid`` by removing these items + and reformatting the values in the items block to ensure 'extend_items' calls can still be run successfully. + + :return: The extracted response as SpotifyTracks. + """ + api_mock.reset_mock() # tests check the number of requests made + key_sub = item_key.rstrip("s") + + # ensure extension of items can be made by reducing available items + limit = response_valid[item_key]["limit"] + response_valid[item_key]["items"] = response_valid[item_key][api.items_key][:limit] + response_items = response_valid[item_key]["items"] + assert len(response_items) < response_valid[item_key]["total"] + + # produce a list of items for input + available_ids = {item[key_sub]["id"] for item in response_items} + limit = len(available_ids) // 2 + items = [SpotifyTrack(response[key_sub]) for response in deepcopy(response_items[:limit])] + for item in response_items: + item[key_sub].pop("popularity") + + # ensure all initially available items are covered by the response items and input items + assert {item[key_sub]["id"] for item in response_items} | {item.id for item in items} == available_ids + + # fix the items block to ensure extension doesn't over/under extend + response_valid[item_key] = api_mock.format_items_block( + url=response_valid[item_key]["href"], + items=response_valid[item_key][api.items_key], + limit=len(response_valid[item_key][api.items_key]), + total=response_valid[item_key]["total"], + ) + + return items + + def test_load_with_all_items( + self, response_valid: dict[str, Any], item_key: str, api: SpotifyAPI, api_mock: SpotifyMock + ): api_mock.reset_mock() # test checks the number of requests made - key = api.collection_item_map[RemoteObjectType.PLAYLIST].name.lower() + "s" - # ensure extension can be made by reducing available items and adding next page URL - response_valid[key]["items"] = response_valid[key]["items"][:response_valid[key]["limit"]] - response_valid[key]["next"] = SpotifyAPI.format_next_url( - url=response_valid[key]["href"], offset=response_valid[key]["limit"], limit=response_valid[key]["limit"] + load_items = [SpotifyTrack(response) for response in response_valid[item_key][api.items_key]] + SpotifyPlaylist.load( + response_valid, api=api, items=load_items, extend_albums=True, extend_tracks=False, extend_features=False ) - # produce a list of items for input and ensure all items have this album assigned - available_ids = {item["track"]["id"] for item in response_valid[key]["items"]} - limit = len(available_ids) // 2 - items = [SpotifyTrack(response["track"]) for response in deepcopy(response_valid[key]["items"][:limit])] - for item in response_valid[key]["items"]: - item["track"].pop("popularity") + assert not api_mock.request_history + + def test_load_with_some_items( + self, + response_valid: dict[str, Any], + item_key: str, + load_items: list[SpotifyTrack], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + kind = RemoteObjectType.PLAYLIST + + result: SpotifyPlaylist = SpotifyPlaylist.load( + response_valid, api=api, items=load_items, extend_tracks=True, extend_features=True + ) - # ensure extension will happen and all initially available items are covered by the response and input items - assert len(response_valid[key]["items"]) < response_valid[key]["total"] - ids = {item["track"]["id"] for item in response_valid[key]["items"]} | {item.id for item in items} - assert ids == available_ids + self.assert_load_with_items_requests( + response=response_valid, result=result, items=load_items, key=item_key, api_mock=api_mock + ) + self.assert_load_with_items_extended( + response=response_valid, result=result, items=load_items, kind=kind, key=item_key, api_mock=api_mock + ) - self.assert_load_with_tracks( - cls=SpotifyPlaylist, items=items, response=response_valid, api=api, api_mock=api_mock + # requests for extension data + expected = api_mock.calculate_pages_from_response(response_valid) + # -1 for not calling initial page + assert len(api_mock.get_requests(re.compile(f"{result.url}/{item_key}"))) == expected - 1 + assert len(api_mock.get_requests(re.compile(f"{api.url}/audio-features"))) == expected + + def test_load_with_some_items_and_no_extension( + self, + response_valid: dict[str, Any], + item_kind: RemoteObjectType, + item_key: str, + load_items: list[SpotifyTrack], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + api.extend_items(response_valid, kind=RemoteObjectType.PLAYLIST, key=item_kind) + api_mock.reset_mock() + + assert len(response_valid[item_key][api.items_key]) == response_valid[item_key]["total"] + assert not api_mock.get_requests(response_valid[item_key]["href"]) + + result: SpotifyPlaylist = SpotifyPlaylist.load( + response_valid, api=api, items=load_items, extend_tracks=True, extend_features=False ) + self.assert_load_with_items_requests( + response=response_valid, result=result, items=load_items, key=item_key, api_mock=api_mock + ) + assert not api_mock.get_requests(response_valid[item_key]["href"]) + + # requests for extension data + assert not api_mock.get_requests(re.compile(f"{result.url}/{item_key}")) # already extended on input + assert not api_mock.get_requests(re.compile(f"{api.url}/audio-features")) + def test_create_playlist(self, api: SpotifyAPI, api_mock: SpotifyMock): api_mock.reset_mock() # test checks the number of requests made @@ -250,7 +347,6 @@ def test_delete_playlist(self, response_valid: dict[str, Any], api: SpotifyAPI, ########################################################################### ## Sync tests set up ########################################################################### - @pytest.fixture def sync_playlist(self, response_valid: dict[str, Any], api: SpotifyAPI) -> SpotifyPlaylist: return SpotifyPlaylist(response=response_valid, api=api) diff --git a/tests/spotify/object/testers.py b/tests/spotify/object/testers.py index f1839c3b..0571684d 100644 --- a/tests/spotify/object/testers.py +++ b/tests/spotify/object/testers.py @@ -3,10 +3,12 @@ from typing import Any from urllib.parse import parse_qs +import pytest + from musify.shared.remote.enum import RemoteObjectType from musify.spotify.api import SpotifyAPI -from musify.spotify.base import SpotifyItem -from musify.spotify.object import SpotifyCollectionLoader, SpotifyTrack +from musify.spotify.base import SpotifyItem, SpotifyObject +from musify.spotify.object import SpotifyCollectionLoader, SpotifyArtist from tests.shared.remote.object import RemoteCollectionTester from tests.spotify.api.mock import SpotifyMock @@ -17,77 +19,104 @@ class SpotifyCollectionLoaderTester(RemoteCollectionTester, metaclass=ABCMeta): def collection_merge_items(self, *args, **kwargs) -> Iterable[SpotifyItem]: raise NotImplementedError + @abstractmethod + def item_kind(self, api: SpotifyAPI) -> RemoteObjectType: + """Yields the RemoteObjectType of items in this collection as a pytest.fixture""" + raise NotImplementedError + + @pytest.fixture + def item_key(self, item_kind: RemoteObjectType) -> str: + """Yields the key of items in this collection as a pytest.fixture""" + return item_kind.name.lower() + "s" + + ########################################################################### + ## Assertions + ########################################################################### + @staticmethod + def assert_load_with_items_requests[T: SpotifyObject]( + response: dict[str, Any], + result: SpotifyCollectionLoader[T], + items: list[T], + key: str, + api_mock: SpotifyMock, + ): + """Run assertions on the requests from load method with given ``items``""" + assert len(result.response[key][result.api.items_key]) == response[key]["total"] + assert len(result.items) == response[key]["total"] + assert not api_mock.get_requests(result.url) # main collection URL was not called + + # ensure none of the input_ids were requested + input_ids = {item.id for item in items} + for request in api_mock.get_requests(f"{result.url}/{key}"): + params = parse_qs(request.query) + if "ids" not in params: + continue + + assert not input_ids.intersection(params["ids"][0].split(",")) + + @staticmethod + def assert_load_with_items_extended[T: SpotifyObject]( + response: dict[str, Any], + result: SpotifyCollectionLoader[T], + items: list[T], + kind: RemoteObjectType, + key: str, + api_mock: SpotifyMock, + ): + """Run assertions on the requests for missing data from load method with given ``items``""" + requests_missing = api_mock.get_requests(f"{result.api.url}/{key}") + limit = response[key]["limit"] + input_ids = {item.id for item in items} + response_item_ids = { + item[key.rstrip("s")]["id"] if kind == RemoteObjectType.PLAYLIST else item["id"] + for item in response[key][result.api.items_key] + } + assert len(requests_missing) == api_mock.calculate_pages(limit=limit, total=len(response_item_ids - input_ids)) + + ########################################################################### + ## Tests + ########################################################################### @staticmethod + @abstractmethod + def get_load_without_items( + loader: SpotifyCollectionLoader, + response_valid: dict[str, Any], + api: SpotifyAPI, + api_mock: SpotifyMock + ): + """Yields the results from 'load' where no items are given as a pytest.fixture""" + raise NotImplementedError + def test_load_without_items( + self, collection: SpotifyCollectionLoader, response_valid: dict[str, Any], + item_key: str, api: SpotifyAPI, api_mock: SpotifyMock ): api_mock.reset_mock() # test checks the number of requests made - unit = collection.__class__.__name__.removeprefix("Spotify") - kind = RemoteObjectType.from_name(unit)[0] - key = api.collection_item_map[kind].name.lower() + "s" - - test = collection.__class__.load(response_valid["href"], api=api, extend_tracks=True) + result = self.get_load_without_items( + loader=collection, response_valid=response_valid, api=api, api_mock=api_mock + ) - assert test.name == response_valid["name"] - assert test.id == response_valid["id"] - assert test.url == response_valid["href"] + assert result.name == response_valid["name"] + assert result.id == response_valid["id"] + assert result.url == response_valid["href"] - requests = api_mock.get_requests(test.url) - requests += api_mock.get_requests(f"{test.url}/{key}") - requests += api_mock.get_requests(f"{collection.api.url}/audio-features") + expected = api_mock.calculate_pages_from_response(result.response, item_key=item_key) + if not isinstance(result, SpotifyArtist): + expected -= 1 # -1 for not calling initial page - # 1 call for initial collection + (pages - 1) for tracks + (pages) for audio-features - assert len(requests) == 2 * api_mock.calculate_pages_from_response(test.response) + assert len(api_mock.get_requests(result.url)) == 1 + assert len(api_mock.get_requests(f"{result.url}/{item_key}")) == expected + assert not api_mock.get_requests(f"{api.url}/audio-features") + assert not api_mock.get_requests(f"{api.url}/audio-analysis") # input items given, but no key to search on still loads - test = collection.__class__.load(response_valid, api=api, items=response_valid.pop(key), extend_tracks=True) + result = collection.load(response_valid, api=api, items=response_valid.pop(item_key), extend_tracks=True) - assert test.name == response_valid["name"] - assert test.id == response_valid["id"] - assert test.url == response_valid["href"] - - @staticmethod - def assert_load_with_tracks( - cls: type[SpotifyCollectionLoader], - items: list[SpotifyTrack], - response: dict[str, Any], - api: SpotifyAPI, - api_mock: SpotifyMock, - ): - """Run test with assertions on load method with given ``items``""" - unit = cls.__name__.removeprefix("Spotify") - kind = RemoteObjectType.from_name(unit)[0] - key = api.collection_item_map[kind].name.lower() + "s" - - test = cls.load(response, api=api, items=items, extend_tracks=True) - assert len(test.response[key]["items"]) == response[key]["total"] - assert len(test.items) == response[key]["total"] - assert not api_mock.get_requests(test.url) # playlist URL was not called - - # requests to extend album start from page 2 onward - requests = api_mock.get_requests(test.url) - requests += api_mock.get_requests(f"{test.url}/{key}") - requests += api_mock.get_requests(f"{api.url}/audio-features") - - # 0 calls for initial collection + (extend_pages - 1) for tracks + (extend_pages) for audio-features - # + (get_pages) for audio-features get on response items not in input items - if kind == RemoteObjectType.PLAYLIST: - input_ids = {item["track"]["id"] for item in response["tracks"]["items"]} - {item.id for item in items} - else: - input_ids = {item["id"] for item in response["tracks"]["items"]} - {item.id for item in items} - get_pages = api_mock.calculate_pages(limit=test.response[key]["limit"], total=len(input_ids)) - extend_pages = api_mock.calculate_pages_from_response(test.response) - assert len(requests) == 2 * extend_pages - 1 + get_pages - - # ensure none of the items ids were requested - input_ids = {item.id for item in items} - for request in api_mock.get_requests(f"{test.url}/{key}"): - params = parse_qs(request.query) - if "ids" not in params: - continue - - assert not input_ids.intersection(params["ids"][0].split(",")) + assert result.name == response_valid["name"] + assert result.id == response_valid["id"] + assert result.url == response_valid["href"] diff --git a/tests/test_report.py b/tests/test_report.py index 3d914758..a6f7877a 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -110,7 +110,7 @@ def test_report_missing_tags(local_library: LocalLibrary): if choice([True, False]): track.bpm = None if choice([True, False]): - track.key = None + track.item_key = None if choice([True, False]): track.disc_total = None