"""Tests for contract logic.""" import datetime from typing import Any from unittest.mock import patch import httpx from marshmallow import ValidationError from owsresponse import response import pytest import sqlalchemy from abacus_contract.constants import constants, error from abacus_contract.constants.constants import ( CONTRACT_KAFKA_EVENT_NAMES, CONTRACT_TYPES ) from abacus_contract.constants.constants import DEFAULT_CONTRACT_EXCLUSIONS from abacus_contract.logic import contract as logic from abacus_contract.schemas.contract import ContractDetailSchema from tests.utils.factories import AccountContractFactory from tests.utils.factories import ContractFactory from tests.utils.factories import ContractLifecycleFactory from tests.utils.factories import ContractLifecycleScheduleFactory from tests.utils.factories import LegacyContractFactory from tests.utils.factories import ReferenceSigningEntityFactory @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, term_start=None, # to be deprecated term_end=None # to be deprecated ) 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, term_start=None, # to be deprecated term_end=None # to be deprecated ) 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, term_start=None, # to be deprecated term_end=None # to be deprecated ) 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, term_start=None, # to be deprecated term_end=None # to be deprecated ) 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, term_start=None, # to be deprecated term_end=None # to be deprecated ) 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 print(res.message) 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, '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, 'sap_created_at': None, 'summary_note': contracts[0].summary_note, 'general_note': contracts[0].general_note }}, {'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, '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, 'sap_created_at': None, 'summary_note': contracts[1].summary_note, 'general_note': contracts[1].general_note }}, {'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, '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, 'sap_created_at': None, 'summary_note': contract.summary_note, 'general_note': contract.general_note }] 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, '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, 'sap_created_at': None, 'summary_note': acc_contract.contract.summary_note, 'general_note': acc_contract.contract.general_note }]} 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.emit_contract_event') @patch('abacus_contract.logic.contract.models') def test_update_contract( mock_models, mock_emit_contract_event ): """Test update_contract method.""" mock_contract = ContractFactory.create() mock_models.Contract.commit_changes.return_value = True put_data = { 'contract_name': 'Test Contract Name' } result = logic.update_contract(mock_contract, **put_data) assert result.status == 200 assert result.message['contract_name'] == 'Test Contract Name' 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') @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, term_start=None, # to be deprecated term_end=None # to be deprecated ) 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 ) mock_models.ReferenceSigningEntity \ .get_signing_entity_by_sap_profit_center_id.assert_not_called() mock_models.ReferenceSapProfitCenter \ .get_sap_profit_center_by_signing_entity_id.assert_not_called() @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 ): """Test update_contract method to update reference_signing_entity_id.""" reference_signing_entity = ReferenceSigningEntityFactory.create( company_code='4926' ) mock_contract = ContractFactory.create() mock_models.Contract.commit_changes.return_value = True 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 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 ) mock_models.ReferenceSigningEntity \ .get_signing_entity_by_sap_profit_center_id.assert_not_called() @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 ): """Test update_contract method to update execution_date.""" mock_contract = ContractFactory.create() 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 ): """Test update_contract method to update initial_start_date.""" mock_contract = ContractFactory.create() 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_create_contract_with_execution_date( mock_models, mock_emit_contract_event ): """Test create contract when execution_date is present.""" mock_contract = ContractFactory.create(term_start=None) 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, '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, term_start=None, # to be deprecated term_end=None # to be deprecated ) 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 ): """Test create contract when is_excluded_from_accounting_run is set to True.""" mock_contract = ContractFactory.create(term_start=None) 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, '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, term_start=None, # to be deprecated term_end=None # to be deprecated ) 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, summary_note=mock_post_request_payload['summary_note'], general_note=mock_post_request_payload['general_note'], initial_start_date=initial_start_date ) 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, summary_note=mock_post_request_payload['summary_note'], general_note=mock_post_request_payload['general_note'], initial_start_date=initial_start_date ) 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): res = logic._create_contract( mock_post_request_payload, initial_start_date) assert res.status == 500 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'] # noqa: E501 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 = \ sqlalchemy.exc.SQLAlchemyError('Contract creating failed') lifecycle_term_start = \ mock_contract_and_lifecycle_post_payload['contract_lifecycle']['lifecycle_term_start'] # noqa: E501 with pytest.raises(Exception): res = logic.create_contract_with_lifecycle_and_schedules( **mock_contract_and_lifecycle_post_payload ) assert res.status == 500 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'] # noqa: E501 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 = \ sqlalchemy.exc.SQLAlchemyError(creation_failed_msg) lifecycle_term_start = \ mock_contract_and_lifecycle_post_payload['contract_lifecycle']['lifecycle_term_start'] # noqa: E501 with pytest.raises(Exception): res = logic.create_contract_with_lifecycle_and_schedules( **mock_contract_and_lifecycle_post_payload ) assert res.status == 500 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 = \ sqlalchemy.exc.SQLAlchemyError(creation_failed_msg) lifecycle_term_start = \ mock_contract_and_lifecycle_post_payload['contract_lifecycle']['lifecycle_term_start'] # noqa: E501 with pytest.raises(Exception): res = logic.create_contract_with_lifecycle_and_schedules( **mock_contract_and_lifecycle_post_payload ) assert res.status == 500 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_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] @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_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()