Skip to content

Subscriber Testing#

Testability is a crucial part of any application, and FastStream provides you with the tools to test your code easily.

Original Application#

Let's take a look at the original application to test

annotation_kafka.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)


@broker.subscriber("test-topic")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1
annotation_confluent.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream
from faststream.confluent import KafkaBroker

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)


@broker.subscriber("test-topic")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1
annotation_rabbit.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream
from faststream.rabbit import RabbitBroker

broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = FastStream(broker)


@broker.subscriber("test-queue")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1
annotation_nats.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream
from faststream.nats import NatsBroker

broker = NatsBroker("nats://localhost:4222")
app = FastStream(broker)


@broker.subscriber("test-subject")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1
annotation_redis.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream
from faststream.redis import RedisBroker

broker = RedisBroker("redis://localhost:6379")
app = FastStream(broker)


@broker.subscriber("test-channel")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1
annotation_redis.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream
from faststream.mqtt import MQTTBroker

broker = MQTTBroker("localhost", port=1883)
app = FastStream(broker)


@broker.subscriber("test-topic")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1

It consumes JSON messages like { "name": "username", "user_id": 1 }

You can test your consume function like a regular one, for sure:

@pytest.mark.asyncio
async def test_handler():
    await handle("John", 1)

But if you want to test your function closer to your real runtime, you should use the special FastStream test client.

In-Memory Testing#

Deploying a whole service with a Message Broker is a bit too much just for testing purposes, especially in your CI environment. Not to mention the possible loss of messages due to network failures when working with real brokers.

For this reason, FastStream has a special TestClient to make your broker work in InMemory mode.

Just use it like a regular async context manager - all published messages will be routed in-memory (without any external dependencies) and consumed by the correct handler.

1
2
3
4
5
6
7
8
9
import pytest
from pydantic import ValidationError

from faststream.kafka import TestKafkaBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

1
2
3
4
5
6
7
8
9
import pytest
from pydantic import ValidationError

from faststream.confluent import TestKafkaBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

1
2
3
4
5
6
7
8
9
import pytest
from pydantic import ValidationError

from faststream.rabbit import TestRabbitBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRabbitBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, queue="test-queue")

1
2
3
4
5
6
7
8
9
import pytest
from pydantic import ValidationError

from faststream.nats import TestNatsBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestNatsBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, subject="test-subject")

1
2
3
4
5
6
7
8
9
import pytest
from pydantic import ValidationError

from faststream.redis import TestRedisBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRedisBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, channel="test-channel")

1
2
3
4
5
6
7
8
9
import pytest
from pydantic import ValidationError

from faststream.mqtt import TestMQTTBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestMQTTBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

Catching Exceptions#

This way you can catch any exceptions that occur inside your handler:

1
2
3
4
5
6
7
@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestKafkaBroker(broker) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", topic="test-topic")

        handle.mock.assert_called_once_with("wrong message")

1
2
3
4
5
6
7
@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestKafkaBroker(broker) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", topic="test-topic")

        handle.mock.assert_called_once_with("wrong message")

1
2
3
4
5
6
7
@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestRabbitBroker(broker) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", queue="test-queue")

        handle.mock.assert_called_once_with("wrong message")

1
2
3
4
5
6
7
@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestNatsBroker(broker) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", subject="test-subject")

        handle.mock.assert_called_once_with("wrong message")

1
2
3
4
5
6
7
@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestRedisBroker(broker) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", channel="test-channel")

        handle.mock.assert_called_once_with("wrong message")

1
2
3
4
5
6
7
@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestMQTTBroker(broker) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", topic="test-topic")

        handle.mock.assert_called_once_with("wrong message")

Full Example#

Let's look at a complete example of creating an app and testing it

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import pytest
from faststream.kafka import KafkaBroker, TestKafkaBroker

broker = KafkaBroker("localhost:9092")


@broker.subscriber("test-topic")
async def handle(msg: str) -> None:
    raise ValueError


@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        with pytest.raises(ValueError):
            await br.publish("hello!", "test-topic")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import pytest
from faststream.confluent import KafkaBroker, TestKafkaBroker

broker = KafkaBroker("localhost:9092")


