"""Unit testcases for GenerateFlowthroughAdjustmentsProcessor processor.""" from unittest.mock import MagicMock, patch import pytest from generate_flowthrough_adjustments.constants.generate_flowthrough_adjustments import ( ERROR_NO_ADJUSTMENT_FILE_FOUND, ERROR_NO_BATCH_CRITERIA_FOUND, ERROR_NO_CURRENT_STATEMENT_PERIOD, MISSING_PAYMENT_SCHEDULES, MISSING_REFERENCE_PAYMENT_ENTITIES, ) from generate_flowthrough_adjustments.error_handling import ( NotFoundError, ValidationError, ) from generate_flowthrough_adjustments.processor import ( GenerateFlowthroughAdjustmentsProcessor, ) @patch('generate_flowthrough_adjustments.processor.get_current_statement_period') def test__get_current_statement_period( mock_get_current_statement_period, mock_event, mock_statement_period ): """Test _get_current_statement_period method.""" mock_sf_executor = MagicMock() mock_get_current_statement_period.return_value = mock_statement_period processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._get_current_statement_period() assert processor._current_statement_month == 1 assert processor._current_statement_year == 2026 @patch('generate_flowthrough_adjustments.processor.get_current_statement_period') def test__get_current_statement_period_error( mock_get_current_statement_period, mock_event ): """Test raises an error if there is no current statement period.""" mock_sf_executor = MagicMock() mock_get_current_statement_period.return_value = {} processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(NotFoundError) as excinfo: processor._get_current_statement_period() assert str(excinfo.value) == ERROR_NO_CURRENT_STATEMENT_PERIOD @patch( 'generate_flowthrough_adjustments.processor.get_statement_period_adjustment_file' ) def test__get_statement_period_adjustment_file( mock_get_statement_period_adjustment_file, mock_event, mock_statement_period_adjustment_file, ): """Test _get_statement_period_adjustment_file method.""" mock_sf_executor = MagicMock() mock_get_statement_period_adjustment_file.return_value = ( mock_statement_period_adjustment_file ) processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) result = processor._get_statement_period_adjustment_file() assert result == mock_statement_period_adjustment_file mock_get_statement_period_adjustment_file.assert_called_once_with( str(mock_event['target_id']) ) @patch( 'generate_flowthrough_adjustments.processor.get_statement_period_adjustment_file' ) def test__get_statement_period_adjustment_file_error( mock_get_statement_period_adjustment_file, mock_event, ): """Test raises an error if there is no record for adjustment file.""" mock_sf_executor = MagicMock() mock_get_statement_period_adjustment_file.return_value = None statement_period_adjustment_file_id = 1 processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._statement_period_adjustment_file_id = statement_period_adjustment_file_id with pytest.raises(NotFoundError) as excinfo: processor._get_statement_period_adjustment_file() assert str(excinfo.value) == ERROR_NO_ADJUSTMENT_FILE_FOUND.format( statement_period_adjustment_file_id=statement_period_adjustment_file_id ) @patch( 'generate_flowthrough_adjustments.processor.get_adjustment_batch_criteria_by_file_id' ) def test__get_statement_period_adjustment_batch_criteria( mock_get_batch_criteria, mock_event, mock_statement_period_adjustment_batch_criteria ): """Test _get_statement_period_adjustment_batch_criteria method.""" mock_sf_executor = MagicMock() mock_get_batch_criteria.return_value = ( mock_statement_period_adjustment_batch_criteria ) processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) result = processor._get_statement_period_adjustment_batch_criteria() assert result == mock_statement_period_adjustment_batch_criteria mock_get_batch_criteria.assert_called_once_with(str(mock_event['target_id'])) @patch( 'generate_flowthrough_adjustments.processor.get_adjustment_batch_criteria_by_file_id' ) def test__get_statement_period_adjustment_batch_criteria_error( mock_get_batch_criteria, mock_event ): """Test raises an error if there is no record for adjustment file batch criteria.""" mock_sf_executor = MagicMock() mock_get_batch_criteria.return_value = {} statement_period_adjustment_file_id = 1 processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._statement_period_adjustment_file_id = statement_period_adjustment_file_id with pytest.raises(NotFoundError) as excinfo: processor._get_statement_period_adjustment_batch_criteria() assert str(excinfo.value) == ERROR_NO_BATCH_CRITERIA_FOUND.format( statement_period_adjustment_file_id=statement_period_adjustment_file_id ) def test_payment_schedules_missing(mock_event): """Test _format_batch_criteria_values method. Throws an error if payment_schedules is missing. """ mock_batch_criteria = { 'batch_criteria': {'reference_payment_entities': [1, 2, 3, 4]}, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(ValidationError) as e: processor._format_batch_criteria_values(mock_batch_criteria) assert str(e.value) == MISSING_PAYMENT_SCHEDULES def test_payment_schedules_empty(mock_event): """Test _format_batch_criteria_values method. Throws an error if payment_schedules is empty. """ mock_batch_criteria = { 'batch_criteria': { 'reference_payment_entities': [1, 2, 3, 4], 'payment_schedules': [], }, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(ValidationError) as e: processor._format_batch_criteria_values(mock_batch_criteria) assert str(e.value) == MISSING_PAYMENT_SCHEDULES def test_payment_schedules_null(mock_event): """Test _format_batch_criteria_values method. Throws an error if payment_schedules is null. """ mock_batch_criteria = { 'batch_criteria': { 'reference_payment_entities': [1, 2, 3, 4], 'payment_schedules': None, }, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(ValidationError) as e: processor._format_batch_criteria_values(mock_batch_criteria) assert str(e.value) == MISSING_PAYMENT_SCHEDULES def test_reference_payment_entities_missing(mock_event): """Test _format_batch_criteria_values method. Throws an error if reference_payment_entities is missing. """ mock_batch_criteria = { 'batch_criteria': { 'payment_schedules': [ '30_days_after_month_end', '45_days_after_month_end', '60_days_after_month_end', ], }, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(ValidationError) as e: processor._format_batch_criteria_values(mock_batch_criteria) assert str(e.value) == MISSING_REFERENCE_PAYMENT_ENTITIES def test_reference_payment_entities_empty(mock_event): """Test _format_batch_criteria_values method. Throws an error if reference_payment_entities is empty. """ mock_batch_criteria = { 'batch_criteria': { 'reference_payment_entities': [], 'payment_schedules': [ '30_days_after_month_end', '45_days_after_month_end', '60_days_after_month_end', ], }, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(ValidationError) as e: processor._format_batch_criteria_values(mock_batch_criteria) assert str(e.value) == MISSING_REFERENCE_PAYMENT_ENTITIES def test_reference_payment_entities_null(mock_event): """Test _format_batch_criteria_values method. Throws an error if reference_payment_entities is null. """ mock_batch_criteria = { 'batch_criteria': { 'reference_payment_entities': None, 'payment_schedules': [ '30_days_after_month_end', '45_days_after_month_end', '60_days_after_month_end', ], }, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(ValidationError) as e: processor._format_batch_criteria_values(mock_batch_criteria) assert str(e.value) == MISSING_REFERENCE_PAYMENT_ENTITIES def test_batch_criteria_empty(mock_event): """Test _format_batch_criteria_values method. Throws an error if batch_criteria is empty. """ mock_batch_criteria = { 'batch_criteria': {}, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(ValidationError) as e: processor._format_batch_criteria_values(mock_batch_criteria) assert str(e.value) == 'batch_criteria is empty.' def test_batch_criteria_null(mock_event): """Test _format_batch_criteria_values method. Throws an error if batch_criteria is null. """ mock_batch_criteria = { 'batch_criteria': None, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) with pytest.raises(ValidationError) as e: processor._format_batch_criteria_values(mock_batch_criteria) assert str(e.value) == 'batch_criteria is empty.' def test__format_batch_criteria_values(mock_event): """Test _format_batch_criteria_values method.""" mock_batch_criteria = { 'batch_criteria': { 'reference_payment_entities': [1, 2, 3, 4], 'payment_schedules': [ '30_days_after_month_end', '45_days_after_month_end', '60_days_after_month_end', ], }, 'statement_period_adjustment_file_id': 1, 'statement_period_adjustment_batch_criteria_id': 1, } mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) payment_schedules, reference_payment_entities = ( processor._format_batch_criteria_values(mock_batch_criteria) ) assert ( payment_schedules == "'30_days_after_month_end','45_days_after_month_end','60_days_after_month_end'" ) assert reference_payment_entities == '1,2,3,4' def test__get_flowthrough_from_snowflake(mock_flowthrough_adjustments, mock_event): """Test _get_flowthrough_from_snowflake method.""" payment_schedules = ( "'30_days_after_month_end','45_days_after_month_end','60_days_after_month_end'" ) reference_payment_entities = '1,2,3,4' GenerateFlowthroughAdjustmentsProcessor._current_statement_period_id = 329 mock_sf_executor = MagicMock() mock_sf_executor.fetchall.return_value = mock_flowthrough_adjustments processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) result = processor._get_flowthrough_from_snowflake( payment_schedules, reference_payment_entities ) assert result == mock_flowthrough_adjustments mock_sf_executor.fetchall.assert_called_once() @patch('generate_flowthrough_adjustments.processor.pd.DataFrame.to_json') def test__export_results_to_gzipped_json_success( mock_to_json, mock_statement_period_adjustment_file, mock_flowthrough_adjustments, mock_event, ): """Test _export_results_to_gzipped_json method.""" mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._current_statement_month = 3 processor._current_statement_year = 2026 zip_path, zip_name = processor._export_results_to_gzipped_json( mock_flowthrough_adjustments, mock_statement_period_adjustment_file ) assert zip_name == 'flowthrough_adjustments_file.json.gz' assert 'flowthrough_adjustments_file.json.gz' in zip_path mock_to_json.assert_called_once() @patch('generate_flowthrough_adjustments.processor.pd.DataFrame.to_json') @patch('generate_flowthrough_adjustments.processor.os.makedirs') def test__export_results_to_gzipped_json_creates_parent_directory( mock_makedirs, mock_to_json, mock_statement_period_adjustment_file, mock_flowthrough_adjustments, mock_event, ): """Test that the parent directory is created before writing the gzip file.""" mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._current_statement_month = 3 processor._current_statement_year = 2026 zip_path, _ = processor._export_results_to_gzipped_json( mock_flowthrough_adjustments, mock_statement_period_adjustment_file ) mock_makedirs.assert_called_once_with(mock_makedirs.call_args[0][0], exist_ok=True) assert mock_makedirs.call_args[1] == {'exist_ok': True} @patch('generate_flowthrough_adjustments.processor.pd.DataFrame.to_json') @patch('generate_flowthrough_adjustments.processor.os.makedirs') def test__export_results_to_gzipped_json_sanitizes_slashes_in_file_name( mock_makedirs, mock_to_json, mock_flowthrough_adjustments, mock_event, ): """Test that slashes in file_name are replaced with dashes to prevent nested paths.""" mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._current_statement_month = 3 processor._current_statement_year = 2026 adjustment_file_with_slash = {'file_name': 'Orchard US UK/30'} zip_path, zip_name = processor._export_results_to_gzipped_json( mock_flowthrough_adjustments, adjustment_file_with_slash ) assert zip_name == 'Orchard-US-UK-30.json.gz' assert '/Orchard-US-UK-30.json.gz' in zip_path assert 'UK/30' not in zip_path @patch('generate_flowthrough_adjustments.processor.pd.DataFrame.to_json') @patch('generate_flowthrough_adjustments.processor.os.makedirs') @patch('generate_flowthrough_adjustments.processor.tempfile.gettempdir') def test__export_results_to_gzipped_json_handles_nonexistent_temp_dir( mock_gettempdir, mock_makedirs, mock_to_json, mock_statement_period_adjustment_file, mock_flowthrough_adjustments, mock_event, ): """Test that makedirs is called even when gettempdir returns a nonexistent subdirectory. Reproduces the Lambda warm-container failure mode where tempfile.tempdir was previously set to a path that was later cleaned up. """ mock_gettempdir.return_value = '/tmp/Orchard-US-UK-30' mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._current_statement_month = 3 processor._current_statement_year = 2026 processor._export_results_to_gzipped_json( mock_flowthrough_adjustments, mock_statement_period_adjustment_file ) mock_makedirs.assert_called_once_with('/tmp/Orchard-US-UK-30', exist_ok=True) @patch('generate_flowthrough_adjustments.processor.config') @patch('generate_flowthrough_adjustments.processor.get_s3_connector') def test__upload_gzip_file_to_s3_success(mock_get_connector, mock_config, mock_event): """Test _upload_gzip_file_to_s3 method.""" mock_sf_executor = MagicMock() processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._statement_period_adjustment_file_id = 999 local_path = '/tmp/test_file.json.gz' file_name = 'test_file.json.gz' expected_s3_key = '999/test_file.json.gz' mock_s3_instance = MagicMock() mock_s3_instance.upload_object.return_value = True mock_get_connector.return_value = mock_s3_instance mock_config.S3_BUCKET_NAME = 'test-bucket' mock_config.S3_ACCOUNT_ID = '123456789' result = processor._upload_gzip_file_to_s3(local_path, file_name) assert result == f's3://{mock_config.S3_BUCKET_NAME}/{expected_s3_key}' mock_get_connector.assert_called_once() mock_s3_instance.upload_object.assert_called_once_with( 'test-bucket', local_path, expected_s3_key, '123456789' ) def test_process( mock_statement_period, mock_statement_period_adjustment_file, mock_statement_period_adjustment_batch_criteria, mock_flowthrough_adjustments, mock_event, ): """Test process method.""" mock_sf_executor = MagicMock() payment_schedules = ( "'30_days_after_month_end','45_days_after_month_end','60_days_after_month_end'" ) reference_payment_entities = '1,2,3,4' processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._get_current_statement_period = MagicMock( return_value=mock_statement_period ) processor._get_statement_period_adjustment_file = MagicMock( return_value=mock_statement_period_adjustment_file ) processor._get_statement_period_adjustment_batch_criteria = MagicMock( return_value=mock_statement_period_adjustment_batch_criteria ) processor._format_batch_criteria_values = MagicMock( return_value=( payment_schedules, reference_payment_entities, ) ) processor._get_flowthrough_from_snowflake = MagicMock( return_value=mock_flowthrough_adjustments ) processor._export_results_to_gzipped_json = MagicMock( return_value=('test_json.json.gz', 'test_json.json') ) processor._upload_gzip_file_to_s3 = MagicMock( return_value='s3://test-bucket/1/test.json.gz' ) processor._update_statement_period_adjustment_file = MagicMock(return_value=True) result = processor.process() assert result == 1 processor._get_current_statement_period.assert_called_once() processor._get_statement_period_adjustment_file.assert_called_once() processor._get_statement_period_adjustment_batch_criteria.assert_called_once() processor._format_batch_criteria_values.assert_called_once_with( mock_statement_period_adjustment_batch_criteria ) processor._get_flowthrough_from_snowflake.assert_called_once_with( payment_schedules, reference_payment_entities ) processor._export_results_to_gzipped_json.assert_called_once_with( mock_flowthrough_adjustments, mock_statement_period_adjustment_file ) processor._upload_gzip_file_to_s3.assert_called_once_with( 'test_json.json.gz', 'test_json.json' ) def test_process_no_flowthrough_records( mock_statement_period, mock_statement_period_adjustment_file, mock_statement_period_adjustment_batch_criteria, mock_event, ): """Test process method when no flowthrough records for given criteria.""" mock_sf_executor = MagicMock() payment_schedules = ( "'30_days_after_month_end','45_days_after_month_end','60_days_after_month_end'" ) reference_payment_entities = '1,2,3,4' processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._get_current_statement_period = MagicMock( return_value=mock_statement_period ) processor._get_statement_period_adjustment_file = MagicMock( return_value=mock_statement_period_adjustment_file ) processor._get_statement_period_adjustment_batch_criteria = MagicMock( return_value=mock_statement_period_adjustment_batch_criteria ) processor._format_batch_criteria_values = MagicMock( return_value=( payment_schedules, reference_payment_entities, ) ) processor._get_flowthrough_from_snowflake = MagicMock(return_value=None) processor._export_results_to_gzipped_json = MagicMock() processor._upload_gzip_file_to_s3 = MagicMock() result = processor.process() assert result == 0 processor._get_current_statement_period.assert_called_once() processor._get_statement_period_adjustment_file.assert_called_once() processor._get_statement_period_adjustment_batch_criteria.assert_called_once() processor._format_batch_criteria_values.assert_called_once_with( mock_statement_period_adjustment_batch_criteria ) processor._get_flowthrough_from_snowflake.assert_called_once_with( payment_schedules, reference_payment_entities ) processor._export_results_to_gzipped_json.assert_not_called() processor._upload_gzip_file_to_s3.assert_not_called() @patch( 'generate_flowthrough_adjustments.processor.update_statement_period_adjustment_file' ) def test__update_adjustment_file_record(mock_update_api, mock_event): """Test that file record is updated.""" mock_sf_executor = MagicMock() s3_bucket_file_path = 's3://test-bucket/1/test.json.gz' processor = GenerateFlowthroughAdjustmentsProcessor(mock_event, mock_sf_executor) processor._statement_period_id = 10 processor._statement_period_adjustment_file_id = 20 processor._update_statement_period_adjustment_file(s3_bucket_file_path) expected_put_body = { 'valid_file_location': s3_bucket_file_path, } mock_update_api.assert_called_once_with(expected_put_body, 10, 20)