from http import HTTPStatus as http_status from unittest.mock import AsyncMock import pytest from apollo_utils.core.constants.dsp import DSP_SPOTIFY_APPLE from server.client.clients.user_data_api import UserDataApiClient from server.client.clients.vendor_api import VendorApiClient from tests.data.base.albums import album_id from tests.data.base.artists import artist_id, artist_name, dsp_to_uri_func as dsp_to_artist_uri_func from tests.data.base.core import dt, image_url, apple_image_url from tests.data.base.tracks import isrc, track_name, track_id, dsp_to_uri_func as dsp_to_track_uri_func, \ spotify_api_track, apple_api_track from tests.data.base.dsp import dsp_getter from tests.api.test_distributors import response_data as distributors_result_data get_dsp = dsp_getter() def favorite_track_item(i, dsp: str = None): dsp = dsp or get_dsp(i) created_at = dt(days_shift=-i, iso=True) return { "entity_id": isrc(i), "entity_type": "track", "favorites_id": i, "data": { "id": track_id(i), "dsp": dsp, "country_code": "us" }, "created_at": created_at, "updated_at": created_at } def favorite_tracks(n=3, count=10, dsp_to_ids=None): id_to_dsp = {_id: dsp for dsp, ids in (dsp_to_ids or {}).items() for _id in ids} or \ {i: get_dsp(i) for i in range(1, n + 1)} return { "count": count, "items": [favorite_track_item(i, dsp=dsp) for i, dsp in id_to_dsp.items()], "next": f"/user-data-api/users/favorites/?" f"entity_type=track&limit={n}&offset={n}", "previous": None } def tracks_meta(n=3, dsp_to_ids=None): id_to_dsp = {_id: dsp for dsp, ids in (dsp_to_ids or {}).items() for _id in ids} or \ {i: get_dsp(i) for i in range(1, n + 1)} return { "spotify": [spotify_api_track(i) for i, dsp in id_to_dsp.items() if dsp == DSP_SPOTIFY_APPLE.SPOTIFY.value], "apple": [apple_api_track(i) for i, dsp in id_to_dsp.items() if dsp == DSP_SPOTIFY_APPLE.APPLE.value] } def starred_track_response_item(i, dsp: str = None, distributor: str = None): dsp = dsp or get_dsp(i) _track_id, _isrc, _track_uri = track_id(i), isrc(i), dsp_to_track_uri_func[dsp](i) _image_url = image_url(i, "track", dsp, 300) if dsp == DSP_SPOTIFY_APPLE.SPOTIFY.value else \ apple_image_url(i, "track", size=300) return { "favorites_id": i, "id": _track_id, "isrc": _isrc, "uri": _track_uri, "vendor": dsp, "created_at": dt(days_shift=-i, iso=True), "streams": 0, "streams_7days_change": 0, "streams_7days_change_rate": 0, "tt_creations_7days_change": 0, "tt_creations_7days_change_rate": 0, "available": True, "distributed_by": distributor, "extra_data": { "id": _track_id, "vendor": dsp, "isrc": _isrc, "uri": _track_uri, "name": track_name(i), "release_date": dt(days_shift=(-30 * i), iso=True, with_time=False), "image": _image_url, "album_ids": [ album_id(i) ], "artists": [ { "id": artist_id(i), "name": artist_name(i), "uri": dsp_to_artist_uri_func[dsp](i) } ], "full_id": f"{dsp}_{_track_id}" }, } def starred_tracks_response(n=3, count=10, dsp_to_ids=None, dist_to_ids=None): id_to_dsp = {_id: dsp for dsp, ids in (dsp_to_ids or {}).items() for _id in ids} or \ {i: get_dsp(i) for i in range(1, n + 1)} id_to_distributors = {_id: dist for dist, ids in (dist_to_ids or {}).items() for _id in ids} return { "count": count, "items": [starred_track_response_item( i, dsp=dsp, distributor=id_to_distributors.get(i)) for i, dsp in id_to_dsp.items()], "next": f"/gate-api/users/favorites/tracks/detailed/?" f"limit={n}&offset={n}", "previous": None } @pytest.mark.skip("dev only") @pytest.mark.parametrize( "params, status, favorites_result, tracks_meta_result, distributors_result, expected_response, tracks_meta_called, distributors_called", ( ({"distributors": ["unknown"]}, http_status.BAD_REQUEST, None, None, None, None, None, None), ( { # distributors = sme by default "market": "global", }, http_status.OK, favorite_tracks(dsp_to_ids={"apple": (1, 2, 3, 4)}), tracks_meta(dsp_to_ids={"apple": (1, 2, 3, 4)}), ([], {}), starred_tracks_response(dsp_to_ids={"apple": (1, 2, 3, 4)}), True, True ), ( { "distributors": ["sme", "theorchard", "awal"], "market": "us", }, http_status.OK, favorite_tracks(dsp_to_ids={"spotify": (1, 3, 4), "apple": (2, 5)}), tracks_meta(dsp_to_ids={"spotify": (1, 3, 4), "apple": (2, 5)}), distributors_result_data( dist_to_ids={"awal": (2,), "theorchard": (4, 5), "sme": (1, 3)}, dsp_to_ids={"spotify": (1, 3, 4), "apple": (2, 5)}, by_track=True, with_extra_data=True), starred_tracks_response( dist_to_ids={"awal": (2,), "theorchard": (4, 5), "sme": (1, 3)}, dsp_to_ids={"spotify": (1, 3, 4), "apple": (2, 5)}), True, True ), ( { "distributors": ["sme", "theorchard", "awal"], "market": "us", }, http_status.OK, {"count": 0, "items": [], "next": None, "previous": None}, None, None, {"count": 0, "items": [], "next": None, "previous": None}, False, False ), ), ) async def test_favorites_tracks_detailed( mocker, client, auth, params, status, favorites_result, tracks_meta_result, distributors_result, expected_response, tracks_meta_called, distributors_called): async_favorite_tracks_mock = AsyncMock(return_value=favorites_result) favorites_mock = mocker.patch.object( UserDataApiClient, "get_user_favorites", side_effect=async_favorite_tracks_mock) async_distributors_mock = AsyncMock(return_value=distributors_result) distributors_mock = mocker.patch( "server.scenarios.distributors.distributors.get_distributors", side_effect=async_distributors_mock) async_search_tracks_mock = AsyncMock(return_value=tracks_meta_result) search_tracks_mock = mocker.patch.object( VendorApiClient, "get_tracks_v1", side_effect=async_search_tracks_mock) headers = {"X-User-Id": "my-user", "X-App-Slug": "apollo", **auth} response = await client.get(f"/api/users/favorites/tracks/detailed/", params=params, headers=headers) assert response.status == status if expected_response is None: return response_json = await response.json() assert response_json["items"] == expected_response["items"] assert favorites_mock.call_count == 1 assert distributors_mock.call_count == int(distributors_called) assert search_tracks_mock.call_count == int(tracks_meta_called)