"""Unit testcases for AdjustmentFileImport processor.""" from unittest.mock import MagicMock from unittest.mock import patch import pymysql import pytest from adjustment_file_import.constants.adjustment_file_import import \ ERROR_NO_REFERENCE_ADJUSTMENT_TYPES_FOUND from adjustment_file_import.constants.adjustment_file_import import \ ERROR_NO_STATEMENT_PERIODS_FOUND from adjustment_file_import.constants.adjustment_file_import import \ FILE_IS_EMPTY from adjustment_file_import.constants.adjustment_file_import import \ INVALID_EXCEL_FILE from adjustment_file_import.constants.mysql_queries import \ INSERT_WORKSHEET_ADJUSTMENT from adjustment_file_import.processor import AdjustmentFileImportProcessor @patch('adjustment_file_import.processor.download_file') @patch('adjustment_file_import.processor.get_statement_period_adjustment_file') def test_get_adjustment_file( mock_get_adjustment_file_adjustment_file, mock_download_s3_file, mock_event ): """Test _get_adjustment_file method.""" mock_statement_period_adjustment_file = { 'valid_file_location': 's3://qa-abacus-adjustments/test.xlsx', 'statement_period_id': 282, 'statement_period_adjustment_file_id': 1 } mock_mysql_conn = MagicMock() mock_get_adjustment_file_adjustment_file.return_value = \ mock_statement_period_adjustment_file mock_download_s3_file.return_value = True adjustmentImport = AdjustmentFileImportProcessor(mock_event, mock_mysql_conn) adjustmentImport._get_adjustment_file() mock_get_adjustment_file_adjustment_file.assert_called_once_with(1) mock_download_s3_file.assert_called_once_with('qa-abacus-adjustments', 'test.xlsx') @patch('adjustment_file_import.processor.pathlib') def test_is_excel_file(mock_pathlib_method, mock_event): """Test is_excel_file method.""" file = 'test.xlsx' mock_pathlib_method.Path.return_value = MagicMock(suffix='.xlsx') mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) result = adjustmentImport._is_excel_file(file) assert result == file @patch('adjustment_file_import.processor.pathlib') def test_is_excel_file_error(mock_pathlib_method, mock_event): """Test is_excel_file method throws an error if file is text file.""" file = 'test.txt' mock_pathlib_method.Path.return_value = MagicMock(suffix='.txt') mock_mysql_conn = MagicMock() with pytest.raises(Exception) as excinfo: adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) adjustmentImport._is_excel_file(file) assert str(excinfo.value) == INVALID_EXCEL_FILE @patch('adjustment_file_import.processor.get_statement_periods_by_years') def test_get_formatted_statement_periods( mock_get_statement_periods, mock_event ): """Test _get_formatted_statement_periods method.""" mock_get_statement_periods.return_value = [{ 'statement_period_id': 1, 'statement_period_name': 'January 2023', 'statement_period_status': 'current', 'statement_month': 1, 'statement_year': 2023 }, { 'statement_period_id': 2, 'statement_period_name': 'February 2023', 'statement_period_status': 'open', 'statement_month': 2, 'statement_year': 2023 }] mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) res = adjustmentImport._get_formatted_statement_periods([2023]) assert res == {2023: {1: 1, 2: 2}} @patch('adjustment_file_import.processor.get_statement_periods_by_years') def test_get_formatted_statement_periods_error( mock_get_statement_periods, mock_event ): """Test error is raised if no statement periods are found.""" statement_years = [2023] mock_get_statement_periods.return_value = [] mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) with pytest.raises(Exception) as excinfo: adjustmentImport._get_formatted_statement_periods([2023]) assert str(excinfo.value) == \ ERROR_NO_STATEMENT_PERIODS_FOUND.format(statement_years) @patch('adjustment_file_import.processor.get_reference_adjustment_types') def test_get_formatted_reference_adjustment_types( mock_get_reference_adjustment_types, mock_event ): """Test _get_formatted_reference_adjustment_types function.""" mock_get_reference_adjustment_types.return_value = { 'items': [{ 'reference_adjustment_type_id': 1, 'type_name': 'label earnings' }, { 'reference_adjustment_type_id': 2, 'type_name': 'reclass between labels' }], 'total_count': 2 } mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) res = adjustmentImport._get_formatted_reference_adjustment_types() assert res == {'label earnings': 1, 'reclass between labels': 2} @patch('adjustment_file_import.processor.get_reference_adjustment_types') def test_get_formatted_reference_adjustment_types_error( mock_get_reference_adjustment_types, mock_event ): """Test error is raised if no adjustment types are found.""" mock_get_reference_adjustment_types.return_value = { 'items': [], 'total_count': 0 } mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) with pytest.raises(Exception) as excinfo: adjustmentImport._get_formatted_reference_adjustment_types() assert str(excinfo.value) == ERROR_NO_REFERENCE_ADJUSTMENT_TYPES_FOUND @patch('adjustment_file_import.processor.bulk_insert_query') def test_insert_worksheet_adjustment_entries( mock_bulk_insert_query, mock_event, mock_worksheet_adjustment_details ): """Test bulk insert for worksheet_adjustment records.""" mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) queue = list() for entry in mock_worksheet_adjustment_details: queue.append(entry) adjustmentImport._insert_worksheet_adjustment_entries( mock_worksheet_adjustment_details ) mock_bulk_insert_query.assert_called_once_with( INSERT_WORKSHEET_ADJUSTMENT, queue, mock_mysql_conn ) @patch('adjustment_file_import.processor.bulk_insert_query') def test_insert_worksheet_adjustment_entries_in_batches( mock_bulk_insert_query, mock_event, mock_worksheet_adjustment_details ): """Test bulk insert for worksheet_adjustment records in batches of 2.""" with patch('adjustment_file_import.processor.MYSQL_BATCH_SIZE', 2): mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) queue = list() for entry in mock_worksheet_adjustment_details: queue.append(entry) adjustmentImport._insert_worksheet_adjustment_entries( mock_worksheet_adjustment_details ) mock_bulk_insert_query.assert_called_once_with( INSERT_WORKSHEET_ADJUSTMENT, queue, mock_mysql_conn ) def test_insert_worksheet_adjustment_detail_entries( mock_event, mock_expense_details_entries ): """Test insert for worksheet_adjustment_details and worksheet_adjustment records.""" mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) worksheet_adjustment_id = 1 cursor_mock = MagicMock() mock_mysql_conn.cursor.return_value.__enter__.return_value = cursor_mock adjustmentImport._insert_worksheet_adjustment = MagicMock( return_value=worksheet_adjustment_id ) adjustmentImport._insert_worksheet_adjustment_details = MagicMock( return_value=True ) adjustmentImport._insert_worksheet_adjustment_detail_entries( mock_expense_details_entries ) adjustmentImport._insert_worksheet_adjustment.assert_called_once_with( mock_expense_details_entries[0]['worksheet_adjustment'], cursor_mock ) adjustmentImport._insert_worksheet_adjustment_details.assert_called_once_with( mock_expense_details_entries[0]['worksheet_adjustment_detail'], worksheet_adjustment_id, cursor_mock ) def test_insert_worksheet_adjustment_details( mock_event, mock_expense_details_entries ): """Test insert for worksheet_adjustment_details records.""" mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) cursor_mock = MagicMock() mock_mysql_conn.cursor.return_value.__enter__.return_value = cursor_mock adjustmentImport._insert_worksheet_adjustment_details( mock_expense_details_entries[0]['worksheet_adjustment_detail'], 1, cursor_mock ) assert cursor_mock.executemany.call_count == 1 def test_insert_worksheet_adjustment_details_deadlock_error( mock_event, mock_expense_details_entries ): """Test to check re-run of _insert_worksheet_adjustment_details in case of deadlock error.""" # noqa: E501 mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) cursor_mock = MagicMock() mock_mysql_conn.cursor.return_value.__enter__.return_value = cursor_mock cursor_mock.executemany.side_effect = pymysql.err.OperationalError( 1213, 'Deadlock found when trying to get lock; try restarting transaction' ) with pytest.raises(Exception): adjustmentImport._insert_worksheet_adjustment_details( mock_expense_details_entries[0]['worksheet_adjustment_detail'], 1, cursor_mock ) assert cursor_mock.executemany.call_count == 3 def test_insert_worksheet_adjustment( mock_event, mock_expense_details_entries ): """Test insert for worksheet_adjustment records.""" mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) cursor_mock = MagicMock() mock_mysql_conn.cursor.return_value.__enter__.return_value = cursor_mock adjustmentImport._insert_worksheet_adjustment( mock_expense_details_entries[0]['worksheet_adjustment'], cursor_mock ) assert cursor_mock.execute.call_count == 1 def test_get_statement_years(mock_event): """Test _get_statement_years method.""" mock_adjustment_file_data = MagicMock() mock_adjustment_file_data.loc.__getitem__.return_value \ .dropna.return_value = [2024, 2023, 2024, 2025] mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) res = adjustmentImport._get_statement_years(mock_adjustment_file_data) assert res == [2023, 2024, 2025] @patch('adjustment_file_import.processor.pd') def test_read_adjustment_file_error(mock_pandas, mock_event): """Test _read_adjustment_file method for empty file.""" mock_adjustment_file = 'file/test.xlsx' mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) mock_pandas.read_excel.return_value.empty = True with pytest.raises(Exception) as excinfo: adjustmentImport._read_adjustment_file(mock_adjustment_file) assert str(excinfo.value) == FILE_IS_EMPTY @patch('adjustment_file_import.processor.pd') def test_read_adjustment_file( mock_pandas, mock_event, mock_adjustment_file_content ): """Test _read_adjustment_file method.""" mock_adjustment_file = 'file/test.xlsx' mock_adjustment_file_data = MagicMock( return_value=enumerate(mock_adjustment_file_content) ) mock_pandas.read_excel.return_value.astype.return_value \ .replace.return_value.dropna.return_value.astype.return_value.where \ .return_value.apply.return_value.iterrows = mock_adjustment_file_data mock_pandas.read_excel.return_value.astype.return_value \ .replace.return_value.dropna.return_value.astype.return_value.where \ .return_value.apply.return_value.empty = False mock_mysql_conn = MagicMock() adjustment_import = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) res = adjustment_import._read_adjustment_file(mock_adjustment_file) assert res.iterrows == mock_adjustment_file_data @patch('adjustment_file_import.processor.pd') def test_read_adjustment_file_special_characters( mock_pandas, mock_event, mock_adjustment_file_content_special_characters ): """Test _read_adjustment_file method with special characters.""" mock_adjustment_file = 'file/test.xlsx' mock_adjustment_file_data = MagicMock( return_value=enumerate(mock_adjustment_file_content_special_characters) ) mock_pandas.read_excel.return_value.astype.return_value \ .replace.return_value.dropna.return_value.astype.return_value.where \ .return_value.apply.return_value.iterrows = mock_adjustment_file_data mock_pandas.read_excel.return_value.astype.return_value \ .replace.return_value.dropna.return_value.astype.return_value.where \ .return_value.apply.return_value.empty = False mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) res = adjustmentImport._read_adjustment_file(mock_adjustment_file) assert res.iterrows == mock_adjustment_file_data @patch('adjustment_file_import.processor.ExpenseDetailsProcessor') @patch('adjustment_file_import.processor.AdjustmentFileProcessor') def test_read_adjustment_file_and_insert_file_content( mock_adjustment_file_processor, mock_expense_details_processor, mock_statement_periods, mock_reference_adjustment_types, mock_adjustment_file_content, mock_event, mock_expense_details_entries, mock_worksheet_adjustment_details ): """Test _read_adjustment_file_and_insert_file_content method.""" mock_adjustment_file = 'file/test.xlsx' iterrows = MagicMock( return_value=enumerate(mock_adjustment_file_content) ) AdjustmentFileImportProcessor._read_adjustment_file = MagicMock( return_value=iterrows ) AdjustmentFileImportProcessor._get_statement_years = MagicMock( return_value=[2024] ) AdjustmentFileImportProcessor._get_formatted_statement_periods = MagicMock( return_value=mock_statement_periods ) AdjustmentFileImportProcessor._get_formatted_reference_adjustment_types = \ MagicMock(return_value=mock_reference_adjustment_types) AdjustmentFileImportProcessor._insert_worksheet_adjustment_entries = MagicMock() AdjustmentFileImportProcessor._insert_worksheet_adjustment_detail_entries = \ MagicMock() mock_adjustment_file_processor.return_value = MagicMock( process=MagicMock(), worksheet_adjustment_entries=[mock_adjustment_file_content[1]], worksheet_adjustment_detail_entries=mock_worksheet_adjustment_details ) mock_expense_details_processor.return_value = MagicMock( process=MagicMock(), expense_detail_entries=mock_expense_details_entries ) mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) adjustmentImport._read_adjustment_file_and_insert_file_content(mock_adjustment_file) AdjustmentFileImportProcessor._insert_worksheet_adjustment_entries \ .assert_called_once_with([mock_adjustment_file_content[1]]) AdjustmentFileImportProcessor._insert_worksheet_adjustment_detail_entries \ .assert_called_once_with(mock_expense_details_entries) def test_process(mock_event): """Test process method.""" mock_adjustment_file = 'file/test.xlsx' AdjustmentFileImportProcessor._read_adjustment_file_and_insert_file_content = \ MagicMock(return_value=True) AdjustmentFileImportProcessor._get_adjustment_file = \ MagicMock(return_value=mock_adjustment_file) mock_mysql_conn = MagicMock() adjustmentImport = AdjustmentFileImportProcessor( mock_event, mock_mysql_conn ) adjustmentImport.process() AdjustmentFileImportProcessor._read_adjustment_file_and_insert_file_content \ .assert_called_once_with(mock_adjustment_file) AdjustmentFileImportProcessor._get_adjustment_file.assert_called_once()