"""Functional tests for worksheet adjustments.""" import os from decimal import Decimal from unittest.mock import patch from abacus_common_logic.constants.error import ERROR_ENTITY_DOES_NOT_EXIST from abacus_worksheet.api import db from abacus_worksheet.constants import constants from abacus_worksheet.constants.error import ( ERROR_DELETE_WORKSHEET_ADJUSTMENTS, ERROR_INVALID_ACCOUNT_IDS, ERROR_INVALID_CONTRACT_IDS, ) from abacus_worksheet.models.worksheet_adjustment import WorksheetAdjustment from abacus_worksheet.models.worksheet_adjustment_detail import ( WorksheetAdjustmentDetail, ) from tests.utils.factories import ( WorksheetAdjustmentDetailFactory, WorksheetAdjustmentFactory, ) def test_get_worksheet_adjustments(fixture_client): """Test to get worksheet adjustments by statement_period_adjustment_file_id.""" statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=statement_period_adjustment_file_id, deleted_by='Test', deleted_at='2019-03-01 00:00:00', ) mock_worksheet_adjustment = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=statement_period_adjustment_file_id, ) res = fixture_client.get( f'/worksheet-adjustments/statement-period-adjustment-file/{statement_period_adjustment_file_id}/?limit=2&offset=0' ) assert res.status_code == 200 assert res.json['total_count'] == 1 assert res.json['items'] == [ { 'worksheet_adjustment_id': mock_worksheet_adjustment.worksheet_adjustment_id, 'statement_period_adjustment_file_id': mock_worksheet_adjustment.statement_period_adjustment_file_id, 'abacus_event_id': mock_worksheet_adjustment.abacus_event_id, 'account_id': mock_worksheet_adjustment.account_id, 'contract_id': mock_worksheet_adjustment.contract_id, 'activity_statement_period_id': mock_worksheet_adjustment.activity_statement_period_id, 'apply_to_statement_period_id': mock_worksheet_adjustment.apply_to_statement_period_id, 'reference_adjustment_type_id': mock_worksheet_adjustment.reference_adjustment_type_id, 'apply_to_flowthrough_payment': None, 'adjustment_amount': str(mock_worksheet_adjustment.adjustment_amount), 'adjustment_currency_code': mock_worksheet_adjustment.adjustment_currency_code, 'account_currency_code': mock_worksheet_adjustment.account_currency_code, 'account_payment_entity_id': mock_worksheet_adjustment.account_payment_entity_id, 'note': mock_worksheet_adjustment.note, 'internal_note': None, 'details': [], } ] def test_get_worksheet_adjustments_with_details_fields(fixture_client): """Test to get worksheet adjustments with details by statement_period_adjustment_file_id.""" statement_period_adjustment_file_id = 1 mock_worksheet_adjustment = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=statement_period_adjustment_file_id, ) mock_worksheet_adjustment_detail = WorksheetAdjustmentDetailFactory.create( worksheet_adjustment=mock_worksheet_adjustment ) res = fixture_client.get( f'/worksheet-adjustments/statement-period-adjustment-file/{statement_period_adjustment_file_id}/?limit=2&offset=0' ) assert res.status_code == 200 assert res.json['total_count'] == 1 assert res.json['items'] == [ { 'worksheet_adjustment_id': mock_worksheet_adjustment.worksheet_adjustment_id, 'statement_period_adjustment_file_id': mock_worksheet_adjustment.statement_period_adjustment_file_id, 'abacus_event_id': mock_worksheet_adjustment.abacus_event_id, 'account_id': mock_worksheet_adjustment.account_id, 'contract_id': mock_worksheet_adjustment.contract_id, 'activity_statement_period_id': mock_worksheet_adjustment.activity_statement_period_id, 'apply_to_statement_period_id': mock_worksheet_adjustment.apply_to_statement_period_id, 'reference_adjustment_type_id': mock_worksheet_adjustment.reference_adjustment_type_id, 'apply_to_flowthrough_payment': None, 'adjustment_amount': str(mock_worksheet_adjustment.adjustment_amount), 'adjustment_currency_code': mock_worksheet_adjustment.adjustment_currency_code, 'account_currency_code': mock_worksheet_adjustment.account_currency_code, 'account_payment_entity_id': mock_worksheet_adjustment.account_payment_entity_id, 'note': mock_worksheet_adjustment.note, 'internal_note': None, 'details': [ { 'account_id': mock_worksheet_adjustment_detail.account_id, 'activity_statement_period_id': mock_worksheet_adjustment_detail.activity_statement_period_id, 'amount': str(mock_worksheet_adjustment_detail.amount), 'apply_to_statement_period_id': mock_worksheet_adjustment_detail.apply_to_statement_period_id, 'contract_id': mock_worksheet_adjustment_detail.contract_id, 'currency_code': mock_worksheet_adjustment_detail.currency_code, 'distribution_type': mock_worksheet_adjustment_detail.distribution_type, 'internal_note': mock_worksheet_adjustment_detail.internal_note, 'note': mock_worksheet_adjustment_detail.note, 'reference_adjustment_type_id': mock_worksheet_adjustment_detail.reference_adjustment_type_id, 'apply_to_flowthrough_payment': None, 'statement_period_adjustment_file_id': mock_worksheet_adjustment_detail.statement_period_adjustment_file_id, 'upc': mock_worksheet_adjustment_detail.upc, 'worksheet_adjustment_detail_id': mock_worksheet_adjustment_detail.worksheet_adjustment_detail_id, 'worksheet_adjustment_id': mock_worksheet_adjustment.worksheet_adjustment_id, } ], } ] @patch('abacus_worksheet.logic.worksheet_adjustment.download_file') @patch('abacus_worksheet.logic.worksheet_adjustment.os') def test_create_worksheet_adjustments_and_adjustment_details( mock_os, mock_download_s3_file, fixture_client ): """Test to create worksheet adjustments and worksheet adjustment details.""" file = os.path.abspath('tests/utils/test_adjustment_files/valid_adjustments.xlsx') mock_os.remove.return_value = False mock_download_s3_file.return_value = file mock_post_request_body = { 'statement_period_adjustment_file_id': 1, 'abacus_event_id': 1, } res = fixture_client.post( '/worksheet-adjustments/import/', json=mock_post_request_body ) worksheet_adjustments = WorksheetAdjustment.query.all() worksheet_adjustment_details = WorksheetAdjustmentDetail.query.all() assert res.status_code == 201 assert res.json['message'] == 'Records inserted succesfully.' assert len(worksheet_adjustments) == 2 assert worksheet_adjustments[0].adjustment_amount == Decimal('-90.10') assert worksheet_adjustments[1].adjustment_amount == Decimal('8580.51') assert worksheet_adjustments[0].reference_adjustment_type_id == 1 assert worksheet_adjustments[1].reference_adjustment_type_id == 65 assert len(worksheet_adjustment_details) == 2 assert worksheet_adjustment_details[0].amount == Decimal('7819.12') assert worksheet_adjustment_details[1].amount == Decimal('761.39') assert worksheet_adjustment_details[0].upc == '4765834587876' assert worksheet_adjustment_details[1].upc == '20286137634' @patch('abacus_worksheet.logic.worksheet_adjustment.download_file') @patch('abacus_worksheet.logic.worksheet_adjustment.os') def test_create_worksheet_adjustments(mock_os, mock_download_s3_file, fixture_client): """Test to create worksheet adjustments.""" file = os.path.abspath( 'tests/utils/test_adjustment_files/valid_adjustments_no_expense_details.xlsx' ) mock_os.remove.return_value = False mock_download_s3_file.return_value = file mock_post_request_body = { 'statement_period_adjustment_file_id': 1, 'abacus_event_id': 1, } res = fixture_client.post( '/worksheet-adjustments/import/', json=mock_post_request_body ) worksheet_adjustments = WorksheetAdjustment.query.all() worksheet_adjustment_details = WorksheetAdjustmentDetail.query.all() assert res.status_code == 201 assert res.json['message'] == 'Records inserted succesfully.' assert len(worksheet_adjustments) == 3 assert worksheet_adjustments[0].adjustment_amount == Decimal('-90.10') assert worksheet_adjustments[1].adjustment_amount == Decimal('7819.12') assert worksheet_adjustments[2].adjustment_amount == Decimal('761.39') assert worksheet_adjustments[0].reference_adjustment_type_id == 1 assert worksheet_adjustments[1].reference_adjustment_type_id == 2 assert worksheet_adjustments[2].reference_adjustment_type_id == 2 assert len(worksheet_adjustment_details) == 0 @patch('abacus_worksheet.logic.worksheet_adjustment.download_file') def test_create_worksheet_adjustments_and_details_invalid_file( mock_download_s3_file, fixture_client ): """Test to create worksheet adjustments and worksheet adjustment details. Throws an error when an adjustment file is empty. """ file = os.path.abspath('tests/utils/test_adjustment_files/invalid_adjustments.xlsx') mock_download_s3_file.return_value = file mock_post_request_body = { 'statement_period_adjustment_file_id': 1, 'abacus_event_id': 1, } res = fixture_client.post( '/worksheet-adjustments/import/', json=mock_post_request_body ) assert res.status_code == 404 assert res.json['message'] == 'No adjustments records found in excel sheet' def test_create_worksheet_adjustments_and_details_invalid_fil_path(fixture_client): """Test to create worksheet adjustments and worksheet adjustment details. Throws an error when file path is NULLE. """ mock_post_request_body = { 'statement_period_adjustment_file_id': 2, 'abacus_event_id': 1, } res = fixture_client.post( '/worksheet-adjustments/import/', json=mock_post_request_body ) assert res.status_code == 404 assert res.json['message'] == 'Invalid adjustment file path' def test_soft_delete_worksheet_adjustments_and_details_by_file_id(fixture_client): """Test to soft delete worksheet adjustments and details by file id.""" statement_period_adjustment_file_id = 3 [ WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=statement_period_adjustment_file_id, ) for _ in range(0, 4) ] mock_worksheet_adjustment = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=statement_period_adjustment_file_id ) [ WorksheetAdjustmentDetailFactory.create( worksheet_adjustment=mock_worksheet_adjustment, statement_period_adjustment_file_id=statement_period_adjustment_file_id, ) for _ in range(0, 2) ] res = fixture_client.delete( f'/worksheet-adjustments/statement-period-adjustment-file/{statement_period_adjustment_file_id}' ) worksheet_adjustments = WorksheetAdjustment.query.all() worksheet_adjustment_details = WorksheetAdjustmentDetail.query.all() assert res.status_code == 204 assert ( all( [ adjustment.deleted_by is not None and adjustment.deleted_at is not None for adjustment in worksheet_adjustments ] ) is True ) assert ( all( [ adjustment_detail.deleted_by is not None and adjustment_detail.deleted_at is not None for adjustment_detail in worksheet_adjustment_details ] ) is True ) def test_soft_delete_worksheet_adjustments_and_details_error(fixture_client): """Test to soft delete worksheet adjustments and details by file id. When statement_period_adjustment_file is not deleted. """ statement_period_adjustment_file_id = 1 res = fixture_client.delete( f'/worksheet-adjustments/statement-period-adjustment-file/{statement_period_adjustment_file_id}' ) assert res.status_code == 400 assert res.json['message'] == ERROR_DELETE_WORKSHEET_ADJUSTMENTS.format( statement_period_adjustment_file_id ) def test_soft_delete_worksheet_adjustments_file_not_exist(fixture_client): """Test to soft delete worksheet adjustments and details by file id. If there is no record for statement_period_adjustment_file. """ statement_period_adjustment_file_id = 120 res = fixture_client.delete( f'/worksheet-adjustments/statement-period-adjustment-file/{statement_period_adjustment_file_id}' ) assert res.status_code == 404 assert res.json['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) def test_get_worksheet_adjustments_and_details(fixture_client): """Test to get worksheet adjustments and details by file id.""" statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create() worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( adjustment_amount=Decimal('180.00'), adjustment_currency_code='EUR' ) workssheet_adjustment_3 = WorksheetAdjustmentFactory.create( adjustment_amount=Decimal('150.00'), adjustment_currency_code='EUR', deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('50.00'), currency_code='EUR', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', worksheet_adjustment=workssheet_adjustment_3, ) res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/adjustments-and-details?limit=2&offset=0' ) assert res.status_code == 200 assert res.json['total_count'] == 3 assert res.json['items'] == [ { 'worksheet_adjustment_id': 1, 'account_id': 1, 'contract_id': 1, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'apply_to_flowthrough_payment': None, 'adjustment_amount': '125.42', 'adjustment_currency_code': 'USD', 'note': 'informative message', 'internal_note': None, 'distribution_type': None, 'upc': None, 'statement_period_adjustment_file_id': 1, 'worksheet_adjustment_detail_id': None, 'deleted_at': None, 'deleted_by': None, }, { 'worksheet_adjustment_id': 2, 'account_id': 1, 'contract_id': 1, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'apply_to_flowthrough_payment': None, 'adjustment_amount': '50.00', 'adjustment_currency_code': 'EUR', 'note': 'informative message', 'internal_note': 'internal informative message', 'distribution_type': 'digital', 'upc': '555444333222111', 'statement_period_adjustment_file_id': 1, 'worksheet_adjustment_detail_id': 1, 'deleted_at': None, 'deleted_by': None, }, ] def test_get_worksheet_adjustments_and_details_file_not_exists(fixture_client): """Test to get worksheet adjustments and details by file id. If there is no record for statement_period_adjustment_file. """ statement_period_adjustment_file_id = 120 res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/adjustments-and-details?limit=2&offset=0' ) assert res.status_code == 404 assert res.json['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) def test_get_worksheet_adjustments_aggregate_file_not_exists(fixture_client): """The aggregate endpoint 404s for a file id that does not exist.""" statement_period_adjustment_file_id = 120 res = fixture_client.get( '/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/worksheet-adjustments/deleted/aggregate' ) assert res.status_code == 404 assert res.json['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) def test_get_worksheet_adjustments_aggregate_soft_deleted_file(fixture_client): """The aggregate endpoint 404s for a soft-deleted file.""" soft_deleted_file_id = 3 # seeded with both deleted_by and deleted_at set res = fixture_client.get( '/statement-period-adjustment-file/' + f'{soft_deleted_file_id}/worksheet-adjustments/deleted/aggregate' ) assert res.status_code == 404 assert res.json['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=soft_deleted_file_id, ) def test_adjustments_and_details_is_deleted_all(fixture_client): """is_deleted=all returns active + deleted with deleted markers populated.""" file_id = 1 WorksheetAdjustmentFactory.create(statement_period_adjustment_file_id=file_id) WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, deleted_by='someone', deleted_at='2024-01-01 00:00:00', ) res = fixture_client.get( f'/worksheet-adjustment/statement-period-adjustment-file/{file_id}' '/adjustments-and-details?is_deleted=all&limit=10&offset=0' ) assert res.status_code == 200 assert res.json['total_count'] == 2 deleted = [r for r in res.json['items'] if r['deleted_at'] is not None] assert len(deleted) == 1 and deleted[0]['deleted_by'] == 'someone' def test_adjustments_and_details_is_deleted_true(fixture_client): """is_deleted=true returns only the deleted row(s).""" file_id = 1 WorksheetAdjustmentFactory.create(statement_period_adjustment_file_id=file_id) deleted_entry = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, deleted_by='someone', deleted_at='2024-01-01 00:00:00', ) res = fixture_client.get( f'/worksheet-adjustment/statement-period-adjustment-file/{file_id}' '/adjustments-and-details?is_deleted=true&limit=10&offset=0' ) assert res.status_code == 200 assert res.json['total_count'] == 1 [row] = res.json['items'] assert row['worksheet_adjustment_id'] == deleted_entry.worksheet_adjustment_id assert row['deleted_at'] is not None assert row['deleted_by'] == 'someone' def test_adjustments_and_details_defaults_to_active(fixture_client): """No is_deleted arg -> active only (unchanged behavior).""" file_id = 1 WorksheetAdjustmentFactory.create(statement_period_adjustment_file_id=file_id) WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, deleted_by='x', deleted_at='2024-01-01 00:00:00', ) res = fixture_client.get( f'/worksheet-adjustment/statement-period-adjustment-file/{file_id}' '/adjustments-and-details?limit=10&offset=0' ) assert res.json['total_count'] == 1 def test_adjustments_and_details_rejects_bad_is_deleted(fixture_client): """An invalid is_deleted value is a 400 (BooleanFilter parse).""" res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/1' '/adjustments-and-details?is_deleted=bogus' ) assert res.status_code == 400 def test_get_pending_worksheet_adjustments_exists(fixture_client): """Test get pending worksheet adjustments. Pending worksheet exists for params """ WorksheetAdjustmentFactory.create() res = fixture_client.get( '/worksheet-adjustments/pending' '?limit=10&offset=0&statement_period_id=2&reference_payment_entity_id=1' ) assert res.status_code == 200 assert res.json['total_count'] == 1 assert res.json['items'] == [ { 'account_id': 1, 'activity_statement_period_id': 1, 'adjustment_amount': '125.42', 'adjustment_currency_code': 'USD', 'apply_to_statement_period_id': 2, 'apply_to_flowthrough_payment': None, 'contract_id': 1, 'internal_note': None, 'note': 'informative message', 'reference_adjustment_type_id': 1, 'statement_period_adjustment_file_id': 1, 'worksheet_adjustment_id': 1, } ] def test_get_pending_worksheet_adjustments_not_exists(fixture_client): """Test get pending worksheet adjustments. Pending worksheet does not exists for params """ WorksheetAdjustmentFactory.create() res = fixture_client.get( '/worksheet-adjustments/pending' '?limit=10&offset=0&statement_period_id=3&reference_payment_entity_id=2' ) assert res.status_code == 200 assert res.json['total_count'] == 0 assert res.json['items'] == [] def test_get_adjustments_by_period_and_type_functional(fixture_client): """Functional test for GET /worksheet-adjustments/period//type/. Ensures response returns items, total_count, and currency-agnostic total amount for matching records, excluding deleted ones. """ period_id = 1 type_id = 1 adj1 = WorksheetAdjustmentFactory.create( activity_statement_period_id=period_id, reference_adjustment_type_id=type_id, adjustment_amount=Decimal('100.00'), adjustment_currency_code='USD', ) adj2 = WorksheetAdjustmentFactory.create( activity_statement_period_id=period_id, reference_adjustment_type_id=type_id, adjustment_amount=Decimal('50.00'), adjustment_currency_code='EUR', ) # Deleted record should be excluded WorksheetAdjustmentFactory.create( activity_statement_period_id=period_id, reference_adjustment_type_id=type_id, adjustment_amount=Decimal('25.00'), adjustment_currency_code='USD', deleted_by='Test User', deleted_at='2024-10-01', ) # Seed applied ledger rows to satisfy inner join filter using raw SQL sql = f""" INSERT INTO ledger_adjustment_applied ( `ledger_adjustment_applied_id`, `abacus_event_id`, `account_id`, `contract_id`, `statement_period_id`, `ledger_adjustment_id`, `worksheet_adjustment_id`, `adjustment_amount`, `adjustment_currency_code`, `adjustment_amount_payee_currency`, `adjustment_payee_currency_code`, `created_by`, `created_at`, `last_modified_by`, `last_modified` ) VALUES ( 1, 1, {adj1.account_id}, {adj1.contract_id if adj1.contract_id is not None else 'NULL'}, {adj1.apply_to_statement_period_id}, NULL, {adj1.worksheet_adjustment_id}, '{adj1.adjustment_amount}', '{adj1.adjustment_currency_code}', '{adj1.adjustment_amount}', '{adj1.adjustment_currency_code}', 'test', NOW(), 'test', NOW() ), ( 2, 1, {adj2.account_id}, {adj2.contract_id if adj2.contract_id is not None else 'NULL'}, {adj2.apply_to_statement_period_id}, NULL, {adj2.worksheet_adjustment_id}, '{adj2.adjustment_amount}', '{adj2.adjustment_currency_code}', '{adj2.adjustment_amount}', '{adj2.adjustment_currency_code}', 'test', NOW(), 'test', NOW() ); """ db.session.execute(sql) db.session.commit() res = fixture_client.get( f'/worksheet-adjustments/period/{period_id}/type/{type_id}' ) assert res.status_code == 200 # Items should be exactly adj1 and adj2 returned_ids = {item['worksheet_adjustment_id'] for item in res.json['items']} assert returned_ids == { adj1.worksheet_adjustment_id, adj2.worksheet_adjustment_id, } assert res.json['total_count'] == 2 assert res.json['currency_agnostic_total_amount'] == float(Decimal('150.00')) def test_get_worksheet_adjustments_and_details_filter_by_contract_ids(fixture_client): """Test to get worksheet adjustments and details for specified file id. And filter the results by contract ids. """ statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create() worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( contract_id=2, adjustment_amount=Decimal('100.00'), adjustment_currency_code='EUR', ) workssheet_adjustment_3 = WorksheetAdjustmentFactory.create( contract_id=3, adjustment_amount=Decimal('80.00'), adjustment_currency_code='EUR', deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentDetailFactory.create( contract_id=2, amount=Decimal('50.00'), currency_code='EUR', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( contract_id=2, amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( contract_id=3, amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', worksheet_adjustment=workssheet_adjustment_3, deleted_by='Test', deleted_at='2024-02-01', ) res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/adjustments-and-details?limit=2&offset=0&contract_ids=2,3' ) assert res.status_code == 200 assert res.json['total_count'] == 2 assert res.json['items'] == [ { 'worksheet_adjustment_id': 2, 'account_id': 1, 'contract_id': 2, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'apply_to_flowthrough_payment': None, 'adjustment_amount': '50.00', 'adjustment_currency_code': 'EUR', 'note': 'informative message', 'internal_note': 'internal informative message', 'distribution_type': 'digital', 'upc': '555444333222111', 'statement_period_adjustment_file_id': 1, 'worksheet_adjustment_detail_id': 1, 'deleted_at': None, 'deleted_by': None, }, { 'worksheet_adjustment_id': 2, 'account_id': 1, 'contract_id': 2, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'apply_to_flowthrough_payment': None, 'adjustment_amount': '50.00', 'adjustment_currency_code': 'EUR', 'note': 'informative message', 'internal_note': 'internal informative message', 'distribution_type': 'digital', 'upc': '1234567891', 'statement_period_adjustment_file_id': 1, 'worksheet_adjustment_detail_id': 2, 'deleted_at': None, 'deleted_by': None, }, ] def test_get_worksheet_adjustments_and_details_filter_by_account_ids(fixture_client): """Test to get worksheet adjustments and details for specified file id. And filter the results by account ids. """ statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create() worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( account_id=3, contract_id=3, adjustment_amount=Decimal('100.00'), adjustment_currency_code='EUR', ) workssheet_adjustment_3 = WorksheetAdjustmentFactory.create( account_id=4, contract_id=4, adjustment_amount=Decimal('80.00'), adjustment_currency_code='EUR', deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=4, contract_id=4, amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', worksheet_adjustment=workssheet_adjustment_3, deleted_by='Test', deleted_at='2024-02-01', ) res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/adjustments-and-details?limit=2&offset=0&account_ids=1,3,4' ) assert res.status_code == 200 assert res.json['total_count'] == 3 assert res.json['items'] == [ { 'worksheet_adjustment_id': 1, 'statement_period_adjustment_file_id': 1, 'account_id': 1, 'contract_id': 1, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'apply_to_flowthrough_payment': None, 'reference_adjustment_type_id': 1, 'adjustment_amount': '125.42', 'adjustment_currency_code': 'USD', 'note': 'informative message', 'internal_note': None, 'distribution_type': None, 'upc': None, 'worksheet_adjustment_detail_id': None, 'deleted_at': None, 'deleted_by': None, }, { 'worksheet_adjustment_id': 2, 'statement_period_adjustment_file_id': 1, 'account_id': 3, 'contract_id': 3, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'apply_to_flowthrough_payment': None, 'reference_adjustment_type_id': 1, 'adjustment_amount': '50.00', 'adjustment_currency_code': 'EUR', 'note': 'informative message', 'internal_note': 'internal informative message', 'distribution_type': 'digital', 'upc': '555444333222111', 'worksheet_adjustment_detail_id': 1, 'deleted_at': None, 'deleted_by': None, }, ] def test_get_worksheet_adjustments_and_details_filter_by_flowthrough_payments( fixture_client, ): """Test to get worksheet adjustments and details for specified file id. And filter the results by apply_to_flowthrough_payment column. """ statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create(apply_to_flowthrough_payment=0) WorksheetAdjustmentFactory.create(apply_to_flowthrough_payment=None) worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( account_id=3, contract_id=3, adjustment_amount=Decimal('100.00'), adjustment_currency_code='EUR', ) workssheet_adjustment_3 = WorksheetAdjustmentFactory.create( account_id=4, contract_id=4, adjustment_amount=Decimal('80.00'), adjustment_currency_code='EUR', ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=4, contract_id=4, amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', apply_to_flowthrough_payment=1, worksheet_adjustment=workssheet_adjustment_3, deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentDetailFactory.create( account_id=4, contract_id=4, amount=Decimal('80.00'), currency_code='EUR', upc='123456789290', apply_to_flowthrough_payment=1, worksheet_adjustment=workssheet_adjustment_3, ) res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/adjustments-and-details?limit=2&offset=0&apply_to_flowthrough_payment=0,1' ) assert res.status_code == 200 assert res.json['total_count'] == 2 assert res.json['items'] == [ { 'worksheet_adjustment_id': 1, 'statement_period_adjustment_file_id': 1, 'account_id': 1, 'contract_id': 1, 'apply_to_flowthrough_payment': False, '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, 'distribution_type': None, 'upc': None, 'worksheet_adjustment_detail_id': None, 'deleted_at': None, 'deleted_by': None, }, { 'worksheet_adjustment_id': 4, 'statement_period_adjustment_file_id': 1, 'account_id': 4, 'contract_id': 4, 'apply_to_flowthrough_payment': True, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'adjustment_amount': '80.00', 'adjustment_currency_code': 'EUR', 'note': 'informative message', 'internal_note': 'internal informative message', 'distribution_type': 'digital', 'upc': '123456789290', 'worksheet_adjustment_detail_id': 4, 'deleted_at': None, 'deleted_by': None, }, ] def test_get_worksheet_adjustments_details_filter_by_all_filter_params(fixture_client): """Test to get worksheet adjustments and details for specified file id. And filter the results by account, contract ids and apply_to_flowthrough_payment. """ statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create() worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( account_id=3, contract_id=3, adjustment_amount=Decimal('100.00'), adjustment_currency_code='EUR', ) workssheet_adjustment_3 = WorksheetAdjustmentFactory.create( account_id=4, contract_id=4, adjustment_amount=Decimal('80.00'), adjustment_currency_code='EUR', deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentFactory.create( account_id=4, contract_id=5, adjustment_amount=Decimal('180.00'), adjustment_currency_code='AUD', ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', apply_to_flowthrough_payment=1, worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', apply_to_flowthrough_payment=1, worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=4, contract_id=4, amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', apply_to_flowthrough_payment=0, worksheet_adjustment=workssheet_adjustment_3, deleted_by='Test', deleted_at='2024-02-01', ) res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/adjustments-and-details?limit=10&offset=0&account_ids=1,3,4&contract_ids=1,3,4,5' + '&apply_to_flowthrough_payment=0,1,null' ) assert res.status_code == 200 assert res.json['total_count'] == 4 assert res.json['items'] == [ { 'worksheet_adjustment_id': 1, 'statement_period_adjustment_file_id': 1, 'account_id': 1, 'contract_id': 1, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'apply_to_flowthrough_payment': None, 'adjustment_amount': '125.42', 'adjustment_currency_code': 'USD', 'note': 'informative message', 'internal_note': None, 'distribution_type': None, 'upc': None, 'worksheet_adjustment_detail_id': None, 'deleted_at': None, 'deleted_by': None, }, { 'worksheet_adjustment_id': 2, 'statement_period_adjustment_file_id': 1, 'account_id': 3, 'contract_id': 3, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'reference_adjustment_type_id': 1, 'apply_to_flowthrough_payment': 1, 'adjustment_amount': '50.00', 'adjustment_currency_code': 'EUR', 'note': 'informative message', 'internal_note': 'internal informative message', 'distribution_type': 'digital', 'upc': '555444333222111', 'worksheet_adjustment_detail_id': 1, 'deleted_at': None, 'deleted_by': None, }, { 'worksheet_adjustment_id': 2, 'statement_period_adjustment_file_id': 1, 'account_id': 3, 'contract_id': 3, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'apply_to_flowthrough_payment': 1, 'reference_adjustment_type_id': 1, 'adjustment_amount': '50.00', 'adjustment_currency_code': 'EUR', 'note': 'informative message', 'internal_note': 'internal informative message', 'distribution_type': 'digital', 'upc': '1234567891', 'worksheet_adjustment_detail_id': 2, 'deleted_at': None, 'deleted_by': None, }, { 'worksheet_adjustment_id': 4, 'statement_period_adjustment_file_id': 1, 'account_id': 4, 'contract_id': 5, 'activity_statement_period_id': 1, 'apply_to_statement_period_id': 2, 'apply_to_flowthrough_payment': None, 'reference_adjustment_type_id': 1, 'adjustment_amount': '180.00', 'adjustment_currency_code': 'AUD', 'note': 'informative message', 'internal_note': None, 'distribution_type': None, 'upc': None, 'worksheet_adjustment_detail_id': None, 'deleted_at': None, 'deleted_by': None, }, ] def test_get_worksheet_adjustments_contracts_file_not_exists(fixture_client): """Test to get worksheet adjustments contracts. If there is no record for statement_period_adjustment_file. """ statement_period_adjustment_file_id = 120 res = fixture_client.get( '/worksheet-adjustments/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/contracts?limit=2&offset=0' ) assert res.status_code == 404 assert res.json['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) def test_get_worksheet_adjustments_contracts_by_contract_name(fixture_client): """Test to get worksheet adjustments contracts by search term contract_name.""" statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create() worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( contract_id=4, adjustment_amount=Decimal('100.00'), adjustment_currency_code='EUR', ) worksheet_adjustment_3 = WorksheetAdjustmentFactory.create( contract_id=2, adjustment_amount=Decimal('80.00'), adjustment_currency_code='EUR', deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('50.00'), currency_code='EUR', contract_id=4, worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', contract_id=4, worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', contract_id=2, worksheet_adjustment=worksheet_adjustment_3, deleted_by='Test', deleted_at='2024-02-01', ) res = fixture_client.get( '/worksheet-adjustments/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/contracts?contract_search_term=Hayward%20Young' + '&limit=2&offset=0' ) assert res.status_code == 200 assert res.json['total_count'] == 1 assert res.json['items'] == [{'contract_id': 4}] def test_get_worksheet_adjustments_contracts_by_any_number(fixture_client): """Test to get worksheet adjustments contracts by any number.""" statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create(contract_id=3) worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( contract_id=4, adjustment_amount=Decimal('100.00'), adjustment_currency_code='EUR', ) worksheet_adjustment_3 = WorksheetAdjustmentFactory.create( contract_id=2, adjustment_amount=Decimal('80.00'), adjustment_currency_code='EUR', deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('50.00'), currency_code='EUR', contract_id=4, worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', contract_id=4, worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', contract_id=2, worksheet_adjustment=worksheet_adjustment_3, deleted_by='Test', deleted_at='2024-02-01', ) res = fixture_client.get( '/worksheet-adjustments/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/contracts?contract_search_term=3' + '&limit=2&offset=0' ) assert res.status_code == 200 assert res.json['total_count'] == 1 assert res.json['items'] == [{'contract_id': 3}] def test_get_worksheet_adjustments_and_details_file_invalid_contract_id(fixture_client): """Test to get worksheet adjustments and details by file id. Throws an error if invalid contract ids are passed. """ statement_period_adjustment_file_id = 1 res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/adjustments-and-details?contract_ids=2,test&limit=2&offset=0' ) assert res.status_code == 400 assert res.json['message'] == str({'_schema': [ERROR_INVALID_CONTRACT_IDS]}) def test_get_worksheet_adjustments_contracts_invalid_params(fixture_client): """Test to get worksheet adjustments contracts. Throws an error if invalid params are passed. """ statement_period_adjustment_file_id = 1 res = fixture_client.get( '/worksheet-adjustments/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/contracts?contract_search_term=3' + '&limit=test&offset=%283' ) assert res.status_code == 400 assert res.json['message'] == str( { 'limit': ['Must be an integer greater or equal to 0.'], 'offset': ['Must be an integer greater or equal to 0.'], } ) def test_get_worksheet_adjustments_accounts_file_not_exists(fixture_client): """Test to get worksheet adjustments accounts. If there is no record for statement_period_adjustment_file. """ statement_period_adjustment_file_id = 120 res = fixture_client.get( '/worksheet-adjustments/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/accounts?limit=2&offset=0' ) assert res.status_code == 404 assert res.json['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriodAdjustmentFile', object_id=statement_period_adjustment_file_id, ) def test_get_worksheet_adjustments_and_details_file_invalid_account_id(fixture_client): """Test to get worksheet adjustments and details by file id. Throws an error if invalid account ids are passed. """ statement_period_adjustment_file_id = 1 res = fixture_client.get( '/worksheet-adjustment/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/adjustments-and-details?account_ids=2,test&limit=2&offset=0' ) assert res.status_code == 400 assert res.json['message'] == str({'_schema': [ERROR_INVALID_ACCOUNT_IDS]}) def test_get_worksheet_adjustments_accounts_invalid_params(fixture_client): """Test to get worksheet adjustments accounts. Throws an error if invalid params are passed. """ statement_period_adjustment_file_id = 1 res = fixture_client.get( '/worksheet-adjustments/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/accounts?account_search_term=3' + '&limit=test&offset=%283' ) assert res.status_code == 400 assert res.json['message'] == str( { 'limit': ['Must be an integer greater or equal to 0.'], 'offset': ['Must be an integer greater or equal to 0.'], } ) def test_get_worksheet_adjustments_accounts_by_account_name(fixture_client): """Test to get worksheet adjustments accounts by search term account_name.""" statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create() worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( account_id=3, contract_id=3, adjustment_amount=Decimal('100.00'), adjustment_currency_code='EUR', ) workssheet_adjustment_3 = WorksheetAdjustmentFactory.create( account_id=4, contract_id=4, adjustment_amount=Decimal('80.00'), adjustment_currency_code='EUR', deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentFactory.create( account_id=4, contract_id=5, adjustment_amount=Decimal('180.00'), adjustment_currency_code='AUD', ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=4, contract_id=4, amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', worksheet_adjustment=workssheet_adjustment_3, deleted_by='Test', deleted_at='2024-02-01', ) res = fixture_client.get( '/worksheet-adjustments/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/accounts?account_search_term=green' ) assert res.status_code == 200 assert res.json['total_count'] == 1 assert res.json['items'] == [{'account_id': 3}] def test_get_worksheet_adjustments_accounts_by_any_number(fixture_client): """Test to get worksheet adjustments accounts by any number.""" statement_period_adjustment_file_id = 1 WorksheetAdjustmentFactory.create() worksheet_adjustment_2 = WorksheetAdjustmentFactory.create( account_id=3, contract_id=3, adjustment_amount=Decimal('100.00'), adjustment_currency_code='EUR', ) workssheet_adjustment_3 = WorksheetAdjustmentFactory.create( account_id=4, contract_id=4, adjustment_amount=Decimal('80.00'), adjustment_currency_code='EUR', deleted_by='Test', deleted_at='2024-02-01', ) WorksheetAdjustmentFactory.create( account_id=4, contract_id=5, adjustment_amount=Decimal('180.00'), adjustment_currency_code='AUD', ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=3, contract_id=3, amount=Decimal('50.00'), currency_code='EUR', upc='1234567891', worksheet_adjustment=worksheet_adjustment_2, ) WorksheetAdjustmentDetailFactory.create( account_id=4, contract_id=4, amount=Decimal('80.00'), currency_code='EUR', upc='1234567892', worksheet_adjustment=workssheet_adjustment_3, deleted_by='Test', deleted_at='2024-02-01', ) res = fixture_client.get( '/worksheet-adjustments/statement-period-adjustment-file/' + f'{statement_period_adjustment_file_id}/accounts?account_search_term=1' ) assert res.status_code == 200 assert res.json['total_count'] == 1 assert res.json['items'] == [{'account_id': 1}] def test_get_worksheet_adjustments_excludes_soft_deleted_details(fixture_client): """A soft-deleted detail is not serialized under its parent's details.""" statement_period_adjustment_file_id = 1 parent = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=statement_period_adjustment_file_id, ) WorksheetAdjustmentDetailFactory.create(worksheet_adjustment=parent) WorksheetAdjustmentDetailFactory.create( worksheet_adjustment=parent, deleted_by='Test', deleted_at='2019-03-01 00:00:00', ) res = fixture_client.get( f'/worksheet-adjustments/statement-period-adjustment-file/' f'{statement_period_adjustment_file_id}/?limit=2&offset=0' ) assert res.status_code == 200 assert res.json['total_count'] == 1 assert len(res.json['items'][0]['details']) == 1 def _post_delete(client, ids): return client.post( '/worksheet-adjustments/delete', json={'worksheet_adjustment_ids': ids}, ) def test_delete_worksheet_adjustments_happy_path(fixture_client): """Delete soft-deletes the entries and their details, returns 204.""" file_id = 1 parent = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id ) WorksheetAdjustmentDetailFactory.create( worksheet_adjustment=parent, statement_period_adjustment_file_id=file_id ) res = _post_delete(fixture_client, [parent.worksheet_adjustment_id]) assert res.status_code == 204 refreshed = WorksheetAdjustment.query.get(parent.worksheet_adjustment_id) assert refreshed.deleted_at is not None and refreshed.deleted_by is not None details = WorksheetAdjustmentDetail.query.filter_by( worksheet_adjustment_id=parent.worksheet_adjustment_id ).all() assert all(d.deleted_at is not None for d in details) def test_delete_worksheet_adjustments_rejects_applied_entry( fixture_client, mock_applied_worksheet_adjustment ): """An applied entry cannot be deleted; nothing is mutated (all-or-nothing).""" applied = mock_applied_worksheet_adjustment live = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=applied.statement_period_adjustment_file_id ) res = _post_delete( fixture_client, [applied.worksheet_adjustment_id, live.worksheet_adjustment_id], ) assert res.status_code == 409 assert ( WorksheetAdjustment.query.get(live.worksheet_adjustment_id).deleted_at is None ) assert ( WorksheetAdjustment.query.get(applied.worksheet_adjustment_id).deleted_at is None ) def test_delete_worksheet_adjustments_rejects_missing_id(fixture_client): """A non-existent id yields 404 and no mutation.""" parent = WorksheetAdjustmentFactory.create() res = _post_delete(fixture_client, [parent.worksheet_adjustment_id, 9999999]) assert res.status_code == 404 assert ( WorksheetAdjustment.query.get(parent.worksheet_adjustment_id).deleted_at is None ) def test_delete_worksheet_adjustments_rejects_applied_batch( fixture_client, insert_apply_file_state ): """When the batch apply_file state is complete, delete is rejected.""" # Reuse the pre-seeded live statement_period_adjustment_file id (1) so the # WorksheetAdjustment FK to statement_period_adjustment_file is satisfied; # insert_apply_file_state's own default id (987) has no such file row. file_id = insert_apply_file_state(file_id=1, status='complete') parent = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id ) res = _post_delete(fixture_client, [parent.worksheet_adjustment_id]) assert res.status_code == 409 assert ( WorksheetAdjustment.query.get(parent.worksheet_adjustment_id).deleted_at is None ) def test_delete_requires_non_empty_ids(fixture_client): """An empty id list is a 400 validation error.""" res = _post_delete(fixture_client, []) assert res.status_code == 400 def test_delete_rejects_multiple_files(fixture_client): """Ids spanning more than one file are rejected (400), no mutation.""" wa_file1 = WorksheetAdjustmentFactory.create(statement_period_adjustment_file_id=1) wa_file2 = WorksheetAdjustmentFactory.create(statement_period_adjustment_file_id=2) res = _post_delete( fixture_client, [wa_file1.worksheet_adjustment_id, wa_file2.worksheet_adjustment_id], ) assert res.status_code == 400 assert ( WorksheetAdjustment.query.get(wa_file1.worksheet_adjustment_id).deleted_at is None ) assert ( WorksheetAdjustment.query.get(wa_file2.worksheet_adjustment_id).deleted_at is None ) def test_delete_worksheet_adjustments_rejects_already_deleted(fixture_client): """Deleting an already soft-deleted entry is rejected (400), no re-delete.""" file_id = 1 parent = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, deleted_by='Test', deleted_at='2019-03-01 00:00:00', ) res = _post_delete(fixture_client, [parent.worksheet_adjustment_id]) assert res.status_code == 400 assert ( WorksheetAdjustment.query.get(parent.worksheet_adjustment_id).deleted_at is not None ) def _post_restore(client, ids): return client.post( '/worksheet-adjustments/restore', json={'worksheet_adjustment_ids': ids}, ) def test_restore_worksheet_adjustments_happy_path(fixture_client): """Restore clears deleted markers on parent and details in lockstep.""" file_id = 1 parent = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, deleted_by='Test', deleted_at='2019-03-01 00:00:00', ) WorksheetAdjustmentDetailFactory.create( worksheet_adjustment=parent, statement_period_adjustment_file_id=file_id, deleted_by='Test', deleted_at='2019-03-01 00:00:00', ) res = _post_restore(fixture_client, [parent.worksheet_adjustment_id]) assert res.status_code == 204 refreshed = WorksheetAdjustment.query.get(parent.worksheet_adjustment_id) assert refreshed.deleted_at is None and refreshed.deleted_by is None details = WorksheetAdjustmentDetail.query.filter_by( worksheet_adjustment_id=parent.worksheet_adjustment_id ).all() assert all(d.deleted_at is None for d in details) def test_restore_rejects_not_deleted_entry(fixture_client): """Restoring a live (not deleted) entry is rejected (400); no mutation.""" file_id = 1 parent = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id ) res = _post_restore(fixture_client, [parent.worksheet_adjustment_id]) assert res.status_code == 400 assert ( WorksheetAdjustment.query.get(parent.worksheet_adjustment_id).deleted_at is None ) def test_restore_rejects_applied_batch(fixture_client, insert_apply_file_state): """Restore is rejected when the batch has been applied (409).""" # Reuse the already-seeded live file id 1 (there is an FK from # worksheet_adjustment.statement_period_adjustment_file_id to # statement_period_adjustment_file; the fixture default 987 has no row). file_id = insert_apply_file_state(file_id=1, status='complete') parent = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, deleted_by='Test', deleted_at='2019-03-01 00:00:00', ) res = _post_restore(fixture_client, [parent.worksheet_adjustment_id]) assert res.status_code == 409 def _insert_ledger_applied_for(worksheet_adjustment_id): """Insert a ledger_adjustment_applied row for the given worksheet_adjustment_id. Mirrors the column list used by the mock_applied_worksheet_adjustment fixture. """ SQL_QUERY = f""" INSERT INTO ledger_adjustment_applied ( `ledger_adjustment_applied_id`, `abacus_event_id`, `account_id`, `contract_id`, `statement_period_id`, `ledger_adjustment_id`, `worksheet_adjustment_id`, `adjustment_amount`, `adjustment_currency_code`, `adjustment_amount_payee_currency`, `adjustment_payee_currency_code`, `created_by`, `created_at`, `last_modified_by`, `last_modified` ) VALUES ( 1, 1, 1, 1, 2, NULL, {worksheet_adjustment_id}, '-100.98', 'USD', '-100.98', 'USD', 'Test account', NOW(), 'Test account', NOW() ); """ db.engine.execute(SQL_QUERY) # Commit so the raw insert is visible to the fresh transaction the test # client's request opens. Without this, the still-open REPEATABLE READ # snapshot from the earlier factory-created row (refreshed on attribute # access after its own commit) predates this insert and would hide it. db.session.commit() def test_whole_batch_delete_rejects_when_entry_applied(fixture_client): """Whole-batch delete refuses to soft-delete a file containing an applied entry.""" soft_deleted_file_id = 3 # seeded soft-deleted by fresh_db applied = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=soft_deleted_file_id ) # Insert a ledger_adjustment_applied row referencing this entry (mirror the # columns used by the mock_applied_worksheet_adjustment fixture). _insert_ledger_applied_for(applied.worksheet_adjustment_id) res = fixture_client.delete( f'/worksheet-adjustments/statement-period-adjustment-file/{soft_deleted_file_id}/' ) assert res.status_code == 409 refreshed = WorksheetAdjustment.query.get(applied.worksheet_adjustment_id) assert refreshed.deleted_at is None def test_delete_rejects_too_many_ids(fixture_client): """A bulk request above the id cap is rejected (400) by schema validation.""" ids = list(range(1, constants.MAX_WORKSHEET_ADJUSTMENT_BULK_IDS + 2)) res = _post_delete(fixture_client, ids) assert res.status_code == 400 def test_whole_batch_delete_preserves_existing_soft_delete_metadata(fixture_client): """A whole-file delete preserves already-set deletion metadata. An entry (or detail) already individually soft-deleted keeps its original deleted_at/deleted_by when the whole file is later deleted. """ file_id = 3 # seeded soft-deleted, so the whole-batch delete proceeds original_at = '2019-01-01 00:00:00' entry = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, deleted_by='original_deleter', deleted_at=original_at, ) detail = WorksheetAdjustmentDetailFactory.create( worksheet_adjustment=entry, statement_period_adjustment_file_id=file_id, deleted_by='original_deleter', deleted_at=original_at, ) res = fixture_client.delete( f'/worksheet-adjustments/statement-period-adjustment-file/{file_id}/' ) assert res.status_code == 204 refreshed = WorksheetAdjustment.query.get(entry.worksheet_adjustment_id) assert refreshed.deleted_by == 'original_deleter' assert str(refreshed.deleted_at) == original_at refreshed_detail = WorksheetAdjustmentDetail.query.get( detail.worksheet_adjustment_detail_id ) assert refreshed_detail.deleted_by == 'original_deleter' assert str(refreshed_detail.deleted_at) == original_at def test_deleted_aggregate_endpoint_round_trip(fixture_client): """The aggregate endpoint reacts to delete then restore, returning to zero.""" file_id = 1 a = WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, adjustment_amount=Decimal('10.00') ) WorksheetAdjustmentFactory.create( statement_period_adjustment_file_id=file_id, adjustment_amount=Decimal('5.00') ) def agg(): r = fixture_client.get( f'/statement-period-adjustment-file/{file_id}/worksheet-adjustments/deleted/aggregate' ) assert r.status_code == 200 return r.json start = agg() assert start['deleted_count'] == 0 assert start['currency_agnostic_deleted_amount'] == '0.00' fixture_client.post( '/worksheet-adjustments/delete', json={'worksheet_adjustment_ids': [a.worksheet_adjustment_id]}, ) mid = agg() assert ( mid['deleted_count'] == 1 and mid['currency_agnostic_deleted_amount'] == '10.00' ) fixture_client.post( '/worksheet-adjustments/restore', json={'worksheet_adjustment_ids': [a.worksheet_adjustment_id]}, ) end = agg() assert end['deleted_count'] == 0 assert end['currency_agnostic_deleted_amount'] == '0.00'