"""Test generation of .csv file and writing project data to it.""" import copy import csv import os import tempfile from flexmock import flexmock from freezegun import freeze_time from oto import response from oto import status as response_status import pytest from label_copy_export.connectors import loggly from label_copy_export.connectors import s3 from label_copy_export.connectors import sentry from label_copy_export.constants import error from label_copy_export.constants import csv as csv_constants from label_copy_export.constants import lce_history from label_copy_export.constants import ows_services from label_copy_export.logic import csv_generation from label_copy_export.models import history from label_copy_export.models import label_copy_export from label_copy_export.models import ows_artist from label_copy_export.models import ows_assets from label_copy_export.models import ows_pricing from label_copy_export.models import ows_product_physical from label_copy_export.models import ows_project_manager from label_copy_export.utils import file_utils from tests.utils import test_sqs_utils @freeze_time('2017-03-30') def test_generate_csv_filename(): """Test CSV file name is being generated.""" project_id = '12345' filename = csv_generation.generate_label_copy_export_filename(project_id) assert filename == '20170330_12345.csv' TEST_PRODUCT_DATA = { 'project_name': 'Test Project', 'project_code': '123456', 'description': 'test description', 'project_highlights': 'test highlights', 'product_name': 'Test Product', 'product_code': 'ABCD1234', 'upc': '2222888899', 'artist_is_individual': 'N', 'label': 'The Hit Factory', 'genre_name': 'Blues', 'subgenre_name': 'Piano Blues', 'version': 'Mexican Version', 'explicit': 'Y', 'product_type': 'product_type', 'sale_start_date': '2016-12-12', 'release_date': '2016-12-12', 'embargo_date': '2016-12-12', 'distribution_format_name': 'Cassette', 'format': 'Single', 'units_per_set': 5, 'value_adds': '2 x CD', 'packaging': 'Box Set', 'initial_stock': 500, 'box_lot': '30', 'wholesale_price': 9.99, 'pricing': 'Priced to move', 'discount': 'discount', 'pline': '1986 Record Label', 'cline': '1986 Record Label', 'exclusive_for': 'Somebody', 'exportable': 'Y', 'manufacturing_obligation': 'Test obligation', 'production_notes': 'Test notes', 'product_highlights': 'Test highlights', 'disc': 1, 'track_number': 2, 'track_name': 'Test Name', 'performer': 'Test Artist', 'isrc': 1234567, 'primary_artist': 'Test Artist', } TEST_PROJECT_MANAGER_MESSAGE = { 'project_name': 'Test name', 'description': 'Project description', 'project_code': 'test code', 'project_highlights': 'Test project highlight data', 'artist_id': 4 } TEST_GET_PROJECT_DATA_MESSAGE = { 'project_name': 'Test name', 'description': 'Project description', 'project_code': 'test code', 'project_highlights': 'Test project highlight data', 'project_artist': 'Adele' } TEST_GENRES_RESPONSE = [ {'id': 4, 'name': 'Genre Four'}, {'id': 1, 'name': 'Genre One'}, {'id': 2, 'name': 'Genre Two'}, {'id': 3, 'name': 'Genre Three'}, ] TEST_SUBGENRES_RESPONSE = [ {'id': 4, 'name': 'Subgenre Four'}, {'id': 1, 'name': 'Subgenre One'}, {'id': 2, 'name': 'Subgenre Two'}, {'id': 3, 'name': 'Subgenre Three'}, ] TEST_PACKAGING_OPTIONS_RESPONSE = [ {'id': 1, 'name': 'Brilliant Case (Jewel Case size, Holds 2 CDs)'}, {'id': 2, 'name': 'Blister Pack'}, {'id': 3, 'name': 'Blu-Ray Packaging'}, ] TEST_PRODUCT_TRACKS = { 'items': [ { 'track_id': 1234, 'track_number': 1, 'track_name': 'Hightail it to Chalupa-ville', 'performer': ['Dave Matthews Bando', 'Yeezus'], 'isrc': '123456789012', 'disc': 1, 'length': { 'formatted': '00:07:23', 'hours': 0, 'minutes': 7, 'seconds': 23 } }, { 'track_id': 1235, 'track_number': 2, 'track_name': 'I Met a Gordita in Tijuana', 'performer': ['Johnny Cash'], 'isrc': '123456789013', 'disc': 1, 'length': { 'formatted': '00:05:12', 'hours': 0, 'minutes': 5, 'seconds': 12 } } ] } TEST_ARTIST_RESPONSE = { 'id': 1, 'name': 'Adele', 'artist_type': 'artist' } TEST_GET_ORCHARD_PRICING_TIERS_RESPONSE = [ { 'sort_order': 1, 'name': 'Front', 'pricing_family_id': 1, 'orchard_pricing_tier_id': 1, 'is_default': False }, { 'sort_order': 2, 'name': 'Theatrical D&D', 'pricing_family_id': 1, 'orchard_pricing_tier_id': 2, 'is_default': False }, { 'sort_order': 3, 'name': 'Mid', 'pricing_family_id': 1, 'orchard_pricing_tier_id': 3, 'is_default': True } ] TEST_GET_ORCHARD_PRICING_TIERS_FOR_PRODUCT_RESPONSE = { 'product_orchard_pricing_tier_id': 1, 'orchard_pricing_tier_id': 2, 'product_id': 1, 'pricing_family_id': 2 } def test_normalize_csv_row_sequence(): """Test product row data is normalized successfully.""" normalized_product_data = csv_generation.normalize_csv_row_sequence( TEST_PRODUCT_DATA, csv_constants.CSV_HEADER_DATA_MAPPING) expected_normalized_data = { 'Release Type': 'product_type', 'ISRC': 1234567, 'Exclusive For': 'Somebody', 'Initial Stock': 500, 'Project Highlights': 'test highlights', 'Product Code': 'ABCD1234', 'Sales Date': '2016-12-12', 'Track Artist': 'Test Artist', 'Individual': 'N', 'Track Name': 'Test Name', 'Product Name': 'Test Product', 'Product Artist': 'Test Artist', 'Release Date': '2016-12-12', 'Suggested List Price': 'Priced to move', 'Imprint': 'The Hit Factory', 'Subgenre': 'Piano Blues', 'Product Highlights': 'Test highlights', 'Embargo Date': '2016-12-12', 'P-Line': '1986 Record Label', 'Wholesale Price': 9.99, 'Track No.': 2, 'Production Notes': 'Test notes', 'Genre': 'Blues', 'Project Description': 'test description', 'Units Per Set': 5, 'Disc': 1, 'Discount Policy': 'discount', 'Project Code': '123456', 'Product Type': 'Cassette', 'C-Line': '1986 Record Label', 'Box Lot': '30', 'Project Name': 'Test Project', 'Version': 'Mexican Version', 'Format': 'Single', 'Manufacturing Obligation': 'Test obligation', 'Value Adds': '2 x CD', 'Packaging': 'Box Set', 'UPC/EAN': '2222888899', 'Explicit': 'Y', 'Exportable': 'Y' } assert normalized_product_data == expected_normalized_data def test_get_project_data_if_get_project_failure_status(): """Test get_project_data_from_microservices. Test get_project_data_from_microservices if failure status in get_project. """ test_error_message = { 'text': 'project is not found' } (flexmock(ows_project_manager) .should_receive('get_project') .and_return(response.Response( message=test_error_message, status=404))).once() result = csv_generation.get_project_data_from_microservices( '12345') assert result.status == 404 assert result.message == test_error_message @pytest.mark.parametrize( 'model_function, model_module, valid_json', [('get_project_manager_response', ows_project_manager, TEST_PROJECT_MANAGER_MESSAGE)] ) def test_get_project_data_success(valid_response_with_message): """Test get_project_data_from_microservices succeeds.""" (flexmock(ows_artist) .should_receive('get_ows_artist') .and_return(response.Response(message=TEST_ARTIST_RESPONSE)) .once()) result = csv_generation.get_project_data_from_microservices( '12345') assert result.status == 200 assert result.message == TEST_GET_PROJECT_DATA_MESSAGE @pytest.mark.parametrize( 'model_function, model_module, valid_json', [('get_project_manager_response', ows_project_manager, TEST_PROJECT_MANAGER_MESSAGE)] ) def test_get_project_data_no_artist(valid_response_with_message): """Test get_project_data_from_microservices succeeds.""" test_error_message = {'error': 'artist is not found'} (flexmock(ows_artist) .should_receive('get_ows_artist') .and_return(response.create_not_found_response( message=test_error_message))) csv_generation.logger = flexmock(loggly.get_current_logger()) flexmock(csv_generation.logger).should_receive('info').once() result = csv_generation.get_project_data_from_microservices( '12345') TEST_GET_PROJECT_DATA_MESSAGE.pop('project_artist') assert result.status == 200 assert result.message == TEST_GET_PROJECT_DATA_MESSAGE @pytest.mark.parametrize( 'model_function, model_module, errors, status', [('get_product_physical_data', ows_product_physical, {'errors': 'product is not found'}, 403)] ) def test_get_data_if_get_ows_product_not_found_status( invalid_response_with_params, feature_engine): """Test get_product_data_from_microservices. Test get_product_data_from_microservices if not found status in get_product_physical_data. """ test_product_id = '12345' result = csv_generation.get_product_data_from_microservices( test_product_id) assert result.status == 403 assert not result def test_get_data_if_get_ows_image_path_failure_status( product_physical_from_ows): """Test get_product_data_from_microservices. Test get_product_data_from_microservices if failure status in get_ows_image_path. """ test_product_id = '12345' test_error_message = { 'text': 'test_error' } (flexmock(ows_product_physical) .should_receive('get_product_physical_data') .and_return(response.Response(message=product_physical_from_ows)) .once()) (flexmock(csv_generation) .should_receive('get_orchard_pricing_tier_name') .and_return(response.Response(message='Test Pricing Tier')) .times(1)) (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .and_return(response.Response( message=test_error_message, status=404))).once() (flexmock(ows_project_manager) .should_receive('get_project_genres') .and_return(response.Response(message=TEST_GENRES_RESPONSE)) .once()) (flexmock(ows_project_manager) .should_receive('get_project_subgenres') .and_return(response.Response(message=TEST_SUBGENRES_RESPONSE)) .once()) (flexmock(label_copy_export) .should_receive('get_distribution_format_name') .and_return(response.Response( message={'distribution_format_name': 'CD'})) .once()) (flexmock(ows_product_physical) .should_receive('get_packaging_options') .and_return(response.Response(message=TEST_PACKAGING_OPTIONS_RESPONSE)) .once()) (flexmock(ows_product_physical) .should_receive('get_product_physical_tracks') .and_return(response.Response(message=TEST_PRODUCT_TRACKS)) .once()) result = csv_generation.get_product_data_from_microservices( test_product_id) assert result.status == 200 assert ows_services.IMAGE_PATH not in result.message[0].keys() @pytest.mark.parametrize( 'model_function, model_module, errors, status', [ ('get_project_genres', ows_project_manager, {'errors': 'error while getting genres'}, 404), ]) def test_get_data_if_get_ows_genres_failure_status( product_physical_from_ows, invalid_response_with_params): """Test get_product_data_from_microservices. Test get_product_data_from_microservices if failure status in get_project_genres. """ test_product_id = '12345' (flexmock(ows_product_physical) .should_receive('get_product_physical_data') .and_return(response.Response(message=product_physical_from_ows)) .once()) (flexmock(csv_generation) .should_receive('get_orchard_pricing_tier_name') .and_return(response.Response(message='Test Pricing Tier')) .times(1)) (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .and_return(response.Response( message='test_path'))).once() result = csv_generation.get_product_data_from_microservices( test_product_id) assert result.status == 404 @pytest.mark.parametrize( 'model_function, model_module, errors, status', [ ('get_project_subgenres', ows_project_manager, {'errors': 'error while getting subgenres'}, 404), ]) def test_get_data_if_get_ows_subgenres_failure_status( product_physical_from_ows, invalid_response_with_params): """Test get_product_data_from_microservices. Test get_product_data_from_microservices if failure status in get_project_subgenres. """ test_product_id = '12345' (flexmock(ows_product_physical) .should_receive('get_product_physical_data') .and_return(response.Response(message=product_physical_from_ows)) .once()) (flexmock(csv_generation) .should_receive('get_orchard_pricing_tier_name') .and_return(response.Response(message='Test Pricing Tier')) .times(1)) (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .and_return(response.Response(message='test_path'))).once() (flexmock(ows_project_manager) .should_receive('get_project_genres') .and_return(response.Response(message=TEST_GENRES_RESPONSE)) .once()) result = csv_generation.get_product_data_from_microservices( test_product_id) assert result.status == 404 @pytest.mark.parametrize( 'model_function, model_module, errors, status', [ ('get_packaging_options', ows_product_physical, {'errors': 'error while packaging options'}, 404), ]) def test_get_data_if_get_packaging_options_fails( product_physical_from_ows, invalid_response_with_params): """Test get_product_data_from_microservices. Test get_product_data_from_microservices if failure status in get_packaging_options. """ test_product_id = '12345' (flexmock(ows_product_physical) .should_receive('get_product_physical_data') .and_return(response.Response(message=product_physical_from_ows)) .once()) (flexmock(csv_generation) .should_receive('get_orchard_pricing_tier_name') .and_return(response.Response(message='Test Pricing Tier')) .times(1)) (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .and_return(response.Response(message='test_path'))).once() (flexmock(ows_project_manager) .should_receive('get_project_genres') .and_return(response.Response(message=TEST_GENRES_RESPONSE)) .once()) (flexmock(ows_project_manager) .should_receive('get_project_subgenres') .and_return(response.Response(message=TEST_SUBGENRES_RESPONSE)) .once()) (flexmock(label_copy_export) .should_receive('get_distribution_format_name') .and_return(response.Response( message={'distribution_format_name': 'CD'})) .once()) result = csv_generation.get_product_data_from_microservices( test_product_id) assert result.status == 404 @pytest.mark.parametrize( 'model_function, model_module, errors, status', [ ('get_product_physical_tracks', ows_product_physical, {'errors': 'error while getting product tracks'}, 404), ]) def test_get_data_if_get_product_tracks_fails( product_physical_from_ows, invalid_response_with_params): """Test get_product_data_from_microservices. Test get_product_data_from_microservices if failure status in get_product_physical_tracks. """ test_product_id = '12345' (flexmock(ows_product_physical) .should_receive('get_product_physical_data') .and_return(response.Response(message=product_physical_from_ows)) .once()) (flexmock(csv_generation) .should_receive('get_orchard_pricing_tier_name') .and_return(response.Response(message='Test Pricing Tier')) .times(1)) (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .and_return(response.Response(message='test_path'))).once() (flexmock(ows_project_manager) .should_receive('get_project_genres') .and_return(response.Response(message=TEST_GENRES_RESPONSE)) .once()) (flexmock(ows_project_manager) .should_receive('get_project_subgenres') .and_return(response.Response(message=TEST_SUBGENRES_RESPONSE)) .once()) (flexmock(label_copy_export) .should_receive('get_distribution_format_name') .and_return(response.Response( message={'distribution_format_name': 'CD'})) .once()) (flexmock(ows_product_physical) .should_receive('get_packaging_options') .and_return(response.Response(message=TEST_PACKAGING_OPTIONS_RESPONSE)) .once()) result = csv_generation.get_product_data_from_microservices( test_product_id) assert result.status == 404 @pytest.mark.parametrize( 'model_function, model_module, errors, status', [ ('get_distribution_format_name', label_copy_export, {'errors': 'error while getting distribution format'}, 404), ]) def test_get_data_if_get_distribution_format_name_fails( product_physical_from_ows, invalid_response_with_params): """Test get_product_data_from_microservices. Test get_product_data_from_microservices if failure status in get_packaging_options. """ test_product_id = '12345' (flexmock(ows_product_physical) .should_receive('get_product_physical_data') .and_return(response.Response(message=product_physical_from_ows)) .once()) (flexmock(csv_generation) .should_receive('get_orchard_pricing_tier_name') .and_return(response.Response(message='Test Pricing Tier')) .times(1)) (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .and_return(response.Response(message='test_path'))).once() (flexmock(ows_project_manager) .should_receive('get_project_genres') .and_return(response.Response(message=TEST_GENRES_RESPONSE)) .once()) (flexmock(ows_project_manager) .should_receive('get_project_subgenres') .and_return(response.Response(message=TEST_SUBGENRES_RESPONSE)) .once()) result = csv_generation.get_product_data_from_microservices( test_product_id) assert result.status == 404 def test_get_product_data_from_microservices_if_product_without_tracks( product_physical_from_ows): """Test get_product_data adds product data to csv if no tracks.""" test_product_id = '12345' (flexmock(ows_product_physical) .should_receive('get_product_physical_data') .and_return(response.Response(message=product_physical_from_ows)) .once()) (flexmock(csv_generation) .should_receive('get_orchard_pricing_tier_name') .and_return(response.Response(message='Test Pricing Tier')) .times(1)) (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .and_return(response.Response(message='test_path'))).once() (flexmock(ows_project_manager) .should_receive('get_project_genres') .and_return(response.Response(message=TEST_GENRES_RESPONSE)) .once()) (flexmock(ows_project_manager) .should_receive('get_project_subgenres') .and_return(response.Response(message=TEST_SUBGENRES_RESPONSE)) .once()) (flexmock(label_copy_export) .should_receive('get_distribution_format_name') .and_return(response.Response( message={ 'distribution_format_name': 'Test distribution format name'})) .once()) (flexmock(ows_product_physical) .should_receive('get_packaging_options') .and_return(response.Response(message=TEST_PACKAGING_OPTIONS_RESPONSE)) .once()) test_product_no_tracks_message = {'items': []} (flexmock(ows_product_physical) .should_receive('get_product_physical_tracks') .and_return(response.Response(message=test_product_no_tracks_message)) .once()) result = csv_generation.get_product_data_from_microservices( test_product_id) assert len(result.message) == 1 assert 'track_name' not in result.message[0] assert 'product_name' in result.message[0] def test_get_product_data_from_microservices_success( product_physical_from_ows): """Test get_product_data_from_microservices succeeds.""" test_product_id = '12345' (flexmock(ows_product_physical) .should_receive('get_product_physical_data') .and_return(response.Response(message=product_physical_from_ows)) .once()) (flexmock(csv_generation) .should_receive('get_orchard_pricing_tier_name') .and_return(response.Response(message='Test Pricing Tier')) .times(1)) (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .and_return(response.Response(message='test_path'))).once() (flexmock(ows_project_manager) .should_receive('get_project_genres') .and_return(response.Response(message=TEST_GENRES_RESPONSE)) .once()) (flexmock(ows_project_manager) .should_receive('get_project_subgenres') .and_return(response.Response(message=TEST_SUBGENRES_RESPONSE)) .once()) (flexmock(label_copy_export) .should_receive('get_distribution_format_name') .and_return(response.Response( message={ 'distribution_format_name': 'Test distribution format name'})) .once()) (flexmock(ows_product_physical) .should_receive('get_packaging_options') .and_return(response.Response(message=TEST_PACKAGING_OPTIONS_RESPONSE)) .once()) (flexmock(ows_product_physical) .should_receive('get_product_physical_tracks') .and_return(response.Response(message=TEST_PRODUCT_TRACKS)) .once()) result = csv_generation.get_product_data_from_microservices( test_product_id) assert result.status == 200 assert result.message[0]['image_path'] == 'test_path' assert result.message[0]['production_notes'] == 'Test notes' assert result.message[0]['manufacturing_obligation'] == 'Test obligation' assert 'genre_name' in result.message[0] assert 'subgenre_name' in result.message[0] assert isinstance(result.message[0]['performer'], str) assert result.message[0][ 'distribution_format_name'] == 'Test distribution format name' assert result.message[0]['isrc'] != result.message[1]['isrc'] assert result.message[0]['release_status'] == 'In Progress' def test_generate_csv( product_physical_from_ows, project_products_from_ows): """Test test_generate_csv succeeds.""" test_project_id = 3780565 product_physical_from_ows.update({ 'image_path': 'test_image_path', 'genre_name': 'genre_name', 'subgenre_name': 'subgenre_name' }) test_user_id = 'oa:344556' for key in ['genre_id', 'subgenre_id', 'artist_id']: product_physical_from_ows.pop(key) csv_header_data_mapping_copy = copy.deepcopy( csv_constants.CSV_HEADER_DATA_MAPPING) (flexmock(csv_constants) .should_receive('CSV_HEADER_DATA_MAPPING') .and_return(csv_header_data_mapping_copy)) (flexmock(csv_generation) .should_receive('get_project_data_from_microservices') .and_return(response.Response( message=TEST_GET_PROJECT_DATA_MESSAGE))).once() (flexmock(ows_project_manager) .should_receive('get_products_for_project') .and_return(response.Response( message=project_products_from_ows))) (flexmock(tempfile) .should_call('NamedTemporaryFile') .with_args(mode='w', suffix='.csv', delete=False).once()) flexmock(csv).should_call('DictWriter').once() (flexmock(csv_generation) .should_receive('get_product_data_from_microservices') .with_args(1002, test_user_id) .and_return(response.Response( message=[product_physical_from_ows]))).once() product_physical_from_ows['orchard_pricing_tier'] = 'Test Tier' csv_generation_result = csv_generation.generate_csv( test_project_id, test_user_id) with open(csv_generation_result.message, newline='') as csvfile: testreader = csv.DictReader(csvfile) for row in testreader: assert row['Release Type'] == 'product_type' assert row['Embargo Date'] == '2016-12-12' assert row['Value Adds'] == '2 x CD' assert row['Production Notes'] == 'Test notes' assert row['Manufacturing Obligation'] == 'Test obligation' assert row.get('Orchard Pricing Tier') == 'Test Tier' assert os.path.isfile(csv_generation_result.message) os.remove(csv_generation_result.message) def test_generate_csv_if_no_phys_products_for_project( product_physical_from_ows, project_products_from_ows): """Test generate_csv succeeds if there are no phys products for project.""" test_project_id = 12345 (flexmock(csv_generation) .should_receive('get_project_data_from_microservices') .and_return(response.Response( message=TEST_GET_PROJECT_DATA_MESSAGE))).once() csv_header_data_mapping_copy = copy.deepcopy( csv_constants.CSV_HEADER_DATA_MAPPING) (flexmock(csv_constants) .should_receive('CSV_HEADER_DATA_MAPPING') .and_return(csv_header_data_mapping_copy)) for product in project_products_from_ows['items']: product['distribution_format']['context'] = 'digital' (flexmock(ows_project_manager) .should_receive('get_products_for_project') .and_return(response.Response( message=project_products_from_ows))).once() (flexmock(tempfile) .should_call('NamedTemporaryFile') .with_args(mode='w', suffix='.csv', delete=False).once()) flexmock(csv).should_call('DictWriter').once() csv_generation_result = csv_generation.generate_csv(test_project_id) with open(csv_generation_result.message, newline='') as csvfile: testreader = csv.DictReader(csvfile) for row in testreader: assert row['Project Name'] == 'Test name' assert os.path.isfile(csv_generation_result.message) os.remove(csv_generation_result.message) def test_generate_csv_get_project_data_from_microservices_error(): """Test test_generate_csv if get_project_data_from_microservices fails.""" test_project_id = 12345 test_error_message = {'errors': 'product is not found'} (flexmock(csv_generation) .should_receive('get_project_data_from_microservices') .and_return(response.create_not_found_response( message=test_error_message))) result = csv_generation.generate_csv(test_project_id) assert result.status == 404 assert result.errors['message'] == test_error_message def test_generate_csv_get_products_for_project_error(): """Test test_generate_csv if get_products_for_project fails.""" test_project_id = 12345 test_error_message = {'errors': 'products for project are not found'} (flexmock(csv_generation) .should_receive('get_project_data_from_microservices') .and_return(response.Response( message=TEST_PROJECT_MANAGER_MESSAGE))).once() (flexmock(ows_project_manager) .should_receive('get_products_for_project') .and_return(response.create_not_found_response( message=test_error_message))).once() result = csv_generation.generate_csv(test_project_id) assert result.status == 404 assert result.errors['message'] == test_error_message def test_generate_csv_error( project_products_from_ows): """Test test_generate_csv if csv Error occurs.""" test_project_id = 12345 (flexmock(csv_generation) .should_receive('get_project_data_from_microservices') .and_return(response.Response( message=TEST_PROJECT_MANAGER_MESSAGE))).once() csv_header_data_mapping_copy = copy.deepcopy( csv_constants.CSV_HEADER_DATA_MAPPING) (flexmock(csv_constants) .should_receive('CSV_HEADER_DATA_MAPPING') .and_return(csv_header_data_mapping_copy)) (flexmock(ows_project_manager) .should_receive('get_products_for_project') .and_return(response.Response( message=project_products_from_ows))).once() (flexmock(tempfile) .should_call('NamedTemporaryFile') .with_args(mode='w', suffix='.csv', delete=False).once()) flexmock(csv).should_receive('DictWriter').and_raise(csv.Error) (flexmock(sentry.sentry_client) .should_receive('captureException') .once()) csv_generation.logger = flexmock(loggly.get_current_logger()) flexmock(csv_generation.logger).should_receive('error').once() (flexmock(file_utils) .should_call('remove_file') .once()) result = csv_generation.generate_csv(test_project_id) file_utils.remove_file(result.errors['message']['csv_file']) assert result.status == 400 def test_generate_csv_get_product_data_from_microservices_error( project_products_from_ows): """Test generate_csv if error occurs in get data from microservices. Test generate_csv if error occurs in get_product_data_from_microservices within temp_csv_file context manager. """ test_project_id = 12345 (flexmock(csv_generation) .should_receive('get_project_data_from_microservices') .and_return(response.Response( message=TEST_PROJECT_MANAGER_MESSAGE))).once() csv_header_data_mapping_copy = copy.deepcopy( csv_constants.CSV_HEADER_DATA_MAPPING) (flexmock(csv_constants) .should_receive('CSV_HEADER_DATA_MAPPING') .and_return(csv_header_data_mapping_copy)) (flexmock(ows_project_manager) .should_receive('get_products_for_project') .and_return(response.Response( message=project_products_from_ows))).once() (flexmock(tempfile) .should_call('NamedTemporaryFile') .with_args(mode='w', suffix='.csv', delete=False).once()) flexmock(csv).should_receive('DictWriter').and_raise(csv.Error) (flexmock(sentry.sentry_client) .should_receive('captureException') .once()) csv_generation.logger = flexmock(loggly.get_current_logger()) flexmock(csv_generation.logger).should_receive('error').once() result = csv_generation.generate_csv(test_project_id) file_utils.remove_file(result.errors['message']['csv_file']) assert result.status == 400 def test_csv_processing(valid_csv_generation_triggering_message): """Test csv_processing success.""" test_project_id = 11223344 test_csv_filename = 'test.csv' orchard_user_id = 'alw:546576' message = test_sqs_utils.create_sqs_message( valid_csv_generation_triggering_message) feature_flag_user_context_attr = { 'feature_flag_user_context': { 'StringValue': orchard_user_id, 'DataType': 'String'}} message.message_attributes.update(feature_flag_user_context_attr) (flexmock(csv_generation) .should_receive('validate_sqs_message') .with_args(message) .and_return(response.Response())).once() (flexmock(message) .should_receive('feature_flag_user_context') .and_return(orchard_user_id)) (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.IN_PROGRESS) .and_return(response.Response( status=response_status.NO_CONTENT))).once() test_s3_key_name = 'date_{project_id}.csv'.format( project_id=test_project_id) (flexmock(csv_generation) .should_receive('generate_csv') .with_args(test_project_id, orchard_user_id) .and_return(response.Response( message=test_csv_filename))).once() (flexmock(csv_generation) .should_receive('generate_label_copy_export_filename') .with_args(test_project_id) .and_return(test_s3_key_name)).once() (flexmock(s3) .should_receive('upload_file_to_s3') .with_args(test_csv_filename, test_s3_key_name) .and_return(response.Response())).once() (flexmock(file_utils) .should_receive('remove_file') .with_args(test_csv_filename)).once() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.COMPLETED) .and_return(response.Response( status=response_status.NO_CONTENT))).once() result = csv_generation.csv_processing(message) assert result def test_csv_processing_generate_csv_fails( valid_csv_generation_triggering_message): """Test csv_processing generate_csv returns error.""" test_project_id = 11223344 test_csv_filename = 'test.csv' message = test_sqs_utils.create_sqs_message( valid_csv_generation_triggering_message) (flexmock(csv_generation) .should_receive('validate_sqs_message') .with_args(message) .and_return(response.Response())).once() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.IN_PROGRESS) .and_return(response.Response( status=response_status.NO_CONTENT))).once() (flexmock(csv_generation) .should_receive('generate_csv') .with_args(test_project_id, None) .and_return(response.create_error_response( code=error.ERROR_GENERATING_CSV, message={ 'exception': 'test_exception', 'csv_file': test_csv_filename}) )).once() (flexmock(file_utils) .should_receive('remove_file') .with_args(test_csv_filename) ).once() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.ERROR) .and_return(response.Response( status=response_status.NO_CONTENT))).once() result = csv_generation.csv_processing(message) assert not result def test_csv_processing_generate_csv_fails_on_collecting_data( valid_csv_generation_triggering_message): """Test csv_processing generate_csv returns error.""" test_project_id = 11223344 message = test_sqs_utils.create_sqs_message( valid_csv_generation_triggering_message) (flexmock(csv_generation) .should_receive('validate_sqs_message') .with_args(message) .and_return(response.Response())).once() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.IN_PROGRESS) .and_return(response.Response( status=response_status.NO_CONTENT))).once() (flexmock(csv_generation) .should_receive('generate_csv') .with_args(test_project_id, None) .and_return(response.create_not_found_response()) .once()) (flexmock(file_utils) .should_receive('remove_file') ).never() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.ERROR) .and_return(response.Response( status=response_status.NO_CONTENT))).once() result = csv_generation.csv_processing(message) assert not result def test_csv_processing_validation_fails( create_sqs_message, valid_csv_generation_triggering_message): """Test csv_processing validation error.""" message = create_sqs_message( valid_csv_generation_triggering_message) (flexmock(csv_generation) .should_receive('validate_sqs_message') .with_args(message) .and_return(response.create_error_response( code=error.ERROR_CODE_CSV_GENERATION, message=error.ERROR_MESSAGE_INVALID_SQS)) .once()) result = csv_generation.csv_processing(message) assert not result def test_csv_processing_upload_to_s3_failed( valid_csv_generation_triggering_message): """Test csv_processing upload_file_to_s3 failed.""" test_project_id = valid_csv_generation_triggering_message['project_id'] test_csv_filename = 'test.csv' test_s3_key_name = 'date_{project_id}.csv'.format( project_id=test_project_id) message = test_sqs_utils.create_sqs_message( valid_csv_generation_triggering_message) (flexmock(csv_generation) .should_receive('validate_sqs_message') .with_args(message) .and_return(response.Response())).once() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.IN_PROGRESS) .and_return(response.Response( status=response_status.NO_CONTENT))).once() (flexmock(csv_generation) .should_receive('generate_csv') .with_args(test_project_id, None) .and_return(response.Response( message=test_csv_filename))).once() (flexmock(csv_generation) .should_receive('generate_label_copy_export_filename') .with_args(test_project_id) .and_return(test_s3_key_name)).once() (flexmock(s3) .should_receive('upload_file_to_s3') .with_args(test_csv_filename, test_s3_key_name) .and_return(response.create_fatal_response())).once() (flexmock(file_utils) .should_receive('remove_file') .with_args(test_csv_filename)).once() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.ERROR) .and_return(response.Response( status=response_status.NO_CONTENT))).once() result = csv_generation.csv_processing(message) assert not result def test_csv_processing_generate_csv_failed( valid_csv_generation_triggering_message): """Test csv_processing generate_csv failed.""" test_project_id = valid_csv_generation_triggering_message['project_id'] response_message_in_case_of_exception = { 'exception': 'exception_args', 'csv_file': 'temp_csv_file_name'} message = test_sqs_utils.create_sqs_message( valid_csv_generation_triggering_message) (flexmock(csv_generation) .should_receive('validate_sqs_message') .with_args(message) .and_return(response.Response())).once() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.IN_PROGRESS) .and_return(response.Response( status=response_status.NO_CONTENT))).once() (flexmock(csv_generation) .should_receive('generate_csv') .with_args(test_project_id, None) .and_return( response.create_error_response( code=error.ERROR_GENERATING_CSV, message=response_message_in_case_of_exception)) ).once() (flexmock(file_utils) .should_receive('remove_file') .with_args('temp_csv_file_name')).once() (flexmock(history) .should_receive('change_status') .with_args(333, lce_history.ERROR) .and_return(response.Response( status=response_status.NO_CONTENT))).once() result = csv_generation.csv_processing(message) assert not result assert result.errors['message'] == response_message_in_case_of_exception def test__handle_csv_generation_error_response(): """Expect correct error response in case if generation fails.""" job_id = 1 error_message = 'test_error_message' csv_generation_and_uploading_result = response.create_error_response( code='test_error_code', message=error_message) (flexmock(history) .should_receive('change_status') .with_args(job_id, lce_history.ERROR) .and_return(response.Response(status=response_status.NO_CONTENT)) .once()) result = csv_generation._handle_csv_generation_error_response( job_id, csv_generation_and_uploading_result) assert not result assert result.errors['message'] == error_message assert result.errors['code'] == error.ERROR_GENERATING_CSV def test__handle_csv_generation_error_response_set_job_error_fails(): """Expect correct error response in case if set_job_error fails.""" job_id = 1 error_message = 'test_error_message' csv_generation_and_uploading_result = response.create_error_response( code='test_error_code', message=error_message) (flexmock(history) .should_receive('change_status') .with_args(job_id, lce_history.ERROR) .and_return( response.create_not_found_response(error.JOB_DOES_NOT_EXIST)) .once()) result = csv_generation._handle_csv_generation_error_response( job_id, csv_generation_and_uploading_result) assert not result assert result.errors['message'] == error.JOB_DOES_NOT_EXIST assert result.errors['code'] == 'not_found_error' def test_get_orchard_pricing_tier_name_success(): """Expect correct response in successful case.""" product_id = 1 pricing_family_id = 2 (flexmock(ows_pricing) .should_receive('get_orchard_pricing_tiers') .with_args(pricing_family_id) .and_return( response.Response(TEST_GET_ORCHARD_PRICING_TIERS_RESPONSE)) .once()) (flexmock(ows_pricing) .should_receive('get_orchard_pricing_tier_for_product') .with_args(product_id, pricing_family_id) .and_return( response.Response(TEST_GET_ORCHARD_PRICING_TIERS_FOR_PRODUCT_RESPONSE)) .once()) result = csv_generation.get_orchard_pricing_tier_name( product_id, pricing_family_id) assert result.message == 'Theatrical D&D' @pytest.mark.parametrize( ( 'get_orchard_pricing_tiers_result', 'get_orchard_pricing_tier_for_product_times_called', 'get_orchard_pricing_tier_for_product_result' ), [ ( response.create_error_response( code='test_error_code', message='test error'), 0, response.Response() ), ( response.Response(TEST_GET_ORCHARD_PRICING_TIERS_RESPONSE), 1, response.create_error_response( code='test_error_code', message='test error') ) ]) def test_get_orchard_pricing_tier_name_failure( get_orchard_pricing_tiers_result, get_orchard_pricing_tier_for_product_times_called, get_orchard_pricing_tier_for_product_result): """Expect correct error response in case of failure.""" product_id = 1 pricing_family_id = 2 (flexmock(ows_pricing) .should_receive('get_orchard_pricing_tiers') .with_args(pricing_family_id) .and_return(get_orchard_pricing_tiers_result) .once()) (flexmock(ows_pricing) .should_receive('get_orchard_pricing_tier_for_product') .with_args(product_id, pricing_family_id) .and_return(get_orchard_pricing_tier_for_product_result) .times(get_orchard_pricing_tier_for_product_times_called)) result = csv_generation.get_orchard_pricing_tier_name( product_id, pricing_family_id) assert not result assert result.errors['message'] == 'test error' def test_get_orchard_pricing_tier_name_not_found(): """Expect correct error response in case of failure.""" product_id = 1 pricing_family_id = 2 get_pricing_tier_for_product_not_existing_id_resp = { 'product_orchard_pricing_tier_id': 1, 'orchard_pricing_tier_id': 200, 'product_id': 1, 'pricing_family_id': 2 } (flexmock(ows_pricing) .should_receive('get_orchard_pricing_tiers') .with_args(pricing_family_id) .and_return(response.Response(TEST_GET_ORCHARD_PRICING_TIERS_RESPONSE)) .once()) (flexmock(ows_pricing) .should_receive('get_orchard_pricing_tier_for_product') .with_args(product_id, pricing_family_id) .and_return(response.Response( get_pricing_tier_for_product_not_existing_id_resp)) .once()) result = csv_generation.get_orchard_pricing_tier_name( product_id, pricing_family_id) assert not result assert result.status == 404