"""Functional tests for contract endpoints.""" from datetime import date from unittest.mock import patch import pytest from abacus_common_logic.connectors.database import db from abacus_common_logic.utils.dates import parse_date from abacus_contract.constants import constants, error from abacus_contract.constants.constants import ( ADVANCE_STATUSES, CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES, CONTRACT_LIFECYCLE_STATUSES, CONTRACT_TYPES, DEFAULT_CONTRACT_EXCLUSIONS, ) from abacus_contract.constants.error import ( ERROR_COLLECTION_PERIOD_DETAIL_REQUIRED, ERROR_CONTRACT_LIFECYCLE_NOT_FOUND, ERROR_CONTRACT_LIFECYCLE_SCHEDULES_LIST_EMPTY, ERROR_CONTRACT_LIFECYCLE_SCHEDULES_NOT_FOUND, ERROR_CONTRACT_LIFECYCLE_START_AND_END_DATE_NOT_VALID, ERROR_CONTRACT_REACTIVATION_INVALID_STATUS, ERROR_INVALID_CONTRACT_STATUS, ERROR_INVALID_CONTRACT_TYPE, ERROR_MORE_THAN_TWO_CONTRACT_LIFECYCLE_SCHEDULES_NOT_ALLOWED, ERROR_ONLY_DISTRIBUTION_PRIMARY_CONTRACT, ERROR_SCHEDULE_END_SHOULD_BE_IN_FUTURE, ERROR_SUBSEQUENT_RENEW_AFTER_CERTAIN_DATE_CAN_NOT_CREATED, ) from abacus_contract.models.contract import Contract from abacus_contract.models.contract_lifecycle import ContractLifecycle from abacus_contract.models.contract_lifecycle_schedule import ContractLifecycleSchedule from abacus_contract.tests.utils.factories import ( AccountContractFactory, ContractAdvanceFactory, ContractFactory, ContractLifecycleFactory, ContractLifecycleScheduleDetailFactory, ContractLifecycleScheduleFactory, LegacyContractFactory, ReferenceSapProfitCenterFactory, ReferenceSigningEntityFactory, SigningEntitySapProfitCenterFactory, ) from royalties.constants.constants import ( ACCOUNTING_PERIOD_STATUSES, ACCOUNTING_RUN_STATUSES, STATEMENT_PERIOD_STATUSES, ) from royalties.tests.utils.factories import ( AccountingPeriodFactory, AccountingRunFactory, RunControllerContractFactory, RunControllerFactory, StatementPeriodFactory, ) # from royalties.tests.utils.factories import RunControllerContractFactory continuously_active_renewal_type = ( CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.CONTINUOUSLY_ACTIVE ) renew_after_certain_date_renewal_type = ( CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_AFTER_CERTAIN_DATE ) renew_periodically_renewal_type = ( CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_a_contract( mock_ows_abacus_account, mock_emit_contract_kafka_event, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test new contract creation.""" contract_post_data = { 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.DISTRIBUTION, 'account_id': 1, 'reference_signing_entity_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', } mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 201 contracts = Contract.query.all() assert len(contracts) == 1 created = contracts[0] assert created.contract_name == contract_post_data['contract_name'] assert created.contract_type == contract_post_data['contract_type'] assert created.sap_created_at is None assert created.reference_signing_entity_id == 1 assert created.is_primary_contract is False assert created.contract_exclusion.exclusions == DEFAULT_CONTRACT_EXCLUSIONS assert created.summary_note == contract_post_data['summary_note'] assert created.general_note == contract_post_data['general_note'] assert created.execution_date is None mock_ows_abacus_account.get_account.assert_called_once_with(1) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_a_contract_with_a_nonexistent_contract_id( mock_ows_abacus_account, mock_emit_contract_kafka_event, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test new contract creation with a nonexistent contract id.""" contract_post_data = { 'contract_id': 123, 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.DISTRIBUTION, 'reference_signing_entity_id': 1, 'account_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', } mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 201 contracts = Contract.query.all() assert len(contracts) == 1 created = contracts[0] assert created.contract_id == contract_post_data['contract_id'] assert created.contract_name == contract_post_data['contract_name'] assert created.contract_type == contract_post_data['contract_type'] assert created.sap_created_at is None assert created.reference_signing_entity_id == 1 assert created.contract_exclusion.exclusions == DEFAULT_CONTRACT_EXCLUSIONS assert created.summary_note == contract_post_data['summary_note'] assert created.general_note == contract_post_data['general_note'] assert created.execution_date is None mock_ows_abacus_account.get_account.assert_called_once_with(1) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_a_contract_with_an_existing_contract_id( mock_ows_abacus_account, mock_emit_contract_kafka_event, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test new contract creation with an existing contract id.""" mock_contract = ContractFactory.create() contract_post_data = { 'contract_id': mock_contract.contract_id, 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.DISTRIBUTION, 'reference_signing_entity_id': 1, 'account_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', } mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) msg = error.ERROR_CONTRACT_ID_ALREADY_EXISTS.format( contract_id=mock_contract.contract_id ) assert response.status_code == 400 assert response.json['message'] == msg mock_ows_abacus_account.get_account.assert_not_called() @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_a_contract_with_execution_date( mock_ows_abacus_account, mock_emit_contract_kafka_event, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test new contract creation with execution date field.""" contract_post_data = { 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.DISTRIBUTION, 'account_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'execution_date': '2019-07-31', 'reference_signing_entity_id': 1, } mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 201 contracts = Contract.query.all() assert len(contracts) == 1 created = contracts[0] assert created.contract_name == contract_post_data['contract_name'] assert created.contract_type == contract_post_data['contract_type'] assert created.sap_created_at is None assert created.reference_signing_entity_id == 1 assert created.contract_exclusion.exclusions == DEFAULT_CONTRACT_EXCLUSIONS assert created.summary_note == contract_post_data['summary_note'] assert created.general_note == contract_post_data['general_note'] assert created.execution_date == parse_date(contract_post_data['execution_date']) mock_ows_abacus_account.get_account.assert_called_once_with(1) def test_get_contracts(mock_contracts, fixture_client): """GET /contracts.""" res = fixture_client.get('/contracts') assert res.status_code == 200 assert len(res.json['items']) == 12 contracts = res.json['items'] assert ( all( [ contract['contract_name'] in [ 'Distribution Contract 1', 'Distribution Contract 2', 'Distribution Contract 3', 'A1 LaFlare\\Amigo Records, LLC', 'Test Distribution Contract', 'NR Contract 1', 'NR Contract 2', 'NR Contract 3', 'Dylan Bukov 100% t/a Dybbukk', 'Test NR Contract', 'Contract 8675309', 'Contract sort order', ] for contract in contracts ] ) is True ) def test_get_contract_by_id( fixture_client, create_mock_account, create_mock_account_payment_term ): """Get /contract/:id.""" new_contract = ContractFactory.create(contract_name='Contract 1') legacy_contract = LegacyContractFactory.create(contract=new_contract) res = fixture_client.get(f'/contract/{new_contract.contract_id}') assert res.status_code == 200 assert res.json == { 'term_start': str(new_contract.term_start), 'contract_id': new_contract.contract_id, 'term_end': str(new_contract.term_end), 'contract_name': new_contract.contract_name, 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': new_contract.summary_note, 'general_note': new_contract.general_note, 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': new_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } def test_get_contracts_by_ids(fixture_client): """POST /contracts/.""" contracts = ContractFactory.create_batch(5) contract_ids = [c.contract_id for c in contracts] res = fixture_client.post('/contracts/', json=contract_ids) assert res.status_code == 200 assert len(res.json) == len(contracts) assert all(contract.get('contract_id') in contract_ids for contract in res.json) def test_get_contracts_by_ids_dataloaded(fixture_client): """POST /contracts/dataloader.""" contracts = ContractFactory.create_batch(2) contract_ids = [c.contract_id for c in contracts] contract_ids.append(9999) res = fixture_client.post('/contracts/dataloader', json=contract_ids) assert res.status_code == 200 assert res.json == [ { '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}, ] def test_get_contracts_by_account(fixture_client, create_mock_account): """GET /contracts/accounts/.""" contracts = ContractFactory.create_batch(3, contract_type='distribution') for idx, contract in enumerate(contracts): account_id = 2 if idx > 0 else 1 AccountContractFactory.create(contract=contract, account_id=account_id) res = fixture_client.get('/contracts/account/2') assert res.status_code == 200 assert len(res.json.get('items')) == 2 def test_get_contracts_by_account_ids(fixture_client, create_mock_account): """POST /contracts/accounts.""" contracts = ContractFactory.create_batch(3, contract_type='distribution') for idx, contract in enumerate(contracts): account_id = 2 if idx > 0 else 1 AccountContractFactory.create(contract=contract, account_id=account_id) account_ids = [1, 2] res = fixture_client.post('/contracts/accounts', json=account_ids) assert res.status_code == 200 assert len(res.json.get('items')) == 3 assert all(c.get('contract_id') for c in res.json.get('items')) assert all(c.get('account_id') for c in res.json.get('items')) def test_get_contracts_by_account_ids_error(fixture_client): """POST /contracts/accounts error.""" account_ids = [1, 2, 'test'] res = fixture_client.post('/contracts/accounts', json=account_ids) assert res.status_code == 400 def test_get_contract_by_account_dataloaded(fixture_client, create_mock_account): """POST /contracts/account/dataloader.""" contracts = ContractFactory.create_batch(3, contract_type='distribution') for idx, contract in enumerate(contracts): account_id = 2 if idx > 0 else 1 AccountContractFactory.create(contract=contract, account_id=account_id) account_ids = [1, 2] res = fixture_client.post('/contracts/account/dataloader', json=account_ids) json = res.json assert res.status_code == 200 assert len(json) == 2 assert all(item.get('data')[0].get('contract_id') for item in json) assert all(item.get('data')[0].get('account_id') for item in json) def test_find_contracts_by_name(mock_contracts, fixture_client): """GET /contracts?contract_name={contract_name}.""" res1 = fixture_client.get('/contracts?contract_name=A1 LaFlare') assert res1.status_code == 200 assert len(res1.json['items']) == 1 res1 = fixture_client.get('/contracts?contract_name=test') assert res1.status_code == 200 assert len(res1.json['items']) == 2 def test_terminate_contract(fixture_client): """PUT /contract//terminate.""" contractLifecycle = ContractLifecycleFactory.create() contract = contractLifecycle.contract json_body = { 'termination_effective': '2024-08-26', 'termination_notice_received': None, } res = fixture_client.put( f'/contract/{contract.contract_id}/terminate', json=json_body ) contract = Contract.get_by_id(contract.contract_id) assert res.status_code == 200 assert ( contract.contract_lifecycle.lifecycle_status == constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) def test_terminate_contract_error(fixture_client): """PUT /contract//terminate.""" 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.INACTIVE, ) json_body = { 'termination_effective': '2024-08-26', 'termination_notice_received': None, } res = fixture_client.put( f'/contract/{contract.contract_id}/terminate', json=json_body ) contract = Contract.get_by_id(contract.contract_id) assert res.status_code == 400 assert ( contract.contract_lifecycle.lifecycle_status == constants.CONTRACT_LIFECYCLE_STATUSES.INACTIVE ) @pytest.mark.skip(reason='This endpoint is deprecated; skip failing tests') def test_get_contracts_vat_info( fixture_client, create_mock_account, create_mock_account_tax_info ): """POST /contracts/vat-info.""" contract_ids = [1, 2] for contract_id in contract_ids: contract = ContractFactory.create(contract_id=contract_id) AccountContractFactory.create(contract=contract, account_id=contract_id) res = fixture_client.post('/contracts/vat-info', json=contract_ids) result = res.json assert res.status_code == 200 assert result == [ { 'account_id': 1, 'contract_id': 1, 'country_of_tax_residence': 'GBR', 'account_is_sba_signed': True, 'client_tax_rate': '20.00', 'supplier_tax_rate': '20.00', }, { 'account_id': 2, 'contract_id': 2, 'country_of_tax_residence': 'GBR', 'account_is_sba_signed': False, 'client_tax_rate': '20.00', 'supplier_tax_rate': None, }, ] def test_get_contract_by_id_sap( fixture_client, create_mock_account, create_mock_account_payment_term ): """Get /contract/:id/sap.""" reference_sap_profit_center = ReferenceSapProfitCenterFactory.create( company_code='4444' ) reference_signing_entity = ReferenceSigningEntityFactory.create( reference_sap_profit_center=reference_sap_profit_center ) contract = ContractFactory.create( term_start='2022-01-01', term_end='2022-01-01', reference_signing_entity=reference_signing_entity, ) account_contract = AccountContractFactory.create(contract=contract) res = fixture_client.get(f'/contract/{contract.contract_id}/sap') assert res.status_code == 200 assert res.json == { 'AccountId': str(account_contract.account_id), 'ContractId': str(contract.contract_id), 'ContractName': contract.contract_name, 'ContractType': contract.contract_type, 'DateTo': '2022-01-01T00:00:00.000000', 'DateFrm': '2022-01-01T00:00:00.000000', 'Bukrs': reference_sap_profit_center.company_code, 'Prctr': reference_sap_profit_center.profit_center, 'BusUnit': None, 'Zzfield1': None, 'Zzfield2': None, 'Zzfield3': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract(mock_emit_contract_event, fixture_client, create_mock_account): """PUT /contract/:id.""" new_contract = ContractFactory.create() AccountContractFactory.create(contract=new_contract) legacy_contract = LegacyContractFactory.create(contract=new_contract) json_body = { 'sap_created_at': '2022-01-01T01:01:02.000000', 'term_start': '2022-08-10', 'term_end': '2023-08-10', 'contract_name': 'Test Contract', } res = fixture_client.put(f'/contract/{new_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': new_contract.account_contract.account_id, 'term_start': json_body['term_start'], 'contract_id': new_contract.contract_id, 'term_end': json_body['term_end'], 'contract_name': json_body['contract_name'], 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': '2022-01-01T01:01:02.000000', 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': new_contract.summary_note, 'general_note': new_contract.general_note, 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': new_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_summary_note( mock_emit_contract_event, fixture_client, create_mock_account ): """PUT /contract/:id.""" new_contract = ContractFactory.create() AccountContractFactory.create(contract=new_contract) legacy_contract = LegacyContractFactory.create(contract=new_contract) json_body = { 'sap_created_at': '2022-01-01T01:01:02.000000', 'term_start': '2022-08-10', 'term_end': '2023-08-10', 'contract_name': 'Test Contract', 'summary_note': 'new summary_note', } res = fixture_client.put(f'/contract/{new_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': new_contract.account_contract.account_id, 'term_start': json_body['term_start'], 'contract_id': new_contract.contract_id, 'term_end': json_body['term_end'], 'contract_name': json_body['contract_name'], 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': '2022-01-01T01:01:02.000000', 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': 'new summary_note', 'general_note': new_contract.general_note, 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': new_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_execution_date( mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating contract's execution_date.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) legacy_contract = LegacyContractFactory.create(contract=mock_contract) json_body = {'execution_date': '2022-08-10'} res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'execution_date': json_body['execution_date'], 'is_paythrough_contract': mock_contract.is_paythrough_contract, 'is_primary_contract': mock_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_initial_start_date( mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating contract's initial_start_date.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) legacy_contract = LegacyContractFactory.create(contract=mock_contract) json_body = {'initial_start_date': '2022-08-10'} res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': mock_contract.execution_date, 'initial_start_date': json_body['initial_start_date'], 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': mock_contract.is_paythrough_contract, 'is_primary_contract': mock_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_no_contract_summary_note( mock_emit_contract_event, fixture_client, create_mock_account ): """PUT /contract/:id.""" new_contract = ContractFactory.create(summary_note=None) AccountContractFactory.create(contract=new_contract) legacy_contract = LegacyContractFactory.create(contract=new_contract) json_body = { 'sap_created_at': '2022-01-01T01:01:02.000000', 'term_start': '2022-08-10', 'term_end': '2023-08-10', 'contract_name': 'Test Contract', } res = fixture_client.put(f'/contract/{new_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': new_contract.account_contract.account_id, 'term_start': json_body['term_start'], 'contract_id': new_contract.contract_id, 'term_end': json_body['term_end'], 'contract_name': json_body['contract_name'], 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': '2022-01-01T01:01:02.000000', 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': None, 'general_note': new_contract.general_note, 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': new_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_empty_contract_summary_note( mock_emit_contract_event, fixture_client, create_mock_account ): """PUT /contract/:id.""" new_contract = ContractFactory.create(summary_note=None) AccountContractFactory.create(contract=new_contract) legacy_contract = LegacyContractFactory.create(contract=new_contract) json_body = { 'sap_created_at': '2022-01-01T01:01:02.000000', 'term_start': '2022-08-10', 'term_end': '2023-08-10', 'contract_name': 'Test Contract', 'summary_note': '', } res = fixture_client.put(f'/contract/{new_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': new_contract.account_contract.account_id, 'term_start': json_body['term_start'], 'contract_id': new_contract.contract_id, 'term_end': json_body['term_end'], 'contract_name': json_body['contract_name'], 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': '2022-01-01T01:01:02.000000', 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': '', 'general_note': new_contract.general_note, 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': new_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_general_note( mock_emit_contract_event, fixture_client, create_mock_account ): """PUT /contract/:id.""" new_contract = ContractFactory.create() AccountContractFactory.create(contract=new_contract) legacy_contract = LegacyContractFactory.create(contract=new_contract) json_body = { 'sap_created_at': '2022-01-01T01:01:02.000000', 'term_start': '2022-08-10', 'term_end': '2023-08-10', 'contract_name': 'Test Contract', 'general_note': 'new general_note', } res = fixture_client.put(f'/contract/{new_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': new_contract.account_contract.account_id, 'term_start': json_body['term_start'], 'contract_id': new_contract.contract_id, 'term_end': json_body['term_end'], 'contract_name': json_body['contract_name'], 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': '2022-01-01T01:01:02.000000', 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': new_contract.summary_note, 'general_note': 'new general_note', 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': new_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_no_contract_general_note( mock_emit_contract_event, fixture_client, create_mock_account ): """PUT /contract/:id.""" new_contract = ContractFactory.create(general_note=None) AccountContractFactory.create(contract=new_contract) legacy_contract = LegacyContractFactory.create(contract=new_contract) json_body = { 'sap_created_at': '2022-01-01T01:01:02.000000', 'term_start': '2022-08-10', 'term_end': '2023-08-10', 'contract_name': 'Test Contract', } res = fixture_client.put(f'/contract/{new_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': new_contract.account_contract.account_id, 'term_start': json_body['term_start'], 'contract_id': new_contract.contract_id, 'term_end': json_body['term_end'], 'contract_name': json_body['contract_name'], 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': '2022-01-01T01:01:02.000000', 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': new_contract.summary_note, 'general_note': None, 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': new_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_empty_contract_general_note( mock_emit_contract_event, fixture_client, create_mock_account ): """PUT /contract/:id.""" new_contract = ContractFactory.create() AccountContractFactory.create(contract=new_contract) legacy_contract = LegacyContractFactory.create(contract=new_contract) json_body = { 'sap_created_at': '2022-01-01T01:01:02.000000', 'term_start': '2022-08-10', 'term_end': '2023-08-10', 'contract_name': 'Test Contract', 'general_note': '', } res = fixture_client.put(f'/contract/{new_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': new_contract.account_contract.account_id, 'term_start': json_body['term_start'], 'contract_id': new_contract.contract_id, 'term_end': json_body['term_end'], 'contract_name': json_body['contract_name'], 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': '2022-01-01T01:01:02.000000', 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'summary_note': new_contract.summary_note, 'general_note': '', 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': new_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_is_excluded_from_accounting_run( mock_emit_contract_event, fixture_client, create_mock_account ): """PUT /contract/:id.""" new_contract = ContractFactory.create() AccountContractFactory.create(contract=new_contract) legacy_contract = LegacyContractFactory.create(contract=new_contract) json_body = { 'contract_name': 'Test Contract', 'is_excluded_from_accounting_run': True, } res = fixture_client.put(f'/contract/{new_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': new_contract.account_contract.account_id, 'term_start': str(new_contract.term_start), 'contract_id': new_contract.contract_id, 'term_end': str(new_contract.term_end), 'contract_name': json_body['contract_name'], 'contract_type': new_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': new_contract.sap_created_at, 'reference_signing_entity_id': new_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': new_contract.reference_sap_profit_center_id, 'summary_note': new_contract.summary_note, 'general_note': new_contract.general_note, 'execution_date': None, 'initial_start_date': str(new_contract.initial_start_date), 'is_excluded_from_accounting_run': True, 'is_paythrough_contract': new_contract.is_paythrough_contract, 'is_primary_contract': new_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_is_paythrough_contract( mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating contract's is_paythrough_contract to True.""" mock_contract = ContractFactory.create(is_paythrough_contract=0) AccountContractFactory.create(contract=mock_contract) legacy_contract = LegacyContractFactory.create(contract=mock_contract) json_body = { 'contract_name': 'Test Contract', 'is_paythrough_contract': True, } res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': json_body['contract_name'], 'contract_type': mock_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': True, 'is_primary_contract': mock_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_is_paythrough_contract_to_false( mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating contract's is_paythrough_contract to False.""" mock_contract = ContractFactory.create(is_paythrough_contract=1) AccountContractFactory.create(contract=mock_contract) legacy_contract = LegacyContractFactory.create(contract=mock_contract) json_body = { 'contract_name': mock_contract.contract_name, 'is_paythrough_contract': False, } res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': json_body['contract_name'], 'contract_type': mock_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': mock_contract.is_primary_contract, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.ows_abacus_account') @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_a_contract_ignores_is_paythrough_contract( mock_emit_contract_event, mock_ows_abacus_account, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test that is_paythrough_contract defaults to 0 during contract creation.""" contract_post_data = { 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.DISTRIBUTION, 'account_id': 1, 'reference_signing_entity_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', } mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 201 contracts = Contract.query.all() assert len(contracts) == 1 created = contracts[0] # Should default to 0 (False) since it cannot be set during creation assert created.is_paythrough_contract == 0 mock_ows_abacus_account.get_account.assert_called_once_with(1) def test_get_contracts_by_contract_type(mock_contracts, fixture_client): """GET /contracts?contract_type={contract_type}.""" res1 = fixture_client.get('/contracts?contract_type=distribution') items = res1.json['items'] assert res1.status_code == 200 assert res1.json['total_count'] == 7 assert ( all( [ contract['contract_name'] in [ 'Distribution Contract 1', 'Distribution Contract 2', 'Distribution Contract 3', 'A1 LaFlare\\Amigo Records, LLC', 'Test Distribution Contract', 'Contract 8675309', 'Contract sort order', ] for contract in items ] ) is True ) res2 = fixture_client.get('/contracts?contract_type=neighbouring_rights') items = res2.json['items'] assert res2.status_code == 200 assert res2.json['total_count'] == 5 assert ( all( [ contract['contract_name'] in [ 'NR Contract 1', 'NR Contract 2', 'NR Contract 3', 'Dylan Bukov 100% t/a Dybbukk', 'Test NR Contract', ] for contract in items ] ) is True ) def test_get_contracts_invalid_contract_type(fixture_client): """GET /contracts?contract_type={contract_type}. Throws an error for invalid contract_type. """ res = fixture_client.get('/contracts?contract_type=test') assert res.status_code == 400 assert res.json['message'] == str( { 'contract_type': [ 'Must be one of distribution, legacy_distribution, neighbouring_rights' ] } ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_a_contract_with_reference_signing_entity( mock_ows_abacus_account, mock_emit_contract_event, fixture_client, create_mock_account, ): """Test new contract creation with reference_signing_entity_id.""" reference_sap_profit_center = ReferenceSapProfitCenterFactory.create( company_code='4909' ) reference_signing_entity = ReferenceSigningEntityFactory.create( company_code='4909', reference_sap_profit_center=reference_sap_profit_center ) contract_post_data = { 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.DISTRIBUTION, 'term_end': '2019-12-31', 'term_start': '2019-06-01', 'reference_signing_entity_id': reference_signing_entity.reference_signing_entity_id, 'account_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', } mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 201 assert response.json == { 'contract_id': response.json['contract_id'], 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.DISTRIBUTION, 'term_end': '2019-12-31', 'term_start': '2019-06-01', 'reference_signing_entity_id': reference_signing_entity.reference_signing_entity_id, 'reference_sap_profit_center_id': reference_signing_entity.reference_sap_profit_center_id, 'oa_contract_id': None, 'sap_created_at': None, 'account_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'execution_date': None, 'initial_start_date': None, 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } mock_ows_abacus_account.get_account.assert_called_once_with(1) @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_lifecycle_and_schedules( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules.""" contract_lifecycle_schedules = ContractLifecycleSchedule.query.all() contract_lifecycles = ContractLifecycle.query.all() assert len(contract_lifecycle_schedules) == 0 assert len(contract_lifecycles) == 0 response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 201 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'term_end': None, 'term_start': None, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_id': 1, 'contract_name': 'Test Contract', 'oa_contract_id': 1, 'account_id': 1, 'sap_created_at': None, 'initial_start_date': '2024-07-30', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } contract_lifecycle_schedules = ContractLifecycleSchedule.query.all() contract_lifecycles = ContractLifecycle.query.all() assert len(contract_lifecycle_schedules) == 1 assert len(contract_lifecycles) == 1 assert contract_lifecycle_schedules[0].contract_id == 1 assert contract_lifecycles[0].contract_id == 1 @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_lifecycle_and_schedules_with_contract_id( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. with a nonexistent contract id. """ contract_lifecycle_schedules = ContractLifecycleSchedule.query.all() contract_lifecycles = ContractLifecycle.query.all() assert len(contract_lifecycle_schedules) == 0 assert len(contract_lifecycles) == 0 custom_contract_id = 125 mock_contract_and_lifecycle_post_payload['contract'].update( {'contract_id': custom_contract_id} ) response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 201 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'term_end': None, 'term_start': None, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_id': custom_contract_id, 'contract_name': 'Test Contract', 'oa_contract_id': 1, 'account_id': 1, 'sap_created_at': None, 'initial_start_date': '2024-07-30', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } contract_lifecycle_schedules = ContractLifecycleSchedule.query.all() contract_lifecycles = ContractLifecycle.query.all() assert len(contract_lifecycle_schedules) == 1 assert len(contract_lifecycles) == 1 assert contract_lifecycle_schedules[0].contract_id == custom_contract_id assert contract_lifecycles[0].contract_id == custom_contract_id @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_lifecycle_and_schedules_with_existing_contract_id( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. with an existing contract id. """ contract_lifecycle_schedules = ContractLifecycleSchedule.query.all() contract_lifecycles = ContractLifecycle.query.all() mock_contract = ContractFactory.create() assert len(contract_lifecycle_schedules) == 0 assert len(contract_lifecycles) == 0 custom_contract_id = mock_contract.contract_id mock_contract_and_lifecycle_post_payload['contract'].update( {'contract_id': custom_contract_id} ) response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) msg = error.ERROR_CONTRACT_ID_ALREADY_EXISTS.format( contract_id=mock_contract.contract_id ) assert response.status_code == 400 assert response.json['message'] == msg @patch('abacus_contract.logic.contract.emit_contract_event') def test_nr_create_contract_with_lifecycle_and_schedules( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating NR contract with lifecycle and schedules.""" contract_lifecycle_schedules = ContractLifecycleSchedule.query.all() contract_lifecycles = ContractLifecycle.query.all() assert len(contract_lifecycle_schedules) == 0 assert len(contract_lifecycles) == 0 mock_contract_and_lifecycle_post_payload['contract'].update( {'contract_type': CONTRACT_TYPES.NEIGHBOURING_RIGHTS} ) mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'] = [ { 'renewal_type': renew_after_certain_date_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', 'schedule_end': '9025-09-10', }, { 'renewal_type': renew_periodically_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', 'renewal_offset_detail_interval': 1, 'renewal_offset_detail_type': 'year', 'collection_period_detail_interval': 2, 'collection_period_detail_type': 'year', 'schedule_end': None, }, ] response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 201 assert response.json == { 'contract_type': 'neighbouring_rights', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'term_end': None, 'term_start': None, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_id': 1, 'contract_name': 'Test Contract', 'oa_contract_id': 1, 'account_id': 1, 'sap_created_at': None, 'initial_start_date': '2024-07-30', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } contract_lifecycle_schedules = ContractLifecycleSchedule.query.all() contract_lifecycles = ContractLifecycle.query.all() assert len(contract_lifecycle_schedules) == 2 assert len(contract_lifecycles) == 1 assert contract_lifecycle_schedules[0].contract_id == 1 assert contract_lifecycle_schedules[1].contract_id == 1 assert contract_lifecycles[0].contract_id == 1 @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_lifecycle_and_schedules_invalid_contract_type( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. returns an error when contract_type is legacy_distribution. """ mock_contract_and_lifecycle_post_payload['contract'].update( { 'contract_type': CONTRACT_TYPES.LEGACY_DISTRIBUTION, } ) response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) msg = ERROR_INVALID_CONTRACT_TYPE.format( contract_type=CONTRACT_TYPES.LEGACY_DISTRIBUTION ) assert response.status_code == 400 assert response.json['message'] == msg @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_lifecycle_and_schedules_invalid_account( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. throws an error when non-existent account_id is passed. """ mock_contract_and_lifecycle_post_payload['contract'].update({'account_id': 9999}) response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 500 @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_lifecycle_and_three_schedules( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. returns an error when contract_lifecycle_schedules POST request field is passed with more than 2 records . """ mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'] = [ { 'renewal_type': continuously_active_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', } ] * 4 response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) msg = ERROR_MORE_THAN_TWO_CONTRACT_LIFECYCLE_SCHEDULES_NOT_ALLOWED assert response.status_code == 400 assert response.json['message'] == msg @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_without_contract_lifecycle_field( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. returns an validation error when contract_lifecycle is missing in the POST request payload. """ del mock_contract_and_lifecycle_post_payload['contract_lifecycle'] response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 400 assert response.json['message'] == {'contract_lifecycle': ['Must be specified.']} @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_empty_contract_lifecycle_schedules( mock_emit_contract_event, mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. returns an error when empty list is passed to the contract_lifecycle_schedules field """ mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'] = [] response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 400 assert ( response.json['message']['_schema'][0] == ERROR_CONTRACT_LIFECYCLE_SCHEDULES_LIST_EMPTY ) @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_lifecycle_and_schedules_schema_validation_error( mock_emit_contract_event, fixture_client, ): """Test creating contract with lifecycle and schedules. returns an error when required fields are not passed """ mock_post_request_payload = { 'contract': { 'account_id': None, 'contract_name': 'Test Contract', 'contract_type': 'distribution', 'reference_signing_entity_id': 1, 'execution_date': None, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', }, 'contract_lifecycle': {'lifecycle_term_start': None}, 'contract_lifecycle_schedules': [ { 'renewal_type': continuously_active_renewal_type, 'termination_notice_detail_interval': None, 'termination_notice_detail_type': 'month', } ], } response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_post_request_payload, ) assert response.status_code == 400 assert response.json['message'] == { 'contract': {'account_id': ['Field may not be null.']}, 'contract_lifecycle': {'lifecycle_term_start': ['Field may not be null.']}, 'contract_lifecycle_schedules': { '0': {'termination_notice_detail_interval': ['Field may not be null.']} }, } def test_create_contract_with_lifecycle_and_schedule_with_past_schedule_end( mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. returns an error when contract_lifecycle_schedules POST request payload is having two records and first record is of renewal type "renew_after_certain_date" and schedule_end date is past date. """ mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'] = [ { 'renewal_type': renew_after_certain_date_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', 'schedule_end': '2024-06-01', }, { 'renewal_type': continuously_active_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', }, ] response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 400 assert response.json['message'] == ERROR_SCHEDULE_END_SHOULD_BE_IN_FUTURE def test_create_contract_with_renew_after_certain_date_subsequent_schedule( mock_contract_and_lifecycle_post_payload, fixture_client ): """Test creating contract with lifecycle and schedules. returns an error when contract_lifecycle_schedules POST request payload is having a second record with type "renew_after_certain_date". """ mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'] = [ { 'renewal_type': renew_after_certain_date_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', 'schedule_end': '2025-09-10', }, { 'renewal_type': renew_after_certain_date_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', 'schedule_end': '2025-09-10', }, ] response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) msg = ERROR_SUBSEQUENT_RENEW_AFTER_CERTAIN_DATE_CAN_NOT_CREATED.format( renew_after_certain_date_renewal_type ) assert response.status_code == 400 assert response.json['message'] == msg def test_nr_create_contract_with_lifecycle_and_schedules_error( mock_contract_and_lifecycle_post_payload, fixture_client ): """Test NR creating contract with lifecycle and schedules.. returns an error when collection_period_detail_type is not passed. """ mock_contract_and_lifecycle_post_payload['contract'].update( {'contract_type': CONTRACT_TYPES.NEIGHBOURING_RIGHTS} ) mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'] = [ { 'renewal_type': continuously_active_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': 6, } ] response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 400 assert response.json['message'] == ERROR_COLLECTION_PERIOD_DETAIL_REQUIRED def test_nr_create_contract_with_lifecycle_with_same_start_and_end_date( mock_contract_and_lifecycle_post_payload, fixture_client ): """Test NR creating contract with lifecycle and schedules. returns an error when lifecycle_term_start is same as lifecycle_term_end. """ schedule_end = lifecycle_term_start = '2050-07-30' mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'] = [ { 'renewal_type': renew_after_certain_date_renewal_type, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day', 'schedule_end': schedule_end, }, { 'renewal_type': continuously_active_renewal_type, 'termination_notice_detail_interval': 1, 'termination_notice_detail_type': 'month', }, ] mock_contract_and_lifecycle_post_payload['contract_lifecycle'] = { 'lifecycle_term_start': lifecycle_term_start } response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) msg = ERROR_CONTRACT_LIFECYCLE_START_AND_END_DATE_NOT_VALID.format( lifecycle_term_start, schedule_end ) assert response.status_code == 400 assert response.json['message'] == msg def test_reactivate_contract_lifecycle_does_not_exist(fixture_client): """Test to reactivate a contract. returns an error if contract_lifecycle does not exist for a contract. """ mock_contract = ContractFactory.create() response = fixture_client.put(f'/contract/{mock_contract.contract_id}/reactivate') assert response.status_code == 400 assert response.json['message'] == ERROR_CONTRACT_LIFECYCLE_NOT_FOUND.format( mock_contract.contract_id ) def test_reactivate_contract_active_lifecycle_status(fixture_client): """Test to reactivate a contract. returns an error if the status of contract_lifecycle is ACTIVE. """ mock_contract = ContractFactory.create() ContractLifecycleFactory.create( contract=mock_contract, lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.ACTIVE ) response = fixture_client.put(f'/contract/{mock_contract.contract_id}/reactivate') assert response.status_code == 400 assert response.json['message'] == ERROR_CONTRACT_REACTIVATION_INVALID_STATUS def test_reactivate_contract_lifecycle_schedule_does_not_exist(fixture_client): """Test to reactivate a contract. returns an error if contract_lifecycle_schedule does not exist for a contract. """ mock_contract = ContractFactory.create() ContractLifecycleFactory.create( contract=mock_contract, lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) response = fixture_client.put(f'/contract/{mock_contract.contract_id}/reactivate') assert response.status_code == 400 assert response.json['message'] == ERROR_CONTRACT_LIFECYCLE_SCHEDULES_NOT_FOUND def test_reactivate_contract_for_future_schedule_end(fixture_client): """Test to reactivate a contract. when current schedule is of renewal type renew_after_certain_date and schedule_end is future date. """ mock_contract = ContractFactory.create() mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=renew_after_certain_date_renewal_type, schedule_end='2090-08-10', ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=renew_periodically_renewal_type ), ] mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED, ) mock_schedule_ids = [ schedule.contract_lifecycle_schedule_id for schedule in mock_contract_lifecycle_schedules ] response = fixture_client.put(f'/contract/{mock_contract.contract_id}/reactivate') assert response.status_code == 200 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'term_end': '2029-06-01', 'term_start': '2019-06-01', 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_id': mock_contract.contract_id, 'contract_name': mock_contract.contract_name, 'oa_contract_id': None, 'sap_created_at': None, 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } contract_lifecycle = ContractLifecycle.get_by_contract_id(mock_contract.contract_id) contract_lifecycle_schedules = ( ContractLifecycleSchedule.get_contract_lifecycle_schedules_by_contract_id( mock_contract.contract_id ) ) assert ( contract_lifecycle.contract_lifecycle_id != mock_contract_lifecycle.contract_lifecycle_id ) assert all( [ schedule.contract_lifecycle_schedule_id in mock_schedule_ids for schedule in contract_lifecycle_schedules ] ) def test_reactivate_contract_for_past_schedule_end(fixture_client): """Test to reactivate a contract. when current schedule is of renewal type renew_after_certain_date and schedule_end is past date. """ mock_contract = ContractFactory.create() mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=renew_after_certain_date_renewal_type, schedule_end='2023-08-10', ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=renew_periodically_renewal_type ), ] mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED, ) response = fixture_client.put(f'/contract/{mock_contract.contract_id}/reactivate') assert response.status_code == 200 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'term_end': '2029-06-01', 'term_start': '2019-06-01', 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_id': mock_contract.contract_id, 'contract_name': mock_contract.contract_name, 'oa_contract_id': None, 'sap_created_at': None, 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } contract_lifecycle = ContractLifecycle.get_by_contract_id(mock_contract.contract_id) contract_lifecycle_schedules = ( ContractLifecycleSchedule.get_contract_lifecycle_schedules_by_contract_id( mock_contract.contract_id ) ) assert ( contract_lifecycle.contract_lifecycle_id != mock_contract_lifecycle.contract_lifecycle_id ) assert len(contract_lifecycle_schedules) == 1 assert ( contract_lifecycle_schedules[0].renewal_type == continuously_active_renewal_type ) def test_reactivate_contract_for_renew_periodically_schedule(fixture_client): """Test to reactivate a contract. when current schedule is of renewal type renew_periodically. """ mock_contract = ContractFactory.create() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=10, period_type='year' ) mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=renew_periodically_renewal_type, renewal_offset_detail=mock_schedule_detail, ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=renew_periodically_renewal_type, renewal_offset_detail=mock_schedule_detail, ), ] mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED, ) mock_schedule_ids = [ schedule.contract_lifecycle_schedule_id for schedule in mock_contract_lifecycle_schedules ] response = fixture_client.put(f'/contract/{mock_contract.contract_id}/reactivate') assert response.status_code == 200 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'term_end': '2029-06-01', 'term_start': '2019-06-01', 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_id': mock_contract.contract_id, 'contract_name': mock_contract.contract_name, 'oa_contract_id': None, 'sap_created_at': None, 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } contract_lifecycle = ContractLifecycle.get_by_contract_id(mock_contract.contract_id) contract_lifecycle_schedules = ( ContractLifecycleSchedule.get_contract_lifecycle_schedules_by_contract_id( mock_contract.contract_id ) ) assert ( contract_lifecycle.contract_lifecycle_id != mock_contract_lifecycle.contract_lifecycle_id ) assert all( [ schedule.contract_lifecycle_schedule_id in mock_schedule_ids for schedule in contract_lifecycle_schedules ] ) def test_reactivate_contract_for_continuously_active_schedule(fixture_client): """Test to reactivate a contract. when current schedule is of renewal type continuously_active. """ mock_contract = ContractFactory.create() mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=continuously_active_renewal_type ) ] mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED, ) response = fixture_client.put(f'/contract/{mock_contract.contract_id}/reactivate') assert response.status_code == 200 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'term_end': '2029-06-01', 'term_start': '2019-06-01', 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_id': mock_contract.contract_id, 'contract_name': mock_contract.contract_name, 'oa_contract_id': None, 'sap_created_at': None, 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } contract_lifecycle = ContractLifecycle.get_by_contract_id(mock_contract.contract_id) contract_lifecycle_schedules = ( ContractLifecycleSchedule.get_contract_lifecycle_schedules_by_contract_id( mock_contract.contract_id ) ) assert ( contract_lifecycle.contract_lifecycle_id != mock_contract_lifecycle.contract_lifecycle_id ) assert len(contract_lifecycle_schedules) == 1 assert ( contract_lifecycle_schedules[0].contract_lifecycle_schedule_id == mock_contract_lifecycle_schedules[0].contract_lifecycle_schedule_id ) def test_snapshot_contracts( create_mock_account, create_mock_account_payment_term, fixture_client ): """Test GET /contracts/snapshot.""" contracts = ContractFactory.create_batch(3) excluded_contract = ContractFactory.create(is_excluded_from_accounting_run=True) contract_ids = list() for contract in contracts: AccountContractFactory.create(contract=contract) contract_ids.append(contract.contract_id) init_contract_lifecycle = ContractLifecycleFactory.create( contract=contracts[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.INIT ) active_contract_lifecycle = ContractLifecycleFactory.create( contract=contracts[1], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.ACTIVE, lifecycle_term_end=date(2050, 1, 1), ) terminated_contract_lifecycle = ContractLifecycleFactory.create( contract=contracts[2], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED, lifecycle_term_start=date(2020, 1, 1), lifecycle_term_end=date(2022, 1, 1), ) res = fixture_client.post('/contracts/snapshot', json=contract_ids) data_without_header = res.data.decode().strip().split('\n')[1:] assert res.status_code == 200 assert res.mimetype == 'text/tsv' assert init_contract_lifecycle.contract_id not in ( row.split('/t')[0] for row in data_without_header ) assert excluded_contract.contract_id not in ( row.split('/t')[0] for row in data_without_header ) assert len(data_without_header) == 2 assert active_contract_lifecycle.lifecycle_term_start == parse_date( data_without_header[0].split('\t')[2] ) assert active_contract_lifecycle.lifecycle_term_end == parse_date( data_without_header[0].split('\t')[3] ) assert terminated_contract_lifecycle.lifecycle_term_start == parse_date( data_without_header[1].split('\t')[2] ) assert terminated_contract_lifecycle.lifecycle_term_end == parse_date( data_without_header[1].split('\t')[3] ) assert contracts[1].term_start != active_contract_lifecycle.lifecycle_term_start assert contracts[1].term_end != active_contract_lifecycle.lifecycle_term_end assert contracts[2].term_start != terminated_contract_lifecycle.lifecycle_term_start assert contracts[2].term_end != terminated_contract_lifecycle.lifecycle_term_end def test_get_contracts_by_included_excluded_run(mock_contracts, fixture_client): """GET /contracts?is_excluded_from_accounting_run={is_excluded_from_accounting_run}.""" res1 = fixture_client.get('/contracts?is_excluded_from_accounting_run=0') items = res1.json['items'] assert res1.status_code == 200 assert res1.json['total_count'] == 6 assert ( all( [ contract['contract_name'] in [ 'Distribution Contract 1', 'Distribution Contract 2', 'Distribution Contract 3', 'Dylan Bukov 100% t/a Dybbukk', 'Contract 8675309', 'Contract sort order', ] for contract in items ] ) is True ) res2 = fixture_client.get('/contracts?is_excluded_from_accounting_run=1') items = res2.json['items'] assert res2.status_code == 200 assert res2.json['total_count'] == 6 assert ( all( [ contract['contract_name'] in [ 'A1 LaFlare\\Amigo Records, LLC', 'Test Distribution Contract', 'NR Contract 1', 'NR Contract 2', 'NR Contract 3', 'Test NR Contract', ] for contract in items ] ) is True ) res3 = fixture_client.get( '/contracts?contract_type=distribution&is_excluded_from_accounting_run=1' ) items = res3.json['items'] assert res3.status_code == 200 assert res3.json['total_count'] == 2 assert ( all( [ contract['contract_name'] in ['A1 LaFlare\\Amigo Records, LLC', 'Test Distribution Contract'] for contract in items ] ) is True ) def test_get_contracts_by_invalid_included_excluded_run(fixture_client): """GET /contracts?is_excluded_from_accounting_run={is_excluded_from_accounting_run}. Throws an error for invalid is_excluded_from_accounting_run. """ res = fixture_client.get('/contracts?is_excluded_from_accounting_run=test') assert res.status_code == 400 assert res.json['message'] == str( {'is_excluded_from_accounting_run': ['Not a valid boolean.']} ) def test_get_contracts_by_invalid_contract_statuses(fixture_client): """GET /contracts?contract_statuses={contract_statuses}. Throws an error for invalid contract_statuses. """ res = fixture_client.get('/contracts?contract_statuses=test1,active,inactive') assert res.status_code == 400 assert res.json['message'] == ERROR_INVALID_CONTRACT_STATUS.format( contract_lifecycle_status='init, active, to_be_terminated, terminated, in_collection_period' ) def test_get_terminated_and_active_contracts(mock_contracts, fixture_client): """GET /contracts?contract_statuses={contract_statuses}. Returns terminated and active contracts. """ res = fixture_client.get('/contracts?contract_statuses=terminated,active') items = res.json['items'] assert res.status_code == 200 assert res.json['total_count'] == 4 assert ( all( [ contract['contract_name'] in [ 'Distribution Contract 3', 'NR Contract 2', 'Contract 8675309', 'Contract sort order', ] for contract in items ] ) is True ) """ def test_get_contracts_filtered_by_run_controller_id( create_mock_account, create_mock_run_controller, fixture_client ): "" "GET /contracts?run_controller_ids=." "" contracts = ContractFactory.create_batch(3, contract_type='distribution') for idx, contract in enumerate(contracts): run_controller_id = 1 if idx > 0 else 2 AccountContractFactory.create(contract=contract, account_id=1) RunControllerContractFactory.create( contract=contract, run_controller_id=run_controller_id ) res = fixture_client.get('/contracts', query_string={'run_controller_ids': 1}) assert res.status_code == 200, res.json assert len(res.json.get('items')) == 2 """ def test_get_not_yet_activated_contracts(mock_contracts, fixture_client): """GET /contracts?contract_statuses={contract_statuses}. Returns contracts that are in 'init' state as well as those contracts for which the contract lifecycle schedules have not been created. """ res = fixture_client.get('/contracts?contract_statuses=init') items = res.json['items'] assert res.status_code == 200 assert res.json['total_count'] == 5 assert ( all( [ contract['contract_name'] in [ 'Distribution Contract 1', 'Dylan Bukov 100% t/a Dybbukk', 'NR Contract 3', 'Test Distribution Contract', 'Test NR Contract', ] for contract in items ] ) is True ) def test_get_to_be_terminated_and_in_collection_period_contracts( mock_contracts, fixture_client ): """GET /contracts?contract_statuses={contract_statuses}. Returns to_be_terminated and in_collection_period contracts. """ res = fixture_client.get( '/contracts?contract_statuses=to_be_terminated,in_collection_period' ) items = res.json['items'] assert res.status_code == 200 assert res.json['total_count'] == 2 assert ( all( [ contract['contract_name'] in ['A1 LaFlare\\Amigo Records, LLC', 'NR Contract 1'] for contract in items ] ) is True ) def test_get_filtered_contracts(mock_contracts, fixture_client): """Get contracts by contract types, status and account ids.""" url = '/contracts?account_ids=1&contract_type=distribution&contract_statuses=to_be_terminated,in_collection_period' res = fixture_client.get(url) items = res.json['items'] assert res.status_code == 200 assert res.json['total_count'] == 1 assert items[0]['contract_name'] == 'A1 LaFlare\\Amigo Records, LLC' def test_get_not_activated_contracts_with_other_filters(mock_contracts, fixture_client): """Get contracts those are in init state and lifecycles are not created.""" url = '/contracts?account_ids=1,2&contract_type=neighbouring_rights&contract_statuses=init' res = fixture_client.get(url) items = res.json['items'] assert res.status_code == 200 assert res.json['total_count'] == 3 assert ( all( [ contract['contract_name'] in ['Dylan Bukov 100% t/a Dybbukk', 'NR Contract 3', 'Test NR Contract'] for contract in items ] ) is True ) def test_can_contract_be_deleted(create_mock_account, fixture_client): """Test `GET /contract/:contract_id/can-be-deleted`.""" contract = ContractFactory.create() AccountContractFactory.create(contract=contract, account_id=1) res = fixture_client.get(f'/contract/{contract.contract_id}/can-be-deleted') assert res.status_code == 200 assert res.json == {'can_be_deleted': True} @pytest.mark.skip(reason='This test case works locally but is failing on Jenkins.') def test_can_not_delete_contract( mock_revenue_associated_with_accounting_run, fixture_client ): """Test `GET /contract/:contract_id/can-be-deleted` when revenue is associated with contract.""" res = fixture_client.get('/contract/9012890/can-be-deleted') assert res.status_code == 200 assert res.json == {'can_be_deleted': False} def test_accounting_run_with_no_revenue_can_be_be_deleted( create_mock_account, fixture_client ): """Test checking if a contract can be deleted if no revenue is associated with contract.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract, account_id=2) mock_run_controller = RunControllerFactory.create() RunControllerContractFactory.create( run_controller=mock_run_controller, contract=mock_contract ) AccountingRunFactory.create(run_controller=mock_run_controller) res = fixture_client.get(f'/contract/{mock_contract.contract_id}/can-be-deleted') assert res.status_code == 200 assert res.json == {'can_be_deleted': True} @pytest.mark.skip(reason='This endpoint is under development.') def test_delete_contract(fixture_client): """Test `DELETE /contract/:contract_id`.""" contract_id = ContractFactory.create().contract_id res = fixture_client.delete(f'/contract/{contract_id}') assert res.status_code == 200 assert res.json == {'deleted': True} existing_contract = Contract.get_by_id(contract_id) assert existing_contract is None @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_a_contract_when_no_other_primary_contract( mock_ows_abacus_account, mock_ff_enabled, mock_emit_contract_kafka_event, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test to create a distribution primary contract.""" contract_post_data = { 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.DISTRIBUTION, 'account_id': 2, 'reference_signing_entity_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', } mock_ff_enabled.return_value = True mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 201 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'is_primary_contract': True, 'term_end': None, 'term_start': None, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'run_controller_id': None, 'contract_id': 1, 'contract_name': 'Test', 'oa_contract_id': None, 'account_id': 2, 'sap_created_at': None, 'initial_start_date': None, 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, } @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_a_nr_primary_contract( mock_ows_abacus_account, mock_ff_enabled, mock_emit_contract_kafka_event, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test to create a NR primary contract.""" contract_post_data = { 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.NEIGHBOURING_RIGHTS, 'account_id': 2, 'reference_signing_entity_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'is_primary_contract': True, } mock_ff_enabled.return_value = True mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 400 assert response.json['message'] == ERROR_ONLY_DISTRIBUTION_PRIMARY_CONTRACT @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_a_nr_non_primary_contract( mock_ows_abacus_account, mock_ff_enabled, mock_emit_contract_kafka_event, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test to create a NR primary contract.""" contract_post_data = { 'contract_name': 'Test', 'contract_type': CONTRACT_TYPES.NEIGHBOURING_RIGHTS, 'account_id': 2, 'reference_signing_entity_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'is_primary_contract': False, } mock_ff_enabled.return_value = True mock_ows_abacus_account.get_account.return_value.status_code = 200 response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 201 assert response.json == { 'contract_type': 'neighbouring_rights', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'is_primary_contract': False, 'term_end': None, 'term_start': None, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'run_controller_id': None, 'contract_id': 1, 'contract_name': 'Test', 'oa_contract_id': None, 'account_id': 2, 'sap_created_at': None, 'initial_start_date': None, 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, } @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') @patch('abacus_contract.logic.contract.ows_abacus_account') def test_create_primary_contract_when_other_primary_already_exist( mock_ows_abacus_account, mock_ff_enabled, mock_emit_contract_kafka_event, fixture_client, create_mock_account, create_mock_reference_signing_entity, ): """Test to create a distribution primary contract when other one already exists.""" account_id = 1 contract_type = CONTRACT_TYPES.DISTRIBUTION primary_contract = ContractFactory.create(is_primary_contract=True) AccountContractFactory.create(contract=primary_contract, account_id=account_id) mock_ows_abacus_account.get_account.return_value.status_code = 200 contract_post_data = { 'contract_name': 'Test', 'contract_type': contract_type, 'account_id': 1, 'reference_signing_entity_id': 1, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'is_primary_contract': True, } mock_ff_enabled.return_value = True response = fixture_client.post('/contract', json=contract_post_data) assert response.status_code == 400 assert response.json[ 'message' ] == error.ERROR_PRIMARY_CONTRACT_ALREADY_EXISTS.format(account_id, contract_type) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_create_primary_contract_with_lifecycle_and_schedules( mock_ff_enabled, mock_emit_contract_kafka_event, mock_contract_and_lifecycle_post_payload, fixture_client, ): """Test to create a distribution primary contract with lifecycle and schedules. Creates a primary contract when no other primary contract exists for same account and contract_type. """ mock_ff_enabled.return_value = True response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 201 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'term_end': None, 'term_start': None, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'contract_id': 1, 'contract_name': 'Test Contract', 'oa_contract_id': 1, 'account_id': 1, 'sap_created_at': None, 'initial_start_date': '2024-07-30', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_create_another_primary_contract_with_lifecycle_and_schedules( mock_ff_enabled, mock_emit_contract_kafka_event, mock_contract_and_lifecycle_post_payload, fixture_client, ): """Test to create another distribution primary contract with lifecycle and schedules. Returns an error when primary contract already exists for same account and contract_type. """ # noqa: E501 account_id = 1 contract_type = CONTRACT_TYPES.DISTRIBUTION primary_contract = ContractFactory.create(is_primary_contract=True) AccountContractFactory.create(contract=primary_contract, account_id=account_id) mock_ff_enabled.return_value = True mock_contract_and_lifecycle_post_payload['contract']['is_primary_contract'] = True response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 400 assert response.json[ 'message' ] == error.ERROR_PRIMARY_CONTRACT_ALREADY_EXISTS.format(account_id, contract_type) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_create_primary_contract_with_lifecycles_all_others_terminated( mock_ff_enabled, mock_emit_contract_kafka_event, mock_contract_and_lifecycle_post_payload, fixture_client, ): """Test to create another distribution primary contract with lifecycle and schedules. Creates a primary contract when all other contracts are terminated that belongs to same account and of same contract_type. """ # noqa: E501 account_id = 1 contracts = ContractFactory.create_batch(3) primary_contract = ContractFactory.create(is_primary_contract=True) AccountContractFactory.create(contract=primary_contract, account_id=2) for contract in contracts: AccountContractFactory.create(contract=contract, account_id=account_id) ContractLifecycleFactory.create( contract=contracts[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) ContractLifecycleFactory.create( contract=contracts[1], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) ContractLifecycleFactory.create( contract=contracts[2], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) mock_ff_enabled.return_value = True mock_contract_and_lifecycle_post_payload['contract']['is_primary_contract'] = True response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 201 assert response.json == { 'contract_type': 'distribution', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'is_primary_contract': True, 'term_end': None, 'term_start': None, 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'run_controller_id': None, 'contract_id': 8, 'contract_name': 'Test Contract', 'oa_contract_id': 1, 'account_id': 1, 'sap_created_at': None, 'initial_start_date': '2024-07-30', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, } @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_create_primary_contract_with_lifecycles_other_one_is_active( mock_ff_enabled, mock_emit_contract_kafka_event, mock_contract_and_lifecycle_post_payload, fixture_client, ): """Test to create another distribution primary contract with lifecycle and schedules. Returns an error when primary contract already exists for same account and contract_type. """ # noqa: E501 account_id = 1 contract_type = CONTRACT_TYPES.DISTRIBUTION contracts = ContractFactory.create_batch(2) contracts.append(ContractFactory.create(is_primary_contract=True)) for contract in contracts: AccountContractFactory.create(contract=contract, account_id=account_id) ContractLifecycleFactory.create( contract=contracts[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) ContractLifecycleFactory.create( contract=contracts[1], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, ) ContractLifecycleFactory.create( contract=contracts[2], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.ACTIVE ) mock_ff_enabled.return_value = True mock_contract_and_lifecycle_post_payload['contract']['is_primary_contract'] = True response = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload, ) assert response.status_code == 400 assert response.json[ 'message' ] == error.ERROR_PRIMARY_CONTRACT_ALREADY_EXISTS.format(account_id, contract_type) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_contract_to_primary( mock_ff_enabled, mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating non primary contract to primary.""" mock_contract = ContractFactory.create() AccountContractFactory.create(contract=mock_contract) legacy_contract = LegacyContractFactory.create(contract=mock_contract) mock_ff_enabled.return_value = True json_body = { 'contract_name': 'Test Contract', 'is_primary_contract': True, } res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': json_body['contract_name'], 'contract_type': mock_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': True, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_primary_contract_to_non_primary( mock_ff_enabled, mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating primary contract to non primary.""" mock_contract = ContractFactory.create(is_primary_contract=True) AccountContractFactory.create(contract=mock_contract) legacy_contract = LegacyContractFactory.create(contract=mock_contract) mock_ff_enabled.return_value = True json_body = { 'contract_name': 'Test Contract', 'is_primary_contract': False, } res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': json_body['contract_name'], 'contract_type': mock_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_primary_contract_to_non_primary_ff_disabled( mock_ff_enabled, mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating primary contract to non primary when ff is disabled.""" mock_contract = ContractFactory.create(is_primary_contract=True) AccountContractFactory.create(contract=mock_contract) legacy_contract = LegacyContractFactory.create(contract=mock_contract) mock_ff_enabled.return_value = False json_body = { 'contract_name': 'Test Contract', 'is_primary_contract': False, } res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': json_body['contract_name'], 'contract_type': mock_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': True, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_non_primary_contract_to_primary_ff_disabled( mock_ff_enabled, mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating non primary contract to primary when ff is disabled.""" mock_contract = ContractFactory.create(is_primary_contract=False) AccountContractFactory.create(contract=mock_contract) legacy_contract = LegacyContractFactory.create(contract=mock_contract) mock_ff_enabled.return_value = False json_body = { 'contract_name': 'Test Contract', 'is_primary_contract': True, } res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': json_body['contract_name'], 'contract_type': mock_contract.contract_type, 'oa_contract_id': legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': None, } @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_non_primary_contract_when_already_exist( mock_ff_enabled, mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating non primary contract when other one is already primary. Returns an error when try to update the non-primary contract to primary when other contract which is already primary and belongs same account and contract_type. """ account_id = 1 mock_primary_contract = ContractFactory.create(is_primary_contract=True) AccountContractFactory.create(contract=mock_primary_contract, account_id=account_id) mock_contract = ContractFactory.create(is_primary_contract=False) AccountContractFactory.create(contract=mock_contract) mock_ff_enabled.return_value = True json_body = { 'contract_name': 'Test Contract', 'is_primary_contract': True, } response = fixture_client.put( f'/contract/{mock_contract.contract_id}', json=json_body ) assert response.status_code == 400 assert response.json[ 'message' ] == error.ERROR_PRIMARY_CONTRACT_ALREADY_EXISTS.format( account_id, CONTRACT_TYPES.DISTRIBUTION ) @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_nr_contract_to_primary( mock_ff_enabled, mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating NR contract to primary.""" mock_contract = ContractFactory.create( is_primary_contract=False, contract_type=CONTRACT_TYPES.NEIGHBOURING_RIGHTS ) AccountContractFactory.create(contract=mock_contract) mock_ff_enabled.return_value = True json_body = { 'contract_name': 'Test Contract', 'is_primary_contract': True, } response = fixture_client.put( f'/contract/{mock_contract.contract_id}', json=json_body ) assert response.status_code == 400 assert response.json['message'] == ERROR_ONLY_DISTRIBUTION_PRIMARY_CONTRACT @patch('abacus_contract.logic.contract.emit_contract_event') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_nr_contract_to_non_primary( mock_ff_enabled, mock_emit_contract_event, fixture_client, create_mock_account ): """Test updating NR contract to non primary which are already non primary.""" mock_contract = ContractFactory.create( is_primary_contract=False, contract_type=CONTRACT_TYPES.NEIGHBOURING_RIGHTS ) AccountContractFactory.create(contract=mock_contract) mock_ff_enabled.return_value = True json_body = { 'contract_name': 'Test Contract', 'is_primary_contract': False, } response = fixture_client.put( f'/contract/{mock_contract.contract_id}', json=json_body ) assert response.status_code == 200 assert { 'contract_type': 'neighbouring_rights', 'execution_date': None, 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'is_primary_contract': False, 'term_end': '2029-06-01', 'term_start': '2019-06-01', 'summary_note': 'This is for the test', 'general_note': 'This is for the test (general_note)', 'run_controller_id': None, 'contract_id': 1, 'contract_name': 'Test Contract', 'oa_contract_id': None, 'account_id': 1, 'sap_created_at': None, 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'is_paythrough_contract': False, } @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_contract_with_same_run_controller( mock_primary_ff_enabled, create_mock_account, fixture_client ): """Test updating contract with same run controller.""" mock_primary_ff_enabled.return_value = True mock_contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=mock_contract) StatementPeriodFactory.create() mock_legacy_contract = LegacyContractFactory.create(contract=mock_contract) mock_accounting_period = AccountingPeriodFactory.create( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN ) mock_run_controller = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) RunControllerContractFactory.create( run_controller=mock_run_controller, contract=mock_contract ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, run_controller=mock_run_controller, ) json_body = { 'is_primary_contract': True, 'run_controller_id': mock_run_controller.run_controller_id, } res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'oa_contract_id': mock_legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': True, 'run_controller_id': mock_run_controller.run_controller_id, } @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_contract_with_different_run_controller_type( mock_primary_ff_enabled, create_mock_account, fixture_client ): """Test updating contract with different run controller type.""" mock_primary_ff_enabled.return_value = True mock_contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=mock_contract) StatementPeriodFactory.create() LegacyContractFactory.create(contract=mock_contract) mock_accounting_period = AccountingPeriodFactory.create( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN ) mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.NEIGHBOURING_RIGHTS ) RunControllerContractFactory.create( run_controller=mock_run_controller_1, contract=mock_contract ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, run_controller=mock_run_controller_1, ) json_body = {'run_controller_id': mock_run_controller_2.run_controller_id} res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 400 assert res.json[ 'message' ] == error.ERROR_CONTRACT_AND_RUN_CONTROLLER_TYPE_MISMATCH.format( contract_type=CONTRACT_TYPES.DISTRIBUTION, ) @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_contract_prevent_run_controller_ff_disabled( mock_primary_ff_enabled, mock_run_controller_ff_enabled, create_mock_account, fixture_client, ): """Test updating contract with different run controller when ff is disabled.""" mock_primary_ff_enabled.return_value = True mock_run_controller_ff_enabled.return_value = False mock_contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=mock_contract) StatementPeriodFactory.create() mock_legacy_contract = LegacyContractFactory.create(contract=mock_contract) mock_accounting_period = AccountingPeriodFactory.create( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN ) mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) RunControllerContractFactory.create( run_controller=mock_run_controller_1, contract=mock_contract ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.COMMITTED, run_controller=mock_run_controller_1, ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, run_controller=mock_run_controller_2, ) json_body = {'run_controller_id': mock_run_controller_2.run_controller_id} res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'oa_contract_id': mock_legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': mock_run_controller_2.run_controller_id, } @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_contract_prevent_run_controller_ff_enabled( mock_primary_ff_enabled, mock_run_controller_ff_enabled, create_mock_account, fixture_client, ): """Test updating contract with different run controller when ff is enabled.""" mock_primary_ff_enabled.return_value = True mock_run_controller_ff_enabled.return_value = False mock_contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=mock_contract) StatementPeriodFactory.create() mock_legacy_contract = LegacyContractFactory.create(contract=mock_contract) mock_accounting_period = AccountingPeriodFactory.create( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN ) mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) RunControllerContractFactory.create( run_controller=mock_run_controller_1, contract=mock_contract ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, run_controller=mock_run_controller_1, ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, run_controller=mock_run_controller_2, ) json_body = {'run_controller_id': mock_run_controller_2.run_controller_id} res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'oa_contract_id': mock_legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': mock_run_controller_2.run_controller_id, } @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_contract_assigned_accounting_run_committed_state( mock_primary_ff_enabled, mock_run_controller_ff_enabled, create_mock_account, fixture_client, ): """Test updating contract with different run controller when accounting_run in committed_state.""" mock_primary_ff_enabled.return_value = True mock_run_controller_ff_enabled.return_value = True mock_contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=mock_contract) mock_statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) LegacyContractFactory.create(contract=mock_contract) mock_accounting_period = AccountingPeriodFactory.create( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN, statement_period=mock_statement_period, ) mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) RunControllerContractFactory.create( run_controller=mock_run_controller_1, contract=mock_contract ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.INVALID, run_controller=mock_run_controller_1, ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.COMMITTED, run_controller=mock_run_controller_1, ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.NO_ACTION_TAKEN, run_controller=mock_run_controller_2, ) json_body = {'run_controller_id': mock_run_controller_2.run_controller_id} res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 400 assert res.json['message'] == error.ERROR_CAN_NOT_UPDATE_RUN_CONTROLLER.format( mock_run_controller_1.run_controller_name ) @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_contract_another_accounting_run_complete_state( mock_primary_ff_enabled, mock_run_controller_ff_enabled, create_mock_account, fixture_client, ): """Test updating contract with different run controller when another accounting_run in complete state.""" mock_primary_ff_enabled.return_value = True mock_run_controller_ff_enabled.return_value = True mock_contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=mock_contract) mock_statement_period = StatementPeriodFactory.create( statement_period_status=STATEMENT_PERIOD_STATUSES.CURRENT ) LegacyContractFactory.create(contract=mock_contract) mock_accounting_period = AccountingPeriodFactory.create( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.OPEN, statement_period=mock_statement_period, ) mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) RunControllerContractFactory.create( run_controller=mock_run_controller_1, contract=mock_contract ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.INVALID, run_controller=mock_run_controller_1, ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.SKIPPED, run_controller=mock_run_controller_1, ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.COMPLETE, run_controller=mock_run_controller_2, ) json_body = {'run_controller_id': mock_run_controller_2.run_controller_id} res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 400 assert res.json['message'] == error.ERROR_CAN_NOT_UPDATE_RUN_CONTROLLER.format( mock_run_controller_2.run_controller_name ) @patch('abacus_contract.logic.contract.is_abacus_prevent_run_controller_update_enabled') @patch('abacus_contract.logic.contract.is_abacus_primary_contract_enabled') def test_update_contract_when_no_open_accounting_period( mock_primary_ff_enabled, mock_run_controller_ff_enabled, create_mock_account, fixture_client, ): """Test updating contract with different run controller when there is no open accounting period.""" mock_primary_ff_enabled.return_value = True mock_run_controller_ff_enabled.return_value = False mock_contract = ContractFactory.create() AccountContractFactory.create(account_id=1, contract=mock_contract) StatementPeriodFactory.create() mock_legacy_contract = LegacyContractFactory.create(contract=mock_contract) mock_accounting_period = AccountingPeriodFactory.create( accounting_period_status=ACCOUNTING_PERIOD_STATUSES.CLOSED, closed_date=date(2025, 2, 5), ) mock_run_controller_1 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) mock_run_controller_2 = RunControllerFactory.create( contract_type=CONTRACT_TYPES.DISTRIBUTION ) RunControllerContractFactory.create( run_controller=mock_run_controller_1, contract=mock_contract ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.COMMITTED, run_controller=mock_run_controller_1, ) AccountingRunFactory.create( accounting_period=mock_accounting_period, run_status=ACCOUNTING_RUN_STATUSES.COMMITTED, run_controller=mock_run_controller_2, ) json_body = {'run_controller_id': mock_run_controller_2.run_controller_id} res = fixture_client.put(f'/contract/{mock_contract.contract_id}', json=json_body) assert res.status_code == 200 assert res.json == { 'account_id': mock_contract.account_contract.account_id, 'term_start': str(mock_contract.term_start), 'contract_id': mock_contract.contract_id, 'term_end': str(mock_contract.term_end), 'contract_name': mock_contract.contract_name, 'contract_type': mock_contract.contract_type, 'oa_contract_id': mock_legacy_contract.oa_contract_id, 'sap_created_at': mock_contract.sap_created_at, 'reference_signing_entity_id': mock_contract.reference_signing_entity_id, 'reference_sap_profit_center_id': mock_contract.reference_sap_profit_center_id, 'summary_note': mock_contract.summary_note, 'general_note': mock_contract.general_note, 'execution_date': None, 'initial_start_date': str(mock_contract.initial_start_date), 'is_excluded_from_accounting_run': mock_contract.is_excluded_from_accounting_run, 'is_paythrough_contract': False, 'is_primary_contract': False, 'run_controller_id': mock_run_controller_2.run_controller_id, } @pytest.mark.db('mysql') @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_pc_writes_old_value_to_contract_history( mock_emit_contract_event, _, fixture_client, create_mock_account ): """The after_contract_update trigger mirrors the OLD PC to contract_history.""" contract = ContractFactory.create() AccountContractFactory.create(contract=contract) original_pc_id = contract.reference_sap_profit_center_id new_pc = ReferenceSapProfitCenterFactory.create() SigningEntitySapProfitCenterFactory.create( reference_signing_entity=contract.reference_signing_entity, reference_sap_profit_center=new_pc, ) res = fixture_client.put( f'/contract/{contract.contract_id}', json={'reference_sap_profit_center_id': new_pc.reference_sap_profit_center_id}, ) assert res.status_code == 200, res.json history_rows = db.engine.execute( f""" SELECT reference_sap_profit_center_id FROM contract_history WHERE contract_id = {contract.contract_id} ORDER BY last_modified """ ).fetchall() pc_audit = [row[0] for row in history_rows] assert original_pc_id in pc_audit, ( f'Expected old PC {original_pc_id} in contract_history audit trail, ' f'got: {pc_audit}' )