from http import HTTPStatus from typing import List import pytest from server.client.clients.dsp_api import DspApiClient from server.client.clients.vendor_api import VendorApiClient def _get_playlist_tracks(tracks_date: str) -> dict: return { "items": [ { "current": 10 + 4 * i, "date": tracks_date if tracks_date else f"2022-06-0{i + 2}", "earliest_position_date": f"2022-03-0{i}", "isrc": f"isrc_{i}", "last_date_change_14_days": f"2022-05-2{i}", "latest_position_date": f"2022-06-0{i}", "num_days_on": 7 * i, "num_weeks_on": i, "previous": 20 + 2 * i, "previous_position_change_14_days": 30 + 3 * i, "trend": -10 + 2 * i, "trend_change_14_days": -20 + i, "playlist": None if i == 1 else { "amazon_playlist_id": f"am_pl_id", "country_code": "us", "dsp_playlist_id": f"pl_id", "image": { "uri": f"https://m.media-amazon.com/images/I/id.jpg", }, "name": f"AmPl", "num_tracks": 40, "playlist_id": f"am_pl_id_us", "rank": 50, }, "amazon_track": ( None if i == 2 else {"amazon_track_id": f"am_tr_id_{i}", "name": f"AmTr{i}", "country_code": "us"} ), "track": ( None if i in (1, 3, 6) else {"name": f"Tr{i}", "release_date": f"2022-02-1{i}", "track_id": f"gras_tr_id_{i}"} ), } for i in range(1, 7) ] } def _get_track_artists(index: int) -> dict: return { "items": ( [{"full_name": ", ".join([f"DeAr{i}" for i in range(1, min(index + 1, 4))])}] if index == 4 else [] ) } def _get_apple_track(index_list: List[int]) -> dict: return { "data": [ { "id": f"apple_track_id_{i}", "attributes": { "artwork": {"url": f"https://is4-ssl.mzstatic.com/image/thumb/{i}/{{w}}x{{h}}.jpg"}, "releaseDate": f"2021-12-2{i}", "name": f"ApTr{i}", "isrc": f"isrc_{i}", }, "relationships": { "artists": { "data": [ {"id": f"ap_ar_{i}", "attributes": {"name": f"ApAr{j}"}} for j in range(1, min(i + 1, 4)) ] } } } for i in index_list if i <= 3 ] } def _get_tracks_result(tracks_date: str) -> dict: return { "items": [ { "current": 10 + 4 * i, "date": tracks_date if tracks_date else f"2022-06-0{i + 2}", "earliest_position_date": f"2022-03-0{i}", "isrc": f"isrc_{i}", "last_date_change_14_days": f"2022-05-2{i}", "latest_position_date": f"2022-06-0{i}", "num_days_on": 7 * i, "num_weeks_on": i, "previous": 20 + 2 * i, "previous_position_change_14_days": 30 + 3 * i, "trend": -10 + 2 * i, "trend_change_14_days": -20 + i, "track": ( { "apple_track_id": f"apple_track_id_{i}", "artists": [{"id": f"ap_ar_{i}", "name": f"ApAr{j}"} for j in range(1, min(i + 1, 4))], "image_url": f"https://is4-ssl.mzstatic.com/image/thumb/{i}/{{w}}x{{h}}.jpg", "name": f"ApTr{i}", "release_date": f"2021-12-2{i}", "track_id": f"gras_tr_id_{i}" if i == 2 else None, **({"amazon_track_id": f"am_tr_id_{i}", "country_code": "us"} if i != 2 else {}), } if i <= 3 else ( { "amazon_track_id": f"am_tr_id_{i}", "artists": [] if i == 5 else [{"name": f"DeAr{i}"} for i in range(1, min(i + 1, 4))], "name": f"Tr{i}", "release_date": f"2022-02-1{i}", "track_id": f"gras_tr_id_{i}", "country_code": "us", } if i in (4, 5) else {"amazon_track_id": f"am_tr_id_{i}", "name": f"AmTr{i}", "country_code": "us"} ) ), } for i in range(1, 7) ], "playlist": { "amazon_playlist_id": f"am_pl_id", "dsp_playlist_id": f"pl_id", "image_url": f"https://m.media-amazon.com/images/I/id.jpg", "name": "AmPl", "num_tracks": 40, "playlist_id": f"am_pl_id_us", "rank": 50, "country_code": "us", }, } @pytest.mark.parametrize( "params,expected_status", ( ({}, HTTPStatus.BAD_REQUEST), ({"country_code": "us"}, HTTPStatus.BAD_REQUEST), ( {"playlist_id": "250", "country_code": "us", "start_date": "2022-05-02", "end_date": "2022-05-01"}, HTTPStatus.BAD_REQUEST, ), ({"playlist_id": "250", "country_code": "us"}, HTTPStatus.OK), ( {"playlist_id": "250", "country_code": "us", "start_date": "2022-05-01", "end_date": "2022-05-01"}, HTTPStatus.OK, ), ), ) async def test_get_playlists_amazon_tracks(params, expected_status, mocker, client, auth): async def dsp_send(url: str, *args, **kwargs): if url == "api/delphi/track-positions/playlists": tracks_date = kwargs["params"].get("end_date") return _get_playlist_tracks(tracks_date) elif url == "api/delphi/artists": isrc = kwargs["params"]["isrc"] index = int(isrc.replace("isrc_", "")) return _get_track_artists(index) async def vendor_send(url: str, *args, **kwargs): isrc_list = kwargs["params"]["isrc"] index_list = [int(isrc.replace("isrc_", "")) for isrc in isrc_list] return _get_apple_track(index_list) apollo_send_mock = mocker.patch.object(DspApiClient, "_send_request", side_effect=dsp_send) vendor_send_mock = mocker.patch.object(VendorApiClient, "_send_request", side_effect=vendor_send) response = await client.get("/api/playlists/amazon/tracks/", params=params, headers=auth) assert response.status == expected_status is_ok = response.status == HTTPStatus.OK assert apollo_send_mock.call_count == (3 if is_ok else 0) assert vendor_send_mock.call_count == (1 if is_ok else 0) if is_ok: response_json = await response.json() response_json["items"] = list(sorted(response_json["items"], key=lambda i: i["isrc"])) assert response_json == _get_tracks_result(params.get("end_date"))