"""Tests for Track Logic.""" from collections import ChainMap from copy import deepcopy import json from unittest.mock import call from unittest.mock import MagicMock import application from oto import response from oto import status as response_code from owslogger import flask_logger import pytest from backend import features from backend.constants import error from backend.constants import feature_flag from backend.constants import field as field_const from backend.constants import header from backend.constants import track_field as tf from backend.constants.product import IS_ERR_CORR_MODE from backend.exceptions import RequestError from backend.logic import isrc as isrc_logic from backend.logic import track as track_logic from backend.logic import track_sample as sample_logic from backend.models import ows_account from backend.models import ows_assets from backend.models import ows_blacklist_manager from backend.models import ows_lyrics from backend.models import ows_product from backend.models import ows_product_digital from backend.models import ows_sound_recordings from backend.models import ows_video from backend.models.track_persister import TrackPersister from backend.utils import localization as loc_util from backend.utils import track_utils from backend.utils import validation from tests.testutils import db from tests.testutils import functions as test_funcs from tests.testutils import mocks def test_get_by_tuid_access_denied(mocker, track_factory, track_localization_factory): """Verifies that a track cannot be accessed from an incorrect vendor.""" track = track_factory( publishers__count=2, performer__count=2, writers__count=1) # This mocks the ownership check to fail for all vendors. mocks.ows_product_verify_product_ownership(mocker, []) mocks.track_persister.get_by_tuid(mocker, track) track_response = track_logic.get_by_tuid( tuid=track.tuid, account_type='vendor', account_id='alw:123', ) assert track_response.status == response_code.BAD_REQUEST def test_get_by_tuid(mocker, track_factory, track_localization_factory): """Verify update fails when metadata lang has confliction localization.""" track = track_factory( publishers__count=2, performer__count=2, writers__count=1) localizations = track_localization_factory(track, [1]) mocks.ows_product_get_track_localizations(mocker, localizations) mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_by_tuid(mocker, track) for localization in localizations: for artist in localization['artists']: del artist['artist_info_id'] track_response = track_logic.get_by_tuid( tuid=track.tuid, account_type='vendor', account_id='alw:123', ) track_dict = track.to_dict() assert track_response.status == response_code.OK assert track_response.message[tf.LOCALIZATIONS] == localizations for field_name, value in track_response.message.items(): if field_name != tf.LOCALIZATIONS: assert value == track_dict[field_name] error_msg = test_funcs.validate_track_response(track_response.message) assert not error_msg def test_get_by_tuid_no_localizations(mocker, track_factory): """Verify get track by tuid without localization.""" track = track_factory( publishers__count=2, performer__count=2, writers__count=1) mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_by_tuid(mocker, track) track_response = track_logic.get_by_tuid( tuid=track.tuid, account_type='vendor', account_id='alw:123', exclude=['localizations'], ) track_dict = track.to_dict() assert track_response.status == response_code.OK assert track_response.message[tf.LOCALIZATIONS] == [] for field_name, value in track_response.message.items(): if field_name != tf.LOCALIZATIONS: assert value == track_dict[field_name] error_msg = test_funcs.validate_track_response(track_response.message) assert not error_msg def test_get_by_tuid_with_samples(mocker): """Verify get track by tuid with track samples when FFlag is ON.""" tuid = 27825628 merged = {'foo': 'bar'} original = {'original': 'response', 'product_id': 1} mocker.patch.object( TrackPersister, 'get_by_tuid', return_value=response.Response(message=original)) mocker.patch.object( track_utils, 'add_track_samples_to_response', return_value=response.Response(message=merged)) mocks.ows_product_verify_product_ownership(mocker) track_response = track_logic.get_by_tuid( tuid=tuid, account_type='vendor', account_id='alw:123', exclude=[tf.LOCALIZATIONS], include=[tf.TRACK_SAMPLES], ) assert track_response.message == {'foo': 'bar', 'localizations': []} def test_get_multiple_by_tuids( mocker, track_factory, track_localization_factory): """Verify update fails when metadata lang has confliction localization.""" tracks = [track_factory(), track_factory()] localizations = ( track_localization_factory(tracks[0], [1]) + track_localization_factory(tracks[1], [1])) mocks.ows_product_get_track_localizations(mocker, localizations) mocks.track_persister.get_multiple_by_tuids(mocker, tracks) track_response = track_logic.get_multiple_by_tuids( [track.tuid for track in tracks]) assert track_response.status == response_code.OK for item in track_response.message['items']: assert item['localizations'] error_msg = test_funcs.validate_track_response(track_response.message) assert not error_msg def test_get_multiple_by_tuids_no_localizations(mocker, track_factory): """Verify get track by tuid without localization.""" tracks = [track_factory(), track_factory()] mocks.track_persister.get_multiple_by_tuids(mocker, tracks) track_response = track_logic.get_multiple_by_tuids( [track.tuid for track in tracks], exclude=['localizations']) assert track_response.status == response_code.OK assert track_response.message['pagination']['total_records'] == 2 assert track_response.message['items'][0]['localizations'] == [] assert track_response.message['items'][1]['localizations'] == [] error_msg = test_funcs.validate_track_response(track_response.message) assert not error_msg def test_get_multiple_by_tuids_with_nones(mocker, track_factory): """Verify get track by tuid with nones.""" tracks = [track_factory(), track_factory()] mocks.track_persister.get_multiple_by_tuids_with_nones(mocker, tracks) track_response = track_logic.get_multiple_by_tuids_with_nones( [track.tuid for track in tracks], ) assert track_response.status == response_code.OK assert track_response.message['pagination']['total_records'] == 2 assert track_response.message['items'][0]['tuid'] == tracks[0].tuid assert track_response.message['items'][1]['tuid'] == tracks[1].tuid error_msg = test_funcs.validate_track_response(track_response.message) assert not error_msg def test_get_all_tracks_by_product_id( mocker, track_factory, track_localization_factory, logic_headers): """Verify update fails when metadata lang has confliction localization.""" tracks = [ track_factory(publishers__count=2, performer__count=2), track_factory(performer__count=2, writers__count=2), track_factory() ] product_id = tracks[0].product_id localizations = track_localization_factory(tracks[0], [1]) localizations.extend(track_localization_factory(tracks[1], [1, 2])) mocks.ows_product_get_track_localizations(mocker, localizations) mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_all_by_product_id(mocker, tracks) track_response = track_logic.get_all_tracks_by_product_id( product_id=product_id, exclude=[], **logic_headers) error_msg = test_funcs.validate_track_response(track_response.message) assert not error_msg track_iter = iter(tracks) for track_res_dict in track_response.message['items']: track = next(track_iter).to_dict() for field_name, value in track_res_dict.items(): if field_name != tf.LOCALIZATIONS: assert value == track[field_name] def test_get_all_tracks_by_product_id_is_overview( mocker, track_factory, track_localization_factory, logic_headers): """Verify getting all tracks by product id for overview is successful.""" tracks = [ track_factory(publishers__count=2, performer__count=2), track_factory(performer__count=2, writers__count=2), track_factory() ] product_id = tracks[0].product_id localizations = track_localization_factory(tracks[0], [1]) localizations.extend(track_localization_factory(tracks[1], [1, 2])) mocks.ows_product_get_track_localizations(mocker, localizations) mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_all_by_product_id_is_overview(mocker, tracks) track_response = track_logic.get_all_tracks_by_product_id( product_id=product_id, exclude=['localizations'], **logic_headers) assert not track_response.errors track_iter = iter(tracks) for track_res_dict in track_response.message['items']: track = next(track_iter).to_dict() for field_name, value in track_res_dict.items(): if field_name != tf.LOCALIZATIONS: assert value == track[field_name] def test_get_all_tracks_by_product_id_light( mocker, track_factory, logic_headers): """Verify getting all tracks by product id light is successful.""" tracks = [ track_factory(publishers__count=2, performer__count=2), track_factory(performer__count=2, writers__count=2), track_factory() ] product_id = tracks[0].product_id mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_all_by_product_id_light(mocker, tracks) track_response = track_logic.get_all_tracks_by_product_id_light( product_id=product_id, **logic_headers) error_msg = test_funcs.validate_track_response(track_response.message) assert not error_msg track_iter = iter(tracks) for track_res_dict in track_response.message['items']: track = next(track_iter).to_dict() for field_name, value in track_res_dict.items(): if field_name != tf.LOCALIZATIONS: assert value == track[field_name] def test_get_all_tracks_by_product_ids_medium(mocker, track_factory): """Verify update fails when metadata lang has confliction localization.""" tracks = [track_factory(), track_factory()] mocks.track_persister.get_all_by_product_ids_medium(mocker, tracks) track_response = track_logic.get_all_tracks_by_product_ids_medium( [tracks[0].product_id]) assert track_response.status == response_code.OK error_msg = test_funcs.validate_track_response(track_response.message) assert not error_msg @pytest.mark.parametrize( ( 'exclude' ), [ ([]), ([tf.LOCALIZATIONS]) ] ) def test_get_all_tracks_by_product_id_excludes( mocker, context, logic_headers, exclude): """Test get_all_tracks_by_product_id() with excludes.""" mocker.patch.object( ows_product, '_get_head_from_ows_product_service', return_value=response.Response()) mocker.patch.object( TrackPersister, 'get_all_by_product_id', return_value=response.Response()) mocker.patch.object( loc_util, 'add_track_localizations_to_response', return_value=response.Response()) mocker.patch.object( track_logic, '_make_track_response', return_value=response.Response()) with context: track_logic.get_all_tracks_by_product_id( product_id=1234, account_type=logic_headers.get('account_type'), account_id=logic_headers.get('account_id'), exclude=exclude) assert loc_util.add_track_localizations_to_response.called !=\ (tf.LOCALIZATIONS in exclude) def test_create(mocker, logic_headers, track_create_data, track_create_expected_response, track_create_expected_logic_response): """Verify Track is created.""" mocks.track_persister.create_track(mocker, track_create_expected_response) mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=100, upc=123456789012) track_response = track_logic.create( track_data=track_create_data, **logic_headers) assert track_response.message == track_create_expected_logic_response def test_create_fail(mocker, logic_headers): """Verify Track failure response works.""" mocks.track_persister.create_track( mocker, response.create_error_response(code='code', message='')) mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=100, upc=123456789012) track_create_data = { 'product_id': 100, 'some_invalid_field': '' } track_response = track_logic.create( track_data=track_create_data, **logic_headers) assert track_response.status == response_code.BAD_REQUEST def test_bulk_create( mocker, logic_headers, track_bulk_create_data, track_bulk_create_new_prod_response, track_bulk_create_new_prod_logic_response): """Verify Track is created.""" mocks.track_persister.bulk_create_tracks( mocker, track_bulk_create_new_prod_response) first_track = track_bulk_create_new_prod_response['items'][0] product_id = first_track[tf.PRODUCT_ID] upc = first_track[tf.UPC] mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=product_id, upc=upc) track_response = track_logic.bulk_create( product_id, track_bulk_create_data, for_volume=None, **logic_headers) assert track_response.message == track_bulk_create_new_prod_logic_response def test_bulk_create_fail( mocker, logic_headers, track_bulk_create_data): """Verify Track failure response works.""" mocks.track_persister.bulk_create_tracks( mocker, response.create_error_response(code='code', message='')) product_id = 100 mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=product_id, upc=123456789012) track_response = track_logic.bulk_create( product_id, track_bulk_create_data, for_volume=None, **logic_headers) assert track_response.status == response_code.BAD_REQUEST def test_update( mocker, logic_headers, track_factory, track_localization_factory, test_product_id): """Verify Track is updated.""" track = track_factory() localizations = track_localization_factory(track, [1]) update_data = track.to_dict() update_data[tf.TRACK_NAME] = 'Updated track name' update_data[tf.EXPLICIT] = 'Y' update_response = test_funcs.make_track_logic_response(update_data) update_response[tf.LOCALIZATIONS] = localizations mocks.ows_product_verify_product_ownership(mocker) mocks.ows_product_get_track_localizations( mocker, localizations) mocks.ows_product_get_localization_languages(mocker) mocks.track_persister.get_by_tuid(mocker, track) mocks.ows_product_digital_get_chinese_product_by_id( mocker, product_id=test_product_id) mocks.track_persister.update_track(mocker, update_data) track_response = track_logic.update( tuid=track.tuid, data=deepcopy(update_data), **logic_headers) assert track_response assert track_response.message == update_response @pytest.mark.parametrize( ( 'single_focus_track_enabled', 'focus_track', 'focus_track_start_date' ), [ ( True, 'Y', None ), ( True, 'N', None ), ( False, 'Y', '2022-10-22' ), ( False, 'N', None ), ] ) def test_update_with_focus_track( mocker, mock_app, single_focus_track_enabled, focus_track, focus_track_start_date, logic_headers, logic_user, track_factory, feature_flags ): """Verify that user information is passed on when focus track.""" feature_flags[feature_flag.INCFEATURES_SINGLE_FOCUS_TRACK] = single_focus_track_enabled mocker.patch.object( features, 'is_incfeatures_single_focus_track_enabled', return_value=single_focus_track_enabled ) track = track_factory() # Create test data track_patch_request_data = { tf.VERSION: 'Meow Mix', tf.EXPLICIT: 'Y', tf.FOCUS_TRACK: focus_track, tf.FOCUS_TRACK_START_DATE: focus_track_start_date } track_patch_response_data = deepcopy(track_patch_request_data) track_patch_response_data[tf.TUID] = track.tuid updated_track_dict = track.to_dict() for key, value in track_patch_request_data.items(): updated_track_dict[key] = value mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_by_tuid(mocker, track) mocks.track_persister.get_all_focus_track_by_product_id(mocker, []) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_chinese_product_by_id(mocker) mock_update_track = mocker.patch( 'backend.models.track_persister.TrackPersister.update_track') track_logic.update( tuid=track.tuid, data=track_patch_request_data, patch_response=True, **ChainMap(logic_headers, logic_user)) _, kwargs = mock_update_track.call_args assert kwargs['data'].get('focus_track') == focus_track if focus_track == 'N': assert kwargs['data'].get('user_id') is None assert kwargs['data'].get('user_type') is None assert kwargs['data'].get('focus_track_start_date') is None else: assert kwargs['data'].get('user_id') == logic_user['user_id'] assert kwargs['data'].get('user_type') == logic_user['user_type'] assert kwargs['data'].get('focus_track_start_date') == focus_track_start_date def test_update_invalid_metadata(mocker, logic_headers, track_factory, test_product_id): """Verify update fails when metadata lang code is invalid.""" track = track_factory() update_data = {tf.META_LANGUAGE_CODE: 'B00'} mocks.ows_product_verify_product_ownership(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.track_persister.get_by_tuid(mocker, track) mocks.ows_product_digital_get_product_by_id( mocker, product_id=test_product_id, meta_language='B00') track_response = track_logic.update( tuid=track.tuid, data=update_data, **logic_headers) assert not track_response assert track_response.errors['code'] == error.VALIDATION_ERROR_CODE assert 'Metalanguage' in track_response.errors['message'] def test_update_conflicting_metadata( mocker, logic_headers, track_factory, track_localization_factory, test_product_id): """Verify update fails when metadata lang has confliction localization.""" track = track_factory() meta_lang_id = 2 meta_lang_code = test_funcs.lookup_language_code(meta_lang_id) localizations = track_localization_factory(track, [1, meta_lang_id]) update_data = {tf.META_LANGUAGE_CODE: meta_lang_code} mocks.ows_product_verify_product_ownership(mocker) mocks.ows_product_get_track_localizations(mocker, localizations) mocks.ows_product_get_localization_languages(mocker) mocks.track_persister.get_by_tuid(mocker, track) mocks.ows_product_digital_get_product_by_id( mocker, product_id=test_product_id, meta_language=meta_lang_code) track_response = track_logic.update( tuid=track.tuid, data=update_data, **logic_headers) assert not track_response assert track_response.errors['code'] == error.VALIDATION_ERROR_CODE assert 'Metalanguage' in track_response.errors['message'] def test_update_patch(mocker, logic_headers, track_factory): """Verify Track is updated using patch data.""" track = track_factory() # Create test data track_patch_request_data = {tf.VERSION: 'Meow Mix', tf.EXPLICIT: 'Y'} track_patch_response_data = deepcopy(track_patch_request_data) track_patch_response_data[tf.TUID] = track.tuid updated_track_dict = track.to_dict() for key, value in track_patch_request_data.items(): updated_track_dict[key] = value mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_by_tuid(mocker, track) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_chinese_product_by_id(mocker) mocks.track_persister.update_track(mocker, updated_track_dict) track_response = track_logic.update( tuid=track.tuid, data=track_patch_request_data, patch_response=True, **logic_headers) assert track_response.message == track_patch_response_data def test_update_patch_offer_type_oa_user( mocker, logic_headers, track_factory): """Verify Track Offer Type is updated for OA users.""" track = track_factory() # Create test data track_patch_request_data = {tf.VERSION: 'Meow Mix', tf.EXPLICIT: 'Y'} track_patch_response_data = deepcopy(track_patch_request_data) track_patch_response_data[tf.TUID] = track.tuid updated_track_dict = track.to_dict() for key, value in track_patch_request_data.items(): updated_track_dict[key] = value mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_by_tuid(mocker, track) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_chinese_product_by_id(mocker) mock_update_track = mocker.patch( 'backend.models.track_persister.TrackPersister.update_track') offer_type = 'stream_only' track_patch_request_data['offer_type'] = offer_type track_logic.update( tuid=track.tuid, data=track_patch_request_data, user_type=header.OA_USER_TYPE, patch_response=True, **logic_headers) _, kwargs = mock_update_track.call_args assert kwargs['data']['offer_type'] == offer_type def test_update_patch_offer_type_non_oa_user( mocker, logic_headers, track_factory): """Verify Track Offer Type is not updated for non OA users.""" track = track_factory() # Create test data track_patch_request_data = {tf.VERSION: 'Meow Mix', tf.EXPLICIT: 'Y'} track_patch_response_data = deepcopy(track_patch_request_data) track_patch_response_data[tf.TUID] = track.tuid updated_track_dict = track.to_dict() for key, value in track_patch_request_data.items(): updated_track_dict[key] = value mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_by_tuid(mocker, track) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_chinese_product_by_id(mocker) mock_update_track = mocker.patch( 'backend.models.track_persister.TrackPersister.update_track') offer_type = 'stream_only' track_patch_request_data['offer_type'] = offer_type track_logic.update( tuid=track.tuid, data=track_patch_request_data, patch_response=True, **logic_headers) _, kwargs = mock_update_track.call_args assert not kwargs['data'].get('offer_type') def test_delete( mocker, test_track, track_localization_put_data, track_update_expected_response): """Verify Track is deleted. A second check is also done to make sure the asset is deleted before the track metadata is deleted. """ src_mock = MagicMock() mocker.patch( 'backend.logic.track.ows_assets.bulk_delete_track_assets', new=src_mock.bulk_delete_track_assets) mocks.ows_product_get_track_localizations( mocker, track_localization_put_data) mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_by_tuid(mocker, track_update_expected_response) src_mock.delete_by_tuid.configure_mock(return_value=response.Response()) mocker.patch.object( TrackPersister, 'delete_by_tuid', new=src_mock.delete_by_tuid) delete_localizations_response = response.Response(status=response_code.OK) mocker.patch.object( ows_lyrics, 'delete_track_lyrics', return_value=response.Response()) mocker.patch.object( ows_video, 'disassociate_tracks', return_value=response.Response()) mocker.patch( 'backend.logic.track.ows_product.delete_track_localizations', return_value=delete_localizations_response) mock_delete_performers = mocker.patch( 'backend.logic.track.performer_logic.delete_performers_by_tuid') mock_delete_audio_attribute_logic = mocker.patch( 'backend.logic.track.audio_attribute_logic.bulk_delete_track_audio_attributes') mock_delete_rights_attribute_logic = mocker.patch( 'backend.logic.track.rights_attribute_logic.bulk_delete_track_rights_attributes') result = track_logic.delete( tuid=test_track[tf.TUID], account_type='test_account', account_id=1) # Order of expected calls for functions of interest expected_calls = [ call.bulk_delete_track_assets([test_track[tf.TUID]]), call.delete_by_tuid( product_id=test_track[tf.PRODUCT_ID], tuid=test_track[tf.TUID])] assert src_mock.mock_calls == expected_calls assert result.status == response_code.OK mock_delete_performers.assert_called_with(test_track[tf.TUID]) mock_delete_audio_attribute_logic.assert_called_with([test_track[tf.TUID]]) mock_delete_rights_attribute_logic.assert_called_with([test_track[tf.TUID]]) def test_update_field_for_all_tracks_in_product( mocker, logic_headers, test_product_id, product_apply_to_all_data, product_apply_to_all_response): """Verify tracks for product are updated.""" mocker.patch.object( TrackPersister, 'update_field_for_all_tracks_in_product', return_value=MagicMock( status=200, message=product_apply_to_all_response)) mocks.head_response(mocker) response = track_logic.update_field_for_all_tracks_in_product( product_id=test_product_id, data=product_apply_to_all_data, **logic_headers) assert response.status == 200 assert response.message == product_apply_to_all_response def test_update_field_for_all_tracks_in_product_with_localization( mocker, tds_product_apply_to_all_version): """Verify tracks for product are updated.""" tds = tds_product_apply_to_all_version for item in tds.persister_response.message['items']: item[tf.ARTISTS] = [] mocker.patch.object( TrackPersister, 'update_field_for_all_tracks_in_product', return_value=test_funcs.copy_response(tds.persister_response)) mocks.head_response(mocker) mocks.put_response(mocker) mocks.ows_product_get_track_localizations( mocker, tds.ows_product_track_localizations.copy()) mocks.ows_product_get_localization_languages(mocker) spy = mocker.spy(ows_product, 'update_localizations') response = track_logic.update_field_for_all_tracks_in_product( product_id=tds.product_id, data=tds.request_data.copy(), **tds.logic_headers) assert response.status == 200 assert response.message == tds.logic_response.message # Make sure proper number of localization updates were sent to ows-product num_updates_done = len(spy.call_args[0][0]) num_locs_to_apply = len(tds.request_data[tf.LOCALIZATIONS]) num_tracks = len(tds.persister_response.message['items']) assert num_updates_done == num_locs_to_apply * num_tracks def test_update_metadata_for_all_tracks_in_product_with_dupe_localization( mocker, track_factory, track_localization_factory, logic_headers): """Verify error if metalanguage update has conflicting localization.""" tracks = [track_factory(), track_factory()] meta_lang_id = 1 localizations = track_localization_factory(tracks[1], [2, meta_lang_id]) meta_lang_code = test_funcs.lookup_language_code(meta_lang_id) data = { tf.META_LANGUAGE_CODE: meta_lang_code } mocks.head_response(mocker) mocks.ows_product_get_track_localizations(mocker, localizations) mocks.ows_product_get_localization_languages(mocker) mocks.track_persister.get_all_by_product_id(mocker, tracks) response = track_logic.update_field_for_all_tracks_in_product( product_id=tracks[0].product_id, data=data, **logic_headers) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.VALIDATION_ERROR_CODE def test_update_field_for_all_tracks_in_product_invalid_data( mocker, test_product_id, logic_headers): """Verify tracks for product return error with invalid data.""" mocks.head_response(mocker) response = track_logic.update_field_for_all_tracks_in_product( product_id=test_product_id, data={}, **logic_headers) assert response.status == 400 assert response.errors['code'] == error.INVALID_VALUE_ERROR_CODE def test_update_field_for_all_tracks_generate_isrc( mocker, logic_headers, test_product_id, product_apply_to_all_generate_isrc_response): """Verify tracks for product are updated with generated ISRC.""" isrc_mock = mocker.patch.object( TrackPersister, 'assign_track_isrcs_for_product_id', return_value=MagicMock( status=200, message=product_apply_to_all_generate_isrc_response)) mocks.head_response(mocker) response = track_logic.update_field_for_all_tracks_in_product( product_id=test_product_id, data={}, generate_isrc=True, **logic_headers) assert isrc_mock.called assert response.status == 200 assert response.message == product_apply_to_all_generate_isrc_response def test_reorder( mocker, track_reorder_data, test_product_id, logic_headers, track_reorder_expected_response): """Verify Track is updated.""" mocker.patch.object(TrackPersister, 'reorder', return_value=MagicMock(status=200, message='success')) mocks.head_response(mocker) mocks.track_persister.get_all_by_product_id( mocker, track_reorder_expected_response['items']) track_response = track_logic.reorder( product_id=test_product_id, arrangement=track_reorder_data, **logic_headers) assert track_response.message == track_reorder_expected_response def test_validate_tracks_for_product( mocker, test_product_id, test_product_tracks_response, logic_headers, logic_user, mock_app ): """Verify number of tracks is returned.""" mocks.head_response(mocker) """ mocker.patch.object( track_logic.track_validators, 'is_enabled', return_value=False) """ mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, product_id=test_product_id) mocks.ows_product_get_localization_languages(mocker) mocks.track_persister.get_all_by_product_id( mocker, test_product_tracks_response['items']) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocker.patch.object( track_logic, 'post_submission_validation', ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) assert validation_response.message[ 'total_tracks'] == len(test_product_tracks_response['items']) def test_validate_tracks_for_classical_product( mocker, logic_headers, logic_user, track_factory, test_product_id, classical_genre_id, classical_subgenre_id, mock_app ): """Verify classical track with missing composer has validation error.""" tracks = [ track_factory(performer__count=1, composer__count=1, writers__count=1), track_factory(performer__count=1, writers__count=1)] subgenres = [ {'orchard_id': classical_subgenre_id, 'name': '20/21st Century', 'genre_id': classical_genre_id} ] mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, product_id=test_product_id, genre_id=classical_genre_id, subgenre_id=classical_subgenre_id) mocks.track_persister.get_all_by_product_id(mocker, tracks) mocks.subgenre_query.get_genre_subgenres(mocker, subgenres) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocker.patch.object( track_logic, 'post_submission_validation', ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user) ) assert validation_response.message['total_tracks'] == len(tracks) assert validation_response.message['valid_tracks'] == 1 validation_error = validation_response.message['errors'][0] assert 'composer' in validation_error assert validation_error['tuid'] == 2 def test_validate_tracks_for_product_with_empty_tracks( mocker, test_product_id, logic_headers, logic_user, ): """Verify number of tracks is zero when product doesn't have tracks.""" empty_tracks = [] mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.track_persister.get_all_by_product_id(mocker, empty_tracks) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user) ) assert validation_response.message['total_tracks'] == len(empty_tracks) @pytest.mark.parametrize('failure_field', [ 'track_name', 'performer', 'explicit', 'isrc', 'p_info', 'writers', 'meta_language_code' ]) def test_validate_tracks_for_product_fail( mocker, test_valid_product_id, test_product_valid_tracks_response, logic_headers, logic_user, failure_field, mock_app ): """Verify number of tracks is returned.""" mocker.patch.object( track_logic, 'post_submission_validation', ) test_product_tracks_fail_items = [] error_list = [] # Remove fields as parameterized for item in test_product_valid_tracks_response['items']: fail_item = item.copy() if failure_field == 'performer': fail_artist_list = [] for artist in fail_item['artists']: if artist['type'] != 'performer': fail_artist_list.append(artist) fail_item['artists'] = fail_artist_list fail_item[failure_field] = '' test_product_tracks_fail_items.append(fail_item) error_list.append( { tf.TUID: item[tf.TUID], failure_field: { field_const.VALIDATOR: 'required', field_const.VALIDATOR_VALUE: True, field_const.MESSAGE: error.VALIDATION_ERROR_MISSING_FIELD_MSG .format(failure_field), field_const.ERROR_CODE: f'missing_{failure_field}' } } ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.track_persister.get_all_by_product_id( mocker, test_product_tracks_fail_items) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists') mocker.patch.object(track_logic.track_validators, 'validate_master_rights') mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=test_valid_product_id, **ChainMap(logic_headers, logic_user) ) assert validation_response.message[ 'total_tracks'] == len(test_product_valid_tracks_response['items']) assert validation_response.message['valid_tracks'] == 0 assert validation_response.message['errors'] == error_list def test_validate_tracks_for_product_audio_file_fail( mocker, track_factory, logic_headers, logic_user, mock_app ): """Should return error because audio file doesn't exist.""" mocker.patch.object( track_logic, 'post_submission_validation', ) track = track_factory.create( writers__count=1, performer__count=1) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_digital_get_product_by_id(mocker) mocks.track_persister.get_all_by_product_id(mocker, [track]) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) expected_error = [ { 'audio_file': validation.format_missing_error( field_name='audio_file', error_code=error.MISSING_AUDIO_FILE ), 'tuid': 1 }] validation_response = track_logic.validate_tracks_for_product( product_id=track.to_dict()['product_id'], **ChainMap(logic_headers, logic_user) ) assert validation_response.message['total_tracks'] == 1 assert validation_response.message['valid_tracks'] == 0 assert validation_response.message['errors'] == expected_error def test_validate_tracks_for_product_master_rights_fail( mocker, track_factory, logic_headers, logic_user, mock_app ): """Should return error because master rights are not set.""" mocker.patch.object( track_logic, 'post_submission_validation', ) track = track_factory.create( writers__count=1, performer__count=1, ownership_rights=None) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_digital_get_product_by_id(mocker) mocks.track_persister.get_all_by_product_id(mocker, [track]) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) expected_error = [ { 'ownership_rights': validation.format_missing_error( field_name='ownership_rights', error_code=error.MISSING_OWNERSHIP_RIGHTS ), 'tuid': 1 }] validation_response = track_logic.validate_tracks_for_product( product_id=track.to_dict()['product_id'], **ChainMap(logic_headers, logic_user) ) assert validation_response.message['total_tracks'] == 1 assert validation_response.message['valid_tracks'] == 0 assert validation_response.message['errors'] == expected_error def test_validate_tracks_for_product_unique_track_name( mocker, logic_headers, logic_user, track_factory, mock_app ): """Validate track_name, version is unique.""" mocker.patch.object( track_logic, 'post_submission_validation', ) tracks = [ track_factory( track_name='Meow Tracks', version='', performer__count=1, writers__count=1), track_factory( track_name='Some other track', version='', performer__count=1, writers__count=1), track_factory( track_name='Meow Tracks', version='', performer__count=1, writers__count=1)] # set the 1st and 3rd track to have the same performing artist tracks[0].artists[0].name = 'An Artist' tracks[0].artists[0].artist_type = 'performer' tracks[2].artists[0].name = 'An Artist' tracks[2].artists[0].artist_type = 'performer' mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker) mocks.track_persister.get_all_by_product_id(mocker, tracks) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=1, **ChainMap(logic_headers, logic_user) ) assert validation_response.message['valid_tracks'] == 2 error = validation_response.message['errors'][0] assert tf.TRACK_NAME in error assert error[tf.TRACK_NAME][field_const.VALIDATOR] == 'used' @pytest.mark.parametrize( ( 'release_correction_status', 'track_corrections_used_in_validation' ), [ ('submitted', True), ('pizza', False) ] ) def test_validate_tracks_for_submitted_corrections( release_correction_status, track_corrections_used_in_validation, mocker, mock_app ): """Test track corrections used for validation when release_correction.status is submitted.""" product_tracks_response_message = {'items': [{'tuid': 38266597, 'product_id': 4201664, 'ownership_rights': 'no_rights', 'track_name': '', 'version': '', 'meta_language_code': 'ENG', 'isrc': 'QMDA72165872', 'p_info': '1999 pline', 'explicit': 'N', 'volume_number': 1, 'track_number': 1, 'upc': 197187490349, 'recording_country_id': None, 'original_rights_holder_country_id': None, 'preview_start_time': None, 'track_type': 'video', 'offer_type': 'all', 'duration_minutes': 0, 'duration_seconds': 24, 'artists': [{'track_artist_id': 99195740, 'type': 'performer', 'name': 'Test Shot Starfish', 'artist_info_id': None}], 'writers': [{'track_writer_id': 70277387, 'type': 'writer', 'name': 'writer22', 'artist_info_id': None}], 'language': {'language_code': 'ENG', 'iso_639_1_code': 'EN', 'iso_code_639_3_code': '', 'language': 'English', 'apple_compatible': True}, 'localizations': []}], 'pagination': {'type': 'none', 'total_records': 1}} # noqa mocker.patch.object( track_logic, 'get_all_tracks_by_product_id', return_value=response.Response(message=product_tracks_response_message)) mocker.patch.object( ows_product, 'verify_product_ownership', return_value=response.Response()) # product not in error correction, setting release_correction.status from test parameter digital_product_response_message = {'artist_id': 2566267, 'description': None, 'distribution_format_id': 1, 'format': 'Single', 'genre_id': 5, 'c_line': '1999 cline', 'imprint': 'Endmusik Records', 'manufacturer_upc': None, 'meta_language': 'ENG', 'p_line': None, 'preorder_date': None, 'product_code': None, 'previewable': None, 'product_id': 4201664, 'product_name': 'Rishi Test 004', 'product_type_id': 1, 'project_id': 4795216, 'product_artists': [{'role': 'primary_artist', 'name': 'Test Shot Starfish', 'id': 8678275, 'artist_info_id': 2566267}], 'delivered_version': None, 'release_date': '2021-11-24', 'release_status': 'transfer_to_content', 'is_error_correction_and_action_required': False, 'sale_start_date': '2021-11-24', 'special_instructions': None, 'subaccount_id': None, 'subgenre_id': 64, 'upc': '197187490349', 'vendor_catalog_number': 'GDWUG3728', 'version': None, 'not_for_distribution': 'N', 'corrections': {'release_correction_id': 404876, 'release_id': 4201664, 'status': release_correction_status, 'items': [{'release_correction_detail_id': 5294804, 'table_name': 'track', 'field_name': 'track_name', 'key_id': 38266597, 'key_value': 'Pizza Power'}, {'release_correction_detail_id': 5294805, 'table_name': 'track', 'field_name': 'lyrics', 'key_id': 38266597, 'key_value': 'asdasdasd\nasdasdasdasd\nasdasdasdas\ndasdasdasdasd'}, {'release_correction_detail_id': 5294806, 'table_name': 'track', 'field_name': 'p_line', 'key_id': 38266597, 'key_value': '2022 Bees'}]}, 'release_approval_status': 'checked_in'} # noqa mocker.patch.object( ows_product_digital, 'get_product_by_product_id', return_value=response.Response(message=digital_product_response_message)) product_response_message = {'product_id': 4201664, 'distribution_format_id': 1, 'project_id': 4795216, 'product_type_id': 1, 'upc': 197187490349, 'context_type': 'digital', 'vendor_id': 16162, 'subaccount_id': 0, 'release_date': '2021-11-24', 'status': 'in_content', 'product_name': 'Rishi Test 004', 'deletions': 'N', 'not_for_distribution': 'N'} # noqa mocker.patch.object( ows_product, 'get_product_by_product_id', return_value=response.Response(message=product_response_message)) asset_response_message = {'items': [{'track_unique_id': 38266597, 'status': 'encoding_completed', 'duration': 45113, 'asset_upload_type': 'stereo'}], 'pagination': {'type': 'none', 'total_records': 1}} # noqa mocker.patch.object( ows_assets, 'get_product_assets_v2', return_value=response.Response(message=asset_response_message)) mocker.patch.object( ows_product, 'get_localization_languages', return_value=response.Response(message={'items': []})) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=1, account_type='vendor', account_id=100, user_type=None, user_id=None ) resp = validation_response.message assert resp['total_tracks'] == 1 if track_corrections_used_in_validation is True: assert resp['valid_tracks'] == 1 assert resp['errors'] == [] assert resp['valid'] is True else: assert resp['valid_tracks'] == 0 assert len(resp['errors']) assert resp['valid'] is False def test_validate_tracks_for_product_post_submission_validation_error( mocker, test_product_id, test_product_tracks_response, logic_headers, logic_user, mock_app ): """Test creates error response for post submission validation.""" mocker.patch.object( features, 'is_validate_missing_producer_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, product_id=test_product_id) mocks.ows_product_get_localization_languages(mocker) mocks.track_persister.get_all_by_product_id( mocker, test_product_tracks_response['items']) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocker.patch.object( track_logic, 'post_submission_validation', side_effect=RequestError( 'Error in post submission validation', http_status=504, error_code='some_error_code_xyz' ) ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) assert validation_response.status == 504 assert validation_response.errors == { 'code': 'some_error_code_xyz', 'message': 'Error in post submission validation' } @db.test_schema def test_import_tracks_no_artists( mocker, logic_headers, tds_copy_localization_single_track_no_artist, request_engine): """Verify track import returns result without artists.""" request_engine['ows-features'].add_spec('GET', r'/features/user/\w+', {}) _test_import_tracks( mocker, logic_headers, tds_copy_localization_single_track_no_artist) @db.test_schema def test_import_tracks_with_artists( mocker, logic_headers, tds_copy_localization_single_track_with_artists, request_engine): """Verify track import returns correct result when using artists.""" request_engine['ows-features'].add_spec('GET', r'/features/user/\w+', {}) _test_import_tracks( mocker, logic_headers, tds_copy_localization_single_track_with_artists) def _test_import_tracks(mocker, logic_headers, tds): """Generic test for track with multiple localizations.""" mocks.track_persister.import_tracks(mocker, tds['source_dest_track_list']) mocks.head_response(mocker) mocks.ows_assets_copy_assets(mocker) mocks.ows_product_update_track_localizations(mocker) mocks.ows_product_get_track_localizations( mocker, tds['source_localizations']) mocks.ows_product_get_product_by_id( mocker, tds['destination_product_id'], tds['destination_product_upc']) mocks.track_persister.get_multiple_by_tuids(mocker, [tds['source_track']]) mocker.patch.object( ows_lyrics, 'copy_tracks_lyrics', return_value=response.Response()) mocker.patch.object( track_logic.logic_util, 'copy_tracks_assets_in_parallel') # No destination track has spatial data mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) mock_get_product_spatial = mocker.patch.object( ows_product_digital, 'get_product_spatial') mock_create_product_spatial = mocker.patch.object( ows_product_digital, 'create_product_spatial') with application.app.app_context() as context: context.g.ows = flask_logger.Ows() context.g.ows.correlation_id = 'correlation_id' context.g.request_context = MagicMock() import_response = track_logic.import_tracks( {'tuids': tds['tuids']}, product_id=tds['destination_product_id'], orchard_user_id=1, correlation_id='1a9', **logic_headers) assert import_response tds['validate_copied_track']( import_response.message['items'][0], validate_logical=True) # No spatial track found — get/create product spatial should not be called mock_get_product_spatial.assert_not_called() mock_create_product_spatial.assert_not_called() @db.test_schema def test_import_tracks_with_spatial( mocker, logic_headers, tds_copy_localization_single_track_no_artist, request_engine): """Verify create_product_spatial is called when a spatial track is found and no existing release spatial record exists.""" request_engine['ows-features'].add_spec('GET', r'/features/user/\w+', {}) tds = tds_copy_localization_single_track_no_artist mocks.track_persister.import_tracks(mocker, tds['source_dest_track_list']) mocks.head_response(mocker) mocks.ows_assets_copy_assets(mocker) mocks.ows_product_update_track_localizations(mocker) mocks.ows_product_get_track_localizations( mocker, tds['source_localizations']) mocks.ows_product_get_product_by_id( mocker, tds['destination_product_id'], tds['destination_product_upc']) mocks.track_persister.get_multiple_by_tuids(mocker, [tds['source_track']]) mocker.patch.object( ows_lyrics, 'copy_tracks_lyrics', return_value=response.Response()) mocker.patch.object( track_logic.logic_util, 'copy_tracks_assets_in_parallel') # Destination track has spatial data destination_tuid = tds['destination_track'][tf.TUID] mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={destination_tuid: 'USRC12345678'})) # No existing release_spatial record mock_get_product_spatial = mocker.patch.object( ows_product_digital, 'get_product_spatial', return_value=response.create_not_found_response()) mock_create_product_spatial = mocker.patch.object( ows_product_digital, 'create_product_spatial', return_value=response.Response()) with application.app.app_context() as context: context.g.ows = flask_logger.Ows() context.g.ows.correlation_id = 'correlation_id' context.g.request_context = MagicMock() import_response = track_logic.import_tracks( {'tuids': tds['tuids']}, product_id=tds['destination_product_id'], orchard_user_id=1, correlation_id='1a9', **logic_headers) assert import_response mock_get_product_spatial.assert_called_once_with(tds['destination_product_id']) mock_create_product_spatial.assert_called_once_with( tds['destination_product_id'], 1) def test_create_release_spatial_if_spatial_isrc_present_with_no_spatial_tracks(mocker): """Verify release spatial is not created when no destination track is spatial.""" import_source_dest_list = [ {'destination': {tf.TUID: 11}}, {'destination': {tf.TUID: 12}}, ] mock_get_spatial_isrc_map = mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={11: None, 12: None})) mock_get_product_spatial = mocker.patch.object( ows_product_digital, 'get_product_spatial') mock_create_product_spatial = mocker.patch.object( ows_product_digital, 'create_product_spatial') track_logic._create_release_spatial_if_spatial_isrc_present( import_source_dest_list, dest_product_id=321, orchard_user_id=9, ) mock_get_spatial_isrc_map.assert_called_once_with([11, 12]) mock_get_product_spatial.assert_not_called() mock_create_product_spatial.assert_not_called() def test_create_release_spatial_if_spatial_isrc_not_found(mocker, mock_app): """Verify release spatial is created when destination has spatial tracks.""" import_source_dest_list = [ {'destination': {tf.TUID: 21}}, ] mock_get_spatial_isrc_map = mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={21: 'USRC12345678'})) mock_get_product_spatial = mocker.patch.object( ows_product_digital, 'get_product_spatial', return_value=response.create_not_found_response()) mock_create_product_spatial = mocker.patch.object( ows_product_digital, 'create_product_spatial', return_value=response.Response()) track_logic._create_release_spatial_if_spatial_isrc_present( import_source_dest_list, dest_product_id=654, orchard_user_id=7, ) mock_get_spatial_isrc_map.assert_called_once_with([21]) mock_get_product_spatial.assert_called_once_with(654) mock_create_product_spatial.assert_called_once_with(654, 7) def test_create_release_spatial_if_spatial_isrc_present_logs_error_on_unexpected_status( mocker, mock_app): """Verify errors are logged when product spatial status is unexpected.""" import_source_dest_list = [ {'destination': {tf.TUID: 31}}, ] mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={31: 'USRC12345678'})) mocker.patch.object( ows_product_digital, 'get_product_spatial', return_value=response.Response(status=response_code.BAD_REQUEST)) mock_create_product_spatial = mocker.patch.object( ows_product_digital, 'create_product_spatial') with mock_app: mock_g = mocker.patch('backend.logic.track.g') track_logic._create_release_spatial_if_spatial_isrc_present( import_source_dest_list, dest_product_id=987, orchard_user_id=5, ) mock_create_product_spatial.assert_not_called() expected_error = 'Failed to fetch spatial data for product 987' mock_g.log.error.assert_called_once_with(expected_error) def test_import_tracks_ownership_failed( mocker, test_product_id, logic_headers, tds_copy_localization_single_track_no_artist, copy_assets_failed_response): """Verify track import error if there is no ownership for a track.""" tds = tds_copy_localization_single_track_no_artist dest_product_id = tds['destination_track'][tf.PRODUCT_ID] mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, tds['destination_product_id'], tds['destination_product_upc']) mocks.ows_product_verify_product_ownership( mocker, owned_product_ids=[dest_product_id]) mocks.track_persister.get_multiple_by_tuids(mocker, [tds['source_track']]) import_response = track_logic.import_tracks( {'tuids': tds['tuids']}, product_id=tds['destination_product_id'], orchard_user_id=1, correlation_id='1a9', **logic_headers) assert not import_response assert import_response.errors['code'] == error.OWS_PRODUCT_ERROR_CODE def test_import_tracks_get_product_upc_failed( mocker, test_product_id, logic_headers, patch_ows_product_get_product_failed, test_track_model_response): """Verify track import error if there is no ownership for a track.""" source_tuid = test_track_model_response[tf.TUID] mocks.ows_product_verify_product_ownership(mocker) mocks.track_persister.get_by_tuid(mocker, test_track_model_response) tuids = {'tuids': [source_tuid]} response = track_logic.import_tracks( tuids, product_id=test_product_id, orchard_user_id=1, correlation_id='1a9', **logic_headers) assert not response assert response.errors['code'] == error.OWS_PRODUCT_ERROR_CODE def test_bulk_delete_tracks( mocker, logic_headers, test_product_id, tracks_delete_response): """Test bulk delete. A second check is also done to make sure the assets are deleted before the tracks metadata are deleted. """ tuids = [1, 2] src_mock = MagicMock() src_mock.delete_many_by_tuids.configure_mock( return_value=tracks_delete_response) mocker.patch.object( TrackPersister, 'delete_many_by_tuids', new=src_mock.delete_many_by_tuids) mocks.ows_product_delete_localizations_for_tracks(mocker, tuids) mocks.ows_product_verify_product_ownership(mocker) mocker.patch.object( ows_lyrics, 'delete_tracks_lyrics', return_value=response.Response()) mocker.patch.object( ows_video, 'disassociate_tracks', return_value=response.Response()) mocker.patch.object( track_logic.performer_logic, 'bulk_delete_performers_by_tuids') mocker.patch.object( sample_logic, 'bulk_delete_samples_by_tuids', return_value=response.Response()) mocker.patch.object( track_logic.audio_attribute_logic, 'bulk_delete_track_audio_attributes') mocker.patch.object( track_logic.rights_attribute_logic, 'bulk_delete_track_rights_attributes') delete_response = track_logic.bulk_delete( tuids, product_id=test_product_id, **logic_headers) # Order of expected calls for functions of interest expected_calls = [ call.delete_many_by_tuids(product_id=test_product_id, tuids=tuids)] assert delete_response assert delete_response.message == tracks_delete_response.message assert src_mock.mock_calls == expected_calls def test_bulk_delete_tracks_delete_localization_failed( mocker, test_product_id, logic_headers, tracks_delete_response, mock_app): """Test bulk delete when deleting a localization failed.""" mocker.patch.object( TrackPersister, 'delete_many_by_tuids', return_value=tracks_delete_response) mocker.patch( 'backend.models.ows_product.delete_track_localizations', return_value=response.create_fatal_response('Server error')) mocks.ows_product_verify_product_ownership(mocker) mocker.patch.object( ows_lyrics, 'delete_tracks_lyrics', return_value=response.Response()) mocker.patch.object( ows_video, 'disassociate_tracks', return_value=response.Response()) mocker.patch.object( track_logic.performer_logic, 'bulk_delete_performers_by_tuids') mocker.patch.object( sample_logic, 'bulk_delete_samples_by_tuids', return_value=response.Response()) mocker.patch.object( track_logic.audio_attribute_logic, 'bulk_delete_track_audio_attributes') mocker.patch.object( track_logic.rights_attribute_logic, 'bulk_delete_track_rights_attributes') tuids = [1, 2] with mock_app: mock_g = mocker.patch('backend.logic.track.g') delete_response = track_logic.bulk_delete( tuids, product_id=test_product_id, **logic_headers) assert delete_response assert delete_response.message == tracks_delete_response.message assert mock_g.log.error.called expected_error = 'Failed to delete track localizations: Server error' mock_g.log.error.assert_called_with(expected_error) def test_bulk_delete_tracks_disassociate_video_failed( mocker, test_product_id, logic_headers, tracks_delete_response, mock_app): """Test bulk delete when disassociating video tracks failed.""" mocker.patch.object( TrackPersister, 'delete_many_by_tuids', return_value=tracks_delete_response) mocks.ows_product_delete_localizations_for_tracks(mocker, [1, 2]) mocks.ows_product_verify_product_ownership(mocker) mocker.patch.object( ows_lyrics, 'delete_tracks_lyrics', return_value=response.Response()) mocker.patch.object( ows_video, 'disassociate_tracks', return_value=response.create_fatal_response('Server error')) mocker.patch.object( track_logic.performer_logic, 'bulk_delete_performers_by_tuids') mocker.patch.object( sample_logic, 'bulk_delete_samples_by_tuids', return_value=response.Response()) mocker.patch.object( track_logic.audio_attribute_logic, 'bulk_delete_track_audio_attributes') mocker.patch.object( track_logic.rights_attribute_logic, 'bulk_delete_track_rights_attributes') tuids = [1, 2] with mock_app: mock_g = mocker.patch('backend.logic.track.g') delete_response = track_logic.bulk_delete( tuids, product_id=test_product_id, **logic_headers) assert delete_response assert delete_response.message == tracks_delete_response.message assert mock_g.log.error.called expected_error = \ 'Failed to disassociate tracks from video products: Server error' mock_g.log.error.assert_called_with(expected_error) def test_bulk_delete_tracks_product_ownership_failed( mocker, test_product_id, logic_headers): """Test bulk delete when product ownership validation failed.""" tuids = [1, 2] mocks.ows_product_verify_product_ownership(mocker, owned_product_ids=[]) delete_response = track_logic.bulk_delete( tuids, product_id=test_product_id, **logic_headers) assert not delete_response assert delete_response.errors['code'] == error.OWS_PRODUCT_ERROR_CODE def test_bulk_delete_tracks_asset_feature_enabled( mocker, logic_headers, test_product_id, tracks_delete_response): """Test bulk delete with bulk_delete_assets enabled. When the feature is enabled, a different function should be called """ tuids = [1, 2] src_mock = MagicMock() src_mock.delete_many_by_tuids.configure_mock( return_value=tracks_delete_response) mocker.patch.object( TrackPersister, 'delete_many_by_tuids', new=src_mock.delete_many_by_tuids) mocks.ows_product_delete_localizations_for_tracks(mocker, tuids) mocks.ows_product_verify_product_ownership(mocker) mocker.patch.object( ows_lyrics, 'delete_tracks_lyrics', return_value=response.Response()) mocker.patch.object( ows_video, 'disassociate_tracks', return_value=response.Response()) mocker.patch.object( track_logic.performer_logic, 'bulk_delete_performers_by_tuids') mocker.patch.object( sample_logic, 'bulk_delete_samples_by_tuids', return_value=response.Response()) mocker.patch.object( track_logic.audio_attribute_logic, 'bulk_delete_track_audio_attributes') mocker.patch.object( track_logic.rights_attribute_logic, 'bulk_delete_track_rights_attributes') delete_response = track_logic.bulk_delete( tuids, product_id=test_product_id, **logic_headers) # Order of expected calls for functions of interest expected_calls = [ call.delete_many_by_tuids(product_id=test_product_id, tuids=tuids)] assert delete_response assert delete_response.message == tracks_delete_response.message assert src_mock.mock_calls == expected_calls def test_bulk_delete_tracks_complex_track_feature_enabled( mocker, logic_headers, test_product_id, tracks_delete_response): """Test bulk delete with bulk_delete_assets enabled. When the feature is enabled, a different function should be called """ tuids = [1, 2] src_mock = MagicMock() src_mock.delete_many_by_tuids.configure_mock( return_value=tracks_delete_response) mocker.patch.object( TrackPersister, 'delete_many_by_tuids', new=src_mock.delete_many_by_tuids) mocker.patch.object( sample_logic, 'bulk_delete_samples_by_tuids', return_value=True) mocks.ows_product_delete_localizations_for_tracks(mocker, tuids) mocks.ows_product_verify_product_ownership(mocker) mocker.patch.object( ows_lyrics, 'delete_tracks_lyrics', return_value=response.Response()) mocker.patch.object( ows_video, 'disassociate_tracks', return_value=response.Response()) mocker.patch.object( track_logic.performer_logic, 'bulk_delete_performers_by_tuids') mocker.patch.object( track_logic.audio_attribute_logic, 'bulk_delete_track_audio_attributes') mocker.patch.object( track_logic.rights_attribute_logic, 'bulk_delete_track_rights_attributes') delete_response = track_logic.bulk_delete( tuids, product_id=test_product_id, **logic_headers) assert delete_response assert delete_response.message == tracks_delete_response.message sample_logic.bulk_delete_samples_by_tuids.assert_called_once_with(tuids) def test_validate_tracks_without_chinese_localization( mocker, logic_headers, logic_user, test_track_model_valid_response, test_product_id, mock_app ): """Should return error because complementary language not localized.""" track = test_track_model_valid_response track['meta_language_code'] = 'CHI' track['localizations'] = [] mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_chinese_product_by_id( mocker, product_id=test_product_id) mocks.track_persister.get_all_by_product_id(mocker, [track]) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) localization_error = { 'tuid': track[tf.TUID], 'localizations': [{ 'validator': 'required', 'validator_value': True, 'message': error.VALIDATION_ERROR_CHINESE_LOCALIZATION_REQUIRED.format( track_language='Chinese Traditional', product_meta_language='Chinese Simplified'), 'error_code': 'missing_chinese_localization' }] } mocker.patch.object( track_logic, 'post_submission_validation', ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) expected_response = { 'errors': [localization_error], 'total_tracks': 1, 'valid_tracks': 0, 'valid': False } validation_response = track_logic.validate_tracks_for_product( product_id=1, **ChainMap(logic_headers, logic_user) ) assert validation_response.message == expected_response def test_validate_tracks_for_corrected_classical_product( classical_genre_id, classical_subgenre_id, logic_headers, logic_user, mocker, track_factory, test_product_id, mock_app ): """Verify corrected tracks use correction data for validation. Ensure that the composer provided correction data is used for validaiton. """ tracks = [ track_factory(performer__count=1, composer__count=1, writers__count=1), track_factory(performer__count=1, writers__count=1)] subgenres = [ {'orchard_id': classical_subgenre_id, 'name': '20/21st Century', 'genre_id': classical_genre_id} ] corrections = { 'items': [{ 'field_name': 'track_artist', 'key_id': tracks[1].tuid, 'key_value': [{ 'name': 'composer dude', 'type': 'composer' }], 'table_name': 'track' }] } mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, genre_id=classical_genre_id, subgenre_id=classical_subgenre_id, product_id=test_product_id, release_status=IS_ERR_CORR_MODE) mocks.track_persister.get_all_by_product_id(mocker, tracks) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocker.patch.object( track_logic, 'post_submission_validation', ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocks.subgenre_query.get_genre_subgenres(mocker, subgenres) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user) ) assert validation_response.message['total_tracks'] == len(tracks) assert validation_response.message['valid_tracks'] == 1 validation_error = validation_response.message['errors'][0] assert 'composer' not in validation_error def test_validate_tracks_for_corrected_product_with_track_writer( logic_headers, logic_user, mocker, track_factory, test_product_id, mock_app ): """Verify corrected tracks use correction data for validation. Ensure that the track writer provided in correction data is used for validaiton. """ tracks = [ track_factory(performer__count=1, writers__count=1), track_factory(performer__count=1, writers__count=0)] corrections = { 'items': [{ 'field_name': 'track_writer', 'key_id': tracks[1].tuid, 'key_value': ['Some writer'], 'table_name': 'track' }] } mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, product_id=test_product_id, release_status=IS_ERR_CORR_MODE) mocks.track_persister.get_all_by_product_id(mocker, tracks) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocker.patch.object( track_logic, 'post_submission_validation', ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user) ) assert validation_response.message['total_tracks'] == len(tracks) assert validation_response.message['valid_tracks'] == 2 assert validation_response.message['errors'] == [] def test_make_track_response(track_factory): """Verify standardize track response removes fields.""" track = track_factory( performer__count=3, writers__count=3, publishers__count=3).to_dict() res = response.Response(test_funcs.wrap_api_results([track])) updated_res = track_logic._make_track_response(res) prev_track = res.message['items'][0] updated_track = updated_res.message['items'][0] assert prev_track == track assert prev_track != updated_track pub_fields = ( tf.PUBLISHERS, tf.US_PUBLISHING_OBLIGATION, tf.THIRD_PARTY_PUBLISHER) assert tf.LOCALIZATIONS not in prev_track assert tf.LOCALIZATIONS in updated_track for field in pub_fields: assert field in prev_track assert field not in updated_track def test_validate_tracks_without_product_chinese_localization( mocker, logic_headers, logic_user, test_track_model_valid_response, test_product_id, mock_app ): """Should return error because complementary language not localized.""" track = test_track_model_valid_response track['meta_language_code'] = 'ENG' track['localizations'] = [] mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_chinese_product_by_id( mocker, product_id=test_product_id) mocks.track_persister.get_all_by_product_id(mocker, [track]) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocker.patch.object( track_logic, 'post_submission_validation', ) error_message = \ error.VALIDATION_ERROR_CHINESE_LOCALIZATION_REQUIRED.format( track_language='Chinese Traditional', product_meta_language='Chinese Simplified') localization_error = { 'tuid': track[tf.TUID], 'localizations': [{ 'validator': 'required', 'validator_value': True, 'message': error_message, 'error_code': 'missing_chinese_localization' }] } expected_response = { 'errors': [localization_error], 'total_tracks': 1, 'valid_tracks': 0, 'valid': False } mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=1, **ChainMap(logic_headers, logic_user) ) assert validation_response.message == expected_response def test_update_with_product_meta_language( mocker, mock_app, logic_headers, track_factory, track_localization_factory): """Product meta language and track localization language should be same.""" track = track_factory() localizations = track_localization_factory(track, [12]) update_data = track.to_dict() update_data[tf.TRACK_NAME] = 'Updated track name' update_data[tf.EXPLICIT] = 'Y' mocks.ows_product_verify_product_ownership(mocker) mocks.ows_product_get_track_localizations( mocker, localizations) mocks.ows_product_get_localization_languages(mocker) mocks.track_persister.get_by_tuid(mocker, track) mocks.ows_product_digital_get_product_by_id(mocker) mocks.track_persister.update_track(mocker, update_data) track_response = track_logic.update( tuid=track.tuid, data=deepcopy(update_data), **logic_headers) assert not track_response assert track_response.status == response_code.BAD_REQUEST def test_validate_tracks_for_corrected_product_with_meta_language( logic_headers, logic_user, mocker, track_factory, test_product_id, mock_app ): """Validate the corrected product uses meta language correction data.""" tracks = [ track_factory(performer__count=1, writers__count=1), track_factory(performer__count=1, writers__count=0)] product_id = 1234 corrections = { 'items': [{ 'field_name': 'meta_language', 'key_id': product_id, 'key_value': 'ENG', 'table_name': 'releases' }] } mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, product_id=test_product_id, release_status=IS_ERR_CORR_MODE) mocks.track_persister.get_all_by_product_id(mocker, tracks) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocker.patch.object( track_logic, 'post_submission_validation', ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user) ) assert validation_response.message['valid_tracks'] == 1 def test_validate_tracks_for_corrected_product_with_featuring_artists( logic_headers, logic_user, mocker, track_factory, test_product_id, mock_app ): """Validate the corrected product uses release level featuring artists correction data.""" tracks = [ track_factory(performer__count=1, writers__count=1, featuring__count=1)] product_artists = [{'role': 'featuring', 'name': 'test', 'id': 123, 'artist_info_id': None}] # Takes the track level featuring artist from mocked data track_featuring_artist = tracks[0].artists[1].artist_name # Corrects release level featuring artist to match track level featuring artist corrections = { 'items': [{ 'field_name': 'featuring', 'key_id': test_product_id, 'key_value': [{'artist_name': track_featuring_artist, 'role': 'featuring'}], 'table_name': 'releases' }] } mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, product_id=test_product_id, release_status=IS_ERR_CORR_MODE, product_artists=product_artists) mocks.track_persister.get_all_by_product_id(mocker, tracks) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocker.patch.object( track_logic, 'post_submission_validation', ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids', return_value=response.Response(message={})) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user) ) assert validation_response.message['valid_tracks'] == 1 def test_get_all_tracks_by_type_and_isrc(mocker, track_factory): """Test retrieving tracks by type and isrc.""" example_tracks = [ track_factory(), track_factory(), track_factory() ] mocks.track_persister.get_all_by_isrc_and_type(mocker, example_tracks) response = track_logic.get_all_tracks_by_type_and_isrc(tf.TRACK_TYPE_MUSIC, example_tracks[0].isrc) assert response.status == response_code.OK assert len(response.message['items']) == len(example_tracks) assert response.message['pagination'] == {'type': 'none', 'total_records': len(example_tracks)} def test_get_account_tracks_by_type_and_isrc(mocker, track_factory): """Test retrieving tracks by type and isrc, filtered by account.""" example_tracks = [ track_factory(product_id=1), track_factory(product_id=2), track_factory(product_id=3), ] mocks.track_persister.get_all_for_account_by_isrc_and_type(mocker, example_tracks) response = track_logic.get_account_tracks_by_type_and_isrc( 'vendor', 11111, tf.TRACK_TYPE_MUSIC, example_tracks[0].isrc, ) assert response.status == response_code.OK assert len(response.message['items']) == 3 assert response.message['pagination'] == {'type': 'none', 'total_records': 3} def test_bulk_create_tracks_with_metadata( mocker, track_bulk_create_track_with_metadata_data, track_bulk_create_track_with_metadata_response, logic_headers): """Verify Tracks are created.""" create_and_update_multiple_tracks_data = response.Response( track_bulk_create_track_with_metadata_response) product_id = track_bulk_create_track_with_metadata_data[ 'tracks'][0]['product_id'] upc = '123456789012' mocks.track_persister.create_and_update_multiple_tracks( mocker, create_and_update_multiple_tracks_data) mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=product_id, upc=upc) track_response = track_logic.bulk_create_track_with_metadata( product_id, track_bulk_create_track_with_metadata_data, **logic_headers) assert track_response.message ==\ track_bulk_create_track_with_metadata_response def test_bulk_create_tracks_with_metadata_fail_on_create( mocker, track_bulk_create_track_with_metadata_data, logic_headers): """Verify Tracks are failed on create.""" product_id = track_bulk_create_track_with_metadata_data[ 'tracks'][0]['product_id'] upc = '123456789012' failed_create_response = response.create_error_response( code='code', message='Error') mocks.track_persister.create_and_update_multiple_tracks( mocker, failed_create_response) mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=product_id, upc=upc) track_response = track_logic.bulk_create_track_with_metadata( product_id, track_bulk_create_track_with_metadata_data, **logic_headers) assert track_response == failed_create_response def test_bulk_create_tracks_with_metadata_fail_on_ownership( mocker, track_bulk_create_track_with_metadata_data, track_bulk_create_track_with_metadata_response, logic_headers): """Verify Tracks are created.""" create_and_update_multiple_tracks_data = response.Response( track_bulk_create_track_with_metadata_response) product_id = track_bulk_create_track_with_metadata_data[ 'tracks'][0]['product_id'] mocks.track_persister.create_and_update_multiple_tracks( mocker, create_and_update_multiple_tracks_data) not_owning_response = response.create_error_response(403, 'not owning') mocker.patch( 'backend.utils.logic.get_product_data_and_validate_ownership', return_value=not_owning_response) track_response = track_logic.bulk_create_track_with_metadata( product_id, track_bulk_create_track_with_metadata_data, **logic_headers) assert not track_response def test_set_product_tracks_create( mocker, track_bulk_create_track_with_metadata_data, test_track_response, logic_headers): """Verify Tracks are created.""" empty_mock_track_response = response.Response({'items': []}) mock_track_response = response.Response({'items': [test_track_response]}) product_id = track_bulk_create_track_with_metadata_data[ 'tracks'][0]['product_id'] upc = '123456789012' payload_tracks = track_bulk_create_track_with_metadata_data['tracks'] payload_tracks[0]['isrc'] = 'ABC' payload_tracks[0]['volume_number'] = 1 payload_tracks[0]['track_number'] = 2 payload_tracks[1]['isrc'] = 'ABC' payload_tracks[1]['volume_number'] = 1 payload_tracks[1]['track_number'] = 2 create_data = [ { 'track_name': payload_tracks[0]['track_name'], 'volume_number': payload_tracks[0]['volume_number'], }, { 'track_name': payload_tracks[1]['track_name'], 'volume_number': payload_tracks[1]['volume_number'], }, ] update_data = [ {**payload_tracks[0], 'tuid': test_track_response['tuid']}, {**payload_tracks[1], 'tuid': test_track_response['tuid']} ] mock_get = mocks.track_persister.get_all_by_product_id(mocker, []) mock_get.side_effect = [empty_mock_track_response, mock_track_response] mock_create = mocks.track_persister.bulk_create_tracks( mocker, mock_track_response) mock_update = mocks.track_persister.update_multiple_tracks( mocker, mock_track_response) mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=product_id, upc=upc) result = track_logic.set_product_tracks( product_id, payload_tracks, **logic_headers) assert result.message['items'] == [test_track_response] mock_create.assert_called_with(product_id, upc, create_data, for_volume=1) mock_update.assert_called_with(update_data, update_track_volume_numbers=True) def test_set_product_tracks_update( mocker, track_bulk_create_track_with_metadata_data, test_track_response, logic_headers): """Verify Tracks are updated.""" mock_track_response = response.Response({'items': [test_track_response]}) product_id = track_bulk_create_track_with_metadata_data[ 'tracks'][0]['product_id'] upc = '123456789012' payload_tracks = track_bulk_create_track_with_metadata_data['tracks'] payload_tracks[0]['isrc'] = 'ABC' payload_tracks[0]['volume_number'] = 1 payload_tracks[0]['track_number'] = 2 payload_tracks[1]['isrc'] = 'ABC' payload_tracks[1]['volume_number'] = 1 payload_tracks[1]['track_number'] = 2 update_data = [ {**payload_tracks[0], 'tuid': test_track_response['tuid']}, {**payload_tracks[1], 'tuid': test_track_response['tuid']} ] mocks.track_persister.get_all_by_product_id( mocker, mock_track_response) mock_create = mocks.track_persister.bulk_create_tracks( mocker, mock_track_response) mock_update = mocks.track_persister.update_multiple_tracks( mocker, mock_track_response) mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=product_id, upc=upc) result = track_logic.set_product_tracks( product_id, payload_tracks, **logic_headers) assert result.message['items'] == [test_track_response] mock_create.assert_not_called() mock_update.assert_called_with(update_data, update_track_volume_numbers=True) def test_set_product_tracks_delete( mocker, track_bulk_create_track_with_metadata_data, test_track_response, logic_headers): """Verify Tracks are deleted.""" mock_track_response = response.Response({'items': [test_track_response]}) product_id = track_bulk_create_track_with_metadata_data[ 'tracks'][0]['product_id'] upc = '123456789012' mocks.track_persister.get_all_by_product_id(mocker, mock_track_response) mock_create = mocks.track_persister.bulk_create_tracks( mocker, mock_track_response) mock_update = mocks.track_persister.update_multiple_tracks( mocker, mock_track_response) mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=product_id, upc=upc) result = track_logic.set_product_tracks( product_id, [], **logic_headers) assert result.message['items'] == [] mock_create.assert_not_called() mock_update.assert_not_called() def test_get_track_by_track_artist_id(mocker, track_factory): """Test retrieving track by track_artist_id.""" track = track_factory(performer__count=3) mocks.track_persister.get_track_by_track_artist_id(mocker, track) response = track_logic.get_track_by_track_artist_id(1234) assert response.status == response_code.OK assert response.message == track.to_dict() def test_bulk_create_for_explicit( mocker, logic_headers, track_bulk_create_data, track_bulk_create_new_prod_response_for_explicit, track_bulk_create_new_prod_logic_response_for_explicit): """Verify Track is created, this is to test for attribute "explicit".""" mocks.track_persister.bulk_create_tracks( mocker, track_bulk_create_new_prod_response_for_explicit) first_track = track_bulk_create_new_prod_response_for_explicit['items'][0] product_id = first_track[tf.PRODUCT_ID] upc = first_track[tf.UPC] mocks.head_response(mocker) mocks.ows_product_get_product_by_id( mocker, product_id=product_id, upc=upc) track_response = track_logic.bulk_create( product_id, track_bulk_create_data, for_volume=None, **logic_headers) assert track_response.message == \ track_bulk_create_new_prod_logic_response_for_explicit def test_claim_new_isrc_success( mocker): """Verify success response from TrackPersister.claim_new_isrc.""" isrc_mock = mocker.patch.object( TrackPersister, 'claim_new_isrc', return_value=MagicMock( status=response_code.OK, message='US1234567890')) response = track_logic.claim_new_isrc() assert isrc_mock.called assert response.status == response_code.OK assert response.message == 'US1234567890' def test_claim_new_isrc_failure( mocker): """Verify error response if no ISRC is claimed.""" isrc_mock = mocker.patch.object( TrackPersister, 'claim_new_isrc', return_value=MagicMock( status=response_code.INTERNAL_ERROR, message='Unable to claim an ISRC.')) response = track_logic.claim_new_isrc() assert isrc_mock.called assert response.status == response_code.INTERNAL_ERROR assert response.message == 'Unable to claim an ISRC.' @pytest.mark.parametrize( ( 'test_description', 'validation_context', 'tracks', 'expected_get_isrc_usage_stats_calls', 'expected_result', ), [ ( 'post_submission context <= max reuse threshold of 5', 'post_submission', [ { 'id': 12321, 'isrc_usage': { 'other': { 'stuff': 'here' }, 'count': 3, }, } ], [call(12321)], [], ), ( 'post_submission context > max reuse threshold of 5', 'post_submission', [ { 'id': 12321, 'isrc_usage': { 'other': { 'stuff': 'here' }, 'count': 6, }, }, { 'id': 32123, 'isrc_usage': { 'other': { 'stuff': 'here' }, 'count': 6, }, } ], [call(12321)], [ { 'isrc': { 'validator': 'max_isrc_reuse', 'validator_value': 5, 'message': '{"id": 12321, "isrc_usage": {"other": {"stuff": "here"}, "count": 6}}', 'code': 'excessive_isrc_reuse', }, 'tuid': 12321, }, { 'isrc': { 'validator': 'max_isrc_reuse', 'validator_value': 5, 'message': '{"id": 32123, "isrc_usage": {"other": {"stuff": "here"}, "count": 6}}', 'code': 'excessive_isrc_reuse', }, 'tuid': 32123, } ], ), ] ) def test__post_submission_validation_isrc_usage( mocker, test_description, validation_context, tracks, expected_get_isrc_usage_stats_calls, expected_result, ): """Test _post_submission_validation_isrc_usage.""" mocker.patch.object(isrc_logic, 'get_isrc_usage_stats', return_value={ 'product': { 'tracks': tracks } }) result = track_logic._post_submission_validation_isrc_usage(12321) assert isrc_logic.get_isrc_usage_stats.mock_calls == ( expected_get_isrc_usage_stats_calls) assert result == expected_result def test__post_submission_validation_potential_audio_infringement(mocker): """Test _post_submission_validation_potential_audio_infringement.""" tracks = [{'tuid': 11, 'artists': []}, {'tuid': 22, 'artists': []}, {'tuid': 33, 'artists': []}] mocker.patch.object( ows_assets, 'get_match_audio_results', return_value=[ { 'asset_final_id': 1, 'code': 'WARNING_MATCHES_FOUND', 'matches': [{'artists': ['Focus Sounds'], 'title': 'PN1', 'label': 'SME' }], 'track_id': 11 }, { 'asset_final_id': 2, 'code': 'SOME_OTHER_CODE', 'track_id': 22 }, { 'asset_final_id': 3, 'code': 'WARNING_MATCHES_FOUND', 'matches': [{'artists': ['Focus Sounds'], 'title': 'PN1', 'label': 'SME' }], 'track_id': 33 }, ] ) result = track_logic._post_submission_validation_potential_audio_infringement(12321, tracks, '') assert result == [ { 'audio_file': { 'code': 'potential_audio_infringement', 'message': '{"asset_final_id": 1, "matches": [{"artists": ["Focus Sounds"], "title": "PN1", "label": "SME"}]}', 'validator': 'potential_audio_infringement', 'validator_value': True }, 'tuid': 11 }, { 'audio_file': { 'code': 'potential_audio_infringement', 'message': '{"asset_final_id": 3, "matches": [{"artists": ["Focus Sounds"], "title": "PN1", "label": "SME"}]}', 'validator': 'potential_audio_infringement', 'validator_value': True }, 'tuid': 33 } ] def test__post_submission_validation_potential_audio_infringement_with_self_matches(mocker): """Test _post_submission_validation_potential_audio_infringement.""" tracks = [ {'tuid': 11, 'artists': [{'name':'Focus Sounds', 'type': 'performer'}],'track_name': 'PN1'}, {'tuid': 22, 'artists': []}, {'tuid': 33, 'artists': [{'name':'Focus Sounds 2', 'type': 'performer'}],'track_name': 'PN2'} ] mocker.patch.object( ows_assets, 'get_match_audio_results', return_value=[ { 'asset_final_id': 1, 'code': 'WARNING_MATCHES_FOUND', 'matches': [{'artists': ['Focus Sounds'], 'title': 'PN1', 'label': 'SME' }], # matches track, remove 'track_id': 11 }, { 'asset_final_id': 2, 'code': 'SOME_OTHER_CODE', 'track_id': 22 }, { 'asset_final_id': 3, 'code': 'WARNING_MATCHES_FOUND', 'matches': [{'artists': ['Focus Sounds 2'], 'title': 'PN2', 'label': 'Some label' }, # noqa: E501, does not match track, keep {'artists': ['Focus Sounds 2'], 'title': 'PN2', 'label': 'SME' }], # matches track, remove 'track_id': 33 }, ] ) result = track_logic._post_submission_validation_potential_audio_infringement(12321, tracks, 'SME') assert result == [ { 'audio_file': { 'code': 'potential_audio_infringement', 'message': '{"asset_final_id": 3, "matches": [{"artists": ["Focus Sounds 2"], "title": "PN2", "label": "Some label"}]}', # noqa: E501 'validator': 'potential_audio_infringement', 'validator_value': True }, 'tuid': 33 } ] @pytest.mark.parametrize( ('as_error', 'code_key'), [ (False, 'code'), (True, 'error_code'), ] ) def test__post_submission_validation_ai_generated_audio_suspected(mocker, as_error, code_key): """Test _post_submission_validation_ai_generated_audio_suspected. As a warning (default) the inner payload is keyed by ``code``; as an error (``as_error=True``) it is keyed by ``error_code``. """ mocker.patch.object( ows_assets, 'get_ai_generated_audio_results', return_value=[ { 'code': 'AI_GENERATED_AUDIO_SUSPECTED', 'asset_final_id': 1, 'track_id': 11 }, { 'code': 'PASS_NO_AI_GENERATED_AUDIO_SUSPECTED', 'asset_final_id': 2, 'track_id': 22 }, { 'code': 'AI_GENERATED_AUDIO_SUSPECTED', 'asset_final_id': 3, 'track_id': 33 }, ] ) result = track_logic._post_submission_validation_ai_generated_audio_suspected( 12321, as_error=as_error) assert result == [ { 'ai_generated_audio': { code_key: 'ai_generated_audio', 'validator': 'ai_generated_audio', 'message': '{"asset_final_id": 1}', 'validator_value': True }, 'tuid': 11 }, { 'ai_generated_audio': { code_key: 'ai_generated_audio', 'validator': 'ai_generated_audio', 'message': '{"asset_final_id": 3}', 'validator_value': True }, 'tuid': 33 } ] @pytest.mark.parametrize( ( 'test_description', 'validation_context', 'expected__post_submission_validation_isrc_usage_calls', '_post_submission_validation_isrc_usage_result', 'expected__post_submission_validation_potential_audio_infringement_calls', '_post_submission_validation_potential_audio_infringement_result', 'ai_generated_audio_error_enabled', 'ai_generated_audio_validation_enabled', 'expected__post_submission_validation_ai_generated_audio_suspected_calls', '_post_submission_validation_ai_generated_audio_suspected_result', 'spotify_watchlist_artist_validation_enabled', 'expected__post_submission_validation_spotify_watchlist_artists_calls', '_post_submission_validation_spotify_watchlist_artists_result', 'expected_warnings', 'expected_errors', ), [ ( 'pre_submission context bypasses validation', 'pre_submission', [], None, [], None, False, False, [], None, False, [], None, ['some existing warning'], [], ), ( 'AI error flag on: suspected AI audio is a blocker merged into errors by track_id.', # noqa 'post_submission', [call(12321)], [ {'tuid': 12321, 'isrc': 'excessive_isrc_reuse warning'}, ], [call(12321, [], '')], [ {'tuid': 12321, 'audio_file': 'potential_audio_infringement warning'}, {'tuid': 32123, 'audio_file': 'potential_audio_infringement warning'}, ], True, False, [call(12321, as_error=True)], [ {'tuid': 12321, 'ai_generated_audio': 'ai_generated_audio error'}, {'tuid': 54321, 'ai_generated_audio': 'ai_generated_audio error'}, ], False, [], [], [ 'some existing warning', { 'isrc': 'excessive_isrc_reuse warning', 'audio_file': 'potential_audio_infringement warning', 'tuid': 12321 }, { 'audio_file': 'potential_audio_infringement warning', 'tuid': 32123 }, ], [ {'tuid': 12321, 'ai_generated_audio': 'ai_generated_audio error'}, {'tuid': 54321, 'ai_generated_audio': 'ai_generated_audio error'}, ], ), ( 'AI validation flag on (error off): suspected AI audio is a warning grouped by track_id.', # noqa 'post_submission', [call(12321)], [ {'tuid': 12321, 'isrc': 'excessive_isrc_reuse warning'}, ], [call(12321, [], '')], [ {'tuid': 12321, 'audio_file': 'potential_audio_infringement warning'}, {'tuid': 32123, 'audio_file': 'potential_audio_infringement warning'}, ], False, True, [call(12321)], [ {'tuid': 12321, 'ai_generated_audio': 'ai_generated_audio warning'}, {'tuid': 54321, 'ai_generated_audio': 'ai_generated_audio warning'}, ], False, [], [], [ 'some existing warning', { 'isrc': 'excessive_isrc_reuse warning', 'ai_generated_audio': 'ai_generated_audio warning', 'audio_file': 'potential_audio_infringement warning', 'tuid': 12321 }, { 'audio_file': 'potential_audio_infringement warning', 'tuid': 32123 }, { 'ai_generated_audio': 'ai_generated_audio warning', 'tuid': 54321 } ], [], ), ( 'Both AI flags on: error flag takes precedence and emits a blocker only.', 'post_submission', [call(12321)], [], [call(12321, [], '')], [], True, True, [call(12321, as_error=True)], [ {'tuid': 99999, 'ai_generated_audio': 'ai_generated_audio error'}, ], False, [], [], [ 'some existing warning', ], [ {'tuid': 99999, 'ai_generated_audio': 'ai_generated_audio error'}, ], ), ( 'Both AI flags disabled, function not called', 'post_submission', [call(12321)], [], [call(12321, [], '')], [], False, False, [], [], False, [], [], ['some existing warning'], [], ), ( 'Spotify watchlist validation enabled and returns warnings', 'post_submission', [call(12321)], [], [call(12321, [], '')], [], False, False, [], [], True, [call([])], [ {'tuid': 11111, 'spotify_watchlist_artist': 'spotify_watchlist_artist warning'}, ], [ 'some existing warning', { 'spotify_watchlist_artist': 'spotify_watchlist_artist warning', 'tuid': 11111 } ], [], ), ] ) def test_post_submission_validation( mocker, test_description, validation_context, expected__post_submission_validation_isrc_usage_calls, _post_submission_validation_isrc_usage_result, expected__post_submission_validation_potential_audio_infringement_calls, _post_submission_validation_potential_audio_infringement_result, ai_generated_audio_error_enabled, ai_generated_audio_validation_enabled, expected__post_submission_validation_ai_generated_audio_suspected_calls, _post_submission_validation_ai_generated_audio_suspected_result, spotify_watchlist_artist_validation_enabled, expected__post_submission_validation_spotify_watchlist_artists_calls, _post_submission_validation_spotify_watchlist_artists_result, expected_warnings, expected_errors ): """Test post_submission_validation.""" mocker.patch.object( track_logic, '_post_submission_validation_isrc_usage', return_value=_post_submission_validation_isrc_usage_result ) mocker.patch.object( track_logic, '_post_submission_validation_potential_audio_infringement', return_value=_post_submission_validation_potential_audio_infringement_result ) mocker.patch.object( track_logic, '_post_submission_validation_ai_generated_audio_suspected', return_value=_post_submission_validation_ai_generated_audio_suspected_result ) mocker.patch.object( track_logic, '_post_submission_validation_spotify_watchlist_artists', return_value=_post_submission_validation_spotify_watchlist_artists_result ) mocker.patch.object( features, 'is_ai_generated_audio_error_enabled', return_value=ai_generated_audio_error_enabled ) mocker.patch.object( features, 'is_ai_generated_audio_validation_enabled', return_value=ai_generated_audio_validation_enabled ) mocker.patch.object( features, 'is_spotify_watchlist_artist_validation_enabled', return_value=spotify_watchlist_artist_validation_enabled ) mocker.patch.object( ows_sound_recordings, 'get_formatted_sound_recording_matches', return_value={}) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response(message='some company brand')) warnings = ['some existing warning'] errors = [] track_logic.post_submission_validation(validation_context, 12321, 100, errors, warnings, [], '') assert track_logic._post_submission_validation_isrc_usage.mock_calls == ( expected__post_submission_validation_isrc_usage_calls) assert track_logic._post_submission_validation_potential_audio_infringement.mock_calls == ( expected__post_submission_validation_potential_audio_infringement_calls) assert track_logic._post_submission_validation_ai_generated_audio_suspected.mock_calls == ( expected__post_submission_validation_ai_generated_audio_suspected_calls) assert track_logic._post_submission_validation_spotify_watchlist_artists.mock_calls == ( expected__post_submission_validation_spotify_watchlist_artists_calls) assert warnings == expected_warnings assert errors == expected_errors def test__merge_post_submission_track_errors_merges_by_tuid(): """AI errors merge into an existing track error entry; new tracks are appended.""" errors = [ {'isrc': 'invalid isrc error', 'tuid': 11}, {'track_name': 'missing track name error', 'tuid': 22}, ] post_submission_errors = [ {'ai_generated_audio': 'ai_generated_audio error', 'tuid': 11}, {'ai_generated_audio': 'ai_generated_audio error', 'tuid': 33}, ] track_logic._merge_post_submission_track_errors(errors, post_submission_errors) assert errors == [ { 'isrc': 'invalid isrc error', 'ai_generated_audio': 'ai_generated_audio error', 'tuid': 11, }, {'track_name': 'missing track name error', 'tuid': 22}, {'ai_generated_audio': 'ai_generated_audio error', 'tuid': 33}, ] @pytest.mark.parametrize( ( 'test_description', 'validate_artists_response', 'expected_result', ), [ ( 'No watchlist matches - returns empty list', {}, [], ), ( 'Watchlist match found - returns formatted warning per matched tuid', { 'validation_errors': { 'track_artists': [ { 'tuid': 2, 'matches': [ { 'name': 'Quadron', 'type': 'performer', 'reason': 'Spotify Watchlist Artist' } ] } ] } }, [ { 'spotify_watchlist_artist': { 'validator': 'spotify_watchlist_artist', 'validator_value': True, 'message': json.dumps([ { "name": "Quadron", "type": "performer", "reason": "Spotify Watchlist Artist" } ]), 'code': 'spotify_watchlist_artist', }, 'tuid': 2, } ], ), ] ) def test__post_submission_validation_spotify_watchlist_artists( mocker, test_description, validate_artists_response, expected_result, ): """Test _post_submission_validation_spotify_watchlist_artists.""" tracks = [ {'tuid': 1, 'artists': [{'name': 'Some Artist', 'type': 'performer'}]}, {'tuid': 2, 'artists': [{'name': 'Quadron', 'type': 'performer'}]}, ] mocker.patch.object( ows_blacklist_manager, 'validate_artists', return_value=validate_artists_response ) result = track_logic._post_submission_validation_spotify_watchlist_artists(tracks) assert result == expected_result def test__post_submission_validation_spotify_watchlist_artists_with_corrections(mocker): """Test that track corrections in a ChainMap are used instead of original artists. When a track has a correction for a specific artist type (e.g. 'remixer'), the corrected artists should be sent to the blacklist manager, not the originals. Non-corrected artist types should fall through to the original artists list. """ from collections import ChainMap original_track = { 'tuid': 1, 'artists': [ {'name': 'Original Performer', 'type': 'performer'}, {'name': 'Original Remixer', 'type': 'remixer'}, ], } correction = {'remixer': [{'name': 'Corrected Remixer', 'type': 'remixer'}]} track_with_correction = ChainMap(correction, original_track) validate_artists_mock = mocker.patch.object( ows_blacklist_manager, 'validate_artists', return_value={}, ) track_logic._post_submission_validation_spotify_watchlist_artists([track_with_correction]) validate_artists_mock.assert_called_once_with({ 'track_artists': [ { 'tuid': 1, 'artists': [ {'name': 'Corrected Remixer', 'type': 'remixer'}, {'name': 'Original Performer', 'type': 'performer'}, ], } ] }) def test_check_profile_track_access( mock_app, mocker, track_factory, request_engine ): """Test check_profile_track_access.""" test_track = track_factory() mocks.track_persister.get_by_tuid(mocker, test_track) product_id = test_track.product_id request_engine['ows-product'].add_spec( 'GET', f'/product/{product_id}', {'vendor_id': 123}, status=200) request_engine['ows-permissions'].add_spec( 'HEAD', '/profile/uuid/abc/resource/label/id/123', status=200) result = track_logic.check_profile_track_access('abc', test_track.tuid) assert result.status == 200