@broker.subscriber("test-topic")
async def handle(msg: str) -> None:
    raise ValueError


@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        with pytest.raises(ValueError):
            await br.publish("hello!", "test-topic")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import pytest
from faststream.rabbit import RabbitBroker, TestRabbitBroker

broker = RabbitBroker("amqp://guest:guest@localhost:5672/")


@broker.subscriber("test-queue")
async def handle(msg: str) -> None:
    raise ValueError


@pytest.mark.asyncio()
async def test_handle() -> None:
    async with TestRabbitBroker(broker) as br:
        with pytest.raises(ValueError):
            await br.publish("hello!", "test-queue")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import pytest
from faststream.nats import NatsBroker, TestNatsBroker

broker = NatsBroker("nats://localhost:4222")


@broker.subscriber("test-subject")
async def handle(msg: str) -> None:
    raise ValueError


@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestNatsBroker(broker) as br:
        with pytest.raises(ValueError):
            await br.publish("hello!", "test-subject")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import pytest
from faststream.redis import RedisBroker, TestRedisBroker

broker = RedisBroker("redis://localhost:6379")


@broker.subscriber("test-channel")
async def handle(msg: str) -> None:
    raise ValueError


@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRedisBroker(broker) as br:
        with pytest.raises(ValueError):
            await br.publish("hello!", "test-channel")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import pytest
from faststream.mqtt import MQTTBroker, TestMQTTBroker

broker = MQTTBroker("localhost", port=1883)


@broker.subscriber("test-topic")
async def handle(msg: str) -> None:
    raise ValueError


@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestMQTTBroker(broker) as br:
        with pytest.raises(ValueError):
            await br.publish("hello!", "test-topic")

Validates Input#

Also, all handlers in test mode have an extra MagicMock object to validate passed arguments and call counts.

1
2
3
4
5
6
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

1
2
3
4
5
6
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

1
2
3
4
5
6
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRabbitBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, queue="test-queue")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

1
2
3
4
5
6
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestNatsBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, subject="test-subject")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

1
2
3
4
5
6
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRedisBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, channel="test-channel")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

1
2
3
4
5
6
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestMQTTBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

Note

The handle mock has a raw JSON message body. This way you can validate the incoming message itself and not a parsed python arguments.

Thus our example checks not mock.assert_called_with(name="John", user_id=1), but mock.assert_called_with({ "name": "John", "user_id": 1 }).

Scoping rule: the mock exists only inside the context manager. Once it exits, handle.mock raises a SetupError instead of answering for calls nobody made.

1
2
3
4
5
6
7
8
9
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

1
2
3
4
5
6
7
8
9
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

1
2
3
4
5
6
7
8
9
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRabbitBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, queue="test-queue")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

1
2
3
4
5
6
7
8
9
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestNatsBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, subject="test-subject")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

1
2
3
4
5
6
7
8
9
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRedisBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, channel="test-channel")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

1
2
3
4
5
6
7
8
9
@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestMQTTBroker(broker) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")

        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

Validates Message Fields#

Every handler also has an assert_called_once_with method. It checks the message body the same way mock.assert_called_once_with does, and beside it the message fields the handler saw: headers, correlation_id, reply_to, content_type and path.

Let's take an example of such an application:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from typing import Annotated

from pydantic import BaseModel

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

broker = KafkaBroker()
app = FastStream(broker)


class Data(BaseModel):
    name: str
    user_id: int


@broker.subscriber("test-topic")
async def handle(
    data: Data,
    trace_id: Annotated[str, Header("trace-id")],
) -> None:
    assert data.name == "John"
    assert data.user_id == 1
    assert trace_id == "42"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from typing import Annotated

from pydantic import BaseModel

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

broker = KafkaBroker()
app = FastStream(broker)


class Data(BaseModel):
    name: str
    user_id: int


@broker.subscriber("test-topic")
async def handle(
    data: Data,
    trace_id: Annotated[str, Header("trace-id")],
) -> None:
    assert data.name == "John"
    assert data.user_id == 1
    assert trace_id == "42"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from typing import Annotated

from pydantic import BaseModel

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

broker = RabbitBroker()
app = FastStream(broker)


class Data(BaseModel):
    name: str
    user_id: int


