Working outside of request context in the FastAPI application unit testing

Working outside of request context in the FastAPI application unit testing

  

While converting Flask application to FastAPI, I got following error in unit tests:

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/testclient.py:633: in post
    return super().post(
/usr/local/python/3.11.7/lib/python3.11/site-packages/httpx/_client.py:1145: in post
    return self.request(
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/testclient.py:516: in request
    return super().request(
/usr/local/python/3.11.7/lib/python3.11/site-packages/httpx/_client.py:827: in request
    return self.send(request, auth=auth, follow_redirects=follow_redirects)
/usr/local/python/3.11.7/lib/python3.11/site-packages/httpx/_client.py:914: in send
    response = self._send_handling_auth(
/usr/local/python/3.11.7/lib/python3.11/site-packages/httpx/_client.py:942: in _send_handling_auth
    response = self._send_handling_redirects(
/usr/local/python/3.11.7/lib/python3.11/site-packages/httpx/_client.py:979: in _send_handling_redirects
    response = self._send_single_request(request)
/usr/local/python/3.11.7/lib/python3.11/site-packages/httpx/_client.py:1015: in _send_single_request
    response = transport.handle_request(request)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/testclient.py:398: in handle_request
    raise exc
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/testclient.py:395: in handle_request
    portal.call(self.app, scope, receive, send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/anyio/from_thread.py:288: in call
    return cast(T_Retval, self.start_task_soon(func, *args).result())
/usr/local/python/3.11.7/lib/python3.11/concurrent/futures/_base.py:456: in result
    return self.__get_result()
/usr/local/python/3.11.7/lib/python3.11/concurrent/futures/_base.py:401: in __get_result
    raise self._exception
/usr/local/python/3.11.7/lib/python3.11/site-packages/anyio/from_thread.py:217: in _call_func
    retval = await retval_or_awaitable
/usr/local/python/3.11.7/lib/python3.11/site-packages/fastapi/applications.py:1054: in __call__
    await super().__call__(scope, receive, send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/applications.py:123: in __call__
    await self.middleware_stack(scope, receive, send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/middleware/errors.py:186: in __call__
    raise exc
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/middleware/errors.py:164: in __call__
    await self.app(scope, receive, _send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/middleware/exceptions.py:65: in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/_exception_handler.py:64: in wrapped_app
    raise exc
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/_exception_handler.py:53: in wrapped_app
    await app(scope, receive, sender)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/routing.py:756: in __call__
    await self.middleware_stack(scope, receive, send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/routing.py:776: in app
    await route.handle(scope, receive, send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/routing.py:297: in handle
    await self.app(scope, receive, send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/routing.py:77: in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/_exception_handler.py:64: in wrapped_app
    raise exc
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/_exception_handler.py:53: in wrapped_app
    await app(scope, receive, sender)
/usr/local/python/3.11.7/lib/python3.11/site-packages/starlette/routing.py:72: in app
    response = await func(request)
/usr/local/python/3.11.7/lib/python3.11/site-packages/fastapi/routing.py:278: in app
    raw_response = await run_endpoint_function(
/usr/local/python/3.11.7/lib/python3.11/site-packages/fastapi/routing.py:191: in run_endpoint_function
    return await dependant.call(**values)
api/adapter/adapter.py:30: in process_email_notification_from_bank
    status_code=status.HTTP_400_BAD_REQUEST, content=request.json
/usr/local/python/3.11.7/lib/python3.11/site-packages/werkzeug/local.py:311: in __get__
    obj = instance._get_current_object()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    def _get_current_object() -> T:
        try:
            obj = local.get()
        except LookupError:
>           raise RuntimeError(unbound_message) from None
E           RuntimeError: Working outside of request context.
E           
E           This typically means that you attempted to use functionality that needed
E           an active HTTP request. Consult the documentation on testing for
E           information about how to avoid this problem.

/usr/local/python/3.11.7/lib/python3.11/site-packages/werkzeug/local.py:508: RuntimeError

This is how my test case looks like:

TEST_DATA = [...]

class MyTest:
    def setup_method(self):
        self.app = create_app()
        self.client = TestClient(self.app)

    @pytest.mark.parametrize(
        "name,messages,expected,status_code",
        TEST_DATA,
    )
    def test_SendMessageChains_ExpectCorrectValuesOfTransaction(
        self, name, messages, expected, status_code
    ):
        for message in messages:
            data = {"plain": message, "envelope": {"to": self.email_adapters[0].email}}
            response = self.client.post(
                "/adapter/email-gateway",
                headers={"Content-Type": "application/json"},
                data=json.dumps(data),
            )
            verify_status(response, status_code)

        transactions = models.transaction.Transaction.objects()
        assert len(transactions) == len(expected)
        for i, transaction in enumerate(transactions):
            self.validate(transaction, expected[i])

It fails in the only test which is parameterized with Pytest.

  1. What exactly does this error mean?
  2. Why a request context is required?
  3. Is it somehow related to parametrization in the Pytest? I observe this problem only in the parametrized test.

Answer

The error RuntimeError: Working outside of request context. indicates that some part of your code is trying to access the request context when it isn't available. This typically happens when you try to access request-specific data (like request.json) outside the actual request handling.

In your case, it seems to happen within the process_email_notification_from_bank function.

Here’s a way to fix it:

Assuming process_email_notification_from_bank is a function that is being called somewhere within your endpoint, you might need to refactor it to pass the JSON data explicitly rather than accessing request.json directly.

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
import json

app = FastAPI()

@app.post("/adapter/email-gateway")
async def process_email_notification_from_bank(request: Request):
    json_data = await request.json()
    return await handle_email_notification(json_data)

async def handle_email_notification(json_data):
    # Do something with json_data
    if some_condition:
        return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content=json_data)
    # Your other logic here

Testing: In your test, you don't need to change much if you're using TestClient correctly:

from fastapi.testclient import TestClient
import pytest
import json

TEST_DATA = [...]

class MyTest:
    def setup_method(self):
        self.app = create_app()
        self.client = TestClient(self.app)

    @pytest.mark.parametrize(
        "name,messages,expected,status_code",
        TEST_DATA,
    )
    def test_SendMessageChains_ExpectCorrectValuesOfTransaction(
        self, name, messages, expected, status_code
    ):
        for message in messages:
            data = {"plain": message, "envelope": {"to": self.email_adapters[0].email}}
            response = self.client.post(
                "/adapter/email-gateway",
                headers={"Content-Type": "application/json"},
                data=json.dumps(data),
            )
            assert response.status_code == status_code

        transactions = models.transaction.Transaction.objects()
        assert len(transactions) == len(expected)
        for i, transaction in enumerate(transactions):
            self.validate(transaction, expected[i])

The key is to ensure that the request object or any request-specific context is accessed within the appropriate context. In FastAPI, this means handling it within async functions and making sure not to pass around request objects or their methods directly outside the request scope. Instead, pass necessary data explicitly as function parameters.

© 2024 Dagalaxy. All rights reserved.