"""Tests for contract handlers.""" import datetime from unittest.mock import MagicMock, patch from flask.testing import FlaskClient from owsresponse import response from owsresponse.adaptors.flask import flaskify import pytest from abacus_contract.constants import constants, error from abacus_contract.utils.request import get_optional_numeric_list_from_params from tests.utils.factories import ( AccountContractFactory, ContractFactory, ContractLifecycleFactory, ContractLifecycleScheduleFactory ) @patch('abacus_contract.blueprints.contract.logic') def test_get_contracts_by_ids(mock_logic, fixture_client, fresh_db_handlers): """Get contracts by a list of contract_ids via POST /contracts/.""" contracts = ContractFactory.create_batch(3) contract_ids = [c.contract_id for c in contracts] mock_logic.get_contracts_by_ids.return_value = contracts res = fixture_client.post('/contracts/', json=contract_ids) assert res mock_logic.get_contracts_by_ids.assert_called_once_with(contract_ids) @patch('abacus_contract.blueprints.contract.logic') def test_get_contracts_by_account(mock_logic, fixture_client): """GET contracts by account_id.""" account_id = 1 fixture_client.get(f'/contracts/account/{account_id}') mock_logic.get_contracts_by_account.assert_called_once_with([account_id]) @patch('abacus_contract.utils.request.request') def test_get_contract_ids_param(mock_req): """Test getting the contract_ids param.""" mock_req.json = None actual = get_optional_numeric_list_from_params(optional=True) assert actual is None mock_req.json = None with pytest.raises(ValueError): get_optional_numeric_list_from_params(optional=False) mock_req.json = [1] actual = get_optional_numeric_list_from_params() assert actual == [1] mock_req.json = [1, 2] actual = get_optional_numeric_list_from_params() assert set(actual) == set([1, 2]) mock_req.json = ['a', 2] with pytest.raises(ValueError): get_optional_numeric_list_from_params() @patch('abacus_contract.blueprints.contract.logic') def test_terminate_contract(mock_logic, fixture_client): """Test to terminate contract.""" contract = ContractFactory.create() contract_lifecycle_schedule = ContractLifecycleScheduleFactory.create( contract=contract, ) ContractLifecycleFactory.create( contract=contract, contract_lifecycle_schedule=contract_lifecycle_schedule, lifecycle_status=constants.CONTRACT_LIFECYCLE_STATUSES.ACTIVE, ) json_body = { 'termination_effective': '2024-08-26', 'termination_notice_received': None, } mock_logic.terminate_contract.return_value = \ response.Response(message='OK', status=200) res = fixture_client.put(f'/contract/{contract.contract_id}/terminate', json=json_body) # noqa: E501 assert res.status_code == 200 mock_logic.terminate_contract.assert_called_once_with( contract.contract_id, termination_effective=datetime.date(2024, 8, 26), termination_notice_received=None, ) @patch('abacus_contract.blueprints.contract.logic') def test_get_contracts_vat_info(mock_logic, fixture_client): """GET contract vat info by a list of contract_ids via POST /contracts/vat-info.""" contract_ids = [1, 2] mock_logic.get_vat_info_by_contract_ids.return_value = response.Response( message='OK', status=200 ) res = fixture_client.post('/contracts/vat-info', json=contract_ids) assert res mock_logic.get_vat_info_by_contract_ids.assert_called_once_with(contract_ids) @patch('abacus_contract.blueprints.contract.logic') def test_get_contracts_vat_info_no_body_error(mock_logic, fixture_client): """Test getting 'Invalid request body' error requesting /contracts/vat-info. Action: Send POST request without body. """ res = fixture_client.post('/contracts/vat-info', json=[]) assert res.json == { 'code': 'error', 'message': error.ERROR_INVALID_BODY.format( expected_body_type='List of Contract IDs' ) } mock_logic.get_vat_info_by_contract_ids.assert_not_called() @patch('abacus_contract.blueprints.contract.logic') def test_get_contracts_vat_info_bad_input_format_error(mock_logic, fixture_client): """Test getting 'IDs must be integers' error requesting /contracts/vat-info. Action: Send list with non-integer in it in POST body. """ res = fixture_client.post('/contracts/vat-info', json=[1, 2, 'test']) assert res.json == { 'code': 'error', 'message': error.ERROR_INVALID_IDS.format(object='Contract') } mock_logic.get_vat_info_by_contract_ids.assert_not_called() @patch('abacus_contract.blueprints.contract.logic') def test_update_contract(mock_logic, fixture_client): """Test to update contract.""" contract = ContractFactory.create() mock_logic.update_contract.return_value = \ response.Response(message='ok', status=200) mock_put_body = { 'contract_name': 'Test Name' } res = fixture_client.put(f'/contract/{contract.contract_id}', json=mock_put_body) assert res.status_code == 200 mock_logic.update_contract.assert_called_once() @patch('abacus_contract.blueprints.contract.logic') def test_create_contract_with_lifecycle_and_schedules( mock_logic, fixture_client, mock_contract_and_lifecycle_post_payload ): """Test creating contract with contract_lifecycle and schedules.""" mock_logic.create_contract_with_lifecycle_and_schedules.return_value = \ response.Response( message='OK', status=201 ) res = fixture_client.post( '/contract/contract-lifecycle-schedule/contract-lifecycle', json=mock_contract_and_lifecycle_post_payload ) mock_contract_lifecycle = { 'lifecycle_term_start': datetime.date(2024, 7, 30) } assert res.status_code == 201 mock_logic.create_contract_with_lifecycle_and_schedules.assert_called_once_with( contract=mock_contract_and_lifecycle_post_payload['contract'], contract_lifecycle=mock_contract_lifecycle, contract_lifecycle_schedules=mock_contract_and_lifecycle_post_payload['contract_lifecycle_schedules'] # noqa: E501 ) @patch('abacus_contract.blueprints.contract.logic') def test_reactivate_contract(mock_logic, fixture_client): """Test to reactivate a contract.""" mock_contract = ContractFactory.create() mock_logic.reactivate_contract.return_value = response.Response( message='OK', status=200 ) res = fixture_client.put( f'/contract/{mock_contract.contract_id}/reactivate', ) assert res.status_code == 200 mock_logic.reactivate_contract.assert_called_once_with(mock_contract.contract_id) @pytest.mark.parametrize( ( 'check_access_result', 'expected_status' ), [ pytest.param( True, 200, id='Access OK' ), pytest.param( False, 403, id='Access not OK' ) ] ) @patch('abacus_contract.blueprints.contract.logic.get_contracts_by_account') @patch('abacus_contract.blueprints.contract.flask_request') @patch('abacus_contract.blueprints.contract.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.ows_client') def test_access_check_get_contracts_by_account( mock_ows_client, mock_authorize_many_accounts, mock_flask_request, mock_get_contracts, fixture_client, check_access_result, expected_status ): """Test the access check for `GET /contracts/account/`.""" mock_authorize_many_accounts.return_value = check_access_result mock_flask_request.verify_rules_access_standalone.return_value = True mock_get_contracts.return_value = response.Response(message='OK') account_id = 1 res = fixture_client.get( f'/contracts/account/{account_id}', headers={ 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': '123' } ) assert res.status_code == expected_status mock_authorize_many_accounts.assert_called_once_with( mock_ows_client, 'LabelProfile', '123', [account_id] ) @pytest.mark.parametrize( ( 'standalone_check_result', 'pdp_check_result', 'permissions_result', 'expected_status' ), [ pytest.param( True, None, True, 200, id='Standalone check OK, permissions check OK' ), pytest.param( True, None, False, 403, id='Standalone check OK, permissions check not OK' ), pytest.param(False, False, None, 403, id='PDP check fail'), pytest.param(False, True, True, 200, id='PDP check OK') ] ) @patch('abacus_contract.blueprints.contract.ItemView.get') @patch('abacus_contract.blueprints.contract.flask_request') @patch('abacus_contract.blueprints.contract.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.pdp_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.ows_client') def test_access_check_get_contract_by_id( mock_ows_client: MagicMock, mock_pdp_authorize_many_accounts: MagicMock, mock_authorize_many_accounts: MagicMock, mock_flask_request: MagicMock, mock_super_get: MagicMock, fixture_client: FlaskClient, standalone_check_result: bool, pdp_check_result: bool | None, permissions_result: bool | None, expected_status: int, ) -> None: """Test the access check for `GET /contract/`.""" mock_authorize_many_accounts.return_value = permissions_result mock_pdp_authorize_many_accounts.return_value = pdp_check_result mock_flask_request.verify_rules_access_standalone.return_value = \ standalone_check_result mock_super_get.return_value = flaskify(response.Response( message={'contract_id': 3, 'account_id': 1} )) res = fixture_client.get( '/contract/3', headers={ 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': '123' } ) assert res.status_code == expected_status if not standalone_check_result: mock_pdp_authorize_many_accounts.assert_called_once_with(account_ids=[1]) if pdp_check_result is not False: mock_authorize_many_accounts.assert_called_once_with( mock_ows_client, 'LabelProfile', '123', [1] ) contract_test_data = [ { 'account_id': 1, 'contract_id': 1, 'contract_name': 'Contract 1', 'contract_type': 'distribution', 'execution_date': None, 'general_note': 'This is for the test (general_note)', 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'oa_contract_id': None, 'reference_signing_entity_id': 2, 'sap_created_at': None, 'summary_note': 'This is for the test', 'term_end': '2029-06-01', 'term_start': '2019-06-01' }, { 'account_id': 2, 'contract_id': 2, 'contract_name': 'Contract 2', 'contract_type': 'neighbouring_rights', 'execution_date': None, 'general_note': 'Another test (general_note)', 'initial_start_date': '2020-01-15', 'is_excluded_from_accounting_run': False, 'oa_contract_id': None, 'reference_signing_entity_id': 4, 'sap_created_at': None, 'summary_note': 'Another test summary', 'term_end': '2030-12-31', 'term_start': '2020-01-01' } ] @pytest.mark.parametrize( ( 'profile_type', 'profile_role', 'authorize_return', 'contract_data', 'expected_status' ), [ pytest.param( 'AbacusProfile', 'administrator', None, contract_test_data, 200, id='standalone check' ), pytest.param( 'Account360Profile', 'account360', True, contract_test_data, 200, id='pdp check, authorized' ), pytest.param( 'Account360Profile', 'account360', False, contract_test_data, 403, id='pdp check, unauthorized' ), pytest.param( 'Account360Profile', 'account360', None, [], 200, id='no data, no check' ) ] ) @patch('abacus_contract.blueprints.contract.pdp_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.get_contracts') def test_get_contracts_with_default_params( mock_get_contracts: MagicMock, mock_pdp_authorize_many_accounts: MagicMock, fixture_client: FlaskClient, profile_type: str, profile_role: str, authorize_return: bool, contract_data: list[dict], expected_status: int ) -> None: """Tests for `GET /contracts` with default parameters.""" # Mock standalone access check based on profile type mock_pdp_authorize_many_accounts.return_value = authorize_return # Mock the return value with actual contracts mock_get_contracts.return_value = response.Response(message={ 'items': contract_data, 'total_count': len(contract_data) }) res = fixture_client.get( '/contracts?account_ids=&limit=20&offset=0&search_term=', 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_get_contracts.assert_called_once() if expected_status == 200: assert res.json == { 'items': contract_data, 'total_count': len(contract_data) } # Check if authorize_resource was called when standalone check failed if profile_type == 'Account360Profile': if contract_data: mock_pdp_authorize_many_accounts.assert_called_once_with([1, 2]) else: mock_pdp_authorize_many_accounts.assert_not_called() else: mock_pdp_authorize_many_accounts.assert_not_called() @pytest.mark.parametrize( ( 'profile_type', 'profile_role', 'authorize_return', 'expected_status' ), [ pytest.param( 'AbacusProfile', 'administrator', None, 200, id='standalone check' ), pytest.param( 'Account360Profile', 'account360', True, 200, id='pdp check, authorized' ), pytest.param( 'Account360Profile', 'account360', False, 403, id='pdp check, unauthorized' ) ] ) @patch('abacus_contract.blueprints.contract.pdp_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.logic.get_contracts_by_account') def test_get_contracts_by_account_ids( mock_get_contracts_by_accounts: MagicMock, mock_pdp_authorize_many_accounts: MagicMock, fixture_client: FlaskClient, profile_type: str, profile_role: str, authorize_return: bool, expected_status: int ): """GET contracts by a list of account_ids via POST /contracts/accounts.""" # Mock standalone access check based on profile type mock_pdp_authorize_many_accounts.return_value = authorize_return # Mock the return value with actual contracts mock_get_contracts_by_accounts.return_value = response.Response( message='OK', status=200 ) account_ids = [1, 2] res = fixture_client.post( '/contracts/accounts', json=account_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 # Check if authorize_many_accounts was called when standalone check failed if profile_type == 'Account360Profile': mock_pdp_authorize_many_accounts.assert_called_once_with([1, 2]) else: mock_pdp_authorize_many_accounts.assert_not_called() if expected_status == 200: mock_get_contracts_by_accounts.assert_called_once_with(account_ids) else: mock_get_contracts_by_accounts.assert_not_called() @pytest.mark.parametrize( ( 'profile_type', 'profile_role', 'pdp_authorize_return', 'permissions_authorize_return', 'expected_status' ), [ pytest.param( 'LabelProfile', 'catalog', None, True, 200, id='Standalone access, permissions check OK' ), pytest.param( 'LabelProfile', 'catalog', None, False, 403, id='Standalone access, permissions check not OK' ), pytest.param( 'Account360Profile', 'account360', True, True, 200, id='standalone access fail, pdp check OK' ), pytest.param( 'Account360Profile', 'account360', False, None, 403, id='standalone access fail, pdp check fail' ) ] ) @patch('abacus_contract.blueprints.contract.logic.get_contracts_by_account') @patch('abacus_contract.blueprints.contract.pdp_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.ows_client') def test_get_contracts_by_account_dataloader( mock_ows_client: MagicMock, mock_permission_authorize_many_accounts: MagicMock, mock_pdp_authorize_many_accounts: MagicMock, mock_get_contracts: MagicMock, fixture_client: FlaskClient, profile_type: str, profile_role: str, pdp_authorize_return: bool, permissions_authorize_return: bool, expected_status: int, ) -> None: """Tests for `POST /contracts/account/dataloader`.""" mock_pdp_authorize_many_accounts.return_value = pdp_authorize_return mock_permission_authorize_many_accounts.return_value = permissions_authorize_return # Mock the return value with actual contracts contracts = [ { 'account_id': 1, 'contract_id': 1, 'contract_name': 'Contract 1', 'contract_type': 'distribution', 'execution_date': None, 'general_note': 'This is for the test (general_note)', 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'oa_contract_id': None, 'reference_signing_entity_id': 2, 'sap_created_at': None, 'summary_note': 'This is for the test', 'term_end': '2029-06-01', 'term_start': '2019-06-01' }, { 'account_id': 2, 'contract_id': 2, 'contract_name': 'Contract 2', 'contract_type': 'distribution', 'execution_date': None, 'general_note': 'This is for the test (general_note)', 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'oa_contract_id': None, 'reference_signing_entity_id': 4, 'sap_created_at': None, 'summary_note': 'This is for the test', 'term_end': '2029-06-01', 'term_start': '2019-06-01' } ] mock_get_contracts.return_value = response.Response(message=[ { 'data': contract, } for contract in contracts] ) account_ids = [1, 2] res = fixture_client.post( '/contracts/account/dataloader', json=account_ids, headers={ 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Profile-Type': profile_type, 'Orchard-Profile-Id': '123', 'Orchard-Roles': profile_role, 'Orchard-Identity-Id': '1234' } ) assert res.status_code == expected_status if expected_status == 200: assert res.json == [ { 'data': contracts[0] }, { 'data': contracts[1] } ] # Check if authorize_many_accounts was called when standalone check failed if profile_type == 'Account360Profile': mock_pdp_authorize_many_accounts.assert_called_once_with(account_ids) else: mock_pdp_authorize_many_accounts.assert_not_called() if pdp_authorize_return is not False: mock_permission_authorize_many_accounts.assert_called_once_with( mock_ows_client, profile_type, '123', account_ids ) else: mock_permission_authorize_many_accounts.assert_not_called() # Check that get_contracts_by_account is called when authorization passes if expected_status == 200: mock_get_contracts.assert_called_once_with(account_ids, dataload=True) else: mock_get_contracts.assert_not_called() @pytest.mark.parametrize( ( 'profile_type', 'profile_role', 'pdp_authorize_return', 'permissions_authorize_return', 'expected_status' ), [ pytest.param( 'ContentProfile', 'manage_nr_ownership', None, True, 200, id='Standalone access, permissions check OK' ), pytest.param( 'ContentProfile', 'manage_nr_ownership', None, False, 403, id='Standalone access, permissions check not OK' ), pytest.param( 'Account360Profile', 'account360', True, True, 200, id='standalone access fail, pdp check OK' ), pytest.param( 'Account360Profile', 'account360', False, None, 403, id='standalone access fail, pdp check not OK' ), ], ) @patch('abacus_contract.blueprints.contract.logic.get_contracts_by_ids') @patch('abacus_contract.blueprints.contract.pdp_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract.ows_client') def test_get_contracts_dataloader( mock_ows_client: MagicMock, mock_permission_authorize_many_accounts: MagicMock, mock_pdp_authorize_many_accounts: MagicMock, mock_get_contracts: MagicMock, fixture_client: FlaskClient, profile_type: str, profile_role: str, pdp_authorize_return: bool, permissions_authorize_return: bool, expected_status: int ): """Tests for `POST /contracts/dataloader`.""" # Mock standalone access check based on profile type mock_pdp_authorize_many_accounts.return_value = pdp_authorize_return mock_permission_authorize_many_accounts.return_value = permissions_authorize_return contracts = [ { 'account_id': 1, 'contract_id': 1, 'contract_name': 'Contract 1', 'contract_type': 'distribution', 'execution_date': None, 'general_note': 'This is for the test (general_note)', 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'oa_contract_id': None, 'reference_signing_entity_id': 2, 'sap_created_at': None, 'summary_note': 'This is for the test', 'term_end': '2029-06-01', 'term_start': '2019-06-01' }, { 'account_id': 2, 'contract_id': 2, 'contract_name': 'Contract 2', 'contract_type': 'distribution', 'execution_date': None, 'general_note': 'This is for the test (general_note)', 'initial_start_date': '2018-07-28', 'is_excluded_from_accounting_run': False, 'oa_contract_id': None, 'reference_signing_entity_id': 4, 'sap_created_at': None, 'summary_note': 'This is for the test', 'term_end': '2029-06-01', 'term_start': '2019-06-01' } ] mock_get_contracts.return_value = contracts contract_ids = [1, 2, 100] # Add a non-existent ID res = fixture_client.post( '/contracts/dataloader', json=contract_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 # Verify JSON response when successful if expected_status == 200: assert res.json == [ { 'data': contracts[0] }, { 'data': contracts[1] }, { 'data': None } ] # Verify get_contracts_by_ids was called mock_get_contracts.assert_called_once_with(contract_ids) # Check if authorize_many_accounts was called when standalone check failed if profile_type == 'Account360Profile': mock_pdp_authorize_many_accounts.assert_called_once_with([1, 2]) else: mock_pdp_authorize_many_accounts.assert_not_called() if pdp_authorize_return is not False: mock_permission_authorize_many_accounts.assert_called_once_with( mock_ows_client, profile_type, '1234', [1, 2] ) else: mock_permission_authorize_many_accounts.assert_not_called() # Check that get_contracts_by_account is called mock_get_contracts.assert_called_once_with(contract_ids) @patch('abacus_contract.blueprints.contract.logic') @patch('abacus_contract.models.contract.Contract') def test_can_contract_be_deleted(mock_contract, mock_logic, fixture_client): """Test `GET /contract/:contract_id/can-be-deleted`.""" mock_account_contract = AccountContractFactory.create() contract_id = mock_account_contract.contract_id mock_result = response.Response( message={'can_be_deleted': True}, status=200 ) mock_contract.get_by_id_or_error.return_value = mock_account_contract mock_logic.can_contract_be_deleted.return_value = mock_result res = fixture_client.get(f'/contract/{contract_id}/can-be-deleted') mock_logic.can_contract_be_deleted.assert_called_once_with(contract_id) assert res.status_code == 200 assert res.json == mock_result.message @patch('abacus_contract.blueprints.contract.logic') def test_delete_contract(mock_logic, fixture_client): """Test `DELETE /contract/:contract_id`.""" contract_id = 1 mock_result = response.Response( message={'deleted': True}, status=200 ) mock_logic.delete_contract.return_value = mock_result res = fixture_client.delete(f'/contract/{contract_id}') mock_logic.delete_contract.assert_called_once_with(contract_id) assert res.status_code == 200 assert res.json == mock_result.message