"""Functional test for statement period adjustment file.""" from unittest.mock import patch import pytest from abacus_common_logic.test_utils.helpers import get_json_body from moto import mock_aws from sqlalchemy import bindparam from abacus_file_upload.tests.utils.factories import FileUploadFactory from royalties.constants import error from royalties.constants.constants import ( STATEMENT_PERIOD_ADJUSTMENT_FILE_ERROR_TYPES, STATEMENT_PERIOD_ADJUSTMENT_FILE_USER_ACTIONS, STATEMENT_PERIOD_STATUSES, ) from royalties.tests.utils.factories import ( StatementPeriodAdjustmentFileFactory, StatementPeriodFactory, ) def test_create_statement_period_adjustment_file(fixture_client): """Create adjustment file and get response from the DB.""" statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) json_request = { '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': '3.000000000001', 'total_rounded_amount_multicurrency': '4.02', 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } response = fixture_client.post( f'/statement-period/{statement_period.statement_period_id}/adjustment-file', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 201 assert json_response == { **json_request, 'batch_type': 'upload', 'statement_period_adjustment_file_id': 1, 'statement_period_id': statement_period.statement_period_id, 'source_file_upload_id': None, 'created_by': 'default_user_id', } def test_create_statement_period_adjustment_file_only_required_fields(fixture_client): """Create adjustment file with only required fields and get response from the DB.""" statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) json_request = { 'file_name': 'test_file.csv', } response = fixture_client.post( f'/statement-period/{statement_period.statement_period_id}/adjustment-file', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 201 assert json_response == { **json_request, 'batch_type': 'upload', 'invalid_file_location': None, 'total_rounded_amount_multicurrency': None, 'invalid_row_count': None, 'md5sum': None, 'valid_file_location': None, 'total_file_amount_multicurrency': None, 'valid_row_count': None, 'statement_period_adjustment_file_id': 1, 'statement_period_id': statement_period.statement_period_id, 'error_type': None, 'source_file_upload_id': None, 'created_by': 'default_user_id', } def test_create_statement_period_adjustment_file_with_source_file_id(fixture_client): """Create adjustment file with source file upload id.""" statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) file_upload = FileUploadFactory.create() json_request = { 'file_name': 'test_file.csv', 'source_file_upload_id': file_upload.file_upload_id, } response = fixture_client.post( f'/statement-period/{statement_period.statement_period_id}/adjustment-file', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 201 assert json_response == { **json_request, 'batch_type': 'upload', 'invalid_file_location': None, 'total_rounded_amount_multicurrency': None, 'invalid_row_count': None, 'md5sum': None, 'valid_file_location': None, 'total_file_amount_multicurrency': None, 'valid_row_count': None, 'statement_period_adjustment_file_id': 1, 'statement_period_id': statement_period.statement_period_id, 'error_type': None, 'source_file_upload_id': file_upload.file_upload_id, 'created_by': 'default_user_id', } def test_create_statement_period_adjustment_file_with_created_by(fixture_client): """Create adjustment file with created by.""" statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) json_request = { 'file_name': 'test_file.csv', 'created_by': '6be406a1-7b6f-4ccd-bb4f-b3254cc1db11', } response = fixture_client.post( f'/statement-period/{statement_period.statement_period_id}/adjustment-file', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 201 assert json_response == { **json_request, 'batch_type': 'upload', 'invalid_file_location': None, 'total_rounded_amount_multicurrency': None, 'invalid_row_count': None, 'md5sum': None, 'valid_file_location': None, 'total_file_amount_multicurrency': None, 'valid_row_count': None, 'statement_period_adjustment_file_id': 1, 'statement_period_id': statement_period.statement_period_id, 'error_type': None, 'source_file_upload_id': None, 'created_by': '6be406a1-7b6f-4ccd-bb4f-b3254cc1db11', } def test_create_statement_period_adjustment_file_no_period(fixture_client): """Create adjustment file error with non-existing period.""" json_request = { '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': '3.000000000001', 'total_rounded_amount_multicurrency': '4.02', 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } response = fixture_client.post( '/statement-period/1/adjustment-file', json=json_request ) json_response = get_json_body(response) assert response.status_code == 400 assert json_response == { 'code': 'error', 'message': '400 Bad Request: StatementPeriod 1 does not exist.', } def test_create_statement_period_adjustment_file_invalid_period_state(fixture_client): """Create adjustment file error with non-existing period.""" statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CLOSED ) json_request = { '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': '3.000000000001', 'total_rounded_amount_multicurrency': '4.02', 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } response = fixture_client.post( f'/statement-period/{statement_period.statement_period_id}/adjustment-file', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 400 assert json_response == { 'code': 'error', 'message': 'Selected statement period status must be current', } def test_update_statement_period_adjustment_file(fixture_client): """Update adjustment file and get response from the DB.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) statement_period = statement_period_adjustment_file.statement_period statement_period_adjustment_file_id = ( statement_period_adjustment_file.statement_period_adjustment_file_id ) json_request = { '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': '3.000000000001', 'total_rounded_amount_multicurrency': '4.02', 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } response = fixture_client.put( f'/statement-period/' f'{statement_period.statement_period_id}/adjustment-file/' f'{statement_period_adjustment_file_id}', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 201 assert json_response == { **json_request, 'batch_type': 'upload', 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'statement_period_id': statement_period.statement_period_id, 'source_file_upload_id': None, 'created_by': 'default_user_id', } def test_update_statement_period_adjustment_file_partial(fixture_client): """Update adjustment file and get response from the DB.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) statement_period = statement_period_adjustment_file.statement_period json_request = {'md5sum': 'c6579ec2950296ed6a04f08f67f64422'} response = fixture_client.put( f'/statement-period/' f'{statement_period.statement_period_id}/adjustment-file/' f'{statement_period_adjustment_file.statement_period_adjustment_file_id}', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 201 assert json_response.items() >= json_request.items() def test_update_statement_period_adjustment_file_with_source_file_id(fixture_client): """Update adjustment file with source file upload id.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) file_upload = FileUploadFactory.create() statement_period = statement_period_adjustment_file.statement_period json_request = { 'md5sum': 'c6579ec2950296ed6a04f08f67f64422', 'source_file_upload_id': file_upload.file_upload_id, } response = fixture_client.put( f'/statement-period/' f'{statement_period.statement_period_id}/adjustment-file/' f'{statement_period_adjustment_file.statement_period_adjustment_file_id}', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 201 assert json_response.items() >= json_request.items() def test_update_statement_period_adjustment_file_invalid_period_state(fixture_client): """Update adjustment file error with non-existing period.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CLOSED ) statement_period = statement_period_adjustment_file.statement_period json_request = { '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': '3.000000000001', 'total_rounded_amount_multicurrency': '4.02', 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } response = fixture_client.post( f'/statement-period/{statement_period.statement_period_id}/adjustment-file', json=json_request, ) json_response = get_json_body(response) assert response.status_code == 400 assert json_response == { 'code': 'error', 'message': 'Selected statement period status must be current', } def test_update_statement_period_adjustment_file_row_count_error(fixture_client): """Update adjustment file with row count error.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) file_id = statement_period_adjustment_file.statement_period_adjustment_file_id json_request = { 'error_type': STATEMENT_PERIOD_ADJUSTMENT_FILE_ERROR_TYPES.ROW_COUNT_ERROR, } response = fixture_client.put( f'/statement-period-adjustment-file/{file_id}', json=json_request ) assert response.status_code == 201 assert ( response.json.get('error_type') == statement_period_adjustment_file.error_type ) def test_update_statement_period_adjustment_file_other_endpoint(fixture_client): """PUT /statement-period-adjustment-file/.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) file_id = statement_period_adjustment_file.statement_period_adjustment_file_id json_request = { 'error_type': STATEMENT_PERIOD_ADJUSTMENT_FILE_ERROR_TYPES.FORMAT_ERROR } response = fixture_client.put( f'/statement-period-adjustment-file/{file_id}', json=json_request ) assert response.status_code == 201 assert ( response.json.get('error_type') == statement_period_adjustment_file.error_type ) def test_get_statement_period_adjustment_file_by_id(fixture_client): """Get a statement_period_adjustment_file by id.""" statement_period = StatementPeriodFactory.create() adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period_adjustment_file_id=80, statement_period=statement_period ) response = fixture_client.get( f'/statement-period-adjustment-file/{adjustment_file.statement_period_adjustment_file_id}/' ) assert response.status_code == 200 assert response.json == { 'batch_type': 'upload', 'total_rounded_amount_multicurrency': None, 'valid_row_count': None, 'invalid_file_location': None, 'valid_file_location': None, 'invalid_row_count': None, 'file_name': 'test_file_name.xlsx', 'total_file_amount_multicurrency': None, 'md5sum': None, 'error_type': None, 'statement_period_id': 1, 'statement_period_adjustment_file_id': 80, 'source_file_upload_id': None, 'created_by': 'default_user_id', } def test_get_statement_period_adjustment_file_by_file_upload_key(fixture_client): """Get a statement_period_adjustment_file by source file key.""" statement_period = StatementPeriodFactory.create() file_upload = FileUploadFactory.create(file_key='file-key') adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period_adjustment_file_id=80, statement_period=statement_period, source_file_upload_id=file_upload.file_upload_id, ) response = fixture_client.get( f'/statement-period-adjustment-file/by-source-file-key/{file_upload.file_key}/' ) assert response.status_code == 200 assert response.json == { 'batch_type': 'upload', 'total_rounded_amount_multicurrency': None, 'valid_row_count': None, 'invalid_file_location': None, 'valid_file_location': None, 'invalid_row_count': None, 'file_name': 'test_file_name.xlsx', 'total_file_amount_multicurrency': None, 'md5sum': None, 'error_type': None, 'statement_period_id': 1, 'statement_period_adjustment_file_id': 80, 'source_file_upload_id': 1, 'created_by': 'default_user_id', } def test_list_statement_period_adjustment_file_success(fixture_client): """List adjustment files.""" adjustment_file = StatementPeriodAdjustmentFileFactory.create() response = fixture_client.get( f'/statement-period/{adjustment_file.statement_period_id}/adjustment-files', ) json_response = get_json_body(response) assert response.status_code == 200 assert json_response == { 'items': [ { 'batch_type': 'upload', 'file_name': adjustment_file.file_name, 'invalid_file_location': adjustment_file.invalid_file_location, 'total_file_amount_multicurrency': adjustment_file.total_file_amount_multicurrency, 'invalid_row_count': adjustment_file.invalid_row_count, 'md5sum': adjustment_file.md5sum, 'statement_period_adjustment_file_id': adjustment_file.statement_period_adjustment_file_id, 'statement_period_id': adjustment_file.statement_period_id, 'valid_file_location': adjustment_file.valid_file_location, 'total_rounded_amount_multicurrency': adjustment_file.total_rounded_amount_multicurrency, 'valid_row_count': adjustment_file.valid_row_count, 'error_type': adjustment_file.error_type, 'source_file_upload_id': adjustment_file.source_file_upload_id, 'created_by': adjustment_file.created_by, } ], 'total_count': 1, } def test_delete_statement_period_adjustment_file_success(fixture_client): """Delete adjustment file and get response from the DB.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) statement_period = statement_period_adjustment_file.statement_period response = fixture_client.delete( f'/statement-period/' f'{statement_period.statement_period_id}/adjustment-file/' f'{statement_period_adjustment_file.statement_period_adjustment_file_id}' ) assert response.status_code == 204 def test_delete_statement_period_adjustment_file_invalid_period_state(fixture_client): """Delete adjustment file error with incorrect period.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CLOSED ) statement_period = statement_period_adjustment_file.statement_period response = fixture_client.delete( f'/statement-period/{statement_period.statement_period_id}/adjustment-file/' f'{statement_period_adjustment_file.statement_period_adjustment_file_id}' ) json_response = get_json_body(response) assert response.status_code == 400 assert json_response == { 'code': 'error', 'message': 'Selected statement period status must be current', } @mock_aws def test_get_abacus_adjustments_template_download(fixture_client): """Get s3 link to download abacus adjustments template.""" res = fixture_client.get('/abacus-adjustments/download/template') urlSubstring = '-abacus-adjustments.s3.amazonaws.com/' assert urlSubstring in res.json['template_url'] @mock_aws def test_get_invalid_adjustment_file_download(fixture_client): """Get s3 link to download invalid adjustment file.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( invalid_file_location='s3://qa-abacus-adjustments/test_fixtures/' 'test_adjustment.xlsx' ) adjustment_id = statement_period_adjustment_file.statement_period_adjustment_file_id res = fixture_client.get( f'/statement-period-adjustment-file/{adjustment_id}/download/error' ) urlSubstring = '-abacus-adjustments.s3.amazonaws.com/' assert urlSubstring in res.json['invalid_report_url'] @mock_aws def test_get_valid_adjustment_file_download(fixture_client): """Get s3 link to download valid adjustment file.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( valid_file_location='s3://qa-abacus-adjustments/test_fixtures/' 'test_adjustment.xlsx' ) adjustment_id = statement_period_adjustment_file.statement_period_adjustment_file_id res = fixture_client.get( f'/statement-period-adjustment-file/{adjustment_id}/download/report' ) urlSubstring = '-abacus-adjustments.s3.amazonaws.com/' assert urlSubstring in res.json['valid_report_url'] @mock_aws def test_get_invalid_adjustment_file_download_failure(fixture_client): """Get s3 link to download invalid adjustment file failure.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( invalid_file_location=None ) adjustment_id = statement_period_adjustment_file.statement_period_adjustment_file_id res = fixture_client.get( f'/statement-period-adjustment-file/{adjustment_id}/download/error' ) assert res.status_code == 400 response_data = (res.data).decode('utf-8') assert ( error.ERROR_STATEMENT_PERIOD_INVALID_REPORT_NOT_FOUND.format( statement_period_adjustment_file_id=statement_period_adjustment_file.statement_period_adjustment_file_id ) in response_data ) @mock_aws def test_get_valid_adjustment_file_download_failure(fixture_client): """Get s3 link to download valid adjustment file failure.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( valid_file_location=None ) adjustment_id = statement_period_adjustment_file.statement_period_adjustment_file_id res = fixture_client.get( f'/statement-period-adjustment-file/{adjustment_id}/download/report' ) assert res.status_code == 400 response_data = (res.data).decode('utf-8') assert ( error.ERROR_STATEMENT_PERIOD_VALID_REPORT_NOT_FOUND.format( statement_period_adjustment_file_id=statement_period_adjustment_file.statement_period_adjustment_file_id ) in response_data ) @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_statement_period_adjustment_files( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files.""" mock_is_ff_enabled.return_value = False result = fixture_client.get( '/statement-period-adjustment-files/?limit=10&offset=0&sort_by=status' ) assert result.status_code == 200 assert result.json == { 'items': [ { 'status': 'not_approved', 'valid_row_count': None, 'statement_period_id': mock_statement_period_adjustment_files[ 1 ].statement_period_id, 'statement_period_adjustment_file_id': 91, 'file_name': 'test_file_name.xlsx', 'total_rounded_amount_multicurrency': None, 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': None, 'approved_by': None, 'date_applied': None, 'applied_by': None, }, { 'status': 'approved', 'valid_row_count': None, 'statement_period_id': mock_statement_period_adjustment_files[ 2 ].statement_period_id, 'statement_period_adjustment_file_id': 92, 'file_name': 'adjustment-file-1-2024.xlsx', 'total_rounded_amount_multicurrency': '15.00', 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': '2023-11-19', 'approved_by': 'Test User', 'date_applied': None, 'applied_by': None, }, { 'status': 'applied', 'valid_row_count': None, 'statement_period_id': mock_statement_period_adjustment_files[ 3 ].statement_period_id, 'statement_period_adjustment_file_id': 93, 'file_name': 'adjustment-file-2-2024.xlsx', 'total_rounded_amount_multicurrency': '20.50', 'created_at': '2023-10-21', 'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c', 'date_approved': '2023-12-14', 'approved_by': 'Test User', 'date_applied': '2023-12-15', 'applied_by': 'Test Apply User', }, ], 'total_count': 3, } @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_statement_period_adjustment_files_ff_enabled( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files when ff is enabled.""" mock_is_ff_enabled.return_value = True result = fixture_client.get( '/statement-period-adjustment-files/?limit=220&offset=0&sort_by=statement_period_adjustment_file_id&sort_order=asc' ) assert result.status_code == 200 assert result.json == { 'items': [ { 'batch_type': 'upload', 'status': 'not_approved', 'valid_row_count': None, 'statement_period_id': 998, 'statement_period_adjustment_file_id': 91, 'file_name': 'test_file_name.xlsx', 'total_rounded_amount_multicurrency': None, 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': None, 'approved_by': None, 'date_applied': None, 'applied_by': None, }, { 'batch_type': 'upload', 'status': 'approved', 'valid_row_count': None, 'statement_period_id': 999, 'statement_period_adjustment_file_id': 92, 'file_name': 'adjustment-file-1-2024.xlsx', 'total_rounded_amount_multicurrency': '15.00', 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': '2023-11-19', 'approved_by': 'Test User', 'date_applied': None, 'applied_by': None, }, { 'batch_type': 'upload', 'status': 'applied', 'valid_row_count': None, 'statement_period_id': 999, 'statement_period_adjustment_file_id': 93, 'file_name': 'adjustment-file-2-2024.xlsx', 'total_rounded_amount_multicurrency': '20.50', 'created_at': '2023-10-21', 'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c', 'date_approved': '2023-12-14', 'approved_by': 'Test User', 'date_applied': '2023-12-15', 'applied_by': 'Test Apply User', }, { 'batch_type': 'auto', 'applied_by': None, 'approved_by': None, 'created_at': '2024-02-01', 'created_by': 'effff8ac3-7e51-4793-8775-50d11282504c', 'date_applied': None, 'date_approved': None, 'file_name': 'adjustment-file-1001-2026.xlsx', 'statement_period_adjustment_file_id': 1001, 'statement_period_id': 998, 'status': 'not_approved', 'total_rounded_amount_multicurrency': None, 'valid_row_count': None, }, { 'batch_type': 'auto', 'applied_by': None, 'approved_by': None, 'created_at': '2024-02-01', 'created_by': 'effff8ac3-7e51-4793-8775-50d11282504c', 'date_applied': None, 'date_approved': None, 'file_name': 'adjustment-file-1002-2026.xlsx', 'statement_period_adjustment_file_id': 1002, 'statement_period_id': 998, 'status': 'failed_to_generate', 'total_rounded_amount_multicurrency': None, 'valid_row_count': None, }, { 'batch_type': 'auto', 'applied_by': None, 'approved_by': None, 'created_at': '2024-02-01', 'created_by': 'effff8ac3-7e51-4793-8775-50d11282504c', 'date_applied': None, 'date_approved': None, 'file_name': 'adjustment-file-1003-2026.xlsx', 'statement_period_adjustment_file_id': 1003, 'statement_period_id': 998, 'status': 'generating', 'total_rounded_amount_multicurrency': None, 'valid_row_count': None, }, { 'batch_type': 'auto', 'applied_by': None, 'approved_by': None, 'created_at': '2024-02-01', 'created_by': 'effff8ac3-7e51-4793-8775-50d11282504c', 'date_applied': None, 'date_approved': None, 'file_name': 'adjustment-file-1004-2026.xlsx', 'statement_period_adjustment_file_id': 1004, 'statement_period_id': 998, 'status': 'no_records', 'total_rounded_amount_multicurrency': None, 'valid_row_count': None, }, ], 'total_count': 7, } def test_get_statement_period_adjustment_files_sort_by_created_at( mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files sorted by created at.""" result = fixture_client.get( '/statement-period-adjustment-files/?limit=10&offset=0&sort_by=created_at' ) assert result.status_code == 200 @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_adjustment_files_filter_by_file_id( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files. Filter by statement_period_adjustment_file_id. """ mock_is_ff_enabled.return_value = True query_params = ( 'limit=10&offset=0&sort_by=status&statement_period_adjustment_file_id=92' ) result = fixture_client.get(f'/statement-period-adjustment-files/?{query_params}') assert result.status_code == 200 assert result.json == { 'items': [ { 'batch_type': 'upload', 'status': 'approved', 'valid_row_count': None, 'statement_period_adjustment_file_id': 92, 'statement_period_id': mock_statement_period_adjustment_files[ 2 ].statement_period_id, 'file_name': 'adjustment-file-1-2024.xlsx', 'total_rounded_amount_multicurrency': '15.00', 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': '2023-11-19', 'approved_by': 'Test User', 'date_applied': None, 'applied_by': None, } ], 'total_count': 1, } @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_adjustment_files_filter_by_status( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files by status filter.""" mock_is_ff_enabled.return_value = True query_params = 'limit=10&offset=0&sort_by=status&status=applied' result = fixture_client.get(f'/statement-period-adjustment-files/?{query_params}') assert result.status_code == 200 assert result.json == { 'items': [ { 'batch_type': 'upload', 'status': 'applied', 'valid_row_count': None, 'statement_period_adjustment_file_id': 93, 'statement_period_id': mock_statement_period_adjustment_files[ 3 ].statement_period_id, 'file_name': 'adjustment-file-2-2024.xlsx', 'total_rounded_amount_multicurrency': '20.50', 'created_at': '2023-10-21', 'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c', 'date_approved': '2023-12-14', 'approved_by': 'Test User', 'date_applied': '2023-12-15', 'applied_by': 'Test Apply User', } ], 'total_count': 1, } @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_adjustment_files_filter_by_status_and_file_id( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files. Filter by status and statement_period_adjustment_file_id. """ mock_is_ff_enabled.return_value = True query_params = 'limit=10&offset=0&sort_by=status&status=not_approved&statement_period_adjustment_file_id=91' result = fixture_client.get(f'/statement-period-adjustment-files/?{query_params}') assert result.status_code == 200 assert result.json == { 'items': [ { 'batch_type': 'upload', 'status': 'not_approved', 'valid_row_count': None, 'statement_period_adjustment_file_id': 91, 'statement_period_id': mock_statement_period_adjustment_files[ 1 ].statement_period_id, 'file_name': 'test_file_name.xlsx', 'total_rounded_amount_multicurrency': None, 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': None, 'approved_by': None, 'date_applied': None, 'applied_by': None, } ], 'total_count': 1, } def test_get_adjustment_files_invalid_status_filter( mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files for invalid status filter.""" query_params = 'limit=10&offset=0&sort_by=status&status=test,applied&statement_period_adjustment_file_id=91' result = fixture_client.get(f'/statement-period-adjustment-files/?{query_params}') assert result.status_code == 400 assert result.json['message'] == str( {'_schema': ['Must be applied, approved, not_approved, failed_to_generate']} ) def test_get_adjustment_file_by_file_id( mock_statement_period_adjustment_files, fixture_client ): """Test to get adjustment file by id.""" result = fixture_client.get('/statement-period-adjustment-file/93') assert result.status_code == 200 assert result.json == { 'batch_type': 'upload', 'error_type': None, 'file_name': 'adjustment-file-2-2024.xlsx', 'invalid_file_location': None, 'invalid_row_count': None, 'md5sum': None, 'statement_period_adjustment_file_id': 93, 'statement_period_id': 999, 'total_file_amount_multicurrency': None, 'total_rounded_amount_multicurrency': '20.50', 'valid_file_location': None, 'valid_row_count': None, 'source_file_upload_id': None, 'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c', } def test_get_deleted_adjustment_file_by_file_id( mock_statement_period_adjustment_files, fixture_client ): """Test to get deleted adjustment file by id.""" result = fixture_client.get('/statement-period-adjustment-file/94') assert result.status_code == 404 assert result.json['message'] == 'StatementPeriodAdjustmentFile 94 does not exist.' def test_update_statement_period_adjustment_file_by_id(fixture_client): """Test to update adjustment file by adjustment file id.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) statement_period = statement_period_adjustment_file.statement_period statement_period_adjustment_file_id = ( statement_period_adjustment_file.statement_period_adjustment_file_id ) json_request = { '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': '3.000000000001', 'total_rounded_amount_multicurrency': '4.02', 'md5sum': 'b6579ec2950296ed6a04f08f67f64422', 'error_type': 'content_error', } result = fixture_client.put( f'/statement-period-adjustment-file/{statement_period_adjustment_file_id}', json=json_request, ) assert result.status_code == 201 assert result.json == { **json_request, 'batch_type': 'upload', 'statement_period_adjustment_file_id': statement_period_adjustment_file_id, 'statement_period_id': statement_period.statement_period_id, 'source_file_upload_id': None, 'created_by': 'default_user_id', } def test_update_deleted_statement_period_adjustment_file_by_id( mock_statement_period_adjustment_files, fixture_client ): """Test to update the deleted adjustment file by adjustment file id.""" json_request = {'file_name': 'new_test_file.csv'} result = fixture_client.put( '/statement-period-adjustment-file/94', json=json_request ) assert result.status_code == 404 assert result.json['message'] == 'StatementPeriodAdjustmentFile 94 does not exist.' def test_delete_statement_period_adjustment_file_by_id(fixture_client): """Test to delete adjustment file by adjustment file id.""" statement_period_adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period__statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) statement_period_adjustment_file_id = ( statement_period_adjustment_file.statement_period_adjustment_file_id ) assert statement_period_adjustment_file.deleted_by is None assert statement_period_adjustment_file.deleted_at is None result = fixture_client.delete( f'/statement-period-adjustment-file/{statement_period_adjustment_file_id}' ) assert result.status_code == 204 assert statement_period_adjustment_file.deleted_by is not None assert statement_period_adjustment_file.deleted_at is not None def test_delete_already_deleted_statement_period_adjustment_file( mock_statement_period_adjustment_files, fixture_client ): """Test to delete adjustment file that are already deleted.""" result = fixture_client.delete('/statement-period-adjustment-file/94') assert result.status_code == 404 assert result.json['message'] == 'StatementPeriodAdjustmentFile 94 does not exist.' @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_adjustment_files_filter_by_file_name( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files by file_name filter.""" mock_is_ff_enabled.return_value = True query_params = 'limit=10&offset=0&sort_by=status&file_name=2024&status=approved' result = fixture_client.get(f'/statement-period-adjustment-files/?{query_params}') assert result.status_code == 200 assert result.json == { 'items': [ { 'batch_type': 'upload', 'status': 'approved', 'valid_row_count': None, 'statement_period_id': mock_statement_period_adjustment_files[ 2 ].statement_period_id, 'statement_period_adjustment_file_id': 92, 'file_name': 'adjustment-file-1-2024.xlsx', 'total_rounded_amount_multicurrency': '15.00', 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': '2023-11-19', 'approved_by': 'Test User', 'date_applied': None, 'applied_by': None, } ], 'total_count': 1, } @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_adjustment_files_filter_by_statement_period( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files by statement_period_id.""" mock_is_ff_enabled.return_value = True query_params = 'limit=10&offset=0&statement_period_id=999&file_name=adjustment' result = fixture_client.get(f'/statement-period-adjustment-files/?{query_params}') assert result.status_code == 200 assert result.json == { 'items': [ { 'batch_type': 'upload', 'status': 'applied', 'valid_row_count': None, 'statement_period_id': 999, 'statement_period_adjustment_file_id': 93, 'file_name': 'adjustment-file-2-2024.xlsx', 'total_rounded_amount_multicurrency': '20.50', 'created_at': '2023-10-21', 'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c', 'date_approved': '2023-12-14', 'approved_by': 'Test User', 'date_applied': '2023-12-15', 'applied_by': 'Test Apply User', }, { 'batch_type': 'upload', 'status': 'approved', 'valid_row_count': None, 'statement_period_id': 999, 'statement_period_adjustment_file_id': 92, 'file_name': 'adjustment-file-1-2024.xlsx', 'total_rounded_amount_multicurrency': '15.00', 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': '2023-11-19', 'approved_by': 'Test User', 'date_applied': None, 'applied_by': None, }, ], 'total_count': 2, } @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_adjustment_files_filter_by_created_by( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get statement period adjustment files by created_by filter.""" mock_is_ff_enabled.return_value = True query_params = 'limit=10&offset=0&sort_by=status&created_by=e5ca8bc3-7e52-4793-8775-50d11282504c&status=approved' result = fixture_client.get(f'/statement-period-adjustment-files/?{query_params}') assert result.status_code == 200 assert result.json == { 'items': [ { 'batch_type': 'upload', 'status': 'approved', 'valid_row_count': None, 'statement_period_id': mock_statement_period_adjustment_files[ 2 ].statement_period_id, 'statement_period_adjustment_file_id': 92, 'file_name': 'adjustment-file-1-2024.xlsx', 'total_rounded_amount_multicurrency': '15.00', 'created_at': '2023-10-21', 'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c', 'date_approved': '2023-11-19', 'approved_by': 'Test User', 'date_applied': None, 'applied_by': None, } ], 'total_count': 1, } def test_get_statement_period_adjustment_file_users( mock_statement_period_adjustment_files, fixture_client ): """Test to get the list of users by user action uploaded-file.""" result = fixture_client.get('/statement-period-adjustment-file/users/uploaded-file') assert result.status_code == 200 assert result.json == { 'items': [ { 'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c', }, {'created_by': 'e5ca8bc3-7e52-4793-8775-50d11282504c'}, ], 'total_count': 2, } def test_get_statement_period_adjustment_file_users_by_invalid_action( mock_statement_period_adjustment_files, fixture_client ): """Test to get the list of users by invalid user action.""" result = fixture_client.get('/statement-period-adjustment-file/users/test-file') assert result.status_code == 400 assert result.json[ 'message' ] == error.ERROR_INVALID_ADJUSTMENT_FILE_USER_ACTION.format( ', '.join(STATEMENT_PERIOD_ADJUSTMENT_FILE_USER_ACTIONS) ) @pytest.mark.db('mysql') @patch('royalties.models.statement_period_adjustment_file.func') def test_get_in_progress_auto_generated_adjustments( mock_utc_func, mock_statement_period_adjustment_files, fixture_client ): """Test to get in-progress auto generated adjustments file.""" mock_utc_func.utc_timestamp.return_value = bindparam( 'mock_now', value='2024-02-01 05:16:00' ) statement_period_id = 998 result = fixture_client.get( f'/statement-period/{statement_period_id}/adjustments/auto-generation/progress', headers={'Orchard-Identity-Id': 'effff8ac3-7e51-4793-8775-50d11282504c'}, ) assert result.status_code == 200 assert result.json == [ { 'batch_type': 'auto', 'status': 'generating', 'statement_period_id': statement_period_id, 'statement_period_adjustment_file_id': 1003, }, { 'batch_type': 'auto', 'status': 'failed_to_generate', 'statement_period_id': statement_period_id, 'statement_period_adjustment_file_id': 1002, }, ] @pytest.mark.db('mysql') def test_get_auto_generated_adjustments_validation_failed( mock_statement_period_adjustment_files, fixture_client ): """Test get_in_progress_auto_generated_adjustments function for invalid statement period.""" statement_period_id = 999 result = fixture_client.get( f'/statement-period/{statement_period_id}/adjustments/auto-generation/progress' ) assert result.status_code == 400 assert result.json['message'] == 'Selected statement period status must be current' @pytest.mark.db('mysql') @patch( 'royalties.models.statement_period_adjustment_file.is_abacus_auto_generate_adjustments_flowthrough_ff_enabled' ) def test_get_failed_and_applied_statement_period_adjustment_files( mock_is_ff_enabled, mock_statement_period_adjustment_files, fixture_client ): """Test to get adjustment files that are applied and failed to generate.""" mock_is_ff_enabled.return_value = True query_params = 'limit=10&offset=0&status=failed_to_generate,applied' result = fixture_client.get(f'/statement-period-adjustment-files/?{query_params}') assert result.status_code == 200 assert result.json == { 'items': [ { 'batch_type': 'auto', 'applied_by': None, 'approved_by': None, 'created_at': '2024-02-01', 'created_by': 'effff8ac3-7e51-4793-8775-50d11282504c', 'date_applied': None, 'date_approved': None, 'file_name': 'adjustment-file-1002-2026.xlsx', 'statement_period_adjustment_file_id': 1002, 'statement_period_id': 998, 'status': 'failed_to_generate', 'total_rounded_amount_multicurrency': None, 'valid_row_count': None, }, { 'batch_type': 'upload', 'applied_by': 'Test Apply User', 'approved_by': 'Test User', 'created_at': '2023-10-21', 'created_by': 'd5ca8ac3-7e51-4793-8775-50d11282504c', 'date_applied': '2023-12-15', 'date_approved': '2023-12-14', 'file_name': 'adjustment-file-2-2024.xlsx', 'statement_period_adjustment_file_id': 93, 'statement_period_id': 999, 'status': 'applied', 'total_rounded_amount_multicurrency': '20.50', 'valid_row_count': None, }, ], 'total_count': 2, }