@broker.subscriber("test-queue")
async def handle(
    data: Data,
    trace_id: Annotated[str, Header("trace-id")],
) -> None:
    assert data.name == "John"
    assert data.user_id == 1
    assert trace_id == "42"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from typing import Annotated

from pydantic import BaseModel

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

broker = NatsBroker()
app = FastStream(broker)


class Data(BaseModel):
    name: str
    user_id: int


@broker.subscriber("test.subject")
async def handle(
    data: Data,
    trace_id: Annotated[str, Header("trace-id")],
) -> None:
    assert data.name == "John"
    assert data.user_id == 1
    assert trace_id == "42"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from typing import Annotated

from pydantic import BaseModel

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

broker = RedisBroker()
app = FastStream(broker)


class Data(BaseModel):
    name: str
    user_id: int


@broker.subscriber("test-channel")
async def handle(
    data: Data,
    trace_id: Annotated[str, Header("trace-id")],
) -> None:
    assert data.name == "John"
    assert data.user_id == 1
    assert trace_id == "42"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from typing import Annotated

from pydantic import BaseModel

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

broker = MQTTBroker()
app = FastStream(broker)


class Data(BaseModel):
    name: str
    user_id: int


@broker.subscriber("test-topic")
async def handle(
    data: Data,
    trace_id: Annotated[str, Header("trace-id")],
) -> None:
    assert data.name == "John"
    assert data.user_id == 1
    assert trace_id == "42"

Using assert_called_once_with, you can check the body and the headers in one statement. The body may be a plain dict or your model: it goes through the broker codec before the comparison, so both spellings mean the same message.

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

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

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            {"name": "John", "user_id": 1},
            headers={"trace-id": "42"},
        )
        # or
        await handle.assert_called_once_with(
            Data(name="John", user_id=1),
            headers={"trace-id": "42"},
        )

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

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

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            {"name": "John", "user_id": 1},
            headers={"trace-id": "42"},
        )
        # or
        await handle.assert_called_once_with(
            Data(name="John", user_id=1),
            headers={"trace-id": "42"},
        )

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

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

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRabbitBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            queue="test-queue",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            {"name": "John", "user_id": 1},
            headers={"trace-id": "42"},
        )
        # or
        await handle.assert_called_once_with(
            Data(name="John", user_id=1),
            headers={"trace-id": "42"},
        )

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

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

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestNatsBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            subject="test.subject",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            {"name": "John", "user_id": 1},
            headers={"trace-id": "42"},
        )
        # or
        await handle.assert_called_once_with(
            Data(name="John", user_id=1),
            headers={"trace-id": "42"},
        )

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

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

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRedisBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            channel="test-channel",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            {"name": "John", "user_id": 1},
            headers={"trace-id": "42"},
        )
        # or
        await handle.assert_called_once_with(
            Data(name="John", user_id=1),
            headers={"trace-id": "42"},
        )

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

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

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestMQTTBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            {"name": "John", "user_id": 1},
            headers={"trace-id": "42"},
        )
        # or
        await handle.assert_called_once_with(
            Data(name="John", user_id=1),
            headers={"trace-id": "42"},
        )

Headers match as a subset: FastStream adds its own headers (content-type, correlation_id) beside yours, and they never get in the way. Every other field matches exactly. When several fields differ, the AssertionError lists all of them at once.

