"""Test Tracks Model.""" from ows_product_physical.connector.mysql import db_session from ows_product_physical.models import track from ows_product_physical.models import track_artist from ows_product_physical.models import track_publisher from ows_product_physical.models import track_writer from ows_product_physical.models.track_artist import TrackArtist def _get_track_artists_list(session, track_id): artists = session.query(TrackArtist) \ .filter(TrackArtist.track_id == track_id) return [artist.to_dict() for artist in artists] def test_delete_single_track( db_fixture, single_track_data, minimal_product_for_tracks): """Assert track is deleted from the database.""" first_track = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], single_track_data).message['items'][0] track_to_delete = [{ 'track_id': first_track['track_id'], 'is_deleted': 'Y' }] result = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], track_to_delete).message['items'] assert result == [] def test_delete_single_track_with_incorrect_flag( db_fixture, single_track_data, minimal_product_for_tracks): """Assert track is not deleted from the database.""" first_track = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], single_track_data).message['items'][0] track_to_delete = [{ 'track_id': first_track['track_id'], 'is_deleted': 'N' }] result = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], track_to_delete).message['items'] expected_response = [{ 'track_id': first_track['track_id'], 'track_name': 'First track', 'track_number': 1, 'side': None, 'song_writers': [], 'performer': ['John Doe'], 'isrc': 'US123456789', 'disc': 1, 'length': { 'hours': 1, 'minutes': 12, 'seconds': 32, 'formatted': '01:12:32' }, 'third_party_publisher': 'Y', 'us_publishing_obligation': 'Composition', 'publisher_names': [] }] assert result == expected_response track_to_delete = [{ 'track_id': first_track['track_id'], 'is_deleted': None }] result = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], track_to_delete).message['items'] assert result == expected_response def test_delete_and_update_multiple_track( db_fixture, multiple_tracks_data, minimal_product_for_tracks): """Assert single track is deleted from the database.""" first_track, second_track = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], multiple_tracks_data).message['items'] publishers = ['John Doe', 'Robert Smith'] with db_session() as session: track_publisher.create(publishers, first_track['track_id'], session) publishing_obligations = track._get_publishing_obligations( minimal_product_for_tracks['product_id'], session) expected_publishing_obligations = [ { 'third_party_publisher': 'Y', 'publisher_names': [ 'John Doe', 'Robert Smith' ], 'track_number': 1, 'track_name': 'First track', 'track_id': 1, 'us_publishing_obligation': 'Composition' }, { 'third_party_publisher': 'Y', 'publisher_names': [ ], 'track_number': 2, 'track_name': 'Second track', 'track_id': 2, 'us_publishing_obligation': 'ControlledByYourLabel' } ] assert publishing_obligations == expected_publishing_obligations tracks_to_update = [{ 'track_id': first_track['track_id'], 'is_deleted': 'Y' }, { 'track_id': second_track['track_id'], 'track_number': 1, 'performer': ['K Chin'] }] with db_session() as session: track_publisher.create(publishers, first_track['track_id'], session) result = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], tracks_to_update).message['items'] expected_response = [{ 'track_id': second_track['track_id'], 'track_name': 'Second track', 'track_number': 1, 'performer': ['K Chin'], 'isrc': 'US123456789X', 'side': None, 'song_writers': [], 'disc': 1, 'length': { 'hours': 0, 'minutes': 4, 'seconds': 23, 'formatted': '00:04:23' }, 'third_party_publisher': 'Y', 'us_publishing_obligation': 'ControlledByYourLabel', 'publisher_names': [] }] assert result == expected_response with db_session() as session: publishing_obligations = track._get_publishing_obligations( minimal_product_for_tracks['product_id'], session) # publishing obligations for the first track were deleted with the # track data expected_publishing_obligations = [ { 'track_number': 1, 'us_publishing_obligation': 'ControlledByYourLabel', 'track_id': 2, 'track_name': 'Second track', 'third_party_publisher': 'Y', 'publisher_names': [] } ] assert publishing_obligations == expected_publishing_obligations def test_delete_track_removes_artists( db_fixture, single_track_data, minimal_product_for_tracks): """Assert track artists are deleted when deleting tracks.""" new_track = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], single_track_data).message['items'][0] track_id = new_track['track_id'] track_to_delete = [{ 'track_id': track_id, 'is_deleted': 'Y' }] track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], track_to_delete) with db_session() as session: result = _get_track_artists_list(session, track_id) assert result == [] def test_delete_all_tracks(mocker): """Test successful deletion of all tracks.""" tracks = [{ 'track_id': 1, }, { 'track_id': 2, }] mocker.patch.object(track_artist, 'bulk_delete_by_track_id') mocker.patch.object(track_publisher, 'bulk_delete_by_track_id') mocker.patch.object(track_writer, 'bulk_delete_by_track_id') mocker.patch.object(track, '_delete_tracks_by_product_id') mocker.patch.object(track, '_get_tracks', return_value=tracks) product_id = 1 session = {} track.delete_all_tracks(product_id, session) track_artist.bulk_delete_by_track_id.assert_called_with([1, 2], session) track_publisher.bulk_delete_by_track_id.assert_called_with([1, 2], session) track_writer.bulk_delete_by_track_id.assert_called_with([1, 2], session) track._delete_tracks_by_product_id.assert_called_with(product_id, session)