"""Tests for Track Logic.""" from oto import response from oto import status as response_code from backend.constants import error from backend.constants import track_field as tf from backend.logic import track_role as track_role_logic from backend.models import ows_product_digital from backend.models import ows_product_workflow from backend.models.track_persister import TrackPersister from backend.utils import api as api_utils from backend.utils import localization as loc_util from tests.testutils import mocks from tests.testutils.functions import make_localization_dict from tests.testutils.functions import wrap_api_results def _mock_ownership_and_get_track(mocker, tuid, product_id): """Mock ownership response and TrackPersister.get_by_tuid.""" mocks.head_response(mocker) mocks.track_persister.get_by_tuid( mocker, {tf.TUID: tuid, tf.PRODUCT_ID: product_id}) def test_create_artist(mocker, test_track, logic_headers): """Verify track artist is added.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) mocker.patch.object( TrackPersister, 'add_track_artist', return_value=response.Response( message={}, status=response_code.CREATED)) track_response = track_role_logic.create( tuid=test_tuid, role='performer', data={'name': 'blah'}, **logic_headers) assert track_response.status == response_code.CREATED def test_create_writer(mocker, test_track, logic_headers): """Verify track writer is added.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) mocker.patch.object( TrackPersister, 'add_track_writer', return_value=response.Response( message={}, status=response_code.CREATED)) track_response = track_role_logic.create( tuid=test_tuid, role='writer', data={'name': 'blah'}, **logic_headers) assert track_response.status == response_code.CREATED def test_create_invalid_no_name( mocker, test_track, logic_headers): """Verify error when name isn't passed.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) track_response = track_role_logic.create( tuid=test_tuid, role='performer', data={}, **logic_headers) assert track_response.status == response_code.BAD_REQUEST track_response = track_role_logic.create( tuid=test_tuid, role='performer', data={'name': ''}, **logic_headers) assert track_response.status == response_code.BAD_REQUEST def test_create_invalid_role( mocker, test_track, logic_headers): """Verify error when invalid role is passed.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) track_response = track_role_logic.create( tuid=test_tuid, role='publisher', data={'name': 'name'}, **logic_headers) assert track_response.status == response_code.BAD_REQUEST assert track_response.errors['code'] == error.INVALID_ROLE_ERROR_CODE def test_delete_artist(mocker, test_track, logic_headers): """Verify artist is deleted.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) mocker.patch.object( TrackPersister, 'remove_track_artist', return_value=response.Response( message={}, status=response_code.OK)) track_response = track_role_logic.delete( tuid=test_tuid, role='performer', track_role_id=1, **logic_headers) assert track_response.status == response_code.OK def test_delete_writer(mocker, test_track, logic_headers): """Verify writer is deleted.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) mocker.patch.object( TrackPersister, 'remove_track_writer', return_value=response.Response( message={}, status=response_code.OK)) track_response = track_role_logic.delete( tuid=test_tuid, role='writer', track_role_id=1, **logic_headers) assert track_response.status == response_code.OK def test_delete_invalid_role( mocker, test_track, logic_headers): """Verify error when invalid role is passed.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) track_response = track_role_logic.delete( tuid=test_tuid, role='publisher', track_role_id=1, **logic_headers) assert track_response.status == response_code.BAD_REQUEST assert track_response.errors['code'] == error.INVALID_ROLE_ERROR_CODE def test_update_remixer_for_all_tracks_in_product( mocker, logic_headers, product_role_apply_to_all_data): """Verify data is synced for artist role.""" expected_msg = { 'items': [{ tf.TUID: 1, tf.META_LANGUAGE_CODE: 'ENG', tf.ARTISTS: product_role_apply_to_all_data['names'] }] } mocker.patch.object( TrackPersister, 'sync_role_for_all_tracks_in_product', return_value=response.Response( message=expected_msg, status=response_code.OK)) mocks.head_response(mocker) sync_response = track_role_logic.update_for_all_tracks_in_product( product_id=1, role_type='remixer', data=product_role_apply_to_all_data, **logic_headers) assert sync_response.status == response_code.OK assert sync_response.message == expected_msg def test_update_for_all_tracks_in_product_with_localization( mocker, logic_headers, track_factory): """Verify artist localizations are applied.""" LANG_ID = 2 NEW_ARTIST_NAME = 'Hoofy' NEW_ARTIST_LOCALIZED_NAME = 'Doofy' # Generate track input data track = track_factory(performer__count=1) artist = track.artists[0] artist.name = NEW_ARTIST_NAME track_dict = track.to_dict( filter_by_fields=[tf.META_LANGUAGE_CODE, tf.ARTISTS]) # Mock Track Persister mocker.patch.object( TrackPersister, 'sync_role_for_all_tracks_in_product', return_value=response.Response( message=wrap_api_results([track_dict]), status=response_code.OK)) # Transformation Data data = { 'names': [NEW_ARTIST_NAME], tf.LOCALIZATIONS: [{ tf.LANGUAGE_ID: LANG_ID, 'names': [NEW_ARTIST_LOCALIZED_NAME]}] } # Expected Output Data artist.name = NEW_ARTIST_LOCALIZED_NAME expected_localization = make_localization_dict( track.tuid, LANG_ID, artists=[artist.to_dict()]) expected_track_delta = track_dict.copy() expected_track_delta[tf.LOCALIZATIONS] = [expected_localization.copy()] del expected_track_delta[tf.META_LANGUAGE_CODE] # Mock OWS Product mocks.head_response(mocker) mocks.put_response(mocker) # use blank data mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker, wrap_api_results([])) spy = mocker.spy(loc_util, 'patch_tracks_localizations') sync_response = track_role_logic.update_for_all_tracks_in_product( product_id=1, role_type='performer', data=data, **logic_headers) assert sync_response assert spy.call_args[0][1] == [{ tf.LANGUAGE_ID: LANG_ID, 'performer_names': [NEW_ARTIST_LOCALIZED_NAME] }] assert sync_response.message['items'] == [expected_track_delta] def test_update_for_all_tracks_in_product_with_extra_localizations( mocker, track_role_localization_data, logic_headers): """Verify failure on extra localizations.""" data = track_role_localization_data data['names'] = data['names'][:1] mocks.head_response(mocker) sync_response = track_role_logic.update_for_all_tracks_in_product( product_id=1, role_type='performer', data=data, **logic_headers) assert not sync_response assert sync_response.errors['code'] == 'validation_error' def test_update_for_all_tracks_in_product_non_artist_role( mocker, track_role_localization_data, logic_headers): """Verify failure when writer is passed in with localizations.""" data = track_role_localization_data mocks.head_response(mocker) sync_response = track_role_logic.update_for_all_tracks_in_product( product_id=1, role_type='writer', data=data, **logic_headers) assert not sync_response assert sync_response.errors['code'] == 'validation_error' def test_update_artist_for_all_tracks_in_product_with_invalid_name( invalid_artist_name, mocker, logic_headers, product_role_apply_to_all_data): """Verify error is returned for an invalid artist name.""" error_response = api_utils.create_validation_error_response( error.INVALID_ARTIST_NAME_ERROR_MSG.format(invalid_artist_name)) expected_msg = error_response.message mocks.head_response(mocker) sync_response = track_role_logic.update_for_all_tracks_in_product( product_id=1, role_type='artist', data=product_role_apply_to_all_data, **logic_headers) assert not sync_response assert sync_response.message == expected_msg def test_update_track_role(mocker, logic_headers): """Verify track artist is updated.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) mocker.patch.object( TrackPersister, 'update_track_artist', return_value=response.Response( message={}, status=response_code.OK )) track_response = track_role_logic.update( tuid=test_tuid, role='performer', role_id=1, data={'name': 'blah'}, **logic_headers ) assert track_response.status == response_code.OK def test_update_track_role_with_writer(mocker, logic_headers): """Verify track writer is updated.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) mocker.patch.object( TrackPersister, 'update_track_writer', return_value=response.Response( message={}, status=response_code.OK )) track_response = track_role_logic.update( tuid=test_tuid, role='writer', role_id=1, data={'name': 'blah'}, **logic_headers ) assert track_response.status == response_code.OK def test_update_track_artist_invalid_role(mocker, logic_headers): """Verify failure when providing an invalid role.""" test_tuid = 1 _mock_ownership_and_get_track(mocker, tuid=test_tuid, product_id=1) track_response = track_role_logic.update( tuid=test_tuid, role='asdf', role_id=1, data={}, **logic_headers ) assert track_response.status == response_code.BAD_REQUEST def test_update_artist_classical(mocker, logic_headers, test_correlation_id, test_orchard_user_id): """Verify valid artists roles for classical genre.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) artist_data = [{'type': 'composer', 'name': 'BTS'}, {'type': 'featuring', 'name': 'Bill'}] mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 1}]}, status=response_code.OK )) mocker.patch.object( TrackPersister, 'get_track_artist_by_track_id', return_value=response.Response( message=artist_data, status=response_code.OK )) mocker.patch.object( TrackPersister, 'remove_track_artist', return_value=response.Response( message={} )) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 12, 'subgenre_id': 263}, **logic_headers, correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == [artist_data[0]] assert artist_response.status == response_code.OK def test_update_artist_non_classical(mocker, logic_headers, test_correlation_id, test_orchard_user_id): """Verify valid artists roles for non-classical genre.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) artist_data = [{'type': 'composer', 'name': 'BTS'}, {'type': 'featuring', 'name': 'Bill'}] mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 1}]}, status=response_code.OK )) mocker.patch.object( TrackPersister, 'get_track_artist_by_track_id', return_value=response.Response( message=artist_data, status=response_code.OK )) mocker.patch.object( TrackPersister, 'remove_track_artist', return_value=response.Response( message={} )) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 13, 'subgenre_id': 691}, **logic_headers, correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == [artist_data[1]] assert artist_response.status == response_code.OK def test_update_artist_soundtrack_and_worldmusic(mocker, logic_headers, test_correlation_id, test_orchard_user_id): """Verify valid artists roles for worldmusic and soundtrack subgenre.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) artist_data = [{'type': 'composer', 'name': 'BTS'}, {'type': 'featuring', 'name': 'Bill'}] mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 1}]}, status=response_code.OK )) mocker.patch.object( TrackPersister, 'get_track_artist_by_track_id', return_value=response.Response( message=artist_data, status=response_code.OK )) mocker.patch.object( TrackPersister, 'remove_track_artist', return_value=response.Response( message={} )) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 27, 'subgenre_id': 1040}, **logic_headers, correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == artist_data assert artist_response.status == response_code.OK def test_update_artist_validation(mocker, logic_headers, test_correlation_id, test_orchard_user_id): """Verify validation when passed empty genre.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': ''}, **logic_headers, correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.status == response_code.BAD_REQUEST def test_update_artist_no_artists_found(mocker, logic_headers, test_correlation_id, test_orchard_user_id): """Verify valid artists roles for worldmusic and soundtrack subgenre.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) artist_data = [{'type': 'featuring', 'name': 'Bill'}] mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 1}]}, status=response_code.OK )) mocker.patch.object( TrackPersister, 'get_track_artist_by_track_id', return_value=response.Response( message=artist_data, status=response_code.OK )) mocker.patch.object( TrackPersister, 'remove_track_artist', return_value=response.Response( message={} )) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 12, 'subgenre_id': 123}, **logic_headers, correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == [] assert artist_response.status == response_code.NO_CONTENT def test_update_artist_classical_in_correction(mocker, logic_headers, test_correlation_id, test_orchard_user_id, test_product_id): """Verify valid artists roles for classical genre.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) res = response.Response(message={}) mocker.patch.object( ows_product_workflow, 'get_last_release_correction', return_value=res) artist_data = [{'type': 'composer', 'name': 'BTS', 'tuid': 123}, {'type': 'performer', 'name': 'BTS2', 'tuid': 123}, {'type': 'featuring', 'name': 'Bill', 'tuid': 456}] mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 1}]}, status=response_code.OK )) mocker.patch.object( TrackPersister, 'get_track_artist_by_track_id', return_value=response.Response( message=artist_data, status=response_code.OK )) res = response.Response(message=[{'type': 'composer', 'tuid': 123, 'name': 'BTS'}]) mocker.patch.object( ows_product_digital, 'post_track_correction', return_value=res) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 12, 'subgenre_id': 263, 'correction_id': 123}, account_id=123, account_type='alw', correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == [artist_data[0]] assert artist_response.status == response_code.OK def test_update_artist_non_classical_in_correction(mocker, logic_headers, test_correlation_id, test_orchard_user_id): """Verify valid artists roles for non-classical genre.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) res = response.Response(message={}) mocker.patch.object( ows_product_workflow, 'get_last_release_correction', return_value=res) artist_data = [{'type': 'composer', 'name': 'BTS', 'tuid': 123}, {'type': 'performer', 'name': 'BTS2', 'tuid': 123}, {'type': 'featuring', 'name': 'Bill', 'tuid': 456}] response_post_track_correction = [ {'key_value': [], 'field_name': 'featuring', 'key_id': 1, 'table_name': 'track'}, {'key_value': [{'type': 'composer', 'name': 'BTS'}, {'type': 'performer', 'name': 'BTS2'}], 'field_name': 'track_artist', 'key_id': 1, 'table_name': 'track'}] mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 1}]}, status=response_code.OK )) mocker.patch.object( TrackPersister, 'get_track_artist_by_track_id', return_value=response.Response( message=artist_data, status=response_code.OK )) res = response.Response(message=response_post_track_correction) mocker.patch.object( ows_product_digital, 'post_track_correction', return_value=res) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 12, 'subgenre_id': 263, 'correction_id': 123}, account_id=123, account_type='alw', correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == response_post_track_correction assert artist_response.status == response_code.OK def test_update_artist_worldmusic_for_error_correction(mocker, logic_headers, test_correlation_id, test_orchard_user_id): """Verify valid artists roles for worldmusic and soundtrack subgenre.""" _mock_ownership_and_get_track(mocker, tuid=123, product_id=100) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) res = response.Response(message={}) mocker.patch.object( ows_product_workflow, 'get_last_release_correction', return_value=res) artist_data = [{'type': 'composer', 'name': 'BTS', 'tuid': 123}, {'type': 'producer', 'name': 'BTS2', 'tuid': 123}, {'type': 'featuring', 'name': 'Bill', 'tuid': 456}] response_post_track_correction = [ {'key_value': [{'type': 'composer', 'name': 'BTS'}], 'field_name': 'track_artist', 'key_id': 1, 'table_name': 'track'}] mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 123}]}, status=response_code.OK )) mocker.patch.object( TrackPersister, 'get_track_artist_by_track_id', return_value=response.Response( message=artist_data, status=response_code.OK )) res = response.Response(message=response_post_track_correction) mocker.patch.object( ows_product_digital, 'post_track_correction', return_value=res) artist_response = track_role_logic.update_artist( product_id=100, request_data={'genre_id': 12, 'subgenre_id': 545, 'correction_id': 123}, **logic_headers, correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == response_post_track_correction assert artist_response.status == response_code.OK def test_update_artist_classical_in_correction_with_correction_details( mocker, logic_headers, test_correlation_id, test_orchard_user_id, test_product_id, track_release_correction_details): """Verify valid artists roles for classical genre with correction detail entries.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) res = response.Response(message=track_release_correction_details) mocker.patch.object( ows_product_workflow, 'get_last_release_correction', return_value=res) mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 30452965}]}, status=response_code.OK )) artist_data = [{'type': 'performer', 'tuid': 30452965, 'name': 'test performer'}] res = response.Response(message=artist_data) mocker.patch.object( ows_product_digital, 'post_track_correction', return_value=res) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 12, 'subgenre_id': 263, 'correction_id': 123}, account_id=123, account_type='alw', correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == artist_data assert artist_response.status == response_code.OK def test_update_artist_non_classical_in_correction_with_correction_details( mocker, logic_headers, test_correlation_id, test_orchard_user_id, track_release_correction_details): """Verify valid artists roles for non-classical genre.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) res = response.Response(message=track_release_correction_details) mocker.patch.object( ows_product_workflow, 'get_last_release_correction', return_value=res) artist_data = [{'key_value': [{'name': 'test performer', 'type': 'performer'}], 'field_name': 'track_artist', 'key_id': 30452965, 'table_name': 'track'}] mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message={'items': [{'tuid': 30452965}]}, status=response_code.OK )) res = response.Response(message=artist_data) mocker.patch.object( ows_product_digital, 'post_track_correction', return_value=res) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 13, 'subgenre_id': 263, 'correction_id': 123}, account_id=123, account_type='alw', correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == artist_data assert artist_response.status == response_code.OK def test_update_artist_classical_in_correction_with_and_without_correction_details( mocker, logic_headers, test_correlation_id, test_orchard_user_id, test_product_id, track_release_correction_details, get_tracks_details, track_correction_details_for_all_tracks): """Verify valid artists roles for classical genre with and without correction detail entries.""" _mock_ownership_and_get_track(mocker, tuid=1, product_id=1) mocks.head_response(mocker) mocks.ows_product_get_product_by_id(mocker, product_id=100, upc=123456789012) res = response.Response(message=track_release_correction_details) mocker.patch.object( ows_product_workflow, 'get_last_release_correction', return_value=res) artist_data = [{'type': 'composer', 'name': 'BTS', 'tuid': 30452966}, {'type': 'producer', 'name': 'BTS2', 'tuid': 30452966}, {'type': 'featuring', 'name': 'Bill', 'tuid': 30452966}] mocker.patch.object( TrackPersister, 'get_all_by_product_id_light', return_value=response.Response( message=get_tracks_details, status=response_code.OK )) mocker.patch.object( TrackPersister, 'get_track_artist_by_track_id', return_value=response.Response( message=artist_data, status=response_code.OK )) res = response.Response(message=track_correction_details_for_all_tracks) mocker.patch.object( ows_product_digital, 'post_track_correction', return_value=res) artist_response = track_role_logic.update_artist( product_id=1, request_data={'genre_id': 12, 'subgenre_id': 263, 'correction_id': 123}, account_id=123, account_type='alw', correlation_id=test_correlation_id, orchard_user_id=test_orchard_user_id ) assert artist_response.message == track_correction_details_for_all_tracks assert artist_response.status == response_code.OK