"""Unit tests for Contract Mechanical Deduction logic.""" import json from unittest.mock import MagicMock, call, patch import pytest from marshmallow import ValidationError from abacus_contract.constants.constants import ( CONTRACT_LIFECYCLE_STATUSES, MECHANICAL_DEDUCTION_ADMIN_TYPES, MECHANICAL_DEDUCTION_TERRITORIES, MECHANICAL_DEDUCTION_TYPES, ) from abacus_contract.constants.error import ( ERROR_MECH_DEDUCTION_CAN_NOT_HAVE_SAME_TERRITORY, ERROR_MECHANICAL_DEDUCTION_ALREADY_EXIST, ERROR_MECHANICAL_TYPE_PHYSICAL_ONLY_ALLOW, ) from abacus_contract.logic import contract_mechanical_deduction as logic from abacus_contract.tests.utils.factories import ( AccountContractFactory, ContractFactory, ContractLifecycleFactory, 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' ) 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 ) @patch('abacus_contract.logic.contract_mechanical_deduction.Contract') def test_get_mechadmin_for_account(mock_contract_model): """Test getting whether or not an account is mechadmin.""" account_id = 1 contract = ContractFactory.build() contract.contract_lifecycle = ContractLifecycleFactory.build( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.ACTIVE ) mech_usa_dig = ContractMechanicalDeductionFactory.build( territory=MECHANICAL_DEDUCTION_TERRITORIES.USA, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.DIGITAL], ) mech_row_phys = ContractMechanicalDeductionFactory.build( territory=MECHANICAL_DEDUCTION_TERRITORIES.ROW, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL], ) contract.contract_mechanical_deductions = [ mech_usa_dig, mech_row_phys, ] inactive_contract = ContractFactory.build() inactive_contract.contract_lifecycle = ContractLifecycleFactory.build( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) inactive_contract.contract_mechanical_deductions = [ ContractMechanicalDeductionFactory.build( territory=MECHANICAL_DEDUCTION_TERRITORIES.USA, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.DIGITAL], ), ContractMechanicalDeductionFactory.build( territory=MECHANICAL_DEDUCTION_TERRITORIES.ROW, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL], ), ] mock_contract_model.get_by_accounts.return_value = [ contract, inactive_contract, ] res = logic.get_mechadmin_for_account(account_id) mock_contract_model.get_by_accounts.assert_called_once_with([account_id]) assert res.status == 200 assert res.message['mechadmin_physical'] is True assert res.message['mechadmin_digital'] is True # Test with no US digital mechs contract.contract_mechanical_deductions = [mech_row_phys] mock_contract_model.get_by_accounts.return_value = [contract] res = logic.get_mechadmin_for_account(account_id) assert res.status == 200 assert res.message['mechadmin_physical'] is True assert res.message['mechadmin_digital'] is False # Test with no physical mechs contract.contract_mechanical_deductions = [mech_usa_dig] mock_contract_model.get_by_accounts.return_value = [contract] res = logic.get_mechadmin_for_account(account_id) assert res.status == 200 assert res.message['mechadmin_physical'] is False assert res.message['mechadmin_digital'] is True # Test with no mechs contract.contract_mechanical_deductions = [] mock_contract_model.get_by_accounts.return_value = [contract] res = logic.get_mechadmin_for_account(account_id) assert res.status == 200 assert res.message['mechadmin_physical'] is False assert res.message['mechadmin_digital'] is False @patch('abacus_contract.logic.contract_mechanical_deduction.Contract') def test_get_mechadmin_for_account_ignores_inactive_contracts(mock_contract_model): """Terminated/inactive contracts must not contribute to mechadmin status.""" account_id = 1 mech_usa_dig = ContractMechanicalDeductionFactory.build( territory=MECHANICAL_DEDUCTION_TERRITORIES.USA, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.DIGITAL], ) mech_row_phys = ContractMechanicalDeductionFactory.build( territory=MECHANICAL_DEDUCTION_TERRITORIES.ROW, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL], ) for status in ( CONTRACT_LIFECYCLE_STATUSES.TERMINATED, CONTRACT_LIFECYCLE_STATUSES.INACTIVE, CONTRACT_LIFECYCLE_STATUSES.INIT, ): contract = ContractFactory.build() contract.contract_lifecycle = ContractLifecycleFactory.build( lifecycle_status=status ) contract.contract_mechanical_deductions = [mech_usa_dig, mech_row_phys] mock_contract_model.get_by_accounts.return_value = [contract] res = logic.get_mechadmin_for_account(account_id) assert res.status == 200 assert res.message['mechadmin_physical'] is False, ( f'status={status} should not set mechadmin_physical' ) assert res.message['mechadmin_digital'] is False, ( f'status={status} should not set mechadmin_digital' )