import copy from datetime import date, datetime, timedelta, timezone from http import HTTPStatus from typing import List, Optional, Union from unittest.mock import AsyncMock import pytest from apollo_utils.core.constants.dsp import DSP from apollo_utils.core.constants.market import Market from pytest_mock import MockerFixture from server.client.clients.apollo_api import ApolloApiClient from server.client.clients.dsp_api import DspApiClient from server.constants.playlists.markets import APPLE_PLAYLISTS_TOP_28_MARKETS from tests.api.playlists.test_tracks import COUNTRY_CODE_LIST, get_dsp, get_isrc, get_playlist, get_playlist_id, \ get_playlist_index, get_tracklist from tests.helpers import read_json_from_file pytest_plugins = ["tests.clients.apollo_api.responses"] def get_date( days_diff: int, base_date: Optional[Union[date, datetime]] = None, as_str: bool = False, with_time: bool = False, trim_seconds: bool = False, ) -> Union[date, datetime, str]: if not base_date: base_date = datetime.now(timezone.utc).replace(microsecond=0) if with_time else date.today() result = base_date - timedelta(days=days_diff) if as_str: result = result.isoformat() if trim_seconds: result = result[0:-6] return result def get_dates(*args, as_str: bool = False) -> List[date] or List[str]: return [get_date(i, as_str=as_str) for i in args] def get_playlists(count: int = 5, with_name: bool = True) -> List[dict]: return [ { "playlistId": f"pl{i}", "trackLastAdded": get_date(i, with_time=True, as_str=True), "fridayLastUpdatedDate": get_date(i if i > 3 else 1, as_str=True), **({"name": f"PlN{i}"} if with_name else {}), } for i in range(1, count + 1) ] def get_delphi_playlists(playlist_id_list: List[str] = None, count: int = 5) -> dict: if not playlist_id_list: playlist_id_list = [f"spotify_pl{i}" for i in range(1, count + 1)] return {"items": [{"playlist_id": pl_id, "name": f"PlN{i}"} for i, pl_id in enumerate(playlist_id_list, 1)]} @pytest.mark.parametrize( "available_dates,playlist_count,outdated_indexes", ( (get_dates(1, 7, 10, 15, as_str=True), 5, [4, 5]), (get_dates(15, 10, 2, as_str=True), 1, []), ), ) async def test_get_playlists_summary_nmf( available_dates: List[str], playlist_count: int, outdated_indexes: List[int], mocker: MockerFixture, auth: dict, client, ): last_updated = get_date(days_diff=2, as_str=True) playlist_data = get_playlists(playlist_count) playlist_data_no_name = [{k: v for k, v in i.items() if k != "name"} for i in playlist_data] async def apollo_send(url: str, *args, **kwargs): if url == "api/spotify/nmf/available-dates/": return available_dates elif url == "api/value/": return last_updated return playlist_data_no_name apollo_send_mock = mocker.patch.object(ApolloApiClient, "_send_request", side_effect=apollo_send) async_dsp_mock = AsyncMock(return_value=get_delphi_playlists(count=playlist_count)) dsp_send_mock = mocker.patch.object(DspApiClient, "_send_request", side_effect=async_dsp_mock) response = await client.get("/api/v1/playlists/summary/nmf/", headers=auth) assert response.status == HTTPStatus.OK assert apollo_send_mock.call_count == 3 dsp_send_mock.assert_called_once() response = await response.json() assert response == { "available_dates": available_dates, "playlists": playlist_data, "outdated_playlists": [pl for i, pl in enumerate(playlist_data) if (i + 1) in outdated_indexes], "last_updated": last_updated, } async def test_get_tracks_playlists_nmf(mocker, client, auth): apollo_response = read_json_from_file("clients/apollo_api/spotify_nmf_track") for item in apollo_response["items"]: del item["playlist_name"] playlist_data = get_delphi_playlists([i["playlist_id"] for i in apollo_response["items"]]) id_name_map = {i["playlist_id"]: i["name"] for i in playlist_data["items"]} apollo_response_copy = { "items": [ {"playlist_name": id_name_map[item["playlist_id"]], **item} for i, item in enumerate(apollo_response["items"]) ], "count": apollo_response["count"], "regions": [copy.deepcopy(i) for i in apollo_response["regions"]], "total": apollo_response["total"], "positions_avg": apollo_response["positions_avg"], "positions_top_total": apollo_response["positions_top_total"], "top_playlist_image_url": apollo_response["top_playlist_image_url"], "previous": apollo_response["previous"], "next": apollo_response["next"].replace("playlists/nmf/track", "v1/tracks/playlists/nmf"), } async_apollo_mock = AsyncMock(return_value=apollo_response) get_apollo_mock = mocker.patch.object(ApolloApiClient, "_send_request", side_effect=async_apollo_mock) async_dsp_mock = AsyncMock(return_value=playlist_data) dsp_send_mock = mocker.patch.object(DspApiClient, "_send_request", side_effect=async_dsp_mock) params = {"isrc": "USSM10600677", "include": ["regions", "top_playlist_image_url"], "limit": 5} response = await client.get("/api/v1/tracks/playlists/nmf/", params=params, headers=auth) assert response.status == HTTPStatus.OK get_apollo_mock.assert_called_once() dsp_send_mock.assert_called_once() response = await response.json() assert response == apollo_response_copy def get_page_url( base_url: str, dsp: DSP, isrc_list: List[str], country_code: Optional[str], offset: Optional[int], limit: Optional[int], count: int, is_next: bool = True, ) -> Optional[str]: offset = offset or 0 if not limit or (is_next and (offset + limit) > count) or (not is_next and (offset - limit) < 0): return None offset = (offset or 0) + (limit if is_next else -min(offset or 0, limit)) if dsp.value == DSP.APPLE.value and country_code: country_code = ",".join(country_code) country_code_arg = ( f"&{'market' if dsp.value == DSP.SPOTIFY.value else 'country'}={country_code}" if country_code else "" ) offset_arg = f"&offset={offset}" if offset else "" args_str = f"isrc={'%2C'.join(isrc_list)}{country_code_arg}&limit={limit}{offset_arg}" return f"{base_url}?{args_str}" def get_by_track_result( streams_country_code_list: List[str], dsp: DSP, count: int = 10, offset: Optional[int] = None, limit: Optional[int] = None, isrc_list: Optional[List[str]] = None, country_code: Optional[str] = None, ) -> dict: item_list = list(range(1, count + 1)) if offset or limit: item_list = item_list[offset or 0: (limit or count) + (offset or 0)] result = [] for index in item_list: dsp = get_dsp(dsp, index) playlist_index = get_playlist_index(dsp, index, False) playlist_id = get_playlist_id(dsp, playlist_index) item_is_personalized = bool(playlist_index % 2) has_previous = not item_is_personalized and index % 4 and (index + 1) % 3 pl_country_code = COUNTRY_CODE_LIST[(playlist_index - 1) % len(COUNTRY_CODE_LIST)] if dsp.value == DSP.SPOTIFY.value: result_item = { "spotifyLink": f"spotify:playlist:{playlist_id}", "name": f"Pl {playlist_index}", "trackCount": 10 + playlist_index, "is_personalized": item_is_personalized, "countryCode": pl_country_code, "subscribers": 80000 + playlist_index, "position": 11 + index, "positionChange": (index + index * (-1 if index % 3 else 1)) if has_previous else None, "positionChangeDate": f"2022-11-1{index - 1}T16:37:41.205Z" if has_previous else None, "added": f"2022-11-0{index}T14:35:26.105Z", "earliestAdded": f"2022-04-0{index}T11:08:52.123Z", "owner": { "accountId": str(playlist_index), "accountName": f"Owner {playlist_index}", "categoryId": playlist_index, "categoryName": f"Cat {playlist_index}", }, "streams": { "global1Day": 51 + index * 5, "global7Days": 251 + index * 10, "local1Day": 52 + index * 5 if streams_country_code_list else None, "local7Days": 252 + index * 10 if streams_country_code_list else None, }, "imageFileName": f"https://images-api.atlas.stream/v2/playlists/by_spotify_id/{playlist_id}", "globalStreams7Days": 5001 + index * 100, "localStreams7Days": 5002 + index * 100 if streams_country_code_list else None, } else: result_item = { "id": playlist_id, "name": f"Pl {playlist_index}", "owner": { "accountId": f"Owner {playlist_index}", "categoryId": playlist_index, "categoryName": f"Cat {playlist_index}", }, "countries": [ { "countryCode": pl_country_code, "playlistTrackCount": 10 + playlist_index, "trackPosition": 11 + index, "positionChange": (index + index * (-1 if index % 3 else 1)) if has_previous else None, "positionChangeDate": f"2022-11-1{index - 1}T16:37:41.205Z" if has_previous else None, "tracksLastAdded": f"2022-11-0{index}T14:35:26.105Z", "earliestAdded": f"2022-04-0{index}T11:08:52.123Z", "isrc": get_isrc(index), "streams7Days": 5002 + index * 100, "streams1Days": 202 + index * 10, "isrcStreams7Days": 252 + index * 10, "isrcStreams1Days": 52 + index * 5, } ], "artwork": ( "https://images-api.atlas.stream/v2/playlists/by_apple_music_id/" + f"{playlist_id}?storefront={pl_country_code}" ), "streamsGlobal7Days": 5001 + index * 100, "streamsGlobal1Days": 201 + index * 10, "isrcStreamsGlobal7Days": 251 + index * 10, "isrcStreamsGlobal1Days": 51 + index * 5, } result.append(result_item) base_url = f"/gate-api/v1/playlists/{dsp.value}/by-track/" return { "items": result, "count": count, "next": get_page_url(base_url, dsp, isrc_list, country_code, offset, limit, count, True), "previous": get_page_url(base_url, dsp, isrc_list, country_code, offset, limit, count, False), } @pytest.mark.parametrize( "country_code_list,dsp,params,status_code", ( ([], DSP.SPOTIFY, {}, HTTPStatus.BAD_REQUEST), ([], DSP.SPOTIFY, {"isrc": "AA11,BB22"}, HTTPStatus.OK), (["us"], DSP.SPOTIFY, {"isrc": "AA11", "market": "us"}, HTTPStatus.OK), ([], DSP.SPOTIFY, {"isrc": "AA11,BB22", "limit": 4}, HTTPStatus.OK), (["ca"], DSP.SPOTIFY, {"isrc": "AA11,BB22", "market": "ca", "offset": 8, "limit": 4}, HTTPStatus.OK), ([], DSP.APPLE, {}, HTTPStatus.BAD_REQUEST), (["local"], DSP.APPLE, {"isrc": "AA11,BB22"}, HTTPStatus.OK), (["local"], DSP.APPLE, {"isrc": "AA11", "country": "us"}, HTTPStatus.OK), (["local"], DSP.APPLE, {"isrc": "AA11,BB22", "limit": 3}, HTTPStatus.OK), (["local"], DSP.APPLE, {"isrc": "AA11,BB22", "offset": 4, "limit": 4}, HTTPStatus.OK), ), ) async def test_get_v1_playlists_by_track_current( country_code_list: Optional[List[str]], dsp: DSP, params: dict, status_code: HTTPStatus, mocker, auth, client ): count = 10 isrc_list, country_code, offset, limit = ( params.get("isrc"), params.get("market" if dsp.value == DSP.SPOTIFY.value else "country"), params.get("offset"), params.get("limit"), ) isrc_list = isrc_list.split(",") if isrc_list else isrc_list async_dsp_mock = AsyncMock( return_value=get_tracklist( [Market.WORLDWIDE] + country_code_list or [], dsp=dsp, count=count, same_playlist=False ) ) dsp_send_mocked = mocker.patch.object(DspApiClient, "send_request", side_effect=async_dsp_mock) response = await client.get(f"/api/v1/playlists/{dsp.value}/by-track/", params=params, headers=auth) assert response.status == status_code if status_code != HTTPStatus.OK: return dsp_send_mocked.assert_called_once_with( "api/delphi/public/track-positions/playlists", params={ "apple_country_code": None, "country_code": ( None if dsp.value == DSP.SPOTIFY.value else (country_code.split(",") if country_code else APPLE_PLAYLISTS_TOP_28_MARKETS) ), "dsp": dsp.value, "end_date": None, "expand_isrc_by_track_positions": False, "include": ["playlists", "streams_for_period", "num_tracks"], "isrc": isrc_list, "min_added_date": None, "playlist_id": None, "playlist_name_contains": None, "playlist_owner_category_id": None, "sort_by": ["streaming_info.worldwide.7.playlist"] if dsp.value == DSP.SPOTIFY.value else ["playlist_id"], "sort_order": ["desc"] if dsp.value == DSP.SPOTIFY.value else ["asc"], "start_date": None, "streams_country_code": ( ["worldwide"] + ( ([country_code] if country_code else []) if dsp.value == DSP.SPOTIFY.value else ["local"] ) ), }, ) response = await response.json() assert response == get_by_track_result( country_code_list, dsp, count, offset=offset, limit=limit, isrc_list=isrc_list, country_code=country_code ) def get_previous_playlists_track(isrc: str, index: int) -> dict: return { "track_id": f"GRAS_tr{index:02d}", "name": f"Tr_{index:02d}", "track_name_suppl": f"Suppl{index:02d}", "product": { "name": f"Tr_{index:02d}", "product_id": f"GRAS_pr{index:02d}", "digital_title_suppl": f"dts{index:02d}", "label_id": f"LBL{index:02d}", "label_name": "SME", "product_version": { "product_version_no": 151 * index, "config_cat_key": "C", "config_cat_name": "Longplay", "is_product_family_head": bool(index % 5), "label_id": f"LBL{index:02d}", "label_name": "SME", "main_artist_id": f"GRAS_ar{index:02d}", "product_class_key": "M", "product_class_name": "Music", "product_family_no": 231 * index, "product_title": f"Ttl{index:02d}", "product_title_suppl": f"pts{index:02d}", "rep_owner_key": f"C{index * 5:02d}", "rep_owner_company": { "company_key": f"C{index * 5:02d}", "company_name": "Epic Records Group" } }, "sub_title": f"ST{index:02d}", "config_key": f"D{index}", "is_explicit": bool(index % 3), "release_date": f"2022-01-1{index % 10}" }, "isrc": isrc, "release_date": f"2011-11-2{index % 10}", "rep_owner_key": f"C{index * 5:02d}", "rep_owner_company": { "company_key": f"C{index * 5:02d}", "company_name": "Epic Records Group" }, "track_type_name": "Simple Track", "lyrics_version": "Explicit", } def get_previous_playlists( isrc_list: List[str] = None, dsp: Optional[DSP] = None, offset: Optional[int] = None, limit: Optional[int] = None, count: int = 10, streams_country_code: Optional[str] = None, ) -> dict: result = [] if isrc_list: for index in range((offset or 0) + 1, min(limit + (offset or 0) if limit else count, count) + 1): dsp = get_dsp(dsp, index) playlist_index = get_playlist_index(dsp, index, False) isrc = isrc_list[index % len(isrc_list)] result.append( { "dsp": dsp.value, "isrc": isrc, "earliest_position_date_time": f"2022-04-0{index % 9 + 1}T11:08:52.123Z", "removed_date_time": f"2023-01-1{index % 5}T15:42:05.659Z", "num_days_on": (5 + index) if index % 3 else 0, "playlist_dates": { "dsp_update_date_time": f"2023-01-1{index % 10}T15:42:05.659Z", "playlist_updated_at_date_time": f"2023-01-1{index % 10}T15:21:07.741Z" }, "playlist_id": f"{dsp.value}_{get_playlist_id(dsp, playlist_index)}", "playlist": get_playlist(dsp, playlist_index, bool(playlist_index % 2)), "track": get_previous_playlists_track(isrc, index), "streaming_info": { streams_country_code: { "1": {"playlist": index * 11 if index % 3 else None}, "7": {"playlist": index * 32 if index % 4 else None}, "14": {"playlist": index * 45 if index % 5 else None}, }, "worldwide": { "1": {"playlist": index * 11 if index % 3 else None}, "7": {"playlist": index * 64 if index % 2 else None}, "14": {"playlist": index * 45 if index % 5 else None}, }, } if streams_country_code else None, } ) return { "items": result, "count": count, "meta": {"owner_categories": [{"owner_category_id": "1", "name": ""}]}, } def get_by_track_history_result( isrc_list: List[str], dsp: Optional[DSP] = None, country_code: str = None, offset: Optional[int] = None, limit: Optional[int] = None, count: int = 10, with_streams: bool = False, ) -> dict: result = [] for index in range((offset or 0) + 1, min(limit + (offset or 0) if limit else count, count) + 1): dsp = get_dsp(dsp, index) playlist_index = get_playlist_index(dsp, index, False) playlist_id = get_playlist_id(dsp, playlist_index) isrc = isrc_list[index % len(isrc_list)] item_country_code = COUNTRY_CODE_LIST[(playlist_index - 1) % len(COUNTRY_CODE_LIST)] if dsp.value == DSP.SPOTIFY.value: result_item = { "isrc": isrc, "earliestTrackDate": f"2022-04-0{index % 9 + 1}T11:08:52.123Z", "latestTrackDate": f"2023-01-1{index % 5}T15:42:05.659Z", "playlist": { "spotifyLink": f"spotify:playlist:{playlist_id}", "name": f"Pl {playlist_index}", "subscribers": 80000 + playlist_index, "countryCode": item_country_code, "is_personalized": bool(playlist_index % 2), "buzzCategoryId": playlist_index, "accountName": f"Owner {playlist_index}", "imageFileName": f"https://images-api.atlas.stream/v2/playlists/by_spotify_id/{playlist_id}", "accountId": str(playlist_index), }, "daysInPlaylist": (5 + index) if index % 3 else 0, "streams": { "global7Days": index * 64 if index % 2 else None, "local7Days": index * 32 if index % 4 else None, } if with_streams else None, } else: result_item = { "id": get_playlist_id(dsp, playlist_index), "name": f"Pl {playlist_index}", "owner": { "accountId": f"Owner {playlist_index}", "categoryId": playlist_index, "categoryName": f"Cat {playlist_index}", }, "countries": [ { "countryCode": item_country_code, "daysInPlaylist": (5 + index) if index % 3 else 0, "earliestAdded": f"2022-04-0{index % 9 + 1}T11:08:52.123Z", "latestDate": f"2023-01-1{index % 5}T15:42:05.659Z", "isrc": isrc, **({"streams_7_days": index * 32 if index % 4 else None} if with_streams else {}) }, ], "artwork": ( "https://images-api.atlas.stream/v2/playlists/by_apple_music_id/" + f"{playlist_id}?storefront={item_country_code}" ), "playlistType": f"pl_type_{playlist_index}", **({"streams_global_7_days": index * 64 if index % 2 else None} if with_streams else {}) } result.append(result_item) base_url = f"/gate-api/v1/playlists/{dsp.value}/by-track/history/" return { "items": result, "count": count, "next": get_page_url(base_url, dsp, isrc_list, country_code, offset, limit, count, True), "previous": get_page_url(base_url, dsp, isrc_list, country_code, offset, limit, count, False), } @pytest.mark.parametrize( "dsp,params,status_code", ( (DSP.SPOTIFY, {}, HTTPStatus.BAD_REQUEST), (DSP.SPOTIFY, {"isrc": "AA11,BB22"}, HTTPStatus.OK), (DSP.SPOTIFY, {"isrc": "AA11", "market": "us"}, HTTPStatus.OK), (DSP.SPOTIFY, {"isrc": "AA11,BB22", "limit": 4}, HTTPStatus.OK), (DSP.SPOTIFY, {"isrc": "AA11,BB22", "market": "ca", "offset": 8, "limit": 4}, HTTPStatus.OK), (DSP.SPOTIFY, {"isrc": "AA11", "market": "us", "include": "streams_for_period"}, HTTPStatus.OK), (DSP.SPOTIFY, {"isrc": "AA11", "include": "streams_for_period", "streams_country_code": "us"}, HTTPStatus.OK), (DSP.APPLE, {}, HTTPStatus.BAD_REQUEST), (DSP.APPLE, {"isrc": "AA11,BB22"}, HTTPStatus.OK), (DSP.APPLE, {"isrc": "AA11", "country": "us"}, HTTPStatus.OK), (DSP.APPLE, {"isrc": "AA11,BB22", "limit": 3}, HTTPStatus.OK), (DSP.APPLE, {"isrc": "AA11,BB22", "offset": 4, "limit": 4}, HTTPStatus.OK), (DSP.APPLE, {"isrc": "AA11,BB22", "include": "streams_for_period"}, HTTPStatus.OK), ), ) async def test_get_v1_playlists_by_track_history(dsp: DSP, params: dict, status_code: HTTPStatus, mocker, auth, client): isrc_list, country_code, offset, limit, include, streams_country_code = ( params.get("isrc"), params.get("market" if dsp.value == DSP.SPOTIFY.value else "country"), params.get("offset"), params.get("limit"), params.get("include"), params.get("streams_country_code", "local"), ) isrc_list = isrc_list.split(",") if isrc_list else isrc_list async_dsp_mock = AsyncMock( return_value=get_previous_playlists( isrc_list, dsp, streams_country_code=streams_country_code if include else None ) ) dsp_send_mocked = mocker.patch.object(DspApiClient, "send_request", side_effect=async_dsp_mock) response = await client.get(f"/api/v1/playlists/{dsp.value}/by-track/history/", params=params, headers=auth) assert response.status == status_code if status_code != HTTPStatus.OK: return dsp_send_mocked.assert_called_once_with( "api/delphi/public/track-positions/previous/playlists", params=dict( dsp=dsp.value, isrc=isrc_list, country_code=( (country_code.split(",") if country_code else APPLE_PLAYLISTS_TOP_28_MARKETS) if dsp.value == DSP.APPLE.value else (country_code if country_code else None) ), include=["playlists"] + ([include] if include else []), playlist_name_contains=None, playlist_owner_category_id=None, min_removed_date=None, sort_by=["removed_date_time"], sort_order=["desc"] if dsp.value == DSP.SPOTIFY.value else ["asc"], streams_country_code=["worldwide", streams_country_code], ), ) response = await response.json() assert response == get_by_track_history_result( isrc_list, dsp, country_code=country_code, offset=offset, limit=limit, with_streams=bool(include) )