"""Unit tests for PaymentGroupPaymentAccount model.""" from decimal import Decimal from abacus_common_logic.connectors.database import db from sqlalchemy import func, select from payment.models.payment_group_payment_account import PaymentGroupPaymentAccount from payment.models.payment_group_payment_account_detail import ( PaymentGroupPaymentAccountDetail, ) from tests.utils.factories import ( PaymentGroupPaymentAccountDetailFactory, PaymentGroupPaymentAccountFactory, PaymentGroupPaymentBatchAccountFactory, PaymentGroupPaymentBatchFactory, PaymentGroupPaymentFactory, WorksheetAccountContractClosingBalanceFactory, WorksheetPayableBalanceAfterTaxFactory, ) def test_create(mock_accounts): """Create a PaymentGroupPaymentAccount.""" payment_group_payment = PaymentGroupPaymentFactory.create() new_payment_account = PaymentGroupPaymentAccount.create( account_id=1, currency_code='USD', payment_group_payment_id=payment_group_payment.payment_group_payment_id, balance_after_tax=0.00, payoneer_program_id=100158970, current_statement_period_id=1, tax_withholding=Decimal('-11.11'), vat_amount=Decimal('10.01'), ) payment_accounts = ( db.session.execute(select(PaymentGroupPaymentAccount)).scalars().all() ) assert len(payment_accounts) == 1 assert payment_group_payment.payment_accounts == payment_accounts payment_account = payment_accounts[0] assert payment_account.account_id == 1 assert payment_account.currency_code == 'USD' assert payment_account.currency_name assert payment_account.current_balance == 0.00 assert not payment_account.deleted_at assert not payment_account.deleted_by assert payment_account.last_payment == 0.00 assert not payment_account.note assert payment_account.payment_group_payment_account_id assert payment_account.payment_group_payment == payment_group_payment assert not payment_account.prior_payment_group_payment_id assert ( payment_account.payoneer_program_id == new_payment_account.payoneer_program_id ) assert payment_account.current_statement_period_id == 1 assert payment_account.tax_withholding == Decimal('-11.11') assert payment_account.vat_amount == Decimal('10.01') def test_get_by_payment_group_payment(mock_accounts, mock_account_payees): """Test getting payment_group_payment_accounts by payment_group_payment.""" payment_group_payment = PaymentGroupPaymentFactory.create() payment_group_payment_id = payment_group_payment.payment_group_payment_id payment_accounts = [ PaymentGroupPaymentAccountFactory.create( account_id=i, payment_group_payment=payment_group_payment, vat_amount=Decimal('20.00'), ) for i in range(1, 7) ] params = { 'limit': 5, 'offset': 0, 'order_by': 'payment_group_payment_account_id', 'order_dir': 'desc', 'payment_group_payment_id': payment_group_payment_id, } res = PaymentGroupPaymentAccount.get_by_payment_group_payment(**params) assert len(payment_accounts) > len(res) assert len(res) == params.get('limit') assert all(item.account_id for item in res) assert all(item.account_name for item in res) assert all(item.contracts_payable for item in res) assert all(item.currency_code for item in res) assert all(item.current_balance for item in res) assert any(item.payment_difference for item in res) assert any(item.percent_difference for item in res) assert all(item.payoneer_program_id for item in res) assert all( item.payment_group_payment_id == payment_group_payment_id for item in res ) assert [item.payment_group_payment_account_id for item in res] == sorted( [pa.payment_group_payment_account_id for pa in payment_accounts], reverse=True )[: params.get('limit')] assert all(item.prior_payment_group_payment_id is None for item in res) assert all(item.balance_after_tax for item in res) assert all(item.tax_withholding is None for item in res) assert all(item.vat_amount == Decimal('20.00') for item in res) def test_get_by_payment_group_payment_not_pending(mock_accounts, mock_account_payees): """Test getting by payment_group_payment excludes pending payment_accounts.""" prior = PaymentGroupPaymentFactory.create() payment_group_payment = PaymentGroupPaymentFactory.create() payment_accounts = [ PaymentGroupPaymentAccountFactory.create( account_id=i, payment_group_payment=payment_group_payment ) for i in range(1, 5) ] pending_payment_account = PaymentGroupPaymentAccountFactory.create( payment_group_payment=payment_group_payment, prior_payment_group_payment_id=prior.payment_group_payment_id, account_id=6, ) params = { 'limit': 5, 'offset': 0, 'order_by': 'currency_code', 'order_dir': 'asc', 'payment_group_payment_id': payment_group_payment.payment_group_payment_id, } res = PaymentGroupPaymentAccount.get_by_payment_group_payment(**params) assert len(res) == len(payment_accounts) assert len(payment_group_payment.payment_accounts) > len(res) assert pending_payment_account.payment_group_payment_account_id not in [ item.payment_group_payment_account_id for item in res ] assert all(item.prior_payment_group_payment_id is None for item in res) def test_get_by_payment_group_payment_pending(mock_accounts, mock_account_payees): """Test getting by payment_group_payment gets only pending payment_accounts.""" prior = PaymentGroupPaymentFactory.create() payment_group_payment = PaymentGroupPaymentFactory.create() pending_payment_accounts = [ PaymentGroupPaymentAccountFactory.create( account_id=i, payment_group_payment=payment_group_payment, prior_payment_group_payment_id=prior.payment_group_payment_id, ) for i in range(1, 5) ] payment_account = PaymentGroupPaymentAccountFactory.create( payment_group_payment=payment_group_payment, account_id=6 ) params = { 'is_pending': True, 'limit': 5, 'offset': 0, 'order_by': 'currency_code', 'order_dir': 'asc', 'payment_group_payment_id': payment_group_payment.payment_group_payment_id, } res = PaymentGroupPaymentAccount.get_by_payment_group_payment(**params) assert len(res) == len(pending_payment_accounts) assert len(payment_group_payment.payment_accounts) > len(res) assert payment_account.payment_group_payment_account_id not in [ item.payment_group_payment_account_id for item in res ] assert all(item.prior_payment_group_payment_id for item in res) def test_get_by_payment_group_search(mock_accounts, mock_account_payees): """Test searching payment_group_payment_accounts by payment_group_payment.""" payment_group_payment = PaymentGroupPaymentFactory.create() payment_group_payment_id = payment_group_payment.payment_group_payment_id payment_accounts = [ PaymentGroupPaymentAccountFactory.create( account_id=i, payment_group_payment=payment_group_payment ) for i in range(1, 7) ] params = { 'limit': 5, 'offset': 0, 'order_by': 'payment_group_payment_account_id', 'order_dir': 'desc', 'payment_group_payment_id': payment_group_payment_id, 'is_pending': False, 'search_term': 'Account 5', } res = PaymentGroupPaymentAccount.get_by_payment_group_payment(**params) assert len(payment_accounts) > len(res) assert len(res) == 1 assert len(res) == 1 assert res[0].account_id == 5 assert res[0].account_name == 'Account 5' def test_get_by_payment_group_payment_count(mock_accounts, mock_account_payees): """Test getting total_count by payment_group_payment.""" payment_group_payment = PaymentGroupPaymentFactory.create() payment_accounts = [ PaymentGroupPaymentAccountFactory.create( account_id=i, payment_group_payment=payment_group_payment ) for i in range(1, 6) ] total_count = PaymentGroupPaymentAccount.get_by_payment_group_payment_count( is_pending=False, payment_group_payment_id=payment_group_payment.payment_group_payment_id, ) assert total_count == len(payment_accounts) def test_get_by_payment_group_payment_count_pending(mock_accounts, mock_account_payees): """Test getting total_count of pending records by payment_group_payment.""" prior = PaymentGroupPaymentFactory.create() payment_group_payment = PaymentGroupPaymentFactory.create() pending_payment_accounts = [ PaymentGroupPaymentAccountFactory.create( account_id=i, prior_payment_group_payment_id=prior.payment_group_payment_id, payment_group_payment=payment_group_payment, ) for i in range(1, 6) ] total_count = PaymentGroupPaymentAccount.get_by_payment_group_payment_count( is_pending=True, payment_group_payment_id=payment_group_payment.payment_group_payment_id, ) assert total_count == len(pending_payment_accounts) def test_group_by_currency_code(mock_accounts): """Test getting totals of payment_group_payment_accounts grouped by currency.""" payment_group_payment = PaymentGroupPaymentFactory.create() payment_group_payment_id = payment_group_payment.payment_group_payment_id eur_payments = [ PaymentGroupPaymentAccountFactory.create( account_id=i, currency_code='EUR', payment_group_payment=payment_group_payment, ) for i in range(1, 3) ] usd_payments = [ PaymentGroupPaymentAccountFactory.create( account_id=i, currency_code='USD', payment_group_payment=payment_group_payment, ) for i in range(3, 7) ] res = PaymentGroupPaymentAccount.group_by_currency_code(payment_group_payment_id) assert len(res) == 2 assert res[0].currency_code == 'USD' assert res[0].account_count == len(usd_payments) assert res[1].currency_code == 'EUR' assert res[1].account_count == len(eur_payments) assert res[0].currency_total > res[1].currency_total assert all(pa.payment_group_payment_id == payment_group_payment_id for pa in res) def test_last_posted_payment_by_account(mock_payments): """Test getting an account's last posted payment.""" account_id = 7 payment_group_payment_id = 501 res = PaymentGroupPaymentAccount.last_posted_payment_by_account(account_id) assert res assert res.account_id == account_id assert res.payment_group_payment_id == payment_group_payment_id assert res.payment_group_payment_account_id == account_id def test_last_posted_payment_by_account_knr(mock_payments_knr): """Test getting a knr account's last posted payment (no payoneer status).""" account_id = 9 payment_group_payment_id = 502 res = PaymentGroupPaymentAccount.last_posted_payment_by_account(account_id) assert res assert res.account_id == account_id assert res.payment_group_payment_id == payment_group_payment_id assert res.payment_group_payment_account_id == account_id def test_last_posted_payment_by_account_none(mock_payments): """Test getting an account's last posted payment when it's rejected.""" account_id = 3 res = PaymentGroupPaymentAccount.last_posted_payment_by_account(account_id) assert res is None def test_soft_delete_by_payment_group_payment(mock_accounts): """Test soft deleting all payment_accounts belonging to a payment_group_payment.""" payment_group_payment = PaymentGroupPaymentFactory.create() payment_accounts = [ PaymentGroupPaymentAccountFactory.create( account_id=i, payment_group_payment=payment_group_payment ) for i in range(1, 4) ] assert all(pa.deleted_at is None for pa in payment_accounts) assert all(pa.deleted_by is None for pa in payment_accounts) PaymentGroupPaymentAccount.soft_delete_by_payment_group_payment( payment_group_payment.payment_group_payment_id ) assert all(pa.deleted_at for pa in payment_accounts) assert all(pa.deleted_by for pa in payment_accounts) def test_hard_delete_by_payment_group_payment(mock_accounts): """Test hard deleting all payment_accounts belonging to a payment_group_payment.""" payment_group_payment = PaymentGroupPaymentFactory.create() payment_accounts = [ PaymentGroupPaymentAccountFactory.create( account_id=i, payment_group_payment=payment_group_payment ) for i in range(1, 4) ] payment_account_ids = [ pa.payment_group_payment_account_id for pa in payment_accounts ] assert ( db.session.execute( select(func.count()).select_from( select(PaymentGroupPaymentAccount) .where( PaymentGroupPaymentAccount.payment_group_payment_account_id.in_( payment_account_ids ) ) .subquery() ) ).scalar_one() == 3 ) PaymentGroupPaymentAccount.hard_delete_by_payment_group_payment( payment_group_payment.payment_group_payment_id ) assert ( db.session.execute( select(func.count()).select_from( select(PaymentGroupPaymentAccount) .where( PaymentGroupPaymentAccount.payment_group_payment_account_id.in_( payment_account_ids ) ) .subquery() ) ).scalar_one() == 0 ) def test_pending_payments_by_account(mock_payments): """Test getting an account's pending payments.""" account_id = 5 payment_group_payment_id = 501 res = PaymentGroupPaymentAccount.pending_payments_by_account(account_id) assert isinstance(res, list) assert len(res) assert res[0].account_id == account_id assert res[0].payment_group_payment_id == payment_group_payment_id assert res[0].payment_group_payment_account_id == account_id def test_pending_payments_by_account_none(mock_accounts, create_mock_payment_state): """Test getting an account's pending payments when it has none.""" account_id = 7 res = PaymentGroupPaymentAccount.pending_payments_by_account(account_id) assert not res assert not len(res) def test_get_payment_totals_grouped_by_program_id( mock_accounts, mock_signing_entities, mock_account_payment_terms ): """Test getting payment totals grouped by program_id.""" account_amounts = { 1: {'amount': 100, 'program_id': 100158970}, 2: {'amount': 200, 'program_id': 100158970}, 3: {'amount': 300, 'program_id': 100158970}, 4: {'amount': 400, 'program_id': 1001}, } payment_group_payment = PaymentGroupPaymentFactory.create() for account_id in account_amounts: PaymentGroupPaymentAccountFactory.create( account_id=account_id, balance_after_tax=account_amounts[account_id]['amount'], payment_group_payment=payment_group_payment, payoneer_program_id=account_amounts[account_id]['program_id'], ) results = PaymentGroupPaymentAccount.get_payment_totals_grouped_by_program_id( payment_group_payment.payment_group_payment_id ) assert len(results) == 2 assert results[0].currency_total == 100 + 200 + 300 assert results[1].currency_total == 400 assert results[0].account_count == 3 assert results[1].account_count == 1 def test_get_by_payment_group_payment_with_payment_status( mock_accounts, mock_account_payees, create_mock_payment_state ): """Test getting payment_group_payment_accounts by payment_group_payment. for each account, get the payment_status and payment_error_code after sending payments to Payoneer. """ payment_group_payment = PaymentGroupPaymentFactory.create() payment_group_payment_id = payment_group_payment.payment_group_payment_id payment_accounts_100158970 = [ PaymentGroupPaymentAccountFactory.create( payment_group_payment_account_id=i, account_id=i, payoneer_program_id=100158970, payment_group_payment=payment_group_payment, ) for i in range(1, 3) ] PaymentGroupPaymentAccountFactory.create( payment_group_payment_account_id=3, account_id=3, payoneer_program_id=1001, payment_group_payment=payment_group_payment, ) payment_batch_100158970 = PaymentGroupPaymentBatchFactory.create( payment_group_payment_batch_id=1, batch_num=1, payment_group_payment=payment_group_payment, payoneer_program_id=100158970, ) [ PaymentGroupPaymentBatchAccountFactory.create( payment_group_payment_account=payment_account, payment_group_payment_batch=payment_batch_100158970, ) for payment_account in payment_accounts_100158970 ] params = { 'limit': 5, 'offset': 0, 'order_by': 'payment_status', 'order_dir': 'desc', 'payment_group_payment_id': payment_group_payment_id, } res = PaymentGroupPaymentAccount.get_by_payment_group_payment(**params) assert len(res) == 3 assert all(item.payment_status == 'Batch Failure' for item in res[0:2]) assert all(item.payment_error_code == 'Batch 001 Has Failed' for item in res[0:2]) assert res[2].payment_status is None assert res[2].payment_error_code is None def test_get_payment_group_payment_account_status_overview(mock_payments): """Test to get payment_group_payment_account's payment status overview.""" payment_group_payment_id = 501 res = PaymentGroupPaymentAccount.get_payment_group_payment_account_status_overview( payment_group_payment_id ) assert int(res.number_of_canceled_payments) == 1 assert int(res.number_of_failed_payments) == 1 assert int(res.number_of_pending_payments) == 2 assert int(res.number_of_successful_payments) == 1 assert int(res.number_of_payments_failed_with_batch) == 2 def test_get_payment_group_payment_account_status_overview_no_group(mock_payments): """Test to get payment_group_payment_account's payment status overview.""" payment_group_payment_id = 9999999999 res = PaymentGroupPaymentAccount.get_payment_group_payment_account_status_overview( payment_group_payment_id ) assert res is None def test_get_by_payment_group_payment_account( mock_accounts, mock_contracts, mock_abacus_event, mock_ledger_account_contracts ): """Test get_by_payment_group_payment_account method.""" worksheet_closing_balance = WorksheetAccountContractClosingBalanceFactory.create() worksheet_after_tax1 = WorksheetPayableBalanceAfterTaxFactory.create( # noqa worksheet_account_contract_closing_balance_id=worksheet_closing_balance.worksheet_account_contract_closing_balance_id ) detail1 = PaymentGroupPaymentAccountDetailFactory.create( # noqa worksheet_account_contract_payable_after_tax_id=worksheet_after_tax1.worksheet_account_contract_payable_after_tax_id ) worksheet_after_tax2 = WorksheetPayableBalanceAfterTaxFactory.create( # noqa worksheet_account_contract_closing_balance_id=worksheet_closing_balance.worksheet_account_contract_closing_balance_id ) PaymentGroupPaymentAccountDetailFactory.create( # noqa worksheet_account_contract_payable_after_tax_id=worksheet_after_tax2.worksheet_account_contract_payable_after_tax_id ) details_items = ( PaymentGroupPaymentAccountDetail.get_by_payment_group_payment_account( # noqa detail1.payment_group_payment_account_id ) ) assert details_items == [detail1]