"""Lambda test module.""" import csv import collections import datetime from io import StringIO import tempfile from unittest.mock import patch from unittest.mock import Mock import pytest from sqlalchemy import exc from sqlalchemy.exc import SQLAlchemyError import index import lambda_exceptions from constants import errors from constants import general from constants import sales_file_structure @pytest.fixture def event_data(): """Fixture for the incoming event.""" return { 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket' }, 'additional_info': {} } @pytest.fixture() def phf_publishing_escrow(): """Test PhfPublishingEscrow to_dict.""" return { 'phf_transaction_id': 1, 'transaction_type': None, 'track_id': '1', 'store': 'Amazon', 'qty': 1, 'royalty_rate_calculated': 1, 'royalty_calculated': 1, 'royalty_rate': None, 'royalty': None, 'gross_revenue': 1, 'net_revenue': 1, 'dist_fee': 1, 'sales_file_name': 'test_name', 'period_id': 1, 'usage_type': None, 'ownership': 1, 'admin_fee': None, 'active': 'Y', 'last_modified': datetime.datetime(2018, 2, 8, 13, 21, 11) } @patch('os.remove') @patch('status.send_failure_status') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_partially_processed( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, send_failure_status_mock, remove_mock, event_data, periods_data): """Test handler function.""" file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) csvfile = StringIO() fieldnames = ['first_name', 'second_name'] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) open_mock.return_value.__enter__.return_value = csvfile identify_company_by_file_headers_mock.return_value = 'Test Company' extract_file_headers_mock.return_value = ','.join(fieldnames) result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) line_number = 3 process_sales_data_record_mock.assert_called_with( collections.OrderedDict( [('first_name', 'One One'), ('second_name', 'One Two')]), 'Test Company', 'test_key', periods_data, line_number ) expected_error_description = { 'partial_processing_errors': { 'not enough values to unpack (expected 2, got 0)': [2, 3]}} send_failure_status_mock.assert_called_with( general.LAMBDA_NAME, 201, event_data['s3_object']['key'], errors.FILE_PARTIALLY_PROCESSED_ERROR_CODE, expected_error_description, event_data['s3_object']['bucket'] ) remove_mock.assert_called_with(file_path) assert result == { 'function': 'phf-sales-data-process', 'status': 201, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'error_code': 'file_partially_processed', 'error_description': expected_error_description, 'additional_info': {'start_index': 200000, 'line_number': 3}} @patch('os.remove') @patch('status.send_failure_status') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_success( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, send_failure_status_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function.""" file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data csvfile = StringIO() fieldnames = ['first_name', 'second_name'] extract_file_headers_mock.return_value = ','.join(fieldnames) writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) open_mock.return_value.__enter__.return_value = csvfile phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.return_value = 'Test Company' bulk_insert_new_phf_metadata_mock.return_value = 200 process_sales_data_record_mock.return_value = ( {'track_id': 'PHF1'}, {'escrow': 'data'}) result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) line_number = 2 process_sales_data_record_mock.assert_called_with( collections.OrderedDict( [('first_name', 'One One'), ('second_name', 'One Two')]), 'Test Company', 'test_key', periods_data, line_number ) send_failure_status_mock.assert_not_called() remove_mock.assert_called_with(file_path) bulk_insert_new_phf_metadata_mock.assert_called_once_with( {'PHF1': {'track_id': 'PHF1'}}, [{'escrow': 'data'}]) assert result == { 'function': 'phf-sales-data-process', 'status': 201, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'additional_info': {'start_index': 200000, 'line_number': 2}} @patch('os.remove') @patch('status.send_failure_status') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_integrity_error( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, send_failure_status_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function restarts on integrity error.""" file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data csvfile = StringIO() fieldnames = ['first_name', 'second_name'] extract_file_headers_mock.return_value = ','.join(fieldnames) writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) open_mock.return_value.__enter__.return_value = csvfile phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.return_value = 'Test Company' bulk_insert_new_phf_metadata_mock.side_effect = exc.IntegrityError( '', '', '') process_sales_data_record_mock.return_value = ( {'track_id': 'PHF1'}, {'escrow': 'data'}) result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) line_number = 2 process_sales_data_record_mock.assert_called_with( collections.OrderedDict( [('first_name', 'One One'), ('second_name', 'One Two')]), 'Test Company', 'test_key', periods_data, line_number ) send_failure_status_mock.assert_not_called() remove_mock.assert_called_with(file_path) bulk_insert_new_phf_metadata_mock.assert_called_once_with( {'PHF1': {'track_id': 'PHF1'}}, [{'escrow': 'data'}]) assert result == { 'function': 'phf-sales-data-process', 'status': 200, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'additional_info': { 'start_index': 0, 'line_number': 1, 'retries_count': 1}} @patch('os.remove') @patch('status.send_failure_status') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_integrity_error_retries_limit_exceeded( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, send_failure_status_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler fails if retries count is exceeded.""" event_data['additional_info']['retries_count'] = 3 file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data csvfile = StringIO() fieldnames = ['first_name', 'second_name'] extract_file_headers_mock.return_value = ','.join(fieldnames) writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) open_mock.return_value.__enter__.return_value = csvfile phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.return_value = 'Test Company' bulk_insert_new_phf_metadata_mock.side_effect = exc.IntegrityError( 'mock', 'mock', 'mock') process_sales_data_record_mock.return_value = ( {'track_id': 'PHF1'}, {'escrow': 'data'}) result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) line_number = 2 process_sales_data_record_mock.assert_called_with( collections.OrderedDict( [('first_name', 'One One'), ('second_name', 'One Two')]), 'Test Company', 'test_key', periods_data, line_number ) send_failure_status_mock.assert_called_once() remove_mock.assert_called_with(file_path) bulk_insert_new_phf_metadata_mock.assert_called_once_with( {'PHF1': {'track_id': 'PHF1'}}, [{'escrow': 'data'}]) assert result == { 'function': 'phf-sales-data-process', 'status': 500, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'error_code': 'file_processing_error', 'error_description': [ "(builtins.str) mock [SQL: 'mock'] [parameters: ""'mock']"]} @patch('os.remove') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') def test_handler_empty_content( guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function.""" guess_file_segment_encoding_mock.return_value = 'utf-8' file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data csvfile = StringIO() fieldnames = ['first_name', 'second_name'] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() csvfile.seek(0) open_mock.return_value.__enter__.return_value = csvfile identify_company_by_file_headers_mock.return_value = 'Test Company' result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) remove_mock.assert_called_with(file_path) assert result == { 'function': 'phf-sales-data-process', 'status': 500, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'error_code': 'missing_data', 'error_description': [( '(ContentIsMissing(...), \'There is no content in the ' 'provided sales file.\')')]} @patch('constants.sales_file_structure.ROWS_TO_PROCESS_PER_ITERATION', 1) @patch('os.remove') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_not_last_iteration( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function.""" file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data csvfile = StringIO() fieldnames = ['first_name', 'second_name'] extract_file_headers_mock.return_value = ','.join(fieldnames) writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) open_mock.return_value.__enter__.return_value = csvfile phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.return_value = 'Test Company' bulk_insert_new_phf_metadata_mock.return_value = 200 process_sales_data_record_mock.return_value = ( {'track_id': 'PHF1'}, {'escrow': 'data'}) result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) line_number = 2 process_sales_data_record_mock.assert_called_with( collections.OrderedDict( [('first_name', 'One One'), ('second_name', 'One Two')]), 'Test Company', 'test_key', periods_data, line_number ) remove_mock.assert_called_with(file_path) bulk_insert_new_phf_metadata_mock.assert_called_once_with( {'PHF1': {'track_id': 'PHF1'}}, [{'escrow': 'data'}]) assert result == { 'function': 'phf-sales-data-process', 'status': 200, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'additional_info': {'start_index': 1, 'line_number': 2}} @patch('os.remove') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') def test_handler_wrong_separator_removed( download_object_mock, set_inactive_mock, get_all_periods_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function.""" tmp = tempfile.NamedTemporaryFile(mode='w', delete=False) file_path = tmp.name with open(file_path, mode='w+') as f: f.write('first_name,second_name\r\nOne One,\nOne Two') f.seek(0) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.return_value = 'Test Company' bulk_insert_new_phf_metadata_mock.return_value = 200 process_sales_data_record_mock.return_value = ( {'track_id': 'PHF1'}, {'escrow': 'data'}) result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], '/tmp/test_key') set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) line_number = 3 process_sales_data_record_mock.assert_called_with( collections.OrderedDict( [('first_name', 'One One'), ('second_name', 'One Two')]), 'Test Company', 'test_key', periods_data, line_number ) remove_mock.assert_called_with(file_path) bulk_insert_new_phf_metadata_mock.assert_called_once_with( {'PHF1': {'track_id': 'PHF1'}}, [{'escrow': 'data'}]) assert result == { 'function': 'phf-sales-data-process', 'status': 201, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'additional_info': {'start_index': 200000, 'line_number': 3}} def test_collect_errors(): """Test collect_errors function.""" collected_errors = {} line_number = 100 error_headers = lambda_exceptions.ContentDoesNotMatchHeaders() result = index.collect_errors(collected_errors, error_headers, line_number) assert result == { 'content_does_not_match_headers': {'line_numbers': [100]}} error_missing_fields = lambda_exceptions.MandatoryFieldsMissing( missing_fields=['field_one', 'field_two']) line_number = 101 result = index.collect_errors( collected_errors, error_missing_fields, line_number) assert result == { 'content_does_not_match_headers': {'line_numbers': [100]}, 'missing_mandatory_fields': {101: ['field_one', 'field_two']}} @patch('constants.publishing.DEFAULT_BATCH_SIZE', 1) @patch('constants.sales_file_structure.ROWS_TO_PROCESS_PER_ITERATION', 40) @patch('os.remove') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_time_limit_exceeded( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function.""" file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data csvfile = StringIO() fieldnames = ['first_name', 'second_name'] extract_file_headers_mock.return_value = ','.join(fieldnames) writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) open_mock.return_value.__enter__.return_value = csvfile phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.return_value = 'Test Company' bulk_insert_new_phf_metadata_mock.return_value = 200 process_sales_data_record_mock.return_value = ( {'track_id': 'PHF1'}, {'escrow': 'data'}) context = Mock() context.get_remaining_time_in_millis.return_value = 100 result = index.handler(event_data, context) context.get_remaining_time_in_millis.assert_called_once() download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) line_number = 2 process_sales_data_record_mock.assert_called_with( collections.OrderedDict( [('first_name', 'One One'), ('second_name', 'One Two')]), 'Test Company', 'test_key', periods_data, line_number ) remove_mock.assert_called_with(file_path) bulk_insert_new_phf_metadata_mock.assert_called_once_with( {'PHF1': {'track_id': 'PHF1'}}, [{'escrow': 'data'}]) assert result == { 'function': 'phf-sales-data-process', 'status': 200, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'additional_info': {'start_index': 1, 'line_number': 2}} @patch('os.remove') @patch('status.send_failure_status') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_database_issue( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, send_failure_status_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function.""" file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data csvfile = StringIO() fieldnames = ['first_name', 'second_name'] extract_file_headers_mock.return_value = ','.join(fieldnames) writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) open_mock.return_value.__enter__.return_value = csvfile phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.return_value = 'Test Company' bulk_insert_new_phf_metadata_mock.side_effect = SQLAlchemyError( 'Something exploded') process_sales_data_record_mock.return_value = ( {'track_id': 'PHF1'}, {'escrow': 'data'}) result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) line_number = 2 process_sales_data_record_mock.assert_called_with( collections.OrderedDict( [('first_name', 'One One'), ('second_name', 'One Two')]), 'Test Company', 'test_key', periods_data, line_number ) send_failure_status_mock.assert_not_called() remove_mock.assert_called_with(file_path) bulk_insert_new_phf_metadata_mock.assert_called_once_with( {'PHF1': {'track_id': 'PHF1'}}, [{'escrow': 'data'}]) assert result == { 'function': 'phf-sales-data-process', 'status': 500, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'error_code': 'database_error', 'error_description': 'Something exploded'} @patch('os.remove') @patch('status.send_failure_status') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_null_byte_faced( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, send_failure_status_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function returns encoding error for null byte file.""" file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.return_value = periods_data csvfile = StringIO() fieldnames = ['first_name', 'second_name'] extract_file_headers_mock.return_value = ','.join(fieldnames) writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) open_mock.return_value.__enter__.return_value = csvfile phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.side_effect = csv.Error result = index.handler(event_data, None) download_object_mock.assert_called_with( event_data['s3_object']['bucket'], event_data['s3_object']['key'], file_path) set_inactive_mock.assert_called_with( event_data['s3_object']['key'] ) get_all_periods_mock.assert_called_once() identify_company_by_file_headers_mock.assert_called_with( set(['first_name', 'second_name'])) process_sales_data_record_mock.assert_not_called() send_failure_status_mock.assert_called_once() remove_mock.assert_called_with(file_path) bulk_insert_new_phf_metadata_mock.assert_not_called() assert result == { 'function': 'phf-sales-data-process', 'status': 500, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'error_code': 'encoding_not_identified', 'error_description': [ '(FileEncodingNotIdentified(...),' ' \'Unexpected encoding is used in the sales file.\')']} @patch('os.remove') @patch('status.send_failure_status') @patch('file_processing.process_sales_data_record') @patch('file_processing.identify_company_by_file_headers') @patch('file_processing.bulk_insert_new_phf_metadata') @patch('builtins.open') @patch('models.period.get_all_periods') @patch( 'models.phf_publishing_escrow' '.set_all_publishing_escrows_from_file_inactive') @patch('s3.download_object') @patch('file_processing.guess_file_segment_encoding') @patch('file_processing.extract_file_headers') def test_handler_database_error_before_processing_start( extract_file_headers_mock, guess_file_segment_encoding_mock, download_object_mock, set_inactive_mock, get_all_periods_mock, open_mock, bulk_insert_new_phf_metadata_mock, identify_company_by_file_headers_mock, process_sales_data_record_mock, send_failure_status_mock, remove_mock, phf_publishing_escrow, event_data, periods_data): """Test handler function.""" file_path = '/tmp/{}'.format(event_data['s3_object']['key']) download_object_mock.return_value = file_path get_all_periods_mock.side_effect = SQLAlchemyError( 'Something exploded') csvfile = StringIO() fieldnames = ['first_name', 'second_name'] extract_file_headers_mock.return_value = ','.join(fieldnames) writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerow({'first_name': 'One One', 'second_name': 'One Two'}) csvfile.seek(0) guess_file_segment_encoding_mock.return_value = ( sales_file_structure.ENCODING_WINDOWS_1252) open_mock.return_value.__enter__.return_value = csvfile phf_publishing_escrow_res = list() phf_publishing_escrow_res.append(phf_publishing_escrow) identify_company_by_file_headers_mock.return_value = 'Test Company' bulk_insert_new_phf_metadata_mock.return_value = 200 process_sales_data_record_mock.return_value = ( {'track_id': 'PHF1'}, {'escrow'}) result = index.handler(event_data, None) get_all_periods_mock.assert_called_once() assert result == { 'function': 'phf-sales-data-process', 'status': 500, 's3_object': { 'key': 'test_key', 'bucket': 'test_bucket'}, 'error_code': 'database_error', 'error_description': 'Something exploded'}