"""Test statement period adjustment file logic.""" from datetime import datetime from decimal import Decimal from unittest.mock import MagicMock, call, patch import pytest from abacus_file_upload.tests.conftest import mock_file_upload from abacus_file_upload.tests.utils.factories import FileUploadFactory from royalties.constants.constants import ( STATEMENT_PERIOD_ADJUSTMENT_FILE_USER_ACTIONS, STATEMENT_PERIOD_STATUSES, ) from royalties.constants.error import ERROR_INVALID_ADJUSTMENT_FILE_USER_ACTION from royalties.logic import statement_period_adjustment_file as logic from royalties.tests.utils.factories import ( StatementPeriodAdjustmentFileFactory, StatementPeriodFactory, ) @patch('royalties.logic.statement_period_adjustment_file.create_presigned_url') @patch('royalties.logic.statement_period_adjustment_file.get_s3_client') def test_get_adjustments_template_xlsx(mock_client, mock_url): """Test getting a presigned S3 url to download adjustments template xlsx.""" mock_client.return_value = 'client' res = logic.get_adjustments_template_xlsx() assert res.status == 200 mock_client.assert_called_once() mock_url.assert_called_once_with( 'client', 'qa-abacus-adjustments', 'abacus_adjustments_template/Adjustment_Template.xlsx', ) @patch('royalties.logic.statement_period_adjustment_file.schemas') @patch('royalties.logic.statement_period_adjustment_file.models') def test_create_statement_period_adjustment_file_success( mock_models, mock_schemas, test_app_request ): """Test create statement period adjustment file success.""" mock_statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.build() mock_statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) mock_models.StatementPeriod.get_by_id_or_error.return_value = mock_statement_period params = { 'statement_period_id': mock_statement_period.statement_period_id, 'file_name': 'test_file.csv', 'valid_file_location': 'valid/location', 'invalid_file_location': 'invalid/location', 'valid_row_count': 1, 'invalid_row_count': 2, 'total_file_amount_multicurrency': Decimal('3.000000000001'), 'total_rounded_amount_multicurrency': Decimal('4.02'), 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', 'source_file_upload_id': 1, } mocked_dump_result = { 'statement_period_adjustment_file_id': 1, 'file_name': 'mocked_file_name.csv', 'valid_file_location': 'valid/location', } mock_models.StatementPeriodAdjustmentFile.create.return_value = ( mock_statement_period_adjustment_file ) mock_schema = mock_schemas.StatementPeriodAdjustmentFileDetailSchema.return_value mock_schema.dump.return_value = mocked_dump_result response = logic.create_statement_period_adjustment_file(**params) mock_models.StatementPeriodAdjustmentFile.create.assert_called_once_with(**params) assert response.status == 201 assert response.message == mocked_dump_result @patch('royalties.logic.statement_period_adjustment_file.schemas') @patch('royalties.logic.statement_period_adjustment_file.models') def test_create_statement_period_adjustment_file_with_created_by_success( mock_models, mock_schemas, test_app_request ): """Test create statement period adjustment file with created_by.""" mock_statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.build() mock_statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) mock_models.StatementPeriod.get_by_id_or_error.return_value = mock_statement_period params = { 'statement_period_id': mock_statement_period.statement_period_id, 'file_name': 'test_file.csv', 'valid_file_location': 'valid/location', 'invalid_file_location': 'invalid/location', 'valid_row_count': 1, 'invalid_row_count': 2, 'total_file_amount_multicurrency': Decimal('3.000000000001'), 'total_rounded_amount_multicurrency': Decimal('4.02'), 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', 'source_file_upload_id': 1, 'created_by': '6be406a1-7b6f-4ccd-bb4f-b3254cc1db11', } mocked_dump_result = { 'statement_period_adjustment_file_id': 1, 'file_name': 'mocked_file_name.csv', 'valid_file_location': 'valid/location', } mock_models.StatementPeriodAdjustmentFile.create.return_value = ( mock_statement_period_adjustment_file ) mock_schema = mock_schemas.StatementPeriodAdjustmentFileDetailSchema.return_value mock_schema.dump.return_value = mocked_dump_result response = logic.create_statement_period_adjustment_file(**params) mock_models.StatementPeriodAdjustmentFile.create.assert_called_once_with(**params) assert response.status == 201 assert response.message == mocked_dump_result @patch('royalties.logic.statement_period_adjustment_file.models') def test_create_statement_period_adjustment_file_failure_period_existence( mock_models, test_app_request ): """Test create statement period adjustment file failure on period existence.""" mock_models.StatementPeriod.get_by_id_or_error.side_effect = Exception('Test error') params = { 'statement_period_id': 1, 'file_name': 'test_file.csv', 'valid_file_location': 'valid/location', 'invalid_file_location': 'invalid/location', 'valid_row_count': 1, 'invalid_row_count': 2, 'total_file_amount_multicurrency': Decimal('3.000000000001'), 'total_rounded_amount_multicurrency': Decimal('4.02'), 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } response = logic.create_statement_period_adjustment_file(**params) assert not mock_models.StatementPeriodAdjustmentFile.create.called assert response.status == 400 assert response.errors == {'code': 'error', 'message': 'Test error'} @patch('royalties.logic.statement_period_adjustment_file.models') def test_create_statement_period_adjustment_file_failure_period_state( mock_models, test_app_request ): """Test create statement period adjustment file failure on bad period.""" mock_statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CLOSED ) mock_models.StatementPeriod.get_by_id_or_error.return_value = mock_statement_period params = { 'statement_period_id': mock_statement_period.statement_period_id, 'file_name': 'test_file.csv', 'valid_file_location': 'valid/location', 'invalid_file_location': 'invalid/location', 'valid_row_count': 1, 'invalid_row_count': 2, 'total_file_amount_multicurrency': Decimal('3.000000000001'), 'total_rounded_amount_multicurrency': Decimal('4.02'), 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } response = logic.create_statement_period_adjustment_file(**params) assert not mock_models.StatementPeriodAdjustmentFile.create.called assert response.status == 400 assert response.errors == { 'code': 'error', 'message': 'Selected statement period status must be current', } @patch('royalties.logic.statement_period_adjustment_file.schemas') def test_update_statement_period_adjustment_file_success( mock_schemas, test_app_request ): """Test update statement period adjustment file success.""" mock_statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) mock_file_upload = FileUploadFactory.create() params = { 'file_name': 'new_test_file.csv', 'valid_file_location': 'new_valid/location', 'invalid_file_location': 'new_invalid/location', 'valid_row_count': 2, 'invalid_row_count': 3, 'total_file_amount_multicurrency': Decimal('3.000000000001'), 'total_rounded_amount_multicurrency': Decimal('4.02'), 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', 'source_file_upload_id': mock_file_upload.file_upload_id, } mocked_dump_result = { 'statement_period_adjustment_file_id': 1, 'file_name': 'mocked_file_name.csv', 'valid_file_location': 'valid/location', } mock_schema = mock_schemas.StatementPeriodAdjustmentFileDetailSchema.return_value mock_schema.dump.return_value = mocked_dump_result response = logic.update_statement_period_adjustment_file( mock_statement_period_adjustment_file, **params ) assert mock_schema.dump.call_args_list == [ call(mock_statement_period_adjustment_file) ] assert response.status == 201 assert response.message == mocked_dump_result @patch('royalties.logic.statement_period_adjustment_file.models') def test_update_statement_period_adjustment_file_failure_period_existence( mock_models, test_app_request ): """Test update statement period adjustment file failure on period existence.""" mock_statement_period_adjustment_file = MagicMock() mock_models.StatementPeriod.get_by_id_or_error.side_effect = Exception('Test error') params = { 'statement_period_id': 1, 'file_name': 'test_file.csv', 'valid_file_location': 'valid/location', 'invalid_file_location': 'invalid/location', 'valid_row_count': 1, 'invalid_row_count': 2, 'total_file_amount_multicurrency': Decimal('3.01'), 'total_rounded_amount_multicurrency': Decimal('4.02'), 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', 'source_file_upload_id': 10, } response = logic.update_statement_period_adjustment_file( mock_statement_period_adjustment_file, **params ) assert not mock_statement_period_adjustment_file.update_attributes.called assert not mock_statement_period_adjustment_file.commit_changes.called assert response.status == 400 assert response.errors == {'code': 'error', 'message': 'Test error'} @patch('royalties.logic.statement_period_adjustment_file.models') def test_update_statement_period_adjustment_file_failure_period_state( mock_models, test_app_request ): """Test create statement period adjustment file failure on bad period.""" mock_statement_period_adjustment_file = MagicMock() mock_statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CLOSED ) mock_models.StatementPeriod.get_by_id_or_error.return_value = mock_statement_period params = { 'file_name': 'test_file.csv', 'valid_file_location': 'valid/location', 'invalid_file_location': 'invalid/location', 'valid_row_count': 1, 'invalid_row_count': 2, 'total_file_amount_multicurrency': Decimal('3.01'), 'total_rounded_amount_multicurrency': Decimal('4.02'), 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } response = logic.update_statement_period_adjustment_file( mock_statement_period_adjustment_file, **params ) assert not mock_statement_period_adjustment_file.update_attributes.called assert not mock_statement_period_adjustment_file.commit_changes.called assert response.status == 400 assert response.errors == { 'code': 'error', 'message': 'Selected statement period status must be current', } @patch('royalties.logic.statement_period_adjustment_file.models') def test_delete_statement_period_adjustment_file_success(mock_models, test_app_request): """Test delete statement period adjustment file success.""" mock_statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) mock_models.StatementPeriod.get_by_id_or_error.return_value = ( mock_statement_period_adjustment_file.statement_period ) response = logic.delete_statement_period_adjustment_file( mock_statement_period_adjustment_file ) assert ( mock_models.StatementPeriodAdjustmentFile.delete_by_id_or_error.call_args_list == [ call( mock_statement_period_adjustment_file.statement_period_adjustment_file_id, soft_delete=True, ) ] ) assert response.status == 204 @patch('royalties.logic.statement_period_adjustment_file.models') def test_delete_statement_period_adjustment_file_failure_period_state( mock_models, test_app_request ): """Test delete statement period adjustment file success.""" mock_statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CLOSED ) mock_models.StatementPeriod.get_by_id_or_error.return_value = ( mock_statement_period_adjustment_file.statement_period ) response = logic.delete_statement_period_adjustment_file( mock_statement_period_adjustment_file ) assert not mock_models.StatementPeriodAdjustmentFile.delete_by_id_or_error.called assert response.status == 400 assert response.errors == { 'code': 'error', 'message': 'Selected statement period status must be current', } @patch('royalties.logic.statement_period_adjustment_file.create_presigned_url') @patch('royalties.logic.statement_period_adjustment_file.get_s3_client') @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_adjustment_invalid_report_file(mock_models, mock_client, mock_url): """Test getting a presigned S3 url to download adjustment error file.""" mock_statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( invalid_file_location='s3://qa-abacus-adjustments/errors/Test7.xlsx' ) mock_models.StatementPeriodAdjustmentFile.get_by_id_or_error.return_value = ( mock_statement_period_adjustment_file ) mock_client.return_value = 'client' res = logic.get_adjustment_file_invalid_report( mock_statement_period_adjustment_file.statement_period_adjustment_file_id ) assert res.status == 200 mock_client.assert_called_once() mock_url.assert_called_once_with( 'client', 'qa-abacus-adjustments', 'errors/Test7.xlsx' ) @patch('royalties.logic.statement_period_adjustment_file.get_s3_client') @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_adjustment_invalid_report_file_failure(mock_models, mock_client): """Test getting a presigned S3 url to download adjustment report file.""" adjustment_file = StatementPeriodAdjustmentFileFactory.create( invalid_file_location=None ) mock_models.StatementPeriodAdjustmentFile.get_by_id_or_error.return_value = ( adjustment_file ) mock_client.return_value = 'client' res = logic.get_adjustment_file_invalid_report( adjustment_file.statement_period_adjustment_file_id ) assert res is None @patch('royalties.logic.statement_period_adjustment_file.create_presigned_url') @patch('royalties.logic.statement_period_adjustment_file.get_s3_client') @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_adjustment_valid_report_file(mock_models, mock_client, mock_url): """Test getting a presigned S3 url to download adjustment error file failure.""" mock_statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( valid_file_location='s3://qa-abacus-adjustments/report/Test7.xlsx' ) mock_models.StatementPeriodAdjustmentFile.get_by_id_or_error.return_value = ( mock_statement_period_adjustment_file ) mock_client.return_value = 'client' res = logic.get_adjustment_file_valid_report( mock_statement_period_adjustment_file.statement_period_adjustment_file_id ) assert res.status == 200 mock_client.assert_called_once() mock_url.assert_called_once_with( 'client', 'qa-abacus-adjustments', 'report/Test7.xlsx' ) @patch('royalties.logic.statement_period_adjustment_file.get_s3_client') @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_adjustment_valid_report_file_failure(mock_models, mock_client): """Test getting a presigned S3 url to download adjustment report file failure.""" adjustment_file = StatementPeriodAdjustmentFileFactory.create( valid_file_location=None ) mock_models.StatementPeriodAdjustmentFile.get_by_id_or_error.return_value = ( adjustment_file ) mock_client.return_value = 'client' res = logic.get_adjustment_file_valid_report( adjustment_file.statement_period_adjustment_file_id ) assert res is None @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_statement_period_adjustment_files(mock_models, test_app_request): """Test get_statement_period_adjustment_files function.""" request_params = { 'sort_by': 'statement_period_adjustment_file_id', 'sort_order': 'desc', 'limit': 10, 'offset': 0, } mock_result = [ { 'status': 'approved', 'valid_row_count': 67, 'statement_period_adjustment_file_id': 1, 'created_at': datetime(2023, 10, 20), 'created_by': 'Test User', 'date_approved': datetime(2023, 11, 11), 'approved_by': 'Joe User', 'date_applied': datetime(2023, 11, 20), } ] mock_models.StatementPeriodAdjustmentFile.get_statement_period_adjustment_files.return_value = ( mock_result, 1, ) result = logic.get_statement_period_adjustment_files(request_params) assert result.status == 200 assert result.message['items'] == [ { 'status': 'approved', 'valid_row_count': 67, 'statement_period_adjustment_file_id': 1, 'created_at': '2023-10-20', 'created_by': 'Test User', 'date_approved': '2023-11-11', 'approved_by': 'Joe User', 'date_applied': '2023-11-20', } ] @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_statement_period_adjustment_file_users(mock_models): """Test get_statement_period_adjustment_file_users function.""" user_action = STATEMENT_PERIOD_ADJUSTMENT_FILE_USER_ACTIONS.UPLOADED_FILE mock_result = [ {'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c'}, {'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c'}, ] mock_models.StatementPeriodAdjustmentFile.get_statement_period_adjustment_file_users.return_value = ( mock_result, 2, ) result = logic.get_statement_period_adjustment_file_users(user_action) assert result.status == 200 assert result.message['items'] == [ {'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c'}, {'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c'}, ] assert result.message['total_count'] == 2 mock_models.StatementPeriodAdjustmentFile.get_statement_period_adjustment_file_users.assert_called_once_with( user_action ) @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_adjustment_file_by_source_file_key(mock_models): """Test get_adjustment_file_by_source_file_key when file exists.""" mock_result = {'statement_period_adjustment_file_id': 123} mock_models.StatementPeriodAdjustmentFile.get_by_source_file_key.return_value = ( mock_result ) file_key = 'file-key' result = logic.get_adjustment_file_by_source_file_key(file_key) assert result.status == 200 assert result.message == mock_result mock_models.StatementPeriodAdjustmentFile.get_by_source_file_key.assert_called_once_with( file_key ) @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_adjustment_file_by_source_file_key_not_existing(mock_models): """Test get_adjustment_file_by_source_file_key when file does not exist.""" mock_models.StatementPeriodAdjustmentFile.get_by_source_file_key.return_value = None file_key = 'file-key' result = logic.get_adjustment_file_by_source_file_key(file_key) assert result.status == 404 mock_models.StatementPeriodAdjustmentFile.get_by_source_file_key.assert_called_once_with( file_key ) @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_statement_period_adjustment_file_users_invalid_action(mock_models): """Test get_statement_period_adjustment_file_users function. An error is thrown for an invalid user action """ user_action = 'invalid action' result = logic.get_statement_period_adjustment_file_users(user_action) assert result.status == 400 assert result.errors['message'] == ERROR_INVALID_ADJUSTMENT_FILE_USER_ACTION.format( ', '.join(STATEMENT_PERIOD_ADJUSTMENT_FILE_USER_ACTIONS) ) mock_models.StatementPeriodAdjustmentFile.get_statement_period_adjustment_file_users.assert_not_called() @patch('royalties.logic.statement_period_adjustment_file.validate_adjustments_common') @patch( 'royalties.logic.statement_period_adjustment_file.AdjustmentsValidationSnowflakeExecutor' ) def test_validate_adjustments(sf_mock, validate_mock, adjustments_fixture): """Test validating a list of manual adjustments.""" statement_period_id = 1 sf_executor_mock = MagicMock() sf_mock.return_value.__enter__.return_value = sf_executor_mock validate_mock.return_value = {0: {'ERROR'}} result = logic.validate_adjustments(adjustments_fixture, statement_period_id) assert result.status == 200 assert result.message == {0: ['ERROR']} @patch( 'royalties.logic.statement_period_adjustment_file._validate_statement_period_state' ) @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_in_progress_auto_generated_adjustments( mock_models, mock__validate_statement_period_state ): """Test get_in_progress_auto_generated_adjustments function.""" mock__validate_statement_period_state.return_value = True identity_id = 'effff8ac3-7e51-4793-8775-50d11282504c' statement_period_id = 1 mock_result = [ { 'batch_type': 'auto', 'status': 'generating', 'statement_period_id': statement_period_id, 'statement_period_adjustment_file_id': 1, } ] mock_models.StatementPeriodAdjustmentFile.get_in_progress_auto_generated_adjustments.return_value = mock_result result = logic.get_in_progress_auto_generated_adjustments( statement_period_id, identity_id ) assert result.status == 200 assert result.message == mock_result mock_models.StatementPeriodAdjustmentFile.get_in_progress_auto_generated_adjustments.assert_called_once_with( statement_period_id, identity_id ) @patch( 'royalties.logic.statement_period_adjustment_file._validate_statement_period_state' ) @patch('royalties.logic.statement_period_adjustment_file.models') def test_get_auto_generated_adjustments_validation_failed( mock_models, mock__validate_statement_period_state ): """Test get_in_progress_auto_generated_adjustments function for invalid statement period.""" mock__validate_statement_period_state.side_effect = Exception( 'Invalid Statement Period' ) identity_id = 'effff8ac3-7e51-4793-8775-50d11282504c' statement_period_id = 1 mock_result = { 'batch_type': 'auto', 'status': 'generating', 'statement_period_id': statement_period_id, 'statement_period_adjustment_file_id': 1, } mock_models.StatementPeriodAdjustmentFile.get_in_progress_auto_generated_adjustments.return_value = mock_result result = logic.get_in_progress_auto_generated_adjustments( statement_period_id, identity_id ) assert result.status == 400 assert result.errors['message'] == 'Invalid Statement Period' mock_models.StatementPeriodAdjustmentFile.get_in_progress_auto_generated_adjustments.assert_not_called()