import operator from itertools import groupby from random import randint from server.constants import DSP from tests.helpers import FieldFabricator class VisitedTracksResponseFabricator(FieldFabricator): """Special fabricator class to create responses from/for api(s) we use in /visited-tracks/""" def _tracks_user_visits(self, number_of_items, vendor: str): """Generates response for """ response = [ { "vendor": vendor, "song_id": self._get_song_id(i), "last_visit_date": self._get_last_visit_datetime(i), "visit_count": randint(1, 20) } for i in range(number_of_items) ] response.sort(key=operator.itemgetter("visit_count", "last_visit_date"), reverse=True) return response def _tracks_track_info(self, tracks_user_visits_data): """Generates response for""" data_length = len(tracks_user_visits_data) result = {} for item, value in zip(tracks_user_visits_data, range(data_length)): result[item["song_id"]] = { "id": item["song_id"], "name": self._get_song_name(value), "artist_name": self._get_artists_name(value), "sony_track": None } return result def _tracks_artists_info(self, tracks_track_info_data): """Generates response for""" data_length = len(tracks_track_info_data.items()) result = [ {"track_id": item["id"], "id": self._get_artist_id(value), "name": item["name"]} for item, value in zip(tracks_track_info_data.values(), range(data_length)) ] return result def _dsp_tracks_for_visited_tracks(self, tracks_user_visits_data, image_height): """Generates response for""" data = [{ "id": track["song_id"], "album": { "images": [ {"url": self._get_image_url(value), "height": image_height} ] } } for track, value in zip(tracks_user_visits_data, range(len(tracks_user_visits_data)))] result = {"tracks": data} return result @staticmethod def get_expected_spotify_response(data, tracks, dsp_tracks, artists, is_sony): """Get an expected response""" dsp_tracks = {track["id"]: {"image": track["album"]["images"][0]["url"]} for track in dsp_tracks["tracks"]} artists = {track_id: list(artists) for track_id, artists in groupby(artists, lambda i: i["track_id"])} for visit in data: if visit["vendor"] == DSP.SPOTIFY.value: song_id = visit["song_id"] visit.update( { "is_sony": song_id in is_sony, "artists": [{"id": a["id"], "name": a["name"]} for a in artists[song_id]] if song_id in artists else [], } ) if song_id in tracks: song = tracks[song_id] visit.update({"name": song["name"], "artist_name": song["artist_name"]}) # There are some tracks that does not exists in our database. Let's mix data from DSP-API. if song_id in dsp_tracks: song = dsp_tracks[song_id] visit["image_url"] = song["image"] if "name" not in visit: visit.update({"name": song["name"], "artists": song.get("artists", [])}) return data def get_dependant_responses(self, items_number: int = 6, vendor: str = "spotify", image_height: int = 640): """Create all needed data responses for all call(s) """ data = self._tracks_user_visits(items_number, vendor) track = self._tracks_track_info(data) artists_info = self._tracks_artists_info(track) dsp_tracks = self._dsp_tracks_for_visited_tracks(data, image_height) is_sony = [] return data, track, artists_info, dsp_tracks, is_sony