"""Tests for contract logic.""" import datetime from typing import Any from unittest.mock import ANY, MagicMock, call, patch import httpx import pytest from abacus_common_logic.connectors.database import db from marshmallow import ValidationError from owsresponse import response from sqlalchemy import and_, text from sqlalchemy.exc import SQLAlchemyError from abacus_account.tests.utils.factories import AccountFactory from abacus_contract.constants import constants, error from abacus_contract.constants.constants import ( CONTRACT_KAFKA_EVENT_NAMES, CONTRACT_TYPES, DEFAULT_CONTRACT_EXCLUSIONS, ) from abacus_contract.logic import contract as logic from abacus_contract.models.contract import Contract from abacus_contract.schemas.contract import ContractDetailSchema from abacus_contract.tests.utils.factories import ( AccountContractFactory, ContractAdvanceFactory, ContractFactory, ContractLifecycleFactory, ContractLifecycleScheduleFactory, LegacyContractFactory, ReferenceSapProfitCenterFactory, ReferenceSigningEntityFactory, SigningEntitySapProfitCenterFactory, ) from royalties.constants.constants import ( ACCOUNTING_PERIOD_STATUSES, ACCOUNTING_RUN_STATUSES, ) from royalties.models.run_controller import RunController from royalties.models.run_controller_contract import RunControllerContract from royalties.tests.utils.factories import ( AccountingPeriodFactory, AccountingRunFactory, RunControllerContractFactory, RunControllerFactory, StatementPeriodAdjustmentFileFactory, StatementPeriodFactory, ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.contract_exclusion.create_contract_exclusions') def test_create_contract_success_with_no_contract_id( mock_create_exclusions, mock_models, mock_emit_contract_event ): """Test success response of create_contract method with no contract id provided.""" mock_contract = ContractFactory.create() mock_create_exclusions.return_value = response.Response(message='OK', status=201) mock_models.Contract.create.return_value = mock_contract reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) post_data = { 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'is_excluded_from_accounting_run': None, } result = logic.create_contract(**post_data) assert result.status == 201 mock_models.Contract.create.assert_called_once_with( **post_data, contract_id=None, execution_date=None, is_primary_contract=False, term_start=None, # to be deprecated term_end=None, # to be deprecated reference_sap_profit_center_id=ANY, ) mock_create_exclusions.assert_called_once_with( mock_contract.contract_id, DEFAULT_CONTRACT_EXCLUSIONS ) mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.contract_exclusion.create_contract_exclusions') def test_create_contract_success_with_contract_id( mock_create_exclusions, mock_models, mock_emit_contract_event ): """Test success response of create_contract method with contract id provided.""" mock_contract = ContractFactory.create() mock_create_exclusions.return_value = response.Response(message='OK', status=201) mock_models.Contract.create.return_value = mock_contract mock_models.Contract.get_by_id.return_value = None reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) post_data = { 'contract_id': 123, 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'is_excluded_from_accounting_run': None, } result = logic.create_contract(**post_data) assert result.status == 201 mock_models.Contract.create.assert_called_once_with( **post_data, execution_date=None, is_primary_contract=False, term_start=None, # to be deprecated term_end=None, # to be deprecated reference_sap_profit_center_id=ANY, ) mock_create_exclusions.assert_called_once_with( mock_contract.contract_id, DEFAULT_CONTRACT_EXCLUSIONS ) mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.contract_exclusion.create_contract_exclusions') def test_create_contract_failure_with_contract_id( mock_create_exclusions, mock_models, ): """Test failure response of create_contract method with existing contract id.""" mock_contract = ContractFactory.create() mock_create_exclusions.return_value = response.Response(message='OK', status=201) mock_models.Contract.create.return_value = mock_contract mock_models.Contract.get_by_id.return_value = mock_contract reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) post_data = { 'contract_id': 123, 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, } result = logic.create_contract(**post_data) assert result.status == 400 mock_models.Contract.create.assert_not_called() @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.legacy_contract') @patch('abacus_contract.logic.contract.contract_exclusion.create_contract_exclusions') def test_create_contract_with_oa_contract_id( mock_create_exclusions, mock_legacy_contract, mock_models, mock_emit_contract_event ): """Test create_contract method with oa_contract_id field.""" oa_contract_id = 1 mock_contract = ContractFactory.create() mock_create_exclusions.return_value = response.Response(message='OK', status=201) mock_models.Contract.create.return_value = mock_contract mock_legacy_contract.create_legacy_contract.return_value = None reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) post_data = { 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'oa_contract_id': oa_contract_id, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'is_excluded_from_accounting_run': None, } res = logic.create_contract(**post_data) assert res.status == 201 mock_models.Contract.create.assert_called_once_with( **{ key: value for key, value in post_data.items() if key not in ['oa_contract_id'] }, contract_id=None, execution_date=None, is_primary_contract=False, term_start=None, # to be deprecated term_end=None, # to be deprecated reference_sap_profit_center_id=ANY, ) mock_legacy_contract.create_legacy_contract.assert_called_once_with( mock_contract.contract_id, oa_contract_id ) mock_create_exclusions.assert_not_called() mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.create_account_contract') @patch('abacus_contract.logic.contract._account_exists') def test_create_contract_with_account_contract_success( mock_account_exists, mock_create_account_contract, mock_models, mock_emit_contract_event, ): """Test create_contract method account_contract data.""" mock_contract = ContractFactory.create() account_id = 1 mock_models.Contract.create.return_value = mock_contract reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) mock_account_exists.return_value = True mock_create_account_contract.return_value = response.Response( message='OK', status=201 ) post_data = { 'account_id': account_id, 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'is_excluded_from_accounting_run': None, } res = logic.create_contract(**post_data) assert res.status == 201 mock_models.Contract.create.assert_called_once_with( **{key: value for key, value in post_data.items() if key not in ['account_id']}, contract_id=None, execution_date=None, is_primary_contract=False, term_start=None, # to be deprecated term_end=None, # to be deprecated reference_sap_profit_center_id=ANY, ) mock_create_account_contract.assert_called_once_with( contract_id=mock_contract.contract_id, account_id=account_id ) mock_models.Contract.create.assert_called_once() mock_account_exists.assert_called_once_with(account_id) mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.contract_exclusion.create_contract_exclusions') def test_create_contract_with_exclusions_success( mock_create_exclusion, mock_models, mock_emit_contract_event ): """Test create_contract method with specific exclusions.""" mock_contract = ContractFactory.create() mock_create_exclusion.return_value = response.Response(message='OK', status=201) mock_models.Contract.create.return_value = mock_contract reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) post_data = { 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'contract_exclusions': {'stores': ['1', '2'], 'countries': ['RUS', 'ALB']}, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, } res = logic.create_contract(**post_data) assert res.status == 201 mock_models.Contract.create.assert_called_once_with( **{ key: value for key, value in post_data.items() if key not in ['contract_exclusions'] }, contract_id=None, execution_date=None, is_primary_contract=False, term_start=None, # to be deprecated term_end=None, # to be deprecated reference_sap_profit_center_id=ANY, ) mock_create_exclusion.assert_called_once_with( mock_contract.contract_id, post_data['contract_exclusions'] ) mock_models.Contract.create.assert_called_once() mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_create_contract_with_legacy_distribution_fail( mock_models, mock_emit_contract_event ): """Test create_contract method for legacy_distribution contract type.""" contract_type = CONTRACT_TYPES.LEGACY_DISTRIBUTION mock_contract = ContractFactory.create(contract_type=contract_type) account_id = 1 reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) mock_models.Contract.create.return_value = mock_contract post_data = { 'account_id': account_id, 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, } res = logic.create_contract(**post_data) assert res.status == 400 assert res.errors['code'] == 'error' assert res.errors['message'] == error.ERROR_INVALID_CONTRACT_TYPE.format( contract_type=contract_type ) mock_models.Contract.create.assert_not_called() mock_emit_contract_event.assert_not_called() @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract._account_exists') def test_create_contract_with_missing_account_fail( mock_account_exists, mock_models, mock_emit_contract_event ): """Test create_contract when account does not exist for account_id.""" mock_contract = ContractFactory.create() account_id = 1 reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) mock_account_exists.return_value = False mock_models.Contract.create.return_value = mock_contract post_data = { 'account_id': account_id, 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, } res = logic.create_contract(**post_data) assert res.status == 400 assert res.errors['code'] == 'error' assert res.errors['message'] == error.ERROR_ACCOUNT_NOT_FOUND.format( account_id=account_id ) mock_models.Contract.create.assert_not_called() mock_emit_contract_event.assert_not_called() @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract._account_exists') def test_create_contract_with_failed_account_check_fail( mock_account_exists, mock_models, mock_emit_contract_event ): """Test create_contract when account check failed.""" mock_contract = ContractFactory.create() account_id = 1 reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) mock_account_exists.side_effect = httpx.ConnectError('Connection error') mock_models.Contract.create.return_value = mock_contract post_data = { 'account_id': account_id, 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, } res = logic.create_contract(**post_data) assert res.status == 500 assert res.errors['code'] == 'error' assert res.errors['message'] == 'Connection error' mock_models.Contract.create.assert_not_called() mock_emit_contract_event.assert_not_called() @patch('abacus_contract.logic.contract.models') def test_get_contracts_by_ids(mock_models: Any) -> None: """Test getting contracts by a list of contract_ids.""" contracts = ContractFactory.create_batch(4) contract_ids = [c.contract_id for c in contracts] mock_models.Contract.get_by_ids.return_value = contracts result = logic.get_contracts_by_ids(contract_ids) assert len(result) == len(contracts) mock_models.Contract.get_by_ids.assert_called_once_with(contract_ids) def test_get_contracts_by_ids_dataloaded(create_mock_account: Any) -> None: """Test getting contracts by a list of contract_ids dataloaded.""" contracts = ContractFactory.create_batch(2) contract_ids = [contract.contract_id for contract in contracts] contract_ids.append(9999) # this method receives serialized contracts and then formats them for dataloading formatted = ContractDetailSchema(many=True).dump(contracts) res = logic.format_contracts_for_dataloader(contract_ids, formatted) assert res.status == 200 assert res.message == [ { 'data': { 'contract_id': contracts[0].contract_id, 'contract_name': contracts[0].contract_name, 'contract_type': contracts[0].contract_type, 'execution_date': contracts[0].execution_date, 'is_excluded_from_accounting_run': contracts[ 0 ].is_excluded_from_accounting_run, 'is_paythrough_contract': contracts[0].is_paythrough_contract, 'is_primary_contract': contracts[0].is_primary_contract, 'initial_start_date': str(contracts[0].initial_start_date), 'oa_contract_id': None, 'term_end': str(contracts[0].term_end), 'term_start': str(contracts[0].term_start), 'reference_signing_entity_id': contracts[0].reference_signing_entity_id, 'reference_sap_profit_center_id': contracts[ 0 ].reference_sap_profit_center_id, 'sap_created_at': None, 'summary_note': contracts[0].summary_note, 'general_note': contracts[0].general_note, 'run_controller_id': None, } }, { 'data': { 'contract_id': contracts[1].contract_id, 'contract_name': contracts[1].contract_name, 'contract_type': contracts[1].contract_type, 'execution_date': contracts[1].execution_date, 'is_excluded_from_accounting_run': contracts[ 1 ].is_excluded_from_accounting_run, 'is_paythrough_contract': contracts[1].is_paythrough_contract, 'is_primary_contract': contracts[1].is_primary_contract, 'initial_start_date': str(contracts[1].initial_start_date), 'oa_contract_id': None, 'term_end': str(contracts[1].term_end), 'term_start': str(contracts[1].term_start), 'reference_signing_entity_id': contracts[1].reference_signing_entity_id, 'reference_sap_profit_center_id': contracts[ 1 ].reference_sap_profit_center_id, 'sap_created_at': None, 'summary_note': contracts[1].summary_note, 'general_note': contracts[1].general_note, 'run_controller_id': None, } }, {'data': None}, ] @patch('abacus_contract.logic.contract.models') def test_get_contracts_by_account(mock_models, create_mock_account): """Test getting contracts by account_id.""" account_id = 1 account_contract = AccountContractFactory.create(account_id=account_id) contract = account_contract.contract mock_models.Contract.get_by_accounts.return_value = [contract] response = logic.get_contracts_by_account([account_id]) assert response.status == 200 assert response.message.get('items') == [ { 'account_id': contract.account_id, 'contract_id': contract.contract_id, 'contract_name': contract.contract_name, 'contract_type': contract.contract_type, 'execution_date': contract.execution_date, 'is_excluded_from_accounting_run': contract.is_excluded_from_accounting_run, 'is_paythrough_contract': contract.is_paythrough_contract, 'is_primary_contract': contract.is_primary_contract, 'initial_start_date': str(contract.initial_start_date), 'oa_contract_id': None, 'term_end': str(contract.term_end), 'term_start': str(contract.term_start), 'reference_signing_entity_id': contract.reference_signing_entity_id, 'reference_sap_profit_center_id': contract.reference_sap_profit_center_id, 'sap_created_at': None, 'summary_note': contract.summary_note, 'general_note': contract.general_note, 'run_controller_id': None, } ] assert response.message.get('total_count') > 0 mock_models.Contract.get_by_accounts.assert_called_once_with([account_id]) @patch('abacus_contract.logic.contract.models') def test_get_contracts_by_account_dataloaded(mock_models, create_mock_account): """Test getting contracts by account_ids dataloaded.""" account_ids = [1, 2] account_contracts = [ AccountContractFactory.create(account_id=account_id) for account_id in account_ids ] mock_models.Contract.get_by_accounts.return_value = [ acc_contract.contract for acc_contract in account_contracts ] response = logic.get_contracts_by_account(account_ids, dataload=True) assert response.status == 200 assert response.message == [ { 'data': [ { 'account_id': acc_contract.contract.account_id, 'contract_id': acc_contract.contract.contract_id, 'contract_name': acc_contract.contract.contract_name, 'contract_type': acc_contract.contract.contract_type, 'execution_date': acc_contract.contract.execution_date, 'is_excluded_from_accounting_run': acc_contract.contract.is_excluded_from_accounting_run, 'is_paythrough_contract': acc_contract.contract.is_paythrough_contract, 'is_primary_contract': acc_contract.contract.is_primary_contract, 'initial_start_date': str(acc_contract.contract.initial_start_date), 'oa_contract_id': None, 'term_end': str(acc_contract.contract.term_end), 'term_start': str(acc_contract.contract.term_start), 'reference_signing_entity_id': acc_contract.contract.reference_signing_entity_id, 'reference_sap_profit_center_id': acc_contract.contract.reference_sap_profit_center_id, 'sap_created_at': None, 'summary_note': acc_contract.contract.summary_note, 'general_note': acc_contract.contract.general_note, 'run_controller_id': None, } ] } for acc_contract in account_contracts ] mock_models.Contract.get_by_accounts.assert_called_once_with(account_ids) @patch('abacus_contract.logic.contract.models') def test_get_contracts_by_oa_contract_ids(mock_models): """Test getting contracts by oa_contract_ids.""" legacy_contract = LegacyContractFactory.create() contract = legacy_contract.contract mock_models.Contract.get_by_legacy_contract_ids.return_value = [contract] response = logic.get_contracts_by_oa_contract_ids([1020]) assert response.status == 200 assert response.message[0]['contract_id'] == contract.contract_id mock_models.Contract.get_by_legacy_contract_ids.assert_called_once_with([1020]) @patch('abacus_contract.logic.contract.models') def test_get_vat_info_by_contract_ids(mock_models): """Test getting contracts vat info by contract_ids.""" contract_id = 1020 mock_models.Contract.get_contract_vat_info_by_contract_ids.return_value = [ { 'account_id': 123, 'contract_id': contract_id, 'country_of_tax_residence': 'GBR', 'account_is_sba_signed': True, 'client_tax_rate': '20', 'supplier_tax_rate': '20', } ] response = logic.get_vat_info_by_contract_ids([1020]) assert response.status == 200 assert response.message[0]['contract_id'] == contract_id mock_models.Contract.get_contract_vat_info_by_contract_ids.assert_called_once_with( [1020] ) @patch('abacus_contract.logic.contract._update_run_controller_and_sibling_contracts') @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract( mock_emit_contract_event, mock__update_run_controller_and_sibling_contracts ) -> None: """Test update_contract method.""" account = AccountFactory.create() account_contract = AccountContractFactory.create( account_id=account.account_id, ) contract = account_contract.contract put_data = {'contract_name': 'Test Contract Name'} result = logic.update_contract(contract, **put_data) assert result.status == 200 assert result.message assert result.message['contract_name'] == 'Test Contract Name' mock_emit_contract_event.assert_called_once_with( contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=False, ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_pc_ff_off_strips_field( mock_models, mock_emit_contract_event, _, create_mock_account ): """FF OFF: reference_sap_profit_center_id in PATCH body is silently ignored.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) original_pc_id = mock_contract.reference_sap_profit_center_id mock_models.Contract.commit_changes.return_value = True put_data = {'reference_sap_profit_center_id': original_pc_id + 999} result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 # Field was stripped — PC unchanged. assert result.message['reference_sap_profit_center_id'] == original_pc_id @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=True, ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_pc_ff_on_valid_pair_applies( mock_models, mock_emit_contract_event, _, create_mock_account ): """FF ON: valid (SE, new_PC) pair in junction → PC change applied.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) new_pc = ReferenceSapProfitCenterFactory.create() SigningEntitySapProfitCenterFactory.create( reference_signing_entity=mock_contract.reference_signing_entity, reference_sap_profit_center=new_pc, ) mock_models.Contract.commit_changes.return_value = True # mock_models.SigningEntitySapProfitCenter.query.filter_by(...).first() returns a # MagicMock (truthy) under the @patch('...models') above, so the junction lookup in # _is_signing_entity_authorized_for_profit_center passes. put_data = {'reference_sap_profit_center_id': new_pc.reference_sap_profit_center_id} result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert ( result.message['reference_sap_profit_center_id'] == new_pc.reference_sap_profit_center_id ) @patch( 'abacus_contract.logic.contract._is_signing_entity_authorized_for_profit_center', return_value=False, ) @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=True, ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_pc_ff_on_invalid_pair_returns_error( mock_models, mock_emit_contract_event, _, __, create_mock_account ): """FF ON: (SE, new_PC) not in junction → ValidationError, no event emitted.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) put_data = {'reference_sap_profit_center_id': 99999} result = logic.update_contract(mock_contract, **put_data) assert result.status == 400 mock_emit_contract_event.assert_not_called() @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=False, ) @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_ff_off_se_change_updates_pc_to_legacy( mock_emit_contract_event, _, create_mock_account ): """FF OFF + SE in PATCH + no PC: PC is set to the new SE's legacy PC.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) new_se = ReferenceSigningEntityFactory.create() put_data = {'reference_signing_entity_id': new_se.reference_signing_entity_id} result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert ( result.message['reference_sap_profit_center_id'] == new_se.reference_sap_profit_center_id ) @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=False, ) @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_ff_off_se_and_pc_change_strips_caller_pc( mock_emit_contract_event, _, create_mock_account ): """FF OFF + SE + PC: PC field is ignored, replaced with new SE's legacy PC.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) new_se = ReferenceSigningEntityFactory.create() unrelated_pc = ReferenceSapProfitCenterFactory.create() put_data = { 'reference_signing_entity_id': new_se.reference_signing_entity_id, 'reference_sap_profit_center_id': unrelated_pc.reference_sap_profit_center_id, } result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 # Caller's PC is dropped; persisted value follows the new SE's legacy. assert ( result.message['reference_sap_profit_center_id'] == new_se.reference_sap_profit_center_id ) @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=False, ) @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_ff_off_se_not_found_returns_error( mock_emit_contract_event, _, create_mock_account ): """FF OFF + SE in PATCH but the SE doesn't exist: ValidationError.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) put_data = {'reference_signing_entity_id': 99999999} result = logic.update_contract(mock_contract, **put_data) assert result.status == 400 mock_emit_contract_event.assert_not_called() @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=True, ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_ff_on_se_change_without_pc_returns_error( mock_models, mock_emit_contract_event, _, create_mock_account ): """FF ON + SE in PATCH + no PC: ValidationError — PC required.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) put_data = {'reference_signing_entity_id': 42} result = logic.update_contract(mock_contract, **put_data) assert result.status == 400 mock_emit_contract_event.assert_not_called() @patch( 'abacus_contract.logic.contract._is_signing_entity_authorized_for_profit_center', return_value=True, ) @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=True, ) @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_ff_on_se_and_pc_change_validates_with_new_se( mock_emit_contract_event, _, mock_authorized, create_mock_account, ): """FF ON + SE + PC: junction lookup uses the NEW SE id, not contract's old SE.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) new_se = ReferenceSigningEntityFactory.create() new_pc = ReferenceSapProfitCenterFactory.create() SigningEntitySapProfitCenterFactory.create( reference_signing_entity=new_se, reference_sap_profit_center=new_pc ) put_data = { 'reference_signing_entity_id': new_se.reference_signing_entity_id, 'reference_sap_profit_center_id': new_pc.reference_sap_profit_center_id, } result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 # Confirm validation was called with the NEW SE, not the contract's old SE. mock_authorized.assert_called_once_with( new_se.reference_signing_entity_id, new_pc.reference_sap_profit_center_id ) @patch('abacus_contract.logic.contract._is_signing_entity_authorized_for_profit_center') @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=True, ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_ff_on_no_se_or_pc_skips_junction_lookup( mock_models, mock_emit_contract_event, _, mock_authorized, create_mock_account ): """FF ON + neither SE nor PC in PATCH: no-op path; junction is never consulted.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) mock_models.Contract.commit_changes.return_value = True put_data = {'contract_name': 'Renamed Contract'} result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert result.message['contract_name'] == 'Renamed Contract' mock_authorized.assert_not_called() @patch( 'abacus_contract.logic.contract._is_signing_entity_authorized_for_profit_center', return_value=False, ) @patch( 'abacus_contract.logic.contract.is_single_supply_chain_company_codes_enabled', return_value=True, ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_ff_on_se_and_pc_invalid_pair_returns_error( mock_models, mock_emit_contract_event, _, mock_authorized, create_mock_account ): """FF ON + new SE + new PC with no junction row: ValidationError, validated against NEW SE.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) new_se = ReferenceSigningEntityFactory.create() new_pc = ReferenceSapProfitCenterFactory.create() put_data = { 'reference_signing_entity_id': new_se.reference_signing_entity_id, 'reference_sap_profit_center_id': new_pc.reference_sap_profit_center_id, } result = logic.update_contract(mock_contract, **put_data) assert result.status == 400 mock_emit_contract_event.assert_not_called() # Validation must use the NEW SE id, not the contract's existing SE. mock_authorized.assert_called_once_with( new_se.reference_signing_entity_id, new_pc.reference_sap_profit_center_id ) @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_primary_to_non_primary_ff_disabled( mock_emit_contract_event, mock_ff_enabled ) -> None: """Test update_contract method to update primary to non primary when ff is disabled.""" # noqa: E501 mock_ff_enabled.return_value = False contract = ContractFactory.create(is_primary_contract=True) account = AccountFactory.create() AccountContractFactory.create( contract=contract, account_id=account.account_id, ) put_data = { 'contract_name': 'Test Contract Name', 'is_primary_contract': False, } result = logic.update_contract(contract, **put_data) assert result.status == 200 assert result.message assert result.message['contract_name'] == 'Test Contract Name' assert result.message['is_primary_contract'] is True @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_existing_run_controller( mock_emit_contract_event, ) -> None: """Test update_contract method on a contract with an existing run controller.""" account = AccountFactory.create() account_contract = AccountContractFactory.create( account_id=account.account_id, ) contract = account_contract.contract run_controller = RunControllerFactory.create(contract_type=contract.contract_type) new_run_controller = RunControllerFactory.create() RunControllerContractFactory.create( run_controller=run_controller, contract=contract, ) assert run_controller.run_controller_id != new_run_controller.run_controller_id assert contract.run_controller_id == run_controller.run_controller_id put_data = {'run_controller_id': new_run_controller.run_controller_id} result = logic.update_contract(contract, **put_data) assert result.status == 200 assert result.message assert result.message['run_controller_id'] == new_run_controller.run_controller_id mock_emit_contract_event.assert_called_once_with( contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_run_controller_and_siblings( mock_emit_contract_event, ) -> None: """Test update_contract new run_controller on a contract with siblings.""" account = AccountFactory.create() account_contract = AccountContractFactory.create( account_id=account.account_id, ) account_contract_sibling = AccountContractFactory.create( account_id=account.account_id, ) contract = account_contract.contract sibling_contract = account_contract_sibling.contract run_controller = RunControllerFactory.create(contract_type=contract.contract_type) new_run_controller = RunControllerFactory.create() RunControllerContractFactory.create( run_controller=run_controller, contract=contract, ) RunControllerContractFactory.create( run_controller=run_controller, contract=sibling_contract, ) assert contract.run_controller_id != new_run_controller.run_controller_id assert sibling_contract.run_controller_id != new_run_controller.run_controller_id put_data = {'run_controller_id': new_run_controller.run_controller_id} result = logic.update_contract(contract, **put_data) assert result.message assert result.message['run_controller_id'] == new_run_controller.run_controller_id updated_contracts = Contract.query.filter( Contract.contract_id.in_([contract.contract_id, sibling_contract.contract_id]) ) assert [new_run_controller.run_controller_id] * 2 == [ c.run_controller_id for c in updated_contracts ] event_name = CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED mock_emit_contract_event.assert_has_calls( [ call(contract.contract_id, event_name), call(sibling_contract.contract_id, event_name), ] ) @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_run_controller_and_non_siblings( mock_emit_contract_event, ) -> None: """Test update_contract new run_controller on a contract. For a contract with non siblings does not update the run controller for non siblings. Non siblings means contracts of different contract types. """ account = AccountFactory.create() account_contract = AccountContractFactory.create( account_id=account.account_id, ) account_contract_sibling = AccountContractFactory.create( account_id=account.account_id, contract__contract_type=constants.CONTRACT_TYPES.NEIGHBOURING_RIGHTS, ) contract = account_contract.contract sibling_contract = account_contract_sibling.contract run_controller = RunControllerFactory.create(contract_type=contract.contract_type) new_run_controller = RunControllerFactory.create() RunControllerContractFactory.create( run_controller=run_controller, contract=contract, ) RunControllerContractFactory.create( run_controller=run_controller, contract=sibling_contract, ) assert contract.run_controller_id != new_run_controller.run_controller_id assert sibling_contract.run_controller_id != new_run_controller.run_controller_id put_data = {'run_controller_id': new_run_controller.run_controller_id} result = logic.update_contract(contract, **put_data) assert ( result.message and result.message['run_controller_id'] == new_run_controller.run_controller_id ) updated_contracts = Contract.query.filter( Contract.contract_id.in_([contract.contract_id, sibling_contract.contract_id]) ) assert [new_run_controller.run_controller_id, run_controller.run_controller_id] == [ c.run_controller_id for c in updated_contracts ] mock_emit_contract_event.assert_called_once_with( contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.contract_exclusion.create_contract_exclusions') def test_create_contract_with_reference_signing_entity( mock_create_exclusions, mock_models, mock_emit_contract_event ): """Test to create_contract function with reference_signing_entity_id in POST.""" mock_contract = ContractFactory.create() mock_create_exclusions.return_value = response.Response(message='OK', status=201) reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) mock_models.Contract.create.return_value = mock_contract post_data = { 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'is_excluded_from_accounting_run': None, } result = logic.create_contract(**post_data) assert result.status == 201 mock_models.Contract.create.assert_called_once_with( **post_data, contract_id=None, execution_date=None, is_primary_contract=False, term_start=None, # to be deprecated term_end=None, # to be deprecated reference_sap_profit_center_id=ANY, ) mock_create_exclusions.assert_called_once_with( mock_contract.contract_id, DEFAULT_CONTRACT_EXCLUSIONS ) mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_with_reference_signing_entity( mock_models, mock_emit_contract_event, create_mock_account ): """Test update_contract method to update reference_signing_entity_id.""" reference_signing_entity = ReferenceSigningEntityFactory.create(company_code='4926') mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) mock_models.Contract.commit_changes.return_value = True # Under FF OFF, _resolve_pc_update looks up the new SE and uses its legacy PC. # Wire the mocked models namespace to return our real SE. mock_models.ReferenceSigningEntity.query.get.return_value = reference_signing_entity put_data = { 'reference_signing_entity_id': reference_signing_entity.reference_signing_entity_id } result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert ( result.message['reference_signing_entity_id'] == reference_signing_entity.reference_signing_entity_id ) # PC follows the new SE's legacy PC (FF OFF behavior). assert ( result.message['reference_sap_profit_center_id'] == reference_signing_entity.reference_sap_profit_center_id ) mock_models.Contract.commit_changes.assert_called_once() mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_with_execution_date( mock_models, mock_emit_contract_event, create_mock_account ): """Test update_contract method to update execution_date.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) mock_models.Contract.commit_changes.return_value = True put_data = {'execution_date': datetime.date(2024, 8, 11)} result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert result.message['execution_date'] == '2024-08-11' mock_models.Contract.commit_changes.assert_called_once() mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_with_initial_start_date( mock_models, mock_emit_contract_event, create_mock_account ): """Test update_contract method to update initial_start_date.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) mock_models.Contract.commit_changes.return_value = True put_data = {'initial_start_date': datetime.date(2024, 8, 11)} result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert result.message['initial_start_date'] == '2024-08-11' mock_models.Contract.commit_changes.assert_called_once() mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_with_is_paythrough_contract( mock_models, mock_emit_contract_event, create_mock_account ): """Test update_contract method to update is_paythrough_contract to True.""" mock_contract = ContractFactory.create(is_paythrough_contract=0) AccountContractFactory.create(contract=mock_contract) mock_models.Contract.commit_changes.return_value = True put_data = {'is_paythrough_contract': True} result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert result.message['is_paythrough_contract'] is True mock_models.Contract.commit_changes.assert_called_once() mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract_with_is_paythrough_contract_to_false( mock_models, mock_emit_contract_event, create_mock_account ): """Test update_contract method to update is_paythrough_contract to False.""" mock_contract = ContractFactory.create(is_paythrough_contract=1) AccountContractFactory.create(contract=mock_contract) mock_models.Contract.commit_changes.return_value = True put_data = {'is_paythrough_contract': False} result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert result.message['is_paythrough_contract'] is False mock_models.Contract.commit_changes.assert_called_once() mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_create_contract_with_execution_date( mock_models, mock_emit_contract_event, create_mock_account ): """Test create contract when execution_date is present.""" mock_contract = ContractFactory.create(term_start=None) AccountContractFactory.create(contract=mock_contract) mock_models.Contract.create.return_value = mock_contract reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) post_data = { 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': '2024-08-11', 'is_excluded_from_accounting_run': None, } response = logic.create_contract(**post_data) assert response.status == 201 mock_models.Contract.create.assert_called_once_with( **post_data, contract_id=None, is_primary_contract=False, term_start=None, # to be deprecated term_end=None, # to be deprecated reference_sap_profit_center_id=ANY, ) mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_create_contract_with_excluded_from_accounting_run( mock_models, mock_emit_contract_event, create_mock_account ): """Test create contract when is_excluded_from_accounting_run is set to True.""" mock_contract = ContractFactory.create(term_start=None) AccountContractFactory.create(contract=mock_contract) mock_models.Contract.create.return_value = mock_contract reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) post_data = { 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': '2024-08-11', 'is_excluded_from_accounting_run': True, } response = logic.create_contract(**post_data) assert response.status == 201 mock_models.Contract.create.assert_called_once_with( **post_data, contract_id=None, is_primary_contract=False, term_start=None, # to be deprecated term_end=None, # to be deprecated reference_sap_profit_center_id=ANY, ) mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.models') def test__create_contract_success_with_no_contract_id(mock_models): """Test creating contract success with no contract id.""" mock_contract = ContractFactory.create() reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) mock_post_request_payload = { 'account_id': 1234, 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_name': 'Contract 6', 'oa_contract_id': None, } initial_start_date = '2024-08-1' mock_models.Contract.build.return_value = mock_contract mock_models.AccountContract.build.return_value = mock_contract.account_contract mock_models.ContractExclusion.build.return_value = mock_contract.contract_exclusion res = logic._create_contract(mock_post_request_payload, initial_start_date) assert res == mock_contract mock_models.Contract.build.assert_called_once_with( contract_id=None, contract_name=mock_post_request_payload['contract_name'], contract_type=mock_post_request_payload['contract_type'], execution_date=mock_post_request_payload['execution_date'], reference_signing_entity_id=reference_signing_entity_id, reference_sap_profit_center_id=ANY, summary_note=mock_post_request_payload['summary_note'], general_note=mock_post_request_payload['general_note'], initial_start_date=initial_start_date, is_primary_contract=False, ) mock_models.AccountContract.build.assert_called_once_with( account_id=mock_post_request_payload['account_id'], contract_id=mock_contract.contract_id, ) mock_models.ContractExclusion.build.assert_called_once_with( contract_id=mock_contract.contract_id, exclusions=DEFAULT_CONTRACT_EXCLUSIONS ) mock_models.LegacyContract.build.assert_not_called() @patch('abacus_contract.logic.contract.models') def test__create_contract_success_with_contract_id(mock_models): """Test creating contract success with contract id.""" mock_contract = ContractFactory.create() reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) mock_post_request_payload = { 'contract_id': 123, 'account_id': 1234, 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_name': 'Contract 6', 'oa_contract_id': None, } initial_start_date = '2024-08-1' mock_models.Contract.get_by_id.return_value = None mock_models.Contract.build.return_value = mock_contract mock_models.AccountContract.build.return_value = mock_contract.account_contract mock_models.ContractExclusion.build.return_value = mock_contract.contract_exclusion res = logic._create_contract(mock_post_request_payload, initial_start_date) assert res == mock_contract mock_models.Contract.build.assert_called_once_with( contract_id=mock_post_request_payload['contract_id'], contract_name=mock_post_request_payload['contract_name'], contract_type=mock_post_request_payload['contract_type'], execution_date=mock_post_request_payload['execution_date'], reference_signing_entity_id=reference_signing_entity_id, reference_sap_profit_center_id=ANY, summary_note=mock_post_request_payload['summary_note'], general_note=mock_post_request_payload['general_note'], initial_start_date=initial_start_date, is_primary_contract=False, ) mock_models.AccountContract.build.assert_called_once_with( account_id=mock_post_request_payload['account_id'], contract_id=mock_contract.contract_id, ) mock_models.ContractExclusion.build.assert_called_once_with( contract_id=mock_contract.contract_id, exclusions=DEFAULT_CONTRACT_EXCLUSIONS ) mock_models.LegacyContract.build.assert_not_called() @patch('abacus_contract.logic.contract.models') def test__create_contract_failure_with_contract_id(mock_models): """Test creating contract failure with existing contract id.""" mock_contract = ContractFactory.create() reference_signing_entity_id = ( mock_contract.reference_signing_entity.reference_signing_entity_id ) mock_post_request_payload = { 'contract_id': 123, 'account_id': 1234, 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': reference_signing_entity_id, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_name': 'Contract 6', 'oa_contract_id': None, } initial_start_date = '2024-08-1' mock_models.Contract.get_by_id.return_value = mock_contract mock_models.Contract.build.return_value = mock_contract mock_models.AccountContract.build.return_value = mock_contract.account_contract mock_models.ContractExclusion.build.return_value = mock_contract.contract_exclusion with pytest.raises(Exception) as exc_info: logic._create_contract(mock_post_request_payload, initial_start_date) assert 'Contract id 123 already exists.' in str(exc_info.value) mock_models.Contract.build.assert_not_called() mock_models.AccountContract.build.assert_not_called() mock_models.ContractExclusion.build.assert_not_called() mock_models.LegacyContract.build.assert_not_called() @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract._create_contract_lifecycle') @patch('abacus_contract.logic.contract._create_contract_lifecycle_schedules') @patch('abacus_contract.logic.contract._validate_request_payload') @patch('abacus_contract.logic.contract._create_contract') def test_create_contract_with_lifecycle_and_schedules_success( mock_create_contract_logic, mock_validate_request_payload, mock_create_contract_lifecycle_schedules, mock_create_contract_lifecycle, mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, ): """Test creating contract with lifecycle and schedules.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule, ) mock_create_contract_logic.return_value = mock_contract mock_validate_request_payload.return_value = True mock_create_contract_lifecycle_schedules.return_value = [ mock_contract_lifecycle_schedule ] mock_create_contract_lifecycle.return_value = mock_contract_lifecycle lifecycle_term_start = mock_contract_and_lifecycle_post_payload[ 'contract_lifecycle' ]['lifecycle_term_start'] res = logic.create_contract_with_lifecycle_and_schedules( **mock_contract_and_lifecycle_post_payload ) assert res.status == 201 mock_create_contract_logic.assert_called_once_with( mock_contract_and_lifecycle_post_payload['contract'], lifecycle_term_start ) mock_validate_request_payload.assert_called_once_with( mock_contract.contract_id, mock_contract.contract_type, [], mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'], ) mock_create_contract_lifecycle_schedules.assert_called_once_with( mock_contract.contract_id, mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'], ) mock_create_contract_lifecycle.assert_called_once_with( mock_contract.contract_id, lifecycle_term_start, mock_contract_lifecycle_schedule, ) mock_emit_contract_event.assert_called_once_with( mock_contract.contract_id, CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_CREATED ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract._create_contract_lifecycle') @patch('abacus_contract.logic.contract._create_contract_lifecycle_schedules') @patch('abacus_contract.logic.contract._validate_request_payload') @patch('abacus_contract.logic.contract._create_contract') def test_create_contract_with_lifecycle_contract_creation_failed( mock_create_contract_logic, mock_validate_request_payload, mock_create_contract_lifecycle_schedules, mock_create_contract_lifecycle, mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, ): """Test creating contract with lifecycle and schedules. throws an error if contract creation fails. """ mock_create_contract_logic.side_effect = SQLAlchemyError('Contract creating failed') lifecycle_term_start = mock_contract_and_lifecycle_post_payload[ 'contract_lifecycle' ]['lifecycle_term_start'] with pytest.raises(Exception) as exc_info: logic.create_contract_with_lifecycle_and_schedules( **mock_contract_and_lifecycle_post_payload ) assert 'Contract creating failed' in str(exc_info.value) mock_create_contract_logic.assert_called_once_with( mock_contract_and_lifecycle_post_payload['contract'], lifecycle_term_start ) mock_validate_request_payload.assert_not_called() mock_create_contract_lifecycle_schedules.assert_not_called() mock_create_contract_lifecycle.assert_not_called() mock_emit_contract_event.assert_not_called() @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract._create_contract_lifecycle') @patch('abacus_contract.logic.contract._create_contract_lifecycle_schedules') @patch('abacus_contract.logic.contract._validate_request_payload') @patch('abacus_contract.logic.contract._create_contract') def test_create_contract_with_lifecycle_and_schedules_validation_error( mock_create_contract_logic, mock_validate_request_payload, mock_create_contract_lifecycle_schedules, mock_create_contract_lifecycle, mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, ): """Test creating contract with lifecycle and schedules. returns an error when contract_lifecycle_schedule validation fails """ mock_contract = ContractFactory.create() mock_create_contract_logic.return_value = mock_contract validation_msg = 'contract lifecycle schedule validation failed' mock_validate_request_payload.side_effect = ValidationError(validation_msg) lifecycle_term_start = mock_contract_and_lifecycle_post_payload[ 'contract_lifecycle' ]['lifecycle_term_start'] res = logic.create_contract_with_lifecycle_and_schedules( **mock_contract_and_lifecycle_post_payload ) assert res.status == 400 assert res.errors['message'] == validation_msg mock_create_contract_logic.assert_called_once_with( mock_contract_and_lifecycle_post_payload['contract'], lifecycle_term_start ) mock_validate_request_payload.assert_called_once_with( mock_contract.contract_id, mock_contract.contract_type, [], mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'], ) mock_create_contract_lifecycle_schedules.assert_not_called() mock_create_contract_lifecycle.assert_not_called() mock_emit_contract_event.assert_not_called() @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract._create_contract_lifecycle') @patch('abacus_contract.logic.contract._create_contract_lifecycle_schedules') @patch('abacus_contract.logic.contract._validate_request_payload') @patch('abacus_contract.logic.contract._create_contract') def test_create_contract_with_lifecycle_and_schedules_creation_error( mock_create_contract_logic, mock_validate_request_payload, mock_create_contract_lifecycle_schedules, mock_create_contract_lifecycle, mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, ): """Test creating contract with lifecycle and schedules. throws an error when contract_lifecycle_schedule creation fails """ mock_contract = ContractFactory.create() mock_create_contract_logic.return_value = mock_contract mock_validate_request_payload.return_value = True creation_failed_msg = 'contract_lifecycle_schedule creation failed' mock_create_contract_lifecycle_schedules.side_effect = SQLAlchemyError( creation_failed_msg ) lifecycle_term_start = mock_contract_and_lifecycle_post_payload[ 'contract_lifecycle' ]['lifecycle_term_start'] with pytest.raises(Exception) as exc_info: logic.create_contract_with_lifecycle_and_schedules( **mock_contract_and_lifecycle_post_payload ) assert 'contract_lifecycle_schedule creation failed' in str(exc_info.value) mock_create_contract_logic.assert_called_once_with( mock_contract_and_lifecycle_post_payload['contract'], lifecycle_term_start ) mock_validate_request_payload.assert_called_once_with( mock_contract.contract_id, mock_contract.contract_type, [], mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'], ) mock_create_contract_lifecycle_schedules.assert_called_once_with( mock_contract.contract_id, mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'], ) mock_create_contract_lifecycle.assert_not_called() mock_emit_contract_event.assert_not_called() @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract._create_contract_lifecycle') @patch('abacus_contract.logic.contract._create_contract_lifecycle_schedules') @patch('abacus_contract.logic.contract._validate_request_payload') @patch('abacus_contract.logic.contract._create_contract') def test_create_contract_with_lifecycle_creation_error( mock_create_contract_logic, mock_validate_request_payload, mock_create_contract_lifecycle_schedules, mock_create_contract_lifecycle, mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, ): """Test creating contract with lifecycle and schedules. throws an error when contract_lifecycle creation fails """ mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) mock_create_contract_logic.return_value = mock_contract mock_validate_request_payload.return_value = True creation_failed_msg = 'contract_lifecycle creation failed' mock_create_contract_lifecycle_schedules.return_value = [ mock_contract_lifecycle_schedule ] mock_create_contract_lifecycle.side_effect = SQLAlchemyError(creation_failed_msg) lifecycle_term_start = mock_contract_and_lifecycle_post_payload[ 'contract_lifecycle' ]['lifecycle_term_start'] with pytest.raises(Exception) as exc_info: logic.create_contract_with_lifecycle_and_schedules( **mock_contract_and_lifecycle_post_payload ) assert 'contract_lifecycle creation failed' in str(exc_info.value) mock_create_contract_logic.assert_called_once_with( mock_contract_and_lifecycle_post_payload['contract'], lifecycle_term_start ) mock_validate_request_payload.assert_called_once_with( mock_contract.contract_id, mock_contract.contract_type, [], mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'], ) mock_create_contract_lifecycle_schedules.assert_called_once_with( mock_contract.contract_id, mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'], ) mock_create_contract_lifecycle.assert_called_once_with( mock_contract.contract_id, lifecycle_term_start, mock_contract_lifecycle_schedule, ) mock_emit_contract_event.assert_not_called() @patch('abacus_contract.logic.contract.terminate_contract_lifecycle') def test_terminate_contract(mock_terminate_handler): """Test terminate_contract method for contract.""" termination_date = datetime.date(2024, 8, 20) contract = ContractFactory.create() contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=contract, ) ContractLifecycleFactory.create( contract=contract, contract_lifecycle_schedule=contract_lifecycle_schedule, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.ACTIVE, ) mock_terminate_handler.return_value = True res = logic.terminate_contract(contract.contract_id, termination_date, None) assert res.status == 200 mock_terminate_handler.assert_called_once_with( contract.contract_id, termination_date, None ) @patch('abacus_contract.logic.contract.terminate_contract_lifecycle') def test_terminate_contract_error(mock_terminate_handler): """Test terminate_contract method for contract with invalid status.""" termination_date = datetime.date(2024, 8, 20) contract = ContractFactory.create() contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=contract, ) ContractLifecycleFactory.create( contract=contract, contract_lifecycle_schedule=contract_lifecycle_schedule, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED, ) mock_terminate_handler.side_effect = ValidationError('Some error') res = logic.terminate_contract(contract.contract_id, termination_date, None) assert res.status == 400 @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.reactivate_contract_lifecycle') def test_reactivate_contract(mock_reactivate_contract_lifecycle, mock_models): """Test reactivate_contract function.""" mock_contract = ContractFactory.create() mock_existing_contract_lifecycle = ContractLifecycleFactory.create() mock_models.get_by_id_or_error.return_value = mock_contract mock_reactivate_contract_lifecycle.return_value = mock_existing_contract_lifecycle res = logic.reactivate_contract(mock_contract.contract_id) assert res.status == 200 mock_reactivate_contract_lifecycle.assert_called_once_with( mock_contract.contract_id ) @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.reactivate_contract_lifecycle') def test_reactivate_contract_error(mock_reactivate_contract_lifecycle, mock_models): """Test to return an error for reactivate_contract function.""" mock_contract = ContractFactory.create() mock_models.get_by_id_or_error.return_value = mock_contract mock_reactivate_contract_lifecycle.side_effect = ValidationError('some_error') res = logic.reactivate_contract(mock_contract.contract_id) assert res.status == 400 assert res.errors['message'] == 'some_error' def test_get_account_id_by_contract_id(create_mock_account) -> None: """Test getting account_id by contract_id.""" contract = ContractFactory.create() AccountContractFactory.create(account_id=2, contract=contract) assert logic.get_account_id_by_contract_id(contract.contract_id) == 2 def test_get_account_id_by_contract_id_not_found() -> None: """Test getting account_id by contract_id when account_id is not found.""" contract = ContractFactory.create() assert logic.get_account_id_by_contract_id(contract.contract_id) is None def test_get_account_id_map_by_contract_ids(create_mock_account) -> None: """Test mapping each contract_id to its account_id for the batch authz step.""" contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=contract) contract2 = ContractFactory.create() AccountContractFactory.create(account_id=2, contract=contract2) assert logic.get_account_id_map_by_contract_ids( [contract.contract_id, contract2.contract_id] ) == {contract.contract_id: 1, contract2.contract_id: 2} def test_get_account_id_map_by_contract_ids_empty() -> None: """Empty input returns an empty map.""" assert logic.get_account_id_map_by_contract_ids([]) == {} def test_get_account_ids_by_contract_ids(create_mock_account) -> None: """Test getting account_ids by contract_ids.""" contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=contract) contract2 = ContractFactory.create() AccountContractFactory.create(account_id=2, contract=contract2) assert logic.get_account_ids_by_contract_ids( [contract.contract_id, contract2.contract_id] ) == [1, 2] def test_get_account_ids_by_contract_ids_not_found(create_mock_account) -> None: """Test getting account_ids by contract_ids when account_id is not found.""" contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=contract) assert logic.get_account_ids_by_contract_ids([contract.contract_id, 99999999]) == [ 1 ] def test_get_can_be_deleted_records_by_contract_ids() -> None: """Flat {contract_id, can_be_deleted} records against the real test DB. A clean contract can be deleted; a contract with accounting activity (an advance) cannot. """ clean_contract = ContractFactory.create() contract_with_activity = ContractFactory.create() ContractAdvanceFactory.create(contract=contract_with_activity) res = logic.get_can_be_deleted_records_by_contract_ids( [clean_contract.contract_id, contract_with_activity.contract_id] ) by_id = {record['contract_id']: record['can_be_deleted'] for record in res} assert by_id == { clean_contract.contract_id: True, contract_with_activity.contract_id: False, } def _insert_abacus_event() -> int: """Insert a minimal abacus_event row and return its id (no factory exists).""" result = db.session.execute( text( 'INSERT INTO abacus_event ' '(event_name, target_type, target_id, event_date, created_by) ' "VALUES ('confirm_advance_payment', 'contract', 1, '2022-09-13', 'test')" ) ) return result.lastrowid def _seed_ledger_account_contract(contract: Contract) -> None: """One ledger_account_contract row for the contract (UNION branch 1).""" account = AccountFactory.create() db.session.execute( text( 'INSERT INTO ledger_account_contract ' '(abacus_event_id, account_id, contract_id, currency_code, ' 'currency_amount, previous_balance, current_balance, ' 'created_by, created_at, last_modified_by, last_modified) ' 'VALUES (:event_id, :account_id, :contract_id, :cc, 10.00, 0.00, 10.00, ' "'test', CURRENT_TIMESTAMP, 'test', CURRENT_TIMESTAMP)" ), { 'event_id': _insert_abacus_event(), 'account_id': account.account_id, 'contract_id': contract.contract_id, 'cc': 'USD', }, ) def _seed_contract_advance(contract: Contract) -> None: """One contract_advance row for the contract (UNION branch 2).""" ContractAdvanceFactory.create(contract=contract) def _seed_worksheet_adjustment(contract: Contract) -> None: """One worksheet_adjustment row for the contract (UNION branch 3).""" account = AccountFactory.create() statement_period = StatementPeriodFactory.create() # Let the DB autoincrement the PK: the factory's default fixed-id sequence # collides with committed adjustment-file rows from other test modules (these # tests commit; there is no per-test rollback). adjustment_file = StatementPeriodAdjustmentFileFactory.create( statement_period=statement_period, statement_period_adjustment_file_id=None, ) db.session.execute( text( 'INSERT INTO worksheet_adjustment ' '(statement_period_adjustment_file_id, abacus_event_id, account_id, ' 'contract_id, activity_statement_period_id, apply_to_statement_period_id, ' 'reference_adjustment_type_id, adjustment_amount, adjustment_currency_code, ' 'created_by, created_at, last_modified_by, last_modified) ' 'VALUES (:file_id, :event_id, :account_id, :contract_id, :sp_id, :sp_id, ' "1, 4.00, 'USD', 'test', CURRENT_TIMESTAMP, 'test', CURRENT_TIMESTAMP)" ), { 'file_id': adjustment_file.statement_period_adjustment_file_id, 'event_id': _insert_abacus_event(), 'account_id': account.account_id, 'contract_id': contract.contract_id, 'sp_id': statement_period.statement_period_id, }, ) def _seed_accounting_run_balance(contract: Contract) -> None: """Seed a larb + accounting_run + run_controller_contract set (branch 4). The 4th UNION branch matches only when a ledger_accounting_run_balance row joins through accounting_run to a run_controller_contract for the same contract, so all three rows must be seeded together. """ run_controller = RunControllerFactory.create() RunControllerContractFactory.create( run_controller=run_controller, contract=contract ) accounting_run = AccountingRunFactory.create(run_controller=run_controller) db.session.execute( text( 'INSERT INTO ledger_accounting_run_balance ' '(accounting_run_id, abacus_event_id, contract_id, currency_code, ' 'total_gross_revenue_amount, total_net_revenue_amount, adjusted_net_revenue, ' 'created_by, created_at, last_modified_by, last_modified) ' 'VALUES (:run_id, :event_id, :contract_id, :cc, 2173.90, 2131.91, 2131.91, ' "'test', CURRENT_TIMESTAMP, 'test', CURRENT_TIMESTAMP)" ), { 'run_id': accounting_run.accounting_run_id, 'event_id': _insert_abacus_event(), 'contract_id': contract.contract_id, 'cc': 'USD', }, ) _ACTIVITY_SEEDS = { 'ledger_account_contract': _seed_ledger_account_contract, 'contract_advance': _seed_contract_advance, 'worksheet_adjustment': _seed_worksheet_adjustment, 'accounting_run_balance': _seed_accounting_run_balance, } @pytest.mark.parametrize('branch', list(_ACTIVITY_SEEDS)) def test_get_can_be_deleted_records_by_contract_ids_each_branch(branch: str) -> None: """Each UNION activity source independently forces can_be_deleted=False. A clean contract can be deleted; a contract with exactly one activity row of this branch's type cannot. """ clean_contract = ContractFactory.create() contract_with_activity = ContractFactory.create() _ACTIVITY_SEEDS[branch](contract_with_activity) res = logic.get_can_be_deleted_records_by_contract_ids( [clean_contract.contract_id, contract_with_activity.contract_id] ) by_id = {record['contract_id']: record['can_be_deleted'] for record in res} assert by_id == { clean_contract.contract_id: True, contract_with_activity.contract_id: False, } def test_get_can_be_deleted_records_by_contract_ids_mixed_batch() -> None: """A single batch with clean and each-branch-active ids returns the correct map.""" clean_a = ContractFactory.create() clean_b = ContractFactory.create() ledger_active = ContractFactory.create() advance_active = ContractFactory.create() adjustment_active = ContractFactory.create() run_active = ContractFactory.create() _seed_ledger_account_contract(ledger_active) _seed_contract_advance(advance_active) _seed_worksheet_adjustment(adjustment_active) _seed_accounting_run_balance(run_active) contract_ids = [ clean_a.contract_id, clean_b.contract_id, ledger_active.contract_id, advance_active.contract_id, adjustment_active.contract_id, run_active.contract_id, ] res = logic.get_can_be_deleted_records_by_contract_ids(contract_ids) by_id = {record['contract_id']: record['can_be_deleted'] for record in res} assert by_id == { clean_a.contract_id: True, clean_b.contract_id: True, ledger_active.contract_id: False, advance_active.contract_id: False, adjustment_active.contract_id: False, run_active.contract_id: False, } @patch('abacus_contract.logic.contract.models') def test_get_can_be_deleted_records_by_contract_ids_empty_list_skips_query( mock_models, ) -> None: """No authorized ids: returns an empty list without querying. The dataloader helper only ever passes authorized ids through. """ res = logic.get_can_be_deleted_records_by_contract_ids([]) assert res == [] mock_models.Contract.can_be_deleted_by_ids.assert_not_called() @patch('abacus_contract.logic.contract.ows_abacus_account') def test__account_exists_success(mock_ows_abacus_account): """Test _account_exists when account exists.""" account_id = 123 mock_ows_abacus_account.get_account.return_value.status_code = 200 assert logic._account_exists(account_id) mock_ows_abacus_account.get_account.assert_called_once_with(account_id) @patch('abacus_contract.logic.contract.ows_abacus_account') def test__account_exists_when_negative_passed(mock_ows_abacus_account): """Test _account_exists when negative is passed.""" account_id = -1 mock_ows_abacus_account.get_account.return_value.status_code = 404 assert not logic._account_exists(account_id) mock_ows_abacus_account.get_account.assert_called_once_with(account_id) @patch('abacus_contract.logic.contract.ows_abacus_account') def test__account_exists_not_found(mock_ows_abacus_account): """Test _account_exists when account does not exist.""" account_id = 123 mock_ows_abacus_account.get_account.return_value.status_code = 404 assert not logic._account_exists(account_id) mock_ows_abacus_account.get_account.assert_called_once_with(account_id) @patch('abacus_contract.logic.contract.models') def test_sap_details(mock_models): """Test sap_details function.""" mock_contract = ContractFactory.create() mock_models.Contract.get_sap_profit_center_by_contract_id.return_value = { 'AccountId': 1, 'ContractId': 1, 'ContractName': 'Test contract', 'ContractType': 'distribution', 'DateTo': '2022-01-01T00:00:00.000000', 'DateFrm': '2022-01-01T00:00:00.000000', 'Bukrs': '123434', 'Prctr': 'USK1234', 'BusUnit': None, 'Zzfield1': None, 'Zzfield2': None, 'Zzfield3': None, } res = logic.sap_details(mock_contract.contract_id) assert res.status == 200 mock_models.Contract.get_sap_profit_center_by_contract_id.assert_called_once_with( contract_id=mock_contract.contract_id ) @patch( 'abacus_contract.models.contract.is_single_supply_chain_company_codes_enabled', return_value=False, ) def test_sap_details_returns_pc_via_legacy_se_join_when_ff_off(_, create_mock_account): """End-to-end: sap_details resolves PC through SE→PC join under FF OFF.""" contract = ContractFactory.create() AccountContractFactory.create(contract=contract) res = logic.sap_details(contract.contract_id) assert res.status == 200 assert ( res.message['Prctr'] == contract.reference_signing_entity.reference_sap_profit_center.profit_center ) @patch( 'abacus_contract.models.contract.is_single_supply_chain_company_codes_enabled', return_value=True, ) def test_sap_details_returns_pc_via_contract_fk_when_ff_on(_, create_mock_account): """End-to-end: sap_details resolves PC directly from contract.PC under FF ON.""" contract = ContractFactory.create() AccountContractFactory.create(contract=contract) res = logic.sap_details(contract.contract_id) assert res.status == 200 # Factory's LazyAttribute keeps contract.PC == SE's legacy PC for backfilled rows, # so the FF-ON direct-FK lookup returns the same profit_center string. assert ( res.message['Prctr'] == contract.reference_signing_entity.reference_sap_profit_center.profit_center ) @patch('abacus_contract.logic.contract.models') def test_can_contract_be_deleted(mock_models): """Test checking if a contract can be deleted.""" contract_id = 1 mock_result = True mock_models.Contract.can_be_deleted.return_value = mock_result result = logic.can_contract_be_deleted(contract_id) mock_models.Contract.can_be_deleted.assert_called_once_with(contract_id) assert result.status == 200 assert result.message['can_be_deleted'] == mock_result @patch('abacus_contract.logic.contract.models') def test_delete_contract(mock_models): """Test deleting a contract.""" contract_id = 1 mock_models.Contract.can_be_deleted.return_value = True result = logic.delete_contract(contract_id) mock_models.Contract.can_be_deleted.assert_called_once_with(contract_id) mock_models.Contract.delete.assert_called_once_with(contract_id) assert result.status == 200 assert result.message['deleted'] is True @patch('abacus_contract.logic.contract.models') def test_delete_contract_cannot_be_deleted(mock_models): """Test deleting a contract when it cannot be deleted.""" contract_id = 1 mock_models.Contract.can_be_deleted.return_value = False with pytest.raises(ValidationError): logic.delete_contract(contract_id) mock_models.Contract.can_be_deleted.assert_called_once_with(contract_id) mock_models.Contract.delete.assert_not_called() @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test__can_mark_contract_as_primary_ff_disabled(mock_ff_enabled): """Test _can_mark_contract_as_primary when ff is disabled.""" mock_ff_enabled.return_value = False assert ( logic._can_mark_contract_as_primary(1, CONTRACT_TYPES.DISTRIBUTION, True) is False ) @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test__can_mark_contract_as_primary_nr_type(mock_ff_enabled): """Test _can_mark_contract_as_primary when contract_type is NR.""" mock_ff_enabled.return_value = True with pytest.raises(Exception) as exc_info: logic._can_mark_contract_as_primary(1, CONTRACT_TYPES.NEIGHBOURING_RIGHTS, True) assert str(exc_info.value) == error.ERROR_ONLY_DISTRIBUTION_PRIMARY_CONTRACT @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test__can_contract_be_created_as_primary_when_already_exists( mock_ff_enabled, mock_models ): """Test _can_mark_contract_as_primary when primary contract already exists.""" account_id = 1 contract_type = CONTRACT_TYPES.DISTRIBUTION contract = ContractFactory.create(is_primary_contract=True) mock_ff_enabled.return_value = True mock_models.Contract.get_primary_contract_by_account_id_and_contract_type.return_value = contract with pytest.raises(Exception) as exc_info: logic._can_mark_contract_as_primary(account_id, contract_type, True) assert str(exc_info.value) == error.ERROR_PRIMARY_CONTRACT_ALREADY_EXISTS.format( account_id, contract_type ) @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test__can_contract_be_updated_as_primary_when_already_exists( mock_ff_enabled, mock_models ): """Test contract can be updated as primary when other primary contract already exists.""" account_id = 1 contract_type = CONTRACT_TYPES.DISTRIBUTION contract = ContractFactory.create(is_primary_contract=True) mock_ff_enabled.return_value = True mock_models.Contract.get_primary_contract_by_account_id_and_contract_type.return_value = contract with pytest.raises(Exception) as exc_info: logic._can_mark_contract_as_primary(account_id, contract_type, True, 101) assert str(exc_info.value) == error.ERROR_PRIMARY_CONTRACT_ALREADY_EXISTS.format( account_id, contract_type ) @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test__can_contract_be_updated_as_primary_when_no_account_contracts_exists( mock_ff_enabled, mock_models ): """Test contract can updated as primary when no other primary contract exists.""" account_id = 1 contract_type = CONTRACT_TYPES.DISTRIBUTION mock_ff_enabled.return_value = True mock_models.Contract.get_primary_contract_by_account_id_and_contract_type.return_value = None response = logic._can_mark_contract_as_primary(account_id, contract_type, True, 1) assert response is True @patch('abacus_contract.logic.contract.models') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test__can_contract_be_updated_as_primary_when_primary_contract_false( mock_ff_enabled, mock_models ): """Test _can_mark_contract_as_primary for existing contract when is_primary_field passed as false.""" # noqa: E501 account_id = 1 contract_type = CONTRACT_TYPES.DISTRIBUTION mock_ff_enabled.return_value = True mock_models.Contract.get_primary_contract_by_account_id_and_contract_type.return_value = None response = logic._can_mark_contract_as_primary(account_id, contract_type, False, 1) assert response is False @patch('abacus_contract.logic.contract.royalties_models') @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') def test__update_same_run_controller(mock_ff_enabled, mock_royalties_models) -> None: """Test _can_update_run_controller function to update contract with same run controller.""" mock_run_controller = RunControllerFactory.build( contract_type=CONTRACT_TYPES.DISTRIBUTION ) result = logic._can_update_run_controller( CONTRACT_TYPES.DISTRIBUTION, mock_run_controller.run_controller_id, mock_run_controller.run_controller_id, ) assert result is None mock_ff_enabled.assert_not_called() mock_royalties_models.StatementPeriod.get_current_statement_period.assert_not_called() mock_royalties_models.AccountingPeriod.get_current_period.assert_not_called() mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.assert_not_called() @patch('abacus_contract.logic.contract.royalties_models') @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') def test__update_run_controller_error(mock_ff_enabled, mock_royalties_models) -> None: """Test throws error for _can_update_run_controller function when contract and run controller type is different.""" mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.NEIGHBOURING_RIGHTS ) with pytest.raises(ValidationError) as exc_info: logic._can_update_run_controller( CONTRACT_TYPES.DISTRIBUTION, mock_run_controller_1.run_controller_id, mock_run_controller_2.run_controller_id, ) assert str( exc_info.value ) == error.ERROR_CONTRACT_AND_RUN_CONTROLLER_TYPE_MISMATCH.format( contract_type=CONTRACT_TYPES.DISTRIBUTION, ) mock_ff_enabled.assert_not_called() mock_royalties_models.StatementPeriod.get_current_statement_period.assert_not_called() mock_royalties_models.AccountingPeriod.get_current_period.assert_not_called() mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.assert_not_called() @patch('abacus_contract.logic.contract.royalties_models') @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') def test__can_update_run_controller_ff_disabled(mock_ff_enabled, mock_royalties_models): """Test _can_update_run_controller function when ff is disabled.""" mock_run_controller_1 = RunControllerFactory.build( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.build( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_ff_enabled.return_value = False result = logic._can_update_run_controller( CONTRACT_TYPES.DISTRIBUTION, mock_run_controller_1.run_controller_id, mock_run_controller_2.run_controller_id, ) assert result is None mock_royalties_models.StatementPeriod.get_current_statement_period.assert_not_called() mock_royalties_models.AccountingPeriod.get_current_period.assert_not_called() mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.assert_not_called() @patch('abacus_contract.logic.contract.royalties_models') @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') def test__can_update_run_controller_no_current_statement_period( mock_ff_enabled, mock_royalties_models ): """Test _can_update_run_controller function when there is no current statement period.""" mock_ff_enabled.return_value = True mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_royalties_models.RunController.get_by_id_or_error.side_effect = ( mock_run_controller_1, mock_run_controller_2, ) mock_royalties_models.StatementPeriod.get_current_statement_period.return_value = ( None ) result = logic._can_update_run_controller( CONTRACT_TYPES.DISTRIBUTION, mock_run_controller_1.run_controller_id, mock_run_controller_2.run_controller_id, ) assert result is None mock_royalties_models.StatementPeriod.get_current_statement_period.assert_called_once() mock_royalties_models.AccountingPeriod.get_current_period.assert_not_called() mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.assert_not_called() @patch('abacus_contract.logic.contract.royalties_models') @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') def test__can_update_run_controller_no_open_accounting_period( mock_ff_enabled, mock_royalties_models ): """Test _can_update_run_controller function when there is no open accounting period.""" mock_statement_period = StatementPeriodFactory.build() mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_ff_enabled.return_value = True mock_royalties_models.RunController.get_by_id_or_error.side_effect = ( mock_run_controller_1, mock_run_controller_2, ) mock_royalties_models.StatementPeriod.get_current_statement_period.return_value = ( mock_statement_period ) mock_royalties_models.AccountingPeriod.get_current_period.return_value = None result = logic._can_update_run_controller( CONTRACT_TYPES.DISTRIBUTION, mock_run_controller_1.run_controller_id, mock_run_controller_2.run_controller_id, ) assert result is None mock_royalties_models.StatementPeriod.get_current_statement_period.assert_called_once() mock_royalties_models.AccountingPeriod.get_current_period.assert_called_once() mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.assert_not_called() @patch('abacus_contract.logic.contract.royalties_models') @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') def test__can_update_run_controller_accounting_run_init_state( mock_ff_enabled, mock_royalties_models ): """Test _can_update_run_controller function when accounting run is not created.""" mock_statement_period = StatementPeriodFactory.build() mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_accounting_period = AccountingPeriodFactory.build( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN ) mock_accounting_run = AccountingRunFactory.build( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, ) mock_ff_enabled.return_value = True mock_royalties_models.RunController.get_by_id_or_error.side_effect = ( mock_run_controller_1, mock_run_controller_2, ) mock_royalties_models.StatementPeriod.get_current_statement_period.return_value = ( mock_statement_period ) mock_royalties_models.AccountingPeriod.get_current_period.return_value = ( mock_accounting_period ) mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.return_value = [ mock_accounting_run ] result = logic._can_update_run_controller( CONTRACT_TYPES.DISTRIBUTION, mock_run_controller_1.run_controller_id, mock_run_controller_2.run_controller_id, ) assert result is None mock_royalties_models.StatementPeriod.get_current_statement_period.assert_called_once() mock_royalties_models.AccountingPeriod.get_current_period.assert_called_once() assert ( mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.call_count == 2 ) @patch('abacus_contract.logic.contract.royalties_models') @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') def test__can_update_run_controller_assigned_accounting_run_committed_state( mock_ff_enabled, mock_royalties_models ): """Test _can_update_run_controller function when assigned accounting run is in committed state.""" mock_statement_period = StatementPeriodFactory.build() mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_accounting_period = AccountingPeriodFactory.build( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN ) mock_accounting_run_1 = AccountingRunFactory.build( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.COMMITTED, ) mock_accounting_run_2 = AccountingRunFactory.build( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, ) mock_ff_enabled.return_value = True mock_royalties_models.RunController.get_by_id_or_error.side_effect = ( mock_run_controller_1, mock_run_controller_2, ) mock_royalties_models.StatementPeriod.get_current_statement_period.return_value = ( mock_statement_period ) mock_royalties_models.AccountingPeriod.get_current_period.return_value = ( mock_accounting_period ) mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.side_effect = ( [mock_accounting_run_1], [mock_accounting_run_2], ) with pytest.raises(ValidationError) as exc_info: logic._can_update_run_controller( CONTRACT_TYPES.DISTRIBUTION, mock_run_controller_1.run_controller_id, mock_run_controller_2.run_controller_id, ) assert str(exc_info.value) == error.ERROR_CAN_NOT_UPDATE_RUN_CONTROLLER.format( mock_run_controller_1.run_controller_name ) mock_royalties_models.StatementPeriod.get_current_statement_period.assert_called_once() mock_royalties_models.AccountingPeriod.get_current_period.assert_called_once() assert ( mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.call_count == 2 ) @patch('abacus_contract.logic.contract.royalties_models') @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') def test__can_update_run_controller_another_accounting_run_committed_state( mock_ff_enabled, mock_royalties_models ): """Test _can_update_run_controller function when another accounting run is in committed state.""" mock_statement_period = StatementPeriodFactory.create() mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_accounting_period = AccountingPeriodFactory.create( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN ) mock_accounting_run_1 = AccountingRunFactory.build( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, run_controller=mock_run_controller_1, ) mock_accounting_run_2 = AccountingRunFactory.build( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.COMMITTED, run_controller=mock_run_controller_2, ) mock_ff_enabled.return_value = True mock_royalties_models.RunController.get_by_id_or_error.side_effect = ( mock_run_controller_1, mock_run_controller_2, ) mock_royalties_models.StatementPeriod.get_current_statement_period.return_value = ( mock_statement_period ) mock_royalties_models.AccountingPeriod.get_current_period.return_value = ( mock_accounting_period ) mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.side_effect = ( [mock_accounting_run_1], [mock_accounting_run_2], ) with pytest.raises(ValidationError) as exc_info: logic._can_update_run_controller( CONTRACT_TYPES.DISTRIBUTION, mock_run_controller_1.run_controller_id, mock_run_controller_2.run_controller_id, ) assert str(exc_info.value) == error.ERROR_CAN_NOT_UPDATE_RUN_CONTROLLER.format( mock_run_controller_2.run_controller_name ) mock_royalties_models.StatementPeriod.get_current_statement_period.assert_called_once() mock_royalties_models.AccountingPeriod.get_current_period.assert_called_once() assert ( mock_royalties_models.AccountingRun.get_by_accounting_period_and_run_controller.call_count == 2 )