import json import pytest import falcon import falcon.errors import falcon.testing as testing from falcon.util.deprecation import DeprecatedWarning from falcon.util.misc import _utcnow _EXPECTED_BODY = {'status': 'ok'} context = {'executed_methods': []} # type: ignore TEST_ROUTE = '/test_path' class CaptureResponseMiddleware: def process_response(self, req, resp, resource, req_succeeded): self.req = req self.resp = resp self.resource = resource self.req_succeeded = req_succeeded class CaptureRequestMiddleware: def process_request(self, req, resp): self.req = req class RequestTimeMiddleware: def process_request(self, req, resp): global context context['start_time'] = _utcnow() def process_resource(self, req, resp, resource, params): global context context['mid_time'] = _utcnow() def process_response(self, req, resp, resource, req_succeeded): global context context['end_time'] = _utcnow() context['req_succeeded'] = req_succeeded async def process_request_async(self, req, resp): self.process_request(req, resp) async def process_resource_async(self, req, resp, resource, params): self.process_resource(req, resp, resource, params) async def process_response_async(self, req, resp, resource, req_succeeded): self.process_response(req, resp, resource, req_succeeded) class TransactionIdMiddleware: def process_request(self, req, resp): global context context['transaction_id'] = 'unique-req-id' def process_resource(self, req, resp, resource, params): global context context['resource_transaction_id'] = 'unique-req-id-2' def process_response(self, req, resp, resource, req_succeeded): pass class TransactionIdMiddlewareAsync: def __init__(self): self._mw = TransactionIdMiddleware() async def process_request(self, req, resp): self._mw.process_request(req, resp) async def process_resource(self, req, resp, resource, params): self._mw.process_resource(req, resp, resource, params) async def process_response(self, req, resp, resource, req_succeeded): self._mw.process_response(req, resp, resource, req_succeeded) class ExecutedFirstMiddleware: def process_request(self, req, resp): global context context['executed_methods'].append( '{}.{}'.format(self.__class__.__name__, 'process_request') ) def process_resource(self, req, resp, resource, params): global context context['executed_methods'].append( '{}.{}'.format(self.__class__.__name__, 'process_resource') ) # NOTE(kgriffs): This also tests that the framework can continue to # call process_response() methods that do not have a 'req_succeeded' # arg. def process_response(self, req, resp, resource, req_succeeded): global context context['executed_methods'].append( '{}.{}'.format(self.__class__.__name__, 'process_response') ) context['req'] = req context['resp'] = resp context['resource'] = resource class ExecutedLastMiddleware(ExecutedFirstMiddleware): pass class RemoveBasePathMiddleware: def process_request(self, req, resp): req.path = req.path.replace('/base_path', '', 1) class ResponseCacheMiddlware: PROCESS_REQUEST_CACHED_BODY = {'cached': True} PROCESS_RESOURCE_CACHED_BODY = {'cached': True, 'resource': True} def process_request(self, req, resp): if req.path == '/cached': resp.media = self.PROCESS_REQUEST_CACHED_BODY resp.complete = True return def process_resource(self, req, resp, resource, params): if req.path == '/cached/resource': resp.media = self.PROCESS_RESOURCE_CACHED_BODY resp.complete = True return class AccessParamsMiddleware: def process_resource(self, req, resp, resource, params): global context params['added'] = True context['params'] = params class MiddlewareClassResource: def on_get(self, req, resp, **kwargs): resp.status = falcon.HTTP_200 resp.text = json.dumps(_EXPECTED_BODY) def on_post(self, req, resp): raise falcon.HTTPForbidden(title=falcon.HTTP_403, description='Setec Astronomy') class EmptySignatureMiddleware: def process_request(self): pass def process_response(self): pass class TestCorsResource: def on_get(self, req, resp, **kwargs): resp.status = falcon.HTTP_200 resp.text = 'Test' class TestMiddleware: def setup_method(self, method): # Clear context global context context = {'executed_methods': []} class TestRequestTimeMiddleware(TestMiddleware): def test_skip_process_resource(self, asgi, util): global context app = util.create_app(asgi, middleware=[RequestTimeMiddleware()]) app.add_route('/', MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path='/404') assert response.status == falcon.HTTP_404 assert 'start_time' in context assert 'mid_time' not in context assert 'end_time' in context assert not context['req_succeeded'] def test_add_invalid_middleware(self, asgi, util): """Test than an invalid class can not be added as middleware""" class InvalidMiddleware: def process_request(self, *args): pass mw_list = [RequestTimeMiddleware(), InvalidMiddleware] with pytest.raises(AttributeError): util.create_app(asgi, middleware=mw_list) mw_list = [RequestTimeMiddleware(), 'InvalidMiddleware'] with pytest.raises(TypeError): util.create_app(asgi, middleware=mw_list) mw_list = [{'process_request': 90}] with pytest.raises(TypeError): util.create_app(asgi, middleware=mw_list) def test_response_middleware_raises_exception(self, asgi, util): """Test that error in response middleware is propagated up""" class RaiseErrorMiddleware: def process_response(self, req, resp, resource): raise Exception('Always fail') app = util.create_app(asgi, middleware=[RaiseErrorMiddleware()]) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) result = client.simulate_request(path=TEST_ROUTE) assert result.status_code == 500 @pytest.mark.parametrize('independent_middleware', [True, False]) def test_log_get_request(self, independent_middleware, asgi, util): """Test that Log middleware is executed""" global context app = util.create_app( asgi, middleware=[RequestTimeMiddleware()], independent_middleware=independent_middleware, ) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path=TEST_ROUTE) assert _EXPECTED_BODY == response.json assert response.status == falcon.HTTP_200 assert 'start_time' in context assert 'mid_time' in context assert 'end_time' in context assert ( context['mid_time'] >= context['start_time'] ), 'process_resource not executed after request' assert ( context['end_time'] >= context['start_time'] ), 'process_response not executed after request' assert context['req_succeeded'] class TestTransactionIdMiddleware(TestMiddleware): def test_generate_trans_id_with_request(self, asgi, util): """Test that TransactionIdmiddleware is executed""" global context middleware = ( TransactionIdMiddlewareAsync() if asgi else TransactionIdMiddleware() ) app = util.create_app(asgi, middleware=middleware) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path=TEST_ROUTE) assert _EXPECTED_BODY == response.json assert response.status == falcon.HTTP_200 assert 'transaction_id' in context assert 'unique-req-id' == context['transaction_id'] class TestSeveralMiddlewares(TestMiddleware): @pytest.mark.parametrize('independent_middleware', [True, False]) def test_generate_trans_id_and_time_with_request( self, independent_middleware, asgi, util ): # NOTE(kgriffs): We test both so that we can cover the code paths # where only a single middleware method is implemented by a # component. creq = CaptureRequestMiddleware() cresp = CaptureResponseMiddleware() global context app = util.create_app( asgi, independent_middleware=independent_middleware, # NOTE(kgriffs): Pass as a generic iterable to verify that works. middleware=iter( [ TransactionIdMiddleware(), RequestTimeMiddleware(), ] ), ) # NOTE(kgriffs): Add a couple more after the fact to test # add_middleware(). app.add_middleware(creq) app.add_middleware(cresp) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path=TEST_ROUTE) assert _EXPECTED_BODY == response.json assert response.status == falcon.HTTP_200 assert 'transaction_id' in context assert 'unique-req-id' == context['transaction_id'] assert 'start_time' in context assert 'mid_time' in context assert 'end_time' in context assert ( context['mid_time'] >= context['start_time'] ), 'process_resource not executed after request' assert ( context['end_time'] >= context['start_time'] ), 'process_response not executed after request' def test_legacy_middleware_called_with_correct_args(self, asgi, util): global context app = util.create_app(asgi, middleware=[ExecutedFirstMiddleware()]) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) assert isinstance(context['req'], falcon.Request) assert isinstance(context['resp'], falcon.Response) assert isinstance(context['resource'], MiddlewareClassResource) def test_middleware_execution_order(self, asgi, util): global context app = util.create_app( asgi, independent_middleware=False, middleware=[ExecutedFirstMiddleware(), ExecutedLastMiddleware()], ) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path=TEST_ROUTE) assert _EXPECTED_BODY == response.json assert response.status == falcon.HTTP_200 # as the method registration is in a list, the order also is # tested expectedExecutedMethods = [ 'ExecutedFirstMiddleware.process_request', 'ExecutedLastMiddleware.process_request', 'ExecutedFirstMiddleware.process_resource', 'ExecutedLastMiddleware.process_resource', 'ExecutedLastMiddleware.process_response', 'ExecutedFirstMiddleware.process_response', ] assert expectedExecutedMethods == context['executed_methods'] def test_independent_middleware_execution_order(self, asgi, util): global context app = util.create_app( asgi, independent_middleware=True, middleware=[ExecutedFirstMiddleware(), ExecutedLastMiddleware()], ) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path=TEST_ROUTE) assert _EXPECTED_BODY == response.json assert response.status == falcon.HTTP_200 # as the method registration is in a list, the order also is # tested expectedExecutedMethods = [ 'ExecutedFirstMiddleware.process_request', 'ExecutedLastMiddleware.process_request', 'ExecutedFirstMiddleware.process_resource', 'ExecutedLastMiddleware.process_resource', 'ExecutedLastMiddleware.process_response', 'ExecutedFirstMiddleware.process_response', ] assert expectedExecutedMethods == context['executed_methods'] def test_multiple_response_mw_throw_exception(self, asgi, util): """Test that error in inner middleware leaves""" global context context['req_succeeded'] = [] class RaiseStatusMiddleware: def process_response(self, req, resp, resource, req_succeeded): raise falcon.HTTPStatus(falcon.HTTP_201) class RaiseErrorMiddleware: def process_response(self, req, resp, resource, req_succeeded): raise falcon.HTTPError(falcon.HTTP_748) class ProcessResponseMiddleware: def process_response(self, req, resp, resource, req_succeeded): context['executed_methods'].append('process_response') context['req_succeeded'].append(req_succeeded) app = util.create_app( asgi, middleware=[ ProcessResponseMiddleware(), RaiseErrorMiddleware(), ProcessResponseMiddleware(), RaiseStatusMiddleware(), ProcessResponseMiddleware(), ], ) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path=TEST_ROUTE) assert response.status == falcon.HTTP_748 expected_methods = ['process_response'] * 3 assert context['executed_methods'] == expected_methods assert context['req_succeeded'] == [True, False, False] def test_inner_mw_throw_exception(self, asgi, util): """Test that error in inner middleware leaves""" global context class MyException(Exception): pass class RaiseErrorMiddleware: def process_request(self, req, resp): raise MyException('Always fail') app = util.create_app( asgi, middleware=[ TransactionIdMiddleware(), RequestTimeMiddleware(), RaiseErrorMiddleware(), ], ) # NOTE(kgriffs): Now that we install a default handler for # Exception, we have to clear them to test the path we want # to trigger with RaiseErrorMiddleware # TODO(kgriffs): Since we always add a default error handler # for Exception, should we take out the checks in the WSGI/ASGI # callable and just always assume it will be handled? If so, # then we would remove the test here... app._error_handlers.clear() app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) with pytest.raises(MyException): client.simulate_request(path=TEST_ROUTE) # RequestTimeMiddleware process_response should be executed assert 'transaction_id' in context assert 'start_time' in context assert 'mid_time' not in context # NOTE(kgriffs): Should not have been added since raising an # unhandled error skips further processing, including response # middleware methods. assert 'end_time' not in context def test_inner_mw_throw_exception_while_processing_resp(self, asgi, util): """Test that error in inner middleware leaves""" global context class MyException(Exception): pass class RaiseErrorMiddleware: def process_response(self, req, resp, resource, req_succeeded): raise MyException('Always fail') app = util.create_app( asgi, middleware=[ TransactionIdMiddleware(), RequestTimeMiddleware(), RaiseErrorMiddleware(), ], ) # NOTE(kgriffs): Now that we install a default handler for # Exception, we have to clear them to test the path we want # to trigger with RaiseErrorMiddleware # TODO(kgriffs): Since we always add a default error handler # for Exception, should we take out the checks in the WSGI/ASGI # callable and just always assume it will be handled? If so, # then we would remove the test here... app._error_handlers.clear() app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) with pytest.raises(MyException): client.simulate_request(path=TEST_ROUTE) # RequestTimeMiddleware process_response should be executed assert 'transaction_id' in context assert 'start_time' in context assert 'mid_time' in context # NOTE(kgriffs): Should not have been added since raising an # unhandled error skips further processing, including response # middleware methods. assert 'end_time' not in context def test_inner_mw_with_ex_handler_throw_exception(self, asgi, util): """Test that error in inner middleware leaves""" global context class RaiseErrorMiddleware: def process_request(self, req, resp, resource): raise Exception('Always fail') app = util.create_app( asgi, middleware=[ TransactionIdMiddleware(), RequestTimeMiddleware(), RaiseErrorMiddleware(), ], ) def handler(req, resp, ex, params): context['error_handler'] = True app.add_error_handler(Exception, handler) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) # RequestTimeMiddleware process_response should be executed assert 'transaction_id' in context assert 'start_time' in context assert 'mid_time' not in context assert 'end_time' in context assert 'error_handler' in context def test_outer_mw_with_ex_handler_throw_exception(self, asgi, util): """Test that error in inner middleware leaves""" global context class RaiseErrorMiddleware: def process_request(self, req, resp): raise Exception('Always fail') app = util.create_app( asgi, middleware=[ TransactionIdMiddleware(), RaiseErrorMiddleware(), RequestTimeMiddleware(), ], ) def handler(req, resp, ex, params): context['error_handler'] = True app.add_error_handler(Exception, handler) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) # Any mw is executed now... assert 'transaction_id' in context assert 'start_time' not in context assert 'mid_time' not in context assert 'end_time' in context assert 'error_handler' in context def test_order_mw_executed_when_exception_in_resp(self, asgi, util): """Test that error in inner middleware leaves""" global context class RaiseErrorMiddleware: def process_response(self, req, resp, resource): raise Exception('Always fail') app = util.create_app( asgi, middleware=[ ExecutedFirstMiddleware(), RaiseErrorMiddleware(), ExecutedLastMiddleware(), ], ) def handler(req, resp, ex, params): pass app.add_error_handler(Exception, handler) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) # Any mw is executed now... expectedExecutedMethods = [ 'ExecutedFirstMiddleware.process_request', 'ExecutedLastMiddleware.process_request', 'ExecutedFirstMiddleware.process_resource', 'ExecutedLastMiddleware.process_resource', 'ExecutedLastMiddleware.process_response', 'ExecutedFirstMiddleware.process_response', ] assert expectedExecutedMethods == context['executed_methods'] def test_order_independent_mw_executed_when_exception_in_resp(self, asgi, util): """Test that error in inner middleware leaves""" global context class RaiseErrorMiddleware: def process_response(self, req, resp, resource): raise Exception('Always fail') app = util.create_app( asgi, independent_middleware=True, middleware=[ ExecutedFirstMiddleware(), RaiseErrorMiddleware(), ExecutedLastMiddleware(), ], ) def handler(req, resp, ex, params): pass app.add_error_handler(Exception, handler) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) # Any mw is executed now... expectedExecutedMethods = [ 'ExecutedFirstMiddleware.process_request', 'ExecutedLastMiddleware.process_request', 'ExecutedFirstMiddleware.process_resource', 'ExecutedLastMiddleware.process_resource', 'ExecutedLastMiddleware.process_response', 'ExecutedFirstMiddleware.process_response', ] assert expectedExecutedMethods == context['executed_methods'] def test_order_mw_executed_when_exception_in_req(self, asgi, util): """Test that error in inner middleware leaves""" global context class RaiseErrorMiddleware: def process_request(self, req, resp): raise Exception('Always fail') class RaiseErrorMiddlewareAsync: async def process_request(self, req, resp): raise Exception('Always fail') rem = RaiseErrorMiddlewareAsync() if asgi else RaiseErrorMiddleware() app = util.create_app( asgi, middleware=[ExecutedFirstMiddleware(), rem, ExecutedLastMiddleware()] ) def handler(req, resp, ex, params): pass app.add_error_handler(Exception, handler) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) # Any mw is executed now... expectedExecutedMethods = [ 'ExecutedFirstMiddleware.process_request', 'ExecutedLastMiddleware.process_response', 'ExecutedFirstMiddleware.process_response', ] assert expectedExecutedMethods == context['executed_methods'] def test_order_independent_mw_executed_when_exception_in_req(self, asgi, util): """Test that error in inner middleware leaves""" global context class RaiseErrorMiddleware: def process_request(self, req, resp): raise Exception('Always fail') class RaiseErrorMiddlewareAsync: async def process_request(self, req, resp): raise Exception('Always fail') rem = RaiseErrorMiddlewareAsync() if asgi else RaiseErrorMiddleware() app = util.create_app( asgi, independent_middleware=True, middleware=[ExecutedFirstMiddleware(), rem, ExecutedLastMiddleware()], ) def handler(req, resp, ex, params): pass app.add_error_handler(Exception, handler) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) # All response middleware still executed... expectedExecutedMethods = [ 'ExecutedFirstMiddleware.process_request', 'ExecutedLastMiddleware.process_response', 'ExecutedFirstMiddleware.process_response', ] assert expectedExecutedMethods == context['executed_methods'] def test_order_mw_executed_when_exception_in_rsrc(self, asgi, util): """Test that error in inner middleware leaves""" global context class RaiseErrorMiddleware: def process_resource(self, req, resp, resource): raise Exception('Always fail') class RaiseErrorMiddlewareAsync: # NOTE(kgriffs): The *_async postfix is not required in this # case, but we include it to make sure it works as expected. async def process_resource_async(self, req, resp, resource): raise Exception('Always fail') rem = RaiseErrorMiddlewareAsync() if asgi else RaiseErrorMiddleware() app = util.create_app( asgi, middleware=[ExecutedFirstMiddleware(), rem, ExecutedLastMiddleware()] ) def handler(req, resp, ex, params): pass app.add_error_handler(Exception, handler) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) # Any mw is executed now... expectedExecutedMethods = [ 'ExecutedFirstMiddleware.process_request', 'ExecutedLastMiddleware.process_request', 'ExecutedFirstMiddleware.process_resource', 'ExecutedLastMiddleware.process_response', 'ExecutedFirstMiddleware.process_response', ] assert expectedExecutedMethods == context['executed_methods'] def test_order_independent_mw_executed_when_exception_in_rsrc(self, asgi, util): """Test that error in inner middleware leaves""" global context class RaiseErrorMiddleware: def process_resource(self, req, resp, resource): raise Exception('Always fail') class RaiseErrorMiddlewareAsync: async def process_resource(self, req, resp, resource): raise Exception('Always fail') rem = RaiseErrorMiddlewareAsync() if asgi else RaiseErrorMiddleware() app = util.create_app( asgi, independent_middleware=True, middleware=[ExecutedFirstMiddleware(), rem, ExecutedLastMiddleware()], ) def handler(req, resp, ex, params): pass app.add_error_handler(Exception, handler) app.add_route(TEST_ROUTE, MiddlewareClassResource()) client = testing.TestClient(app) client.simulate_request(path=TEST_ROUTE) # Any mw is executed now... expectedExecutedMethods = [ 'ExecutedFirstMiddleware.process_request', 'ExecutedLastMiddleware.process_request', 'ExecutedFirstMiddleware.process_resource', 'ExecutedLastMiddleware.process_response', 'ExecutedFirstMiddleware.process_response', ] assert expectedExecutedMethods == context['executed_methods'] class TestRemoveBasePathMiddleware(TestMiddleware): def test_base_path_is_removed_before_routing(self, asgi, util): """Test that RemoveBasePathMiddleware is executed before routing""" app = util.create_app(asgi, middleware=RemoveBasePathMiddleware()) # We dont include /base_path as it will be removed in middleware app.add_route('/sub_path', MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path='/base_path/sub_path') assert _EXPECTED_BODY == response.json assert response.status == falcon.HTTP_200 response = client.simulate_request(path='/base_pathIncorrect/sub_path') assert response.status == falcon.HTTP_404 class TestResourceMiddleware(TestMiddleware): @pytest.mark.parametrize('independent_middleware', [True, False]) def test_can_access_resource_params(self, asgi, util, independent_middleware): """Test that params can be accessed from within process_resource""" global context class Resource: def on_get(self, req, resp, **params): resp.text = json.dumps(params) app = util.create_app( asgi, middleware=AccessParamsMiddleware(), independent_middleware=independent_middleware, ) app.add_route('/path/{id}', Resource()) client = testing.TestClient(app) response = client.simulate_request(path='/path/22') assert 'params' in context assert context['params'] assert context['params']['id'] == '22' assert response.json == {'added': True, 'id': '22'} class TestEmptySignatureMiddleware(TestMiddleware): def test_dont_need_params_in_signature(self, asgi, util): """ Verify that we don't need parameters in the process_* signatures (for side-effect-only middlewares, mostly). Makes no difference on py27 but does affect py36. https://github.com/falconry/falcon/issues/1254 """ util.create_app(asgi, middleware=EmptySignatureMiddleware()) class TestErrorHandling(TestMiddleware): def test_error_composed_before_resp_middleware_called(self, asgi, util): mw = CaptureResponseMiddleware() app = util.create_app(asgi, middleware=mw) app.add_route('/', MiddlewareClassResource()) client = testing.TestClient(app) response = client.simulate_request(path='/', method='POST') assert response.status == falcon.HTTP_403 assert mw.resp.status == response.status composed_body = json.loads(mw.resp.data.decode()) assert composed_body['title'] == response.status assert not mw.req_succeeded # NOTE(kgriffs): Sanity-check the other params passed to # process_response() assert isinstance(mw.req, falcon.Request) assert isinstance(mw.resource, MiddlewareClassResource) def test_http_status_raised_from_error_handler(self, asgi, util): mw = CaptureResponseMiddleware() app = util.create_app(asgi, middleware=mw) app.add_route('/', MiddlewareClassResource()) client = testing.TestClient(app) # NOTE(kgriffs): Use the old-style error handler signature to # ensure our shim for that works as expected. def _http_error_handler(error, req, resp, params): raise falcon.HTTPStatus(falcon.HTTP_201) async def _http_error_handler_async(req, resp, error, params): raise falcon.HTTPStatus(falcon.HTTP_201) h = _http_error_handler_async if asgi else _http_error_handler # NOTE(kgriffs): This will take precedence over the default # handler for facon.HTTPError. if asgi: # NOTE(vytas): The ASGI flavour supports no reordering shim. app.add_error_handler(falcon.HTTPError, h) else: with pytest.warns(DeprecatedWarning, match='deprecated signature'): app.add_error_handler(falcon.HTTPError, h) response = client.simulate_request(path='/', method='POST') assert response.status == falcon.HTTP_201 assert mw.resp.status == response.status class TestShortCircuiting(TestMiddleware): def setup_method(self, method): super(TestShortCircuiting, self).setup_method(method) def _make_client(self, asgi, util, independent_middleware=True): mw = [ RequestTimeMiddleware(), ResponseCacheMiddlware(), TransactionIdMiddleware(), ] app = util.create_app( asgi, middleware=mw, independent_middleware=independent_middleware ) app.add_route('/', MiddlewareClassResource()) app.add_route('/cached', MiddlewareClassResource()) app.add_route('/cached/resource', MiddlewareClassResource()) return testing.TestClient(app) def test_process_request_not_cached(self, asgi, util): response = self._make_client(asgi, util).simulate_get('/') assert response.status == falcon.HTTP_200 assert response.json == _EXPECTED_BODY assert 'transaction_id' in context assert 'resource_transaction_id' in context assert 'mid_time' in context assert 'end_time' in context @pytest.mark.parametrize('independent_middleware', [True, False]) def test_process_request_cached(self, asgi, util, independent_middleware): response = self._make_client(asgi, util, independent_middleware).simulate_get( '/cached' ) assert response.status == falcon.HTTP_200 assert response.json == ResponseCacheMiddlware.PROCESS_REQUEST_CACHED_BODY # NOTE(kgriffs): Since TransactionIdMiddleware was ordered after # ResponseCacheMiddlware, the response short-circuiting should have # skipped it. assert 'transaction_id' not in context assert 'resource_transaction_id' not in context # NOTE(kgriffs): RequestTimeMiddleware only adds this in # process_resource(), which should be skipped when # ResponseCacheMiddlware sets resp.completed = True in # process_request(). assert 'mid_time' not in context # NOTE(kgriffs): Short-circuiting does not affect process_response() assert 'end_time' in context @pytest.mark.parametrize('independent_middleware', [True, False]) def test_process_resource_cached(self, asgi, util, independent_middleware): response = self._make_client(asgi, util, independent_middleware).simulate_get( '/cached/resource' ) assert response.status == falcon.HTTP_200 assert response.json == ResponseCacheMiddlware.PROCESS_RESOURCE_CACHED_BODY # NOTE(kgriffs): This should be present because it is added in # process_request(), but the short-circuit does not occur until # process_resource(). assert 'transaction_id' in context # NOTE(kgriffs): Since TransactionIdMiddleware was ordered after # ResponseCacheMiddlware, the response short-circuiting should have # skipped it. assert 'resource_transaction_id' not in context # NOTE(kgriffs): RequestTimeMiddleware only adds this in # process_resource(), which will not be skipped in this case because # RequestTimeMiddleware is ordered before ResponseCacheMiddlware. assert 'mid_time' in context # NOTE(kgriffs): Short-circuiting does not affect process_response() assert 'end_time' in context class TestCORSMiddlewareWithAnotherMiddleware(TestMiddleware): @pytest.mark.parametrize( 'mw,allowed', [ (CaptureResponseMiddleware(), True), ([CaptureResponseMiddleware()], True), ((CaptureResponseMiddleware(),), True), (iter([CaptureResponseMiddleware()]), True), (falcon.CORSMiddleware(), False), ([falcon.CORSMiddleware()], False), ], ) def test_api_initialization_with_cors_enabled_and_middleware_param( self, mw, asgi, util, allowed ): if allowed: app = util.create_app(asgi, middleware=mw, cors_enable=True) app.add_route('/', TestCorsResource()) client = testing.TestClient(app) result = client.simulate_get(headers={'Origin': 'localhost'}) assert result.headers['Access-Control-Allow-Origin'] == '*' else: with pytest.raises(ValueError, match='CORSMiddleware'): app = util.create_app(asgi, middleware=mw, cors_enable=True) def test_async_postfix_method_must_be_coroutine(util): class FaultyComponentA: def process_request_async(self, req, resp): pass class FaultyComponentB: def process_resource_async(self, req, resp, resource, params): pass class FaultyComponentC: def process_response_async(self, req, resp, resource, req_succeeded): pass if util.HAS_CYTHON: pytest.skip(reason='Cythonized coroutine functions cannot be detected') for mw in (FaultyComponentA, FaultyComponentB, FaultyComponentC): with pytest.raises(falcon.errors.CompatibilityError): util.create_app(True, middleware=[mw()])