"""Contract Term Condition handler tests.""" from unittest.mock import MagicMock, patch import pytest from flask import testing as flask_testing from owsresponse import response from core.config import Config @patch('abacus_contract.blueprints.contract_term_condition.logic') def test_bulk_create_contract_term_conditions(mock_logic, fixture_client): """Test POST /contract-term//conditions.""" contract_term_id = 33 mock_logic.create_contract_term_conditions.return_value = response.Response( message='ok', status=201 ) json_body = { 'conditions': {'countries': ['USA', 'IND'], 'stores': ['Spotify']}, 'priority': 1, 'term_rate': 25.00, } res = fixture_client.post( f'/contract-term/{contract_term_id}/conditions', json=[json_body] ) assert res.status_code == 201 mock_logic.create_contract_term_conditions.assert_called_once_with( contract_term_id, [json_body] ) @patch('abacus_contract.blueprints.contract_term_condition.logic') def test_bulk_soft_delete_contract_term_conditions(mock_logic, fixture_client): """Test PUT /contract-term-conditions/soft-delete.""" mock_logic.soft_delete_contract_term_conditions.return_value = response.Response( message='ok', status=200 ) json_body = [1, 3, 5, 7] res = fixture_client.put('/contract-term-conditions/soft-delete', json=json_body) assert res.status_code == 200 mock_logic.soft_delete_contract_term_conditions.assert_called_once_with(json_body) @patch('abacus_contract.blueprints.contract_term_condition.validated_request_body') @patch('abacus_contract.blueprints.contract_term_condition.logic') def test_bulk_update_contract_term_conditions( mock_logic, mock_validate, fixture_client ): """Test PUT /contract-term//conditions.""" contract_term_id = 123 mock_logic.update_contract_term_conditions.return_value = response.Response( message='ok', status=200 ) json_body = { 'contract_term_condition_id': 456, 'conditions': {'countries': ['USA', 'IND'], 'stores': ['Spotify']}, 'priority': 1, 'term_rate': 25.00, } mock_validate.return_value = [json_body] res = fixture_client.put( f'/contract-term/{contract_term_id}/conditions', json=[json_body] ) assert res.status_code == 200 mock_logic.update_contract_term_conditions.assert_called_once_with( contract_term_id, [json_body] ) @pytest.mark.parametrize( [ 'get_account_id_result', 'profile_type', 'profile_role', 'pdp_auth_result', 'permissions_result', 'expected_status', ], [ pytest.param( 1, 'ContentProfile', 'manage_nr_ownership', None, True, 200, id='Standalone check OK, Permissions check OK', ), pytest.param( 1, 'ContentProfile', 'manage_nr_ownership', None, False, 403, id='Standalone check OK, Permissions check not OK', ), pytest.param( 1, 'Account360Profile', 'account360', True, True, 200, id='Standalone check not OK, PDP check OK', ), pytest.param( 1, 'Account360Profile', 'account360', False, None, 403, id='Standalone check not OK, PDP check not OK', ), pytest.param( None, 'Account360Profile', 'account360', None, None, 404, id='Account not found', ), ], ) @patch('abacus_contract.blueprints.contract_term_condition.ows_client') @patch( 'abacus_contract.blueprints.contract_term_condition.permissions_authorize_many_accounts' ) @patch('abacus_contract.blueprints.contract_term_condition.logic') @patch('abacus_contract.blueprints.contract_term_condition.authorization') @patch('abacus_contract.blueprints.contract_term_condition.contract_term_logic') def test_get_conditions_by_term( mock_contract_term_logic: MagicMock, mock_authorization: MagicMock, mock_logic: MagicMock, mock_permissions_authorize_many_accounts: MagicMock, mock_ows_client: MagicMock, get_account_id_result: int | None, profile_type: str, profile_role: str, pdp_auth_result: bool | None, permissions_result: bool | None, expected_status: int, fixture_client: flask_testing.FlaskClient, ) -> None: """Test GET /contract-term//conditions authorization.""" contract_term_id = 33 mock_contract_term_logic.get_account_id_by_contract_term_id.return_value = ( get_account_id_result ) mock_authorization.pdp_authorize_many_accounts.return_value = pdp_auth_result mock_permissions_authorize_many_accounts.return_value = permissions_result mock_logic.get_conditions_by_term_id.return_value = response.Response() res = fixture_client.get( f'/contract-term/{contract_term_id}/conditions', headers={ 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Profile-Type': profile_type, 'Orchard-Profile-Id': '1234', 'Orchard-Roles': profile_role, 'Orchard-Identity-Id': '1234', }, ) assert res.status_code == expected_status mock_contract_term_logic.get_account_id_by_contract_term_id.assert_called_once_with( contract_term_id ) # Check if the PDP check was called when the standalone check failed if profile_type == 'Account360Profile': if get_account_id_result: mock_authorization.pdp_authorize_many_accounts.assert_called_once_with( [get_account_id_result] ) else: mock_authorization.pdp_authorize_many_accounts.assert_not_called() else: mock_authorization.pdp_authorize_many_accounts.assert_not_called() if pdp_auth_result 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_logic.get_conditions_by_term_id.assert_called_once_with(contract_term_id) else: mock_logic.get_conditions_by_term_id.assert_not_called() @patch('abacus_contract.blueprints.contract_term_condition.contract_term_logic') def test_get_conditions_by_term_ids_dataloader_over_cap( mock_contract_term_logic: MagicMock, fixture_client: flask_testing.FlaskClient, ) -> None: """Over-cap batches are rejected with a 400 before any lookup happens.""" ids = list(range(Config.OWS_BATCH_LIMIT + 1)) res = fixture_client.post('/contract-term-conditions/dataloader', json=ids) assert res.status_code == 400 mock_contract_term_logic.get_account_ids_by_contract_term_ids.assert_not_called() @pytest.mark.parametrize( [ 'profile_type', 'profile_role', 'pdp_auth_result', 'permissions_result', 'expected_status', ], [ pytest.param( 'ContentProfile', 'manage_nr_ownership', None, True, 200, id='Standalone check OK, Permissions check OK', ), pytest.param( 'ContentProfile', 'manage_nr_ownership', None, False, 403, id='Standalone check OK, Permissions check not OK', ), pytest.param( 'Account360Profile', 'account360', True, True, 200, id='Standalone check not OK, PDP check OK', ), pytest.param( 'Account360Profile', 'account360', False, None, 403, id='Standalone check not OK, PDP check not OK', ), ], ) @patch('abacus_contract.utils.dataloader.ows_client') @patch('abacus_contract.utils.dataloader.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract_term_condition.logic') @patch('abacus_contract.utils.dataloader.authorization') @patch('abacus_contract.blueprints.contract_term_condition.contract_term_logic') def test_get_conditions_by_term_ids_dataloader( mock_contract_term_logic: MagicMock, mock_authorization: MagicMock, mock_logic: MagicMock, mock_permissions_authorize_many_accounts: MagicMock, mock_ows_client: MagicMock, profile_type: str, profile_role: str, pdp_auth_result: bool | None, permissions_result: bool | None, expected_status: int, fixture_client: flask_testing.FlaskClient, ) -> None: """Batch authorization mirrors the single-term handler, over the whole account set.""" contract_term_ids = [1, 2] account_ids_by_term_id = {1: 9, 2: 10} mock_contract_term_logic.get_account_ids_by_contract_term_ids.return_value = ( account_ids_by_term_id ) mock_authorization.pdp_authorize_many_accounts.return_value = pdp_auth_result mock_permissions_authorize_many_accounts.return_value = permissions_result mock_logic.get_condition_records_by_term_ids.return_value = [] res = fixture_client.post( '/contract-term-conditions/dataloader', json=contract_term_ids, headers={ 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Profile-Type': profile_type, 'Orchard-Profile-Id': '1234', 'Orchard-Roles': profile_role, 'Orchard-Identity-Id': '1234', }, ) assert res.status_code == expected_status mock_contract_term_logic.get_account_ids_by_contract_term_ids.assert_called_once_with( contract_term_ids ) if profile_type == 'Account360Profile': mock_authorization.pdp_authorize_many_accounts.assert_called_once() assert sorted( mock_authorization.pdp_authorize_many_accounts.call_args.args[0] ) == [9, 10] else: mock_authorization.pdp_authorize_many_accounts.assert_not_called() if pdp_auth_result is not False: mock_permissions_authorize_many_accounts.assert_called_once() perm_args = mock_permissions_authorize_many_accounts.call_args.args assert perm_args[:3] == (mock_ows_client, profile_type, '1234') assert sorted(perm_args[3]) == [9, 10] else: mock_permissions_authorize_many_accounts.assert_not_called() if expected_status == 200: mock_logic.get_condition_records_by_term_ids.assert_called_once_with( list(account_ids_by_term_id.keys()) ) else: mock_logic.get_condition_records_by_term_ids.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_condition.logic') @patch('abacus_contract.utils.dataloader.authorization') @patch('abacus_contract.blueprints.contract_term_condition.contract_term_logic') def test_get_conditions_by_term_ids_dataloader_no_accounts_found( mock_contract_term_logic: MagicMock, mock_authorization: MagicMock, mock_logic: MagicMock, mock_permissions_authorize_many_accounts: MagicMock, mock_ows_client: MagicMock, fixture_client: flask_testing.FlaskClient, ) -> None: """None of the requested ids resolve to an account: skip auth, return empty lists.""" mock_contract_term_logic.get_account_ids_by_contract_term_ids.return_value = {} mock_logic.get_condition_records_by_term_ids.return_value = [] res = fixture_client.post('/contract-term-conditions/dataloader', json=[1, 2]) assert res.status_code == 200 mock_authorization.pdp_authorize_many_accounts.assert_not_called() mock_permissions_authorize_many_accounts.assert_not_called() mock_logic.get_condition_records_by_term_ids.assert_called_once_with([]) @patch('abacus_contract.utils.dataloader.ows_client') @patch('abacus_contract.utils.dataloader.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract_term_condition.logic') @patch('abacus_contract.utils.dataloader.authorization') @patch('abacus_contract.blueprints.contract_term_condition.contract_term_logic') def test_get_conditions_by_term_ids_dataloader_heterogeneous_forbidden( mock_contract_term_logic: MagicMock, mock_authorization: MagicMock, mock_logic: MagicMock, mock_permissions_authorize_many_accounts: MagicMock, mock_ows_client: MagicMock, fixture_client: flask_testing.FlaskClient, ) -> None: """A batch spanning two accounts where only one is denied returns 403 for the whole batch. permissions_authorize_many_accounts is mocked at its true call boundary (called once with the full account set), so this covers a "some accounts forbidden" outcome rather than a single wholesale allow/deny bool. """ contract_term_ids = [1, 2] account_ids_by_term_id = {1: 9, 2: 10} mock_contract_term_logic.get_account_ids_by_contract_term_ids.return_value = ( account_ids_by_term_id ) def fake_permissions_authorize_many_accounts( client, profile_type, profile_id, account_ids ): # Account 9 is permitted, account 10 is denied: the batch as a whole # must be forbidden, not partially returned. return set(account_ids) <= {9} mock_permissions_authorize_many_accounts.side_effect = ( fake_permissions_authorize_many_accounts ) res = fixture_client.post( '/contract-term-conditions/dataloader', json=contract_term_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 == 403 mock_permissions_authorize_many_accounts.assert_called_once() perm_args = mock_permissions_authorize_many_accounts.call_args.args assert perm_args[:3] == (mock_ows_client, 'ContentProfile', '1234') assert sorted(perm_args[3]) == [9, 10] mock_logic.get_condition_records_by_term_ids.assert_not_called()