from typing import List, Tuple from http import HTTPStatus import pytest from server.client.clients.user_data_api import UserDataApiClient from server.client.clients.dsp_api import DspApiClient from server.client.clients.vendor_api import VendorApiClient from server.constants.charts.tracks.streams import ChartsTracksInclude from tests.helpers import value_from def _get_is_starred(index: int) -> bool: return index % 3 == 0 async def get_is_starred(*args, **kwargs) -> dict: isrc_list = kwargs["params"]["entity_id"] result = [] for isrc in isrc_list: isrc_idx = int(isrc.replace("isrc_", "")) if _get_is_starred(isrc_idx): result.append({"entity_id": isrc, "favorites_id": isrc_idx}) return {"items": result} def _get_spotify_track(index: int, track_id: str = None) -> dict: size_list = (64, 300, 640) if not track_id: track_id = f"track_id_{index}" return { "album": { "images": [ {"url": f"tr_img_{index}_{j}", "width": size_list[j % 3]} for j in range(1, min(index, 3) + 1) ], }, "uri": f"spotify:track:{track_id}", } def _get_spotify_episode(index: int, episode_id: str = None) -> dict: size_list = (64, 300, 640) if not episode_id: episode_id = f"episode_id_{index}" return { "images": [ {"url": f"ep_img_{index}_{j}", "width": size_list[j % 3]} for j in range(1, min(index, 3) + 1) ], "uri": f"spotify:episode:{episode_id}", } async def get_spotify_data(url: str, *args, **kwargs) -> dict: id_list = kwargs["params"]["ids"] if url == "api/spotify/v1/tracks": return { "tracks": [_get_spotify_track(int(track_id.replace("track_id_", "")), track_id) for track_id in id_list] } elif url == "api/spotify/v1/episodes": return { "episodes": [ _get_spotify_episode(int(episode_id.replace("episode_id_", "")), episode_id) for episode_id in id_list ] } async def get_distributors_data(track_id, country_code, distributors, track_list, include) -> Tuple[List[dict], dict]: get_distributor = value_from([d.value for d in (distributors or [])], with_none=True) return [ { "track_id": t_id, "distributed_by": get_distributor(int(t_id.split('_')[-1])) } for t_id in track_id ], {} def _get_juno_data( country_code: List[str], isrc_country_code: List[str], offset: int, limit: int, include: list = None, distributors: list = None, ) -> List[dict]: if not include: include = [] get_distributor = value_from(distributors or [], with_none=True) return [ { "isrc": f"isrc_{i}", "latest_entry_date": f"2021-0{i}-15", "product_sale_date": f"2021-0{i}-16", "country_code": country_code[i % len(country_code)], "isrc_country_code": isrc_country_code[i % len(isrc_country_code)], "distributed_by": get_distributor(i) if i % 3 else None, "spotify_top_track_meta": { "spotify_track_id": f"some_id_{i}", "track_name": f"Tr_{i}", "artist_names": [f"Ar_{i}_{j}" for j in range(1, min(i, 2) + 1)], "track_uri": f"spotify:track:track_id_{i}" if i % 3 else f"spotify:episode:episode_id_{i}", }, "gras_track_meta": { "track_id": f"GRAS_{i}", "track_name": f"GrTr_{i}", "release_date": f"2020-0{i}-17", "label": {"name": f"L_{i} LLC", "label_id": f"l_id_{i}"}, "artists": [ { "artist_id": f"gr_ar_id_{i}_{j}", "full_name": f"GrAr {i}_{j}", "first_name": "Gr", "last_name": "Ar", "apple_artist_id": i * 1000 + j, "spotify_artist_id": f"spotify:artist:{i}_{j}", "spotify_popularity": i * 10 + j, "image": {"uri": f"https://img_{i}_{j}.jpg", "height": 300, "width": 300}, } for j in range(1, min(i, 2) + 1) ], }, "metrics": { "totals": {"streams_days_7": 200 * i + i, "streams_days_8_14": 300 * i + i}, "change": {"streams_change_7_to_14_days": 40 * i, "streams_change_percent_7_to_14_days": 10.2 * i}, "lean_back": {"streams_days_7": 70 * i, "streams_days_8_14": 80 * i}, "lean_forward": {"streams_days_7": 10 * i, "streams_days_8_14": 20 * i}, "demographics": { "male": {"streams_days_7": 30 * i, "streams_days_8_14": 25 * i}, "female": {"streams_days_7": 20 * i, "streams_days_8_14": 40 * i}, } }, **( {"spotify_meta": _get_spotify_track(i) if i % 3 else _get_spotify_episode(i)} if ChartsTracksInclude.SPOTIFY_META.value in include else {} ), **( {"image_url": f"{'tr' if i % 3 else 'ep'}_img_{i}_2" if i >= 2 else None} if ChartsTracksInclude.IMAGE_URL.value in include else {} ), **({"is_starred": _get_is_starred(i), "favorites_id": i if _get_is_starred(i) else None} if ChartsTracksInclude.IS_STARRED.value in include else {}), } for i in range(offset + 1, limit + offset + 1) ] async def get_juno_tracks(*args, **kwargs) -> dict: params = kwargs["params"] return { "items": _get_juno_data( params["country_code"], params.get("isrc_country_code", ["us", "gb", "ca", "de"]), params.get("offset", 0), params.get("limit", 5), ), } def get_result(params: dict) -> List[dict]: distributors = params.get("distributors") if distributors: distributors = distributors.split(",") return _get_juno_data( params["country_code"].split(","), params.get("isrc_country_code", "us,gb,ca,de").split(","), params.get("offset", 0), params.get("limit", 5), params.get("include", "").split(","), distributors, ) @pytest.mark.parametrize( "params,status,call_count", ( ({}, HTTPStatus.BAD_REQUEST, (0, 0, 0, 0)), ({"country_code": "us,gb"}, HTTPStatus.OK, (0, 1, 0, 0)), ({"country_code": "us,gb", "include": "country_codes,spotify_meta"}, HTTPStatus.OK, (0, 1, 2, 0)), ({"country_code": "ca,gb", "include": "is_starred"}, HTTPStatus.OK, (1, 1, 0, 0)), ( {"country_code": "us,gb", "include": "image_url,is_starred", "distributors": "theorchard"}, HTTPStatus.OK, (1, 1, 2, 1) ), ({"country_code": "us,gb", "distributors": "sme,awal"}, HTTPStatus.OK, (0, 1, 0, 1)), ( { "country_code": "de,am,es", "isrc_country_code": "us,ca", "percent_change": 20.2, "max_product_sale_date": "2022-06-01", "min_product_sale_date": "2000-01-01", "sort_by": "streams_change_7_to_14_days", "sort_order": "desc", "limit": 10, "offset": 2, "spotify_country_code": "ca", "include": "image_url,is_starred,country_codes,spotify_meta", }, HTTPStatus.OK, (1, 1, 2, 0), ), ), ) async def test_get_charts_tracks_streams_spotify(params, status, call_count, mocker, auth, client): ud_send_mock = mocker.patch.object(UserDataApiClient, "_send_request", side_effect=get_is_starred) dsp_send_mock = mocker.patch.object(DspApiClient, "_send_request", side_effect=get_juno_tracks) vendor_send_mock = mocker.patch.object(VendorApiClient, "_send_request", side_effect=get_spotify_data) distributors_mock = mocker.patch( "server.scenarios.distributors.distributors.get_distributors", side_effect=get_distributors_data) response = await client.get("/api/charts/tracks/streams/spotify/", params=params, headers=auth) assert response.status == status assert ud_send_mock.call_count == call_count[0] assert dsp_send_mock.call_count == call_count[1] assert vendor_send_mock.call_count == call_count[2] assert distributors_mock.call_count == call_count[3] if status == HTTPStatus.OK: response = await response.json() assert response == get_result(params)