"""BasePostTaxDetailsInput tests""" import logging from unittest.mock import _Call, call, MagicMock, Mock, patch from faker import Faker import pytest from src.connectors.exceptions import ( OwsAbacusAccountException, OwsPayeeException, ) from src.models import ( AccountTaxInfo, AccountTaxInfoBulk, Address, BasePostTaxDetailsInput, NewTaxDetails, TaxDetailsDataloaderItem, UpdateAccountTaxInfo, ) from src.processors.base import LogEntry from src.processors.exceptions import ProcessingError from src.processors.tax_details.base_post_tax_details import BasePostTaxDetailsProcessor from tests.unit.factories import ( AccountTaxInfoFactory, TaxDetailsDataloaderItemFactory, ) from tests.utils import create_sample_csv_buffer DEFAULT_BATCH_SIZE = BasePostTaxDetailsProcessor.DEFAULT_BATCH_SIZE class TestBasePostTaxDetailsInput(BasePostTaxDetailsInput): def get_account_payee_tax_details(self) -> NewTaxDetails: return NewTaxDetails( is_vat_registered=self.is_vat_registered, business_name=self.business_name, country_of_tax_residency_code=self.country_of_tax_residence_code, address=Address( country=self.country_code, address1=self.address1, address2=self.address2, city=self.city, postal_code=self.zip, state=self.province, ) if self.is_vat_registered else None, ) def get_account_tax_info(self) -> UpdateAccountTaxInfo: return UpdateAccountTaxInfo( country_of_tax_residence=self.country_of_tax_residence_code ) class TestBasePostTaxDetailsProcessor: """BasePostTaxDetailsProcessor test suite.""" @patch.object(BasePostTaxDetailsProcessor, '_load_data') @patch.object(BasePostTaxDetailsProcessor, '_load_account_payee_ids') @patch.object(BasePostTaxDetailsProcessor, '_load_account_tax_info') @patch.object(BasePostTaxDetailsProcessor, '_load_account_payee_tax_details') @patch.object(BasePostTaxDetailsProcessor, '_validate_data') @patch.object(BasePostTaxDetailsProcessor, '_post_data') def test_process( self, mock_post_data: Mock, mock_validate_data: Mock, mock_load_account_payee_tax_details: Mock, mock_load_account_tax_info: Mock, mock_load_account_payee_ids: Mock, mock_load_data: Mock, faker: Faker, ) -> None: """Test process method.""" processor = BasePostTaxDetailsProcessor(faker.pystr(), faker.pystr()) processor.process() mock_load_data.assert_called_once() mock_load_account_payee_ids.assert_called_once() mock_load_account_tax_info.assert_called_once() mock_load_account_payee_tax_details.assert_called_once() mock_validate_data.assert_called_once() mock_post_data.assert_called_once() _headers = ( 'vendor_id', 'business_name', 'first_name', 'last_name', 'is_vat_registered', 'address1', 'address2', 'province', 'city', 'zip', 'country_code', 'country_of_tax_residence_code', ) def test_load_data_success(self) -> None: """Test loading data from CSV file success.""" buffer = create_sample_csv_buffer( ( tuple(list(self._headers) + ['override']), ( '5', 'business name', 'first name', 'last name', 'TRUE', '123 Baldwin Ave.', 'apt. 321', 'NY', 'New york', '10001', 'USA', 'ESP', '', ), ( '6', '', 'first name', 'last name', 'FALSE', '', '', '', '', '', 'USA', 'ESP', 'TRUE', ), ) ) processor = BasePostTaxDetailsProcessor('', '') processor._input_type = TestBasePostTaxDetailsInput processor._input_file_buffer = buffer processor._load_data() assert processor._data == self._get_mock_data() def test_load_data_failure_no_file(self) -> None: """Test loading data from CSV file failure.""" processor = BasePostTaxDetailsProcessor('', '') with pytest.raises(ProcessingError, match='Unable to read the file'): processor._load_data() def test_load_data_failure_no_content(self) -> None: """Test loading data from CSV file failure, no content.""" buffer = create_sample_csv_buffer((self._headers,)) processor = BasePostTaxDetailsProcessor('', '') processor._input_type = TestBasePostTaxDetailsInput processor._input_file_buffer = buffer with pytest.raises(ProcessingError, match='No items in file'): processor._load_data() def test_load_data_failure_bad_content(self) -> None: """Test loading data from CSV file failure, bad content.""" buffer = create_sample_csv_buffer( ( self._headers, ( 'aaa', 'business name', 'first name', 'last name', 'true', '123 Baldwin Ave.', 'apt. 321', 'NY', 'New york', '10001', 'USA', 'ESP', ), ) ) processor = BasePostTaxDetailsProcessor('', '') processor._input_file_buffer = buffer with pytest.raises(ProcessingError): processor._load_data() @patch('src.processors.tax_details.base_post_tax_details.ows_abacus_account') def test_load_account_payee_ids( self, mock_ows_abacus_account: Mock, faker: Faker ) -> None: """Test _load_account_payee_ids method.""" def generate_ids_map(start: int, end: int) -> dict[int, int]: return {10 + i: 600 + i for i in range(start, end)} items_count = round(DEFAULT_BATCH_SIZE * 1.4) mock_ows_abacus_account.get_payees_by_accounts.side_effect = ( generate_ids_map(0, DEFAULT_BATCH_SIZE), generate_ids_map(DEFAULT_BATCH_SIZE, items_count), ) processor = BasePostTaxDetailsProcessor(faker.pystr(), faker.pystr()) processor._data = generate_ids_map(0, items_count) # type: ignore processor._load_account_payee_ids() assert mock_ows_abacus_account.get_payees_by_accounts.call_args_list == [ call(list(generate_ids_map(0, DEFAULT_BATCH_SIZE).keys())), call(list(generate_ids_map(DEFAULT_BATCH_SIZE, items_count).keys())), ] assert processor._account_id_account_payee_id == generate_ids_map( 0, items_count ) @patch('src.processors.tax_details.base_post_tax_details.ows_abacus_account') def test_load_account_tax_info( self, mock_ows_abacus_account: Mock, faker: Faker, ) -> None: """Test _load_account_tax_info method.""" def generate_map( start: int, end: int, with_value: bool = False ) -> dict[int, AccountTaxInfo | None]: return { 30 + i: ( AccountTaxInfoFactory.build(account_id=30 + i) if with_value else None ) for i in range(start, end) } items_count = round(DEFAULT_BATCH_SIZE * 1.1) data = ( generate_map(0, DEFAULT_BATCH_SIZE, True), generate_map(DEFAULT_BATCH_SIZE, items_count, True), ) mock_ows_abacus_account.get_account_tax_info_bulk.side_effect = [ AccountTaxInfoBulk(items=list(i.values()), total_count=len(i)) for i in data ] processor = BasePostTaxDetailsProcessor(faker.pystr(), faker.pystr()) processor._data = generate_map(0, items_count, False) # type: ignore processor._load_account_tax_info() assert mock_ows_abacus_account.get_account_tax_info_bulk.call_args_list == [ call( list(generate_map(0, DEFAULT_BATCH_SIZE, False).keys()), limit=DEFAULT_BATCH_SIZE, ), call( list(generate_map(DEFAULT_BATCH_SIZE, items_count, False).keys()), limit=DEFAULT_BATCH_SIZE, ), ] assert processor._account_id_account_tax_info == data[0] | data[1] @patch('src.processors.tax_details.base_post_tax_details.ows_payee') def test_load_account_payee_tax_details( self, mock_ows_payee: Mock, faker: Faker, ) -> None: """Test _load_account_payee_tax_details method.""" def generate_ids(start: int, end: int) -> list[int]: return [120 + i for i in range(start, end)] def generate_map(start: int, end: int) -> dict[int, TaxDetailsDataloaderItem]: return { i: TaxDetailsDataloaderItemFactory.build(account_payee_id=i) for i in generate_ids(start, end) } items_count = round(DEFAULT_BATCH_SIZE * 1.15) data = (generate_map(0, 5), generate_map(DEFAULT_BATCH_SIZE, 3)) mock_ows_payee.get_tax_details_bulk.side_effect = [ list(i.values()) for i in data ] processor = BasePostTaxDetailsProcessor(faker.pystr(), faker.pystr()) processor._account_id_account_payee_id = dict( zip(range(0, items_count), generate_ids(0, items_count)) ) processor._load_account_payee_tax_details() assert mock_ows_payee.get_tax_details_bulk.call_args_list == [ call(generate_ids(0, DEFAULT_BATCH_SIZE)), call(generate_ids(DEFAULT_BATCH_SIZE, items_count)), ] assert processor._account_payee_ids_with_tax_details == sum( [list(i.keys()) for i in data], [] ) def test_validate_data_success(self) -> None: """Test _validate_data""" processor = BasePostTaxDetailsProcessor('', '') processor._data = self._get_mock_data() processor._account_id_account_payee_id = {5: 1, 6: 15} processor._account_id_account_tax_info = { 5: AccountTaxInfoFactory().build(account_tax_info_id=2), 6: AccountTaxInfoFactory().build(account_tax_info_id=16), } processor._account_payee_ids_with_tax_details = [] processor._validate_data() assert processor._data == self._get_mock_data(validated=True) def test_validate_data_failure_no_payee(self) -> None: """Test _validate_data""" processor = BasePostTaxDetailsProcessor('', '') processor._data = self._get_mock_data() processor._account_id_account_payee_id = {} processor._validate_data() assert processor._data == {} assert processor._logs == { logging.ERROR: [ LogEntry( message='Payee does not exist, skipped', group_condition='account_payee_not_exist', additional_data={'account_id': 5}, ), LogEntry( message='Payee does not exist, skipped', group_condition='account_payee_not_exist', additional_data={'account_id': 6}, ), ] } def test_validate_data_failure_no_tax_info(self) -> None: """Test _validate_data""" processor = BasePostTaxDetailsProcessor('', '') processor._data = self._get_mock_data() processor._account_id_account_payee_id = {5: 61, 6: 72} processor._account_id_account_tax_info = {} processor._validate_data() assert processor._data == {} assert processor._logs == { logging.ERROR: [ LogEntry( message='Tax info does not exist, skipped', group_condition='tax_info_not_exist', additional_data={'account_id': 5, 'account_payee_id': 61}, ), LogEntry( message='Tax info does not exist, skipped', group_condition='tax_info_not_exist', additional_data={'account_id': 6, 'account_payee_id': 72}, ), ], } def test_validate_data_failure_tax_details_exist(self) -> None: """Test _validate_data""" processor = BasePostTaxDetailsProcessor('', '') processor._data = self._get_mock_data() processor._account_id_account_payee_id = {5: 55, 6: 15} mock_tax_info = AccountTaxInfoFactory.build(account_tax_info_id=16) processor._account_id_account_tax_info = {5: mock_tax_info, 6: mock_tax_info} processor._account_payee_ids_with_tax_details = [55, 15] processor._validate_data() assert processor._data == self._get_mock_data(validated=True, key=6) assert processor._logs == { logging.INFO: [ LogEntry( message='Tax details already exist, overridden', group_condition='tax_details_exist_overridden', additional_data={'account_id': 6, 'account_payee_id': 15}, ), ], logging.WARNING: [ LogEntry( message='Tax details already exist, skipped', group_condition='tax_details_exist_skipped', additional_data={'account_id': 5, 'account_payee_id': 55}, ), ], } @patch('src.processors.tax_details.base_post_tax_details.ows_payee') @patch('src.processors.tax_details.base_post_tax_details.ows_abacus_account') def test_post_data_success( self, mock_ows_abacus_account: MagicMock, mock_ows_payee: MagicMock, ) -> None: """Test _post_data""" processor = BasePostTaxDetailsProcessor('', '') processor._input_type = TestBasePostTaxDetailsInput processor._data = self._get_mock_data(validated=True) processor._post_data() assert ( mock_ows_payee.save_tax_details.call_args_list == self._get_tax_details_calls(processor._data) ) assert ( mock_ows_abacus_account.update_account_tax_info.call_args_list == self._get_tax_info_calls(processor._data) ) assert not processor._logs @patch('src.processors.tax_details.base_post_tax_details.ows_payee') @patch('src.processors.tax_details.base_post_tax_details.ows_abacus_account') def test_post_data_failure_tax_details( self, mock_ows_abacus_account: MagicMock, mock_ows_payee: MagicMock, ) -> None: """Test _post_data""" processor = BasePostTaxDetailsProcessor('', '') processor._data = self._get_mock_data(validated=True) mock_ows_payee.save_tax_details.side_effect = OwsPayeeException( 'Test', error_code='Err1', additional_data={'account_payee_id': 105} ) processor._post_data() assert ( mock_ows_payee.save_tax_details.call_args_list == self._get_tax_details_calls(processor._data) ) mock_ows_abacus_account.update_account_tax_info.assert_not_called() assert processor._logs == { logging.ERROR: [ LogEntry( message='ows-payee failure: Test', group_condition='Err1', additional_data={'account_id': 5, 'account_payee_id': 105}, ), LogEntry( message='ows-payee failure: Test', group_condition='Err1', additional_data={'account_id': 6, 'account_payee_id': 105}, ), ] } @patch('src.processors.tax_details.base_post_tax_details.ows_payee') @patch('src.processors.tax_details.base_post_tax_details.ows_abacus_account') def test_post_data_failure_tax_info( self, mock_ows_abacus_account: MagicMock, mock_ows_payee: MagicMock, ) -> None: """Test _post_data""" processor = BasePostTaxDetailsProcessor('', '') processor._data = self._get_mock_data(validated=True) mock_ows_abacus_account.update_account_tax_info.side_effect = ( OwsAbacusAccountException('Test') ) processor._post_data() assert ( mock_ows_payee.save_tax_details.call_args_list == self._get_tax_details_calls(processor._data) ) assert ( mock_ows_abacus_account.update_account_tax_info.call_args_list == self._get_tax_info_calls(processor._data) ) assert processor._logs == { logging.ERROR: [ LogEntry( message='Unable to save account tax info for 5/2: ows-abacus-account failure: Test' ), LogEntry( message='Unable to save account tax info for 6/16: ows-abacus-account failure: Test' ), ] } @staticmethod def _get_mock_data( validated: bool = False, key: int | None = None ) -> dict[int, BasePostTaxDetailsInput]: data = { 5: TestBasePostTaxDetailsInput( vendor_id=5, business_name='business name', first_name='first name', last_name='last name', is_vat_registered=True, address1='123 Baldwin Ave.', address2='apt. 321', province='NY', city='New york', zip='10001', country_code='USA', country_of_tax_residence_code='ESP', **( { 'account_payee_id': 1, 'account_tax_info_id': 2, } if validated else {} ), ), 6: TestBasePostTaxDetailsInput( vendor_id=6, first_name='first name', last_name='last name', is_vat_registered=False, country_code='USA', country_of_tax_residence_code='ESP', override=True, **( { 'account_payee_id': 15, 'account_tax_info_id': 16, } if validated else {} ), ), } if key: data = {key: data[key]} return data # type: ignore @staticmethod def _get_tax_details_calls(data: dict[int, BasePostTaxDetailsInput]) -> list[_Call]: return [ call( data_item.account_payee_id, NewTaxDetails( is_vat_registered=data_item.is_vat_registered, business_name=data_item.business_name, country_of_tax_residency_code=data_item.country_of_tax_residence_code, address=Address( country=data_item.country_code, address1=data_item.address1, address2=data_item.address2, city=data_item.city, postal_code=data_item.zip, state=data_item.province, ) if data_item.is_vat_registered else None, ), ) for data_item in data.values() ] @staticmethod def _get_tax_info_calls(data: dict[int, BasePostTaxDetailsInput]) -> list[_Call]: return [ call( data_item.account_tax_info_id, UpdateAccountTaxInfo( country_of_tax_residence=data_item.country_of_tax_residence_code ), ) for data_item in data.values() ]