"""Unit tests for WorksheetAdjustment logic.""" from decimal import Decimal from unittest.mock import MagicMock, call, patch import pytest from abacus_common_logic.constants.error import ERROR_ENTITY_DOES_NOT_EXIST from marshmallow import ValidationError from sqlalchemy.exc import OperationalError from abacus_worksheet.constants import constants from abacus_worksheet.constants.error import ( ERROR_DELETE_WORKSHEET_ADJUSTMENTS, ERROR_ENTRY_APPLIED, ) from abacus_worksheet.logic import worksheet_adjustment from abacus_worksheet.schemas.worksheet_adjustment import WorksheetAdjustmentListSchema from tests.utils.factories import WorksheetAdjustmentFactory @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') def test_get_by_statement_period_adjustment_file_id(mock_model): """Test get_by_statement_period_adjustment_file_id function.""" statement_period_adjustment_file_id = 1 mock_worksheet_adjustments = WorksheetAdjustmentFactory.create_batch(5) request_params = {'limit': 25, 'offset': 0} mock_model.get_by_statement_period_adjustment_file_id.return_value = ( mock_worksheet_adjustments, len(mock_worksheet_adjustments), ) res = worksheet_adjustment.get_by_statement_period_adjustment_file_id( statement_period_adjustment_file_id, request_params ) assert res.status == 200 assert res.message.get('items') assert res.message.get('total_count') == 5 mock_model.get_by_statement_period_adjustment_file_id.assert_called_once_with( statement_period_adjustment_file_id, **request_params ) @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriod') def test_get_formatted_statement_periods(mock_model): """Test get_formatted_statement_periods function.""" mock_model.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, }, ] res = worksheet_adjustment.get_formatted_statement_periods([2023]) assert res == {2023: {1: 1, 2: 2}} @patch('abacus_worksheet.logic.worksheet_adjustment.ReferenceAdjustmentType') def test_get_formatted_reference_adjustment_types(mock_model): """Test get_formatted_reference_adjustment_types function.""" mock_model.get_adjustment_types.return_value = [ {'reference_adjustment_type_id': 1, 'type_name': 'label earnings'}, {'reference_adjustment_type_id': 2, 'type_name': 'reclass between labels'}, ] res = worksheet_adjustment.get_formatted_reference_adjustment_types() assert res == {'label earnings': 1, 'reclass between labels': 2} @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') def test_formatted_worksheet_adjustment_detail(mock_model, mock_adjustment_row): """Test formatted_worksheet_adjustment_detail function.""" statement_periods = {2023: {1: 1, 2: 2}} reference_adjustment_types = {'label earnings': 1, 'reclass between labels': 2} statement_period_adjustment_file_id = 1 res = worksheet_adjustment.formatted_worksheet_adjustment_detail( mock_adjustment_row, reference_adjustment_types, statement_periods, statement_period_adjustment_file_id, ) assert res == { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'account_id': '68952', 'contract_id': '533565', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'amount': Decimal('-50.8979990'), 'currency_code': 'USD', 'upc': '196626383143', 'distribution_type': 'digital', 'note': None, 'internal_note': None, } @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') def test_formatted_worksheet_adjustment(mock_model): """Test formatted_worksheet_adjustment function.""" abacus_event_id = 1 statement_periods = {2023: {1: 1, 2: 2}} reference_adjustment_types = {'label earnings': 1, 'reclass between labels': 2} statement_period_adjustment_file_id = 1 adjustment_row = { 'account_id': '1234', 'contract_id': '1234', 'activity_year': '2023', 'activity_month': '1', 'statement_year': '2023', 'statement_month': '2', 'adjustment_type': 'label earnings', 'amount': '-90.122', 'currency': 'USD', 'upc': '1234567890', 'distribution_type': 'digital', 'client_facing_comments': None, 'internal_note': None, } res = worksheet_adjustment.formatted_worksheet_adjustment( abacus_event_id, adjustment_row, reference_adjustment_types, statement_periods, statement_period_adjustment_file_id, ) assert res == { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'abacus_event_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'adjustment_amount': Decimal('-90.122'), 'adjustment_currency_code': 'USD', 'note': None, 'internal_note': None, } @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') def test_get_expense_details_records(mock_model): """Test get_expense_details_records function.""" abacus_event_id = 1 statement_period_adjustment_file_id = 1 worksheet_adjustment_details = [ { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'amount': Decimal('-90.122'), 'currency_code': 'USD', 'upc': '1234567890', 'distribution_type': 'digital', 'note': None, 'internal_note': None, }, { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'amount': Decimal('1897.678'), 'currency_code': 'USD', 'upc': '196925277303', 'distribution_type': 'digital', 'note': None, 'internal_note': None, }, ] res = worksheet_adjustment.get_expense_details_records( statement_period_adjustment_file_id, abacus_event_id, worksheet_adjustment_details, ) assert res == [ { 'worksheet_adjustment': { 'statement_period_adjustment_file_id': 1, 'abacus_event_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 65, 'adjustment_amount': Decimal('1807.556'), 'adjustment_currency_code': 'USD', }, 'worksheet_adjustment_detail': [ { 'statement_period_adjustment_file_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'amount': Decimal('-90.122'), 'currency_code': 'USD', 'upc': '1234567890', 'distribution_type': 'digital', 'note': None, 'internal_note': None, }, { 'statement_period_adjustment_file_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'amount': Decimal('1897.678'), 'currency_code': 'USD', 'upc': '196925277303', 'distribution_type': 'digital', 'note': None, 'internal_note': None, }, ], } ] @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') def test__create_worksheet_adjustments(mock_model): """Test _create_worksheet_adjustments function.""" mock_worksheet_adjustment = WorksheetAdjustmentFactory.create() mock_model.build.return_value = mock_worksheet_adjustment mock_model.commit_changes.return_value = True params = [ { 'statement_period_adjustment_file_id': 1, 'abacus_event_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'adjustment_amount': '125.42', 'adjustment_currency_code': 'USD', 'note': 'informative message', 'internal_note': None, } ] result = worksheet_adjustment._create_worksheet_adjustments(params) assert result == [mock_worksheet_adjustment] assert mock_model.build.call_count == 1 mock_model.build.assert_called_once() mock_model.commit_changes.assert_called_once() @patch( 'abacus_worksheet.logic.worksheet_adjustment.create_worksheet_adjustment_details' ) @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') def test__create_worksheet_adjustments_and_details( mock_model, mock_adjustment_detail_logic ): """Test _create_worksheet_adjustments_and_details function.""" mock_worksheet_adjustment = WorksheetAdjustmentFactory.create() mock_model.build.return_value = mock_worksheet_adjustment mock_model.commit_changes.return_value = True mock_adjustment_detail_logic.return_value = True expense_details = [ { 'worksheet_adjustment': { 'statement_period_adjustment_file_id': 1, 'abacus_event_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 65, 'adjustment_amount': Decimal('1807.556'), 'adjustment_currency_code': 'USD', }, 'worksheet_adjustment_detail': [ { 'statement_period_adjustment_file_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'amount': Decimal('-90.122'), 'currency_code': 'USD', 'upc': '1234567890', 'distribution_type': 'digital', 'note': None, 'internal_note': None, } ], } ] result = worksheet_adjustment._create_worksheet_adjustments_and_details( expense_details ) assert result == [mock_worksheet_adjustment] assert mock_model.build.call_count == 1 mock_adjustment_detail_logic.assert_called_once() mock_model.build.assert_called_once() mock_model.commit_changes.assert_called_once() @patch( 'abacus_worksheet.logic.worksheet_adjustment._create_worksheet_adjustments_and_details' ) @patch('abacus_worksheet.logic.worksheet_adjustment._create_worksheet_adjustments') @patch('abacus_worksheet.logic.worksheet_adjustment.get_expense_details_records') @patch('abacus_worksheet.logic.worksheet_adjustment.formatted_worksheet_adjustment') @patch( 'abacus_worksheet.logic.worksheet_adjustment.formatted_worksheet_adjustment_detail' ) @patch( 'abacus_worksheet.logic.worksheet_adjustment.get_formatted_reference_adjustment_types' ) @patch('abacus_worksheet.logic.worksheet_adjustment.get_formatted_statement_periods') @patch('abacus_worksheet.logic.worksheet_adjustment.download_file') @patch('abacus_worksheet.logic.worksheet_adjustment.os') @patch('abacus_worksheet.logic.worksheet_adjustment.pd') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_create_worksheet_adjustments( mock_model, mock_pandas, mock_os, mock_download_s3_file, mock_formatted_statement_periods, mock_formatted_adjustment_types, mock_formatted_worksheet_adjustment_detail, mock_formatted_worksheet_adjustment, mock_get_expense_details_records, mock_create_worksheet_adjustments, mock_create_worksheet_adjustments_and_details, ): """Test create_worksheet_adjustments_and_details function. When there are no expense details. """ file = 'file/test.xlsx' statement_period_adjustment_file_id = 1 abacus_event_id = 1 mock_os.remove.return_value = True mock_model.get_statement_period_adjustment_file_by_id.return_value = { 'valid_file_location': 's3://qa-abacus-adjustments/test.xlsx', 'statement_period_id': 282, 'statement_period_adjustment_file_id': 1, 'deleted_by': None, 'deleted_at': None, } mock_pandas.read_excel.return_value.astype.return_value.replace.return_value.dropna.return_value.astype.return_value.where.return_value.empty = False mock_download_s3_file.return_value = MagicMock(name=file) mock_formatted_statement_periods.return_value = [] mock_formatted_adjustment_types.return_value = [] mock_formatted_worksheet_adjustment_detail.return_value = [] mock_formatted_worksheet_adjustment.return_value = [] mock_get_expense_details_records.return_value = [] mock_create_worksheet_adjustments.return_value = [] mock_create_worksheet_adjustments_and_details.return_value = [] res = worksheet_adjustment.create_worksheet_adjustments_and_details( statement_period_adjustment_file_id, abacus_event_id ) assert res.status == 201 assert res.message == {'message': 'Records inserted succesfully.'} mock_create_worksheet_adjustments.assert_called_once() mock_create_worksheet_adjustments_and_details.assert_not_called() @patch( 'abacus_worksheet.logic.worksheet_adjustment._create_worksheet_adjustments_and_details' ) @patch('abacus_worksheet.logic.worksheet_adjustment._create_worksheet_adjustments') @patch('abacus_worksheet.logic.worksheet_adjustment.get_expense_details_records') @patch('abacus_worksheet.logic.worksheet_adjustment.formatted_worksheet_adjustment') @patch( 'abacus_worksheet.logic.worksheet_adjustment.formatted_worksheet_adjustment_detail' ) @patch( 'abacus_worksheet.logic.worksheet_adjustment.get_formatted_reference_adjustment_types' ) @patch('abacus_worksheet.logic.worksheet_adjustment.get_formatted_statement_periods') @patch('abacus_worksheet.logic.worksheet_adjustment.download_file') @patch('abacus_worksheet.logic.worksheet_adjustment.os') @patch('abacus_worksheet.logic.worksheet_adjustment.pd') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_create_worksheet_adjustments_and_details( mock_model, mock_pandas, mock_os, mock_download_s3_file, mock_formatted_statement_periods, mock_formatted_adjustment_types, mock_formatted_worksheet_adjustment_detail, mock_formatted_worksheet_adjustment, mock_get_expense_details_records, mock_create_worksheet_adjustments, mock_create_worksheet_adjustments_and_details, mock_adjustment_row, ): """Test create_worksheet_adjustments_and_details function. When the file includes data for adjustment expanses as well. """ file = 'file/test.xlsx' statement_period_adjustment_file_id = 1 abacus_event_id = 1 mock_os.remove.return_value = True mock_model.get_statement_period_adjustment_file_by_id.return_value = { 'valid_file_location': 's3://qa-abacus-adjustments/test.xlsx', 'statement_period_id': 282, 'statement_period_adjustment_file_id': 1, 'deleted_by': None, 'deleted_at': None, } adjustments = [ { 'statement_period_adjustment_file_id': 1, 'abacus_event_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 65, 'adjustment_amount': Decimal('1807.556'), 'adjustment_currency_code': 'USD', } ] expense_details = { 'statement_period_adjustment_file_id': 1, 'account_id': '1234', 'contract_id': '1234', 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'amount': Decimal('-90.122'), 'currency_code': 'USD', 'upc': '1234567890', 'distribution_type': 'digital', 'note': None, 'internal_note': None, } mock_download_s3_file.return_value = MagicMock(name=file) mock_pandas.read_excel.return_value.astype.return_value.replace.return_value.dropna.return_value.astype.return_value.where.return_value.iterrows = MagicMock( return_value=enumerate([mock_adjustment_row]) ) mock_pandas.read_excel.return_value.astype.return_value.replace.return_value.dropna.return_value.astype.return_value.where.return_value.empty = False mock_formatted_statement_periods.return_value = [] mock_formatted_adjustment_types.return_value = [] mock_formatted_worksheet_adjustment_detail.return_value = expense_details mock_formatted_worksheet_adjustment.return_value = adjustments mock_get_expense_details_records.return_value = [] mock_create_worksheet_adjustments.return_value = [] mock_create_worksheet_adjustments_and_details.return_value = [] res = worksheet_adjustment.create_worksheet_adjustments_and_details( statement_period_adjustment_file_id, abacus_event_id ) assert res.status == 201 assert res.message == {'message': 'Records inserted succesfully.'} mock_create_worksheet_adjustments.assert_called_once() mock_create_worksheet_adjustments_and_details.assert_called_once() mock_get_expense_details_records.assert_called_once() @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_create_worksheet_adjustments_and_details_invalid_path(mock_model): """Test create_worksheet_adjustments_and_details function for invalid file path.""" statement_period_adjustment_file_id = 1 abacus_event_id = 1 mock_model.get_statement_period_adjustment_file_by_id.return_value = { 'valid_file_location': None, 'statement_period_id': 282, 'statement_period_adjustment_file_id': 1, 'deleted_by': None, 'deleted_at': None, } res = worksheet_adjustment.create_worksheet_adjustments_and_details( statement_period_adjustment_file_id, abacus_event_id ) assert res.status == 404 assert res.errors['message'] == 'Invalid adjustment file path' @patch('abacus_worksheet.logic.worksheet_adjustment.pd') @patch('abacus_worksheet.logic.worksheet_adjustment.download_file') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_create_worksheet_adjustments_and_details_empty_data( mock_model, mock_pandas, mock_download_s3_file ): """Test create_worksheet_adjustments_and_details function for empty data.""" statement_period_adjustment_file_id = 1 abacus_event_id = 1 file = 'file/test.xlsx' mock_model.get_statement_period_adjustment_file_by_id.return_value = { 'valid_file_location': 's3://qa-abacus-adjustments/test.xlsx', 'statement_period_id': 282, 'statement_period_adjustment_file_id': 1, 'deleted_by': None, 'deleted_at': None, } mock_download_s3_file.return_value = MagicMock(name=file) mock_pandas.read_excel.return_value.empty = True res = worksheet_adjustment.create_worksheet_adjustments_and_details( statement_period_adjustment_file_id, abacus_event_id ) assert res.status == 404 assert res.errors['message'] == 'No adjustments records found in excel sheet' @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustmentDetail') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_soft_delete_worksheet_adjustments_and_details( mock_adjustment_file_model, mock_adjustment_detail_model, mock_adjustment_model ): """Test soft_delete_worksheet_adjustments_and_details function.""" statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'deleted_by': 'Test User', 'deleted_at': '02-09-2024', } mock_adjustment_model.get_applied_ids_for_file.return_value = set() mock_adjustment_model.soft_delete_worksheet_adjustments.return_value = True mock_adjustment_detail_model.soft_delete_worksheet_adjustment_details.return_value = True res = worksheet_adjustment.soft_delete_worksheet_adjustments_and_details( statement_period_adjustment_file_id ) assert res.status == 204 mock_adjustment_model.get_applied_ids_for_file.assert_called_once_with( statement_period_adjustment_file_id ) mock_adjustment_detail_model.soft_delete_worksheet_adjustment_details.assert_called_once_with( statement_period_adjustment_file_id ) mock_adjustment_model.soft_delete_worksheet_adjustments.assert_called_once_with( statement_period_adjustment_file_id ) @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustmentDetail') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_soft_delete_worksheet_adjustments_and_details_rejects_applied( mock_adjustment_file_model, mock_adjustment_detail_model, mock_adjustment_model ): """Whole-batch delete refuses to soft-delete when an entry is already applied.""" statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'deleted_by': 'Test User', 'deleted_at': '02-09-2024', } mock_adjustment_model.get_applied_ids_for_file.return_value = {7} res = worksheet_adjustment.soft_delete_worksheet_adjustments_and_details( statement_period_adjustment_file_id ) assert res.status == 409 assert res.errors['message'] == ERROR_ENTRY_APPLIED.format([7]) mock_adjustment_detail_model.soft_delete_worksheet_adjustment_details.assert_not_called() mock_adjustment_model.soft_delete_worksheet_adjustments.assert_not_called() @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustmentDetail') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_soft_delete_worksheet_adjustments_and_details_maps_trigger_1644_to_409( mock_adjustment_file_model, mock_adjustment_detail_model, mock_adjustment_model ): """A trigger errno 1644 during the whole-batch cascade returns 409, not 500.""" statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'deleted_by': 'Test User', 'deleted_at': '02-09-2024', } mock_adjustment_model.get_applied_ids_for_file.return_value = set() err = OperationalError('UPDATE ...', {}, Exception()) err.orig = MagicMock() err.orig.args = (1644, 'applied') mock_adjustment_detail_model.soft_delete_worksheet_adjustment_details.side_effect = err res = worksheet_adjustment.soft_delete_worksheet_adjustments_and_details( statement_period_adjustment_file_id ) assert res.status == 409 @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_soft_delete_worksheet_adjustments_and_details_error( mock_adjustment_file_model, ): """Test soft_delete_worksheet_adjustments_and_details function. When statement_period_adjustment_file is not deleted. """ statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'deleted_by': None, 'deleted_at': None, } res = worksheet_adjustment.soft_delete_worksheet_adjustments_and_details( statement_period_adjustment_file_id ) assert res.status == 400 assert res.errors['message'] == ERROR_DELETE_WORKSHEET_ADJUSTMENTS.format( statement_period_adjustment_file_id ) @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_soft_delete_worksheet_adjustments_file_not_exist(mock_adjustment_file_model): """Test soft_delete_worksheet_adjustments_and_details function. If there is no record for statement_period_adjustment_file. """ statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = None res = worksheet_adjustment.soft_delete_worksheet_adjustments_and_details( statement_period_adjustment_file_id ) assert res.status == 404 assert res.errors['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_get_worksheet_adjustments_and_details_by_file_id( mock_adjustment_file_model, mock_worksheet_adjustment_model ): """Test get_worksheet_adjustments_and_details_by_file_id function.""" mock_worksheet_adjustments = WorksheetAdjustmentFactory.create_batch(2) statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'deleted_by': None, 'deleted_at': None, } mock_worksheet_adjustment_model.get_worksheet_adjustments_and_details.return_value = ( mock_worksheet_adjustments, 2, ) query_params = dict(limit=10, offset=0) res = worksheet_adjustment.get_worksheet_adjustments_and_details_by_file_id( statement_period_adjustment_file_id, query_params ) assert res.status == 200 assert res.message['total_count'] == 2 assert ( res.message['items'][0]['worksheet_adjustment_id'] == mock_worksheet_adjustments[0].worksheet_adjustment_id ) mock_worksheet_adjustment_model.get_worksheet_adjustments_and_details.assert_called_once_with( statement_period_adjustment_file_id, **query_params, is_deleted=False, ) @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_get_worksheet_adjustments_and_details_file_not_exist( mock_adjustment_file_model, ): """Test get_worksheet_adjustments_and_details_by_file_id function.. If there is no record for statement_period_adjustment_file. """ statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = None query_params = dict(limit=10, offset=0) res = worksheet_adjustment.get_worksheet_adjustments_and_details_by_file_id( statement_period_adjustment_file_id, query_params ) assert res.status == 404 assert res.errors['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustmentListSchema') @patch( 'abacus_worksheet.logic.worksheet_adjustment.' 'PendingWorksheetAdjustmentsRequestParamsSchema' ) @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') def test_get_pending_worksheet_adjustments_success( mock_model, mock_load_schema, mock_dump_schema ): """Test get_pending_worksheet_adjustments function.""" mock_worksheet_adjustments = WorksheetAdjustmentFactory.create_batch(2) test_params = { 'limit': 10, 'offset': 0, 'statement_period_id': 1, 'reference_payment_entity_id': 2, } mock_dumped_worksheet_adjustments = WorksheetAdjustmentListSchema(many=True).dump( mock_worksheet_adjustments ) mock_load_schema.return_value.load.return_value = test_params mock_dump_schema.return_value.dump.return_value = mock_dumped_worksheet_adjustments mock_model.get_pending_worksheet_adjustments.return_value = ( mock_worksheet_adjustments, len(mock_worksheet_adjustments), ) res = worksheet_adjustment.get_pending_worksheet_adjustments(test_params) assert mock_load_schema.return_value.load.call_args_list == [call(test_params)] assert mock_model.get_pending_worksheet_adjustments.call_args_list == [ call(**test_params) ] assert mock_dump_schema.return_value.dump.call_args_list == [ call(mock_worksheet_adjustments) ] assert res.status == 200 assert res.message['items'] == mock_dumped_worksheet_adjustments assert res.message['total_count'] == len(mock_worksheet_adjustments) @patch( 'abacus_worksheet.logic.worksheet_adjustment.' 'PendingWorksheetAdjustmentsRequestParamsSchema' ) def test_get_pending_worksheet_adjustments_failure_validation(mock_load_schema): """Test get_pending_worksheet_adjustments function.""" test_params = { 'limit': 10, 'offset': 0, 'statement_period_id': 1, 'reference_payment_entity_id': 2, } mock_load_schema.return_value.load.side_effect = ValidationError('test error') res = worksheet_adjustment.get_pending_worksheet_adjustments(test_params) assert mock_load_schema.return_value.load.call_args_list == [call(test_params)] assert res.status == 400 assert res.errors == {'code': 'error', 'message': 'test error'} @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_get_worksheet_adjustments_contracts_by_file_id( mock_adjustment_file_model, mock_worksheet_adjustment_model ): """Test get_worksheet_adjustments_contracts_by_file_id function.""" mock_worksheet_adjustments = [WorksheetAdjustmentFactory.create()] statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'deleted_by': None, 'deleted_at': None, } mock_worksheet_adjustment_model.get_worksheet_adjustments_contracts_by_file_id.return_value = ( mock_worksheet_adjustments, 1, ) query_params = {'contract_search_term': 'foo-1-1', 'limit': 10, 'offset': 0} result = worksheet_adjustment.get_worksheet_adjustments_contracts_by_file_id( statement_period_adjustment_file_id, query_params ) assert result.status == 200 assert result.message['items'] == [{'contract_id': 1}] mock_worksheet_adjustment_model.get_worksheet_adjustments_contracts_by_file_id.assert_called_once_with( statement_period_adjustment_file_id, **query_params ) @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_get_worksheet_adjustments_contracts_file_not_exist(mock_adjustment_file_model): """Test get_worksheet_adjustments_contracts_by_file_id function.. If there is no record for statement_period_adjustment_file. """ statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = None query_params = dict(limit=10, offset=0) res = worksheet_adjustment.get_worksheet_adjustments_contracts_by_file_id( statement_period_adjustment_file_id, query_params ) assert res.status == 404 assert res.errors['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_get_worksheet_adjustments_accounts_by_file_id( mock_adjustment_file_model, mock_worksheet_adjustment_model ): """Test get_worksheet_adjustments_accounts_by_file_id function.""" mock_worksheet_adjustments = [WorksheetAdjustmentFactory.create()] statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = { 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'deleted_by': None, 'deleted_at': None, } mock_worksheet_adjustment_model.get_worksheet_adjustments_accounts_by_file_id.return_value = ( mock_worksheet_adjustments, 1, ) query_params = {'account_search_term': 'Test', 'limit': 10, 'offset': 0} result = worksheet_adjustment.get_worksheet_adjustments_accounts_by_file_id( statement_period_adjustment_file_id, query_params ) assert result.status == 200 assert result.message['items'] == [{'account_id': 1}] mock_worksheet_adjustment_model.get_worksheet_adjustments_accounts_by_file_id.assert_called_once_with( statement_period_adjustment_file_id, **query_params ) @patch('abacus_worksheet.logic.worksheet_adjustment.StatementPeriodAdjustmentFile') def test_get_worksheet_adjustments_accounts_file_not_exist(mock_adjustment_file_model): """Test get_worksheet_adjustments_accounts_by_file_id function.. If there is no record for statement_period_adjustment_file. """ statement_period_adjustment_file_id = 1 mock_adjustment_file_model.get_statement_period_adjustment_file_by_id.return_value = None query_params = dict(limit=10, offset=0) res = worksheet_adjustment.get_worksheet_adjustments_accounts_by_file_id( statement_period_adjustment_file_id, query_params ) assert res.status == 404 assert res.errors['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) @patch('abacus_worksheet.logic.worksheet_adjustment.WorksheetAdjustment') def test_get_adjustments_by_period_and_type_id(mock_model): """Test get_adjustments_by_period_and_type_id function.""" statement_period_id = 1 reference_adjustment_type_id = 2 mock_adjustments = WorksheetAdjustmentFactory.create_batch(2) mock_model.get_adjustments_by_period_and_type_id.return_value = ( mock_adjustments, 2, 100.00, ) res = worksheet_adjustment.get_adjustments_by_period_and_type_id( statement_period_id, reference_adjustment_type_id ) assert res.status == 200 assert res.message['total_count'] == 2 assert res.message['currency_agnostic_total_amount'] == 100.00 assert ( res.message['items'][0]['worksheet_adjustment_id'] == mock_adjustments[0].worksheet_adjustment_id ) mock_model.get_adjustments_by_period_and_type_id.assert_called_once_with( statement_period_id, reference_adjustment_type_id ) def test_handle_applied_trigger_maps_1644_to_409(): """Errno 1644 from the trigger maps to a 409 applied-entry error.""" err = OperationalError('UPDATE ...', {}, Exception()) err.orig = MagicMock() err.orig.args = (1644, 'Cannot soft-delete applied worksheet_adjustment') res = worksheet_adjustment._handle_applied_trigger_or_raise(err) assert res.status == 409 def test_handle_applied_trigger_reraises_other_operational_errors(): """A non-1644 OperationalError is re-raised, not swallowed.""" err = OperationalError('UPDATE ...', {}, Exception()) err.orig = MagicMock() err.orig.args = (2013, 'Lost connection') with pytest.raises(OperationalError): worksheet_adjustment._handle_applied_trigger_or_raise(err)