from datetime import date, datetime from unittest.mock import call, Mock, patch from abacus_common_logic.models.base import db import pytest from payee.config import Config from payee.constants.constants import TAX_FORM_TYPES, TAX_FORM_TYPES_W8 from payee.logic import tax_form as logic from payee.logic.exceptions import LogicError from payee.logic.secure_document import ( create_secure_document_details, has_secure_document_details, ) from payee.models.account_payee import AccountPayee from payee.models.tax_form import TaxFormInfo, TaxFormInfoDetails, USTaxFormW9Document from payee.utils.models import object_as_dict from tests.constants import MOCK_AUDIT_FIELDS, MOCK_TAX_FORM_DETAILS from tests.utils.factories import TaxFormInfoFactory def test_get_tax_form_info( fresh_db, account_fixtures, payoneer_program_fixtures, account_payee_fixtures ): """Test successfully retrieving tax from info.""" w8_tax_form_info = TaxFormInfoFactory.create() payload = { 'limit': 10, 'offset': 0, 'is_active': True, 'expiration_date_start': '2022-11-30', 'expiration_date_end': '2027-11-30', 'account_payee_ids': [1, 2, 3], } result = logic.get_records(**payload) assert result == {'items': [w8_tax_form_info], 'total_count': 1} def test_get_tax_form_info_no_results(fresh_db): """Test successfully retrieving tax from info.""" payload = { 'limit': 10, 'offset': 0, 'is_active': True, 'expiration_date_start': '2022-11-30', 'expiration_date_end': '2027-11-30', 'account_payee_ids': [1, 2, 3], } result = logic.get_records(**payload) assert result == {'items': [], 'total_count': 0} def test_get_tax_form_info_no_filters( fresh_db, account_fixtures, payoneer_program_fixtures, account_payee_fixtures ): """Test successfully retrieving tax from info.""" w8_tax_form_info = TaxFormInfoFactory.create() payload = { 'limit': 10, 'offset': 0, } result = logic.get_records(**payload) assert result == {'items': [w8_tax_form_info], 'total_count': 1} @patch('payee.logic.secure_document.get_audit_fields') @patch('payee.logic.tax_form.get_records') def test_get_records_details( mock_get_records, mock_get_audit_fields, fresh_db, account_fixtures, payoneer_program_fixtures, account_payee_fixtures, mock_config, ): """Test successfully retrieving tax form details.""" mock_get_audit_fields.return_value = MOCK_AUDIT_FIELDS w9_tax_form_info = TaxFormInfoFactory.create(tax_form_type='W9') document_class = w9_tax_form_info.tax_form_document_class test_secure_data = { 'account_payee_id': w9_tax_form_info.account_payee_id, 'tax_id_country': 'USA', 'tin_type': 'test tin type', 'tin': 'test tin', 'tax_name': 'test tax name', 'tax_classification': 'test classification', } secure_document_details = create_secure_document_details( mock_config.SDM_CONFIG, document_class, **test_secure_data ) mock_get_records.return_value = {'items': [w9_tax_form_info], 'total_count': 1} payload = { 'limit': 10, 'offset': 0, 'is_active': True, 'expiration_date_start': '2022-11-30', 'expiration_date_end': '2027-11-30', 'account_payee_ids': [1, 2, 3], } result = logic.get_records_details(obscure_pii=False, **payload) assert mock_get_records.call_args_list == [ call([1, 2, 3], '2022-11-30', '2027-11-30', True, 10, 0) ] assert result == { 'items': [ TaxFormInfoDetails( **object_as_dict(w9_tax_form_info, ignore_fields=('revision_id',)), details=secure_document_details, ) ], 'total_count': 1, } result = logic.get_records_details(obscure_pii=True, **payload) secure_document_details['tin'] = None assert result == { 'items': [ TaxFormInfoDetails( **object_as_dict(w9_tax_form_info, ignore_fields=('revision_id')), details=secure_document_details, ) ], 'total_count': 1, } @pytest.mark.parametrize( 'tax_form_type, details_fixture', MOCK_TAX_FORM_DETAILS.items() ) @patch('payee.logic.secure_document.get_audit_fields') def test_create_or_update_record_new_success( mock_get_audit_fields, fresh_db, account_fixtures, payoneer_program_fixtures, account_payee_fixtures, tax_form_type, details_fixture, faker, mocker, ): """Test create_or_update_record function.""" mock_get_audit_fields.return_value = MOCK_AUDIT_FIELDS account_payee_id = 1 signed_date = None if tax_form_type != TAX_FORM_TYPES.W9: signed_date = faker.past_date() mocker.spy(logic, 'db') result = logic.create_or_update_record( account_payee_id, tax_form_type, signed_date, **details_fixture ) expiration_date = None if signed_date and tax_form_type in TAX_FORM_TYPES_W8: expiration_date = date(signed_date.year + 3, 12, 31) assert result.account_payee_id == account_payee_id assert result.tax_form_type == tax_form_type assert result.signed_date == signed_date assert result.expiration_date == expiration_date for key, value in details_fixture.items(): assert getattr(result, key) == value, f'Actual {key} = {value}' assert logic.db.session.commit.called account_payee = AccountPayee.get_by_id(account_payee_id) tax_form_info = account_payee.tax_form_info document_class = tax_form_info.tax_form_document_class document = has_secure_document_details(account_payee_id, document_class) details = document_class.build_response(account_payee_id, document) assert ( result.details == TaxFormInfoDetails( **object_as_dict(tax_form_info, ignore_fields=('revision_id',)), details=details, ).details ) def test_create_or_update_record_new_failure_no_type( fresh_db, account_fixtures, payoneer_program_fixtures, account_payee_fixtures, faker, mocker, ): """Test create_or_update_record function.""" account_payee_id = 1 tax_form_type = 'wrong_form_type' signed_date = faker.past_date() details_fixture = list(MOCK_TAX_FORM_DETAILS.values())[0] mocker.spy(logic, 'db') with pytest.raises(Exception, match='Tax form type is not supported.'): logic.create_or_update_record( account_payee_id, tax_form_type, signed_date, **details_fixture ) assert not logic.db.session.commit.called @pytest.mark.parametrize( 'tax_form_type, details_fixture', MOCK_TAX_FORM_DETAILS.items() ) @patch('payee.logic.secure_document.get_audit_fields') def test_create_or_update_record_existing_success( mock_get_audit_fields, fresh_db, account_fixtures, payoneer_program_fixtures, account_payee_fixtures, tax_form_type, details_fixture, faker, mock_config, mocker, ): """Test create_or_update_record function.""" mock_get_audit_fields.return_value = MOCK_AUDIT_FIELDS account_payee_id = 1 mocker.spy(logic, 'db') tax_form = TaxFormInfoFactory.create(tax_form_type=tax_form_type) create_secure_document_details( mock_config.SDM_CONFIG, tax_form.tax_form_document_class, account_payee_id=account_payee_id, **{ 'tax_id_country': 'AFG', 'tin_type': 'initial test tin type', 'tin': 'initial test tin', 'tax_name': 'initial test tax name', 'tax_classification': 'initial test classification', }, ) signed_date = faker.past_date() result = logic.create_or_update_record( account_payee_id, tax_form_type, signed_date, **details_fixture ) expiration_date = None if signed_date and tax_form_type in TAX_FORM_TYPES_W8: expiration_date = date(signed_date.year + 3, 12, 31) assert result.account_payee_id == account_payee_id assert result.tax_form_type == tax_form_type assert result.signed_date == signed_date assert result.expiration_date == expiration_date for key, value in details_fixture.items(): assert getattr(result, key) == value, f'Actual {key} = {value}' assert logic.db.session.commit.called account_payee = AccountPayee.get_by_id(account_payee_id) tax_form_info = account_payee.tax_form_info document_class = tax_form_info.tax_form_document_class document = has_secure_document_details(account_payee_id, document_class) details = document_class.build_response(account_payee_id, document) assert ( result.details == TaxFormInfoDetails( **object_as_dict(tax_form_info, ignore_fields=('revision_id',)), details=details, ).details ) @pytest.mark.parametrize( 'tax_form_type, details_fixture', MOCK_TAX_FORM_DETAILS.items() ) @patch('payee.logic.tax_form.db') @patch('payee.logic.secure_document.get_audit_fields') def test_create_or_update_record_existing_failure_rollback_document( mock_get_audit_fields, mock_db, fresh_db, account_fixtures, payoneer_program_fixtures, account_payee_fixtures, tax_form_type, details_fixture, faker, mock_config, mocker, ): """Test create_or_update_record function.""" mock_get_audit_fields.return_value = MOCK_AUDIT_FIELDS account_payee_id = 1 mock_db.session.commit.side_effect = Exception('Test error') mocker.spy(logic, '_rollback_to_document') initial_details = { 'tax_id_country': 'AFG', 'tin_type': 'initial test tin type', 'tin': 'initial test tin', 'tax_name': 'initial test tax name', } if tax_form_type == TAX_FORM_TYPES.W9: initial_details['tax_classification'] = 'initial test classification' tax_form = TaxFormInfoFactory.create(tax_form_type=tax_form_type) create_secure_document_details( mock_config.SDM_CONFIG, tax_form.tax_form_document_class, account_payee_id=account_payee_id, **initial_details, ) signed_date = faker.past_date() with pytest.raises(Exception, match='Test error'): logic.create_or_update_record( account_payee_id, tax_form_type, signed_date, **details_fixture ) assert logic._rollback_to_document.called db.session.refresh(tax_form) document_class = tax_form.tax_form_document_class document = has_secure_document_details(account_payee_id, document_class) details = document_class.build_response(account_payee_id, document) for k, v in initial_details.items(): assert details.get(k) == v, f'No rollback for {k}: {v}' @pytest.mark.parametrize( 'sql_exists,dynamo_exists,sql_can_delete', ( (True, True, True), (False, True, True), (True, False, True), (True, True, False), ), ) @patch('payee.logic.tax_form.save_secure_document_details_by_payee_id') @patch('payee.logic.tax_form.delete_secure_document') @patch('payee.logic.tax_form.save_secure_document_details_version') @patch('payee.logic.tax_form.has_secure_document_details') @patch('payee.logic.tax_form.TaxFormInfo') def test_delete_record( mock_tax_form_info: Mock, mock_has_secure_document_details: Mock, mock_save_secure_document_details_version: Mock, mock_delete_secure_document: Mock, mock_save_secure_document_details_by_payee_id: Mock, mock_config: Config, tax_form_info_w9_document: USTaxFormW9Document, sql_exists: bool, dynamo_exists: bool, sql_can_delete: bool, ): """Test delete_record function.""" revision_id = 'rev01' tax_form_info = TaxFormInfoFactory.build(tax_form_type='W9') tax_form_info_id = tax_form_info.account_payee_tax_form_info_id account_payee_id = tax_form_info.account_payee_id document_class = tax_form_info_w9_document.__class__ if sql_exists: mock_tax_form_info.get_by_id_or_error.return_value = tax_form_info else: mock_tax_form_info.get_by_id_or_error.side_effect = Exception('not found') if not sql_can_delete: mock_tax_form_info.delete_with_revision.side_effect = Exception('Test error') mock_has_secure_document_details.return_value = ( tax_form_info_w9_document if dynamo_exists else None ) mock_save_secure_document_details_version.return_value = revision_id if sql_exists and dynamo_exists and sql_can_delete: deleted_revision_id, deleted_document = logic.delete_record(tax_form_info_id) assert deleted_revision_id == revision_id assert deleted_document == tax_form_info_w9_document else: with pytest.raises(Exception): logic.delete_record(tax_form_info_id) mock_tax_form_info.get_by_id_or_error.assert_called_once_with(tax_form_info_id) if sql_exists: mock_has_secure_document_details.assert_called_once_with( account_payee_id, document_class ) else: mock_has_secure_document_details.assert_not_called() if sql_exists and dynamo_exists: mock_save_secure_document_details_version.assert_called_once_with( account_payee_id, document_class, **tax_form_info_w9_document.values, ) mock_delete_secure_document.assert_called_once_with( account_payee_id, document_class ) mock_tax_form_info.delete_with_revision.assert_called_once_with( tax_form_info, revision_id, True ) else: mock_save_secure_document_details_version.assert_not_called() mock_delete_secure_document.assert_not_called() mock_tax_form_info.delete_with_revision.assert_not_called() if sql_can_delete or not sql_exists or not dynamo_exists: mock_save_secure_document_details_by_payee_id.assert_not_called() else: mock_save_secure_document_details_by_payee_id.assert_called_once_with( account_payee_id, document_class, **tax_form_info_w9_document.values, ) @patch('payee.logic.tax_form.create_secure_document_details') @patch.object(TaxFormInfo, 'build') @patch('payee.logic.tax_form.has_secure_document_details') @patch.object(TaxFormInfo, 'get_by_account_payee_id') @patch('payee.logic.tax_form.db') def test_create_or_update_record_change_type_failure( mock_db: Mock, mock_get_by_account_payee_id: Mock, mock_has_secure_document_details: Mock, mock_build: Mock, mock_create_secure_document_details: Mock, ) -> None: """Test create_or_update_record change type failure.""" account_payee_id = 1 mock_get_by_account_payee_id.return_value.tax_form_type = TAX_FORM_TYPES.W8BEN with pytest.raises(LogicError, match='Tax form type field is not editable.'): logic.create_or_update_record( account_payee_id, TAX_FORM_TYPES.W9, datetime.now(), **{'test_data': 123} ) mock_get_by_account_payee_id.assert_called_once_with(account_payee_id) mock_db.session.commit.assert_not_called() mock_get_by_account_payee_id.update_attributes.assert_not_called() mock_has_secure_document_details.assert_not_called() mock_build.assert_not_called() mock_create_secure_document_details.assert_not_called()