"""Test Tracks Model.""" from oto import response from ows_product_physical.connector import mysql from ows_product_physical.connector.mysql import db_session from ows_product_physical.constant import error from ows_product_physical.models import track from ows_product_physical.models import track_artist def test_update_existing_track( db_fixture, single_track_data, minimal_product_for_tracks): """Assert track and artists with given track ID is updated.""" new_track = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], single_track_data) new_track_id = new_track.message['items'][0]['track_id'] updated_track_data = [ { 'track_id': new_track_id, 'track_number': 1, 'track_name': 'My First track', 'performer': ['John Shmoe'], 'isrc': 'US123456789', 'disc': 1, 'length': '1:12:32' } ] result = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], updated_track_data) assert result.status == 201 expected_message = { 'items': [{ 'track_id': new_track_id, 'track_number': 1, 'track_name': 'My First track', 'side': None, 'song_writers': [], 'performer': ['John Shmoe'], 'isrc': 'US123456789', 'disc': 1, 'length': { 'formatted': '01:12:32', 'hours': 1, 'minutes': 12, 'seconds': 32 }, 'us_publishing_obligation': 'Composition', 'third_party_publisher': 'Y', 'publisher_names': [] }], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 1, 'total_records': 1}} assert result.message == expected_message def test_update_existing_track_with_nulls( db_with_data, single_track_data, minimal_product_for_tracks): """Assert track and artists with given track ID is updated.""" new_track = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], single_track_data) new_track_id = new_track.message['items'][0]['track_id'] updated_track_data = [ { 'track_id': new_track_id, 'isrc': None, 'disc': None, 'length': None } ] result = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], updated_track_data) assert result.status == 201 expected_message = { 'items': [{ 'track_id': new_track_id, 'track_number': 1, 'track_name': 'First track', 'side': None, 'song_writers': [], 'performer': ['John Doe'], 'isrc': None, 'disc': 1, 'length': None, 'us_publishing_obligation': 'Composition', 'third_party_publisher': 'Y', 'publisher_names': [] }], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 1, 'total_records': 1}} assert result.message == expected_message def test_update_multiple_existing_tracks( db_fixture, multiple_tracks_for_update, minimal_product_for_tracks): """Assert track and artists for multiple track IDs are updated.""" first_track, second_track, third_track = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], multiple_tracks_for_update).message['items'] updated_track_data = [ { 'track_id': first_track['track_id'], 'track_name': 'My First track', 'length': '1:12:32' }, { 'track_id': second_track['track_id'], 'performer': ['John Ron'] }, { 'track_id': third_track['track_id'], 'disc': 1 } ] result = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], updated_track_data) assert result.status == 201 expected_message = {'items': [{ 'track_id': first_track['track_id'], 'track_number': 1, 'track_name': 'My First track', 'side': None, 'song_writers': [], 'performer': ['John Doe'], 'isrc': 'US123456789', 'disc': 1, 'length': { 'formatted': '01:12:32', 'hours': 1, 'minutes': 12, 'seconds': 32 }, 'us_publishing_obligation': 'Composition', 'third_party_publisher': 'Y', 'publisher_names': [] }, { 'track_id': second_track['track_id'], 'track_number': 2, 'track_name': 'Second track', 'side': None, 'song_writers': [], 'performer': ['John Ron'], 'isrc': 'US123456789X', 'disc': 1, 'length': { 'formatted': '00:04:23', 'hours': 0, 'minutes': 4, 'seconds': 23 }, 'us_publishing_obligation': 'ControlledByYourLabel', 'third_party_publisher': 'Y', 'publisher_names': [] }, { 'track_id': third_track['track_id'], 'track_number': 3, 'track_name': 'Third Track', 'side': None, 'song_writers': [], 'performer': ['John Doe Brown'], 'isrc': 'US987654321X', 'disc': 1, 'length': { 'formatted': '00:04:23', 'hours': 0, 'minutes': 4, 'seconds': 23 }, 'us_publishing_obligation': 'PublicDomain', 'third_party_publisher': 'N', 'publisher_names': [] }], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 3, 'total_records': 3 }} assert result.message == expected_message def _get_isrc(track): return track.get('isrc') def test_update_tracks_uppercase_isrc( db_fixture, tracks_with_lowercase_isrc, minimal_product_for_tracks): """Test that a db rollback happens when track business logic fails.""" update_response = track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], tracks_with_lowercase_isrc) updated_tracks_isrcs = \ list(map(_get_isrc, update_response.message.get('items'))) original_tracks_isrcs_upcased = \ [isrc.upper() for isrc in map(_get_isrc, tracks_with_lowercase_isrc)] assert updated_tracks_isrcs == original_tracks_isrcs_upcased def test_update_track_artists(mocker): """Test track_artist model methods called on update.""" session = mocker.patch.object(mysql, 'db_session') mocker.spy(track_artist, 'delete_by_track_id') mocker.spy(track_artist, 'create') track_id = 1 artists = ['Joe', 'Blow'] track._update_track_artists(track_id, artists, session) track_artist.delete_by_track_id.assert_called_once_with(track_id, session) track_artist.create.assert_called_once_with(artists, track_id, session) def test_update_track_physical_will_write_row_if_not_there( mocker, db_with_data): """Test the track physical will write the row even if its not there.""" mocker.spy(track, '_add_track_physical') mocker.patch.object( track, '_add_track_physical', return_value=response.Response()) with db_session() as session: track_id = 100000 track._update_track_physical(track_id, 'side', session) assert track._add_track_physical.call_count == 1 def test_bulk_update_publishing_obligations( mocker, valid_put_product_publishing_obligation_data): """Test bulk_update_publishing_obligations success.""" product_id = 1 upc = 1 mocker.spy(track, '_update_track_publishers') mocker.spy(track, 'process_tracks') mocker.patch.object( track, '_update_track_publishers', return_value=response.Response()) mocker.patch.object( track, 'process_tracks', return_value=response.Response(status=201)) result = track.bulk_update_publishing_obligations( product_id, upc, valid_put_product_publishing_obligation_data) assert track._update_track_publishers.call_count == 3 track.process_tracks.assert_called_once_with( product_id, upc, valid_put_product_publishing_obligation_data) assert result def test_bulk_update_publishing_obligations_update_publishers_fails( mocker, valid_put_product_publishing_obligation_data): """Test bulk_update_publishing_obligations update publishers failure.""" product_id = 1 upc = 1 test_error_message = 'test mysql error' mocker.spy(track, '_update_track_publishers') mocker.patch.object( track, '_update_track_publishers', return_value=response.create_error_response( code=error.INTERNAL_ERROR, message=test_error_message, status=500)) result = track.bulk_update_publishing_obligations( product_id, upc, valid_put_product_publishing_obligation_data) assert track._update_track_publishers.call_count == 1 assert not result assert result.errors['message'] == test_error_message def test_bulk_update_publishing_obligations_process_tracks_fails( mocker, valid_put_product_publishing_obligation_data): """Test bulk_update_publishing_obligations process tracks failure.""" product_id = 1 upc = 1 test_error_message = 'test mysql error' mocker.spy(track, '_update_track_publishers') mocker.spy(track, 'process_tracks') mocker.patch.object( track, '_update_track_publishers', return_value=response.Response()) mocker.patch.object( track, 'process_tracks', return_value=response.create_error_response( code=error.INTERNAL_ERROR, message=test_error_message, status=500)) result = track.bulk_update_publishing_obligations( product_id, upc, valid_put_product_publishing_obligation_data) assert track._update_track_publishers.call_count == 3 track.process_tracks.assert_called_once_with( product_id, upc, valid_put_product_publishing_obligation_data) assert not result assert result.errors['message'] == test_error_message def test_bulk_update_publishing_obligations_with_test_db( db_fixture, minimal_product_for_tracks, multiple_tracks_for_update, valid_put_product_publishing_obligation_data): """Test bulk_update_publishing_obligations with real data in database.""" track.process_tracks( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], multiple_tracks_for_update) publishing_obligations_of_existing_product = [ { 'track_id': 1, 'track_number': 1, 'us_publishing_obligation': 'Composition', 'third_party_publisher': 'Y', 'publisher_names': [], 'track_name': 'First track' }, { 'track_id': 2, 'track_number': 2, 'us_publishing_obligation': 'ControlledByYourLabel', 'third_party_publisher': 'Y', 'publisher_names': [], 'track_name': 'Second track' }, { 'track_id': 3, 'track_number': 3, 'us_publishing_obligation': 'PublicDomain', 'third_party_publisher': 'N', 'publisher_names': [], 'track_name': 'Third Track' } ] publishing_obligations_for_product = track.get_publishing_obligations( minimal_product_for_tracks['product_id']) assert publishing_obligations_for_product.message['items'] == ( publishing_obligations_of_existing_product) track.bulk_update_publishing_obligations( minimal_product_for_tracks['product_id'], minimal_product_for_tracks['upc'], valid_put_product_publishing_obligation_data) updated_data = publishing_obligations_of_existing_product updated_data[1]['us_publishing_obligation'] = ( valid_put_product_publishing_obligation_data[1][ 'us_publishing_obligation']) updated_data[1]['third_party_publisher'] = ( valid_put_product_publishing_obligation_data[1][ 'third_party_publisher']) updated_publishing_obligations = track.get_publishing_obligations( minimal_product_for_tracks['product_id']) assert updated_publishing_obligations.message['items'] != ( publishing_obligations_of_existing_product)