A handler that saw several messages answers with two more methods, named as in unittest.mock and taking the same arguments: assert_called_with checks the last message, as the mock's does, and assert_any_call passes when any of the messages matches. When none does, the AssertionError lists every message the handler saw with its own mismatches.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import pytest
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_several_messages() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
            correlation_id="first",
        )
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
            correlation_id="second",
        )

        # the last message, as `mock.assert_called_with` reads it
        await handle.assert_called_with(
            Data(name="John", user_id=1),
            correlation_id="second",
        )
        # any of the messages
        await handle.assert_any_call(
            Data(name="John", user_id=1),
            correlation_id="first",
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import pytest
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_several_messages() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
            correlation_id="first",
        )
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
            correlation_id="second",
        )

        # the last message, as `mock.assert_called_with` reads it
        await handle.assert_called_with(
            Data(name="John", user_id=1),
            correlation_id="second",
        )
        # any of the messages
        await handle.assert_any_call(
            Data(name="John", user_id=1),
            correlation_id="first",
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import pytest
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_several_messages() -> None:
    async with TestRabbitBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            queue="test-queue",
            headers={"trace-id": "42"},
            correlation_id="first",
        )
        await br.publish(
            Data(name="John", user_id=1),
            queue="test-queue",
            headers={"trace-id": "42"},
            correlation_id="second",
        )

        # the last message, as `mock.assert_called_with` reads it
        await handle.assert_called_with(
            Data(name="John", user_id=1),
            correlation_id="second",
        )
        # any of the messages
        await handle.assert_any_call(
            Data(name="John", user_id=1),
            correlation_id="first",
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import pytest
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_several_messages() -> None:
    async with TestNatsBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            subject="test.subject",
            headers={"trace-id": "42"},
            correlation_id="first",
        )
        await br.publish(
            Data(name="John", user_id=1),
            subject="test.subject",
            headers={"trace-id": "42"},
            correlation_id="second",
        )

        # the last message, as `mock.assert_called_with` reads it
        await handle.assert_called_with(
            Data(name="John", user_id=1),
            correlation_id="second",
        )
        # any of the messages
        await handle.assert_any_call(
            Data(name="John", user_id=1),
            correlation_id="first",
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import pytest
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_several_messages() -> None:
    async with TestRedisBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            channel="test-channel",
            headers={"trace-id": "42"},
            correlation_id="first",
        )
        await br.publish(
            Data(name="John", user_id=1),
            channel="test-channel",
            headers={"trace-id": "42"},
            correlation_id="second",
        )

        # the last message, as `mock.assert_called_with` reads it
        await handle.assert_called_with(
            Data(name="John", user_id=1),
            correlation_id="second",
        )
        # any of the messages
        await handle.assert_any_call(
            Data(name="John", user_id=1),
            correlation_id="first",
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import pytest
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_several_messages() -> None:
    async with TestMQTTBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
            correlation_id="first",
        )
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
            correlation_id="second",
        )

        # the last message, as `mock.assert_called_with` reads it
        await handle.assert_called_with(
            Data(name="John", user_id=1),
            correlation_id="second",
        )
        # any of the messages
        await handle.assert_any_call(
            Data(name="John", user_id=1),
            correlation_id="first",
        )

To check only a part of the body, put a dirty-equals matcher in its place. Anything the fields above do not cover, such as the Kafka message key, lives in the context: check it through context by the same path you would give to Context(), walking attributes and dict keys from a context name.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
import pytest
from dirty_equals import IsPartialDict
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_message_context() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
            key=b"user-1",
        )

        await handle.assert_called_once_with(
            IsPartialDict(name="John"),
            context={"message.raw_message.key": b"user-1"},
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import pytest
from dirty_equals import IsPartialDict
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_message_context() -> None:
    async with TestKafkaBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            IsPartialDict(name="John"),
            context={"log_context.topic": "test-topic"},
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import pytest
from dirty_equals import IsPartialDict
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_message_context() -> None:
    async with TestRabbitBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            queue="test-queue",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            IsPartialDict(name="John"),
            context={"message.raw_message.routing_key": "test-queue"},
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import pytest
from dirty_equals import IsPartialDict
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_message_context() -> None:
    async with TestNatsBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            subject="test.subject",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            IsPartialDict(name="John"),
            context={"message.raw_message.subject": "test.subject"},
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import pytest
from dirty_equals import IsPartialDict
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_message_context() -> None:
    async with TestRedisBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            channel="test-channel",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            IsPartialDict(name="John"),
            context={"message.raw_message.channel": "test-channel"},
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import pytest
from dirty_equals import IsPartialDict
from pydantic import BaseModel

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

@pytest.mark.asyncio
async def test_message_context() -> None:
    async with TestMQTTBroker(broker) as br:
        await br.publish(
            Data(name="John", user_id=1),
            topic="test-topic",
            headers={"trace-id": "42"},
        )

        await handle.assert_called_once_with(
            IsPartialDict(name="John"),
            context={"message.raw_message.topic": "test-topic"},
        )

Note

A context path reads attributes and keys, it never calls. Where a raw message answers with methods, as the Confluent one does, reach for what FastStream put in the context beside it, such as log_context.

Note

Both handle.mock and the three assertion methods exist only inside the test broker. Outside of it they raise a SetupError instead of answering for a handler nobody has called.

Real Broker Testing#

If you want to test your application in a real environment, you shouldn't have to rewrite all your tests: just pass with_real optional parameter to your TestClient context manager. This way, TestClient supports all the testing features but uses an unpatched broker to send and consume messages.

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

from faststream.exceptions import SetupError
from faststream.kafka import TestKafkaBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker, with_real=True) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")
        await handle.wait_call(timeout=3)
        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestKafkaBroker(broker, with_real=True) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", topic="test-topic")
            await handle.wait_call(timeout=3)

        handle.mock.assert_called_once_with("wrong message")

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

