"""Unit tests for Ledger Reserve Taken logic.""" from unittest.mock import patch import pytest from ledger.constants.constants import ( DEFAULT_PAGE_OFFSET, LEDGER_ACCOUNT_ENTRIES_SUCCESS_MSG, ) from ledger.constants.error import ERROR_UNKNOWN_CURRENCY from ledger.logic import ledger_reserve_taken as logic from tests.utils.factories import LedgerReserveTakenFactory @patch('ledger.logic.ledger_reserve_taken.LedgerReserveTaken') @patch('ledger.logic.ledger_reserve_taken._validate_entry') @patch('ledger.logic.ledger_reserve_taken._validate_post_request_body') @patch('ledger.logic.ledger_reserve_taken._get_existing_ledger_records') @patch('ledger.logic.ledger_reserve_taken.db') def test_bulk_create_ledger_reserve_taken_success( mock_db, mock_get_existing_records, mock_validate_post_request_body, mock_validate_entry, mock_model, mock_bulk_ledger_reserve_taken_body, ): """Test successfully bulk creating ledger_reserve_taken entries.""" params = mock_bulk_ledger_reserve_taken_body mock_validate_post_request_body.return_value = None mock_get_existing_records.return_value = ['am i real?'] mock_validate_entry.return_value = True mock_model.return_value = '' mock_model.get_by_abacus_events_accounting_runs_contract_reserves.return_value = ( mock_bulk_ledger_reserve_taken_body ) result = logic.bulk_create(params) assert result.status == 201 assert len(result.message) == len(params) assert mock_validate_entry.call_count == len(params) assert mock_model.call_count == len(params) mock_validate_post_request_body.assert_called_once_with(params) mock_get_existing_records.assert_called_once() mock_model.get_by_abacus_events_accounting_runs_contract_reserves.assert_called_once() mock_db.session.bulk_save_objects.assert_called_once() mock_db.session.commit.assert_called_once() mock_db.session.close.assert_called_once() @patch('ledger.logic.ledger_reserve_taken.LedgerReserveTaken') @patch('ledger.logic.ledger_reserve_taken._validate_entry') @patch('ledger.logic.ledger_reserve_taken._validate_post_request_body') @patch('ledger.logic.ledger_reserve_taken._get_existing_ledger_records') def test_bulk_create_ledger_reserve_taken_duplicates_in_payload( mock_get_existing_records, mock_validate_post_request_body, mock_validate_entry, mock_model, mock_bulk_ledger_reserve_taken_body, ): """Test failure bulk creating entries when POST body has duplicates.""" params = [ mock_bulk_ledger_reserve_taken_body[0], mock_bulk_ledger_reserve_taken_body[0], ] mock_validate_post_request_body.side_effect = Exception('This data is a mess') mock_get_existing_records.return_value = [] mock_validate_entry.return_value = True mock_model.build.return_value = True mock_model.commit_changes.return_value = True res = logic.bulk_create(params) assert res.status == 400 assert res.errors mock_validate_post_request_body.assert_called_once_with(params) mock_get_existing_records.assert_not_called() mock_validate_entry.assert_not_called() mock_model.build.assert_not_called() mock_model.commit_changes.assert_not_called() @patch('ledger.logic.ledger_reserve_taken.LedgerReserveTaken') @patch('ledger.logic.ledger_reserve_taken._validate_entry') @patch('ledger.logic.ledger_reserve_taken._validate_post_request_body') @patch('ledger.logic.ledger_reserve_taken._get_existing_ledger_records') def test_bulk_create_ledger_reserve_taken_record_already_exists( mock_get_existing_records, mock_validate_post_request_body, mock_validate_entry, mock_model, mock_bulk_ledger_reserve_taken_body, ): """Test failure bulk creating ledger_reserve_taken entries when record exists.""" params = mock_bulk_ledger_reserve_taken_body mock_validate_post_request_body.return_value = None mock_get_existing_records.return_value = [] mock_validate_entry.side_effect = Exception('there can be only one') mock_model.build.return_value = True mock_model.commit_changes.return_value = True res = logic.bulk_create(params) assert res.status == 400 assert res.errors assert mock_validate_entry.call_count == 1 mock_validate_post_request_body.assert_called_once_with(params) mock_get_existing_records.assert_called_once() mock_validate_entry.assert_called_with([], params[0]) mock_model.build.assert_not_called() mock_model.commit_changes.assert_not_called() @patch('ledger.logic.ledger_reserve_taken.validate_pagination_params') @patch('ledger.logic.ledger_reserve_taken.LedgerReserveTaken') def test_get_ledger_reserve_taken_by_accounting_run_id( mock_model, mock_validation, mock_event_fixtures ): """Test get_ledger_reserve_taken_by_accounting_period_id function.""" accounting_run_id = 1 ledger_reserve_taken = LedgerReserveTakenFactory.create_batch(2) request_params = {'limit': 10, 'offset': 0} mock_model.get_by_accounting_run.return_value = ( ledger_reserve_taken, len(ledger_reserve_taken), ) mock_validation.return_value = request_params res = logic.get_ledger_reserve_taken_by_accounting_run_id( accounting_run_id, request_params ) assert res.status == 200 assert res.message.get('items') assert res.message.get('total_count') == 2 mock_validation.assert_called_once_with(10, 0) mock_model.get_by_accounting_run.assert_called_once_with( accounting_run_id, **request_params ) @patch('ledger.logic.ledger_reserve_taken.build_account_ledger_entries') @patch('ledger.logic.ledger_reserve_taken.LedgerReserveTaken') @patch('ledger.logic.ledger_reserve_taken.db') def test_debit_reserves_from_ledger_account_by_run( mock_db, mock_ledger_reserve_taken_model, mock_build_ledger_entires ): """Test debit_reserves_from_ledger_account_by_run function.""" accounting_run_id = 1 mock_ledger_reserve_taken = [ { 'abacus_event_id': 22, 'gross_returns': '-12.10', 'currency_code': 'CAD', 'account_id': 35970, 'ledger_reserve_taken_id': 1, 'contract_id': 6, 'net_revenue': '123.10', 'gross_sales': '100.20', 'net_revenue_after_reserve': '110.66', 'accounting_run_id': 1, 'statement_period_id': 253, 'reserve_amount': '-12.34', 'contract_reserve_id': 1, }, { 'abacus_event_id': 22, 'gross_returns': '-100.10', 'currency_code': 'AUD', 'account_id': 35971, 'ledger_reserve_taken_id': 2, 'contract_id': 7, 'net_revenue': '90.10', 'gross_sales': '23.90', 'net_revenue_after_reserve': '119.10', 'accounting_run_id': 1, 'statement_period_id': 253, 'reserve_amount': '-111.34', 'contract_reserve_id': 2, }, ] mock_ledger_account_contract = [ { 'abacus_event_id': 22, 'account_id': 35970, 'contract_id': 6, 'currency_code': 'CAD', 'currency_amount': '-12.34', }, { 'abacus_event_id': 22, 'account_id': 35971, 'contract_id': 7, 'currency_code': 'AUD', 'currency_amount': '-111.34', }, ] mock_ledger_reserve_taken_model.get_by_accounting_run.return_value = ( mock_ledger_reserve_taken, len(mock_ledger_reserve_taken), ) mock_build_ledger_entires.return_value = ( mock_ledger_account_contract, mock_ledger_account_contract, ) result = logic.debit_reserves_from_ledger_account_by_run(accounting_run_id) assert result.status == 201 assert result.message['message'] == LEDGER_ACCOUNT_ENTRIES_SUCCESS_MSG.format( len(mock_ledger_reserve_taken) ) mock_ledger_reserve_taken_model.get_by_accounting_run.assert_called_once_with( accounting_run_id, 100000, DEFAULT_PAGE_OFFSET ) mock_build_ledger_entires.assert_called_once_with(mock_ledger_account_contract) mock_db.session.bulk_save_objects.assert_any_call(mock_ledger_account_contract) mock_db.session.commit.assert_called_once() mock_db.session.close.assert_called_once() @patch('ledger.logic.ledger_reserve_taken.LedgerReserveTaken') def test_get_existing_ledger_records(mock_model, mock_event_fixtures): """Test getting existing ledger_reserve_taken entries.""" abacus_event_ids = [1, 2, 3] accounting_run_ids = [1, 2, 3] contract_reserve_ids = [1, 2, 3] ledger_reserve_taken = LedgerReserveTakenFactory.create( abacus_event_id=1, accounting_run_id=3, contract_reserve_id=2 ) mock_model.get_by_abacus_events_accounting_runs_contract_reserves.return_value = [ ledger_reserve_taken ] res = logic._get_existing_ledger_records( abacus_event_ids, accounting_run_ids, contract_reserve_ids ) assert res assert res[0].get('abacus_event_id') == ledger_reserve_taken.abacus_event_id assert res[0].get('accounting_run_id') == ledger_reserve_taken.accounting_run_id assert res[0].get('contract_reserve_id') == ledger_reserve_taken.contract_reserve_id mock_model.get_by_abacus_events_accounting_runs_contract_reserves.assert_called_once_with( abacus_event_ids, accounting_run_ids, contract_reserve_ids ) @patch('ledger.logic.ledger_reserve_taken.LedgerReserveTaken') def test_get_existing_ledger_records_no_records(mock_model): """Test getting existing ledger_reserve_taken entries when none exist.""" abacus_event_ids = [1, 2, 3] accounting_run_ids = [1, 2, 3] contract_reserve_ids = [1, 2, 3] mock_model.get_by_abacus_events_accounting_runs_contract_reserves.return_value = [] res = logic._get_existing_ledger_records( abacus_event_ids, accounting_run_ids, contract_reserve_ids ) assert not res mock_model.get_by_abacus_events_accounting_runs_contract_reserves.assert_called_once_with( abacus_event_ids, accounting_run_ids, contract_reserve_ids ) @patch('ledger.logic.ledger_reserve_taken.currency_exists') def test_validate_entry_success(mock_currency_exists): """Test _validate_entry for valid currency_code.""" entry = {'currency_code': 'USD'} mock_currency_exists.return_value = True result = logic._validate_entry([], entry) assert result mock_currency_exists.assert_called_once_with('USD') @patch('ledger.logic.ledger_reserve_taken.currency_exists') def test_validate_entry_invalid_currency_code(mock_currency_exists): """Test _validate_entry for an invalid currency_code.""" entry = {'currency_code': 'Test'} mock_currency_exists.return_value = False with pytest.raises( Exception, match=ERROR_UNKNOWN_CURRENCY.format(code=entry['currency_code']) ): logic._validate_entry([], entry) mock_currency_exists.assert_called_once_with('Test') @patch('ledger.logic.ledger_reserve_taken.currency_exists') def test_validate_entry_duplicate_entry(mock_currency_exists): """Test _validate_entry for a duplicate record.""" entry = { 'currency_code': 'EUR', 'abacus_event_id': 1, 'accounting_run_id': 234, 'contract_reserve_id': 567, } existing_records = [ {'abacus_event_id': 1, 'accounting_run_id': 234, 'contract_reserve_id': 567} ] mock_currency_exists.return_value = True with pytest.raises(Exception): logic._validate_entry(existing_records, entry) mock_currency_exists.assert_called_once_with('EUR') def test_validate_post_body_valid(mock_bulk_ledger_reserve_taken_body): """Test that no exception is raised when POST body is valid.""" assert not logic._validate_post_request_body(mock_bulk_ledger_reserve_taken_body) def test_validate_post_body_invalid(mock_bulk_ledger_reserve_taken_body): """Test that an exception is raised when POST body contains duplicates.""" params = [ mock_bulk_ledger_reserve_taken_body[0], mock_bulk_ledger_reserve_taken_body[0], ] with pytest.raises(Exception): logic._validate_post_request_body(params)