"""Functional tests for contract endpoints.""" from datetime import date from unittest.mock import patch from abacus_common_logic.utils.dates import parse_date import pytest from abacus_contract.constants import constants, error from abacus_contract.constants.constants import \ CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES from abacus_contract.constants.constants import \ CONTRACT_LIFECYCLE_STATUSES from abacus_contract.constants.constants import CONTRACT_TYPES from abacus_contract.constants.constants import DEFAULT_CONTRACT_EXCLUSIONS from abacus_contract.constants.error import \ ERROR_COLLECTION_PERIOD_DETAIL_REQUIRED from abacus_contract.constants.error import \ ERROR_CONTRACT_LIFECYCLE_NOT_FOUND from abacus_contract.constants.error import \ ERROR_CONTRACT_LIFECYCLE_SCHEDULES_LIST_EMPTY from abacus_contract.constants.error import \ ERROR_CONTRACT_LIFECYCLE_SCHEDULES_NOT_FOUND from abacus_contract.constants.error import \ ERROR_CONTRACT_LIFECYCLE_START_AND_END_DATE_NOT_VALID from abacus_contract.constants.error import \ ERROR_CONTRACT_REACTIVATION_INVALID_STATUS from abacus_contract.constants.error import ERROR_INVALID_CONTRACT_STATUS from abacus_contract.constants.error import \ ERROR_INVALID_CONTRACT_TYPE from abacus_contract.constants.error import \ ERROR_MORE_THAN_TWO_CONTRACT_LIFECYCLE_SCHEDULES_NOT_ALLOWED from abacus_contract.constants.error import \ ERROR_SCHEDULE_END_SHOULD_BE_IN_FUTURE from abacus_contract.constants.error import \ 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 tests.utils.factories import AccountContractFactory from tests.utils.factories import ContractFactory from tests.utils.factories import ContractLifecycleFactory from tests.utils.factories import ContractLifecycleScheduleDetailFactory from tests.utils.factories import ContractLifecycleScheduleFactory from tests.utils.factories import LegacyContractFactory from tests.utils.factories import ReferenceSapProfitCenterFactory from tests.utils.factories import ReferenceSigningEntityFactory from 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.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, '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 } 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, 'initial_start_date': str(contracts[0].initial_start_date), 'oa_contract_id': None, 'term_end': str(contracts[0].term_end), 'term_start': str(contracts[0].term_start), 'reference_signing_entity_id': contracts[0].reference_signing_entity_id, 'sap_created_at': None, 'summary_note': contracts[0].summary_note, 'general_note': contracts[0].general_note }}, {'data': { 'contract_id': contracts[1].contract_id, 'contract_name': contracts[1].contract_name, 'contract_type': contracts[1].contract_type, 'execution_date': contracts[1].execution_date, 'is_excluded_from_accounting_run': contracts[1].is_excluded_from_accounting_run, 'initial_start_date': str(contracts[1].initial_start_date), 'oa_contract_id': None, 'term_end': str(contracts[1].term_end), 'term_start': str(contracts[1].term_start), 'reference_signing_entity_id': contracts[1].reference_signing_entity_id, 'sap_created_at': None, 'summary_note': contracts[1].summary_note, 'general_note': contracts[1].general_note }}, {'data': None} ] 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 @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') def test_terminate_contract(mock_emit_contract_event, 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) # noqa: E501 contract = Contract.get_by_id(contract.contract_id) assert res.status_code == 200 assert contract.contract_lifecycle.lifecycle_status == \ constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') def test_terminate_contract_error(mock_emit_contract_event, 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) # noqa: E501 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() 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 == { '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, '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 } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_summary_note(mock_emit_contract_event, fixture_client): """PUT /contract/:id.""" new_contract = ContractFactory.create() 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 == { '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, '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 } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_execution_date(mock_emit_contract_event, fixture_client): """Test updating contract's execution_date.""" mock_contract = ContractFactory.create() 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 == { '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, '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'] } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_initial_start_date(mock_emit_contract_event, fixture_client): """Test updating contract's initial_start_date.""" mock_contract = ContractFactory.create() 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 == { '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, '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 } @patch('abacus_contract.logic.contract.emit_contract_event') def test_no_contract_summary_note(mock_emit_contract_event, fixture_client): """PUT /contract/:id.""" new_contract = ContractFactory.create(summary_note=None) 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 == { '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, '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 } @patch('abacus_contract.logic.contract.emit_contract_event') def test_empty_contract_summary_note(mock_emit_contract_event, fixture_client): """PUT /contract/:id.""" new_contract = ContractFactory.create(summary_note=None) 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 == { '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, '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 } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_general_note(mock_emit_contract_event, fixture_client): """PUT /contract/:id.""" new_contract = ContractFactory.create() 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 == { '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, '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 } @patch('abacus_contract.logic.contract.emit_contract_event') def test_no_contract_general_note(mock_emit_contract_event, fixture_client): """PUT /contract/:id.""" new_contract = ContractFactory.create(general_note=None) 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 == { '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, '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 } @patch('abacus_contract.logic.contract.emit_contract_event') def test_empty_contract_general_note(mock_emit_contract_event, fixture_client): """PUT /contract/:id.""" new_contract = ContractFactory.create() 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 == { '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, '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 } @patch('abacus_contract.logic.contract.emit_contract_event') def test_update_contract_is_excluded_from_accounting_run( mock_emit_contract_event, fixture_client ): """PUT /contract/:id.""" new_contract = ContractFactory.create() 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 == { '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, '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 } 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']}) # noqa:E501 @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, '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 } 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, '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 } 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, '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 } 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, '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 } 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.']} } } @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_lifecycle_and_schedule_with_past_schedule_end( 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 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 @patch('abacus_contract.logic.contract.emit_contract_event') def test_create_contract_with_renew_after_certain_date_subsequent_schedule( 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 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 @patch('abacus_contract.logic.contract.emit_contract_event') def test_nr_create_contract_with_lifecycle_and_schedules_error( mock_emit_contract_event, 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 @patch('abacus_contract.logic.contract.emit_contract_event') def test_nr_create_contract_with_lifecycle_with_same_start_and_end_date( mock_emit_contract_event, 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 @patch('abacus_contract.logic.contract.emit_contract_event') def test_reactivate_contract_lifecycle_does_not_exist( mock_emit_contract_event, 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 ) @patch('abacus_contract.logic.contract.emit_contract_event') def test_reactivate_contract_active_lifecycle_status( mock_emit_contract_event, 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, '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, } 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, '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, } 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, '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, } 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, '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, } 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}.""" # noqa: 501 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. """ # noqa:E501 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. """ # noqa:E501 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' # noqa: E501 ) 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') # noqa:E501 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' # noqa:E501 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' # noqa:E501 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='Skip failing test') 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