"""Contract Term handler tests.""" from unittest.mock import MagicMock, patch import pytest from flask.testing import FlaskClient from owsresponse import response from abacus_contract.constants.constants import CONTRACT_TERM_TYPES from abacus_contract.tests.utils.factories import ContractFactory, ContractTermFactory @patch('abacus_contract.blueprints.contract_term.logic') def test_create_contract_term(mock_logic, fixture_client): """POST contract_term.""" contract_id = 123 json_body = {'attachments': ['1', '2'], 'term_type': CONTRACT_TERM_TYPES.LABEL} mock_logic.create_contract_term.return_value = response.Response( message='ok', status=201 ) res = fixture_client.post(f'/contract/{contract_id}/contract-term', json=json_body) assert res.status_code == 201 mock_logic.create_contract_term.assert_called_once_with( contract_id=contract_id, **json_body ) @patch('abacus_contract.blueprints.contract_term.logic') def test_update_contract_term(mock_logic, fixture_client): """PUT contract_term.""" contract_term = ContractTermFactory.create() json_body = {'attachments': ['1'], 'contract_term_name': 'test12345'} mock_logic.update_contract_term.return_value = response.Response( message='ok', status=200 ) res = fixture_client.put( f'/contract-term/{contract_term.contract_term_id}', json=json_body ) assert res.status_code == 200 mock_logic.update_contract_term.assert_called_once_with(contract_term, **json_body) @patch('abacus_contract.blueprints.contract_term.logic') def test_update_contract_term_error(mock_logic, fixture_client): """PUT contract_term for invalid contract term id.""" json_body = {'attachments': ['1']} res = fixture_client.put('/contract-term/1234', json=json_body) assert res.status_code == 404 mock_logic.update_contract_term.assert_not_called() @patch('abacus_contract.blueprints.contract_term.logic') def test_get_contract_terms_for_account_and_term_type(mock_logic, fixture_client): """GET contract terms by an account and term_type.""" account_id = 123 params = {'attachments': ['01234'], 'term_type': 'product'} mock_logic.get_contract_terms_for_account_and_term_type.return_value = ( response.Response(message='ok', status=200) ) res = fixture_client.post(f'/account/{account_id}/contract-terms/', json=params) assert res.status_code == 200 mock_logic.get_contract_terms_for_account_and_term_type.assert_called_once_with( account_id, params ) @patch('abacus_contract.blueprints.contract_term.logic') def test_soft_delete_contract_term(mock_logic, fixture_client): """DELETE contract_term and attached term conditions.""" contract = ContractFactory.create() contract_term = ContractTermFactory.create(contract=contract) mock_logic.soft_delete_contract_term_and_conditions.return_value = ( response.Response(status=204) ) res = fixture_client.delete(f'/contract-term/{contract_term.contract_term_id}/') assert res.status_code == 204 mock_logic.soft_delete_contract_term_and_conditions.assert_called_once_with( contract_term ) @pytest.mark.parametrize( ( 'profile_type', 'profile_role', 'pdp_authorize_return', 'permissions_authorize_return', 'expected_status', 'get_account_id_result', ), [ pytest.param( 'ContentProfile', 'manage_nr_ownership', None, True, 200, 1, id='Standalone check OK, Permissions check OK', ), pytest.param( 'ContentProfile', 'manage_nr_ownership', None, False, 403, 1, id='Standalone check OK, Permissions check not OK', ), pytest.param( 'Account360Profile', 'account360', True, True, 200, 1, id='Standalone check not OK, PDP check OK', ), pytest.param( 'Account360Profile', 'account360', False, None, 403, 1, id='Standalone check not OK, PDP check not OK', ), pytest.param( 'Account360Profile', 'account360', False, None, 404, None, id='Account not found', ), ], ) @patch('abacus_contract.blueprints.contract_term.ows_client') @patch('abacus_contract.blueprints.contract_term.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract_term.pdp_authorize_many_accounts') @patch('abacus_contract.blueprints.contract_term.logic') @patch('abacus_contract.blueprints.contract_term.contract_logic') def test_get_contracts_terms_by_contract( mock_contract_logic: MagicMock, mock_contract_terms_logic: MagicMock, mock_pdp_authorize_many_accounts: MagicMock, mock_permissions_authorize_many_accounts: MagicMock, mock_ows_client: MagicMock, fixture_client: FlaskClient, profile_type: str, profile_role: str, pdp_authorize_return: bool, permissions_authorize_return: bool, expected_status: int, get_account_id_result: int | None, ) -> None: """GET contract terms by contract_id.""" contract_id = 123 mock_contract_logic.get_account_id_by_contract_id.return_value = ( get_account_id_result ) mock_pdp_authorize_many_accounts.return_value = pdp_authorize_return mock_permissions_authorize_many_accounts.return_value = permissions_authorize_return mock_contract_terms_logic.get_contract_terms_by_contract.return_value = ( response.Response(message='ok', status=200) ) result = fixture_client.get( f'/contracts/{contract_id}/contract-terms', headers={ 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Profile-Type': profile_type, 'Orchard-Profile-Id': '1234', 'Orchard-Roles': profile_role, 'Orchard-Identity-Id': '1234', }, ) assert result.status_code == expected_status mock_contract_logic.get_account_id_by_contract_id.assert_called_once_with( contract_id ) # Check if the PDP check was called when the standalone check failed if profile_type == 'Account360Profile': if get_account_id_result: mock_pdp_authorize_many_accounts.assert_called_once_with( [get_account_id_result] ) else: mock_pdp_authorize_many_accounts.assert_not_called() else: mock_pdp_authorize_many_accounts.assert_not_called() if pdp_authorize_return is not False: if get_account_id_result: mock_permissions_authorize_many_accounts.assert_called_once_with( mock_ows_client, profile_type, '1234', [get_account_id_result] ) else: mock_permissions_authorize_many_accounts.assert_not_called() else: mock_permissions_authorize_many_accounts.assert_not_called() if expected_status == 200: ( mock_contract_terms_logic.get_contract_terms_by_contract.assert_called_once_with( contract_id ) ) else: mock_contract_terms_logic.get_contract_terms_by_contract.assert_not_called() @patch('abacus_contract.utils.dataloader.ows_client') @patch('abacus_contract.utils.dataloader.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract_term.logic') @patch('abacus_contract.utils.dataloader.authorization') @patch('abacus_contract.blueprints.contract_term.contract_logic') def test_get_contract_terms_by_contract_ids_dataloader( mock_contract_logic: MagicMock, mock_authorization: MagicMock, mock_logic: MagicMock, mock_permissions_authorize_many_accounts: MagicMock, mock_ows_client: MagicMock, fixture_client: FlaskClient, ) -> None: """Happy path: terms come back grouped per contract_id in request order. An id that does not resolve to an account (3) is dropped from the fetch and surfaces as ``data: None``. """ contract_ids = [1, 2, 3] mock_contract_logic.get_account_id_map_by_contract_ids.return_value = {1: 9, 2: 10} mock_authorization.pdp_authorize_many_accounts.return_value = True mock_permissions_authorize_many_accounts.return_value = True mock_logic.get_contract_term_records_by_contract_ids.return_value = [ {'contract_term_id': 100, 'contract_id': 1}, {'contract_term_id': 101, 'contract_id': 1}, {'contract_term_id': 200, 'contract_id': 2}, ] res = fixture_client.post( '/contract-terms/dataloader', json=contract_ids, headers={ 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Profile-Type': 'ContentProfile', 'Orchard-Profile-Id': '1234', 'Orchard-Roles': 'manage_nr_ownership', 'Orchard-Identity-Id': '1234', }, ) assert res.status_code == 200 mock_contract_logic.get_account_id_map_by_contract_ids.assert_called_once_with( contract_ids ) # Only the account-resolved (and authorized) ids reach the fetch. mock_logic.get_contract_term_records_by_contract_ids.assert_called_once_with([1, 2]) # Grouped per contract_id in request order; the unresolved id 3 is data: null. assert res.get_json() == [ { 'data': [ {'contract_term_id': 100, 'contract_id': 1}, {'contract_term_id': 101, 'contract_id': 1}, ] }, {'data': [{'contract_term_id': 200, 'contract_id': 2}]}, {'data': None}, ]