How to setup TaskIQ for testing when building FastAPI applications

[devraj]

We build most of our applications using FastAPI and TaskIQ works beautifully with the integration. We also use the recommend method for testing FastAPI applications where we provide a test client as a fixture:

@pytest.fixture(scope='session')
def test_client():
    from fastapi.testclient import TestClient
    from labs.api import app
    return TestClient(app)

and then each test is able to use it to call the endpoints:

def test_signup(test_client, signup_request):

    response = test_client.post(
        "/signup", 
        json=signup_request.dict()
    )

    expected_response = SignupResponse(
        success=True,
        email=signup_request.email
    )

    parsed_body = SignupResponse.parse_obj(response.json())

    assert response.status_code == status.HTTP_201_CREATED
    assert parsed_body == expected_response

the quoted example spins off a TaskIQ based job to send a verification email. However when the handler gets to that portion, we get the following error:

======================================================== warnings summary ========================================================
tests/test_auth.py::test_signup
  /usr/local/lib/python3.11/site-packages/taskiq/abc/broker.py:81: TaskiqDeprecationWarning: Setting result backend with constructor is deprecated. Please use `with_result_backend` instead.
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
==================================================== short test summary info =====================================================
FAILED tests/test_auth.py::test_signup - taskiq.exceptions.SendTaskError
============================================= 1 failed, 5 passed, 1 warning in 0.74s =============================================
task: Failed to run task "dev:test": exit status 1

I assume this is because of the way FastAPI's test client executes the API endpoints when testing which would bypass the TaskIQ bootstrapping process. I am wondering what the solution could be where the test suite can invoke the TaskIQ jobs to achieve a wholistic test suite.

Any thoughts on this will be very welcome.

All files referenced above are available on this repository

[s3rius]

This happens because pytest reacts to warning. Consider switching to new BuilderAPI for brokers. Before, we used to pass result backends as parameters to brokers, but now we use builder pattern to configure parameters that are not related to brokers.

So this

broker = MyBroker(result_backend=MyResBackend())

Should become this:

broker = MyBroker().with_result_backend(MyResBackend())

Also, I'm creating a repo with example projects to show all possible pitfalls with different frameworks.

[s3rius]

Repository with examples: https://github.com/taskiq-python/examples. I'm going to add new examples very soon.

Also, currently we're updating documentation to add more details about framework integrations.

[devraj]

If I am able to contribute to the examples and docs I will.

Much appreciate your response.

[s3rius]

It's fully open to contributions. Please fork it and make PR.

Thanks for that.