"""Test for the flask plugin.""" from collections import namedtuple import json from typing import Any from unittest.mock import MagicMock from unittest.mock import patch import uuid from _pytest.monkeypatch import MonkeyPatch import flask from flask import g from jwtauth import JWTAuth from owsresponse import response from owsresponse import status from owsresponse.adaptors.flask import flaskify import pytest from owsrequest import access from owsrequest import auth from owsrequest import flask_request from owsrequest import request from owsrequest import rules from owsrequest.constants import errors from owsrequest.constants import headers from owsrequest.constants.environment import PROD_ENVIRONMENT from owsrequest.constants.environment import QA_ENVIRONMENT from owsrequest.constants.environment import UAT_ENVIRONMENT from tests.fixtures import flask_global monkeypatch = MonkeyPatch() Context = namedtuple( 'Context', 'identity_id orchard_user_id context_type ' + 'profile_id profile_type identity_uuid' ) def setup_function(function): """Invoked for every test function. Args: function (callable): the function that will run the test. """ flask_global.mock(monkeypatch) def teardown_function(function): """Invoked everytime a test function has completed. Args: function (callable): the function that has run. """ monkeypatch.undo() def test_get_ows(): """Test getting ows.""" ows = flask_request.get_ows() assert isinstance(ows, flask_request.Ows) assert isinstance(g.ows, flask_request.Ows) assert g.ows is ows @patch('owsrequest.flask_request.has_app_context') @patch('owsrequest.flask_request.jwt_auth_from_environment') def test_get_jwt_auth_client_requires_app_context( mock_jwt_auth_from_environment, mock_has_app_context, ): """Test getting jwt_auth_client requires app_context.""" mock_has_app_context.return_value = False with pytest.raises(AssertionError): flask_request.get_jwt_auth_client() mock_jwt_auth_from_environment.assert_not_called() @patch('owsrequest.flask_request.has_app_context') @patch('owsrequest.flask_request.current_app') @patch('owsrequest.flask_request.jwt_auth_from_environment') def test_get_jwt_auth_client_uses_existing_client( mock_jwt_auth_from_environment, mock_current_app, mock_has_app_context, ): """Test getting existing jwt_auth_client from current_app.""" mock_has_app_context.return_value = True mock_jwt_auth_client = MagicMock(spec=JWTAuth) mock_current_app.jwt_auth_client = mock_jwt_auth_client actual = flask_request.get_jwt_auth_client() mock_jwt_auth_from_environment.assert_not_called() assert mock_current_app.jwt_auth_client == actual assert mock_current_app.jwt_auth_client == mock_jwt_auth_client @patch('owsrequest.flask_request.has_app_context') @patch('owsrequest.flask_request.current_app') @patch('owsrequest.flask_request.hasattr') @patch('owsrequest.flask_request.jwt_auth_from_environment') def test_get_jwt_auth_client_creates_client( mock_jwt_auth_from_environment, mock_hasattr, mock_current_app, mock_has_app_context, ): """Test new jwt_auth_client is created when it has not been set on current_app.""" mock_has_app_context.return_value = True mock_hasattr.return_value = False mock_jwt_auth_client = MagicMock(spec=JWTAuth) mock_jwt_auth_from_environment.return_value = mock_jwt_auth_client assert not mock_current_app.jwt_auth_client == mock_jwt_auth_client actual = flask_request.get_jwt_auth_client() mock_hasattr.assert_called_once_with(mock_current_app, 'jwt_auth_client') mock_jwt_auth_from_environment.assert_called_once() assert mock_current_app.jwt_auth_client == actual assert mock_current_app.jwt_auth_client == mock_jwt_auth_client def test_setup(monkeypatch): """Test setting up a flask request on the application.""" application_name = 'ows-product' environment = PROD_ENVIRONMENT service_name = 'service-name' path = '/path' application = flask.Flask(application_name) monkeypatch.setattr( flask_request, 'process', MagicMock(spec=flask_request.process)) monkeypatch.setattr( auth, 'initialize_cache', MagicMock(return_value=None)) monkeypatch.setattr( auth, 'cache_jwks', MagicMock(return_value=None)) flask_request.setup(application, environment, uwsgi_cache_enabled=True) for method in flask_request.METHODS: assert hasattr(request, method.lower()) for method in flask_request.METHODS: attribute = method.lower() getattr(request, attribute)(service_name, path, headers={'foo': 'bar'}) arguments = flask_request.process.call_args[0] options = flask_request.process.call_args[1] assert arguments == ( application_name, environment, method.upper(), service_name, path) assert options == {'headers': {'foo': 'bar'}, 'uwsgi_cache_enabled': True} @pytest.mark.parametrize('auth_header_required', [(True), (False)]) @pytest.mark.parametrize('authorization_token, validation_method, module', [ ('xxx', 'confirm_authorization', request), ('Bearer xxx', 'validate_and_decode_jwt_token', auth), ]) def test_authorize_request( monkeypatch, authorization_token, validation_method, module, auth_header_required): """Testing an authorized request.""" monkeypatch.setattr( module, validation_method, MagicMock( return_value=response.Response(), spec=getattr(module, validation_method), ), ) monkeypatch.setattr(flask_request, 'get_jwt_auth_client', MagicMock()) current_request = MagicMock() current_request.headers = { 'Authorization': authorization_token, 'Correlation-Id': 'cid' } monkeypatch.setattr(flask_request, 'current_request', current_request) auth_response = flask_request.authorize_request( PROD_ENVIRONMENT, auth_header_required=auth_header_required) assert not isinstance(auth_response, flask.Response) assert getattr(module, validation_method).called assert not g.log.error.called if validation_method == 'validate_and_decode_jwt_token': assert flask_request.get_jwt_auth_client.called @pytest.mark.parametrize('environment', [ (QA_ENVIRONMENT), (PROD_ENVIRONMENT), (UAT_ENVIRONMENT), ]) @pytest.mark.parametrize( 'unauthorized_message, authorization_token, validation_method, module', [ (errors.AUTH_EXPIRED, 'xxx', 'confirm_authorization', request), (errors.AUTH_NOT_FOUND, 'xxx', 'confirm_authorization', request), (errors.HMAC_MISMATCH, 'xxx', 'confirm_authorization', request), (errors.SENDER_MISMATCH, 'xxx', 'confirm_authorization', request), ('some unauthorized message', 'Bearer xxx', 'validate_and_decode_jwt_token', auth), ]) def test_authorize_invalid_request( monkeypatch, unauthorized_message, authorization_token, validation_method, module, environment): """Testing an authorizing an invalid request for prod, qa, and uat environments.""" monkeypatch.setattr( module, validation_method, MagicMock( return_value=response.create_error_response( code=errors.UNAUTHORIZED_CODE, message=errors.UNAUTHORIZED_MESSAGE.get( unauthorized_message) if validation_method == 'confirm_authorization' else errors.UNAUTHORIZED_JWT_MESSAGE.get( unauthorized_message), status=status.UNAUTHORIZED ), spec=getattr(module, validation_method))) monkeypatch.setattr(flask_request, 'get_jwt_auth_client', MagicMock()) correlation_id = 'cid' authorization = authorization_token orchard_user_id = 'alw:123' current_request = MagicMock() current_request.headers = { 'Authorization': authorization, 'Correlation-Id': correlation_id, 'Orchard-User-Id': orchard_user_id } monkeypatch.setattr(flask_request, 'current_request', current_request) auth_response = flask_request.authorize_request(environment) expected = { 'code': errors.UNAUTHORIZED_CODE, 'message': errors.UNAUTHORIZED_MESSAGE.get(unauthorized_message) if validation_method == 'confirm_authorization' else errors.UNAUTHORIZED_JWT_MESSAGE.get(unauthorized_message) } assert isinstance(auth_response, flask.Response) assert auth_response.status_code == status.UNAUTHORIZED assert json.loads(auth_response.data.decode('utf8')) == expected g.ows.log.warning.assert_called_with( errors.UNAUTHORIZED_LOGGLY_MESSAGE.format( environment=environment, user_id=orchard_user_id, correlation_id=correlation_id, error_status=status.UNAUTHORIZED, error_code=errors.UNAUTHORIZED_CODE, error_message=errors.UNAUTHORIZED_MESSAGE.get( unauthorized_message) if validation_method == 'confirm_authorization' else errors.UNAUTHORIZED_JWT_MESSAGE.get(unauthorized_message) )) # Assert the module / validation method was called getattr(module, validation_method).called @pytest.mark.parametrize( 'authorization_header_passed, url, is_logged, auth_header_required', [ (True, 'test/other-route', True, False), (False, 'test/other-route', True, True), (False, 'test/health', False, False), (False, 'test/hello', False, True), (True, 'test/health', False, False), (True, 'test/hello', False, True), ]) def test_authorize_request_with_missing_authorization( monkeypatch, authorization_header_passed, url, is_logged, auth_header_required): """Testing an authorized request with missing or blank authorization.""" monkeypatch.setattr( request, 'confirm_authorization', MagicMock( return_value=response.Response(), spec=request.confirm_authorization)) monkeypatch.setattr(auth, 'validate_and_decode_jwt_token', MagicMock()) current_request = MagicMock() orchard_user_id = 123 correlation_id = 'cid' environment = PROD_ENVIRONMENT header_dict = { 'Correlation-Id': correlation_id, 'Orchard-User-Id': orchard_user_id } if authorization_header_passed: header_dict['Authorization'] = '' current_request.headers = header_dict current_request.url = url monkeypatch.setattr(flask_request, 'current_request', current_request) flask_response = flask_request.authorize_request(environment, auth_header_required) assert not auth.validate_and_decode_jwt_token.called assert not request.confirm_authorization.called assert not g.log.error.called if is_logged and auth_header_required: assert flask_response.status_code == 403 assert flask_response.response == [ b'{"code": "request_unauthorized", "message": "authorization not found"}' ] if is_logged and not auth_header_required: g.ows.log.debug.assert_called_with( errors.NO_AUTHORIZATION_MESSAGE.format( request_method=current_request.method, request_body=current_request.get_data(), request_url=current_request.url, environment=environment, user_id=orchard_user_id, correlation_id=correlation_id)) @pytest.mark.no_owsrequest_patch def test_processing(monkeypatch): """Ensuring that the process methods proxies the request.""" environment = PROD_ENVIRONMENT service_name = 'service-name' path = '/path' method = 'GET' application_name = 'app_name' correlation_id = str(uuid.uuid1()) headers = {'foo': 'bar'} value = {} monkeypatch.setattr(request, 'process', MagicMock( spec=request.process, return_value=value)) response = flask_request.process( application_name, environment, method, service_name, path, headers=headers, correlation_id=correlation_id, uwsgi_cache_enabled=True) assert response is value assert request.process.call_args[0][0] == application_name assert request.process.call_args[0][1] == environment assert request.process.call_args[0][2] == method assert request.process.call_args[0][3] == service_name assert request.process.call_args[0][4] == path assert request.process.call_args[0][5] == correlation_id assert request.process.call_args[1].get('uwsgi_cache_enabled') is True assert request.process.call_args[1].get('headers') == headers @pytest.mark.parametrize('correlation_id', (None, uuid.uuid1())) def test_next_correlation_id(correlation_id): """Test getting the next correlation id.""" for call_count in range(1, 5): response = flask_request.next_correlation_id(correlation_id) response_parts = response.split('.') if correlation_id: assert response_parts[0] == str(correlation_id) continue assert len(response_parts) == 2 assert response_parts[1] == str(call_count) def test_next_correlation_id_on_missing_counter(): """Missing counter should set it to 0.""" delattr(g, 'request_counter') flask_request.next_correlation_id() assert g.ows.request_counter == 1 def test_next_correlation_id_when_missing(): """Test the unlikely scenario where the correlation id has not been set.""" delattr(g, 'correlation_id') with pytest.raises(AssertionError): flask_request.next_correlation_id() def test_get_grass_headers(monkeypatch): """Test relevant Grass headers are returned as a tuple.""" expected_type = 'account-type' expected_id = 'account-id' mock_request = MagicMock() mock_request.headers.get.side_effect = [expected_type, expected_id] result = flask_request.get_grass_headers(mock_request) assert result == (expected_type, expected_id) @pytest.mark.parametrize('required', [True, False]) def test_verify_grass_headers(required, mocker): """Test that the generic validation is called with values from headers.""" expected_type = 'account-type' expected_id = 'account-id' mock_request = MagicMock() mock_request.headers.get.side_effect = [expected_type, expected_id] access_response = response.Response(message='lgtm') verify_grass_headers = mocker.patch.object( access, 'verify_grass_headers', autospec=True, return_value=access_response) result = flask_request.verify_grass_headers( mock_request, required=required) verify_grass_headers.assert_called_with( expected_type, expected_id, required=required) assert result == access_response def test_verify_grass_headers_default_required(mocker): """Test that headers are not required by default.""" expected_type = 'account-type' expected_id = 'account-id' mock_request = MagicMock() mock_request.headers.get.side_effect = [expected_type, expected_id] verify_grass_headers = mocker.patch.object( access, 'verify_grass_headers', autospec=True) flask_request.verify_grass_headers(mock_request) verify_grass_headers.assert_called_with( expected_type, expected_id, required=False) def test_grass_access_verification(mocker): """Test Grass Access.""" expected_type = 'account-type' expected_id = 'account-id' expected_account_id = 'request-account-id' expected_subaccount_id = 'request-subaccount-id' mock_request = MagicMock() mock_request.headers.get.side_effect = [expected_type, expected_id] verify_grass_access = mocker.patch.object( access, 'verify_grass_access', autospec=True) flask_request.verify_grass_access( mock_request, required=True, account=expected_account_id, subaccount=expected_subaccount_id) verify_grass_access.assert_called_with( expected_type, expected_id, required=True, account=expected_account_id, subaccount=expected_subaccount_id) @pytest.mark.parametrize('header_data', [ [headers.GRASS_ACCOUNT_TYPE_VENDOR, '10982'], [headers.GRASS_ACCOUNT_TYPE_VENDOR, ''], [headers.GRASS_ACCOUNT_TYPE_VENDOR, None], [headers.GRASS_ACCOUNT_TYPE_SUBACCOUNT, '10982'], [headers.GRASS_ACCOUNT_TYPE_SUBACCOUNT, ''], [headers.GRASS_ACCOUNT_TYPE_SUBACCOUNT, None], ['', '10909282'], [None, '10909282']]) def test_verify_grass_ownership(header_data): """Test verifying Grass ownership with valid headers.""" mock_request = MagicMock() mock_request.headers.get.side_effect = header_data ownership_response = response.Response() method = MagicMock(return_value=ownership_response) method_arguments = ('some', 'values', 'to', 'provide') method_named_parameters = {'additional': 'data'} ownership = flask_request.verify_grass_ownership( mock_request, method, *method_arguments, **method_named_parameters) assert ownership is ownership_response method.assert_called_with( *method_arguments, account_type=header_data[0], account_id=header_data[1], **method_named_parameters) def test_verify_grass_ownership_without_grass_headers(): """Test verifying Grass ownership with no grass headers.""" method = MagicMock() method_arguments = ('some', 'values', 'to', 'provide') method_named_parameters = {'additional': 'data'} mock_request = MagicMock() mock_request.headers = {} ownership = flask_request.verify_grass_ownership( mock_request, method, *method_arguments, **method_named_parameters) assert ownership.status == status.OK assert not method.called @pytest.mark.parametrize(('profile_type', 'profile_id'), [ ('a', '1'), ('foo', 'bar'), (None, None), ]) @patch('flask.request') def test_get_profile_headers(mock_request, profile_type, profile_id): """Test get_profile_headers.""" header_data = { headers.ORCHARD_PROFILE_TYPE: profile_type, headers.ORCHARD_PROFILE_ID: profile_id } mock_request.headers = header_data profile_type, profile_id = flask_request.get_profile_headers(mock_request) assert profile_type == header_data[headers.ORCHARD_PROFILE_TYPE] assert profile_id == header_data[headers.ORCHARD_PROFILE_ID] @pytest.mark.parametrize(('profile_type', 'profile_id', 'expected'), [ ('dummy', 'id', 200), ('', '', 200), (None, None, 200), ('', 'id', 400), ('dummy', '', 400), (None, 123, 400), ('dummy', None, 400), ]) @patch('flask.request') def test_verify_profile_headers( mock_request, profile_type, profile_id, expected): """Test verify_profile_headers.""" header_data = { headers.ORCHARD_PROFILE_TYPE: profile_type, headers.ORCHARD_PROFILE_ID: profile_id } mock_request.headers = header_data result = flask_request.verify_profile_headers(mock_request) assert result.status == expected @pytest.mark.parametrize(( 'head_profile_type', 'head_profile_id', 'profile_type', 'profile_id', 'expected' ), [ # direct calls. (None, None, 'dummy', '123', 200), ('', '', 'dummy', '123', 200), # via grass ('dummy', '456', 'dummy', '123', 403), ('random', '123', 'dummy', '123', 403), ('dummy', 123, 'dummy', '123', 200), ('dummy', '123', 'dummy', '123', 200), # fake calls (None, '123', 'dummy', '123', 400), ('dummy', None, 'dummy', '123', 400), ]) @patch('flask.request') def test_verify_profile_headers_match_route( mock_request, head_profile_type, head_profile_id, profile_type, profile_id, expected): """Test verify_profile_headers_match_route.""" header_data = { headers.ORCHARD_PROFILE_TYPE: head_profile_type, headers.ORCHARD_PROFILE_ID: head_profile_id } mock_request.headers = header_data result = flask_request.verify_profile_headers_match_route( mock_request, profile_type, profile_id) assert result.status == expected def test_ows_get_headers_full_profile(monkeypatch): """Test it can fill out full profile headers.""" context = Context( identity_id=1, orchard_user_id=1, context_type='profile', profile_id=2, profile_type='Podcast', identity_uuid=6, ) monkeypatch.setattr( flask_request, '_request_context', MagicMock(return_value=context)) assert flask_request.get_ows_headers() == { 'Orchard-Identity-Id': 1, 'Orchard-Identity-UUID': 6, 'Orchard-Profile-Id': 2, 'Orchard-Profile-Type': 'Podcast' } def test_ows_get_headers_partial_profile(monkeypatch): """Test it can fill out partial profile headers.""" context = Context( identity_id=None, orchard_user_id=1, context_type='profile', profile_id=2, profile_type=None, identity_uuid=None, ) monkeypatch.setattr( flask_request, '_request_context', MagicMock(return_value=context)) assert flask_request.get_ows_headers() == { 'Orchard-Profile-Id': 2, } def test_ows_get_headers_full_account(monkeypatch): """Test it can fill out full account headers.""" context = Context( identity_id=1, orchard_user_id=1, context_type='account', profile_id=2, profile_type='Cow', identity_uuid=None, ) monkeypatch.setattr( flask_request, '_request_context', MagicMock(return_value=context)) assert flask_request.get_ows_headers() == { 'Grass-Account-Id': 2, 'Grass-Account-Type': 'Cow', 'Orchard-User-Id': 1 } def test_ows_get_headers_partial_account(monkeypatch): """Test it can fill out partial account headers.""" context = Context( identity_id=1, orchard_user_id=1, context_type='account', profile_id=None, profile_type=None, identity_uuid=None, ) monkeypatch.setattr( flask_request, '_request_context', MagicMock(return_value=context)) assert flask_request.get_ows_headers() == { 'Orchard-User-Id': 1 } @pytest.mark.parametrize(( 'requestor_service_name', 'log_only', 'profile_type', 'access_denied', 'path', 'skip' ), [ # exact match services ('graphql-product', True, 'LabelProfile', False, '/holds/234', False), ('graphql-product', True, 'IncorrectProfile', True, '/holds/234', False), ('ows-grass', False, 'IncorrectProfile', True, '/holds/234', False), ('ows-users', True, 'LabelProfile', False, '/health', True), ('ows-users', True, 'LabelProfile', False, '/hello', True), ('ows-dummy', True, 'LabelProfile', False, '/hello', True), # regex match services ('graphql-user', False, 'IncorrectProfile', True, '/holds/234', False), ('graphql-new', False, 'LabelProfile', True, '/report', False), ('graphql-new', False, 'IncorrectProfile', True, '/holds/234', False), ('graphql-new-name', False, 'LabelProfile', False, '/holds/234', False), ('graphql-user', False, 'LabelProfile', False, '/holds/234', False), # no service name (None, True, 'LabelProfile', False, '/holds/234', True), ]) def test_verify_rules_access( monkeypatch, requestor_service_name, log_only, profile_type, access_denied, path, skip): """Verify rules access setup.""" current_request = MagicMock() current_request.headers = { 'Orchard-Requester-Service': requestor_service_name, 'Correlation-Id': 'cid' } current_request.method = 'PUT' current_request.path = path request_context = MagicMock() request_context.requestor_service_name = requestor_service_name request_context.profile_type = profile_type request_context.profile_id = 'abc' request_context.roles = ['catalog'] monkeypatch.setattr( flask_request, '_request_context', MagicMock( return_value=request_context)) validator = rules.EndpointRulesValidator('tests/fixtures/rules.yaml') monkeypatch.setattr(flask_request, 'current_request', current_request) response = flask_request.verify_rules_access( current_request, validator, log_only) if skip: g.ows.log.debug.assert_called_with( errors.SKIP_ROLE_CHECK_MESSAGE.format(requestor_service_name)) if access_denied: g.ows.log.info.assert_called_with( errors.ACCESS_DENIED_MESSAGE.format( method='PUT', path=path, profile_id='abc', profile_type=profile_type, roles=['catalog'], correlation_id='cid' )) if not log_only: assert response.status_code == 403 assert response.json['code'] == errors.ERROR_CODE_FORBIDDEN @pytest.mark.parametrize('path,access_denied', [ ('/users/identity/a0b8f767-707a-432c-989e-b3b3f74d9cb7/devices', True), ('/users/identity/e058c7f5-5102-4e80-9400-5d20d972bccd', True), ('/users/identity/e058c7f5-5102-4e80-9400-5d20d972bccd/profiles', True), ('/identity/c3497c61-e018-4da6-af38-b57624980fa6/applications', True), ('/hello/', False), ('/users/identity/a0b8f767-707a-432c-989e-b3b3f74d9cb7/application/orchard-go/profiles', False), # noqa ('/users/identity/d907a6b7-c3e2-4d6f-be27-97ab89fde179/application/orchard-go/profiles', False), # noqa ('/users/identity/e058c7f5-5102-4e80-9400-5d20d972bccd/application/frontend-publishing/profiles', False), # noqa ('/users/identity/c3497c61-e018-4da6-af38-b57624980fa6/application/frontend-insights/profiles', False), # noqa ]) def test_verify_rules_access_exclude_paths(monkeypatch, path, access_denied): """Verify rules access exclude_paths.""" current_request = MagicMock() current_request.headers = { 'Orchard-Requester-Service': 'graphql-product', 'Correlation-Id': 'cid' } current_request.method = 'GET' current_request.path = path request_context = MagicMock() request_context.requestor_service_name = 'graphql-product' request_context.profile_type = 'LabelProfile' request_context.profile_id = 'abc' request_context.roles = ['catalog'] monkeypatch.setattr( flask_request, '_request_context', MagicMock( return_value=request_context)) exclude_paths = [ '/hello/', '/users/identity//application//profiles' ] validator = rules.EndpointRulesValidator('tests/fixtures/rules.yaml') monkeypatch.setattr(flask_request, 'current_request', current_request) verify_response = flask_request.verify_rules_access( current_request, validator, False, exclude_paths) if access_denied: assert verify_response.status_code == 403 g.ows.log.info.assert_called_with( errors.ACCESS_DENIED_MESSAGE.format( method=current_request.method, path=current_request.path, profile_id=request_context.profile_id, profile_type=request_context.profile_type, roles=request_context.roles, correlation_id='cid' )) else: assert verify_response is None @patch('owsrequest.flask_request.verify_rules_access') @patch('owsrequest.flask_request.get_rules_validator') def test_verify_rules_access_standalone_uses_app_context( mock_get_rules_validator: MagicMock, mock_verify_rules_access: MagicMock, ) -> None: """Test verify_rules_access_standalone uses app_context.""" mock_request = MagicMock() mock_get_rules_validator.return_value = None result = flask_request.verify_rules_access_standalone(mock_request) mock_get_rules_validator.assert_called_once() mock_verify_rules_access.assert_not_called() assert not result @pytest.mark.parametrize( 'mock_verify_rules_access_return_value,expected_result,description', [ (None, True, 'None value is ok'), (True, True, 'True value is ok'), ( flaskify(response.create_error_response('no', 'forbidden', status=403)), False, 'Error response is not ok', ), ], ) @patch('owsrequest.flask_request.verify_rules_access') @patch('owsrequest.flask_request.get_rules_validator') def test_verify_rules_access_standalone_skips_app_context( mock_get_rules_validator: MagicMock, mock_verify_rules_access: MagicMock, mock_verify_rules_access_return_value: Any, expected_result: bool, description: str, ) -> None: """Test verify_rules_access_standalone uses app_context.""" mock_request = MagicMock() mock_get_rules_validator.return_value = MagicMock(spec=rules.EndpointRulesValidator) mock_rules_validator = MagicMock(spec=rules.EndpointRulesValidator) mock_verify_rules_access.return_value = mock_verify_rules_access_return_value result = flask_request.verify_rules_access_standalone(mock_request, mock_rules_validator) mock_get_rules_validator.assert_not_called() mock_verify_rules_access.assert_called_once_with( mock_request, mock_rules_validator, log_only=False, ) assert result == expected_result, description @patch('owsrequest.flask_request.has_app_context') def test_get_rules_validator_requires_app_context( mock_has_app_context, ): """Test getting rules validator requires app_context.""" mock_has_app_context.return_value = False with pytest.raises(AssertionError): flask_request.get_rules_validator() mock_has_app_context.assert_called_once() @patch('owsrequest.flask_request.has_app_context') @patch('owsrequest.flask_request.current_app') def test_get_rules_validator_uses_app_rules_validator( mock_current_app, mock_has_app_context, ): """Test existing get_rules_validator from current_app.""" mock_has_app_context.return_value = True mock_rules_validator = MagicMock(spec=rules.EndpointRulesValidator) mock_current_app.rules_validator = mock_rules_validator actual = flask_request.get_rules_validator() mock_has_app_context.assert_called_once() assert mock_current_app.rules_validator == actual assert mock_current_app.rules_validator == mock_rules_validator @patch('owsrequest.flask_request.has_app_context') @patch('owsrequest.flask_request.current_app') @patch('owsrequest.flask_request.hasattr') def test_get_rules_validator_returns_none( mock_hasattr, mock_current_app, mock_has_app_context, ): """Test None is returned when rules_validator has not been set on current_app.""" mock_has_app_context.return_value = True mock_hasattr.return_value = False actual = flask_request.get_rules_validator() mock_has_app_context.assert_called_once() mock_hasattr.assert_called_once_with(mock_current_app, 'rules_validator') assert actual is None def test_set_rules_validator(): """Test rules_file is loaded and set on application.""" mock_application = MagicMock() rules_file = 'tests/fixtures/rules.yaml' flask_request.set_rules_validator(mock_application, rules_file) assert isinstance(mock_application.rules_validator, rules.EndpointRulesValidator)