"""Unit tests for contract_lifecycle logic.""" import datetime from unittest.mock import MagicMock from unittest.mock import patch from abacus_common_logic.constants.error import ERROR_ENTITY_DOES_NOT_EXIST from abacus_common_logic.utils.dates import current_timestamp from freezegun import freeze_time from marshmallow import ValidationError 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.error import \ ERROR_CAN_NOT_TERMINATE_CONTRACT_LIFECYCLE from abacus_contract.constants.error import \ ERROR_CONTRACT_LIFECYCLE_ALREADY_EXIST from abacus_contract.constants.error import \ ERROR_CONTRACT_LIFECYCLE_NOT_FOUND from abacus_contract.constants.error import \ ERROR_CONTRACT_LIFECYCLE_SCHEDULE_NOT_MATCHED 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_CURRENT_CONTRACT_LIFECYCLE_SCHEDULE_NOT_FOUND from abacus_contract.constants.error import \ ERROR_INVALID_RENEWAL_EFFECTIVE_DATE from abacus_contract.constants.error import \ ERROR_INVALID_STATUS_FOR_ACTIVATING_CONTRACT_LIFECYCLE from abacus_contract.constants.error import \ ERROR_INVALID_STATUS_FOR_RENEWING_CONTRACT_LIFECYCLE from abacus_contract.constants.error import \ ERROR_INVALID_STATUS_FOR_TERMINATING_CONTRACT_LIFECYCLE from abacus_contract.logic import contract_lifecycle 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.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle.Contract') def test_get_contract_lifecycle_by_contract_id( mock_contract_model, mock_contract_lifecycle_model ): """Test getting a contract lifecycle by contract_id.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule ) contract_id = mock_contract.contract_id mock_contract_model.get_by_id_or_error.return_value = mock_contract mock_contract_lifecycle_model.get_by_contract_id.return_value = \ mock_contract_lifecycle res = logic.get_contract_lifecycle_by_contract_id(contract_id) assert res.status == 200 assert res.message['contract_lifecycle_id'] == \ mock_contract_lifecycle.contract_lifecycle_id mock_contract_lifecycle_model.get_by_contract_id.assert_called_once_with( contract_id ) @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') @patch('abacus_contract.logic.contract_lifecycle.is_feature_enabled') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle.Contract') def test_update_contract_lifecycle_by_contract_id( mock_contract_model, mock_contract_lifecycle_model, mock_is_feature_enabled, mock_emit_contract_kafka_event, ): """Test terminate_contract_lifecycle for past.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule, ) mock_is_feature_enabled.return_value = True contract_id = mock_contract.contract_id mock_contract_model.get_by_id_or_error.return_value = mock_contract mock_contract_lifecycle_model.get_by_contract_id.return_value = ( mock_contract_lifecycle ) mock_params = {'lifecycle_term_start': datetime.date(2025, 6, 3)} result = logic.update_contract_lifecycle_by_contract_id(contract_id, mock_params) lifecycle = result.message assert lifecycle is not None assert lifecycle['contract_id'] == contract_id assert str(lifecycle['lifecycle_term_start']) \ == str(mock_params['lifecycle_term_start']) @patch('abacus_contract.logic.contract_lifecycle.Contract') def test_get_deleted_contract_lifecycle_by_contract_id(mock_model): """Test getting a deleted contract lifecycle by contract_id.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule, deleted_by='Test User' ) contract_id = mock_contract.contract_id mock_model.get_by_id_or_error.return_value = mock_contract res = logic.get_contract_lifecycle_by_contract_id(contract_id) assert res.status == 200 assert not res.message @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_get_contract_lifecycles_by_contract_ids(mock_model): """Test getting a list of contract lifecycle by contract ids.""" mock_contract_lifecycles = [ ContractLifecycleFactory.create(), ContractLifecycleFactory.create() ] ids = [ contract_lifecycle.contract_id for contract_lifecycle in mock_contract_lifecycles ] ids.append(12345) mock_model.get_by_contract_ids.return_value = mock_contract_lifecycles res = logic.get_contract_lifecycles_by_contract_ids(ids) assert res.status == 200 assert len(res.message) == len(ids) assert res.message == [{ 'data': { 'contract_lifecycle_id': mock_contract_lifecycles[0].contract_lifecycle_id, 'contract_id': mock_contract_lifecycles[0].contract_id, 'contract_lifecycle_schedule_id': mock_contract_lifecycles[0].contract_lifecycle_schedule_id, 'last_renewed': None, 'lifecycle_status': mock_contract_lifecycles[0].lifecycle_status, 'lifecycle_term_start': str(mock_contract_lifecycles[0].lifecycle_term_start), 'lifecycle_term_end': mock_contract_lifecycles[0].lifecycle_term_end, 'renewal_effective': mock_contract_lifecycles[0].renewal_effective, 'termination_notice_deadline': mock_contract_lifecycles[0].termination_notice_deadline, 'termination_notice_received': mock_contract_lifecycles[0].termination_notice_received, 'termination_effective': mock_contract_lifecycles[0].termination_effective, 'collection_start': mock_contract_lifecycles[0].collection_start, 'collection_end': mock_contract_lifecycles[0].collection_end } }, { 'data': { 'contract_lifecycle_id': mock_contract_lifecycles[1].contract_lifecycle_id, 'contract_id': mock_contract_lifecycles[1].contract_id, 'contract_lifecycle_schedule_id': mock_contract_lifecycles[1].contract_lifecycle_schedule_id, 'last_renewed': None, 'lifecycle_status': mock_contract_lifecycles[1].lifecycle_status, 'lifecycle_term_start': str(mock_contract_lifecycles[1].lifecycle_term_start), 'lifecycle_term_end': mock_contract_lifecycles[1].lifecycle_term_end, 'renewal_effective': mock_contract_lifecycles[1].renewal_effective, 'termination_notice_deadline': mock_contract_lifecycles[1].termination_notice_deadline, 'termination_notice_received': mock_contract_lifecycles[1].termination_notice_received, 'termination_effective': mock_contract_lifecycles[1].termination_effective, 'collection_start': mock_contract_lifecycles[1].collection_start, 'collection_end': mock_contract_lifecycles[1].collection_end } }, { 'data': None }] @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_get_contract_lifecycles_by_ids(mock_model): """Test getting a list of contract lifecycle by ids.""" mock_contract_lifecycles = [ ContractLifecycleFactory.create(), ContractLifecycleFactory.create() ] ids = [ contract_lifecycle.contract_lifecycle_id for contract_lifecycle in mock_contract_lifecycles ] ids.append(12345) mock_model.get_by_ids.return_value = mock_contract_lifecycles res = logic.get_contract_lifecycles_by_ids(ids) assert res.status == 200 assert len(res.message) == len(ids) assert res.message == [{ 'data': { 'contract_lifecycle_id': mock_contract_lifecycles[0].contract_lifecycle_id, 'contract_id': mock_contract_lifecycles[0].contract_id, 'contract_lifecycle_schedule_id': mock_contract_lifecycles[0].contract_lifecycle_schedule_id, 'last_renewed': None, 'lifecycle_status': mock_contract_lifecycles[0].lifecycle_status, 'lifecycle_term_start': str(mock_contract_lifecycles[0].lifecycle_term_start), 'lifecycle_term_end': mock_contract_lifecycles[0].lifecycle_term_end, 'renewal_effective': mock_contract_lifecycles[0].renewal_effective, 'termination_notice_deadline': mock_contract_lifecycles[0].termination_notice_deadline, 'termination_notice_received': mock_contract_lifecycles[0].termination_notice_received, 'termination_effective': mock_contract_lifecycles[0].termination_effective, 'collection_start': mock_contract_lifecycles[0].collection_start, 'collection_end': mock_contract_lifecycles[0].collection_end } }, { 'data': { 'contract_lifecycle_id': mock_contract_lifecycles[1].contract_lifecycle_id, 'contract_id': mock_contract_lifecycles[1].contract_id, 'contract_lifecycle_schedule_id': mock_contract_lifecycles[1].contract_lifecycle_schedule_id, 'last_renewed': None, 'lifecycle_status': mock_contract_lifecycles[1].lifecycle_status, 'lifecycle_term_start': str(mock_contract_lifecycles[1].lifecycle_term_start), 'lifecycle_term_end': mock_contract_lifecycles[1].lifecycle_term_end, 'renewal_effective': mock_contract_lifecycles[1].renewal_effective, 'termination_notice_deadline': mock_contract_lifecycles[1].termination_notice_deadline, 'termination_notice_received': mock_contract_lifecycles[1].termination_notice_received, 'termination_effective': mock_contract_lifecycles[1].termination_effective, 'collection_start': mock_contract_lifecycles[1].collection_start, 'collection_end': mock_contract_lifecycles[1].collection_end } }, { 'data': None }] @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycleSchedule') def test_get_contract_lifecycle_by_schedule_id(mock_model): """Test getting a contract lifecycle by contract_lifecycle_schedule_id.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule ) contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_model.get_by_id_or_error.return_value = mock_contract_lifecycle_schedule res = logic.get_contract_lifecycle_by_schedule_id(contract_lifecycle_schedule_id) assert res.status == 200 assert res.message['contract_lifecycle_id'] == \ mock_contract_lifecycle.contract_lifecycle_id @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycleSchedule') def test_get_deleted_contract_lifecycle_by_schedule_id(mock_model): """Test getting a deleted contract lifecycle by contract_lifecycle_schedule_id.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule, deleted_by='Test User' ) contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_model.get_by_id_or_error.return_value = mock_contract_lifecycle_schedule res = logic.get_contract_lifecycle_by_schedule_id(contract_lifecycle_schedule_id) assert res.status == 200 assert not res.message @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycleSchedule') def test_get_contract_lifecycle_by_deleted_schedule_id(mock_model): """Test getting a contract lifecycle by deleted contract_lifecycle_schedule_id.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract, deleted_by='Test User' ) ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule, deleted_by='Test User', deleted_at='2028-07-15' ) contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_model.get_by_id_or_error.return_value = mock_contract_lifecycle_schedule res = logic.get_contract_lifecycle_by_schedule_id(contract_lifecycle_schedule_id) assert res.status == 404 assert not res.message == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='ContractLifecycleSchedule', object_id=mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id ) def test_get_lifecycle_term_end_for_continuously_active(): """Test getting lifecycle_term_end date. returns None when renewal type is continuously_active. """ mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.CONTINUOUSLY_ACTIVE ) lifecycle_start_date = '2024-07-30' res = logic._get_lifecycle_term_end( lifecycle_start_date, mock_contract_lifecycle_schedule ) assert res is None def test_get_lifecycle_term_end_for_renew_after_certain_date(): """Test getting lifecycle_term_end date. returns schedule_end date when renewal type is renew_after_certain_date. """ schedule_end = '2025-07-30' mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_AFTER_CERTAIN_DATE, # noqa: E501 schedule_end=schedule_end ) lifecycle_start_date = '2024-07-30' res = logic._get_lifecycle_term_end( lifecycle_start_date, mock_contract_lifecycle_schedule ) assert str(res) == schedule_end def test_get_lifecycle_term_end_for_renew_periodically(): """Test getting lifecycle_term_end date. returns date by adding lifecycle_start_date and renewal_offset schedule when renewal type is renew_periodically. """ mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=20, period_type='year' ) mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ) lifecycle_start_date = datetime.date(2024, 7, 30) res = logic._get_lifecycle_term_end( lifecycle_start_date, mock_contract_lifecycle_schedule ) assert str(res) == '2044-07-30' def test_get_termination_notice_deadline_with_none_term_end(): """Test getting termination_notice_deadline date. returns None when lifecycle_term_end is None. """ mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=45, period_type='day' ) lifecycle_term_end = None res = logic._get_termination_notice_deadline( lifecycle_term_end, mock_schedule_detail.contract_lifecycle_schedule_detail_id ) assert res is None def test_get_termination_notice_deadline_with_term_end_not_none(): """Test getting termination_notice_deadline date. returns date by subtracting lifecycle_term_end and termination_notice_deadline when lifecycle_term_end is not None. """ mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=45, period_type='day' ) lifecycle_term_end = datetime.date(2025, 7, 30) res = logic._get_termination_notice_deadline( lifecycle_term_end, mock_schedule_detail.contract_lifecycle_schedule_detail_id ) assert str(res) == '2025-06-15' def test_get_lifecycle_status_with_future_term_start(): """Test getting lifecycle_status when lifecycle_term_start is future date.""" lifecycle_term_start = datetime.date(2050, 7, 30) lifecycle_term_end = None res = logic._get_lifecycle_status( lifecycle_term_start, lifecycle_term_end ) assert res == CONTRACT_LIFECYCLE_STATUSES.INIT def test_get_lifecycle_status_with_past_term_start(): """Test getting lifecycle_status when lifecycle_term_start is past date.""" lifecycle_term_start = datetime.date(2024, 6, 30) lifecycle_term_end = None res = logic._get_lifecycle_status( lifecycle_term_start, lifecycle_term_end ) assert res == CONTRACT_LIFECYCLE_STATUSES.ACTIVE def test_get_lifecycle_status_with_current_term_start(): """Test getting lifecycle_status when lifecycle_term_start is current date. And lifecycle_term_end is future date. """ lifecycle_term_start = current_timestamp().date() lifecycle_term_end = datetime.date(2050, 12, 31) res = logic._get_lifecycle_status( lifecycle_term_start, lifecycle_term_end ) assert res == CONTRACT_LIFECYCLE_STATUSES.ACTIVE @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_validate_contract_lifecycle_already_exist(mock_model): """Test _validate_and_get_contract_lifecycle_schedule function. throws an error when a contract_lifecycle for specified contract_id and contract_lifecycle_schedule_id already exist. """ mock_contract_lifecycle = ContractLifecycleFactory.create() contract_id = mock_contract_lifecycle.contract_id contract_lifecycle_schedule_id = \ mock_contract_lifecycle.contract_lifecycle_schedule_id mock_model.get_by_contract_id_and_schedule_id.return_value = \ mock_contract_lifecycle msg = ERROR_CONTRACT_LIFECYCLE_ALREADY_EXIST.format( contract_id, contract_lifecycle_schedule_id ) with pytest.raises(Exception) as excinfo: logic._validate_and_get_contract_lifecycle_schedule( contract_id, contract_lifecycle_schedule_id ) assert str(excinfo.value) == msg @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle.Contract') def test_validate_contract_not_exist( mock_contract_model, mock_contract_lifecycle_model ): """Test _validate_and_get_contract_lifecycle_schedule function. throws an error when a contract doesn't exist. """ msg = 'contract not exist' mock_contract_model.get_by_id_or_error.side_effect = Exception(msg) with pytest.raises(Exception) as excinfo: logic._validate_and_get_contract_lifecycle_schedule(999, 1) assert str(excinfo.value) == msg mock_contract_lifecycle_model.get_by_contract_id_and_schedule_id \ .assert_not_called() @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycleSchedule') def test_validate_contract_lifecycle_schedule_not_exist( mock_contract_lifecycle_schedule_model, mock_contract_lifecycle_model ): """Test _validate_and_get_contract_lifecycle_schedule function. throws an error when a contract_lifecycle_schedule doesn't exist. """ mock_contract_lifecycle = ContractLifecycleFactory.create() contract_id = mock_contract_lifecycle.contract_id msg = 'contract_lifecycle_schedule not exist' mock_contract_lifecycle_schedule_model.get_by_id_or_error.side_effect = \ Exception(msg) with pytest.raises(Exception) as excinfo: logic._validate_and_get_contract_lifecycle_schedule( contract_id, 9999 ) assert str(excinfo.value) == msg mock_contract_lifecycle_model.get_by_contract_id_and_schedule_id \ .assert_not_called() @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_validate_schedule_is_not_associated_with_contract( mock_contract_lifecycle_model ): """Test _validate_and_get_contract_lifecycle_schedule function. throws an error when a contract_lifecycle_schedule is not associated with specified contract. """ mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create() contract_id = \ mock_contract.contract_id contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_contract_lifecycle_model.get_by_contract_id_and_schedule_id.return_value = \ None with pytest.raises(Exception) as excinfo: logic._validate_and_get_contract_lifecycle_schedule( contract_id, contract_lifecycle_schedule_id ) assert str(excinfo.value) == ERROR_CONTRACT_LIFECYCLE_SCHEDULE_NOT_MATCHED.format( contract_lifecycle_schedule_id, contract_id ) @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_validate_schedule_associated_with_contract( mock_contract_lifecycle_model ): """Test _validate_and_get_contract_lifecycle_schedule function. returns a contract_lifecycle_schedule when contract_lifecycle_schedule is associated with specified contract. """ mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) contract_id = \ mock_contract.contract_id contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_contract_lifecycle_model.get_by_contract_id_and_schedule_id.return_value = \ None res = logic._validate_and_get_contract_lifecycle_schedule( contract_id, contract_lifecycle_schedule_id ) assert res == mock_contract_lifecycle_schedule @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_validate_contract_lifecycle_schedule_is_deleted( mock_contract_lifecycle_model ): """Test _validate_and_get_contract_lifecycle_schedule function. throws an error a contract_lifecycle_schedule is associated with the contract but its already deleted. """ mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract, deleted_by='Test User' ) contract_id = \ mock_contract.contract_id contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_contract_lifecycle_model.get_by_contract_id_and_schedule_id.return_value = \ None with pytest.raises(Exception) as excinfo: logic._validate_and_get_contract_lifecycle_schedule( contract_id, contract_lifecycle_schedule_id ) assert str(excinfo.value) == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='ContractLifecycleSchedule', object_id=contract_lifecycle_schedule_id ) @patch('abacus_contract.logic.contract_lifecycle._generate_contract_lifecycle_params') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test__create_contract_lifecycle( mock_model, mock_generate_contract_lifecycle_params ): """Test creating a contract_lifecycle.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule ) contract_id = \ mock_contract.contract_id contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_model.build.return_value = mock_contract_lifecycle mock_contract_lifecycle_params = { 'contract_id': mock_contract.contract_id, 'contract_lifecycle_schedule_id': contract_lifecycle_schedule_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.INIT, 'lifecycle_term_start': '2024-09-25', 'lifecycle_term_end': None, 'renewal_effective': None, 'termination_notice_deadline': None } mock_generate_contract_lifecycle_params.return_value = \ mock_contract_lifecycle_params lifecycle_term_start = '2024-07-30' res = logic._create_contract_lifecycle( contract_id, lifecycle_term_start, mock_contract_lifecycle_schedule ) assert res == mock_contract_lifecycle mock_model.build.assert_called_once_with(**mock_contract_lifecycle_params) @patch('abacus_contract.logic.contract_lifecycle._get_lifecycle_term_end') def test_create_contract_lifecycle_with_same_start_and_end_date( mock_get_lifecycle_term_end ): """Test creating a contract_lifecycle. throws an error when lifecycle_term_start is same as lifecycle_term_end. """ mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) contract_id = \ mock_contract.contract_id lifecycle_term_start = lifecycle_term_end = '2024-07-30' mock_get_lifecycle_term_end.return_value = lifecycle_term_end with pytest.raises(Exception) as excinfo: logic._create_contract_lifecycle( contract_id, lifecycle_term_start, mock_contract_lifecycle_schedule ) assert str(excinfo.value) == \ ERROR_CONTRACT_LIFECYCLE_START_AND_END_DATE_NOT_VALID.format( lifecycle_term_start, lifecycle_term_end) @patch('abacus_contract.logic.contract_lifecycle._create_contract_lifecycle') @patch('abacus_contract.logic.contract_lifecycle.' '_validate_and_get_contract_lifecycle_schedule') def test_create_contract_lifecycle( mock_validation, mock_create_contract_lifecycle_logic, ): """Test create_contract_lifecycle function.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule ) contract_id = \ mock_contract.contract_id contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_validation.return_value = mock_contract_lifecycle_schedule mock_create_contract_lifecycle_logic.return_value = mock_contract_lifecycle lifecycle_term_start = '2024-07-30' res = logic.create_contract_lifecycle( contract_id, contract_lifecycle_schedule_id, lifecycle_term_start ) assert res.status == 201 mock_create_contract_lifecycle_logic.assert_called_once_with( contract_id, lifecycle_term_start, mock_contract_lifecycle_schedule ) @patch('abacus_common_logic.models.base.CRUDMixin.update_attributes') @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') @patch('abacus_contract.logic.contract_lifecycle.is_feature_enabled') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle._validate_contract_lifecycle_dates') def test_update_contract_lifecycle( mock_validate_handler, mock_contract_lifecycle, mock_is_feature_enabled, mock_emit_contract_kafka_event, mock_update_attributes, ): """Test _update_contract_lifecycle function.""" contract_lifecycle = ContractLifecycleFactory.create() termination_date = datetime.date(2024, 8, 29) termination_notice_received = datetime.date(2024, 8, 20) data = { 'termination_effective': termination_date, 'termination_notice_received': termination_notice_received, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED } mock_validate_handler.return_value = True mock_is_feature_enabled.return_value = False mock_contract_lifecycle.commit_changes.return_value = contract_lifecycle mock_update_attributes.return_value = contract_lifecycle logic._update_contract_lifecycle( contract_lifecycle, **data ) mock_validate_handler.assert_called_once_with( contract_lifecycle, data, ) mock_update_attributes.assert_called_once_with( termination_effective=termination_date, termination_notice_received=termination_notice_received, lifecycle_term_end=termination_date, renewal_effective=None, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED, ) mock_contract_lifecycle.commit_changes.assert_called_once() mock_is_feature_enabled.assert_called_once() mock_emit_contract_kafka_event.assert_called_once_with( contract_lifecycle.contract_id, constants.CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_common_logic.models.base.CRUDMixin.update_attributes') @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') @patch('abacus_contract.logic.contract_lifecycle.is_feature_enabled') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle._validate_contract_lifecycle_dates') def test_update_contract_lifecycle_with_feature_flag( mock_validate_handler, mock_contract_lifecycle, mock_is_feature_enabled, mock_emit_contract_kafka_event, mock_update_attributes, ): """Test _update_contract_lifecycle function. Does not emit kafka event when ABACUS_CONTRACT_SIGNING_ENTITY feature flag on. """ contract_lifecycle = ContractLifecycleFactory.create() termination_date = datetime.date(2024, 8, 29) termination_notice_received = datetime.date(2024, 8, 20) data = { 'termination_effective': termination_date, 'termination_notice_received': termination_notice_received, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED } mock_validate_handler.return_value = True mock_is_feature_enabled.return_value = True mock_contract_lifecycle.commit_changes.return_value = contract_lifecycle mock_update_attributes.return_value = contract_lifecycle logic._update_contract_lifecycle( contract_lifecycle, **data ) mock_validate_handler.assert_called_once_with( contract_lifecycle, data, ) mock_update_attributes.assert_called_once_with( termination_effective=termination_date, termination_notice_received=termination_notice_received, lifecycle_term_end=termination_date, renewal_effective=None, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED, ) mock_contract_lifecycle.commit_changes.assert_called_once() mock_is_feature_enabled.assert_called_once() mock_emit_contract_kafka_event.assert_not_called() @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') @patch('abacus_contract.logic.contract_lifecycle.is_feature_enabled') @patch('abacus_contract.logic.contract_lifecycle._validate_contract_lifecycle_dates') def test_update_contract_lifecycle_error( mock_validate_handler, mock_is_feature_enabled, mock_emit_contract_kafka_event ): """Test _update_contract_lifecycle function error.""" contract_lifecycle = ContractLifecycleFactory.create() termination_date = datetime.date(2024, 8, 27) data = { 'termination_effective': termination_date, 'termination_notice_received': None, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED } mock_validate_handler.side_effect = ValidationError( error.ERROR_TERMINATION_EFFECTIVE_DATE_GREATER_THAN_TERM_END ) mock_is_feature_enabled.return_value = False with pytest.raises( ValidationError, match=error.ERROR_TERMINATION_EFFECTIVE_DATE_GREATER_THAN_TERM_END ): logic._update_contract_lifecycle( contract_lifecycle, **data ) mock_validate_handler.assert_called_once_with( contract_lifecycle, data ) mock_is_feature_enabled.assert_not_called() mock_emit_contract_kafka_event.assert_not_called() def test_validate_contract_lifecycle_dates_valid(): """Test _validate_contract_lifecycle_dates method for valid dates.""" termination_date = datetime.date(2024, 8, 27) mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_term_end=datetime.date(2025, 2, 1) ) data = { 'termination_effective': termination_date } res = logic._validate_contract_lifecycle_dates( mock_contract_lifecycle, data ) assert res is True def test_validate_contract_lifecycle_dates_invalid(): """Test _validate_contract_lifecycle_dates termination date is after end date.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_term_end=datetime.date(2025, 2, 1) ) termination_date = datetime.date(2025, 2, 2) data = { 'termination_effective': termination_date } with pytest.raises( ValidationError, match=error.ERROR_TERMINATION_EFFECTIVE_DATE_GREATER_THAN_TERM_END # noqa: E501 ): logic._validate_contract_lifecycle_dates( mock_contract_lifecycle, data ) @patch('abacus_contract.logic.contract_lifecycle._update_contract_lifecycle') @patch('abacus_contract.logic.contract_lifecycle._updated_fields_for_contract_lifecycle_termination') # noqa: E501 @patch('abacus_contract.logic.contract_lifecycle._validate_contract_lifecycle_termination') # noqa: E501 @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_terminate_contract_lifecycle( mock_model, mock_validate_handler, mock_updated_fields_handler, mock_update_handler, ): """Test terminate_contract_lifecycle for past.""" termination_date = datetime.date(2024, 8, 20) contract_lifecycle = ContractLifecycleFactory.create() mock_model.get_by_contract_id.return_value = contract_lifecycle mock_validate_handler.return_value = True mock_updated_fields_handler.return_value = { 'termination_effective': termination_date, 'termination_notice_received': None, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'lifecycle_status': constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED, } logic.terminate_contract_lifecycle( contract_lifecycle.contract_id, termination_date, None ) mock_update_handler.assert_called_once_with( contract_lifecycle, termination_effective=termination_date, termination_notice_received=None, lifecycle_term_end=termination_date, renewal_effective=None, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED, ) @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_terminate_contract_lifecycle_error( mock_model, ): """Test terminate_contract_lifecycle error.""" termination_date = datetime.date(2024, 8, 20) contract_lifecycle = ContractLifecycleFactory.create() contract_id = contract_lifecycle.contract_id mock_model.get_by_contract_id.return_value = None with pytest.raises( ValidationError, match=error.ERROR_CONTRACT_LIFECYCLE_NOT_FOUND.format(contract_id) ): logic.terminate_contract_lifecycle(contract_id, termination_date, None) def test_validate_contract_lifecycle_termination_status_init(): """Test _validate_contract_lifecycle_termination method for init status.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.INIT ) res = logic._validate_contract_lifecycle_termination(mock_contract_lifecycle) assert res is True def test_validate_contract_lifecycle_termination_status_active(): """Test _validate_contract_lifecycle_termination method for active status.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.ACTIVE ) res = logic._validate_contract_lifecycle_termination(mock_contract_lifecycle) assert res is True def test_validate_contract_lifecycle_termination_status_to_be_terminated(): """Test _validate_contract_lifecycle_termination for to_be_terminated status.""" with pytest.raises( ValidationError, match=error.ERROR_CONTRACT_TERMINATION_INVALID_STATUS ): mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED ) logic._validate_contract_lifecycle_termination(mock_contract_lifecycle) def test_validate_contract_lifecycle_termination_status_terminated(): """Test _validate_contract_lifecycle_termination for terminated status.""" with pytest.raises( ValidationError, match=error.ERROR_CONTRACT_TERMINATION_INVALID_STATUS ): mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) logic._validate_contract_lifecycle_termination(mock_contract_lifecycle) def test_validate_contract_lifecycle_termination_status_in_collection_period(): """Test _validate_contract_lifecycle_termination for in_collection_period status.""" with pytest.raises( ValidationError, match=error.ERROR_CONTRACT_TERMINATION_INVALID_STATUS ): mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.IN_COLLECTION_PERIOD ) logic._validate_contract_lifecycle_termination(mock_contract_lifecycle) def test_validate_contract_lifecycle_termination_status_inactive(): """Test _validate_contract_lifecycle_termination for inactive status.""" with pytest.raises( ValidationError, match=error.ERROR_CONTRACT_TERMINATION_INVALID_STATUS ): mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.INACTIVE ) logic._validate_contract_lifecycle_termination(mock_contract_lifecycle) def test_updated_fields_for_contract_lifecycle_termination_past(): """Test _updated_fields_for_contract_lifecycle_termination for past termination.""" termination_date = datetime.date(2024, 8, 20) contract_lifecycle = ContractLifecycleFactory.create() res = logic._updated_fields_for_contract_lifecycle_termination( contract_lifecycle, termination_date, None ) assert res == { 'termination_effective': termination_date, 'termination_notice_received': None, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'lifecycle_status': constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED, } def test_updated_fields_for_contract_lifecycle_for_termination_with_notice(): """Test _updated_fields_for_contract_lifecycle_termination with notice.""" termination_date = datetime.date(2024, 8, 20) termination_notice_received = datetime.date(2024, 8, 15) contract_lifecycle = ContractLifecycleFactory.create() res = logic._updated_fields_for_contract_lifecycle_termination( contract_lifecycle, termination_date, termination_notice_received ) assert res == { 'termination_effective': termination_date, 'termination_notice_received': termination_notice_received, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'lifecycle_status': constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED, } @freeze_time('2024-08-25') def test_updated_fields_for_contract_lifecycle_for_termination_future(): """Test _updated_fields_for_contract_lifecycle_termination for future.""" termination_date = datetime.date(2024, 8, 26) contract_lifecycle = ContractLifecycleFactory.create() res = logic._updated_fields_for_contract_lifecycle_termination( contract_lifecycle, termination_date, None ) assert res == { 'termination_effective': termination_date, 'termination_notice_received': None, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'lifecycle_status': constants.CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, } def test_updated_fields_for_contract_lifecycle_for_termination_past_collection_period(): # noqa: E501 """Test _updated_fields_for_contract_lifecycle_termination for past collection period.""" # noqa: E501 termination_date = datetime.date(2024, 7, 27) contract = ContractFactory.create() collection_period_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=30, period_type=constants.CONTRACT_LIFECYCLE_SCHEDULE_DETAIL_PERIOD_TYPES.DAY, ) contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=contract, collection_period_detail=collection_period_detail, ) contract_lifecycle = ContractLifecycleFactory.create( contract=contract, contract_lifecycle_schedule=contract_lifecycle_schedule, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.ACTIVE, ) res = logic._updated_fields_for_contract_lifecycle_termination( contract_lifecycle, termination_date, None ) assert res == { 'termination_effective': termination_date, 'termination_notice_received': None, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'collection_start': termination_date, 'collection_end': datetime.date(2024, 8, 26), 'lifecycle_status': constants.CONTRACT_LIFECYCLE_STATUSES.TERMINATED, } @freeze_time('2024-08-25') def test_updated_fields_for_contract_lifecycle_for_termination_ongoing_collection_period(): # noqa: E501 """Test _updated_fields_for_contract_lifecycle_termination for ongoing collection period.""" # noqa: E501 termination_date = datetime.date(2024, 2, 27) contract = ContractFactory.create() collection_period_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=6, period_type=constants.CONTRACT_LIFECYCLE_SCHEDULE_DETAIL_PERIOD_TYPES.MONTH, ) contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=contract, collection_period_detail=collection_period_detail, ) contract_lifecycle = ContractLifecycleFactory.create( contract=contract, contract_lifecycle_schedule=contract_lifecycle_schedule, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.ACTIVE, ) res = logic._updated_fields_for_contract_lifecycle_termination( contract_lifecycle, termination_date, None ) assert res == { 'termination_effective': termination_date, 'termination_notice_received': None, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'collection_start': termination_date, 'collection_end': datetime.date(2024, 8, 27), 'lifecycle_status': constants.CONTRACT_LIFECYCLE_STATUSES.IN_COLLECTION_PERIOD, } @freeze_time('2024-08-25') def test_updated_fields_for_contract_lifecycle_for_termination_future_collection_period(): # noqa: E501 """Test _updated_fields_for_contract_lifecycle_termination for future collection period.""" # noqa: E501 termination_date = datetime.date(2024, 8, 26) contract = ContractFactory.create() collection_period_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=2, period_type=constants.CONTRACT_LIFECYCLE_SCHEDULE_DETAIL_PERIOD_TYPES.YEAR, ) contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=contract, collection_period_detail=collection_period_detail, ) contract_lifecycle = ContractLifecycleFactory.create( contract=contract, contract_lifecycle_schedule=contract_lifecycle_schedule, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.ACTIVE, ) res = logic._updated_fields_for_contract_lifecycle_termination( contract_lifecycle, termination_date, None ) assert res == { 'termination_effective': termination_date, 'termination_notice_received': None, 'lifecycle_term_end': termination_date, 'renewal_effective': None, 'collection_start': termination_date, 'collection_end': datetime.date(2026, 8, 26), 'lifecycle_status': constants.CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, } def test_validate_contract_lifecycle_reactivation(): """Test _validate_contract_lifecycle_reactivation function.""" contract_lifecycle_status = CONTRACT_LIFECYCLE_STATUSES.TERMINATED res = logic._validate_contract_lifecycle_reactivation(contract_lifecycle_status) assert res is True def test_validate_contract_lifecycle_reactivation_error(): """Test error is thrown if contract_lifecycle status is invalid.""" contract_lifecycle_status = CONTRACT_LIFECYCLE_STATUSES.ACTIVE with pytest.raises(Exception) as excinfo: logic._validate_contract_lifecycle_reactivation(contract_lifecycle_status) assert str(excinfo.value) == ERROR_CONTRACT_REACTIVATION_INVALID_STATUS def test_get_current_contract_lifecycle_schedule(): """Test _get_current_contract_lifecycle_schedule function.""" mock_contract_lifecycle_schedules = \ ContractLifecycleScheduleFactory.create_batch(2) mock_contract_lifecycle = ContractLifecycleFactory.create( contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0] ) res = logic._get_current_contract_lifecycle_schedule( mock_contract_lifecycle, mock_contract_lifecycle_schedules ) assert res == mock_contract_lifecycle_schedules[0] def test_get_current_contract_lifecycle_schedule_error(): """Test error is thrown if current contract_lifecycle_schedule is not found.""" mock_contract_lifecycle_schedules = \ ContractLifecycleScheduleFactory.create_batch(2) mock_contract_lifecycle = ContractLifecycleFactory.create() with pytest.raises(Exception) as excinfo: logic._get_current_contract_lifecycle_schedule( mock_contract_lifecycle, mock_contract_lifecycle_schedules ) assert str(excinfo.value) == ERROR_CURRENT_CONTRACT_LIFECYCLE_SCHEDULE_NOT_FOUND @patch('abacus_contract.logic.contract_lifecycle.db') @patch('abacus_contract.logic.contract_lifecycle._create_contract_lifecycle') def test_soft_delete_and_create_contract_lifecycle( mock_create_contract_lifecycle_logic, mock_db ): """Test _soft_delete_and_create_contract_lifecycle function.""" current_date = current_timestamp().date() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=10, period_type='year' ) mock_contract = ContractFactory.create() mock_existing_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ) ] mock_existing_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_existing_contract_lifecycle_schedules[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) mock_existing_contract_lifecycle._soft_delete = MagicMock() mock_contract_lifecycle = ContractLifecycleFactory.create() mock_create_contract_lifecycle_logic.return_value = mock_contract_lifecycle mock_db.session.commit = MagicMock() res = logic._soft_delete_and_create_contract_lifecycle( mock_contract.contract_id, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, mock_existing_contract_lifecycle_schedules[0] ) assert res == mock_contract_lifecycle mock_create_contract_lifecycle_logic.assert_called_once_with( mock_contract.contract_id, current_date, mock_existing_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' '._delete_and_create_contract_lifecycle_and_schedules') def test_soft_delete_and_create_lifecycle_renew_after_certain_date_schedule( mock_delete_and_create_contract_lifecycle_and_schedules, ): """Test _soft_delete_and_create_contract_lifecycle function. soft delete both contract_lifecycle and contract_lifecycle_schedules when renewal_type is renew_after_certain_date and schedule_end is past date. """ current_date = current_timestamp().date() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=10, period_type='year' ) mock_contract = ContractFactory.create() mock_existing_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_AFTER_CERTAIN_DATE, # noqa: E501, schedule_end='2023-08-10' ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ) ] mock_existing_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_existing_contract_lifecycle_schedules[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) mock_existing_contract_lifecycle._soft_delete = MagicMock() mock_contract_lifecycle = ContractLifecycleFactory.create() mock_delete_and_create_contract_lifecycle_and_schedules.return_value = \ (None, mock_contract_lifecycle) res = logic._soft_delete_and_create_contract_lifecycle( mock_contract.contract_id, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, mock_existing_contract_lifecycle_schedules[0] ) continuously_active_schedule = [{ 'renewal_type': CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.CONTINUOUSLY_ACTIVE, 'termination_notice_detail_interval': 30, 'termination_notice_detail_type': 'day' }] assert res == mock_contract_lifecycle mock_delete_and_create_contract_lifecycle_and_schedules.assert_called_once_with( mock_contract.contract_id, continuously_active_schedule, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, current_date, True ) @patch('abacus_contract.logic.contract_lifecycle.db') @patch('abacus_contract.logic.contract_lifecycle._create_contract_lifecycle') def test_soft_delete_and_create_contract_lifecycle_future_schedule_end( mock_create_contract_lifecycle_logic, mock_db ): """Test _soft_delete_and_create_contract_lifecycle function. soft delete only contract_lifecycle when renewal_type is renew_after_certain_date and schedule_end is future date. """ current_date = current_timestamp().date() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=10, period_type='year' ) mock_contract = ContractFactory.create() mock_existing_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_AFTER_CERTAIN_DATE, # noqa: E501, schedule_end='2090-08-10' ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ) ] mock_existing_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_existing_contract_lifecycle_schedules[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) mock_existing_contract_lifecycle._soft_delete = MagicMock() mock_contract_lifecycle = ContractLifecycleFactory.create() mock_create_contract_lifecycle_logic.return_value = mock_contract_lifecycle mock_db.session.commit = MagicMock() res = logic._soft_delete_and_create_contract_lifecycle( mock_contract.contract_id, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, mock_existing_contract_lifecycle_schedules[0] ) assert res == mock_contract_lifecycle mock_create_contract_lifecycle_logic.assert_called_once_with( mock_contract.contract_id, current_date, mock_existing_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' '._soft_delete_and_create_contract_lifecycle') @patch('abacus_contract.logic.contract_lifecycle' '._get_current_contract_lifecycle_schedule') @patch('abacus_contract.logic.contract_lifecycle' '._validate_contract_lifecycle_reactivation') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycleSchedule') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_reactivate_contract_lifecycle( mock_contract_lifecycle_model, mock_contract_lifecycle_schedule_model, mock_validation, mock_get_current_contract_lifecycle_schedule, mock_soft_delete_and_create_contract_lifecycle ): """Test reactivate_contract_lifecycle function.""" mock_contract = ContractFactory.create() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=10, period_type='year' ) mock_existing_contract_lifecycle_schedules = [ ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ), ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ) ] mock_existing_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_existing_contract_lifecycle_schedules[0], lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TERMINATED ) mock_contract_lifecycle = ContractLifecycleFactory.create() mock_contract_lifecycle_model.get_by_contract_id.return_value = \ mock_existing_contract_lifecycle mock_contract_lifecycle_schedule_model \ .get_contract_lifecycle_schedules_by_contract_id.return_value = \ mock_existing_contract_lifecycle_schedules mock_validation.return_value = True mock_get_current_contract_lifecycle_schedule.return_value = \ mock_existing_contract_lifecycle_schedules[0] mock_soft_delete_and_create_contract_lifecycle.return_value = \ mock_contract_lifecycle res = logic.reactivate_contract_lifecycle(mock_contract.contract_id) assert res == mock_contract_lifecycle mock_contract_lifecycle_model.get_by_contract_id.assert_called_once_with( mock_contract.contract_id ) mock_contract_lifecycle_schedule_model \ .get_contract_lifecycle_schedules_by_contract_id \ .assert_called_once_with(mock_contract.contract_id) mock_get_current_contract_lifecycle_schedule.assert_called_once_with( mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules ) mock_soft_delete_and_create_contract_lifecycle.assert_called_once_with( mock_contract.contract_id, mock_existing_contract_lifecycle, mock_existing_contract_lifecycle_schedules, mock_existing_contract_lifecycle_schedules[0] ) @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_reactivate_contract_lifecycle_does_not_exist(mock_contract_lifecycle_model): """Test error is thrown if contract_lifecycle doesn't exist.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_model.get_by_contract_id.return_value = None with pytest.raises(Exception) as excinfo: logic.reactivate_contract_lifecycle(mock_contract.contract_id) assert str(excinfo.value) == ERROR_CONTRACT_LIFECYCLE_NOT_FOUND.format( mock_contract.contract_id ) @patch('abacus_contract.logic.contract_lifecycle' '._validate_contract_lifecycle_reactivation') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycleSchedule') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_reactivate_contract_lifecycle_schedule_does_not_exist( mock_contract_lifecycle_model, mock_contract_lifecycle_schedule_model, mock_validation, ): """Test error is thrown if contract_lifecycle_schedule doesn't exist.""" mock_contract = ContractFactory.create() mock_existing_contract_lifecycle = ContractLifecycleFactory.create() mock_contract_lifecycle_model.get_by_contract_id.return_value = \ mock_existing_contract_lifecycle mock_contract_lifecycle_schedule_model \ .get_contract_lifecycle_schedules_by_contract_id.return_value = \ None mock_validation.return_value = True with pytest.raises(ValidationError) as excinfo: logic.reactivate_contract_lifecycle(mock_contract.contract_id) assert str(excinfo.value) == ERROR_CONTRACT_LIFECYCLE_SCHEDULES_NOT_FOUND def test_validate_contract_lifecycle_termination_status(): """Test _validate_contract_lifecycle_termination_status function.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED ) res = logic._validate_contract_lifecycle_termination_status( mock_contract_lifecycle ) assert res is True def test_validate_contract_lifecycle_termination_status_error(): """Test error is thrown if contract_lifecycle termination status is invalid.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.ACTIVE ) with pytest.raises(Exception) as excinfo: logic._validate_contract_lifecycle_termination_status( mock_contract_lifecycle ) assert str(excinfo.value) == \ ERROR_INVALID_STATUS_FOR_TERMINATING_CONTRACT_LIFECYCLE.format( mock_contract_lifecycle.contract_lifecycle_id) def test_get_updated_lifecycle_termination_status_future_collection_end(): """Test _get_updated_contract_lifecycle_termination_status function. returns contract_lifecycle record with "in_collection_period" lifecycle_status if collection_end is future date. """ today = datetime.date.today() mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, termination_effective=today, collection_start=today, collection_end=datetime.datetime(2050, 10, 20) ) res = logic._get_updated_contract_lifecycle_termination_status( mock_contract_lifecycle ) assert res == { 'contract_lifecycle_id': mock_contract_lifecycle.contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.IN_COLLECTION_PERIOD } def test_get_updated_lifecycle_termination_status_past_collection_end(): """Test _get_updated_contract_lifecycle_termination_status function. returns contract_lifecycle record with "terminated" lifecycle_status if collection_end is past date. """ today = datetime.date.today() mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, termination_effective=today, collection_start=today, collection_end=datetime.datetime(2024, 8, 20) ) res = logic._get_updated_contract_lifecycle_termination_status( mock_contract_lifecycle ) assert res == { 'contract_lifecycle_id': mock_contract_lifecycle.contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED } def test_get_updated_lifecycle_termination_status_current_collection_end(): """Test _get_updated_contract_lifecycle_termination_status function. returns contract_lifecycle record with "terminated" lifecycle_status if collection_end is today's date. """ today = datetime.date.today() mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, termination_effective=datetime.datetime(2024, 8, 20), collection_start=datetime.datetime(2024, 8, 20), collection_end=today ) res = logic._get_updated_contract_lifecycle_termination_status( mock_contract_lifecycle ) assert res == { 'contract_lifecycle_id': mock_contract_lifecycle.contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED } def test_get_updated_lifecycle_termination_status_past_termination_effective(): """Test _get_updated_contract_lifecycle_termination_status function. returns contract_lifecycle record with "terminated" status if termination_effective date is in past and no collection dates are assigned. """ mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, termination_effective=datetime.datetime(2024, 9, 1), collection_start=None, collection_end=None ) res = logic._get_updated_contract_lifecycle_termination_status( mock_contract_lifecycle ) assert res == { 'contract_lifecycle_id': mock_contract_lifecycle.contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED } def test_get_updated_lifecycle_termination_status_future_termination_effective(): """Test _get_updated_contract_lifecycle_termination_status function. throws an error if termination_effective is future date. """ mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, termination_effective=datetime.datetime(2100, 9, 1), collection_start=None, collection_end=None ) with pytest.raises(Exception) as excinfo: logic._get_updated_contract_lifecycle_termination_status( mock_contract_lifecycle ) assert str(excinfo.value) == ERROR_CAN_NOT_TERMINATE_CONTRACT_LIFECYCLE.format( mock_contract_lifecycle.contract_lifecycle_id ) def test_get_updated_lifecycle_termination_status_null_termination_effective(): """Test _get_updated_contract_lifecycle_termination_status function. throws an error if termination_effective is Null. """ mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, termination_effective=None, collection_start=None, collection_end=None ) with pytest.raises(Exception) as excinfo: logic._get_updated_contract_lifecycle_termination_status( mock_contract_lifecycle ) assert str(excinfo.value) == ERROR_CAN_NOT_TERMINATE_CONTRACT_LIFECYCLE.format( mock_contract_lifecycle.contract_lifecycle_id ) @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') @patch('abacus_contract.logic.contract_lifecycle.is_feature_enabled') @patch('abacus_contract.logic.contract_lifecycle' '._get_updated_contract_lifecycle_termination_status') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle' '._validate_contract_lifecycle_termination_status') def test_terminate_contract_lifecycles( mock_validation, mock_contract_lifecycle_model, mock_get_updated_contract_lifecycle_termination_status, mock_is_feature_enabled, mock_emit_contract_kafka_event ): """Test terminate_contract_lifecycles function.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED ) mock_contract_lifecycle_model.get_by_id_or_error.return_value = \ mock_contract_lifecycle contract_lifecycle_id = mock_contract_lifecycle.contract_lifecycle_id mock_validation.return_value = True mock_get_updated_contract_lifecycle_termination_status.return_value = { 'contract_lifecycle_id': contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED } mock_contract_lifecycle_model.bulk_update.return_value = True mock_is_feature_enabled.return_value = False res = logic.terminate_contract_lifecycles([contract_lifecycle_id]) assert res.status == 200 mock_contract_lifecycle_model.bulk_update.assert_called_once_with([{ 'contract_lifecycle_id': contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED }]) mock_is_feature_enabled.assert_called_once() mock_emit_contract_kafka_event.assert_called_once_with( mock_contract_lifecycle.contract_id, constants.CONTRACT_KAFKA_EVENT_NAMES.CONTRACT_UPDATED ) @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') @patch('abacus_contract.logic.contract_lifecycle.is_feature_enabled') @patch('abacus_contract.logic.contract_lifecycle' '._get_updated_contract_lifecycle_termination_status') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle' '._validate_contract_lifecycle_termination_status') def test_terminate_contract_lifecycles_with_feature_flag( mock_validation, mock_contract_lifecycle_model, mock_get_updated_contract_lifecycle_termination_status, mock_is_feature_enabled, mock_emit_contract_kafka_event ): """Test terminate_contract_lifecycles function. Does not emit kafka event when ABACUS_CONTRACT_SIGNING_ENTITY feature flag is on. """ mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED ) mock_contract_lifecycle_model.get_by_id_or_error.return_value = \ mock_contract_lifecycle contract_lifecycle_id = mock_contract_lifecycle.contract_lifecycle_id mock_validation.return_value = True mock_get_updated_contract_lifecycle_termination_status.return_value = { 'contract_lifecycle_id': contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED } mock_contract_lifecycle_model.bulk_update.return_value = True mock_is_feature_enabled.return_value = True res = logic.terminate_contract_lifecycles([contract_lifecycle_id]) assert res.status == 200 mock_contract_lifecycle_model.bulk_update.assert_called_once_with([{ 'contract_lifecycle_id': contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.TERMINATED }]) mock_is_feature_enabled.assert_called_once() mock_emit_contract_kafka_event.assert_not_called() @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') @patch('abacus_contract.logic.contract_lifecycle.is_feature_enabled') def test_terminate_deleted_contract_lifecycles( mock_is_feature_enabled, mock_emit_contract_kafka_event ): """Test terminate_contract_lifecycles function. returns an error if contract_lifecycle is deleted. """ mock_contract_lifecycles = [ ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED, deleted_by='Test User' ), ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED ) ] ids = [ contract_lifecycle.contract_lifecycle_id for contract_lifecycle in mock_contract_lifecycles ] mock_is_feature_enabled.return_value = False res = logic.terminate_contract_lifecycles(ids) assert res.errors['message'] == ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='ContractLifecycle', object_id=mock_contract_lifecycles[0].contract_lifecycle_id ) mock_is_feature_enabled.assert_not_called() mock_emit_contract_kafka_event.assert_not_called() @patch('abacus_contract.logic.contract_lifecycle.emit_contract_event') @patch('abacus_contract.logic.contract_lifecycle.is_feature_enabled') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') @patch('abacus_contract.logic.contract_lifecycle' '._validate_contract_lifecycle_termination_status') def test_terminate_contract_lifecycles_validation_error( mock_validation, mock_contract_lifecycle_model, mock_is_feature_enabled, mock_emit_contract_kafka_event ): """Test terminate_contract_lifecycles function. returns an error if there is any validation error. """ mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.TO_BE_TERMINATED ) mock_contract_lifecycle_model.get_by_id_or_error.return_value = \ mock_contract_lifecycle contract_lifecycle_id = mock_contract_lifecycle.contract_lifecycle_id mock_error = ERROR_INVALID_STATUS_FOR_TERMINATING_CONTRACT_LIFECYCLE.format( contract_lifecycle_id ) mock_validation.side_effect = ValidationError(mock_error) mock_is_feature_enabled.return_value = False res = logic.terminate_contract_lifecycles([contract_lifecycle_id]) assert res.errors['message'] == mock_error mock_is_feature_enabled.assert_not_called() mock_emit_contract_kafka_event.assert_not_called() def test_validate_contract_lifecycle_activation_status_for_valid_record(): """Test _validate_contract_lifecycle_activation_status function.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.INIT ) res = logic._validate_contract_lifecycle_activation_status( mock_contract_lifecycle ) assert res is True def test_validate_contract_lifecycle_activation_status_for_invalid_record(): """Test _validate_contract_lifecycle_activation_status function.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.INACTIVE ) with pytest.raises(Exception) as excinfo: logic._validate_contract_lifecycle_activation_status( mock_contract_lifecycle ) assert str(excinfo.value) == \ ERROR_INVALID_STATUS_FOR_ACTIVATING_CONTRACT_LIFECYCLE.format( mock_contract_lifecycle.contract_lifecycle_id) @patch('abacus_contract.logic.contract_lifecycle._validate_contract_lifecycle_activation_status') # noqa: E501 @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_activate_contract_lifecycles( mock_contract_lifecycle_model, mock_validate_contract_lifecycle_activation_status, ): """Test activate_contract_lifecycles function.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.INIT ) mock_contract_lifecycle_model.get_by_id_or_error.return_value = \ mock_contract_lifecycle mock_validate_contract_lifecycle_activation_status.return_value = True mock_contract_lifecycle_model.bulk_update.return_value = True contract_lifecycle_ids = [mock_contract_lifecycle.contract_lifecycle_id] res = logic.activate_contract_lifecycles(contract_lifecycle_ids) assert res.status == 200 mock_validate_contract_lifecycle_activation_status.assert_called_once_with( mock_contract_lifecycle ) mock_contract_lifecycle_model.bulk_update.assert_called_once_with([ { 'contract_lifecycle_id': mock_contract_lifecycle.contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.ACTIVE } ]) def test_activate_contract_lifecycles_with_nonexisting_records(): """Test activate_contract_lifecycles function. Returns an error since this record doesn't exist. """ contract_lifecycle_ids = [999] res = logic.activate_contract_lifecycles(contract_lifecycle_ids) assert res.status == 400 assert res.errors['message'] == \ '400 Bad Request: ContractLifecycle 999 does not exist.' def test_activate_contract_lifecycles_with_deleted_records(): """Test activate_contract_lifecycles function. returns an error since this record was deleted. """ deleted_record = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.INIT, deleted_by='Test User' ) records_ids = [deleted_record.contract_lifecycle_id] res = logic.activate_contract_lifecycles(records_ids) assert res.status == 404 assert res.errors['message'] == \ 'Contract Lifecycle {} does not exist.'.format( deleted_record.contract_lifecycle_id) @patch('abacus_contract.logic.contract_lifecycle.current_timestamp') @patch('abacus_contract.logic.contract_lifecycle.g') def test_generate_contract_lifecycle_schedule_soft_delete_params( mock_g, mock_current_timestamp ): """Test _generate_contract_lifecycle_schedule_soft_delete_params function.""" user_id = '123' mock_g.user_details = {'id': user_id} mock_current_timestamp.return_value = '2024-09-25 07:34:37' mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create() res = logic._generate_contract_lifecycle_schedule_soft_delete_params( mock_contract_lifecycle_schedule ) assert res == { 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id, 'last_modified': '2024-09-25 07:34:37', 'last_modified_by': user_id, 'deleted_by': user_id, 'deleted_at': '2024-09-25 07:34:37' } @patch('abacus_contract.logic.contract_lifecycle.current_timestamp') @patch('abacus_contract.logic.contract_lifecycle.g') def test_generate_contract_lifecycle_soft_delete_params( mock_g, mock_current_timestamp ): """Test _generate_contract_lifecycle_soft_delete_params function.""" user_id = '123' mock_g.user_details = {'id': user_id} mock_current_timestamp.return_value = '2024-09-25 07:34:37' mock_contract_lifecycle = ContractLifecycleFactory.create() res = logic._generate_contract_lifecycle_soft_delete_params( mock_contract_lifecycle ) assert res == { 'contract_lifecycle_id': mock_contract_lifecycle.contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.INACTIVE, 'last_modified': '2024-09-25 07:34:37', 'last_modified_by': user_id, 'deleted_by': user_id, 'deleted_at': '2024-09-25 07:34:37' } @patch('abacus_contract.logic.contract_lifecycle._get_lifecycle_status') @patch('abacus_contract.logic.contract_lifecycle._get_termination_notice_deadline') @patch('abacus_contract.logic.contract_lifecycle._get_lifecycle_term_end') def test__generate_contract_lifecycle_params( mock_get_lifecycle_term_end, mock_get_termination_notice_deadline, mock_get_lifecycle_status ): """Test generating contract_lifecycle params.""" mock_contract = ContractFactory.create() mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract ) ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedule ) contract_id = mock_contract.contract_id contract_lifecycle_schedule_id = \ mock_contract_lifecycle_schedule.contract_lifecycle_schedule_id mock_get_lifecycle_term_end.return_value = None mock_get_termination_notice_deadline.return_value = None mock_get_lifecycle_status.return_value = CONTRACT_LIFECYCLE_STATUSES.ACTIVE lifecycle_term_start = '2024-09-25' res = logic._generate_contract_lifecycle_params( contract_id, lifecycle_term_start, mock_contract_lifecycle_schedule ) assert res == { 'contract_id': contract_id, 'contract_lifecycle_schedule_id': contract_lifecycle_schedule_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.ACTIVE, 'lifecycle_term_start': lifecycle_term_start, 'lifecycle_term_end': None, 'renewal_effective': None, 'termination_notice_deadline': None } @patch('abacus_contract.logic.contract_lifecycle._generate_contract_lifecycle_params') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_renew_contract_lifecycle_and_schedule_from_subsequent_schedule( mock_model, mock_generate_contract_lifecycle_params ): """Test _renew_contract_lifecycle_and_schedule_from_subsequent_schedule function. returns only contract_lifecycle. """ # noqa: E501 mock_contract = ContractFactory.create() mock_contract_lifecycle_schedules = ContractLifecycleScheduleFactory.create_batch( 2, contract=mock_contract ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0], renewal_effective='2024-09-24' ) mock_contract_lifecycle_params = { 'contract_id': mock_contract.contract_id, 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[1].contract_lifecycle_schedule_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.INIT, 'lifecycle_term_start': '2024-09-25', 'lifecycle_term_end': None, 'renewal_effective': None, 'termination_notice_deadline': None } mock_contract_lifecycle_response = ContractLifecycleFactory.create( **mock_contract_lifecycle_params ) mock_model.return_value = mock_contract_lifecycle_response mock_generate_contract_lifecycle_params.return_value = \ mock_contract_lifecycle_params res = logic._renew_contract_lifecycle_and_schedule_from_subsequent_schedule( mock_contract_lifecycle, mock_contract_lifecycle_schedules ) assert res == (mock_contract_lifecycle_response, None) @patch('abacus_contract.logic.contract_lifecycle._generate_contract_lifecycle_params') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycleSchedule') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_renew_contract_lifecycle_schedule_from_subsequent_schedule( mock_contract_lifecycle_model, mock_contract_lifecycle_schedule_model, mock_generate_contract_lifecycle_params ): """Test _renew_contract_lifecycle_and_schedule_from_subsequent_schedule function. returns contract_lifecycle and subsequent contract_lifecycle_schedule if renewal_type id "renew_periodically". """ # noqa: E501 mock_contract = ContractFactory.create() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=2, period_type='year' ) mock_contract_lifecycle_schedules = ContractLifecycleScheduleFactory.create_batch( 2, contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ) mock_contract_lifecycle = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0], renewal_effective='2024-09-24' ) mock_contract_lifecycle_params = { 'contract_id': mock_contract.contract_id, 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[1].contract_lifecycle_schedule_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.INIT, 'lifecycle_term_start': '2024-09-25', 'lifecycle_term_end': None, 'renewal_effective': None, 'termination_notice_deadline': None } mock_contract_lifecycle_schedule_params = { 'contract_id': mock_contract.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 } mock_contract_lifecycle_response = ContractLifecycleFactory.create( **mock_contract_lifecycle_params ) mock_contract_lifecycle_schedule_response = \ ContractLifecycleScheduleFactory.create( **mock_contract_lifecycle_schedule_params ) mock_contract_lifecycle_model.return_value = mock_contract_lifecycle_response mock_contract_lifecycle_schedule_model.return_value = \ mock_contract_lifecycle_schedule_response mock_generate_contract_lifecycle_params.return_value = \ mock_contract_lifecycle_params res = logic._renew_contract_lifecycle_and_schedule_from_subsequent_schedule( mock_contract_lifecycle, mock_contract_lifecycle_schedules ) assert res == \ (mock_contract_lifecycle_response, mock_contract_lifecycle_schedule_response) def test__validate_contract_lifecycle_renewal_status_and_date(): """Test _validate_contract_lifecycle_renewal_status_and_date function.""" mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.ACTIVE, renewal_effective=datetime.date.today() ) res = logic._validate_contract_lifecycle_renewal_status_and_date( mock_contract_lifecycle ) assert res is True def test__validate_contract_lifecycle_invalid_renewal_status(): """Test _validate_contract_lifecycle_renewal_status_and_date function. error is thrown if lifecycle_status is not active. """ mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.INACTIVE, renewal_effective=datetime.date.today() ) msg = ERROR_INVALID_STATUS_FOR_RENEWING_CONTRACT_LIFECYCLE.format( mock_contract_lifecycle.contract_lifecycle_id ) with pytest.raises(Exception) as excinfo: logic._validate_contract_lifecycle_renewal_status_and_date( mock_contract_lifecycle ) assert str(excinfo.value) == msg def test__validate_contract_lifecycle_invalid_renewal_effective_date(): """Test _validate_contract_lifecycle_renewal_status_and_date function. error is thrown if renewal_effective date is None. """ mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.ACTIVE, renewal_effective=None ) msg = ERROR_INVALID_RENEWAL_EFFECTIVE_DATE.format( mock_contract_lifecycle.contract_lifecycle_id ) with pytest.raises(Exception) as excinfo: logic._validate_contract_lifecycle_renewal_status_and_date( mock_contract_lifecycle ) assert str(excinfo.value) == msg def test__validate_contract_lifecycle_future_renewal_effective_date(): """Test _validate_contract_lifecycle_renewal_status_and_date function. error is thrown if renewal_effective date is in future. """ mock_contract_lifecycle = ContractLifecycleFactory.create( lifecycle_status=CONTRACT_LIFECYCLE_STATUSES.ACTIVE, renewal_effective='2100-09-01' ) msg = ERROR_INVALID_RENEWAL_EFFECTIVE_DATE.format( mock_contract_lifecycle.contract_lifecycle_id ) with pytest.raises(Exception) as excinfo: logic._validate_contract_lifecycle_renewal_status_and_date( mock_contract_lifecycle ) assert str(excinfo.value) == msg @patch('abacus_contract.logic.contract_lifecycle.db') @patch('abacus_contract.logic.contract_lifecycle' '._generate_contract_lifecycle_schedule_soft_delete_params') @patch('abacus_contract.logic.contract_lifecycle' '._generate_contract_lifecycle_soft_delete_params') @patch('abacus_contract.logic.contract_lifecycle' '._get_current_contract_lifecycle_schedule') @patch('abacus_contract.logic.contract_lifecycle' '._renew_contract_lifecycle_and_schedule_from_subsequent_schedule') @patch('abacus_contract.logic.contract_lifecycle' '._validate_contract_lifecycle_renewal_status_and_date') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycleSchedule') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_renew_contract_lifecycles( mock_contract_lifecycle_model, mock_schedule_model, mock_validation, mock_renew_lifecycle_and_schedule, mock_current_lifecycle_schedule, mock_generate_contract_lifecycle_soft_delete_params, mock_generate_schedule_soft_delete_params, mock_db ): """Test renew_contract_lifecycles function.""" mock_contract = ContractFactory.create() mock_schedule_detail = ContractLifecycleScheduleDetailFactory.create( period_interval=2, period_type='year' ) mock_contract_lifecycle_schedules = ContractLifecycleScheduleFactory.create_batch( 2, contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ) mock_contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=mock_contract, renewal_type=CONTRACT_LIFECYCLE_SCHEDULE_RENEWAL_TYPES.RENEW_PERIODICALLY, renewal_offset_detail=mock_schedule_detail ) mock_contract_lifecycle_1 = ContractLifecycleFactory.create( contract=mock_contract, contract_lifecycle_schedule=mock_contract_lifecycle_schedules[0], renewal_effective='2024-09-24' ) mock_contract_lifecycle_2 = ContractLifecycleFactory.create( contract_lifecycle_schedule=mock_contract_lifecycle_schedule, renewal_effective='2030-09-24' ) mock_contract_lifecycle_model.get_by_id_or_error.return_value = \ mock_contract_lifecycle_1 mock_validation.return_value = True mock_schedule_model.return_value = mock_contract_lifecycle_schedules mock_renew_lifecycle_and_schedule.return_value = \ (mock_contract_lifecycle_2, mock_contract_lifecycle_schedule) mock_current_lifecycle_schedule.return_value = mock_contract_lifecycle_schedules[0] mock_generate_contract_lifecycle_soft_delete_params.return_value = { 'contract_lifecycle_id': mock_contract_lifecycle_1.contract_lifecycle_id, 'lifecycle_status': CONTRACT_LIFECYCLE_STATUSES.INACTIVE, 'last_modified_by': 'Test User', 'last_modified': '2024-09-25', 'deleted_by': 'Test User', 'deleted_at': '2024-09-25', } mock_generate_schedule_soft_delete_params.return_value = { 'contract_lifecycle_schedule_id': mock_contract_lifecycle_schedules[0].contract_lifecycle_schedule_id, 'last_modified_by': 'Test User', 'last_modified': '2024-09-25', 'deleted_by': 'Test User', 'deleted_at': '2024-09-25', } res = logic.renew_contract_lifecycles( [mock_contract_lifecycle_1.contract_lifecycle_id] ) assert res.status == 200 assert mock_db.session.bulk_save_objects.call_count == 2 assert mock_db.session.bulk_update_mappings.call_count == 2 mock_db.session.commit.assert_called_once() mock_db.session.close.assert_called_once() @patch('abacus_contract.logic.contract_lifecycle' '._validate_contract_lifecycle_renewal_status_and_date') @patch('abacus_contract.logic.contract_lifecycle.ContractLifecycle') def test_renew_contract_lifecycles_error( mock_contract_lifecycle_model, mock_validation ): """Test renew_contract_lifecycles function error.""" mock_contract_lifecycle = ContractLifecycleFactory.create() mock_contract_lifecycle_model.get_by_id_or_error.return_value = \ mock_contract_lifecycle mock_validation.side_effect = Exception('Error Message') with pytest.raises(Exception) as excinfo: logic.renew_contract_lifecycles( [mock_contract_lifecycle.contract_lifecycle_id] ) assert str(excinfo.value) == 'Error Message'