"""Unit tests for contract_lifecycle_schedule logic.""" import datetime from unittest.mock import MagicMock, patch import pytest from abacus_common_logic.constants.error import ERROR_ENTITY_DOES_NOT_EXIST from marshmallow import ValidationError from sqlalchemy.exc import SQLAlchemyError from abacus_contract.constants.constants import ( CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES, CONTRACT_LIFECYCLE_STATUSES, CONTRACT_TYPES, ) from abacus_contract.constants.error import ( ERROR_COLLECTION_PERIOD_DETAIL_NOT_REQUIRED, ERROR_COLLECTION_PERIOD_DETAIL_REQUIRED, ERROR_CONTRACT_LIFECYCLE_SCHEDULE_NOT_MATCHED, ERROR_MORE_THAN_TWO_CONTRACT_LIFECYCLE_SCHEDULES_NOT_ALLOWED, ERROR_SCHEDULE_END_REQUIRED, ERROR_SCHEDULE_END_SHOULD_BE_IN_FUTURE, ERROR_SUBSEQUENT_CONTRACT_LIFECYCLE_SCHEDULES_CAN_NOT_CREATED, ERROR_SUBSEQUENT_RENEW_AFTER_CERTAIN_DATE_CAN_NOT_CREATED, ERROR_TWO_CONTRACT_LIFECYCLE_SCHEDULES_ALLOWED, ) from abacus_contract.logic import contract_lifecycle_schedule as logic from abacus_contract.tests.utils.factories import ( ContractFactory, ContractLifecycleFactory, ContractLifecycleScheduleDetailFactory, ContractLifecycleScheduleFactory, ) @patch('abacus_contract.logic.contract_lifecycle_schedule.ContractLifecycleSchedule') def test_get_contract_lifecycle_schedules(mock_model): """Test getting a list of contract lifecycle schedules by contract_id.""" mock_contract_lifecycle_schedules = [ContractLifecycleScheduleFactory.create()] contract_id = mock_contract_lifecycle_schedules[0].contract_id mock_model.get_contract_lifecycle_schedules_by_contract_id.return_value = ( mock_contract_lifecycle_schedules ) res = logic.get_contract_lifecycle_schedules_by_contract_id(contract_id) assert res.status == 200 assert len(res.message) == len(mock_contract_lifecycle_schedules) mock_model.get_contract_lifecycle_schedules_by_contract_id.assert_called_once_with( contract_id ) @patch('abacus_contract.logic.contract_lifecycle_schedule.ContractLifecycleSchedule') def test_get_contract_lifecycle_schedules_by_ids(mock_model): """Test getting a list of contract lifecycle schedules by ids.""" mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create(), ContractLifecycleScheduleFactory.create(), ] ids = [ schedule.contract_lifecycle_schedule_id for schedule in mock_contract_lifecycle_schedules ] ids.append(12345) mock_model.get_by_ids.return_value = mock_contract_lifecycle_schedules res = logic.get_contract_lifecycle_schedules_by_ids(ids) assert res.status == 200 assert len(res.message) == len(ids) assert res.message == [ { 'data': { 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'contract_id': mock_contract_lifecycle_schedules[0].contract_id, 'termination_notice_detail_id': mock_contract_lifecycle_schedules[ 0 ].termination_notice_detail_id, 'renewal_offset_detail_id': mock_contract_lifecycle_schedules[ 0 ].renewal_offset_detail_id, 'collection_period_detail_id': mock_contract_lifecycle_schedules[ 0 ].collection_period_detail_id, 'renewal_type': mock_contract_lifecycle_schedules[0].renewal_type, 'schedule_end': mock_contract_lifecycle_schedules[0].schedule_end, 'contract_lifecycle': None, } }, { 'data': { 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[ 1 ].contract_lifecycle_schedule_id, 'contract_id': mock_contract_lifecycle_schedules[1].contract_id, 'termination_notice_detail_id': mock_contract_lifecycle_schedules[ 1 ].termination_notice_detail_id, 'renewal_offset_detail_id': mock_contract_lifecycle_schedules[ 1 ].renewal_offset_detail_id, 'collection_period_detail_id': mock_contract_lifecycle_schedules[ 1 ].collection_period_detail_id, 'renewal_type': mock_contract_lifecycle_schedules[1].renewal_type, 'schedule_end': mock_contract_lifecycle_schedules[1].schedule_end, 'contract_lifecycle': None, } }, {'data': None}, ] @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '.find_or_create_contract_lifecycle_schedule_detail' ) def test_get_detail_ids(mock_schedule_detail_logic): """Test getting ids for termination_notice_detail, renewal_offset_detail and collection_period_detail.""" mock_schedule_details = ContractLifecycleScheduleDetailFactory.create_batch(3) request_body = { 'renewal_offset_detail_interval': mock_schedule_details[0].period_interval, 'renewal_offset_detail_type': mock_schedule_details[0].period_type, 'termination_notice_detail_interval': mock_schedule_details[1].period_interval, 'termination_notice_detail_type': mock_schedule_details[1].period_type, 'collection_period_detail_interval': mock_schedule_details[2].period_interval, 'collection_period_detail_type': mock_schedule_details[2].period_type, } mock_schedule_detail_logic.side_effect = [ mock_schedule_details[1], mock_schedule_details[0], mock_schedule_details[2], ] res = logic._get_detail_ids(request_body) assert res == ( mock_schedule_details[1].contract_lifecycle_schedule_detail_id, mock_schedule_details[0].contract_lifecycle_schedule_detail_id, mock_schedule_details[2].contract_lifecycle_schedule_detail_id, ) assert mock_schedule_detail_logic.call_count == 3 @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '.find_or_create_contract_lifecycle_schedule_detail' ) def test_get_detail_ids_for_termination_notice_detail(mock_schedule_detail_logic): """Test getting ids for termination_notice_detail.""" mock_schedule_details = ContractLifecycleScheduleDetailFactory.create_batch(1) request_body = { 'renewal_offset_detail_interval': None, 'renewal_offset_detail_type': None, 'termination_notice_detail_interval': mock_schedule_details[0].period_interval, 'termination_notice_detail_type': mock_schedule_details[0].period_type, 'collection_period_detail_interval': None, 'collection_period_detail_type': None, } mock_schedule_detail_logic.return_value = mock_schedule_details[0] res = logic._get_detail_ids(request_body) assert res == ( mock_schedule_details[0].contract_lifecycle_schedule_detail_id, None, None, ) assert mock_schedule_detail_logic.call_count == 1 def test_validate_collection_period_details_for_nr(): """Test _validate_collection_period_details for NR contract.""" request_body = [ { 'collection_period_detail_interval': 1, 'collection_period_detail_type': 'month', } ] contract_type = CONTRACT_TYPES.NEIGHBOURING_RIGHTS res = logic._validate_collection_period_details(contract_type, request_body) assert res is True def test_validate_collection_period_details_error_for_nr(): """Test _validate_collection_period_details for NR contract. Throws an error if both collection_period_detail_type and collection_period_detail_interval are None or either of them is None. """ request_body = [ { 'collection_period_detail_interval': None, 'collection_period_detail_type': 'month', } ] contract_type = CONTRACT_TYPES.NEIGHBOURING_RIGHTS with pytest.raises(ValidationError) as excinfo: logic._validate_collection_period_details(contract_type, request_body) msg = ERROR_COLLECTION_PERIOD_DETAIL_REQUIRED assert excinfo.value.messages[0] == msg def test_validate_request_payload_with_more_records(): """Testing _validate_request_payload. when POST/PUT request payload has more than two records. """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', } ] * 3 msg = ERROR_MORE_THAN_TWO_CONTRACT_LIFECYCLE_SCHEDULES_NOT_ALLOWED with pytest.raises(Exception) as excinfo: logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert excinfo.value.messages[0] == msg def test_validate_request_payload_with_one_record_error(): """Testing _validate_request_payload. throws an error when POST request payload is having a record other than renewal type "continuously_active". """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_after_certain_date', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'schedule_end': '2024-07-30', } ] msg = ERROR_TWO_CONTRACT_LIFECYCLE_SCHEDULES_ALLOWED with pytest.raises(Exception) as excinfo: logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert excinfo.value.messages[0] == msg def test_validate_request_payload_with_one_record_success(): """Testing _validate_request_payload. return true when POST/PUT request payload is having a record of renewal type "continuously_active". """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', } ] res = logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert res is True def test_validate_request_payload_with_one_record_allow(): """Testing _validate_request_payload. return true when PUT request payload is having a record of renewal type "renew_periodically". """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'renew_offset_detail_interval': 1, 'renew_offset_detail_type': 'month', } ] allow_one_record_in_payload = True res = logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, allow_one_record_in_payload, ) assert res is True def test_validate_request_payload_with_one_record_allow_error(): """Testing _validate_request_payload. throws an error when PUT request payload is having a record of renewal type "renew_after_certain_date". """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_after_certain_date', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'schedule_end': '2025-09-10', } ] allow_one_record_in_payload = True msg = ERROR_SUBSEQUENT_RENEW_AFTER_CERTAIN_DATE_CAN_NOT_CREATED.format( CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_AFTER_CERTAIN_DATE ) with pytest.raises(Exception) as excinfo: logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, allow_one_record_in_payload, ) assert str(excinfo.value) == msg def test_validate_request_payload_with_first_continuously_active(): """Testing _validate_request_payload. throws an error when POST/PUT request payload is having two records and first record is of renewal type "continuously_active". """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', }, { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'renew_offset_detail_interval': 1, 'renew_offset_detail_type': 'month', }, ] msg = ERROR_SUBSEQUENT_CONTRACT_LIFECYCLE_SCHEDULES_CAN_NOT_CREATED.format( CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.CONTINUOUSLY_ACTIVE ) with pytest.raises(Exception) as excinfo: logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert str(excinfo.value) == msg def test_validate_request_with_subsequent_renew_after_certain_date(): """Testing _validate_request_payload. throws an error when POST/PUT request payload is having two records and second record is of renewal type "renew_after_certain_date". """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'renew_offset_detail_interval': 1, 'renew_offset_detail_type': 'month', }, { 'renewal_type': 'renew_after_certain_date', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'schedule_end': '2024-07-30', }, ] msg = ERROR_SUBSEQUENT_RENEW_AFTER_CERTAIN_DATE_CAN_NOT_CREATED.format( CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_AFTER_CERTAIN_DATE ) with pytest.raises(Exception) as excinfo: logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert str(excinfo.value) == msg def test_validate_request_with_first_renew_after_certain_date(): """Testing _validate_request_payload. throws an error when POST/PUT 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_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_after_certain_date', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'schedule_end': datetime.date(2019, 6, 1), }, { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'renew_offset_detail_interval': 1, 'renew_offset_detail_type': 'month', }, ] with pytest.raises(Exception) as excinfo: logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert excinfo.value.messages[0] == ERROR_SCHEDULE_END_SHOULD_BE_IN_FUTURE def test_validate_request_payload_with_none_schedule_end(): """Testing _validate_request_payload. throws an error when POST/PUT request payload is having two records and first record is of renewal type "renew_after_certain_date" and schedule_end date is None. """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_after_certain_date', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'schedule_end': None, }, { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'renew_offset_detail_interval': 1, 'renew_offset_detail_type': 'month', }, ] with pytest.raises(Exception) as excinfo: logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert excinfo.value.messages[0] == ERROR_SCHEDULE_END_REQUIRED.format( CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_AFTER_CERTAIN_DATE ) @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._validate_collection_period_details' ) def test_validate_request_payload_with_end_date_in_the_past( mock_collection_period_details_validation, ): """Testing _validate_request_payload with end date in the past.""" mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_after_certain_date', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'schedule_end': datetime.date(2000, 6, 1), }, { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'renew_offset_detail_interval': 1, 'renew_offset_detail_type': 'month', 'collection_period_detail_interval': 1, 'collection_period_detail_type': 'year', }, ] mock_collection_period_details_validation.return_value = True res = logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, allow_end_date_in_past=True, ) assert res is True @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._validate_collection_period_details' ) def test_validate_request_payload_with_collection_period_details( mock_collection_period_details_validation, ): """Testing _validate_request_payload. throws an error when POST/PUT request payload is having two records and collection_period_details provided for distribution contract. """ mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_after_certain_date', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'schedule_end': datetime.date(2100, 6, 1), }, { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'renew_offset_detail_interval': 1, 'renew_offset_detail_type': 'month', 'collection_period_detail_interval': 1, 'collection_period_detail_type': 'year', }, ] msg = ERROR_COLLECTION_PERIOD_DETAIL_NOT_REQUIRED mock_collection_period_details_validation.side_effect = Exception(msg) with pytest.raises(Exception) as excinfo: logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert str(excinfo.value) == msg @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._validate_collection_period_details' ) def test_validate_request_payload_success(mock_collection_period_details_validation): """Testing _validate_request_payload.""" mock_contract_lifecycle_schedules = [] mock_contract = ContractFactory.create() mock_request = [ { 'renewal_type': 'renew_after_certain_date', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'schedule_end': datetime.date(2100, 6, 1), }, { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 90, 'termination_notice_detail_type': 'day', 'renew_offset_detail_interval': 1, 'renew_offset_detail_type': 'month', 'collection_period_detail_interval': 1, 'collection_period_detail_type': 'year', }, ] mock_collection_period_details_validation.return_value = True res = logic._validate_request_payload( mock_contract.contract_id, mock_contract.contract_type, mock_contract_lifecycle_schedules, mock_request, ) assert res is True @patch( 'abacus_contract.logic.contract_lifecycle_schedule._generate_contract_lifecycle_params' ) @patch('abacus_contract.logic.contract_lifecycle_schedule.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle_schedule.ContractLifecycleSchedule') @patch('abacus_contract.logic.contract_lifecycle_schedule._get_detail_ids') def test__create_contract_lifecycle_schedules( mock_get_detail_ids, mock_model, mock_contract_lifecycle_model, mock__generate_contract_lifecycle_params, ): """Testing _create_contract_lifecycle_schedules.""" mock_contract = ContractFactory.create() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create() detail_id = mock_schedule_detail.contract_lifecycle_schedule_detail_id mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type='continuously_active', termination_notice_detail_id=detail_id, ) ] mock_get_detail_ids.return_value = (detail_id, None, None) mock_model.build.return_value = mock_contract_lifecycle_schedules[0] mock_post_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': mock_schedule_detail.period_interval, 'termination_notice_detail_type': mock_schedule_detail.period_type, } ] res = logic._create_contract_lifecycle_schedules( mock_contract.contract_id, mock_post_request ) assert res == mock_contract_lifecycle_schedules mock_model.build.assert_called_once() mock__generate_contract_lifecycle_params.assert_not_called() mock_contract_lifecycle_model.assert_not_called() @patch( 'abacus_contract.logic.contract_lifecycle_schedule._generate_contract_lifecycle_params' ) @patch('abacus_contract.logic.contract_lifecycle_schedule.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle_schedule.ContractLifecycleSchedule') @patch('abacus_contract.logic.contract_lifecycle_schedule._get_detail_ids') def test__create_contract_lifecycle_schedules_with_lifecycle( mock_get_detail_ids, mock_model, mock_contract_lifecycle_model, mock__generate_contract_lifecycle_params, ): """Testing _create_contract_lifecycle_schedules with lifecycle.""" mock_contract = ContractFactory.create() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create() detail_id = mock_schedule_detail.contract_lifecycle_schedule_detail_id mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type='continuously_active', termination_notice_detail_id=detail_id, ) ] mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0], ) mock_contract_lifecycle_params = { 'contract_id': mock_contract.contract_id, 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.ACTIVE, 'lifecycle_term_start': '2025-01-01', 'lifecycle_term_end': None, 'renewal_effective': None, 'termination_notice_deadline': None, } mock__generate_contract_lifecycle_params.return_value = ( mock_contract_lifecycle_params ) mock_get_detail_ids.return_value = (detail_id, None, None) mock_model.build.return_value = mock_contract_lifecycle_schedules[0] mock_contract_lifecycle_model.build.return_value = mock_contract_lifecycle mock_post_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': mock_schedule_detail.period_interval, 'termination_notice_detail_type': mock_schedule_detail.period_type, 'contract_lifecycle': {'lifecycle_term_start': '2025-01-01'}, } ] res = logic._create_contract_lifecycle_schedules( mock_contract.contract_id, mock_post_request ) assert res == mock_contract_lifecycle_schedules mock_model.build.assert_called_once() mock__generate_contract_lifecycle_params.assert_called_once_with( mock_contract.contract_id, mock_contract_lifecycle_params.get('lifecycle_term_start'), mock_contract_lifecycle_schedules[0], ) mock_contract_lifecycle_model.build.assert_called_once_with( **mock_contract_lifecycle_params ) @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._create_contract_lifecycle_schedules' ) @patch('abacus_contract.logic.contract_lifecycle_schedule._validate_request_payload') def test_create_contract_lifecycle_schedules( mock_post_request_validation, mock_create_lifecycle_schedule_logic ): """Testing create_contract_lifecycle_schedules.""" mock_contract = ContractFactory.create() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create() mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type='continuously_active', termination_notice_detail=mock_schedule_detail, ) ] mock_post_request_validation.return_value = True mock_create_lifecycle_schedule_logic.return_value = ( mock_contract_lifecycle_schedules ) mock_post_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': mock_schedule_detail.period_interval, 'termination_notice_detail_type': mock_schedule_detail.period_type, } ] res = logic.create_contract_lifecycle_schedules( mock_contract.contract_id, mock_post_request ) assert res.status == 201 assert res.message == [ { 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'contract_id': mock_contract.contract_id, 'termination_notice_detail_id': mock_schedule_detail.contract_lifecycle_schedule_detail_id, 'renewal_offset_detail_id': None, 'collection_period_detail_id': None, 'renewal_type': 'continuously_active', 'schedule_end': None, 'contract_lifecycle': None, } ] mock_create_lifecycle_schedule_logic.assert_called_once() @patch('abacus_contract.logic.contract_lifecycle_schedule._validate_request_payload') def test_create_contract_lifecycle_schedules_with_lifecycle( mock_post_request_validation, ): """Testing create_contract_lifecycle_schedules with lifecycle.""" mock_post_request_validation.return_value = True mock_contract = ContractFactory.create() mock_post_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 1, 'termination_notice_detail_type': 'DAY', 'contract_lifecycle': { 'lifecycle_term_start': datetime.datetime(2020, 1, 1).date(), }, } ] res = logic.create_contract_lifecycle_schedules( mock_contract.contract_id, mock_post_request ) assert res.status == 201 assert res.message == [ { 'contract_lifecycle_schedule_id': 1, 'contract_id': 1, 'termination_notice_detail_id': 1, 'renewal_offset_detail_id': None, 'collection_period_detail_id': None, 'renewal_type': 'continuously_active', 'schedule_end': None, 'contract_lifecycle': {'lifecycle_term_start': '2020-01-01'}, } ] def test_create_contract_lifecycle_schedules_with_lifecycle_no_start_date(): """Testing create_contract_lifecycle_schedules with lifecycle.""" mock_contract = ContractFactory.create() mock_post_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 1, 'termination_notice_detail_type': 'DAY', 'contract_lifecycle': { 'lifecycle_term_start': None, }, } ] res = logic.create_contract_lifecycle_schedules( mock_contract.contract_id, mock_post_request ) assert res.status == 400 def test_create_contract_lifecycle_schedules_with_lifecycle_invalid(): """Testing create_contract_lifecycle_schedules with lifecycle.""" mock_contract = ContractFactory.create() mock_post_request = [ { 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 1, 'termination_notice_detail_type': 'DAY', 'contract_lifecycle': { 'lifecycle_term_start': None, }, } ] res = logic.create_contract_lifecycle_schedules( mock_contract.contract_id, mock_post_request ) assert res.status == 400 def test_validate_contract_lifecycle_schedules_for_current_schedule(): """Test _validate_and_get_existing_contract_lifecycle_schedules. If the PUT request includes the current lifecycle schedule id, then the _validate_and_get_existing_contract_lifecycle_schedules function returns both the current and subsequent lifecycle schedule records. """ is_updating_current_lifecycle_schedule = True mock_contract = ContractFactory.create() mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ] mock_put_request = [ { 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_after_certain_date', 'schedule_end': '2030-08-08', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, }, { 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[ 1 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, 'renewal_offset_detail_interval': 150, 'renewal_offset_detail_type': 'day', }, ] res = logic._validate_and_get_existing_contract_lifecycle_schedules( mock_contract.contract_id, mock_put_request, is_updating_current_lifecycle_schedule, ) assert res == mock_contract_lifecycle_schedules def test_validate_contract_lifecycle_schedules_for_subsequent_schedule(): """Test _validate_and_get_existing_contract_lifecycle_schedules. If the PUT request includes only subsequent lifecycle schedule id, then the _validate_and_get_existing_contract_lifecycle_schedules function returns only the subsequent lifecycle schedule. """ is_updating_current_lifecycle_schedule = False mock_contract = ContractFactory.create() mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ] mock_put_request = [ { 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[ 1 ].contract_lifecycle_schedule_id, 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, } ] res = logic._validate_and_get_existing_contract_lifecycle_schedules( mock_contract.contract_id, mock_put_request, is_updating_current_lifecycle_schedule, ) assert res == [mock_contract_lifecycle_schedules[1]] def test_validate_contract_lifecycle_schedules_throws_deleted_error(): """Test _validate_and_get_existing_contract_lifecycle_schedules. throws an error if contract_lifecycle_schedule is deleted. """ is_updating_current_lifecycle_schedule = False mock_contract = ContractFactory.create() mock_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, deleted_at='2024-08-08', deleted_by='Test User', ), ] contract_lifecycle_schedule_id = mock_contract_lifecycle_schedules[ 1 ].contract_lifecycle_schedule_id mock_put_request = [ { 'contract_lifecycle_schedule_id': contract_lifecycle_schedule_id, 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, } ] with pytest.raises(ValidationError) as excinfo: logic._validate_and_get_existing_contract_lifecycle_schedules( mock_contract.contract_id, mock_put_request, is_updating_current_lifecycle_schedule, ) msg = ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='ContractLifecycleSchedule', object_id=contract_lifecycle_schedule_id, ) assert excinfo.value.messages[0] == msg def test_validate_contract_lifecycle_schedules_throws_error(): """Test _validate_and_get_existing_contract_lifecycle_schedules. throws an error if contract_lifecycle_schedule is not associated with contract. """ is_updating_current_lifecycle_schedule = False mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.CONTINUOUSLY_ACTIVE ) mock_contract = ContractFactory.create() contract_lifecycle_schedule_id = ( mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id ) mock_put_request = [ { 'contract_lifecycle_schedule_id': contract_lifecycle_schedule_id, 'renewal_type': 'continuously_active', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, } ] with pytest.raises(ValidationError) as excinfo: logic._validate_and_get_existing_contract_lifecycle_schedules( mock_contract.contract_id, mock_put_request, is_updating_current_lifecycle_schedule, ) msg = ERROR_CONTRACT_LIFECYCLE_SCHEDULE_NOT_MATCHED.format( contract_lifecycle_schedule_id, mock_contract.contract_id ) assert excinfo.value.messages[0] == msg @patch('abacus_contract.logic.contract_lifecycle_schedule.db') @patch('abacus_contract.logic.contract_lifecycle_schedule._create_contract_lifecycle') @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._create_contract_lifecycle_schedules' ) def test_delete_and_create_contract_lifecycle_and_schedules( mock_create_contract_lifecycle_schedules_logic, mock_create_contract_lifecycle_logic, mock_db, ): """Test _delete_and_create_contract_lifecycle_and_schedules function. deletes and creates the both current and subsequent lifecycle schedules. also deletes and creates the contract_lifecycle. """ mock_contract = ContractFactory.create() mock_existing_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ] mock_existing_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_existing_contract_lifecycle_schedules[0], ) mock_contract_lifecycle_schedules = ContractLifecycleScheduleFactory.create_batch(2) mock_create_contract_lifecycle_schedules_logic.return_value = ( mock_contract_lifecycle_schedules ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0] ) mock_create_contract_lifecycle_logic.return_value = mock_contract_lifecycle mock_db.session.commit = MagicMock() for schedule in mock_existing_contract_lifecycle_schedules: schedule._soft_delete = MagicMock() mock_existing_contract_lifecycle._soft_delete = MagicMock() mock_put_request = [ { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_after_certain_date', 'schedule_end': '2030-08-08', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, }, { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 1 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, 'renewal_offset_detail_interval': 150, 'renewal_offset_detail_type': 'day', }, ] is_updating_current_lifecycle_schedule = True new_lifecycle_schedules, new_contract_lifecycle = ( logic._delete_and_create_contract_lifecycle_and_schedules( mock_contract.contract_id, mock_put_request, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, mock_existing_contract_lifecycle.lifecycle_term_start, is_updating_current_lifecycle_schedule, ) ) assert new_lifecycle_schedules == mock_contract_lifecycle_schedules assert new_contract_lifecycle == mock_contract_lifecycle mock_existing_contract_lifecycle_schedules[0]._soft_delete.assert_called_once() mock_existing_contract_lifecycle_schedules[1]._soft_delete.assert_called_once() mock_create_contract_lifecycle_schedules_logic.assert_called_once_with( mock_contract.contract_id, mock_put_request ) mock_create_contract_lifecycle_logic.assert_called_once_with( mock_contract.contract_id, mock_existing_contract_lifecycle.lifecycle_term_start, mock_contract_lifecycle_schedules[0], ) mock_existing_contract_lifecycle._soft_delete.assert_called_once() mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.contract_lifecycle_schedule.db') @patch('abacus_contract.logic.contract_lifecycle_schedule._create_contract_lifecycle') @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._create_contract_lifecycle_schedules' ) def test_delete_and_create_contract_schedules( mock_create_contract_lifecycle_schedules_logic, mock_create_contract_lifecycle_logic, mock_db, ): """Test _delete_and_create_contract_lifecycle_and_schedules function. delete and creates only subsequent lifecycle when subsequent lifecycle is updated. """ mock_contract = ContractFactory.create() mock_existing_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ] mock_existing_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_existing_contract_lifecycle_schedules[0], ) mock_contract_lifecycle_schedules = [ContractLifecycleScheduleFactory.create()] mock_create_contract_lifecycle_schedules_logic.return_value = ( mock_contract_lifecycle_schedules ) mock_db.session.commit = MagicMock() for schedule in mock_existing_contract_lifecycle_schedules: schedule._soft_delete = MagicMock() mock_existing_contract_lifecycle._soft_delete = MagicMock() mock_put_request = [ { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_after_certain_date', 'schedule_end': '2030-08-08', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, }, { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 1 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, 'renewal_offset_detail_interval': 150, 'renewal_offset_detail_type': 'day', }, ] is_updating_current_lifecycle_schedule = False new_lifecycle_schedules, new_contract_lifecycle = ( logic._delete_and_create_contract_lifecycle_and_schedules( mock_contract.contract_id, mock_put_request, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, mock_existing_contract_lifecycle.lifecycle_term_start, is_updating_current_lifecycle_schedule, ) ) assert new_lifecycle_schedules == mock_contract_lifecycle_schedules assert new_contract_lifecycle == mock_existing_contract_lifecycle assert mock_create_contract_lifecycle_logic[0].call_count == 0 mock_existing_contract_lifecycle_schedules[1]._soft_delete.assert_called_once() mock_create_contract_lifecycle_schedules_logic.assert_called_once_with( mock_contract.contract_id, mock_put_request ) mock_create_contract_lifecycle_logic.assert_not_called() mock_existing_contract_lifecycle._soft_delete.assert_not_called() mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.contract_lifecycle_schedule.db') @patch('abacus_contract.logic.contract_lifecycle_schedule._create_contract_lifecycle') @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._create_contract_lifecycle_schedules' ) def test_delete_and_create_contract_lifecycle_and_schedules_error( mock_create_contract_lifecycle_schedules_logic, mock_create_contract_lifecycle_logic, mock_db, ): """Test _delete_and_create_contract_lifecycle_and_schedules function. throws an error in case lifecycle schedule creation fails """ mock_contract = ContractFactory.create() mock_existing_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, ), ] mock_existing_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_existing_contract_lifecycle_schedules[0], ) mock_create_contract_lifecycle_schedules_logic.side_effect = SQLAlchemyError( 'creating ContractLifecycleSchedule failed' ) mock_db.session.commit = MagicMock() for schedule in mock_existing_contract_lifecycle_schedules: schedule._soft_delete = MagicMock() mock_existing_contract_lifecycle._soft_delete = MagicMock() mock_put_request = [ { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_after_certain_date', 'schedule_end': '2030-08-08', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, }, { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 1 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, 'renewal_offset_detail_interval': 150, 'renewal_offset_detail_type': 'day', }, ] is_updating_current_lifecycle_schedule = False with pytest.raises(Exception) as exc_info: logic._delete_and_create_contract_lifecycle_and_schedules( mock_contract.contract_id, mock_put_request, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, mock_existing_contract_lifecycle.lifecycle_term_start, is_updating_current_lifecycle_schedule, ) assert 'creating ContractLifecycleSchedule failed' in str(exc_info.value) assert mock_create_contract_lifecycle_logic[0].call_count == 0 assert mock_create_contract_lifecycle_logic[1].call_count == 0 mock_create_contract_lifecycle_schedules_logic.assert_called_once_with( mock_contract.contract_id, mock_put_request ) mock_create_contract_lifecycle_logic.assert_not_called() mock_existing_contract_lifecycle._soft_delete.assert_not_called() mock_db.session.commit.assert_not_called() @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._delete_and_create_contract_lifecycle_and_schedules' ) @patch('abacus_contract.logic.contract_lifecycle_schedule._validate_request_payload') @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._validate_and_get_existing_contract_lifecycle_schedules' ) @patch('abacus_contract.logic.contract_lifecycle_schedule.ContractLifecycle') def test_update_contract_lifecycle_schedules( mock_contract_lifecycle_model, mock_validate_contract_lifecycle_schedules, mock_validate_request_payload, mock_delete_and_create_logic, ): """Test update_contract_lifecycle_schedules function. successfully deletes and creates the contract lifecycle and schedules. """ mock_contract = ContractFactory.create() mock_existing_contract_lifecycle = ContractLifecycleFactory.create() mock_existing_contract_lifecycle_schedules = ( ContractLifecycleScheduleFactory.create_batch(2) ) mock_contract_lifecycle_schedules = ContractLifecycleScheduleFactory.create_batch(2) mock_contract_lifecycle_model.get_by_contract_id.return_value = ( mock_existing_contract_lifecycle ) mock_validate_contract_lifecycle_schedules.return_value = ( mock_existing_contract_lifecycle_schedules ) mock_validate_request_payload.return_value = True mock_delete_and_create_logic.return_value = ( mock_contract_lifecycle_schedules, mock_existing_contract_lifecycle, ) mock_put_request = [ { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_after_certain_date', 'schedule_end': '2030-08-08', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, }, { 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, 'renewal_offset_detail_interval': 150, 'renewal_offset_detail_type': 'day', }, ] res = logic.update_contract_lifecycle_schedules( mock_contract.contract_id, mock_put_request ) assert res.status == 200 mock_contract_lifecycle_model.get_by_contract_id.assert_called_once_with( mock_contract.contract_id ) mock_validate_contract_lifecycle_schedules.assert_called_once_with( mock_contract.contract_id, mock_put_request, False ) mock_validate_request_payload.assert_called_once_with( mock_contract.contract_id, 'distribution', [], mock_put_request, True, False ) mock_delete_and_create_logic.assert_called_once_with( mock_contract.contract_id, mock_put_request, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, mock_existing_contract_lifecycle.lifecycle_term_start, False, ) @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._delete_and_create_contract_lifecycle_and_schedules' ) @patch('abacus_contract.logic.contract_lifecycle_schedule._validate_request_payload') @patch( 'abacus_contract.logic.contract_lifecycle_schedule' '._validate_and_get_existing_contract_lifecycle_schedules' ) @patch('abacus_contract.logic.contract_lifecycle_schedule.ContractLifecycle') def test_update_contract_lifecycle_schedules_error( mock_contract_lifecycle_model, mock_validate_contract_lifecycle_schedules, mock_validate_request_payload, mock_delete_and_create_logic, ): """Test update_contract_lifecycle_schedules function. returns an error in case PUT request validation fails. """ mock_contract = ContractFactory.create() mock_existing_contract_lifecycle = ContractLifecycleFactory.create() mock_existing_contract_lifecycle_schedules = ( ContractLifecycleScheduleFactory.create_batch(2) ) mock_contract_lifecycle_model.get_by_contract_id.return_value = ( mock_existing_contract_lifecycle ) mock_validate_contract_lifecycle_schedules.return_value = ( mock_existing_contract_lifecycle_schedules ) validation_msg = 'contract lifecycle schedule validation failed' mock_validate_request_payload.side_effect = ValidationError(validation_msg) mock_put_request = [ { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 0 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_after_certain_date', 'schedule_end': '2030-08-08', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, }, { 'contract_lifecycle_schedule_id': mock_existing_contract_lifecycle_schedules[ 1 ].contract_lifecycle_schedule_id, 'renewal_type': 'renew_periodically', 'termination_notice_detail_interval': 60, 'termination_notice_detail_type': 'day', 'collection_period_detail_interval': None, 'collection_period_detail_type': None, 'renewal_offset_detail_interval': 150, 'renewal_offset_detail_type': 'day', }, ] res = logic.update_contract_lifecycle_schedules( mock_contract.contract_id, mock_put_request ) assert res.status == 400 assert res.errors['message'] == validation_msg mock_contract_lifecycle_model.get_by_contract_id.assert_called_once_with( mock_contract.contract_id ) mock_validate_contract_lifecycle_schedules.assert_called_once_with( mock_contract.contract_id, mock_put_request, False ) mock_validate_request_payload.assert_called_once_with( mock_contract.contract_id, 'distribution', [], mock_put_request, True, False ) mock_delete_and_create_logic.assert_not_called()