from faststream.exceptions import SetupError
from faststream.confluent import TestKafkaBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestKafkaBroker(broker, with_real=True) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic-confluent")
        await handle.wait_call(timeout=30)
        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestKafkaBroker(broker, with_real=True) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", topic="test-topic-confluent")
            await handle.wait_call(timeout=30)

        handle.mock.assert_called_once_with("wrong message")

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

from faststream.exceptions import SetupError
from faststream.rabbit import TestRabbitBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRabbitBroker(broker, with_real=True) as br:
        await br.publish({"name": "John", "user_id": 1}, queue="test-queue")
        await handle.wait_call(timeout=3)
        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestRabbitBroker(broker, with_real=True) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", queue="test-queue")
            await handle.wait_call(timeout=3)

        handle.mock.assert_called_once_with("wrong message")

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

from faststream.exceptions import SetupError
from faststream.nats import TestNatsBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestNatsBroker(broker, with_real=True) as br:
        await br.publish({"name": "John", "user_id": 1}, subject="test-subject")
        await handle.wait_call(timeout=3)
        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestNatsBroker(broker, with_real=True) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", subject="test-subject")
            await handle.wait_call(timeout=3)

        handle.mock.assert_called_once_with("wrong message")

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

from faststream.exceptions import SetupError
from faststream.redis import TestRedisBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestRedisBroker(broker, with_real=True) as br:
        await br.publish({"name": "John", "user_id": 1}, channel="test-channel")
        await handle.wait_call(timeout=3)
        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestRedisBroker(broker, with_real=True) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", channel="test-channel")
            await handle.wait_call(timeout=3)

        handle.mock.assert_called_once_with("wrong message")

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

from faststream.exceptions import SetupError
from faststream.mqtt import TestMQTTBroker

@pytest.mark.asyncio
async def test_handle() -> None:
    async with TestMQTTBroker(broker, with_real=True) as br:
        await br.publish({"name": "John", "user_id": 1}, topic="test-topic")
        await handle.wait_call(timeout=3)
        handle.mock.assert_called_once_with({"name": "John", "user_id": 1})

    with pytest.raises(SetupError):  # the mock leaves with the test broker
        handle.mock.assert_not_called()

@pytest.mark.asyncio
async def test_validation_error() -> None:
    async with TestMQTTBroker(broker, with_real=True) as br:
        with pytest.raises(ValidationError):
            await br.publish("wrong message", topic="test-topic")
            await handle.wait_call(timeout=3)

        handle.mock.assert_called_once_with("wrong message")

Tip

When you're using a patched broker to test your consumers, the publish method is called synchronously with a consumer one, so you need not wait until your message is consumed. But in the real broker's case, it doesn't.

For this reason, you have to wait for message consumption manually with the special handler.wait_call(timeout) method. Also, inner handler exceptions will be raised in this function, not broker.publish(...).

A Little Tip#

It can be very useful to set the with_real flag using an environment variable. This way, you will be able to choose the testing mode right from the command line:

WITH_REAL=True/False pytest ...

To learn more about managing your application configuration visit this page.

What's Next#

The same patched broker also captures what your handlers publish, so you can assert on outgoing messages without a real broker. See Publisher Testing.

If your test needs the on_startup / on_shutdown hooks or the lifespan context to run, wrap the application in TestApp. See Events Testing.