"""Tests for contract term logic.""" import datetime import re from decimal import Decimal from unittest.mock import MagicMock, patch import pytest from marshmallow import ValidationError from abacus_contract.constants.constants import CONTRACT_TERM_TYPES from abacus_contract.constants.error import ( ERROR_ATTACHMENT_ALREADY_EXISTS, ERROR_ATTACHMENTS_ALREADY_EXISTS_OTHER_CONTRACT, ERROR_ATTACHMENTS_CAN_NOT_BE_NULL, ERROR_BASE_TERM_ALREADY_EXISTS, ERROR_MISSING_ATTACHMENT, ERROR_MISSING_BASE_TERM, ERROR_MISSING_SCHEDULE_IDS, ) from abacus_contract.logic import contract_term as logic from abacus_contract.tests.utils.factories import ( AccountContractFactory, ContractFactory, ContractTermConditionFactory, ContractTermFactory, ContractTermScheduleFactory, ) @patch('abacus_contract.logic.contract_term._validate_contract_term') @patch('abacus_contract.logic.contract_term.models') def test_bulk_create_contract_terms_and_conditions( mock_models, mock_validation, create_mock_account ): """Test creating multiple contract terms and conditions.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) AccountContractFactory.create(account_id=1, contract=contract) term_condition_params = [ { 'conditions': {'countries': ['USA'], 'stores': [], 'transaction_types': []}, 'priority': 1, 'term_rate': 20.21, 'contract_term_condition_name': 'USA Only', } ] params = [ { 'attachments': ['88', '99'], 'contract_id': contract.contract_id, 'is_base_term': True, 'term_type': CONTRACT_TERM_TYPES.LABEL, 'contract_term_conditions': term_condition_params, } ] mock_validation.return_value = None mock_models.ContractTerm.build.return_value = contract_term mock_models.ContractTermCondition.commit_changes.return_value = True res = logic.bulk_create_contract_terms_and_conditions(params) assert res.status == 201 assert len(res.message['contract_terms']) == len(params) assert mock_validation.call_count == len(params) assert mock_models.ContractTerm.build.call_count == len(params) assert mock_models.ContractTermCondition.build.call_count == len( term_condition_params ) mock_models.ContractTermCondition.build.assert_called_once_with( contract_term_id=contract_term.contract_term_id, contract_term_condition_name=term_condition_params[0][ 'contract_term_condition_name' ], conditions=term_condition_params[0]['conditions'], priority=term_condition_params[0]['priority'], term_rate=Decimal(str(term_condition_params[0]['term_rate'])), commission=Decimal('79.79'), ) mock_models.ContractTermCondition.commit_changes.assert_called_once() @patch('abacus_contract.logic.contract_term._validate_contract_term') @patch('abacus_contract.logic.contract_term.models') def test_bulk_create_contract_terms_and_conditions_invalid_term_rate( mock_models, mock_validation, create_mock_account ): """Test creating multiple contract terms and conditions with invalid term_rate.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) AccountContractFactory.create(account_id=1, contract=contract) term_condition_params = [ { 'conditions': {'countries': ['USA'], 'stores': [], 'transaction_types': []}, 'priority': 1, 'term_rate': 180.1, } ] params = [ { 'attachments': ['88', '99'], 'contract_id': contract.contract_id, 'is_base_term': True, 'term_type': CONTRACT_TERM_TYPES.LABEL, 'contract_term_conditions': term_condition_params, } ] mock_validation.return_value = None mock_models.ContractTerm.build.return_value = contract_term mock_models.ContractTermCondition.commit_changes.return_value = True res = logic.bulk_create_contract_terms_and_conditions(params) assert res.status == 201 assert mock_models.ContractTerm.build.call_count == 0 assert mock_models.ContractTermCondition.build.call_count == 0 assert ( 'Must be greater than or equal to 0.0 and less than or equal to 100.0.' in res.message['invalid_contract_terms'][0]['error'] ) @patch('abacus_contract.logic.contract_term.models') def test_get_contract_terms_by_contract(mock_models): """Test getting contracts by contract_id.""" contract = ContractFactory.create() contract_terms = ContractTermFactory.create(contract=contract) mock_models.Contract.get_by_id_or_error.return_value = contract response = logic.get_contract_terms_by_contract(contract.contract_id) assert response.status == 200 mock_models.Contract.get_by_id_or_error.assert_called_once_with( contract_terms.contract_id ) @patch('abacus_contract.logic.contract_term.create_or_update_contract_term_schedules') @patch('abacus_contract.logic.contract_term._validate_contract_term') @patch('abacus_contract.logic.contract_term.models') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_create_contract_term_success( mock_ff_enabled, mock_models, mock_validation, mock_create_or_update_contract_term_schedules, ): """Test create contract term success.""" mock_contract = ContractFactory.create() mock_contract_term = ContractTermFactory.build(contract=mock_contract) mock_models.ContractTerm.build.return_value = mock_contract_term mock_ff_enabled.return_value = False mock_validation.return_value = None params = { 'contract_id': mock_contract_term.contract_id, 'term_type': mock_contract_term.term_type, 'attachments': mock_contract_term.attachments, 'is_base_term': mock_contract_term.is_base_term, } result = logic.create_contract_term(**params) assert result.status == 201 mock_validation.assert_called_once_with(params) mock_models.ContractTerm.build.assert_called_once_with(**params) mock_create_or_update_contract_term_schedules.assert_not_called() @patch('abacus_contract.logic.contract_term.create_or_update_contract_term_schedules') @patch('abacus_contract.logic.contract_term._validate_contract_term') @patch('abacus_contract.logic.contract_term.models') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_create_contract_term_for_contributor_schedule( mock_ff_enabled, mock_models, mock_validation, mock_create_or_update_contract_term_schedules, ): """Test create contract term for contributor_schedule term type.""" mock_contract = ContractFactory.create() mock_contract_term = ContractTermFactory.build( contract=mock_contract, term_type=CONTRACT_TERM_TYPES.CONTRIBUTOR_SCHEDULE, attachments=None, ) mock_models.ContractTerm.build.return_value = mock_contract_term mock_ff_enabled.return_value = False mock_validation.return_value = None params = { 'contract_id': mock_contract_term.contract_id, 'term_type': mock_contract_term.term_type, 'schedule_ids': ['2', '3'], 'is_base_term': mock_contract_term.is_base_term, } result = logic.create_contract_term(**params) assert result.status == 201 del params['schedule_ids'] params.update({'attachments': None, 'attachments_relations': None}) mock_validation.assert_called_once_with(params) mock_models.ContractTerm.build.assert_called_once_with(**params) mock_create_or_update_contract_term_schedules.assert_called_once_with( ['2', '3'], mock_contract_term.contract_term_id ) @patch('abacus_contract.logic.contract_term._validate_contract_term') @patch('abacus_contract.logic.contract_term.models') def test_create_contract_term_error(mock_models, mock_validation): """Test create contract term failure.""" mock_validation.side_effect = ValidationError('NOPE') params = {} result = logic.create_contract_term(**params) assert result.status == 400 assert result.errors['message'] == 'NOPE' mock_validation.assert_called_once_with(params) mock_models.ContractTerm.create.assert_not_called() @patch('abacus_contract.logic.contract_term.create_or_update_contract_term_schedules') @patch('abacus_contract.logic.contract_term.models') def test_update_contract_term(models_mock, mock_contract_term_schedule_logic): """Test update contract term logic.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) params = dict(attachments=['1'], contract_term_name='test12345') result = logic.update_contract_term(contract_term, **params) assert result.status == 200 models_mock.ContractTerm.commit_changes.assert_called_once() mock_contract_term_schedule_logic.assert_not_called() @patch('abacus_contract.logic.contract_term.create_or_update_contract_term_schedules') @patch('abacus_contract.logic.contract_term.models') def test_update_contract_term_for_contributor_schedule_term_type( models_mock, mock_contract_term_schedule_logic, create_mock_schedule ): """Test update contract term logic for contributor_schedule term_type.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create( contract=contract, term_type=CONTRACT_TERM_TYPES.CONTRIBUTOR_SCHEDULE ) contract_term.update_attributes = MagicMock() params = dict( schedule_ids=['1', '2'], term_type=CONTRACT_TERM_TYPES.CONTRIBUTOR_SCHEDULE ) result = logic.update_contract_term(contract_term, **params) assert result.status == 200 models_mock.ContractTerm.commit_changes.assert_called_once() mock_contract_term_schedule_logic.assert_called_once_with( ['1', '2'], contract_term.contract_term_id ) @patch('abacus_contract.logic.contract_term.create_or_update_contract_term_schedules') @patch('abacus_contract.logic.contract_term.models') def test_update_contract_term_for_contributor_schedule_term_type_fails( models_mock, mock_contract_term_schedule_logic ): """Test update contract term logic fails. if different term_type is passed but schedule_ids are missing. """ contract = ContractFactory.create() contract_term = ContractTermFactory.create( contract=contract, term_type=CONTRACT_TERM_TYPES.CONTRIBUTOR_SCHEDULE ) contract_term.update_attributes = MagicMock() params = dict(term_type=CONTRACT_TERM_TYPES.CONTRIBUTION_SCHEDULE) result = logic.update_contract_term(contract_term, **params) assert result.status == 400 assert result.errors['message'] == ERROR_MISSING_SCHEDULE_IDS models_mock.ContractTerm.commit_changes.assert_not_called() contract_term.update_attributes.assert_not_called() mock_contract_term_schedule_logic.assert_not_called() @patch('abacus_contract.logic.contract_term.models') def test_validate_contract_term_base_term_already_exists_feature_flag_enabled( mock_models, ): """Test duplicate base terms is valid when feature flag is enabled.""" contract = ContractFactory.create() base_term = ContractTermFactory.create(contract=contract, is_base_term=True) mock_models.Contract.get_by_id_or_error.return_value = contract mock_models.ContractTerm.get_contract_base_term.return_value = base_term term_condition_params = [ { 'conditions': {'countries': ['USA'], 'stores': [], 'transaction_types': []}, 'priority': 1, 'term_rate': 180.1, } ] params = { 'attachments': ['88', '99'], 'contract_id': contract.contract_id, 'is_base_term': True, 'term_type': CONTRACT_TERM_TYPES.LABEL, 'contract_term_conditions': term_condition_params, } res = logic._validate_contract_term(params) assert res is None @patch('abacus_contract.logic.contract_term.models') def test_validate_contract_term_base_term_missing_feature_flag_enabled(mock_models): """Test missing base term is valid when feature flag is enabled.""" contract = ContractFactory.create() mock_models.Contract.get_by_id_or_error.return_value = contract mock_models.ContractTerm.get_contract_base_term.return_value = None term_condition_params = [ { 'conditions': {'countries': ['USA'], 'stores': [], 'transaction_types': []}, 'priority': 1, 'term_rate': 180.1, } ] params = { 'attachments': ['88', '99'], 'contract_id': contract.contract_id, 'is_base_term': False, 'term_type': CONTRACT_TERM_TYPES.LABEL, 'contract_term_conditions': term_condition_params, } res = logic._validate_contract_term(params) assert not res @patch('abacus_contract.logic.contract_term.models') def test_validate_contract_term_attachments_missing(mock_models): """Test validation raises error when attachments list is empty.""" contract = ContractFactory.create() mock_models.Contract.get_by_id_or_error.return_value = contract mock_models.ContractTerm.get_contract_base_term.return_value = None params = { 'attachments': [], 'contract_id': contract.contract_id, 'is_base_term': True, } with pytest.raises(ValidationError, match=ERROR_MISSING_ATTACHMENT): logic._validate_contract_term(params) @pytest.mark.parametrize( 'term_type, other_attachments, attachments, error_attachments', [ (CONTRACT_TERM_TYPES.PRODUCT, ['111', '222'], ['777', '888', '111'], ['111']), (CONTRACT_TERM_TYPES.TRACK, ['222', '111'], ['111', '333', '555'], ['111']), ], ) def test_validate_attachment_uniqueness( term_type, other_attachments, attachments, error_attachments, create_mock_account ): """Test validation raises error for attachments to other contract.""" account_id = 1 other_contract = ContractFactory.create(contract_id=1) AccountContractFactory.create(account_id=account_id, contract=other_contract) ContractTermFactory.create( attachments=other_attachments, contract=other_contract, term_type=term_type ) contract = ContractFactory.create(contract_id=2) AccountContractFactory.create(account_id=account_id, contract=contract) expected_error = ERROR_ATTACHMENTS_ALREADY_EXISTS_OTHER_CONTRACT.format( attachments=error_attachments, term_type=term_type.capitalize() ) with pytest.raises(ValidationError, match=re.escape(expected_error)): logic._validate_attachment_uniqueness(contract, attachments, term_type) def test_build_contract_term_row(): """Test formatting a stringified tsv row for contract term export.""" contract_term_condition = ContractTermConditionFactory.create() contract_term = contract_term_condition.contract_term attachments = ','.join(contract_term.attachments) conditions = str( { key: value for (key, value) in contract_term_condition.conditions.items() if len(value) } ) res = logic._build_contract_term_row( contract_term, contract_term_condition, attachments, conditions ) assert isinstance(res, str) assert res.endswith('\n') assert len(res.split('\t')) == len(logic.CONTRACT_TERMS_SNAPSHOT_HEADER) def test_format_attachments(): """Test formatting attachments for export.""" contract_term = ContractTermFactory.create() res = logic._format_attachments(contract_term) assert res == ','.join(contract_term.attachments) def test_format_attachments_track_terms(): """Test formatting track attachments for export.""" isrcs = ['ISRC123', 'ISRC456'] contract_term = ContractTermFactory.create( attachments=isrcs, term_type=CONTRACT_TERM_TYPES.TRACK ) res = logic._format_attachments(contract_term) assert res == ','.join(isrcs) def test_format_term_conditions(): """Test formatting contract_term_conditions as stringified json for export.""" conditions = {'countries': ['USA'], 'stores': [], 'transaction_types': []} contract_term_condition = ContractTermConditionFactory.create(conditions=conditions) res = logic._format_term_conditions(contract_term_condition) expected = '{"countries": ["USA"]}' assert res == expected def test_format_term_conditions_no_conditions(): """Test formatting contract_term_conditions when there are no exceptions.""" conditions = {'countries': [], 'stores': [], 'transaction_types': []} contract_term_condition = ContractTermConditionFactory.create(conditions=conditions) res = logic._format_term_conditions(contract_term_condition) assert res == '' def test_validate_conflicting_attachments(): """Test validation raises error when exceptions are duplicated. And doesn't raises any errors if there are no duplications. """ contract = ContractFactory.create() attachments = ['1111', '2222', '3333'] ContractTermFactory.create( attachments=attachments, contract=contract, term_type=CONTRACT_TERM_TYPES.PRODUCT, ) expected_error = ERROR_ATTACHMENT_ALREADY_EXISTS.format(attachment='2222, 3333') with pytest.raises(ValidationError) as err: logic._validate_conflicting_attachments( contract_terms=contract.contract_terms, term_type=CONTRACT_TERM_TYPES.PRODUCT, attachments=['5555', '2222', '4444', '3333'], ) assert err.value.messages[0] == expected_error assert ( logic._validate_conflicting_attachments( contract_terms=contract.contract_terms, term_type=CONTRACT_TERM_TYPES.PRODUCT, attachments=['4444', '5555'], ) is None ) @patch('abacus_contract.logic.contract_term.models') def test_get_contract_terms_for_account_and_term_type(mock_models): """Test getting contract terms for a specified account and term_type.""" params = {'term_type': CONTRACT_TERM_TYPES.TRACK, 'attachments': ['128776']} contract_term = ContractTermFactory.create(term_type=CONTRACT_TERM_TYPES.TRACK) mock_models.ContractTerm.get_contract_terms_by_account_and_term_type.return_value = [ contract_term ] response = logic.get_contract_terms_for_account_and_term_type(1234, params) assert response.status == 200 mock_models.ContractTerm.get_contract_terms_by_account_and_term_type.assert_called_once_with( 1234, **params ) @patch('abacus_contract.logic.contract_term.models') def test_get_contract_terms_for_account_and_term_type_error(mock_models): """Test getting contract terms for a specified account and term_type.""" params = {'term_type': CONTRACT_TERM_TYPES.TRACK, 'attachments': ['128776']} ContractTermFactory.create(term_type=CONTRACT_TERM_TYPES.TRACK) mock_models.ContractTerm.get_contract_terms_by_account_and_term_type.return_value = [] response = logic.get_contract_terms_for_account_and_term_type(1234, params) assert response.status == 404 def test_soft_delete_distribution_contract_term_and_conditions(): """Test to soft delete distribution contract term and conditions.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) contract_term_condition_1 = ContractTermConditionFactory.create( contract_term=contract_term ) contract_term_condition_2 = ContractTermConditionFactory.create( contract_term=contract_term ) contract_term.conditions = [contract_term_condition_1, contract_term_condition_2] contract_term._soft_delete = MagicMock() for condition in contract_term.conditions: condition._soft_delete = MagicMock() response = logic.soft_delete_contract_term_and_conditions(contract_term) assert response.status == 204 contract_term._soft_delete.assert_called_once() contract_term_condition_1._soft_delete.assert_called_once() contract_term_condition_2._soft_delete.assert_called_once() def test_soft_delete_nr_contract_term_and_schedules_and_conditions( create_mock_schedule, ): """Test to soft delete NR contract term, term schedules and conditions.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) contract_term_condition_1 = ContractTermConditionFactory.create( contract_term=contract_term ) contract_term_condition_2 = ContractTermConditionFactory.create( contract_term=contract_term ) contract_term_schedule = ContractTermScheduleFactory.create( contract_term=contract_term, schedule_id=2 ) contract_term.conditions = [contract_term_condition_1, contract_term_condition_2] contract_term.contract_term_schedule = [contract_term_schedule] contract_term._soft_delete = MagicMock() contract_term_schedule._soft_delete = MagicMock() for condition in contract_term.conditions: condition._soft_delete = MagicMock() response = logic.soft_delete_contract_term_and_conditions(contract_term) assert response.status == 204 contract_term._soft_delete.assert_called_once() contract_term_schedule._soft_delete.assert_called_once() contract_term_condition_1._soft_delete.assert_called_once() contract_term_condition_2._soft_delete.assert_called_once() def test_get_account_id_by_contract_term_id(create_mock_account) -> None: """Test getting account_id by contract_term_id.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) AccountContractFactory.create(account_id=2, contract=contract) assert logic.get_account_id_by_contract_term_id(contract_term.contract_term_id) == 2 def test_get_account_id_by_contract_term_id_not_found() -> None: """Test getting account_id by contract_term_id when account_id is not found.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) assert ( logic.get_account_id_by_contract_term_id(contract_term.contract_term_id) is None ) def test_get_account_ids_by_contract_term_ids(create_mock_account) -> None: """Test batch-resolving account_id for a set of contract_term_ids in one query.""" contract_1 = ContractFactory.create() contract_2 = ContractFactory.create() contract_term_1 = ContractTermFactory.create(contract=contract_1) contract_term_2 = ContractTermFactory.create(contract=contract_2) contract_term_no_account = ContractTermFactory.create() AccountContractFactory.create(account_id=2, contract=contract_1) AccountContractFactory.create(account_id=3, contract=contract_2) result = logic.get_account_ids_by_contract_term_ids( [ contract_term_1.contract_term_id, contract_term_2.contract_term_id, contract_term_no_account.contract_term_id, ] ) assert result == { contract_term_1.contract_term_id: 2, contract_term_2.contract_term_id: 3, } def test_get_account_ids_by_contract_term_ids_empty() -> None: """Test batch-resolving account ids for an empty id list.""" assert logic.get_account_ids_by_contract_term_ids([]) == {} @patch('abacus_contract.logic.contract_term.models') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_validate_contract_term_attachments_missing_ff_enabled( mock_is_feature_enabled, mock_models ): """Test validation raises error when attachments list is empty.""" mock_is_feature_enabled.return_value = True contract = ContractFactory.create() mock_models.Contract.get_by_id_or_error.return_value = contract mock_models.ContractTerm.get_contract_base_term.return_value = None params = { 'attachments': [], 'contract_id': contract.contract_id, 'is_base_term': True, } with pytest.raises(ValidationError, match=ERROR_MISSING_ATTACHMENT): logic._validate_contract_term(params) @patch('abacus_contract.logic.contract_term.models') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_validate_contract_term_null_attachments_ff_enabled( mock_is_feature_enabled, mock_models ): """Test returns true when attachments is null.""" mock_is_feature_enabled.return_value = True contract = ContractFactory.create() mock_models.Contract.get_by_id_or_error.return_value = contract mock_models.ContractTerm.get_contract_base_term.return_value = None params = { 'attachments': None, 'contract_id': contract.contract_id, 'is_base_term': True, } result = logic._validate_contract_term(params) assert result is True @patch('abacus_contract.logic.contract_term.models') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_validate_contract_term_null_attachments_ff_disabled( mock_is_feature_enabled, mock_models ): """Test validation raises error when attachments are null.""" mock_is_feature_enabled.return_value = False contract = ContractFactory.create() mock_models.Contract.get_by_id_or_error.return_value = contract mock_models.ContractTerm.get_contract_base_term.return_value = None params = { 'attachments': None, 'contract_id': contract.contract_id, 'is_base_term': True, 'term_type': CONTRACT_TERM_TYPES.LABEL, } with pytest.raises(ValidationError, match=ERROR_ATTACHMENTS_CAN_NOT_BE_NULL): logic._validate_contract_term(params) @patch('abacus_contract.logic.contract_term.models') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_validate_label_contract_term_null_attachments_ff_disabled( mock_is_feature_enabled, mock_models ): """Test validation raises error when attachments are null for label terms.""" mock_is_feature_enabled.return_value = True contract = ContractFactory.create() mock_models.Contract.get_by_id_or_error.return_value = contract mock_models.ContractTerm.get_contract_base_term.return_value = None params = { 'attachments': None, 'contract_id': contract.contract_id, 'is_base_term': True, 'term_type': CONTRACT_TERM_TYPES.LABEL, } with pytest.raises(ValidationError, match=ERROR_ATTACHMENTS_CAN_NOT_BE_NULL): logic._validate_contract_term(params) @patch('abacus_contract.logic.contract_term.create_or_update_contract_term_schedules') @patch('abacus_contract.logic.contract_term.models') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_update_contract_term_with_null_attachments_ff_disabled( mock_is_feature_enabled, models_mock, mock_contract_term_schedule_logic ): """Test validation raises error when try to update the contract term with null attachments.""" mock_is_feature_enabled.return_value = False contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) params = dict(attachments=None, contract_term_name='test12345') result = logic.update_contract_term(contract_term, **params) assert result.status == 400 assert result.errors['message'] == ERROR_ATTACHMENTS_CAN_NOT_BE_NULL models_mock.ContractTerm.commit_changes.assert_not_called() mock_contract_term_schedule_logic.assert_not_called() @patch('abacus_contract.logic.contract_term.create_or_update_contract_term_schedules') @patch('abacus_contract.logic.contract_term.models') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_update_label_contract_term_with_null_attachments_ff_enabled( mock_is_feature_enabled, models_mock, mock_contract_term_schedule_logic ): """Test validation raises error when try to update the label contract term with null attachments.""" mock_is_feature_enabled.return_value = True contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) params = dict(attachments=None, contract_term_name='test12345') result = logic.update_contract_term(contract_term, **params) assert result.status == 400 assert result.errors['message'] == ERROR_ATTACHMENTS_CAN_NOT_BE_NULL models_mock.ContractTerm.commit_changes.assert_not_called() mock_contract_term_schedule_logic.assert_not_called() @patch('abacus_contract.logic.contract_term.create_or_update_contract_term_schedules') @patch('abacus_contract.logic.contract_term.models') @patch('abacus_contract.logic.contract_term.null') @patch( 'abacus_contract.logic.contract_term.is_abacus_terms_with_no_tracks_and_products_enabled' ) def test_update_contract_term_with_null_attachments_ff_enabled( mock_is_feature_enabled, mock_sqlalchemy_null, models_mock, mock_contract_term_schedule_logic, ): """Test to update contract term with null attachments when ff is enabled.""" mock_is_feature_enabled.return_value = True mock_sqlalchemy_null.return_value = None contract = ContractFactory.create() contract_term = ContractTermFactory.create( contract=contract, term_type=CONTRACT_TERM_TYPES.TRACK ) params = dict(attachments=None, contract_term_name='test12345') result = logic.update_contract_term(contract_term, **params) assert result.status == 200 models_mock.ContractTerm.commit_changes.assert_called_once() mock_contract_term_schedule_logic.assert_not_called() def test_get_contract_term_records_by_contract_ids(): """Serialize non-deleted contract terms for authorized contract ids as flat records.""" contract = ContractFactory.create() other_contract = ContractFactory.create() term_1 = ContractTermFactory.create(contract=contract) term_2 = ContractTermFactory.create(contract=other_contract) # A soft-deleted term (deleted_at set) must be excluded, matching the single # endpoint semantics. ContractTermFactory.create( contract=contract, deleted_at=datetime.datetime(2020, 1, 1) ) res = logic.get_contract_term_records_by_contract_ids( [contract.contract_id, other_contract.contract_id] ) # A flat list of records, not the {'data': ...} dataload wrapping. returned_term_ids = {record['contract_term_id'] for record in res} assert returned_term_ids == {term_1.contract_term_id, term_2.contract_term_id} assert all(record.get('contract_id') is not None for record in res) def test_get_contract_term_records_by_contract_ids_empty_list_skips_query(): """No authorized ids: returns an empty list without querying. The dataloader helper only ever passes the authorized ids through, so an id that never resolved to an account must never reach this function's query. """ with patch('abacus_contract.logic.contract_term.models') as mock_models: res = logic.get_contract_term_records_by_contract_ids([]) assert res == [] mock_models.ContractTerm.get_non_deleted_by_contract_ids.assert_not_called()