Skip to content

Events Testing#

In the most cases you are testing your subsriber/publisher functions, but sometimes you need to trigger some lifespan hooks in your tests too.

For this reason, FastStream has a special TestApp patcher working as a regular async context manager.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
import pytest

from faststream import FastStream, TestApp
from faststream.kafka import KafkaBroker, TestKafkaBroker

app = FastStream(KafkaBroker())


@app.after_startup
async def handle():
    print("Calls in tests too!")


@pytest.mark.asyncio
async def test_lifespan():
    async with TestKafkaBroker(app.broker, connect_only=True):
        async with TestApp(app):
            # test something
            pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
import pytest

from faststream import FastStream, TestApp
from faststream.rabbit import RabbitBroker, TestRabbitBroker

app = FastStream(RabbitBroker())


@app.after_startup
async def handle():
    print("Calls in tests too!")


@pytest.mark.asyncio
async def test_lifespan():
    async with TestRabbitBroker(app.broker, connect_only=True):
        async with TestApp(app):
            # test something
            pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
import pytest

from faststream import FastStream, TestApp
from faststream.nats import NatsBroker, TestNatsBroker

app = FastStream(NatsBroker())


@app.after_startup
async def handle():
    print("Calls in tests too!")


@pytest.mark.asyncio
async def test_lifespan():
    async with TestNatsBroker(app.broker, connect_only=True):
        async with TestApp(app):
            # test something
            pass

Tip

If you are using a connected broker inside withing your lifespan hooks, it's advisable to patch the broker first (before applying the application patch).

Also, because FastStream calls broker.start() inside, you need to prevent TestClient broker starting to respect the original lifespan hooks ordering by connect_only=True option. Without this one, all FastStream hooks will be called after broker was started, what can breaks some @app.on_startup logic.

Last update: 2023-10-02