"""Unit tests for contract_lifecycle_schedule logic.""" import datetime from unittest.mock import MagicMock, patch from abacus_common_logic.constants.error import ERROR_ENTITY_DOES_NOT_EXIST from marshmallow import ValidationError import pytest import sqlalchemy from abacus_contract.constants.constants import \ CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES from abacus_contract.constants.constants import CONTRACT_TYPES from abacus_contract.constants.error import \ ERROR_COLLECTION_PERIOD_DETAIL_NOT_REQUIRED from abacus_contract.constants.error import \ ERROR_COLLECTION_PERIOD_DETAIL_REQUIRED from abacus_contract.constants.error import \ ERROR_CONTRACT_LIFECYCLE_SCHEDULE_NOT_MATCHED from abacus_contract.constants.error import \ ERROR_MORE_THAN_TWO_CONTRACT_LIFECYCLE_SCHEDULES_NOT_ALLOWED from abacus_contract.constants.error import \ ERROR_SCHEDULE_END_REQUIRED from abacus_contract.constants.error import \ ERROR_SCHEDULE_END_SHOULD_BE_IN_FUTURE from abacus_contract.constants.error import \ ERROR_SUBSEQUENT_CONTRACT_LIFECYCLE_SCHEDULES_CAN_NOT_CREATED from abacus_contract.constants.error import \ ERROR_SUBSEQUENT_RENEW_AFTER_CERTAIN_DATE_CAN_NOT_CREATED from abacus_contract.constants.error import \ ERROR_TWO_CONTRACT_LIFECYCLE_SCHEDULES_ALLOWED from abacus_contract.logic import contract_lifecycle_schedule as logic from tests.utils.factories import ContractFactory from tests.utils.factories import ContractLifecycleFactory from tests.utils.factories import ContractLifecycleScheduleDetailFactory from tests.utils.factories import 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.""" # noqa: E501 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 ) 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.""" # noqa: E501 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 ) 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.ContractLifecycleSchedule') @patch('abacus_contract.logic.contract_lifecycle_schedule' '._get_detail_ids') def test__create_contract_lifecycle_schedules(mock_get_detail_ids, mock_model): """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() @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), } }] 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_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, # noqa: E501 '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, # noqa: E501 '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, # noqa: E501 '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, # noqa: E501 '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 = \ sqlalchemy.exc.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, # noqa: E501 '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, # noqa: E501 '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): res = 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 res.status == 500 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, # noqa: E501 '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, # noqa: E501 '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, # noqa: E501 '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()