"""Unit tests for Contract Mechanical Deduction logic.""" import json from unittest.mock import call from unittest.mock import MagicMock from unittest.mock import patch from marshmallow import ValidationError import pytest from abacus_contract.constants.constants import MECHANICAL_DEDUCTION_ADMIN_TYPES from abacus_contract.constants.constants import MECHANICAL_DEDUCTION_TERRITORIES from abacus_contract.constants.constants import MECHANICAL_DEDUCTION_TYPES from abacus_contract.constants.error import \ ERROR_MECH_DEDUCTION_CAN_NOT_HAVE_SAME_TERRITORY from abacus_contract.constants.error import ERROR_MECHANICAL_DEDUCTION_ALREADY_EXIST from abacus_contract.constants.error import ERROR_MECHANICAL_TYPE_PHYSICAL_ONLY_ALLOW from abacus_contract.logic import contract_mechanical_deduction as logic from tests.utils.factories import ContractFactory from tests.utils.factories import ContractMechanicalDeductionFactory @patch('abacus_contract.logic.contract_mechanical_deduction.Contract') def test_get_contract_mechanical_deductions_by_contract_id( mock_contract_model ): """Test getting list of contract mechanical deductions by contract_id.""" mock_contract = ContractFactory.create() mock_contract_mechanical_deduction = ContractMechanicalDeductionFactory.create( contract=mock_contract ) contract_id = mock_contract.contract_id mock_contract_model.get_by_id_or_error.return_value = mock_contract (res, _) = logic.get_contract_mechanical_deductions_by_contract_id(contract_id) assert res.status == 200 assert len(res.message) == 1 assert res.message == [{ 'contract_id': mock_contract_mechanical_deduction.contract_id, 'contract_mechanical_deduction_id': mock_contract_mechanical_deduction.contract_mechanical_deduction_id, 'territory': mock_contract_mechanical_deduction.territory, 'mechanical_type': list(mock_contract_mechanical_deduction.mechanical_type), 'admin_type': mock_contract_mechanical_deduction.admin_type, 'admin_fee': str(mock_contract_mechanical_deduction.admin_fee) }] @patch('abacus_contract.logic.contract_mechanical_deduction.Contract') def test_get_contract_mechanical_deductions_by_contract_id_with_soft_deleted_deductions( mock_contract_model ): """Test getting list of contract mechanical deductions by contract_id. filters out soft deleted contract mechanical deductions. """ mock_contract = ContractFactory.create() mock_contract_mechanical_deductions = [ ContractMechanicalDeductionFactory.create( contract=mock_contract ), ContractMechanicalDeductionFactory.create( contract=mock_contract ) ] mock_contract_mechanical_deductions[0].deleted_at = '2022-01-01' contract_id = mock_contract.contract_id mock_contract_model.get_by_id_or_error.return_value = mock_contract (res, _) = logic.get_contract_mechanical_deductions_by_contract_id(contract_id) assert res.status == 200 assert len(res.message) == 1 def test_validate_territory_uniqueness(): """Test _validate_territory_uniqueness method.""" mock_contract = ContractFactory.create() ContractMechanicalDeductionFactory.create( contract=mock_contract, territory=MECHANICAL_DEDUCTION_TERRITORIES.USA ) territory = MECHANICAL_DEDUCTION_TERRITORIES.ROW res = logic._validate_territory_uniqueness(mock_contract, territory) assert res is True def test_validate_territory_uniqueness_with_existing_mech_deduction(): """Test _validate_territory_uniqueness method. throws an error when the payload contains mech deductions with the 'ROW' territory that are already present for a contract in db table. """ mock_contract = ContractFactory.create() ContractMechanicalDeductionFactory.create( contract=mock_contract, territory=MECHANICAL_DEDUCTION_TERRITORIES.ROW ) territory = MECHANICAL_DEDUCTION_TERRITORIES.ROW with pytest.raises(ValidationError) as excinfo: logic._validate_territory_uniqueness(mock_contract, territory) assert str(excinfo.value) == \ ERROR_MECH_DEDUCTION_CAN_NOT_HAVE_SAME_TERRITORY.format( mock_contract.contract_id, territory) @patch('abacus_contract.logic.contract_mechanical_deduction.db') @patch('abacus_contract.logic.contract_mechanical_deduction' '._validate_mechanical_type') @patch('abacus_contract.logic.contract_mechanical_deduction' '._validate_territory_uniqueness') @patch('abacus_contract.logic.contract_mechanical_deduction' '.ContractMechanicalDeduction') def test_create_contract_mechanical_deduction( mock_contract_mechanical_deduction_model, mock_validate_territory_uniqueness, mock_validate_mechanical_type, mock_db ): """Test creating contract_mechanical_deduction.""" mock_contract = ContractFactory.create() mock_contract_mechanical_deduction = ContractMechanicalDeductionFactory.create( contract=mock_contract, admin_fee=2.10, admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.CUSTOMER, territory=MECHANICAL_DEDUCTION_TERRITORIES.USA, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL] ) mock_payload_request = { 'admin_fee': 2.10, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.CUSTOMER, 'territory': MECHANICAL_DEDUCTION_TERRITORIES.USA, 'mechanical_type': [MECHANICAL_DEDUCTION_TYPES.PHYSICAL] } mock_validate_territory_uniqueness.return_value = True mock_validate_mechanical_type.return_value = True mock_contract_mechanical_deduction_model.build.return_value = \ mock_contract_mechanical_deduction mock_db.session.commit.return_value = True res = logic.create_contract_mechanical_deduction( mock_contract.contract_id, **mock_payload_request ) assert res.status == 201 mock_contract_mechanical_deduction_model.build.assert_called_once_with( contract_id=mock_contract.contract_id, admin_fee=2.10, admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.CUSTOMER, territory=MECHANICAL_DEDUCTION_TERRITORIES.USA, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL] ) mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.contract_mechanical_deduction.db') @patch('abacus_contract.logic.contract_mechanical_deduction' '._validate_mechanical_type') @patch('abacus_contract.logic.contract_mechanical_deduction' '._validate_territory_uniqueness') @patch('abacus_contract.logic.contract_mechanical_deduction' '.ContractMechanicalDeduction') def test_create_contract_mechanical_deduction_validation_error( mock_contract_mechanical_deduction_model, mock_validate_territory_uniqueness, mock_validate_mechanical_type, mock_db ): """Test validation error is thrown when creating contract_mechanical_deduction.""" mock_contract = ContractFactory.create() mock_payload_request = { 'admin_fee': 1.09, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, 'territory': MECHANICAL_DEDUCTION_TERRITORIES.ROW, 'mechanical_type': [MECHANICAL_DEDUCTION_TYPES.DIGITAL] } mock_validate_territory_uniqueness.side_effect = ValidationError('Same Territories') mock_validate_mechanical_type.return_value = False mock_contract_mechanical_deduction_model.build.return_value = False mock_db.session.commit.return_value = False res = logic.create_contract_mechanical_deduction( mock_contract.contract_id, **mock_payload_request ) assert res.status == 400 assert res.errors['message'] == 'Same Territories' mock_contract_mechanical_deduction_model.build.assert_not_called() mock_db.session.commit.assert_not_called() def test_validate_mechanical_type_for_row_territory(): """Test _validate_mechanical_type function for territory "ROW".""" territory = MECHANICAL_DEDUCTION_TERRITORIES.ROW mechanical_type = [MECHANICAL_DEDUCTION_TYPES.PHYSICAL] res = logic._validate_mechanical_type(mechanical_type, territory) assert res is True def test_validate_mechanical_type_for_usa_territory(): """Test _validate_mechanical_type function for territory "USA".""" territory = MECHANICAL_DEDUCTION_TERRITORIES.USA mechanical_type = [ MECHANICAL_DEDUCTION_TYPES.PHYSICAL, MECHANICAL_DEDUCTION_TYPES.DIGITAL ] res = logic._validate_mechanical_type(mechanical_type, territory) assert res is True def test_validate_mechanical_type_for_row_territory_error(): """Test _validate_mechanical_type function for territory "ROW". throws an error if mechanical_type is not physical. """ territory = MECHANICAL_DEDUCTION_TERRITORIES.ROW mechanical_type = [MECHANICAL_DEDUCTION_TYPES.DIGITAL] with pytest.raises(ValidationError) as excinfo: logic._validate_mechanical_type(mechanical_type, territory) assert str(excinfo.value) == ERROR_MECHANICAL_TYPE_PHYSICAL_ONLY_ALLOW @patch('abacus_contract.logic.contract_mechanical_deduction.db') @patch('abacus_contract.logic.contract_mechanical_deduction' '._validate_mechanical_type') def test_update_contract_mechanical_deduction( mock_validation, mock_db ): """Test updating contract_mechanical_deduction.""" mock_contract_mechanical_deduction = ContractMechanicalDeductionFactory.create( admin_fee=2.0, admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.DIGITAL], territory=MECHANICAL_DEDUCTION_TERRITORIES.USA, ) mock_put_request_body = { 'admin_fee': 1.78, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.BUSINESS, 'mechanical_type': [MECHANICAL_DEDUCTION_TYPES.PHYSICAL] } mock_validation.return_value = True mock_contract_mechanical_deduction.update_attributes = MagicMock() mock_db.session.commit.return_value = True res = logic.update_contract_mechanical_deduction( mock_contract_mechanical_deduction, **mock_put_request_body ) assert res.status == 200 mock_contract_mechanical_deduction.update_attributes.assert_called_once_with( **mock_put_request_body ) mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.contract_mechanical_deduction.db') @patch('abacus_contract.logic.contract_mechanical_deduction' '._validate_mechanical_type') def test_update_contract_mechanical_deduction_validation_error( mock_validation, mock_db ): """Test updating contract_mechanical_deduction.""" mock_contract_mechanical_deduction = ContractMechanicalDeductionFactory.create( admin_fee=2.0, admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.DIGITAL], territory=MECHANICAL_DEDUCTION_TERRITORIES.USA ) mock_put_request_body = { 'admin_fee': 1.78, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.BUSINESS, 'mechanical_type': [MECHANICAL_DEDUCTION_TYPES.PHYSICAL] } mock_validation.side_effect = ValidationError('Validation Error') mock_contract_mechanical_deduction.update_attributes = MagicMock() mock_db.session.commit.return_value = False res = logic.update_contract_mechanical_deduction( mock_contract_mechanical_deduction, **mock_put_request_body ) assert res.status == 400 assert res.errors['message'] == 'Validation Error' mock_contract_mechanical_deduction.update_attributes.assert_not_called() mock_db.session.commit.assert_not_called() @patch('abacus_contract.logic.contract_mechanical_deduction.ContractMechanicalDeduction') # noqa E501 def test_soft_delete_contract_mechanical_deduction_by_id( mock_models ): """Test soft deleting contract mechanical deduction by id.""" mock_contract_mechanical_deduction = ContractMechanicalDeductionFactory.create() mock_contract_mechanical_deduction._soft_delete = MagicMock() res = logic.soft_delete_contract_mechanical_deduction( mock_contract_mechanical_deduction) assert res.status == 204 assert res.message is None mock_contract_mechanical_deduction._soft_delete.assert_called_once() mock_models.commit_changes.assert_called_once() @patch('abacus_contract.logic.contract_mechanical_deduction.db') @patch('abacus_contract.logic.contract_mechanical_deduction' '.ContractMechanicalDeduction') def test_create_contract_mechanical_deductions_worldwide( mock_contract_mechanical_deduction_model, mock_db ): """Test creating contract_mechanical_deductions worldwide.""" mock_contract = ContractFactory.create() mock_payload_request = { 'admin_fee': 20.20, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, 'mechanical_type': [ MECHANICAL_DEDUCTION_TYPES.DIGITAL, MECHANICAL_DEDUCTION_TYPES.PHYSICAL ] } mock_contract_mechanical_deduction_model.build.return_value = True mock_db.session.commit.return_value = True res = logic.create_contract_mechanical_deductions_worldwide( mock_contract.contract_id, **mock_payload_request ) assert res.status == 201 mock_contract_mechanical_deduction_model.build.assert_has_calls([ call( contract_id=mock_contract.contract_id, admin_fee=20.20, admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, territory=MECHANICAL_DEDUCTION_TERRITORIES.CAN, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL] ), call( contract_id=mock_contract.contract_id, admin_fee=20.20, admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, territory=MECHANICAL_DEDUCTION_TERRITORIES.ROW, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL] ), call( contract_id=mock_contract.contract_id, admin_fee=20.20, admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, territory=MECHANICAL_DEDUCTION_TERRITORIES.USA, mechanical_type=[ MECHANICAL_DEDUCTION_TYPES.DIGITAL, MECHANICAL_DEDUCTION_TYPES.PHYSICAL ] ) ]) mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.contract_mechanical_deduction.db') @patch('abacus_contract.logic.contract_mechanical_deduction' '.ContractMechanicalDeduction') def test_create_contract_mechanical_deductions_worldwide_error( mock_contract_mechanical_deduction_model, mock_db ): """Test creating contract_mechanical_deductions worldwide. throws an error when contract already has a mech deduction. """ mock_contract = ContractFactory.create() mock_contract_mechanical_deductions = [ ContractMechanicalDeductionFactory.create(contract=mock_contract) ] mock_payload_request = { 'admin_fee': 20.20, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, 'mechanical_type': [ MECHANICAL_DEDUCTION_TYPES.DIGITAL, MECHANICAL_DEDUCTION_TYPES.PHYSICAL ] } mock_contract_mechanical_deduction_model.build.return_value = \ mock_contract_mechanical_deductions mock_db.session.commit.return_value = True res = logic.create_contract_mechanical_deductions_worldwide( mock_contract.contract_id, **mock_payload_request ) assert res.status == 400 assert res.errors['message'] == \ ERROR_MECHANICAL_DEDUCTION_ALREADY_EXIST.format(mock_contract.contract_id) mock_contract_mechanical_deduction_model.build.assert_not_called() mock_db.session.commit.assert_not_called() @patch('abacus_contract.logic.contract_mechanical_deduction' '.ContractMechanicalDeduction') def test_get_active_contracts_with_mechanical_deductions_logic(mock_model): """Test logic.get_active_contracts_with_mechanical_deductions.""" mock_date = '2025-01-01' mock_model.get_active_contracts_with_mechanical_deductions.return_value = [ { 'account_id': 1, 'contract_id': 101, 'term_type': 'track', 'attachments': json.dumps(['1111111111111', '2222222222222']), 'mechanical_type': 'digital,physical' }, { 'account_id': 2, 'contract_id': 102, 'term_type': 'product', 'attachments': None, 'mechanical_type': '' } ] res = logic.get_active_contracts_with_mechanical_deductions(mock_date) assert res.status == 200 assert len(res.message) == 2 assert set(res.message[0]['attachments']) == {'1111111111111', '2222222222222'} assert res.message[0]['mechanical_type'] == ['digital', 'physical'] assert res.message[1]['attachments'] == [] assert res.message[1]['mechanical_type'] == [] mock_model.get_active_contracts_with_mechanical_deductions.assert_called_once_with( mock_date )