"""Test cases for re run failed custom export executions script.""" import argparse import datetime from unittest import mock import pytest from processing_accounting.bin import rerun_failed_custom_export @mock.patch('processing_accounting.bin.rerun_failed_custom_export.boto3') def test_get_session(boto3): """Test get_session utility function.""" assert rerun_failed_custom_export.get_session() boto3.session.Session.assert_called_once_with(profile_name='default') @mock.patch('processing_accounting.bin.rerun_failed_custom_export.boto3') def test_get_session_with_profile_name(boto3): """Test get_session utility function.""" profile_name = 'production' assert rerun_failed_custom_export.get_session(profile_name) boto3.session.Session.assert_called_once_with(profile_name=profile_name) @mock.patch('processing_accounting.bin.rerun_failed_custom_export.tasks') def test_get_dynamodb_key_from_input(tasks): """Test get_dynamodb_key_from_input utility function.""" expected_user_id_type = 'test user id type' expected_user_params = 'test user params' tasks._get_user_id_type.return_value = expected_user_id_type tasks._get_user_params.return_value = expected_user_params expected_user_id = '4242' expected_user_type = 'test' expected_period_ids = [1, 2, 3] expected_transaction_types = ['TST', 'TSTS'] expected_locale = 'ts_TST' expected_file_format = 'tst' execution_input = { 'user_id': expected_user_id, 'user_type': expected_user_type, 'period_ids': expected_period_ids, 'transaction_types': expected_transaction_types, 'locale': expected_locale, 'file_format': expected_file_format } expected_result = { 'user_id_type': expected_user_id_type, 'user_params': expected_user_params} result = rerun_failed_custom_export.get_dynamodb_key_from_input( execution_input) assert expected_result == result tasks._get_user_id_type.assert_called_once_with( user_id=expected_user_id, user_type=expected_user_type) tasks._get_user_params.assert_called_once_with( period_ids=expected_period_ids, transaction_types=expected_transaction_types, locale=expected_locale, file_format=expected_file_format) def test_get_date(): """Test get_date utility function.""" expected_result = datetime.datetime(2018, 1, 15) result = rerun_failed_custom_export.get_date('01/15/2018') assert expected_result == result def test_get_date_raises(): """Test get_date utility function.""" with pytest.raises(argparse.ArgumentTypeError): rerun_failed_custom_export.get_date('15/01/2018') @pytest.mark.parametrize('status', ['PENDING', 'GENERATING', '']) @mock.patch( 'processing_accounting.bin.rerun_failed_custom_export' '.get_dynamodb_key_from_input') @mock.patch('processing_accounting.bin.rerun_failed_custom_export.dynamodb') @mock.patch('processing_accounting.bin.rerun_failed_custom_export.swf') def test_process_items(swf, dynamodb, get_dynamodb_key_from_input, status): """Test process_items utility function.""" domain_name = 'test domain' table_name = 'test table' session = 'test session' item_execution = 'test item execution' item = {'execution': item_execution} items = [item] execution_input = 'test execution input' expected_result = [{ 'execution_data': item, 'input': execution_input}] expected_key = 'test dynamodb key' dynamodb.get_status_using_session.return_value = status get_dynamodb_key_from_input.return_value = expected_key execution_history = mock.MagicMock() swf.get_execution_history.return_value = execution_history swf.get_execution_input.return_value = execution_input result = rerun_failed_custom_export.process_items( items, domain_name=domain_name, table_name=table_name, session=session) assert expected_result == list(result) swf.get_execution_history.assert_called_once_with( domain_name, item_execution, session=session) swf.get_execution_input.assert_called_once_with(execution_history) get_dynamodb_key_from_input.assert_called_once_with(execution_input) dynamodb.get_status_using_session.assert_called_once_with( key=expected_key, table_name=table_name, session=session) assert swf.re_run_execution.call_count == 0 @mock.patch( 'processing_accounting.bin.rerun_failed_custom_export' '.get_dynamodb_key_from_input') @mock.patch('processing_accounting.bin.rerun_failed_custom_export.dynamodb') @mock.patch('processing_accounting.bin.rerun_failed_custom_export.swf') def test_process_items_re_run(swf, dynamodb, get_dynamodb_key_from_input): """Test process_items utility function.""" domain_name = 'test domain' table_name = 'test table' session = 'test session' item_execution = 'test item execution' item = {'execution': item_execution} items = [item] execution_input = 'test execution input' re_run_response = 'test re run' expected_result = [{ 'execution_data': item, 'input': execution_input, 'rerun_response': re_run_response}] expected_key = 'test dynamodb key' dynamodb.get_status_using_session.return_value = 'PENDING' get_dynamodb_key_from_input.return_value = expected_key execution_history = mock.MagicMock() swf.get_execution_history.return_value = execution_history swf.get_execution_input.return_value = execution_input swf.re_run_execution.return_value = re_run_response result = rerun_failed_custom_export.process_items( items, domain_name=domain_name, table_name=table_name, session=session, re_run=True) assert expected_result == list(result) swf.get_execution_history.assert_called_once_with( domain_name, item_execution, session=session) swf.get_execution_input.assert_called_once_with(execution_history) get_dynamodb_key_from_input.assert_called_once_with(execution_input) dynamodb.get_status_using_session.assert_called_once_with( key=expected_key, table_name=table_name, session=session) swf.re_run_execution.assert_called_once_with( execution=item, execution_input=execution_input, domain_name=domain_name, session=session) @mock.patch( 'processing_accounting.bin.rerun_failed_custom_export' '.get_dynamodb_key_from_input') @mock.patch('processing_accounting.bin.rerun_failed_custom_export.dynamodb') @mock.patch('processing_accounting.bin.rerun_failed_custom_export.swf') def test_process_items_wrong_status( swf, dynamodb, get_dynamodb_key_from_input): """Test process_items utility function.""" domain_name = 'test domain' table_name = 'test table' session = 'test session' item_execution = 'test item execution' item = {'execution': item_execution} items = [item] expected_result = [] dynamodb.get_status_using_session.return_value = 'GENERATED' result = rerun_failed_custom_export.process_items( items, domain_name=domain_name, table_name=table_name, session=session) assert expected_result == list(result) assert swf.re_run_execution.call_count == 0 @mock.patch('processing_accounting.bin.rerun_failed_custom_export.logging') @mock.patch( 'processing_accounting.bin.rerun_failed_custom_export.process_items') @mock.patch( 'processing_accounting.bin.rerun_failed_custom_export.get_session') @mock.patch('processing_accounting.bin.rerun_failed_custom_export.swf') def test_main(swf, get_session, process_items, logging): """Test main script function.""" profile_name = 'test profile' domain_name = 'test domain' table_name = 'test table' latest_date = mock.MagicMock() time_delta = 42 re_run = True session = 'test session' get_session.return_value = session expected_failed_executions = [1, 2] swf.get_latest_failed_executions.return_value = expected_failed_executions process_items.return_value = [{}, {}] rerun_failed_custom_export.main( profile_name=profile_name, domain_name=domain_name, table_name=table_name, latest_date=latest_date, time_delta=time_delta, re_run=re_run) get_session.assert_called_once_with(profile_name=profile_name) swf.get_latest_failed_executions.assert_called_once_with( domain=domain_name, latest_date=latest_date, session=session, days_range=time_delta) process_items.assert_called_once_with( expected_failed_executions, domain_name=domain_name, table_name=table_name, session=session, re_run=re_run)