"""PostTaxCorrectionsProcessor tests.""" from collections import defaultdict from decimal import Decimal from unittest.mock import call, MagicMock, patch from faker import Faker import pytest from src.constants import DEFAULT_BATCH_SIZE, TaxCorrectionStatuses, TaxCorrectionTypes from src.models import GetTaxCorrectionsResponse, NewTaxCorrection from src.processors.exceptions import ProcessingError from src.processors.tax_corrections import PostTaxCorrectionsProcessor from tests.unit.factories import NewTaxCorrectionFactory, TaxCorrectionFactory from tests.utils import create_sample_csv_buffer class TestPostTaxCorrectionsProcessor: """PostTaxCorrectionsProcessor test suite.""" @patch('src.processors.tax_corrections.post_tax_corrections.is_feature_enabled') def test_process_ff_enabled(self, mock_feature: MagicMock, faker: Faker) -> None: """Test process method.""" mock_feature.return_value = True processor = PostTaxCorrectionsProcessor(faker.pystr(), faker.pystr()) processor._load_data = MagicMock() # type: ignore processor._validate_data = MagicMock() # type: ignore processor._cleanup_pending_data = MagicMock() # type: ignore processor._create_tax_corrections = MagicMock() # type: ignore processor.process() assert processor._load_data.called assert processor._validate_data.called assert processor._cleanup_pending_data.called assert processor._create_tax_corrections.called assert mock_feature.called @patch('src.processors.tax_corrections.post_tax_corrections.is_feature_enabled') def test_process_ff_disabled(self, mock_feature: MagicMock, faker: Faker) -> None: """Test process method.""" mock_feature.return_value = False processor = PostTaxCorrectionsProcessor(faker.pystr(), faker.pystr()) processor._load_data = MagicMock() # type: ignore processor._validate_data = MagicMock() # type: ignore processor._cleanup_pending_data = MagicMock() # type: ignore processor._create_tax_corrections = MagicMock() # type: ignore processor.process() assert processor._load_data.called processor._validate_data.assert_not_called() processor._cleanup_pending_data.assert_not_called() assert processor._create_tax_corrections.called assert mock_feature.called def test_load_data_success(self) -> None: """Test loading data from CSV file success.""" buffer = create_sample_csv_buffer( ( ( 'account_id', 'contract_id', 'correction_statement_period_id', 'correction_type', 'amount', 'currency_code', 'note', ), ('36034', '505338', '305', 'wht', '13.00', 'USD', 'note'), ('36035', '505339', '306', 'vat', '15.00', 'GBP', 'note2'), ) ) processor = PostTaxCorrectionsProcessor('', '') processor._input_file_buffer = buffer processor._load_data() assert processor._data == [ NewTaxCorrection( account_id=36034, contract_id=505338, correction_statement_period_id=305, correction_type=TaxCorrectionTypes.wht, amount=Decimal('13.00'), currency_code='USD', note='note', ), NewTaxCorrection( account_id=36035, contract_id=505339, correction_statement_period_id=306, correction_type=TaxCorrectionTypes.vat, amount=Decimal('15.00'), currency_code='GBP', note='note2', ), ] def test_load_data_failure_no_file(self) -> None: """Test loading data from CSV file failure.""" processor = PostTaxCorrectionsProcessor('', '') with pytest.raises(ProcessingError, match='Unable to read the file'): processor._load_data() def test_load_data_failure_invalid_content(self) -> None: """Test loading data from CSV file failure.""" buffer = create_sample_csv_buffer( ( ( 'account_id', 'contract_id', 'correction_statement_period_id', 'correction_type', 'amount', 'currency_code', 'note', ), ('36034a', '505338b', '305c', '4d', '13.00e', 'USD', 'note'), ('36035', '505339', '306', '5', '15.00', 'GBP', 'note2'), ) ) processor = PostTaxCorrectionsProcessor('', '') processor._input_file_buffer = buffer with pytest.raises(ProcessingError, match='Invalid file content'): processor._load_data() def test_load_data_failure_no_items(self) -> None: """Test loading data from CSV file failure.""" buffer = create_sample_csv_buffer( ( ( 'account_id', 'contract_id', 'correction_statement_period_id', 'correction_type', 'amount', 'currency_code', 'note', ), ) ) processor = PostTaxCorrectionsProcessor('', '') processor._input_file_buffer = buffer with pytest.raises(ProcessingError, match='No items in file to post'): processor._load_data() @patch('src.processors.tax_corrections.post_tax_corrections.ows_payment') def test_validate_data_success( self, mock_ows_payment: MagicMock, faker: Faker ) -> None: """Test data validation success.""" processor = PostTaxCorrectionsProcessor(faker.pystr(), faker.pystr()) processor._data = [ NewTaxCorrection( account_id=36034, contract_id=505337, correction_statement_period_id=305, correction_type=TaxCorrectionTypes.wht, amount=Decimal('13.00'), currency_code='USD', note='note', ), NewTaxCorrection( account_id=36034, contract_id=505338, correction_statement_period_id=305, correction_type=TaxCorrectionTypes.wht, amount=Decimal('13.00'), currency_code='USD', note='note', ), NewTaxCorrection( account_id=36035, contract_id=505339, correction_statement_period_id=306, correction_type=TaxCorrectionTypes.vat, amount=Decimal('15.00'), currency_code='GBP', note='note2', ), ] mock_ows_payment.get_tax_corrections.return_value = GetTaxCorrectionsResponse( items=[], total_count=0 ) processor._validate_data() assert processor._data_combinations == { (305, TaxCorrectionTypes.wht): [505337, 505338], (306, TaxCorrectionTypes.vat): [505339], } assert mock_ows_payment.get_tax_corrections.call_args_list == [ call( TaxCorrectionTypes.wht, TaxCorrectionStatuses.active, 305, [505337, 505338], ), call(TaxCorrectionTypes.vat, TaxCorrectionStatuses.active, 306, [505339]), ] @patch('src.processors.tax_corrections.post_tax_corrections.ows_payment') def test_validate_data_failure( self, mock_ows_payment: MagicMock, faker: Faker ) -> None: """Test data validation failure.""" processor = PostTaxCorrectionsProcessor(faker.pystr(), faker.pystr()) processor._data = [ NewTaxCorrection( account_id=36034, contract_id=505337, correction_statement_period_id=305, correction_type=TaxCorrectionTypes.wht, amount=Decimal('13.00'), currency_code='USD', note='note', ), NewTaxCorrection( account_id=36034, contract_id=505338, correction_statement_period_id=305, correction_type=TaxCorrectionTypes.wht, amount=Decimal('13.00'), currency_code='USD', note='note', ), NewTaxCorrection( account_id=36035, contract_id=505339, correction_statement_period_id=306, correction_type=TaxCorrectionTypes.vat, amount=Decimal('15.00'), currency_code='GBP', note='note2', ), ] mock_ows_payment.get_tax_corrections.return_value = GetTaxCorrectionsResponse( items=[TaxCorrectionFactory.build()], total_count=1 ) with pytest.raises( ProcessingError, match='Active corrections exist for the data' ): processor._validate_data() assert processor._data_combinations == { (305, TaxCorrectionTypes.wht): [505337, 505338], (306, TaxCorrectionTypes.vat): [505339], } assert mock_ows_payment.get_tax_corrections.call_args_list == [ call( TaxCorrectionTypes.wht, TaxCorrectionStatuses.active, 305, [505337, 505338], ) ] @patch('src.processors.tax_corrections.post_tax_corrections.ows_payment') def test_get_pending_corrections_success( self, mock_ows_payment: MagicMock, faker: Faker ) -> None: """Test _get_pending_corrections method.""" processor = PostTaxCorrectionsProcessor(faker.pystr(), faker.pystr()) processor._data_combinations = defaultdict( list, { (305, TaxCorrectionTypes.wht): [505337, 505338], (306, TaxCorrectionTypes.vat): [505339], }, ) test_tax_corrections = TaxCorrectionFactory.batch(DEFAULT_BATCH_SIZE * 2) test_total_count = len(test_tax_corrections) mock_ows_payment.get_tax_corrections.side_effect = [ GetTaxCorrectionsResponse( items=test_tax_corrections[:DEFAULT_BATCH_SIZE], total_count=test_total_count - 100, ), GetTaxCorrectionsResponse( items=test_tax_corrections[DEFAULT_BATCH_SIZE : test_total_count - 100], total_count=test_total_count - 100, ), GetTaxCorrectionsResponse( items=test_tax_corrections[test_total_count - 100 :], total_count=100 ), ] assert processor._get_pending_corrections() == test_tax_corrections assert mock_ows_payment.get_tax_corrections.call_args_list == [ call( TaxCorrectionTypes.wht, TaxCorrectionStatuses.pending, 305, [505337, 505338], DEFAULT_BATCH_SIZE, 0, ), call( TaxCorrectionTypes.wht, TaxCorrectionStatuses.pending, 305, [505337, 505338], DEFAULT_BATCH_SIZE, DEFAULT_BATCH_SIZE, ), call( TaxCorrectionTypes.vat, TaxCorrectionStatuses.pending, 306, [505339], DEFAULT_BATCH_SIZE, 0, ), ] @patch('src.processors.tax_corrections.post_tax_corrections.ows_payment') def test_cleanup_pending_data_with_data( self, mock_ows_payment: MagicMock, faker: Faker ) -> None: """Test _cleanup_pending_data method.""" test_pending_corrections = TaxCorrectionFactory.batch(faker.pyint(1, 10)) processor = PostTaxCorrectionsProcessor(faker.pystr(), faker.pystr()) processor._get_pending_corrections = ( # type: ignore lambda: test_pending_corrections ) processor._cleanup_pending_data() assert mock_ows_payment.delete_tax_corrections.call_args_list == [ call( [ correction.worksheet_tax_correction_id for correction in test_pending_corrections ] ) ] @patch('src.processors.tax_corrections.post_tax_corrections.ows_payment') def test_cleanup_pending_data_no_data( self, mock_ows_payment: MagicMock, faker: Faker ) -> None: """Test _cleanup_pending_data method.""" processor = PostTaxCorrectionsProcessor(faker.pystr(), faker.pystr()) processor._get_pending_corrections = lambda: [] # type: ignore processor._cleanup_pending_data() assert not mock_ows_payment.delete_tax_corrections.called @patch('src.processors.tax_corrections.post_tax_corrections.ows_payment') def test_create_tax_corrections( self, mock_ows_payment: MagicMock, faker: Faker ) -> None: """Test _create_tax_corrections method.""" processor = PostTaxCorrectionsProcessor(faker.pystr(), faker.pystr()) processor._data = NewTaxCorrectionFactory.batch(faker.pyint(1, 10)) processor._create_tax_corrections() assert mock_ows_payment.create_tax_corrections.call_args_list == [ call(processor._data) ]