Skip to content

Events Testing#

In most cases you are testing your subscriber and 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
20
21
22
import pytest

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

broker = KafkaBroker()
app = FastStream(broker)


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


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

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

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

broker = KafkaBroker()
app = FastStream(broker)


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


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

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

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

broker = RabbitBroker()
app = FastStream(broker)


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


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

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

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

broker = NatsBroker()
app = FastStream(broker)


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


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

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

from faststream import FastStream, TestApp
from faststream.redis import RedisBroker, TestRedisBroker

broker = RedisBroker()
app = FastStream(broker)


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


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

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

from faststream import FastStream, TestApp
from faststream.mqtt import MQTTBroker, TestMQTTBroker

broker = MQTTBroker()
app = FastStream(broker)


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


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

Using with TestBroker#

If you want to use an In-Memory patched broker in your tests, it's advised to patch the broker first (before applying the application patch).

Also, TestApp and TestBroker are both calling broker.start(). According to the original logic, the broker should be started in the FastStream application, but if TestBroker is applied first – it breaks this behavior. For this reason, TestApp prevents TestBroker broker.start() call if it is placed inside the TestBroker context.

This behavior is controlled by the connect_only argument to TestBroker. While None is the default value, TestApp can set it to True or False during the code execution. If the connect_only argument is provided manually, it would not be changed.

Warning

With connect_only=False, all FastStream hooks will be called after the broker start, which can break some @app.on_startup logic.