"""Tests for track-level asset operations.""" import flexmock from oto import response import pytest from assets import config from assets.constants import asset_types from assets.constants import asset_legacy from assets.constants import error from assets.logic.legacy import track from assets.logic.legacy import asset_received from assets.logic.legacy import import_asset from assets.logic.legacy import asset_location from assets.models import ows_product from assets.models import ows_track from assets.models.legacy import asset_type from assets.models.legacy import asset @pytest.fixture def correct_copy_track_data(): """Correct data for copy_track_asset call.""" return { 'from_tuid': 54321, 'to_tuid': 92345, 'full_user_id': 'alw:12345', 'user_id': 'alw:12345', 'asset_type': asset_types.TYPE_IMPORT_AUDIO, 'source_upc': 54321, 'destination_upc': 12345, 'import_type': asset_legacy.IMPORT_TYPE_COPY } @pytest.fixture def copy_track_asset_mocks(): """Success mocks for copy_track_asset.""" ok_response = response.Response('ok') source_file_names_response = response.Response({ 'source_upc': 54321, 'source_filename': '54321_888_123.mp3', 'track': { 'upc': 54321, 'volume_number': 888, 'track_number': 123, }, 'product': { 'upc': 54321, 'status': 'in_content' } }) destination_file_names_response = response.Response({ 'destination_upc': 12345, 'destination_filename': '54321_888_123.wav', 'product': {'upc': 54321, 'status': 'in_progress'}, 'track': { 'upc': 54321, 'volume_number': 888, 'track_number': 123, }, }) return { '_prepare_source_track': source_file_names_response, '_prepare_destination_track': destination_file_names_response, 'import_asset_items': response.Response({'import_asset_id': 1, 'asset_type': 'MP3'}), 'direct_delivery_items': ok_response, '_get_track_path': response.Response({'track_path': ''}) } @pytest.fixture def get_track_and_product_by_track_id_mocks(): """Success mocks for _get_track_and_product_by_track_id.""" return { 'ows_product': response.Response({'status': 'label_processing'}), 'ows_track': response.Response({'product_id': 1, 'track_number': 123, 'volume_number': 888, 'original_file_name': 'name'}), } @pytest.fixture def generate_prepare_tracks_data_mocks( from_product_status='in_content', from_original_name='', to_product_status='label_processing'): """Generate ows_track, ows_product responses mocks for copy_track_asset.""" from_track_uid_response = response.Response({ 'track': {'product_id': 1, 'track_number': 123, 'upc': 12312, 'volume_number': 888, 'original_file_name': from_original_name}, 'product': {'upc': 12312, 'status': from_product_status}, }) to_track_uid_response = response.Response({ 'track': {'product_id': 2, 'tuid': 999, 'track_number': 321, 'upc': 76767, 'volume_number': 111, 'original_file_name': 'name'}, 'product': {'upc': 76767, 'status': to_product_status} }) return { 'source_track': from_track_uid_response, 'destination_track': to_track_uid_response, } @pytest.fixture def tracks_id_for_copying(): """Correct input data for preparing functions calls.""" return { 'from_tuid': 54321, 'to_tuid': 92345, } @pytest.fixture def get_track_path_inputs(): """Input parameters for _get_track_path calling.""" return { 'args': { 'product': { 'upc': 54321, 'status': 'in_content' }, 'track': { 'upc': 54321, 'volume_number': 888, 'track_number': 123 }, 'asset_type_name': asset_types.TYPE_FILE_WAV, }, 'get_path_params': { 'local_ip': 'test_ip', 'initial_folder': 'mp3/', 'folder_structure': '{upc(0,6)}/{upc(6,3)}/{upc}', 'params': {'upc': '54321'}, 'filename': '54321_888_123.mp3', 'path_structure_type': asset_types.TYPE_PATH_STRUCTURE_DIRECTORY }, } @pytest.fixture def get_track_path_mocks(): """Success mocks for _get_track_path.""" get_streaming_info = { 'local_ip': 'test_ip', 'initial_folder': 'mp3/', 'folder_structure': '{upc(0,6)}/{upc(6,3)}/{upc}', 'upc': 54321, 'filename': '54321_888_123.mp3', 'path_structure_type': asset_types.TYPE_PATH_STRUCTURE_DIRECTORY } return { 'get_asset_type_by_name': response.Response({'id': 1}), 'get_streaming_info': response.Response(get_streaming_info), 'get_path': 'path_to_file', } def test_post_asset_success_with_in_content_product( correct_copy_track_data, copy_track_asset_mocks): """Test for successful copy_track_asset call.""" copy_track_asset_mocks['_prepare_destination_track'] = response.Response({ 'destination_upc': 12345, 'destination_filename': '54321_888_123.wav', 'product': {'status': 'in_content'} }) (flexmock(track) .should_receive('_prepare_destination_track') .with_args(correct_copy_track_data['to_tuid']) .and_return(copy_track_asset_mocks['_prepare_destination_track'])) (flexmock(track) .should_receive('_prepare_source_track') .with_args(correct_copy_track_data['from_tuid']) .and_return(copy_track_asset_mocks['_prepare_source_track'])) result = track.copy_track_asset( correct_copy_track_data['from_tuid'], correct_copy_track_data['to_tuid'], correct_copy_track_data['user_id'], ) assert result.status == 400 assert result.errors['message'] == error.ERROR_MESSAGE_PRODUCT_IN_CONTENT assert result.errors['code'] == error.ERROR_CODE_WRONG_VALUE def test_post_asset_success( correct_copy_track_data, copy_track_asset_mocks, get_track_path_inputs): """Test for successful copy_track_asset call.""" (flexmock(track) .should_receive('_prepare_source_track') .with_args(correct_copy_track_data['from_tuid']) .and_return(copy_track_asset_mocks['_prepare_source_track'])) (flexmock(track) .should_receive('_prepare_destination_track') .with_args(correct_copy_track_data['to_tuid']) .and_return(copy_track_asset_mocks['_prepare_destination_track'])) (flexmock(track) .should_receive('_get_track_path') .with_args(get_track_path_inputs['args']['product'], get_track_path_inputs['args']['track'], get_track_path_inputs['args']['asset_type_name']) .and_return(copy_track_asset_mocks['_get_track_path'])) (flexmock(import_asset) .should_receive('create_import_asset_items') .with_args(correct_copy_track_data['full_user_id'], '54321_888_123.mp3', correct_copy_track_data['asset_type'], correct_copy_track_data['source_upc'], correct_copy_track_data['from_tuid'], correct_copy_track_data['import_type'], ) .and_return(copy_track_asset_mocks['import_asset_items'])) (flexmock(asset_received) .should_receive('create_direct_delivery_items') .with_args( asset_types.TYPE_FILE_WAV, 1, '', correct_copy_track_data['destination_upc'], correct_copy_track_data['to_tuid'], '54321_888_123.wav', correct_copy_track_data['import_type'], ) .and_return(copy_track_asset_mocks['direct_delivery_items'])) result = track.copy_track_asset( correct_copy_track_data['from_tuid'], correct_copy_track_data['to_tuid'], correct_copy_track_data['user_id'], ) assert result @pytest.mark.parametrize('failing_mock', [ '_prepare_source_track', 'import_asset_items', 'direct_delivery_items', '_get_track_path', '_prepare_destination_track', ]) def test_post_asset_failure( correct_copy_track_data, copy_track_asset_mocks, failing_mock, get_track_path_inputs): """Test for failure copy_track_asset call.""" copy_track_asset_mocks[failing_mock] = response.create_fatal_response( message='%s_failure' % failing_mock) (flexmock(track) .should_receive('_prepare_source_track') .with_args(correct_copy_track_data['from_tuid']) .and_return(copy_track_asset_mocks['_prepare_source_track'])) (flexmock(track) .should_receive('_prepare_destination_track') .with_args(correct_copy_track_data['to_tuid']) .and_return(copy_track_asset_mocks['_prepare_destination_track'])) (flexmock(track) .should_receive('_get_track_path') .with_args(get_track_path_inputs['args']['product'], get_track_path_inputs['args']['track'], get_track_path_inputs['args']['asset_type_name']) .and_return(copy_track_asset_mocks['_get_track_path'])) (flexmock(import_asset) .should_receive('create_import_asset_items') .with_args(correct_copy_track_data['full_user_id'], '54321_888_123.mp3', correct_copy_track_data['asset_type'], correct_copy_track_data['source_upc'], correct_copy_track_data['from_tuid'], correct_copy_track_data['import_type'], ) .and_return(copy_track_asset_mocks['import_asset_items'])) (flexmock(asset_received) .should_receive('create_direct_delivery_items') .with_args( asset_types.TYPE_FILE_WAV, 1, '', correct_copy_track_data['destination_upc'], correct_copy_track_data['to_tuid'], '54321_888_123.wav', correct_copy_track_data['import_type'], ) .and_return(copy_track_asset_mocks['direct_delivery_items'])) result = track.copy_track_asset( correct_copy_track_data['from_tuid'], correct_copy_track_data['to_tuid'], correct_copy_track_data['user_id'], ) assert not result assert result.errors['message'] == '%s_failure' % failing_mock def test_copy_track_asset_with_empty_user_id(correct_copy_track_data): """Test for failure copy_track_asset call.""" result = track.copy_track_asset( correct_copy_track_data['from_tuid'], correct_copy_track_data['to_tuid'], '', ) assert not result assert result.status == 400 assert result.errors['code'] == error.ERROR_CODE_WRONG_VALUE assert result.errors['message'] == error.ERROR_MESSAGE_USER_ID_IS_REQUIRED def test_prepare_source_track_failure(tracks_id_for_copying): """Test for failure _prepare_source_track call.""" mock_message = 'get_track_and_product_by_track_ids_failure' failing_mock = response.create_fatal_response(message=mock_message) (flexmock(track) .should_receive('_get_track_and_product_by_track_id') .with_args(tracks_id_for_copying['from_tuid']) .and_return(failing_mock) .ordered()) result = track._prepare_source_track(tracks_id_for_copying['from_tuid']) assert not result assert result.errors['message'] == mock_message def test_prepare_destination_track_success( tracks_id_for_copying, generate_prepare_tracks_data_mocks): """Test for successful _prepare_destination_track call.""" (flexmock(track) .should_receive('_get_track_and_product_by_track_id') .with_args(tracks_id_for_copying['to_tuid']) .and_return(generate_prepare_tracks_data_mocks['destination_track']) .ordered()) result = track._prepare_destination_track(tracks_id_for_copying['to_tuid']) assert result.message['destination_filename'] == '76767_999.wav' def test_prepare_source_track_success( tracks_id_for_copying, generate_prepare_tracks_data_mocks): """Test for successful _prepare_source_track call.""" (flexmock(track) .should_receive('_get_track_and_product_by_track_id') .with_args(tracks_id_for_copying['from_tuid']) .and_return(generate_prepare_tracks_data_mocks['source_track']) .ordered()) result = track._prepare_source_track(tracks_id_for_copying['from_tuid']) assert result assert result.message['source_filename'] == '12312_888_123' def test_prepare_destination_track_failure(tracks_id_for_copying): """Test for failure _prepare_destination_track call.""" mock_message = 'get_track_and_product_by_track_ids_failure' failing_mock = response.create_fatal_response(message=mock_message) (flexmock(track) .should_receive('_get_track_and_product_by_track_id') .with_args(tracks_id_for_copying['to_tuid']) .and_return(failing_mock) .ordered()) result = track._prepare_destination_track(tracks_id_for_copying['to_tuid']) assert not result assert result.errors['message'] == mock_message @pytest.mark.parametrize('failing_mock', ['ows_track', 'ows_product']) def test_get_track_and_product_by_track_id_failure( tracks_id_for_copying, get_track_and_product_by_track_id_mocks, failing_mock): """Test for failure _get_track_and_product_by_track_id call.""" get_track_and_product_by_track_id_mocks[failing_mock] = ( response.create_fatal_response(message='%s_failure' % failing_mock)) (flexmock(ows_track) .should_receive('get_track_by_id') .with_args(tracks_id_for_copying['from_tuid']) .and_return(get_track_and_product_by_track_id_mocks['ows_track'])) (flexmock(ows_product) .should_receive('get_product_by_id') .with_args(1) .and_return(get_track_and_product_by_track_id_mocks['ows_product'])) result = track._get_track_and_product_by_track_id( tracks_id_for_copying['from_tuid'] ) assert not result assert result.errors['message'] == '%s_failure' % failing_mock def test_get_track_and_product_by_track_id_success( tracks_id_for_copying, get_track_and_product_by_track_id_mocks): """Test for successful _get_track_and_product_by_track_id call.""" (flexmock(ows_track) .should_receive('get_track_by_id') .with_args(tracks_id_for_copying['from_tuid']) .and_return(get_track_and_product_by_track_id_mocks['ows_track'])) (flexmock(ows_product) .should_receive('get_product_by_id') .with_args(1) .and_return(get_track_and_product_by_track_id_mocks['ows_product'])) result = track._get_track_and_product_by_track_id( tracks_id_for_copying['from_tuid'] ) assert result def test_get_track_path_success( get_track_path_inputs, get_track_path_mocks): """Test for successful _get_track_path call.""" (flexmock(asset_type) .should_receive('get_asset_type_by_name') .with_args(get_track_path_inputs['args']['asset_type_name']) .and_return(get_track_path_mocks['get_asset_type_by_name'])) (flexmock(asset) .should_receive('get_streaming_info') .with_args( get_track_path_inputs['args']['product']['upc'], 1, config.SERVER_LOCATION_AVL, '54321_888_123.%') .and_return(get_track_path_mocks['get_streaming_info'])) (flexmock(asset_location) .should_receive('get_path') .with_args( get_track_path_inputs['get_path_params']['local_ip'], get_track_path_inputs['get_path_params']['initial_folder'], get_track_path_inputs['get_path_params']['folder_structure'], get_track_path_inputs['get_path_params']['params'], get_track_path_inputs['get_path_params']['filename'], get_track_path_inputs['get_path_params']['path_structure_type']) .and_return(get_track_path_mocks['get_path'])) result = track._get_track_path(**get_track_path_inputs['args']) assert result assert result.message['track_path'] == 'path_to_file' @pytest.mark.parametrize('failing_mock', [ 'get_asset_type_by_name', 'get_streaming_info']) def test_get_track_path_failure( get_track_path_inputs, get_track_path_mocks, failing_mock): """Test for failure _get_track_path call.""" get_track_path_mocks[failing_mock] = ( response.create_fatal_response(message='%s_failure' % failing_mock)) (flexmock(asset_type) .should_receive('get_asset_type_by_name') .with_args(get_track_path_inputs['args']['asset_type_name']) .and_return(get_track_path_mocks['get_asset_type_by_name'])) (flexmock(asset) .should_receive('get_streaming_info') .with_args( get_track_path_inputs['args']['product']['upc'], 1, config.SERVER_LOCATION_AVL, '54321_888_123.%') .and_return(get_track_path_mocks['get_streaming_info'])) (flexmock(asset_location) .should_receive('get_path') .with_args( get_track_path_inputs['get_path_params']['local_ip'], get_track_path_inputs['get_path_params']['initial_folder'], get_track_path_inputs['get_path_params']['folder_structure'], get_track_path_inputs['get_path_params']['params'], get_track_path_inputs['get_path_params']['filename'], get_track_path_inputs['get_path_params']['path_structure_type']) .and_return(get_track_path_mocks['get_path'])) result = track._get_track_path(**get_track_path_inputs['args']) assert not result assert result.errors['message'] == '%s_failure' % failing_mock @pytest.fixture def correct_track_mapping(): """Correct track mapping data.""" return { 101: { 'to_tuid': 201, 'volume': 1, 'number': 1 }, 102: { 'to_tuid': 202, 'volume': 1, 'number': 2 }, 103: { 'to_tuid': 203, 'volume': 1, 'number': 3 } } @pytest.fixture def copy_product_track_assets_mocks(): """Return mocks for copy_product_track_assets.""" return { 'create_import_asset_items': response.Response( message={'import_asset_id': 321}), 'create_direct_delivery_items': response.Response(), '_get_track_path': response.Response( message={'track_path': 'uploaded_path'}) } def test_copy_product_track_assets_batch_success( correct_track_mapping, copy_product_track_assets_mocks): """Test for success of copy_product_track_assets.""" (flexmock(import_asset) .should_receive( 'create_import_asset_items') .and_return( copy_product_track_assets_mocks['create_import_asset_items']) .times(3)) (flexmock(asset_received) .should_receive( 'create_direct_delivery_items') .and_return( copy_product_track_assets_mocks['create_direct_delivery_items']) .times(3)) (flexmock(track) .should_receive( '_get_track_path') .and_return( copy_product_track_assets_mocks['_get_track_path']) .times(3)) result = track.copy_product_track_assets( correct_track_mapping, {'upc': 1}, {'upc': 2}, 'alw:123') assert result def test_copy_product_track_assets_success( correct_track_mapping, copy_product_track_assets_mocks): """Test for success of copy_product_track_assets.""" correct_track_mapping = { 101: correct_track_mapping[101] } asset_batch_data = {'import_asset_batch_id': 12345} (flexmock(import_asset) .should_receive( 'create_import_asset_items') .with_args( full_user_id='alw:123', filename='10001_101.wav', asset_type='audio', track_unique_id=101, upc=10001, import_type='copy', import_asset_batch_body=asset_batch_data) .and_return( copy_product_track_assets_mocks['create_import_asset_items'])) (flexmock(asset_received) .should_receive( 'create_direct_delivery_items') .with_args( asset_name='WAV', import_asset_id=321, uploaded_asset_filename='uploaded_path', upc=20001, track_unique_id=201, destination_name='20001_201.wav', job_type='copy') .and_return( copy_product_track_assets_mocks['create_direct_delivery_items'])) (flexmock(track) .should_receive( '_get_track_path') .with_args( {'upc': 10001}, {'tuid': 101, 'volume_number': 1, 'track_number': 1}, 'WAV' ) .and_return( copy_product_track_assets_mocks['_get_track_path'])) result = track.copy_product_track_assets( correct_track_mapping, {'upc': 10001}, {'upc': 20001}, 'alw:123', asset_batch_data=asset_batch_data ) assert result @pytest.mark.parametrize('failing_mock', [ 'create_import_asset_items', 'create_direct_delivery_items', '_get_track_path' ]) def test_copy_product_track_assets_failure( correct_track_mapping, copy_product_track_assets_mocks, failing_mock): """Test for failure of copy_product_track_assets.""" fatal_response = response.create_fatal_response( message='%s_failure' % failing_mock ) copy_product_track_assets_mocks[failing_mock] = fatal_response (flexmock(import_asset) .should_receive( 'create_import_asset_items') .and_return( copy_product_track_assets_mocks['create_import_asset_items'])) (flexmock(asset_received) .should_receive( 'create_direct_delivery_items') .and_return( copy_product_track_assets_mocks['create_direct_delivery_items'])) (flexmock(track) .should_receive( '_get_track_path') .and_return( copy_product_track_assets_mocks['_get_track_path'])) result = track.copy_product_track_assets( correct_track_mapping, {'upc': 1}, {'upc': 2}, 'alw:123') assert not result assert result.errors['message'] == '%s_failure' % failing_mock @pytest.fixture def get_image_path_inputs(): """Input parameters for _get_image_path calling.""" return { 'args': { 'upc': 54321, 'asset_type_name': asset_types.TYPE_FILE_TIF, }, 'get_path_params': { 'local_ip': 'test_ip', 'initial_folder': 'tif123/', 'path_structure': '{upc(0,6)}/{upc(6,3)}/{upc}.tif', 'params': {'upc': '54321'}, 'filename': '54321.tif', 'path_structure_type': asset_types.TYPE_PATH_STRUCTURE_FILE }, } @pytest.fixture def get_image_path_mocks(): """Success mocks for _get_image_path.""" get_streaming_info = { 'local_ip': 'test_ip', 'initial_folder': 'tif123/', 'folder_structure': '{upc(0,6)}/{upc(6,3)}/{upc}.tif', 'upc': 54321, 'filename': '54321.tif', 'path_structure_type': asset_types.TYPE_PATH_STRUCTURE_FILE } return { 'get_asset_type_by_name': response.Response({'id': 1}), 'get_streaming_info': response.Response(get_streaming_info), 'get_path': 'path_to_file', } def test_get_image_path_success( get_image_path_inputs, get_image_path_mocks): """Test for successful _get_image_path call.""" (flexmock(asset_type) .should_receive('get_asset_type_by_name') .with_args(get_image_path_inputs['args']['asset_type_name']) .and_return(get_image_path_mocks['get_asset_type_by_name'])) (flexmock(asset) .should_receive('get_streaming_info') .with_args( get_image_path_inputs['args']['upc'], 1, config.SERVER_LOCATION_AVL, '54321.tif', True) .and_return(get_image_path_mocks['get_streaming_info'])) (flexmock(asset_location) .should_receive('get_path') .with_args( get_image_path_inputs['get_path_params']['local_ip'], get_image_path_inputs['get_path_params']['initial_folder'], get_image_path_inputs['get_path_params']['path_structure'], get_image_path_inputs['get_path_params']['params'], get_image_path_inputs['get_path_params']['filename'], get_image_path_inputs['get_path_params']['path_structure_type']) .and_return(get_image_path_mocks['get_path'])) result = track._get_image_path(**get_image_path_inputs['args']) assert result assert result.message['image_path'] == 'path_to_file' @pytest.mark.parametrize('failing_mock', [ 'get_asset_type_by_name', 'get_streaming_info']) def test_get_image_path_failure( get_image_path_inputs, get_image_path_mocks, failing_mock): """Test for successful _get_image_path call.""" get_image_path_mocks[failing_mock] = ( response.create_fatal_response(message='%s_failure' % failing_mock)) (flexmock(asset_type) .should_receive('get_asset_type_by_name') .with_args(get_image_path_inputs['args']['asset_type_name']) .and_return(get_image_path_mocks['get_asset_type_by_name'])) (flexmock(asset) .should_receive('get_streaming_info') .with_args( get_image_path_inputs['args']['upc'], 1, config.SERVER_LOCATION_AVL, '54321.tif', True) .and_return(get_image_path_mocks['get_streaming_info'])) (flexmock(asset_location) .should_receive('get_path') .with_args( get_image_path_inputs['get_path_params']['local_ip'], get_image_path_inputs['get_path_params']['initial_folder'], get_image_path_inputs['get_path_params']['path_structure'], get_image_path_inputs['get_path_params']['params'], get_image_path_inputs['get_path_params']['filename'], get_image_path_inputs['get_path_params']['path_structure_type']) .and_return(get_image_path_mocks['get_path'])) result = track._get_image_path(**get_image_path_inputs['args']) assert not result assert result.errors['message'] == '%s_failure' % failing_mock @pytest.fixture def correct_copy_image_data(): """Correct data for copy_product_image_asset call.""" return { 'from_upc': 54321, 'to_upc': 20001, 'full_user_id': 'alw:32123', 'asset_type': asset_types.TYPE_FILE_TIF, 'import_type': asset_legacy.IMPORT_TYPE_COPY } @pytest.fixture def copy_image_asset_mocks(): """Success mocks for copy_product_image_asset.""" ok_response = response.Response('ok') return { 'import_asset_items': response.Response({'import_asset_id': 1, 'asset_type': 'TIF'}), 'direct_delivery_items': ok_response, '_get_image_path': response.Response({'image_path': ''}) } def test_copy_product_image_asset_success( correct_copy_image_data, copy_image_asset_mocks, get_image_path_inputs): """Test for successful copy_product_image_asset call.""" (flexmock(track) .should_receive('_get_image_path') .with_args( get_image_path_inputs['args']['upc'], get_image_path_inputs['args']['asset_type_name']) .and_return(copy_image_asset_mocks['_get_image_path'])) (flexmock(import_asset) .should_receive('create_import_asset_items') .with_args( correct_copy_image_data['full_user_id'], '54321.tif', 'image', correct_copy_image_data['from_upc'], 0, correct_copy_image_data['import_type'], {'import_asset_batch_id': 111}) .and_return(copy_image_asset_mocks['import_asset_items'])) (flexmock(asset_received) .should_receive('create_direct_delivery_items') .with_args( 'TIF', 1, '', correct_copy_image_data['to_upc'], 0, '20001.tif', correct_copy_image_data['import_type'],) .and_return(copy_image_asset_mocks['direct_delivery_items'])) result = track.copy_product_image_asset( correct_copy_image_data['from_upc'], correct_copy_image_data['to_upc'], correct_copy_image_data['full_user_id'], asset_batch_data={'import_asset_batch_id': 111} ) assert result @pytest.mark.parametrize('failing_mock', [ '_get_image_path', 'import_asset_items', 'direct_delivery_items' ]) def test_copy_image_asset_failure( copy_image_asset_mocks, correct_copy_image_data, failing_mock): """Test copy_product_image_asset failure.""" copy_image_asset_mocks[failing_mock] = response.create_fatal_response( message='%s_failure' % failing_mock) (flexmock(import_asset) .should_receive('create_import_asset_items') .and_return(copy_image_asset_mocks['import_asset_items'])) (flexmock(track) .should_receive('_get_image_path') .and_return(copy_image_asset_mocks['_get_image_path'])) (flexmock(asset_received) .should_receive('create_direct_delivery_items') .and_return(copy_image_asset_mocks['direct_delivery_items'])) result = track.copy_product_image_asset( correct_copy_image_data['from_upc'], correct_copy_image_data['to_upc'], correct_copy_image_data['full_user_id'], asset_batch_data={'import_asset_batch_id': 111} ) assert not result assert result.errors['message'] == '%s_failure' % failing_mock