"""Test for Track Persister.""" import datetime from unittest.mock import MagicMock from oto import response from oto import status as response_code import pytest import sqlalchemy import sqlalchemy.orm.session from backend.connectors import mysql from backend.constants import api as api_consts from backend.constants import error from backend.constants import track_field as tf from backend.models.track import Track from backend.models.track import TrackArtist from backend.models.track import TrackMasterRights from backend.models.track import TrackProducerNationality from backend.models.track import TrackPublisher from backend.models.track import TrackWriter from backend.models.track_additional_isrc import TrackAdditionalIsrc from backend.models.track_additional_isrc import TrackAdditionalIsrcType from backend.models.track_persister import TrackPersister from backend.models.track_spatial import TrackSpatial from backend.models.track_query import TrackQuery from tests.testutils import db from tests.testutils import functions as test_funcs from tests.testutils.constants import TEST_SINGLE_TRACK_PRODUCT_ID @db.test_schema def test_get_by_tuid(test_track): """Test getting track by tuid is successful.""" response = TrackPersister.get_by_tuid(test_track[tf.TUID]) assert response.status == response_code.OK track = response.message for field in test_track: assert track[field] == test_track[field] @db.test_schema def test_get_by_invalid_tuid(): """Test getting track by invalid tuid.""" response = TrackPersister.get_by_tuid(0) assert response.status == response_code.NOT_FOUND @db.test_schema def test_get_all_by_isrc_and_type(test_track): """Test getting by isrc and type is successful.""" results = TrackPersister.get_all_by_isrc_and_type( test_track['isrc'], tf.TRACK_TYPE_MUSIC ) assert len(results) != 0 for track in results: assert track['isrc'] == test_track['isrc'] assert track['track_type'] == tf.TRACK_TYPE_MUSIC @db.test_schema_no_seed def test_get_mutlipe_by_tuids(track_factory): """Test getting tracks by a list of tuids is successful.""" tracks = [track_factory(), track_factory(), track_factory()] db.merge_model_objects(tracks) tuids = [track.tuid for track in tracks] response = TrackPersister.get_multiple_by_tuids(tuids) assert response.status == response_code.OK assert response.message['pagination']['total_records'] == len(tuids) @db.test_schema_no_seed def test_get_mutlipe_by_tuids_invalid_tuid(track_factory): """Test getting tracks by a list of tuids is successful.""" tracks = [track_factory(), track_factory()] db.merge_model_objects(tracks) tuids = [track.tuid for track in tracks] tuids.append(tuids[-1] + 1) response = TrackPersister.get_multiple_by_tuids(tuids) assert response.status == response_code.NOT_FOUND @db.test_schema_no_seed def test_get_mutlipe_by_tuids_with_nones(track_factory): """Test getting tracks by a list of tuids with nones is successful.""" tracks = [track_factory(), track_factory(), track_factory()] db.merge_model_objects(tracks) tuids = [ 1000, tracks[0].tuid, 1001, tracks[1].tuid, tracks[2].tuid, 10002, ] response = TrackPersister.get_multiple_by_tuids_with_nones(tuids) assert response.status == response_code.OK assert response.message['pagination']['total_records'] == len(tuids) assert response.message['items'] == [ None, tracks[0].to_dict(), None, tracks[1].to_dict(), tracks[2].to_dict(), None, ] @db.test_schema def test_get_all_tracks_by_product_id(test_multi_tracks_product): """Test getting all tracks by product_id is successful.""" test_product_id = test_multi_tracks_product['product_id'] test_volumes_tup = test_multi_tracks_product['volumes'] response = TrackPersister.get_all_by_product_id(test_product_id) assert response.status == response_code.OK items = response.message[api_consts.ITEMS] pagination = response.message[api_consts.PAGINATION] vol_dict = {} # Dictionary of found volumes {vol_number: count} tuids_dict = {} # Lookup of track tuids (to verify uniqueness) for track in items: assert track[tf.PRODUCT_ID] == test_product_id # Validate tuids tuid = track[tf.TUID] assert tuid not in tuids_dict tuids_dict[tuid] = 1 # Update volume dict count vol_num = track[tf.VOLUME_NUMBER] vol_dict[vol_num] = vol_dict.get(vol_num, 0) + 1 # Validate volume and track counts assert len(test_volumes_tup) == len(vol_dict) assert test_volumes_tup == (vol_dict[1], vol_dict[2], vol_dict[3]) assert pagination['type'] == api_consts.PAGINATION_TYPE_NONE assert pagination['total_records'] == len(items) @db.test_schema def test_get_tuids_by_product_ids_with_order(test_product_ids): """Test getting all tracks by product_id is successful.""" tracks_query = TrackPersister.get_tuids_by_product_ids_with_order( test_product_ids, order_by_fields=[{'column_name': 'volume_number', 'order': 'asc'}], ) product_ids = [t['product_id'] for t in tracks_query] assert set(test_product_ids) == set(product_ids) @db.test_schema def test_get_tuids_with_none_by_product_ids_with_order( test_invalid_product_ids, ): """Test tracks by correct product_id and None with incorrect one.""" tracks_query = TrackPersister.get_tuids_by_product_ids_with_order( test_invalid_product_ids, order_by_fields=[{'column_name': 'track_number', 'order': 'asc'}], ) product_ids = [t['product_id'] for t in tracks_query] assert product_ids == [TEST_SINGLE_TRACK_PRODUCT_ID] @db.test_schema def test_get_all_tracks_by_product_id_light(test_multi_tracks_product): """Test getting all tracks by product_id light is successful.""" test_product_id = test_multi_tracks_product['product_id'] test_volumes_tup = test_multi_tracks_product['volumes'] response = TrackPersister.get_all_by_product_id_light(test_product_id) assert response.status == response_code.OK items = response.message[api_consts.ITEMS] pagination = response.message[api_consts.PAGINATION] vol_dict = {} # Dictionary of found volumes {vol_number: count} tuids_dict = {} # Lookup of track tuids (to verify uniqueness) for track in items: assert track[tf.PRODUCT_ID] == test_product_id # Validate tuids tuid = track[tf.TUID] assert tuid not in tuids_dict tuids_dict[tuid] = 1 # Update volume dict count vol_num = track[tf.VOLUME_NUMBER] vol_dict[vol_num] = vol_dict.get(vol_num, 0) + 1 # Validate volume and track counts assert len(test_volumes_tup) == len(vol_dict) assert test_volumes_tup == (vol_dict[1], vol_dict[2], vol_dict[3]) assert pagination['type'] == api_consts.PAGINATION_TYPE_NONE assert pagination['total_records'] == len(items) @db.test_schema_no_seed def test_get_all_by_product_ids_medium(track_factory): """Test getting all tracks by product_ids medium is successful.""" tracks = [track_factory(), track_factory(), track_factory()] db.merge_model_objects(tracks) product_ids = [tracks[0].product_id] response = TrackPersister.get_all_by_product_ids_medium(product_ids) assert response.status == response_code.OK assert response.message['pagination']['total_records'] == len(tracks) @db.test_schema_no_seed def test_get_all_focus_tracks_by_product_id(track_factory): """Test getting all focus tracks by product_id is successful.""" tracks = [ track_factory( focus_track=1, focus_track__start_date=datetime.date.fromisoformat('2022-10-21'), focus_track__end_date=datetime.date.fromisoformat('2022-10-25')), track_factory( focus_track=1, focus_track__start_date=datetime.date.fromisoformat('2022-11-14'), focus_track__end_date=datetime.date.fromisoformat('2022-11-19')), track_factory() ] db.merge_model_objects(tracks) test_product_id = 1 response = TrackPersister.get_all_focus_track_by_product_id(test_product_id) assert response.status == response_code.OK items = response.message[api_consts.ITEMS] pagination = response.message[api_consts.PAGINATION] total_focus_tracks = 0 tuids_dict = {} # Lookup of track tuids (to verify uniqueness) for track in items: assert track[tf.PRODUCT_ID] == test_product_id # Validate tuids tuid = track[tf.TUID] assert tuid not in tuids_dict tuids_dict[tuid] = 1 if track[tf.FOCUS_TRACK] == 'Y': total_focus_tracks += 1 assert pagination['type'] == api_consts.PAGINATION_TYPE_NONE assert pagination['total_records'] == 2 @db.test_schema def test_create_track_new_product( track_create_data, track_create_expected_response): """Test create track is successful for a new product.""" product_id = track_create_data.pop(tf.PRODUCT_ID) upc = track_create_expected_response[tf.UPC] response = TrackPersister.create_track( product_id, upc, track_create_data) assert response.status == response_code.CREATED assert response.message == track_create_expected_response # Same test, but explicitly passes a volume number product_id += 1 track_create_data[tf.VOLUME_NUMBER] = 1 track_create_expected_response[tf.TUID] += 1 track_create_expected_response[tf.PRODUCT_ID] = product_id response = TrackPersister.create_track(product_id, upc, track_create_data) assert response.status == response_code.CREATED assert response.message == track_create_expected_response @db.test_schema def test_create_track_existing_product( test_multi_tracks_product, track_max_tuid, test_upc): """Test create track is successful for existing product.""" test_product_id = test_multi_tracks_product['product_id'] test_volumes_tup = test_multi_tracks_product['volumes'] response = TrackPersister.create_track( test_product_id, test_upc, track_data={'track_name': 'Some Track'}) assert response.status == response_code.CREATED track = response.message assert track[tf.TUID] > track_max_tuid assert track[tf.TRACK_NAME] == 'Some Track' assert track[tf.PRODUCT_ID] == test_product_id assert track[tf.VOLUME_NUMBER] == len(test_volumes_tup) assert track[tf.TRACK_NUMBER] == test_volumes_tup[-1] + 1 @db.test_schema def test_create_track_existing_product_with_existing_volume( test_multi_tracks_product, track_max_tuid): """Test create track is successful for product with existing volume. The volume is one less than the last volume. """ test_product_id = test_multi_tracks_product['product_id'] test_volumes_tup = test_multi_tracks_product['volumes'] test_upc = test_multi_tracks_product['upc'] test_volume_num = len(test_volumes_tup) - 1 test_volume_num_tracks = test_volumes_tup[test_volume_num - 1] response = TrackPersister.create_track( test_product_id, test_upc, track_data={'track_name': 'Some Track'}, for_volume=test_volume_num) assert response.status == response_code.CREATED track = response.message assert track[tf.TUID] > track_max_tuid assert track[tf.TRACK_NAME] == 'Some Track' assert track[tf.PRODUCT_ID] == test_product_id assert track[tf.VOLUME_NUMBER] == test_volume_num assert track[tf.TRACK_NUMBER] == test_volume_num_tracks + 1 @db.test_schema def test_create_track_existing_product_with_new_volume( test_multi_tracks_product, track_max_tuid): """Test create track is successful with a new volume. New volume is one higher than an existing volume. """ test_product_id = test_multi_tracks_product['product_id'] test_volumes_tup = test_multi_tracks_product['volumes'] test_upc = test_multi_tracks_product['upc'] response = TrackPersister.create_track( test_product_id, test_upc, track_data={'track_name': 'Some Track'}, for_volume=4) assert response.status == response_code.CREATED track = response.message assert track[tf.TUID] > track_max_tuid assert track[tf.TRACK_NAME] == 'Some Track' assert track[tf.PRODUCT_ID] == test_product_id assert track[tf.VOLUME_NUMBER] == len(test_volumes_tup) + 1 assert track[tf.TRACK_NUMBER] == 1 assert track[tf.US_PUBLISHING_OBLIGATION] is None assert track[tf.THIRD_PARTY_PUBLISHER] == 'N' @db.test_schema def test_create_track_new_product_with_invalid_volume( test_upc, track_create_data, track_create_expected_response, track_max_tuid): """Test create track fails when invalid vol for new product is passed.""" product_id = track_create_data.pop(tf.PRODUCT_ID) response = TrackPersister.create_track( product_id, test_upc, track_data={'track_name': 'Some Track'}, for_volume=2) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.VALIDATION_ERROR_CODE assert response.errors['message'] == \ error.VALIDATION_SEQ_INTEGRITY_ERROR_MSG @db.test_schema def test_create_track_existing_product_with_invalid_volume( test_multi_tracks_product, track_max_tuid): """Test create track fails when invalid volume is passed. The new volume is more than one higher than an existing volume. """ test_product_id = test_multi_tracks_product['product_id'] test_volumes_tup = test_multi_tracks_product['volumes'] test_upc = test_multi_tracks_product['upc'] response = TrackPersister.create_track( test_product_id, test_upc, track_data={'track_name': 'Some Track'}, for_volume=len(test_volumes_tup) + 2) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.VALIDATION_ERROR_CODE assert response.errors['message'] == \ error.VALIDATION_SEQ_INTEGRITY_ERROR_MSG response2 = TrackPersister.create_track( test_product_id, test_upc, track_data={'track_name': 'Some Track'}, for_volume=0) assert response2.status == response_code.BAD_REQUEST assert response2.errors['code'] == error.VALIDATION_ERROR_CODE @db.test_schema def test_create_track_invalid_upc_type(track_max_product_id): """Test create track fails when invalid UPC type is passed in.""" product_id = track_max_product_id + 1 response = TrackPersister.create_track( product_id, None, track_data={'track_name': 'Some Track'}, for_volume=1) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.VALIDATION_ERROR_CODE @db.test_schema def test_create_track_invalid_data(track_max_product_id, test_upc): """Test create track fails when track_number is passed in.""" response = TrackPersister.create_track( track_max_product_id + 1, test_upc, track_data={'track_name': 'Some Track', 'track_number': 1}, for_volume=1) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.VALIDATION_ERROR_CODE @db.test_schema def test_create_track_invalid_field(track_max_product_id, test_upc): """Test create track fails when invalid field is passed in.""" response = TrackPersister.create_track( track_max_product_id + 1, test_upc, track_data={'track_name': 'Some Track', 'not_a_real_field': 'fail'}) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.VALIDATION_ERROR_CODE @db.test_schema def test_bulk_create_tracks_new_product( track_bulk_create_data, track_bulk_create_new_prod_response, mocker): """Test create track is successful for a new product.""" first_response = track_bulk_create_new_prod_response['items'][0] product_id = first_response[tf.PRODUCT_ID] upc = first_response[tf.UPC] rollback_spy = mocker.spy(sqlalchemy.orm.session.Session, 'rollback') response = TrackPersister.bulk_create_tracks( product_id, upc, track_bulk_create_data) assert response.status == response_code.CREATED assert response.message == track_bulk_create_new_prod_response assert rollback_spy.call_count == 0 @db.test_schema def test_bulk_create_tracks_existing_product( track_bulk_create_data, track_bulk_create_existing_prod_response, mocker): """Test create track is successful for an existing product.""" first_response = track_bulk_create_existing_prod_response['items'][0] product_id = first_response[tf.PRODUCT_ID] upc = first_response[tf.UPC] rollback_spy = mocker.spy(sqlalchemy.orm.session.Session, 'rollback') response = TrackPersister.bulk_create_tracks( product_id, upc, track_bulk_create_data) assert response.status == response_code.CREATED assert response.message == track_bulk_create_existing_prod_response assert rollback_spy.call_count == 0 @db.test_schema def test_update_track( track_update_tuid, track_update_data, track_update_expected_response_without_localizations): """Test updating track is successful and changes persist.""" response = TrackPersister.update_track( track_update_tuid, track_update_data) assert response.status == response_code.OK assert ( response.message == track_update_expected_response_without_localizations) # Verify database was updated response = TrackPersister.get_by_tuid(track_update_tuid) assert response.status == response_code.OK assert ( response.message == track_update_expected_response_without_localizations) @db.test_schema_no_seed def test_update_track_clean_persisted_data( mocker, track_factory, test_upc): """Bad data that is already persisted should be fixed on update.""" track = track_factory( third_party_publisher='', us_publishing_obligation='') db.merge_model_objects([track]) with mysql.db_session() as session: track = TrackQuery.get_by_tuid(tuid=1, session=session) assert track.third_party_publisher == '' assert track.us_publishing_obligation == '' TrackPersister._apply_track_update(track, {}) assert track.third_party_publisher == 'N' assert track.us_publishing_obligation is None with mysql.db_session() as session: result = mysql.db_engine.execute( 'select third_party_publisher, us_publishing_obligation ' 'from track where id = 1') row = next(iter(result)) assert row[0] == 'N' assert row[1] is None @pytest.fixture @db.test_schema_no_seed def test_update_track_with_different_artist(track_factory): """Test updating track changes artist.""" track = track_factory(performer__count=1) db.merge_model_objects([track]) update_data = {'artists': [{ 'type': 'performer', 'name': 'Performer 1', }]} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response test_funcs.validate_artists_are_same( response.message[tf.ARTISTS], update_data['artists']) @pytest.fixture @db.test_schema_no_seed def test_update_track_artist_with_artist_info_id(track_factory): """Test updating track artist changes artist_info_id.""" track = track_factory(performer__count=1) db.merge_model_objects([track]) update_data = {'artists': [{ 'track_artist_id': 1, 'type': 'performer', 'name': 'Performer 1', 'artist_info_id': 1 }]} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response test_funcs.validate_artists_are_same( response.message[tf.ARTISTS], update_data['artists']) @pytest.fixture @db.test_schema_no_seed def test_update_track_artist_with_artist_info_id_none(track_factory): """Test updating track artist accepts null artist_info_id.""" track = track_factory(performer__count=1) db.merge_model_objects([track]) update_data = {'artists': [{ 'track_artist_id': 1, 'type': 'performer', 'name': 'Performer 1', 'artist_info_id': None }]} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response test_funcs.validate_artists_are_same( response.message[tf.ARTISTS], update_data['artists']) @db.test_schema_no_seed def test_update_track_artist_with_artist_info_id_error(track_factory): """Test updating track artist with artist_info_id error.""" track = track_factory(performer__count=1) db.merge_model_objects([track]) update_data = {'artists': [{ 'type': 'performer', 'name': 'Performer 1', 'artist_info_id': 1 }]} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == 'validation_error' @pytest.fixture @db.test_schema_no_seed def test_update_track_writer_with_artist_info_id(track_factory): """Test updating track writer changes artist_info_id.""" track = track_factory(writers__count=1) db.merge_model_objects([track]) update_data = { 'writers': [ { 'track_writer_id': 1, 'type': 'writer', 'name': 'Julius', 'artist_info_id': 1 } ] } update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response test_funcs.validate_writers_are_same( response.message[tf.WRITERS], update_data['writers']) @pytest.fixture @db.test_schema_no_seed def test_update_track_writer_with_artist_info_id_none(track_factory): """Test updating track writer accepts null artist_info_id.""" track = track_factory(writers__count=1) db.merge_model_objects([track]) update_data = { 'writers': [ { 'track_writer_id': 1, 'type': 'writer', 'name': 'Julius', 'artist_info_id': None } ] } update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response test_funcs.validate_writers_are_same( response.message[tf.WRITERS], update_data['writers']) @db.test_schema_no_seed def test_update_track_writer_with_artist_info_id_error(track_factory): """Test updating track writer with artist_info_id but no id.""" track = track_factory(writers__count=1) db.merge_model_objects([track]) update_data = { 'writers': [ { 'type': 'writer', 'name': 'Julius', 'artist_info_id': 1 } ] } update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == 'validation_error' @pytest.fixture @db.test_schema_no_seed def test_update_track_with_no_artists(track_factory): """Test updating track removes artists.""" track = track_factory(performer__count=2) db.merge_model_objects([track]) update_data = {'artists': []} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response assert len(response.message[tf.ARTISTS]) == 0 @pytest.fixture @db.test_schema_no_seed def test_update_track_with_legacy_artist(track_factory): """Test updating track changes artist.""" track = track_factory(performer__count=1) db.merge_model_objects([track]) update_data = {'artists': [{ 'type': 'soloist', 'name': 'Performer 1' }]} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response test_funcs.validate_artists_are_same( response.message[tf.ARTISTS], update_data['artists']) @pytest.fixture @db.test_schema_no_seed def test_update_track_with_writers(track_factory): """Test updating track changes artist.""" track = track_factory() db.merge_model_objects([track]) update_data = {'writers': [{ 'type': 'writer', 'name': 'Writer 1' }]} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response assert tf.WRITERS in response.message writers = response.message[tf.WRITERS] assert len(writers) == 1 assert writers[0]['name'] == 'Writer 1' @db.test_schema def test_update_track_invalid_track(track_update_data): """Test updating non-existent track.""" response = TrackPersister.update_track(1000, track_update_data) assert response.status == response_code.NOT_FOUND, \ 'Track should not be found' @db.test_schema_no_seed def test_update_track_with_invalid_artist(track_factory): """Test updating track fails with invalid artist.""" track = track_factory() db.merge_model_objects([track]) update_data = {'artists': [{'name': 'Performer', 'type': 'invalid_role'}]} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response.status == response_code.BAD_REQUEST response.errors['code'] == 'validation_error' @db.test_schema_no_seed def test_update_track_with_id(track_factory): """Test updating track fails when .""" track = track_factory() db.merge_model_objects([track]) update_data = {'artists': [{'name': 'Performer', 'type': 'invalid_role'}]} update_request = test_funcs.create_track_update_request( track.to_dict(), update_data) response = TrackPersister.update_track( track.tuid, update_request) assert response.status == response_code.BAD_REQUEST response.errors['code'] == 'validation_error' @db.test_schema def test_update_track_invalid_field(test_track): """Test update track fails when invalid field is passed in.""" tuid = test_track[tf.TUID] # Update the track track_updates = { tf.TRACK_NAME: 'New Name', tf.VERSION: 'New Version', 'not_a_real_field': 'This field causes a failure' } response = TrackPersister.update_track(tuid, track_updates) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.VALIDATION_ERROR_CODE def test_update_track_integrity_error(track_factory, mocker): """Test update track fails when invalid field is passed in.""" track = track_factory() mocker.patch.object(TrackQuery, 'get_by_tuids', return_value=[track]) mock_sentry = mocker.patch('backend.models.track_persister.capture_exception') session = MagicMock() session.merge = MagicMock( side_effect=sqlalchemy.exc.IntegrityError( statement='sql', params=[], orig='orig')) # Update the track track_updates = {tf.TRACK_NAME: 'New Name'} response = TrackPersister.update_track( track.tuid, track_updates, session=session) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.TOO_MANY_WRITES_CODE assert mock_sentry.called @db.test_schema def test_update_track_with_artist_info_ids_to_set( track_update_tuid, track_update_data, mocker): """Test updating track is successful if artist_info_ids_to_set.""" track_update_data['artists'] = [ { 'artist_info_id': 1, 'type': 'performer', 'name': 'Snowball' } ] _apply_track_update_spy = mocker.spy( TrackPersister, '_apply_track_update') response = TrackPersister.update_track( track_update_tuid, track_update_data, get_artist_info_ids_from_existing_rows=True) assert response.status == response_code.OK response = TrackPersister.get_by_tuid(track_update_tuid) assert response.status == response_code.OK assert response.message['artists'] == [ { 'artist_info_id': 1, 'name': 'Snowball', 'track_artist_id': 1, 'type': 'performer' }] assert _apply_track_update_spy.call_args[1]['artist_info_ids_to_set'] == ( { 'artists': {('performer', 'Snowball'): 1}, 'writers': {} }) @db.test_schema def test_update_multiple_tracks(test_tuids): """Test updating track is successful and changes persist.""" track_update_list = [] i = 1 for tuid in sorted(test_tuids): track_update_list.append({ tf.TUID: tuid, tf.VERSION: 'version {}'.format(i) }) i += 1 response = TrackPersister.update_multiple_tracks(track_update_list) assert response assert response.message[api_consts.PAGINATION]['total_records'] == 5 i = 1 for track in response.message[api_consts.ITEMS]: assert track[tf.VERSION] == 'version {}'.format(i) i += 1 i = 1 with mysql.db_session() as session: tracks = TrackQuery.get_by_tuids(test_tuids, session) for track in tracks: assert track.version == 'version {}'.format(i) i += 1 @db.test_schema def test_update_multiple_tracks_with_volume_and_track_number(test_tuids): """Test updating tracks with volume / track number.""" track_update_list = [] i = 1 for tuid in sorted(test_tuids): track_update_list.append({ tf.TUID: tuid, tf.VERSION: 'version {}'.format(i), tf.VOLUME_NUMBER: 1, tf.TRACK_NUMBER: i + 100 }) i += 1 response = TrackPersister.update_multiple_tracks( track_update_list, update_track_volume_numbers=True) assert response assert response.message[api_consts.PAGINATION]['total_records'] == 5 i = 1 for track in response.message[api_consts.ITEMS]: assert track[tf.VERSION] == 'version {}'.format(i) i += 1 i = 1 with mysql.db_session() as session: tracks = TrackQuery.get_by_tuids(test_tuids, session) for track in tracks: assert track.version == 'version {}'.format(i) assert track.volume_number == 1 assert track.track_number == i + 100 i += 1 @db.test_schema def test_update_multiple_tracks_with_get_artist_info_ids_from_existing_rows( test_tuids): """Test updating tracks with get_artist_info_ids_from_existing_rows.""" track_update_list = [] i = 1 test_tuids = test_tuids[:3] for tuid in sorted(test_tuids): track_update_list.append({ tf.TUID: tuid, tf.VERSION: 'version {}'.format(i), tf.ARTISTS: [{ 'name': 'Artist1', 'type': 'performer', 'artist_info_id': i + 200 }] }) i += 1 response = TrackPersister.update_multiple_tracks( track_update_list, get_artist_info_ids_from_existing_rows=True) assert response assert response.message[api_consts.PAGINATION]['total_records'] == 3 with mysql.db_session() as session: tracks = TrackQuery.get_by_tuids(test_tuids, session) i = 1 for track in tracks: assert track.version == 'version {}'.format(i) assert track.artists[0].artist_info_id == i + 200 i += 1 @db.test_schema def test_update_field_for_all_tracks_in_product(test_track): """Test tracks field are all updated for a product.""" product_id = test_track[tf.PRODUCT_ID] field_updates = ( (tf.VERSION, 'radio edit',), (tf.META_LANGUAGE_CODE, 'ITA',), (tf.P_INFO, '2020 Make Great Meows',), (tf.EXPLICIT, 'Y',), (tf.OWNERSHIP_RIGHTS, 'original_owner',), (tf.RECORDING_COUNTRY_ID, 7,), (tf.ORIGINAL_RIGHTS_HOLDER_COUNTRY_ID, 7,), ) # Do Apply to All operation for each field for (field_name, field_value,) in field_updates: response = TrackPersister.update_field_for_all_tracks_in_product( product_id, field_name, field_value) assert response.status == response_code.OK deltas = response.message['items'] # Make sure 9 tracks are updated assert len(deltas) == 9 # Verify response of deltas are correct for delta in deltas: assert len(delta) == 2 assert delta[field_name] == field_value assert delta[tf.TUID] # Make sure changes are saved response = TrackPersister.get_all_by_product_id(product_id) for track in response.message['items']: for (field_name, field_value,) in field_updates: assert track[field_name] == field_value, \ 'Value for field {} does not match'.format(field_name) @db.test_schema def test_update_field_for_all_tracks_in_product_invalid(): """Test tracks field are not updated for a product when invalid.""" response = TrackPersister.update_field_for_all_tracks_in_product( 1, tf.TRACK_NAME, 'bad bad kitty') assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.VALIDATION_ERROR_CODE @db.test_schema def test_sync_role_for_all_tracks_in_product( test_valid_product_id, test_product_valid_tracks_response, product_role_apply_to_all_data, product_role_apply_to_all_role_type): """Test remixer role is synced for tracks in product.""" role_type = product_role_apply_to_all_role_type names = product_role_apply_to_all_data['names'] response = TrackPersister.sync_role_for_all_tracks_in_product( test_valid_product_id, role_type, names) assert response.status == response_code.OK test_funcs.validate_artists_persister_response( before_apply_response=test_product_valid_tracks_response, names_list=product_role_apply_to_all_data['names'], deltas_list=response.message['items'], role_type=product_role_apply_to_all_role_type) def test_sync_role_for_all_tracks_in_product_invalid_role( test_product_id, test_product_tracks_response, product_role_apply_to_all_data, product_role_apply_to_all_role_type): """Test remixer role is synced for tracks in product.""" response = TrackPersister.sync_role_for_all_tracks_in_product( 1, 'invalid_role', []) assert response.status == response_code.BAD_REQUEST assert response.errors['code'] == error.INVALID_ROLE_ERROR_CODE def test_role_type_to_field(): """Test role_type to track field conversion.""" assert TrackPersister._role_type_to_field('performer') == tf.ARTISTS assert TrackPersister._role_type_to_field('composer') == tf.ARTISTS assert TrackPersister._role_type_to_field('publisher') == tf.PUBLISHERS assert TrackPersister._role_type_to_field('writer') == tf.WRITERS with pytest.raises(ValueError): TrackPersister._role_type_to_field('invalid_role_type') @db.test_schema def test_add_track_artist(test_track): """Test adding artist to track.""" tuid = test_track[tf.TUID] response = TrackPersister.add_track_artist( tuid, 'performer', 'Some new artist') assert response.status == response_code.CREATED, \ 'Artist should be created' new_artist = response.message # Verify artist is saved in track track_response = TrackPersister.get_by_tuid(tuid) track_artists = track_response.message['artists'] found = False for artist in track_artists: if new_artist == artist: found = True break assert found, 'Created artist should be saved' @db.test_schema def test_add_track_artist_invalid_type(test_track): """Test adding artist to track.""" tuid = test_track[tf.TUID] response = TrackPersister.add_track_artist( tuid, 'bad_artist_type', 'Some new artist') assert response.status == response_code.BAD_REQUEST, \ 'Adding new artist with bad artist type should fail' assert response.errors['code'] == error.INVALID_VALUE_ERROR_CODE @db.test_schema def test_add_track_artist_invalid_track(): """Test adding artist to non-existent track.""" response = TrackPersister.add_track_artist( 10000, 'performer', 'Some new artist') assert response.status == response_code.NOT_FOUND, \ 'Track should not be found' @db.test_schema def test_add_track_artist_duplicate(test_track): """Test adding artist with duplicate name to track.""" tuid = test_track[tf.TUID] response = TrackPersister.add_track_artist(tuid, 'performer', 'Snowball') assert response.status == response_code.BAD_REQUEST, \ 'Duplicate artist must not be created' assert response.errors['code'] == error.DUPLICATE_CODE @db.test_schema_no_seed def test_update_track_artist(track_factory): """Test updating a track artist.""" track = track_factory(performer__count=1) db.merge_model_objects([track]) tuid = track.tuid TrackPersister.update_track_artist( tuid, 'performer', 1, {'name': 'hello world'} ) with mysql.db_session() as session: track = session.query(TrackArtist).filter_by(track_artist_id=1).first() assert track.name == 'hello world' @db.test_schema_no_seed def test_update_track_artist_not_found(): """Test updating a track artist that doesn't exist.""" response = TrackPersister.update_track_artist( 1, 'performer', 1, {'name': 'hello world'} ) assert response.status == response_code.NOT_FOUND @db.test_schema def test_remove_track_artist(test_track): """Test removing artist from track.""" tuid = test_track[tf.TUID] # Find artist_id track_response = TrackPersister.get_by_tuid(tuid) track_artist = track_response.message['artists'][0] artist_type = track_artist['type'] artist_id = track_artist['track_artist_id'] response = TrackPersister.remove_track_artist( tuid, artist_type, artist_id) assert response.status == response_code.OK # Verify artist was deleted track_response = TrackPersister.get_by_tuid(tuid) track_artists = track_response.message['artists'] for artist in track_artists: assert track_artist != artist, 'Artist should not exist' @db.test_schema def test_remove_track_artist_invalid_track(): """Test removing artist from non-existent track.""" response = TrackPersister.remove_track_artist(10000, 'performer', 1) assert response.status == response_code.NOT_FOUND, \ 'Track should not be found' @db.test_schema def test_remove_invalid_track_artist(test_track): """Test removing invalid artist from track.""" tuid = test_track[tf.TUID] response = TrackPersister.remove_track_artist(tuid, 'performer', 1) assert response.status == response_code.NOT_FOUND, \ 'Track artist should not be found' @db.test_schema def test_remove_track_artist_invalid_artist_type(test_track): """Test removing artist from track with invalid role specified.""" tuid = test_track[tf.TUID] # Find artist_id track_response = TrackPersister.get_by_tuid(tuid) track_artist = track_response.message['artists'][0] artist_id = track_artist['track_artist_id'] response = TrackPersister.remove_track_artist( tuid, 'bad_artist_type', artist_id) assert response.status == response_code.BAD_REQUEST, \ 'Invalid Track artist type should cause error response' @db.test_schema def test_remove_track_artist_wrong_artist_type(test_track): """Test removing artist from track with wrong role specified.""" tuid = test_track[tf.TUID] # Find artist_id track_response = TrackPersister.get_by_tuid(tuid) track_artist = track_response.message['artists'][0] wrong_artist_type = 'performer' \ if track_artist['type'] == 'remixer' else 'remixer' artist_id = track_artist['track_artist_id'] response = TrackPersister.remove_track_artist( tuid, wrong_artist_type, artist_id) assert response.status == response_code.NOT_FOUND, \ 'Wrong track artist type should cause not found response' @db.test_schema def test_add_track_publisher(test_track): """Test adding publisher to track.""" tuid = test_track[tf.TUID] response = TrackPersister.add_track_publisher( tuid, 'Some new publisher') assert response.status == response_code.CREATED, \ 'publisher should be created' new_publisher = response.message # Verify publisher is saved in track track_response = TrackPersister.get_by_tuid(tuid) track_publishers = track_response.message['publishers'] found = False for publisher in track_publishers: if new_publisher == publisher: found = True break assert found, 'Created publisher should be saved' @db.test_schema def test_add_track_publisher_invalid_track(): """Test adding publisher to non-existent track.""" response = TrackPersister.add_track_publisher( 10000, 'Some new publisher') assert response.status == response_code.NOT_FOUND, \ 'Track should not be found' @db.test_schema def test_add_track_publisher_duplicate(test_track): """Test adding publisher with duplicate name to track.""" tuid = test_track[tf.TUID] response = TrackPersister.add_track_publisher(tuid, 'Meow Meow Meow') assert response.status == response_code.BAD_REQUEST, \ 'Duplicate publisher must not be created' assert response.errors['code'] == error.DUPLICATE_CODE @db.test_schema def test_remove_track_publisher(test_track): """Test removing publisher from track.""" tuid = test_track[tf.TUID] # Find publisher_id track_response = TrackPersister.get_by_tuid(tuid) track_publisher = track_response.message['publishers'][0] publisher_id = track_publisher['track_publisher_id'] response = TrackPersister.remove_track_publisher(tuid, publisher_id) assert response.status == response_code.OK # Verify publisher was deleted track_response = TrackPersister.get_by_tuid(tuid) track_publishers = track_response.message['publishers'] for publisher in track_publishers: assert track_publisher != publisher, 'Publisher should not exist' @db.test_schema def test_remove_track_publisher_invalid_track(): """Test removing publisher from non-existent track.""" response = TrackPersister.remove_track_publisher(10000, 1) assert response.status == response_code.NOT_FOUND, \ 'Track should not be found' @db.test_schema def test_remove_invalid_track_publisher(test_track): """Test removing invalid publisher from track.""" tuid = test_track[tf.TUID] response = TrackPersister.remove_track_publisher(tuid, 1) assert response.status == response_code.NOT_FOUND, \ 'Track publisher should not be found' @db.test_schema def test_add_track_writer(test_track): """Test adding writer to track.""" tuid = test_track[tf.TUID] response = TrackPersister.add_track_writer( tuid, 'Some new writer') assert response.status == response_code.CREATED, \ 'Writer should be created' new_writer = response.message # Verify writer is saved in track track_response = TrackPersister.get_by_tuid(tuid) track_writers = track_response.message['writers'] found = False for writer in track_writers: if new_writer == writer: found = True break assert found, 'Created writer should be saved' @db.test_schema def test_add_track_writer_invalid_track(): """Test adding writer to non-existent track.""" response = TrackPersister.add_track_writer( 10000, 'Some new writer') assert response.status == response_code.NOT_FOUND, \ 'Track should not be found' @db.test_schema def test_add_track_writer_duplicate(test_track): """Test adding writer with duplicate name to track.""" tuid = test_track[tf.TUID] response = TrackPersister.add_track_writer(tuid, 'Furball') assert response.status == response_code.BAD_REQUEST, \ 'Duplicate writer must not be created' assert response.errors['code'] == error.DUPLICATE_CODE @db.test_schema def test_update_track_writer(test_track): """Test updating a track writer.""" tuid = test_track[tf.TUID] # Find writer_id track_response = TrackPersister.get_by_tuid(tuid) track_writer = track_response.message['writers'][0] writer_id = track_writer['track_writer_id'] TrackPersister.update_track_writer( tuid, writer_id, {'name': 'hello world'} ) with mysql.db_session() as session: track = session\ .query(TrackWriter).filter_by(track_writer_id=writer_id).first() assert track.name == 'hello world' @db.test_schema_no_seed def test_update_track_writer_not_found(): """Test updating a track writer that does not exist.""" response = TrackPersister.update_track_writer(1, 1, {'name': 'hello'}) assert response.status == response_code.NOT_FOUND @db.test_schema def test_update_track_duration(track_update_tuid): """Test update track duration is successful.""" mock_track_update_data = { 'duration_minutes': 1, 'duration_seconds': 1 } response = TrackPersister.update_track_duration( track_update_tuid, mock_track_update_data) assert response.status == response_code.OK # Verify database was updated response = TrackPersister.get_by_tuid(track_update_tuid) assert response.status == response_code.OK assert response.message['duration_minutes'] == 1 assert response.message['duration_seconds'] == 1 @db.test_schema def test_remove_track_writer(test_track): """Test removing writer from track.""" tuid = test_track[tf.TUID] # Find writer_id track_response = TrackPersister.get_by_tuid(tuid) track_writer = track_response.message['writers'][0] writer_id = track_writer['track_writer_id'] response = TrackPersister.remove_track_writer(tuid, writer_id) assert response.status == response_code.OK # Verify writer was deleted track_response = TrackPersister.get_by_tuid(tuid) track_writers = track_response.message['writers'] for writer in track_writers: assert track_writer != writer, 'Writer should not exist' @db.test_schema def test_remove_track_writer_invalid_track(): """Test removing writer from non-existent track.""" response = TrackPersister.remove_track_writer(10000, 1) assert response.status == response_code.NOT_FOUND, \ 'Track should not be found' @db.test_schema def test_remove_invalid_track_writer(test_track): """Test removing invalid writer from track.""" tuid = test_track[tf.TUID] response = TrackPersister.remove_track_writer(tuid, 1) assert response.status == response_code.NOT_FOUND, \ 'Track writer should not be found' def test_validate_tracks_order_correct(): """Test track order validation function for correct cases.""" tracks = [ Track(track_name='Track 1', volume_number=1, track_number=1), Track(track_name='Track 2', volume_number=1, track_number=2), Track(track_name='Track 3', volume_number=1, track_number=3) ] is_valid, _ = TrackPersister._validate_tracks_order(tracks) assert is_valid, 'Tracks are valid for a single volume' tracks = [ Track(track_name='Track 1', volume_number=1, track_number=1), Track(track_name='Track 2', volume_number=1, track_number=2), Track(track_name='Track 3', volume_number=2, track_number=1), Track(track_name='Track 3', volume_number=3, track_number=1), Track(track_name='Track 3', volume_number=3, track_number=2) ] is_valid, _ = TrackPersister._validate_tracks_order(tracks) assert is_valid, 'Multiple volumes with tracks' def test_validate_tracks_order_incorrect(): """Test track order validation function for incorrect cases.""" tracks = [Track(track_name='Track 1', volume_number=1.0, track_number=1)] is_valid, error_code = TrackPersister._validate_tracks_order(tracks) assert not is_valid, 'Volume must be integer' assert error_code == error.VALIDATION_ERROR_CODE tracks = [Track(track_name='Track 1', volume_number=1, track_number=1.0)] is_valid, error_code = TrackPersister._validate_tracks_order(tracks) assert not is_valid, 'Track must be integer' assert error_code == error.VALIDATION_ERROR_CODE tracks = [ Track(track_name='Track 1', volume_number=1, track_number=1), Track(track_name='Track 2', volume_number=1, track_number=1) ] is_valid, error_code = TrackPersister._validate_tracks_order(tracks) assert not is_valid, 'Volume should not have duplicate track numbers' assert error_code == error.ERROR_NOT_UNIQUE_CODE tracks = [ Track(track_name='Track 1', volume_number=1, track_number=2), Track(track_name='Track 2', volume_number=1, track_number=3) ] is_valid, error_code = TrackPersister._validate_tracks_order(tracks) assert not is_valid, 'Track number 1 must exist' assert error_code == error.ERROR_SEQ_RANGE_CODE tracks = [ Track(track_name='Track 1', volume_number=1, track_number=1), Track(track_name='Track 2', volume_number=3, track_number=1) ] is_valid, error_code = TrackPersister._validate_tracks_order(tracks) assert not is_valid, 'Volume cannot be skipped' assert error_code == error.ERROR_SEQ_INTEGRITY_CODE @db.test_schema def test_reorder_track(test_multi_tracks_product): """Test track reordering is successful.""" test_product_id = test_multi_tracks_product[tf.PRODUCT_ID] response = TrackPersister.get_all_by_product_id(test_product_id) items = response.message[api_consts.ITEMS] # Rearrange neighboring tracks track_start1 = items[0] track_start2 = items[1] arrgt = test_funcs.make_switch_tracks_arrgt(track_start1, track_start2) response = TrackPersister.reorder(test_product_id, arrgt) assert response.status == response_code.OK # Verify items are rearranged in database with mysql.db_session() as session: tracks = TrackQuery.get_all_by_product_id(test_product_id, session) assert tracks[0].tuid == track_start2[tf.TUID] assert tracks[1].tuid == track_start1[tf.TUID] is_valid, _ = TrackPersister._validate_tracks_order(tracks) assert is_valid, 'Track order is valid' @db.test_schema def test_reorder_track_fail( test_multi_tracks_product): """Test track reordering is successful.""" test_product_id = test_multi_tracks_product[tf.PRODUCT_ID] response = TrackPersister.get_all_by_product_id(test_product_id) items = response.message[api_consts.ITEMS] # rearrange neighboring tracks track_start1 = items[0] track_start2 = items[1] arrgt = test_funcs.make_switch_tracks_arrgt(track_start1, track_start2) arrgt[0][tf.TUID] = 999 response = TrackPersister.reorder(test_product_id, arrgt) assert response.status == response_code.BAD_REQUEST, \ 'Fail on invalid TUID' assert response.errors['code'] == error.VALIDATION_ERROR_CODE arrgt = test_funcs.make_switch_tracks_arrgt(track_start1, track_start2) arrgt[0][tf.TUID] = arrgt[1][tf.TUID] response = TrackPersister.reorder(test_product_id, arrgt) assert response.status == response_code.BAD_REQUEST, \ 'Fail on duplicate TUID' assert response.errors['code'] == error.VALIDATION_ERROR_CODE # rearrange neighboring tracks arrgt = test_funcs.make_switch_tracks_arrgt(track_start1, track_start2) arrgt[0][tf.VOLUME_NUMBER] = 10 response = TrackPersister.reorder(test_product_id, arrgt) assert response.status == response_code.BAD_REQUEST, \ 'Fail on bad track order' assert response.errors['code'] == error.ERROR_SEQ_INTEGRITY_CODE @db.test_schema def test_assign_track_isrcs_for_product_id( mock_claim_new_isrcs, test_multi_tracks_product): """Test assign_track_isrcs_for_product_id is successful.""" test_product_id = test_multi_tracks_product[tf.PRODUCT_ID] response = TrackPersister.assign_track_isrcs_for_product_id( test_product_id) assert response.status == response_code.OK # Verify items are rearranged in database with mysql.db_session() as session: tracks = TrackQuery.get_all_by_product_id(test_product_id, session) for track in tracks: assert track.isrc is not None @db.test_schema def test_assign_track_isrcs_for_product_id_with_no_tracks( mock_claim_new_isrcs, test_multi_tracks_product): """Test assign_track_isrcs_for_product_id with no tracks returns 404.""" # some non-existing product test_product_id = 999999 response = TrackPersister.assign_track_isrcs_for_product_id( test_product_id) assert response.status == response_code.NOT_FOUND @db.test_schema def test_import_tracks(test_product_id, test_upc): """Test import tracks successfull.""" test_source_tuid = 2 with mysql.db_session() as session: original_track_count = TrackQuery.get_all_by_product_id( test_product_id, session, eager_loading=False).count() response = TrackPersister.import_tracks( [test_source_tuid], test_product_id, test_upc) assert response.status == response_code.OK import_source_dest_dict = response.message[api_consts.ITEMS][0] old_tuid = import_source_dest_dict['source'][tf.TUID] destination_track = import_source_dest_dict['destination'] assert test_source_tuid == old_tuid new_tuid = destination_track['tuid'] with mysql.db_session() as session: track_count = TrackQuery.get_all_by_product_id( test_product_id, session, eager_loading=False).count() assert track_count == original_track_count + 1 imported_track = TrackQuery.get_by_tuid(new_tuid, session) assert imported_track @db.test_schema def test_import_tracks_not_found_result(test_product_id, test_upc): """Test import tracks fails when a track doesn't exist.""" test_source_tuid = 2 non_existing_tuid = 99999 tuids = [test_source_tuid, non_existing_tuid] with mysql.db_session() as session: original_track_count = TrackQuery.get_all_by_product_id( test_product_id, session, eager_loading=False).count() response = TrackPersister.import_tracks( tuids, test_product_id, test_upc) assert response.status == response_code.NOT_FOUND with mysql.db_session() as session: track_count = TrackQuery.get_all_by_product_id( test_product_id, session, eager_loading=False).count() assert track_count == original_track_count @db.test_schema @pytest.mark.parametrize('exclude_fields', [None, ['version']]) def test_copy_product_tracks( test_product_id, test_upc, exclude_fields): """Test copy product tracks successful.""" test_source_tuid = 9 test_new_product_id = 100 with mysql.db_session() as session: original_track_count = TrackQuery.get_all_by_product_id( test_new_product_id, session, eager_loading=False).count() assert not original_track_count src_track_count = TrackQuery.get_all_by_product_id( test_product_id, session, eager_loading=False).count() response = TrackPersister.copy_product_tracks( src_product_id=test_product_id, dest_product_id=test_new_product_id, dest_upc=test_upc, exclude_fields=exclude_fields) assert response.status == response_code.OK import_source_dest_dict = response.message[api_consts.ITEMS][0] old_tuid = import_source_dest_dict['source'][tf.TUID] destination_track = import_source_dest_dict['destination'] assert test_source_tuid == old_tuid new_tuid = destination_track['tuid'] with mysql.db_session() as session: track_count = TrackQuery.get_all_by_product_id( test_new_product_id, session, eager_loading=False).count() assert track_count == original_track_count + src_track_count imported_track = TrackQuery.get_by_tuid(new_tuid, session) assert imported_track @db.test_schema_no_seed def test_copy_product_tracks_if_new_product_has_tracks( track_factory, test_upc): """Test copy product fails when the new product has tracks already.""" src_product_id = 1 dest_product_id = 2 tracks = [ track_factory(product_id=src_product_id), track_factory(product_id=dest_product_id)] db.merge_model_objects(tracks) response = TrackPersister.copy_product_tracks( src_product_id, dest_product_id, test_upc) assert response.errors['code'] == error.COPY_TRACKS_ERROR_CODE @db.test_schema @pytest.mark.parametrize('num_tracks_to_copy', [0, 1, 3]) def test_copy_product_tracks_with_track_list( test_product_id, test_upc, num_tracks_to_copy): """Test copy product tracks successful and only copies requested tracks.""" source_product_id = 1 destination_product_id = 100 with mysql.db_session() as session: original_track_count = TrackQuery.get_all_by_product_id( destination_product_id, session, eager_loading=False).count() assert not original_track_count source_track_ids = [ track.tuid for track in TrackQuery.get_all_by_product_id( source_product_id, session, eager_loading=False ) ] track_list = source_track_ids[0:num_tracks_to_copy] or None response = TrackPersister.copy_product_tracks( src_product_id=test_product_id, dest_product_id=destination_product_id, dest_upc=test_upc, track_list=track_list ) assert response.status == response_code.OK with mysql.db_session() as session: copied_tracks = TrackQuery.get_all_by_product_id( destination_product_id, session, eager_loading=False ) destination_track_tuids = [track.tuid for track in copied_tracks] if not num_tracks_to_copy: assert len(destination_track_tuids) == len(source_track_ids) else: assert len(destination_track_tuids) == num_tracks_to_copy volumes = [track.volume_number for track in copied_tracks] track_numbers = [track.track_number for track in copied_tracks] assert volumes == [1] * num_tracks_to_copy assert track_numbers == [ track_number for track_number in range(1, num_tracks_to_copy + 1) ] @db.test_schema_no_seed def test_delete_track_relationships(track_factory): """Test delete track and all relationships.""" track = track_factory( performer__count=1, writers__count=1, publishers__count=1, ownership_rights='original_owner', original_rights_holder_country_id=1) db.merge_model_objects([track]) tuid = track.tuid res = TrackPersister.delete_by_tuid(tuid, track.product_id) assert res dependent_models = [ TrackArtist, TrackWriter, TrackPublisher, TrackMasterRights, TrackProducerNationality] with mysql.db_session() as session: assert TrackQuery.get_by_tuid(tuid, session) is None for model in dependent_models: assert session.query(model).filter_by(tuid=tuid).first() is None @db.test_schema_no_seed def test_delete_track_with_spatial_record(track_factory): """Verify a track with a track_spatial row can be deleted. track_spatial has a FK on track.id; without the cascade relationship the DELETE raises a MySQL IntegrityError 1217. """ track = track_factory() db.merge_model_objects([track]) tuid = track.tuid with mysql.db_session() as session: session.add(TrackSpatial(track_id=tuid, isrc='USORCHARD00001')) session.commit() res = TrackPersister.delete_by_tuid(tuid, track.product_id) assert res with mysql.db_session() as session: assert TrackQuery.get_by_tuid(tuid, session) is None assert session.query(TrackSpatial).filter_by(track_id=tuid).first() is None @db.test_schema_no_seed def test_delete_track_not_found(): """Test not found response is returned if track doesn't exist.""" response = TrackPersister.delete_by_tuid(tuid=1234567890, product_id=12356) assert response.status == response_code.NOT_FOUND @db.test_schema_no_seed def test_delete_multiple_tracks(track_factory): """Verify tracks for release with one volume deletes/reorders correctly. The order of tracks in the database is structured in such a way """ tracks = [ track_factory(tuid=1, track_number=3), track_factory(tuid=2, track_number=2), track_factory(tuid=3, track_number=1)] db.merge_model_objects(tracks) res = TrackPersister.delete_many_by_tuids( tuids=[2], product_id=tracks[0].product_id) expected_arrangement = [ {'tuid': 3, 'volume_number': 1, 'track_number': 1}, {'tuid': 1, 'volume_number': 1, 'track_number': 2}] _assert_remaining_tracks_arrangement_after_delete( res, tracks[0].product_id, expected_arrangement) @db.test_schema_no_seed def test_delete_multiple_tracks_volume_deleted(track_factory): """Verify volumes are reordered when tracks for volume are deleted.""" tracks = [ track_factory(tuid=1, volume_number=3, track_number=1), track_factory(tuid=2, volume_number=2, track_number=1), track_factory(tuid=3, volume_number=2, track_number=2), track_factory(tuid=4, volume_number=1, track_number=1)] db.merge_model_objects(tracks) res = TrackPersister.delete_many_by_tuids( tuids=[2, 3], product_id=tracks[0].product_id) expected_arrangement = [ {'tuid': 4, 'volume_number': 1, 'track_number': 1}, {'tuid': 1, 'volume_number': 2, 'track_number': 1}] _assert_remaining_tracks_arrangement_after_delete( res, tracks[0].product_id, expected_arrangement) @db.test_schema_no_seed def test_delete_multiple_tracks_various_deleted(track_factory): """Verify tracks are reordered with edge cases.""" tracks = [ track_factory(tuid=51, volume_number=5, track_number=1), track_factory(tuid=41, volume_number=4, track_number=1), track_factory(tuid=32, volume_number=3, track_number=2), track_factory(tuid=31, volume_number=3, track_number=1), track_factory(tuid=23, volume_number=2, track_number=3), track_factory(tuid=22, volume_number=2, track_number=2), track_factory(tuid=21, volume_number=2, track_number=1), track_factory(tuid=11, volume_number=1, track_number=1)] db.merge_model_objects(tracks) res = TrackPersister.delete_many_by_tuids( tuids=[11, 22, 32, 51], product_id=tracks[0].product_id) expected_arrangement = [ {'tuid': 21, 'volume_number': 1, 'track_number': 1}, {'tuid': 23, 'volume_number': 1, 'track_number': 2}, {'tuid': 31, 'volume_number': 2, 'track_number': 1}, {'tuid': 41, 'volume_number': 3, 'track_number': 1}] _assert_remaining_tracks_arrangement_after_delete( res, tracks[0].product_id, expected_arrangement) def _assert_remaining_tracks_arrangement_after_delete( res, product_id, track_arrangements): """Assert remaining track arrangement is correct.""" assert res.message['items'] == track_arrangements # Verify changes are persisted remaining_tracks = TrackPersister.get_all_by_product_id( product_id).message['items'] track_arragements_iter = iter(track_arrangements) for track in remaining_tracks: arrangement = next(track_arragements_iter) assert track['tuid'] == arrangement['tuid'] assert track['volume_number'] == arrangement['volume_number'] assert track['track_number'] == arrangement['track_number'] @db.test_schema_no_seed def test_delete_multiple_tracks_track_not_found(track_factory): """Test delete multiple tracks when one if the tracks is not found.""" good_tuid = 1 bad_tuid = 2 tracks = [track_factory(tuid=good_tuid)] db.merge_model_objects(tracks) response = TrackPersister.delete_many_by_tuids( tuids=[good_tuid, bad_tuid], product_id=tracks[0].product_id) assert not response assert response.errors['message'] == error.ERROR_MESSAGE_TRACK_NOT_FOUND # Verify original track was not deleted track = TrackPersister.get_by_tuid(good_tuid) assert track @db.test_schema_no_seed def test_delete_multiple_tracks_duplicate(track_factory): """Verify tracks for release with one volume deletes/reorders correctly. The order of tracks in the database is structured in such a way """ res = TrackPersister.delete_many_by_tuids( tuids=[1, 2, 3, 1], product_id=1) assert not res assert res.errors['code'] == error.DUPLICATE_CODE @db.test_schema_no_seed @pytest.mark.parametrize( 'exclude_fields, expected_version', [ ([], 'Live'), (['version'], None)]) def test_import_track( track_factory, test_upc, exclude_fields, expected_version): """Test import track copies data correctly to new track.""" dest_product_id = 10 duration_mins = 1 duration_secs = 20 new_volume_number = 1 new_track_number = 1 track = track_factory( version='Live', ownership_rights='no_rights', original_rights_holder_country_id=2, duration_minutes=duration_mins, duration_seconds=duration_secs, performer__count=2, featuring__count=3, remixer__count=1, writers__count=1, publishers__count=2) db.merge_model_objects([track]) # We want the source data saved here to make sure it isn't modified src_track_dict = track.to_dict() with mysql.db_session() as session: src_track, dest_track = TrackPersister._import_track( track, dest_product_id, test_upc, new_volume_number, new_track_number, session, exclude_fields=exclude_fields) is_copy, error_msg = test_funcs.validate_track_is_copy( src_track_dict, dest_track.to_dict(), exclude_fields=exclude_fields) session.commit() assert is_copy, error_msg assert dest_track.version == expected_version assert dest_track.product_id == dest_product_id assert dest_track.upc == test_upc assert dest_track.track_number == new_track_number assert dest_track.volume_number == new_volume_number assert dest_track.duration_minutes == duration_mins assert dest_track.duration_seconds == duration_secs @db.test_schema_no_seed def test_import_track_mirrors_spatial_to_track_additional_isrc( track_factory, test_upc): """Cloning a spatial source track writes the dual-write mirror for the copy. Guards the clone path (copy-product / import-from-catalog): _import_track copies src._spatial via add_spatial, which must also write the track_additional_isrc mirror for the destination track. """ track = track_factory() track.add_spatial('US1234567890') db.merge_model_objects([track]) with mysql.db_session() as session: _, dest_track = TrackPersister._import_track( track, 10, test_upc, 1, 1, session) session.commit() mirror = session.query(TrackAdditionalIsrc).filter( TrackAdditionalIsrc.track_id == dest_track.tuid).one() assert mirror.type == TrackAdditionalIsrcType.ATMOS assert mirror.isrc == 'US1234567890' assert mirror.deleted_at is None @db.test_schema_no_seed def test_import_track_legacy_track_artist(track_factory, test_upc): """Test import track skips track_artist import for legacy artist_type.""" dest_product_id = 10 new_volume_number = 1 new_track_number = 1 track = track_factory(performer__count=1, soloist__count=1) db.merge_model_objects([track]) with mysql.db_session() as session: src_track, dest_track = TrackPersister._import_track( track, dest_product_id, test_upc, new_volume_number, new_track_number, session) session.commit() assert len(dest_track.artists) == 1 assert dest_track.artists[0].artist_type != 'soloist' @db.test_schema @pytest.mark.parametrize( 'method_causing_error', ['add_artist', 'add_writer', 'add_publisher']) def test_import_track_is_rolled_back_on_role_error( mocker, test_product_id, test_upc, method_causing_error): """Should rollback transaction if error happened while adding roles.""" mock_sentry_capture_exception = mocker.patch('backend.models.track_persister.capture_exception') test_source_tuid = 2 mocker.patch( 'backend.models.track_persister.Track.{}'.format(method_causing_error), side_effect=Exception) with mysql.db_session() as session: mock_rollback = mocker.patch.object(session, 'rollback') import_track_response = TrackPersister.import_tracks( [test_source_tuid], test_product_id, test_upc, session=session) assert mock_rollback.called assert import_track_response.status == response_code.INTERNAL_ERROR assert mock_sentry_capture_exception.called @db.test_schema @pytest.mark.parametrize( 'method_causing_error', ['add']) def test_import_track_is_rolled_back_on_session_error( mocker, test_product_id, test_upc, method_causing_error): """Should rollback transaction if error happened while add/commit.""" test_source_tuid = 2 mock_sentry_capture_exception = mocker.patch('backend.models.track_persister.capture_exception') with mysql.db_session() as session: original_method = getattr(session, method_causing_error) mocker.patch.object( session, method_causing_error, side_effect=Exception) mock_rollback = mocker.patch.object(session, 'rollback') import_track_response = TrackPersister.import_tracks( [test_source_tuid], test_product_id, test_upc, session=session) # this is needed because mysql.db_session also calls commit on exit setattr(session, method_causing_error, original_method) assert mock_rollback.called assert import_track_response.status == response_code.INTERNAL_ERROR assert mock_sentry_capture_exception.called def test_apply_track_update_with_role(track_factory): """Verify track role is updated.""" track = track_factory(tuid=1, publishers__count=2) publisher_names = ['Lap', 'Wap'] data = { tf.TUID: 1, tf.VERSION: 'red', 'publisher_names': publisher_names.copy() } result = TrackPersister._apply_track_update(track, data) assert result assert track.version == 'red' assert publisher_names == [ publisher.name for publisher in track.publishers] def test_apply_track_update_with_duplicate_role(track_factory): """Verify track role is updated.""" track = track_factory(tuid=1, publishers__count=2) publisher_names = ['Lap', 'Wap', 'Lap'] data = { tf.TUID: 1, tf.VERSION: 'red', 'publisher_names': publisher_names.copy() } result = TrackPersister._apply_track_update(track, data) assert not result assert result.errors['message'] == \ 'Duplicate data found for field `publishers`' @db.test_schema_no_seed def test_apply_track_update_invalid_field(track_factory): """Verify invalid field is not updated.""" track = track_factory(tuid=1) data = {tf.TUID: 1, tf.VOLUME_NUMBER: 2} result = TrackPersister._apply_track_update(track, data) assert not result @db.test_schema_no_seed def test_apply_track_update_bad_role(track_factory): """Verify invalid field is not updated.""" track = track_factory(tuid=1) data = {tf.TUID: 1, 'facilities_names': ['Janitor Jack', 'Plumber Bob']} result = TrackPersister._apply_track_update(track, data) assert not result @db.test_schema_no_seed @pytest.mark.parametrize('exception', [TypeError, ValueError]) def test_apply_track_update_handle_exception(track_factory, mocker, exception): """Verify validation response on TypeError when updating.""" track = track_factory(tuid=1) data = {tf.TUID: 1, tf.VERSION: 'red'} mocker.patch( 'backend.models.track.Track.update', side_effect=exception('msg')) result = TrackPersister._apply_track_update(track, data) assert not result assert result.errors['message'] == 'msg' @db.test_schema_no_seed def test_apply_track_update_sqlalchemy_integrity_error(track_factory, mocker): """Verify validation response on IntegrityError when updating.""" track = track_factory(tuid=1) data = {tf.TUID: 1, tf.VERSION: 'red'} ex = sqlalchemy.exc.IntegrityError(statement='sql', params=[], orig='orig') mocker.patch( 'backend.models.track.Track.update', side_effect=ex) result = TrackPersister._apply_track_update(track, data) assert not result assert result.errors['message'] == 'orig' @db.test_schema def test_apply_track_update_with_artist_info_ids(track_factory): """Verify track artist artist_info_id is updated.""" track = track_factory(tuid=1, publishers__count=2) data = { tf.TUID: 1, 'performer_names': ['performer 1'], 'writer_names': ['writer 1'] } artist_info_ids_to_set = { 'artists': { ('performer', 'performer 1', ): 10 }, 'writers': { ('writer 1', ): 20 } } result = TrackPersister._apply_track_update( track, data, artist_info_ids_to_set=artist_info_ids_to_set) assert result assert track.artists[0].artist_info_id == 10 assert track.writers[0].artist_info_id == 20 def test_apply_track_update_with_focus_track(track_factory): """Verify focus track is updated.""" track = track_factory(tuid=1) data = { tf.TUID: 1, tf.FOCUS_TRACK: 'Y', tf.FOCUS_TRACK_START_DATE: datetime.date.fromisoformat('2022-11-14') } result = TrackPersister._apply_track_update(track, data) assert result assert track.focus_track == 'Y' assert track.focus_track_start_date == datetime.date.fromisoformat('2022-11-14') @db.test_schema def test_apply_track_update_with_focus_track_and_user_info( track_factory): """Verify focus track user data is updated.""" track = track_factory(tuid=1) data = { tf.TUID: 1, tf.FOCUS_TRACK: 'Y', tf.FOCUS_TRACK_START_DATE: datetime.date.fromisoformat('2022-11-14'), 'user_id': 123, 'user_type': 'alw' } result = TrackPersister._apply_track_update(track, data) assert result assert track.focus_track == 'Y' assert track.focus_track_start_date == datetime.date.fromisoformat('2022-11-14') assert track._focus_track.updated_by_user_id == 123 assert track._focus_track.updated_by_user_type == 'alw' @db.test_schema def test_create_and_update_multiple_tracks_success( mocker, mock_app, track_factory): """Verify that tracks are created and saved to DB.""" track_1 = track_factory(track_name='Unique track name 1').to_dict() del track_1['tuid'] del track_1['publishers'] del track_1['product_id'] update_track = track_1.copy() product_id = 12345 upc = '123456' with mock_app: mocker.patch('backend.models.track_persister.g') rollback_spy = mocker.spy(sqlalchemy.orm.session.Session, 'rollback') update_multiple_tracks_spy = mocker.spy( TrackPersister, 'update_multiple_tracks') result = TrackPersister.create_and_update_multiple_tracks( product_id, upc, [update_track] ) assert result received_track = result.message['items'][0] del received_track['tuid'] # The language data from the language table will not be updated # since the data should only be read-only track_1['language'] = None expected_track = dict( track_1, product_id=product_id, publishers=[]) assert received_track == expected_track assert rollback_spy.call_count == 0 with mysql.db_session() as session: assert session.query(Track).filter_by( track_name='Unique track name 1').count() == 1 assert update_multiple_tracks_spy.call_args[1][ 'get_artist_info_ids_from_existing_rows'] is True assert update_multiple_tracks_spy.call_args[1][ 'update_track_volume_numbers'] is True @pytest.fixture @db.test_schema def test_create_and_update_multiple_tracks_failure(track_factory, mocker): """Verify that no tracks are saved to DB in case of failure.""" track_1 = track_factory(track_name='Unique track name 1').to_dict() track_2 = track_factory(track_name='Unique track name 2').to_dict() track_2['artists'] = [{ 'artist_name': 'Artist1', 'artist_type': 'Not existing type' }] for track in [track_1, track_2]: del track['tuid'] del track['publishers'] del track['product_id'] product_id = 12345 upc = '123456' rollback_spy = mocker.spy(sqlalchemy.orm.session.Session, 'rollback') result = TrackPersister.create_and_update_multiple_tracks( product_id, upc, [track_1.copy(), track_2.copy()]) assert result.status == response_code.BAD_REQUEST assert result.errors['code'] == error.VALIDATION_ERROR_CODE assert rollback_spy.call_count == 1 with mysql.db_session() as session: assert session.query(Track).filter_by( track_name='Unique track name 1').count() == 0 assert session.query(Track).filter_by( track_name='Unique track name 2').count() == 0 def test_format_artists_for_update_success(): """Test for successful processing.""" track_data = { 'artists': [ { 'track_artist_id': 101, 'type': 'performer', 'name': 'Performer 1', 'artist_info_id': 201 }, { 'track_artist_id': 102, 'type': 'bass', 'name': 'Bass 1', 'artist_info_id': None }, { 'track_artist_id': None, 'type': 'drums', 'name': 'Drums 1', 'artist_info_id': None } ] } result = TrackPersister._format_artists_for_update(track_data) assert result assert track_data['performer_names'] == ['Performer 1'] assert track_data['bass_names'] == ['Bass 1'] assert track_data['drums_names'] == ['Drums 1'] assert track_data['artist_info_ids_for_artists'] == [(101, 201)] def test_format_artists_for_update_track_artist_id_none(): """Test for failure on None track_artist_id and filled artist_info_id.""" track_data = { 'artists': [ { 'track_artist_id': None, 'type': 'performer', 'name': 'Performer 1', 'artist_info_id': 201 } ] } result = TrackPersister._format_artists_for_update(track_data) assert result.status == response_code.BAD_REQUEST assert result.errors['code'] == error.VALIDATION_ERROR_CODE def test_format_artists_for_update_track_bad_role(): """Test for failure on bad artist type.""" track_data = { 'artists': [ { 'track_artist_id': 101, 'type': 'BAD_TYPE', 'name': 'BAD_TYPE 1', 'artist_info_id': 201 } ] } result = TrackPersister._format_artists_for_update(track_data) assert result.status == response_code.BAD_REQUEST assert result.errors['code'] == error.VALIDATION_ERROR_CODE def test_format_writers_for_update_success(): """Test for successful processing.""" track_data = { 'writers': [ { 'track_writer_id': 101, 'type': 'writer', 'name': 'writer 1', 'artist_info_id': 201 }, { 'track_writer_id': 102, 'type': 'writer', 'name': 'Writer 2', 'artist_info_id': None }, { 'track_writer_id': None, 'type': 'writer', 'name': 'Writer 3', 'artist_info_id': None } ] } result = TrackPersister._format_writers_for_update(track_data) assert result assert track_data['writer_names'] == ['writer 1', 'Writer 2', 'Writer 3'] assert track_data['artist_info_ids_for_writers'] == [(101, 201)] def test_format_writers_for_update_track_writer_id_none(): """Test for failure on None track_writer_id and filled artist_info_id.""" track_data = { 'writers': [ { 'track_writer_id': None, 'type': 'performer', 'name': 'Performer 1', 'artist_info_id': 201 } ] } result = TrackPersister._format_writers_for_update(track_data) assert result.status == response_code.BAD_REQUEST assert result.errors['code'] == error.VALIDATION_ERROR_CODE def test_format_artists_and_writers_success(mocker): """Test for success of _format_artists_and_writers_for_update.""" update_data = { 'artists': [], 'writers': [] } mocker.patch.object( TrackPersister, '_format_artists_for_update', return_value=response.Response()) mocker.patch.object( TrackPersister, '_format_artists_for_update', return_value=response.Response()) result = TrackPersister._format_artists_and_writers_for_update( update_data) assert result def test_format_artists_and_writers_failure_of_artists(mocker): """Test for failure on artists update.""" update_data = { 'artists': [], 'writers': [] } mocker.patch.object( TrackPersister, '_format_artists_for_update', return_value=response.create_error_response('', '')) mocker.patch.object( TrackPersister, '_format_writers_for_update', return_value=response.Response()) result = TrackPersister._format_artists_and_writers_for_update( update_data) assert not result def test_format_artists_and_writers_failure_of_writers(mocker): """Test for failure on writers update.""" update_data = { 'artists': [], 'writers': [] } mocker.patch.object( TrackPersister, '_format_writers_for_update', return_value=response.create_error_response('', '')) mocker.patch.object( TrackPersister, '_format_artists_for_update', return_value=response.Response()) result = TrackPersister._format_artists_and_writers_for_update( update_data) assert not result def test_extract_track_artist_and_writer_ids(mocker): """Test of _extract_track_artist_and_writer_ids.""" update_data = { 'artists': [ { 'name': 'performer 1', 'type': 'performer', }, { 'name': 'performer 2', 'type': 'performer', 'artist_info_id': 2 }, { 'name': 'performer 3', 'type': 'performer', 'artist_info_id': 3, 'track_artist_id': 103 } ], 'writers': [ { 'name': 'writer 1', 'type': 'writer', }, { 'name': 'writer 2', 'type': 'writer', 'artist_info_id': 12 }, { 'name': 'writer 3', 'type': 'writer', 'artist_info_id': 33, 'track_writer_id': 203 } ] } result = TrackPersister._extract_track_artist_and_writer_ids(update_data) assert update_data == { 'artists': [ { 'name': 'performer 1', 'type': 'performer', }, { 'name': 'performer 2', 'type': 'performer', }, { 'name': 'performer 3', 'type': 'performer', 'artist_info_id': 3, 'track_artist_id': 103 } ], 'writers': [ { 'name': 'writer 1', 'type': 'writer', }, { 'name': 'writer 2', 'type': 'writer', }, { 'name': 'writer 3', 'type': 'writer', 'artist_info_id': 33, 'track_writer_id': 203 } ] } assert result.message == { 'artists': { ('performer', 'performer 2'): 2 }, 'writers': { ('writer 2',): 12} } @db.test_schema_no_seed def test_get_track_by_track_artist_id(track_factory): """Test getting track by tuid is successful.""" track = track_factory(performer__count=1) db.merge_model_objects([track]) expected_tuid = track.tuid response = TrackPersister.get_track_by_track_artist_id(1) assert response.status == response_code.OK assert expected_tuid == response.message['tuid'] @db.test_schema_no_seed def test_get_track_by_track_artist_id_not_found(track_factory): """Test getting track by tuid is successful.""" track = track_factory(performer__count=1) db.merge_model_objects([track]) response = TrackPersister.get_track_by_track_artist_id(1001) assert response.status == response_code.NOT_FOUND @db.test_schema def test_get_track_artist_by_track_id(): """Test track artist by track unique id.""" result = TrackPersister.get_track_artist_by_track_id(tuid=1) assert result.message == [ { 'track_artist_id': 1, 'type': 'performer', 'name': 'Snowball', 'artist_info_id': None }, { 'track_artist_id': 2, 'type': 'featuring', 'name': 'Catatomic', 'artist_info_id': None }, { 'track_artist_id': 3, 'type': 'remixer', 'name': 'Purrfect', 'artist_info_id': None } ] assert result.status == 200 @db.test_schema def test_get_track_artist_by_track_id_not_found(): """Track data should not be found.""" result = TrackPersister.get_track_artist_by_track_id(tuid=999) assert result.status == response_code.NOT_FOUND, \ 'Track should not be found' @db.test_schema def test_claim_new_isrc_success(mock_claim_new_isrcs): """Test claiming a new ISRC.""" isrc_response = TrackPersister.claim_new_isrc() assert isrc_response.status == response_code.OK assert list(isrc_response.message.keys()) == ['isrc'] assert isinstance(isrc_response.message['isrc'], str) assert len(isrc_response.message['isrc']) == 16 @db.test_schema def test_claim_new_isrc_failure(mocker): """Test claiming a new ISRC failure.""" mocker.patch.object(TrackQuery, 'claim_new_isrcs', return_value=[]) mock_send_to_sentry = mocker.patch('backend.models.track_persister.send_to_sentry') isrc_response = TrackPersister.claim_new_isrc() assert isrc_response.status == response_code.INTERNAL_ERROR assert isrc_response.errors['message'] == 'Unable to claim an ISRC.' assert mock_send_to_sentry.called @db.test_schema def test_claim_new_isrcs_success(mock_claim_new_isrcs): """Test claiming multiple new ISRCs.""" isrc_response = TrackPersister.claim_new_isrcs(3) assert isrc_response.status == response_code.OK assert 'isrcs' in isrc_response.message assert isinstance(isrc_response.message['isrcs'], list) assert len(isrc_response.message['isrcs']) == 3 for isrc in isrc_response.message['isrcs']: assert isinstance(isrc, str) assert len(isrc) == 16 @db.test_schema def test_claim_new_isrcs_zero(mock_claim_new_isrcs): """Test claiming zero ISRCs returns empty list.""" isrc_response = TrackPersister.claim_new_isrcs(0) assert isrc_response.status == response_code.OK assert isrc_response.message == {'isrcs': []} @db.test_schema def test_claim_new_isrcs_failure(mocker): """Test claiming multiple ISRCs failure.""" mocker.patch.object(TrackQuery, 'claim_new_isrcs', return_value=['only_one']) mock_send_to_sentry = mocker.patch('backend.models.track_persister.send_to_sentry') isrc_response = TrackPersister.claim_new_isrcs(3) assert isrc_response.status == response_code.INTERNAL_ERROR assert isrc_response.errors['message'] == 'Unable to claim 3 ISRCs.' assert mock_send_to_sentry.called @db.test_schema_no_seed def test_get_spatial_isrc_map_by_product_id(track_factory): """Test returns mapping of track_id to ISRC for tracks with spatial data.""" product_id = 1 track = track_factory(product_id=product_id) db.merge_model_objects([track]) db.merge_model_objects([TrackSpatial(track_id=track.tuid, isrc='US1234567890')]) result = TrackPersister.get_spatial_isrc_map_by_product_id(product_id) assert result assert result.message == {track.tuid: 'US1234567890'} @db.test_schema_no_seed def test_get_spatial_isrc_map_by_product_id_empty(track_factory): """Test returns empty dict when no spatial records exist for product.""" product_id = 1 track = track_factory(product_id=product_id) db.merge_model_objects([track]) result = TrackPersister.get_spatial_isrc_map_by_product_id(product_id) assert result assert result.message == {}