I am implementing exception handler logic to intercept RetryError
s occurring at service level.
For context, in that specific case, the second argument of the handler's signature: async def my_handler(request: Request, call_next)
is, in fact, not a callable, but the actual RetryError
.
My manual test with the application runtime seems to indicate my fix resolves the issue.
However, I am having considerable trouble adding non-E2E coverage for the change in the handler.
TL;DR My exact problem is that I cannot seem to make my test execute the handler function in test, whereas it does work at runtime.
Here's my test code:
@fixture(scope="session", name="exception_handler_test_client")
def get_test_client() -> TestClient:
app = FastAPI()
router = APIRouter()
# see below for the "service" function that raises RetryError
router.add_api_route("/test_exception_handler/", retryable_failure, methods=["GET"])
app.include_router(router)
# the real test would replace foo with the real exception handler I want to test
app.add_exception_handler(Exception, foo)
test_client = TestClient(router)
return test_client
async def foo(request: Request, call_next):
# here for convenience, the real one is somewhere else in the code base
# and neither the real one nor this one execute on exception raised
pass
# this does raise RetryError as expected
@retry(wait_fixed=100, stop_max_delay=1000, stop_max_attempt_number=2, retry_on_result=lambda Any: True)
def retryable_failure():
"""
Programmatically fails a retryable function for testing purposes.
"""
pass
# the actual test
@pytest.mark.asyncio
def test_retry_error_handled_without_hard_failure(exception_handler_test_client: TestClient) -> None:
# TODO once the handler works I would assert on the JSON response and HTTP status
exception_handler_test_client.get("/test_exception_handler/")
The test fails because it raises RetryError
as expected, but the handler is never executed.
I am aware of this post, describing a similar issue.
The only answer mentions using pytest.raises
, but that doesn't solve my problem at all - it just passes the test without executing the handler.
Extra note:
The exception handler associated with the app in test doesn't execute even if the programmatically failed retryable function is replaced with a garden-variety, non-retryable function just raising Exception.
In your get_test_client()
fixture, you are building your test_client
from the API router instance instead of the FastAPI app instance. The TestClient
constructor expects a FastAPI app to initialize itself. This would also explain why the exception handler is not getting executed, since you are adding the exception handler to the app
and not the router
object.
I suggest replacing the line to create your test client with this:
test_client = TestClient(app)
As a side note and like @Christos Palyvos suggested, Pytest encourages using yield
over return
in fixtures.