from collections import namedtuple from http import HTTPStatus as http_status from unittest.mock import AsyncMock import pytest from server.scenarios.distributors import distributors from server.client.clients.apollo_api import ApolloApiClient from server.client.clients.dsp_api import DspApiClient from tests.data.base.distributors import get_tracks_brands_tagging_mock, get_distributors_mock from tests.helpers import seq from tests.data.base.albums import upc from tests.data.base.tracks import track_id, spotify_track_id, apple_track_id def sme_data(n=3, by_track=False, ids=None, ids_to_upc=None): # by default return n odds: 1, 3, 5, ... ids = ids or [2 * i + 1 for i in range(n)] ids_to_upc = ids_to_upc or {i: i for i in ids} if by_track: track_ids = [track_id(i) for i in ids] return track_ids, {"track_upc_mapping": {track_id(i): upc(j) for i, j in ids_to_upc.items()}} return [upc(i) for i in ids] non_sme_data = get_tracks_brands_tagging_mock response_data = get_distributors_mock MockResult = namedtuple("MockResult", ("count", "args", "kwargs")) @pytest.mark.parametrize( "params,status,sme_response,non_sme_response,expected_response,sme_track_called,sme_album_called,non_sme_called", ( ({"country_code": "us"}, http_status.BAD_REQUEST, None, None, None, None, None, None), ( {"track_id": seq(track_id, 3), "country_code": "us"}, http_status.BAD_REQUEST, None, None, None, None, None, None ), ( {"track_id": seq(spotify_track_id, 3), "upc": seq(upc, 3), "country_code": "us"}, http_status.BAD_REQUEST, None, None, None, None, None, None ), ({"track_id": seq(apple_track_id, 3)}, http_status.BAD_REQUEST, None, None, None, None, None, None), ( { "track_id": seq(spotify_track_id, items=(1, 2, 3)) + seq(apple_track_id, items=(4, 5)), "country_code": "us" }, http_status.OK, sme_data(ids=(1,), by_track=True), non_sme_data( dist_to_ids={"awal": (2, 4), "theorchard": (5,)}, dsp_to_ids={"apple": (4, 5), "spotify": (1, 2, 3)}, unknown_ids=(1, 3,), by_track=True), response_data( dist_to_ids={"awal": (2, 4), "theorchard": (5,), "sme": (1,)}, dsp_to_ids={"apple": (4, 5), "spotify": (1, 2)}, by_track=True), MockResult( 1, [], { "track_id_list": seq(track_id, items=(1, 2, 3)), "country_code": "us", "track_list": None, "album_list": None, "include": {'track_upc_mapping'} }), MockResult(0, None, None), MockResult(1, None, None) ), ( { "track_id": seq(spotify_track_id, items=(1, 2, 3)) + seq(apple_track_id, items=(4, 5, 6)), "country_code": "global", "distributors": ["sme", "theorchard", "awal"] }, http_status.OK, sme_data(ids=(1, 3), by_track=True), non_sme_data( dist_to_ids={"awal": (2, 4), "theorchard": (1, 4, 5)}, dsp_to_ids={"apple": (4, 5, 6), "spotify": (1, 2, 3)}, unknown_ids=(3, 6), by_track=True), response_data( dist_to_ids={"awal": (2,), "theorchard": (4, 5), "sme": (1, 3)}, dsp_to_ids={"apple": (4, 5), "spotify": (1, 2, 3)}, by_track=True), MockResult( 1, [], { "track_id_list": seq(track_id, items=(1, 2, 3)), "country_code": "global", "track_list": None, "album_list": None, "include": {'track_upc_mapping'} }), MockResult(0, None, None), MockResult(1, ["api/delphi/tracks/brands-tagging"], {"params": { "track_id": seq(spotify_track_id, items=(1, 2, 3)) + seq(apple_track_id, items=(4, 5, 6)), "country_code": "worldwide", "upc": None, }}) ), ( { "upc": seq(upc, 3), "country_code": "worldwide", "distributors": ["sme"] }, http_status.OK, sme_data(ids=(1, 3), by_track=False), None, response_data(dist_to_ids={"sme": (1, 3)}, by_track=False), MockResult(0, None, None), MockResult(1, [], {"upc": seq(upc, 3), "market": "global"}), MockResult(0, None, None) ), ( { "upc": seq(upc, 5), "country_code": "us", "distributors": ["sme", "awal"] }, http_status.OK, sme_data(ids=(1, 3, 5), by_track=False), non_sme_data(dist_to_ids={"awal": (1, 2, 4), "theorchard": (4, 5)}, by_track=False), response_data(dist_to_ids={"awal": (2, 4), "sme": (1, 3, 5)}, by_track=False), None, None, None ), ), ) async def test_distributors_check( mocker, client, auth, params, status, sme_response, non_sme_response, expected_response, sme_track_called, sme_album_called, non_sme_called): async_dsp_mock = AsyncMock(return_value=non_sme_response) non_sme_mock = mocker.patch.object(DspApiClient, "send_request", side_effect=async_dsp_mock) async_album_sony_mock = AsyncMock(return_value=sme_response) sme_album_mock = mocker.patch.object(ApolloApiClient, "get_albums_is_sony", side_effect=async_album_sony_mock) async_check_sony_mock = AsyncMock(return_value=sme_response) sme_track_mock = mocker.patch.object(distributors, "tracks_is_sony", side_effect=async_check_sony_mock) response = await client.get(f"/api/distributors/check/", params=params, headers=auth) assert response.status == status if expected_response is None: return response_json = await response.json() expected_response.sort(key=lambda x: x["upc"]) response_json.sort(key=lambda x: x["upc"]) assert response_json == expected_response for mock_result, mock in ( (sme_track_called, sme_track_mock), (sme_album_called, sme_album_mock), (non_sme_called, non_sme_mock)): if mock_result is not None: if mock_result.count is not None: assert mock.call_count == mock_result.count if mock_result.args is not None or mock_result.kwargs is not None: mock.assert_called_with(*(mock_result.args or []), **(mock_result.kwargs or {}))