"""Shared fixtures for tests.""" import datetime import json import os from abacus_common_logic.connectors.database import db import pytest from abacus_account.api import create_app from abacus_account.config import Config from tests.utils.factories import AccountFactory from tests.utils.factories import AccountPayeeFactory from tests.utils.factories import AccountPaymentTermFactory from tests.utils.factories import AccountTaxInfoFactory from tests.utils.factories import PaymentEntityPayoneerProgramFactory from tests.utils.factories import ReferencePaymentTypeFactory from tests.utils.factories import ReferencePayoneerProgramFactory class TestConfig(Config): """Test configuration.""" MYSQL_DB_NAME = os.environ.get( 'MYSQL_TEST_DB_NAME', Config.MYSQL_DB_NAME + '_test') @pytest.fixture(scope='session', autouse=True) def test_app(): """Create a test application.""" return create_app(TestConfig) @pytest.fixture(autouse=True) def test_app_in_context(test_app): """Push the test app onto the context.""" with test_app.app_context(): yield test_app @pytest.fixture(autouse=True) def test_app_request(test_app_in_context): """Push the test app onto the context and trigger preprocessing.""" with test_app_in_context.test_request_context(): test_app_in_context.preprocess_request() yield test_app_in_context @pytest.fixture def fixture_client(test_app_in_context): """Create a client fixture.""" with test_app_in_context.test_client() as test_client: yield test_client @pytest.fixture def fresh_db(): """Refresh the test database.""" top_level_tables = ( 'abacus_state', 'account', 'account_contract', 'account_contract_vat_detail', 'account_payee', 'account_payee_history', 'account_payment_term', 'account_payment_term_template', 'account_tax_info', 'account_tax_info_history', 'contract', 'ledger_account_current_balance', 'ledger_account', 'ledger_account_contract_current_balance', 'ledger_account_contract', 'payment_entity_payoneer_program', 'payment_group', 'payment_group_payment', 'payment_group_payment_account', 'payment_hold', 'payment_hold_history', 'payment_minimum', 'reference_payoneer_program', 'reference_payment_type', 'reference_sap_profit_center', 'reference_signing_entity' ) con = db.engine.connect() con.execute('SET FOREIGN_KEY_CHECKS = 0;') trans = con.begin() for table in top_level_tables: con.execute(f'TRUNCATE TABLE `{table}`') trans.commit() con.execute('SET FOREIGN_KEY_CHECKS = 1;') con.execute( 'DELETE FROM reference_payment_entity ' 'WHERE reference_payment_entity_id in (5, 6, 8)' ) ReferencePaymentTypeFactory.create( reference_payment_type_id=7 ) def payment_eligibility_fixtures( status_payment_eligibility: str = 'approved', status_tax_eligibility: str = 'complete', payoneer_payee_id: int = 100, payoneer_program_id: int = 1, contracts_payable: list = [ { 'contract_id': 111, 'currency_code': 'GBP', 'current_balance': 5000 }], reference_payment_entity_id=5 ): """Insert data for testing an account's payment eligibility.""" account_id = 1 account = AccountFactory.create(account_id=account_id, account_name='Account 1') if payoneer_program_id is None: account_payee = AccountPayeeFactory.create( account=account, payoneer_payee_id=payoneer_payee_id, reference_payoneer_program=None ) else: account_payee = AccountPayeeFactory.create( account=account, payoneer_payee_id=payoneer_payee_id, reference_payoneer_program__payoneer_program_id=payoneer_program_id ) AccountTaxInfoFactory.create(account=account) AccountPaymentTermFactory.create( account=account, currency_code='GBP', payment_entity_id=reference_payment_entity_id ) contract_insert = """ INSERT INTO contract( contract_id, reference_signing_entity_id, contract_name, contract_type, created_by, created_at, last_modified_by, last_modified, reference_sap_profit_center_id ) VALUES {contract_values}; """ # noqa: E501 contract_values = [] for contract in contracts_payable: contract_values.append(f"({contract['contract_id']}, 1, 'Test Contract {contract['contract_id']}', 'distribution', 'user_id', now(), 'user_id', now(), 1)") # noqa: E501 account_contract_insert = """ INSERT INTO account_contract( account_id, contract_id ) VALUES {account_contract_values}; """ account_contract_values = [] for contract in contracts_payable: account_contract_values.append(f"({account_id}, {contract['contract_id']})") payment_minimum_insert = """ INSERT INTO payment_minimum( currency_code, check_amount, wire_transfer_amount, western_union_amount, created_by, created_at, last_modified_by, last_modified ) VALUES ('GBP', 30.00, 30.00, 30.00, 'user_id', NOW(), 'user_id', NOW()); """ ledger_account_contract_insert = """ INSERT INTO ledger_account_contract ( abacus_event_id, account_id, contract_id, currency_code, currency_amount, previous_balance, current_balance, created_by, created_at, last_modified_by, last_modified ) VALUES {ledger_values}; """ ledger_values = [] for contract in contracts_payable: ledger_values.append("""( 123, {account_id}, {contract_id}, '{currency_code}', {current_balance}, 0.00, {current_balance}, 'user_id', NOW(), 'user_id', NOW() )""".format( account_id=account_id, contract_id=contract['contract_id'], currency_code=contract['currency_code'], current_balance=contract['current_balance'], )) group_criteria = { 'currency_codes': ['GBP'], 'reference_payment_entities': [5], 'payment_schedules': ['30_days_after_month_end'], 'reference_agreement_types': [1], } payment_group_insert = """ INSERT INTO payment_group( payment_group_id, group_name, group_criteria, is_reusable, created_at, created_by, last_modified, last_modified_by ) VALUES ( 123, 'Group', '{group_criteria}', 0, NOW(), 'user_id', NOW(), 'user_id' ); """ payment_group_payment_insert = """ INSERT INTO payment_group_payment( payment_group_payment_id, payment_group_id, payment_name, statement_period_id, created_at, created_by, last_modified, last_modified_by ) VALUES (456, 123, 'Payment', 285, NOW(), 'user_id', NOW(), 'user_id'); """ abacus_state_insert = """ INSERT INTO abacus_state( abacus_state_id, parent_table_id, parent_table_name, action_name, action_status, created_at, created_by, last_modified, last_modified_by ) VALUES ( 1, {account_payee_id}, 'account_payee', 'payment_eligibility', '{payment_eligibility_status}', NOW(), 'user_id', NOW(), 'user_id' ), ( 2, {account_payee_id}, 'account_payee', 'tax_eligibility', '{tax_eligibility_status}', NOW(), 'user_id', NOW(), 'user_id' ); """ con = db.engine.connect() trans = con.begin() con.execute('SET FOREIGN_KEY_CHECKS = 0;') con.execute(contract_insert.format(contract_values=','.join(contract_values))) con.execute(account_contract_insert.format( account_contract_values=','.join(account_contract_values)) ) con.execute(payment_minimum_insert) con.execute( ledger_account_contract_insert.format( account_id=account_id, ledger_values=','.join(ledger_values) ) ) con.execute(payment_group_insert.format(group_criteria=json.dumps(group_criteria))) con.execute(payment_group_payment_insert) con.execute( abacus_state_insert.format( account_payee_id=account_payee.account_payee_id, payment_eligibility_status=status_payment_eligibility, tax_eligibility_status=status_tax_eligibility ) ) con.execute('SET FOREIGN_KEY_CHECKS = 1;') trans.commit() @pytest.fixture def account_payment_eligible(reference_payment_entity_fixture): """Create fixtures for accounts eligible for payment.""" payment_eligibility_fixtures() @pytest.fixture def account_payment_eligible_via_closing_balance(reference_payment_entity_fixture): """Create fixtures for accounts eligible for payment via closing balance.""" payment_eligibility_fixtures(contracts_payable=[ {'contract_id': 111, 'currency_code': 'GBP', 'current_balance': 10.00}] ) @pytest.fixture def account_payment_ineligible_balance(reference_payment_entity_fixture): """Create fixtures for accounts ineligible for payment: below payment_minimum.""" payment_eligibility_fixtures(contracts_payable=[ {'contract_id': 111, 'currency_code': 'GBP', 'current_balance': 10.00}]) @pytest.fixture def account_payment_multiple_contract_balances(reference_payment_entity_fixture): """Create fixtures for account with multiple contract balances.""" payment_eligibility_fixtures(contracts_payable=[ {'contract_id': 111, 'currency_code': 'GBP', 'current_balance': 100.00}, {'contract_id': 222, 'currency_code': 'GBP', 'current_balance': -90.00}, {'contract_id': 333, 'currency_code': 'GBP', 'current_balance': 325.86}, ]) @pytest.fixture def account_payment_multiple_contract_balances_via_closing_balance( reference_payment_entity_fixture ): """Create fixtures for account with multiple contract balances.""" payment_eligibility_fixtures(contracts_payable=[ {'contract_id': 111, 'currency_code': 'GBP', 'current_balance': 10.00}, {'contract_id': 222, 'currency_code': 'GBP', 'current_balance': -11.00}, {'contract_id': 333, 'currency_code': 'GBP', 'current_balance': 20.86}, ]) @pytest.fixture def account_payment_ineligible_payee_info( reference_payment_entity_fixture ): """Create fixtures for accounts ineligible for payment: incomplete payee info.""" payment_eligibility_fixtures(status_payment_eligibility='rejected') @pytest.fixture def account_payment_ineligible_tax_details(reference_payment_entity_fixture): """Create fixtures for accounts ineligible for payment: missing tax details.""" payment_eligibility_fixtures(status_tax_eligibility='init') @pytest.fixture def account_payment_ineligible_payoneer_program_id(reference_payment_entity_fixture): """Create fixtures for accounts ineligible for payment: missing payoneer program id.""" # noqa: E501 payment_eligibility_fixtures(payoneer_program_id=None) @pytest.fixture def account_payment_ineligible_payoneer_payee_id(reference_payment_entity_fixture): """Create fixtures for accounts ineligible for payment: missing payoneer payee id.""" # noqa: E501 payment_eligibility_fixtures(payoneer_payee_id=None) @pytest.fixture def account_payment_eligible_via_closing_balance_with_account_ids( reference_payment_entity_fixture ): """Create fixtures for accounts eligible for payment.""" # Create ineligible reference_payment_type for one of the accounts ReferencePaymentTypeFactory.create(reference_payment_type_id=9) account_ids = [1000, 1001, 1002, 1003] account_payee_ids = [] con = db.engine.connect() trans = con.begin() con.execute('SET FOREIGN_KEY_CHECKS = 0;') for idx, account_id in enumerate(account_ids): account = AccountFactory.create( account_id=account_id, account_name=f'Account {account_id}' ) account_payee = AccountPayeeFactory.create( account=account, payoneer_payee_id=account_id + 100, sap_vendor_id=account_id + 1000, reference_payment_type_id=9 if account_id == 1000 else 7, ) account_payee_ids.append(account_payee.account_payee_id) AccountTaxInfoFactory.create(account=account) AccountPaymentTermFactory.create( account=account, currency_code='GBP', payment_entity_id=5 ) contract_values = [ f"({account_id}, 1, 'Test Contract {account_id}', " f"'distribution', 'user_id', NOW(), 'user_id', NOW(), 1)" for account_id in account_ids ] con.execute(f""" INSERT INTO contract(contract_id, reference_signing_entity_id, contract_name, contract_type, created_by, created_at, last_modified_by, last_modified, reference_sap_profit_center_id) VALUES {','.join(contract_values)}; """) account_contract_values = [ f'({account_id}, {account_id})' for account_id in account_ids ] con.execute(f""" INSERT INTO account_contract(account_id, contract_id) VALUES {','.join(account_contract_values)}; """) abacus_state_values = [] for account_payee_id in account_payee_ids: abacus_state_values.extend([ f"({account_payee_id}, 'account_payee', " f"'payment_eligibility', 'approved', NOW(), 'user_id', " f"NOW(), 'user_id')", f"({account_payee_id}, 'account_payee', 'tax_eligibility', " f"'complete', NOW(), 'user_id', NOW(), 'user_id')" ]) con.execute(f""" INSERT INTO abacus_state(parent_table_id, parent_table_name, action_name, action_status, created_at, created_by, last_modified, last_modified_by) VALUES {','.join(abacus_state_values)}; """) con.execute(""" INSERT INTO payment_minimum(currency_code, check_amount, wire_transfer_amount, western_union_amount, created_by, created_at, last_modified_by, last_modified) VALUES ('GBP', 30.00, 30.00, 30.00, 'user_id', NOW(), 'user_id', NOW()) ON DUPLICATE KEY UPDATE currency_code=currency_code; """) group_criteria = {'account_ids': [1000, 1001, 1002, 1003]} con.execute(f""" INSERT INTO payment_group(payment_group_id, group_name, group_criteria, is_reusable, created_at, created_by, last_modified, last_modified_by) VALUES (124, 'Group with multiple account_ids', '{json.dumps(group_criteria)}', 0, NOW(), 'user_id', NOW(), 'user_id'); """) con.execute('SET FOREIGN_KEY_CHECKS = 1;') trans.commit() @pytest.fixture def account_fixtures(): """Insert account data for testing.""" return [AccountFactory.create( account_id=ind, account_name=f'Account {ind}' ) for ind in range(1, 6)] @pytest.fixture def payoneer_programs_fixture(reference_payment_entity_fixture): """Insert data for payoneer_programs.""" ReferencePayoneerProgramFactory.create( payoneer_program_id=10017819, payoneer_program_name='AWAL Core - GBP', funding_currency='GBP' ) ReferencePayoneerProgramFactory.create( payoneer_program_id=10017820, payoneer_program_name='AWAL Core - USD', funding_currency='USD' ) PaymentEntityPayoneerProgramFactory.create( payoneer_program_id=10017819, reference_payment_entity_id=5, payment_currency='GBP', agreement_type_id=1 ) PaymentEntityPayoneerProgramFactory.create( payoneer_program_id=10017820, reference_payment_entity_id=6, payment_currency='USD', agreement_type_id=1 ) PaymentEntityPayoneerProgramFactory.create( payoneer_program_id=10017820, reference_payment_entity_id=8, payment_currency='EUR', agreement_type_id=1 ) @pytest.fixture def account_payee_data_fixtures(): """Insert account payee data for testing.""" return [{ 'account_id': ind, 'account_payee_id': ind, 'last_modified': datetime.datetime.now(), } for ind in range(1, 6)] @pytest.fixture def account_contract_vat_detail_fixtures( account_ids: list = [1, 2], contract_ids: list = [1, 2] ): """Insert data for testing an account_contract_vat_detail.""" for account_id in account_ids: AccountFactory.create( account_id=account_id, account_name=f'Account {account_id}') contract_insert = """ INSERT INTO contract( contract_id, reference_signing_entity_id, contract_name, contract_type, created_by, created_at, last_modified_by, last_modified, reference_sap_profit_center_id ) VALUES {contract_values}; """ # noqa: E501 contract_values = [] for contract_id in contract_ids: contract_values.append(f"({contract_id}, 1, 'Test Contract {contract_id}', 'distribution', 'user_id', now(), 'user_id', now(), 1)") # noqa: E501 con = db.engine.connect() con.execute(contract_insert.format(contract_values=','.join(contract_values))) @pytest.fixture def reference_payment_entity_fixture(): """ Insert reference_payment_entities for orchard-us, orchard-es. This fixture is required, because some of new payment entities were not added using the changelod (but should). """ con = db.engine.connect() reference_payment_entity = """ INSERT IGNORE INTO reference_payment_entity( reference_payment_entity_id, payment_entity_name, country_of_tax_reporting, created_by, created_at, last_modified_by, last_modified ) VALUES (1,'AWAL-UK','GBR','Test User',NOW(),'Test User','2024-12-03 14:46:28'), (2,'AWAL-US','USA','Test User',NOW(),'Test User','2024-12-03 14:46:28'), (3,'KNR-UK','GBR','Test User',NOW(),'Test User','2024-12-03 14:46:28'), (4,'KNR-NL','NLD','Test User',NOW(),'Test User','2024-12-03 14:46:28'), (5,'ORCHARD-US','USA','Test User',NOW(),'Test User','2024-12-03 14:46:28'), (6,'ORCHARD-ES','ESP','Test User',NOW(),'Test User','2024-12-03 14:46:28'), (8,'ORCHARD-DE','DEU','Test User',NOW(),'Test User','2024-12-03 14:46:28'); """ con.execute(reference_payment_entity) @pytest.fixture def reference_signing_entity_fixture(reference_payment_entity_fixture): """Insert reference_signing_entity.""" con = db.engine.connect() reference_sap_profit_center = """ INSERT INTO `reference_sap_profit_center` ( `reference_sap_profit_center_id`, `profit_center`, `company_code`, `business_group`, `display_name`, `created_by`, `created_at`, `last_modified_by`, `last_modified` ) VALUES ( 1, 'UK4923', '4923', 'ORC', 'UK4923', 'Test User', NOW(), 'Test User', '2024-12-03 14:46:28' ), ( 2, 'UK4912', '4912', 'ORC', 'UK4912', 'Test User', NOW(), 'Test User', '2024-12-03 14:46:28' ); """ con.execute(reference_sap_profit_center) reference_signing_entity = """ INSERT INTO `reference_signing_entity`( `reference_signing_entity_id`, `reference_payment_entity_id`, `company_code`, `legal_name`, `vat_number`, `company_registration_number`, `address`, `reference_sap_profit_center_id` ) VALUES (1, 5, '2981', 'Test 1', NULL, NULL, NULL, 1), (2, 6, '6020', 'Test 2', NULL, NULL, NULL, 2); """ con.execute(reference_signing_entity)