Publisher Testing#

If you are working with a Publisher object (either as a decorator or directly), you have several testing features available:

• In-memory TestClient
• Publishing locally with error propogation
• Checking the incoming message body

Base Application#

Let's take a look at a simple application example with a publisher as a decorator or as a direct call:

1
2
3
4
5
6

publisher = broker.publisher("another-topic")

@publisher
@broker.subscriber("test-topic")
async def handle() -> str:
    return "Hi!"

1
2
3
4
5
6

publisher = broker.publisher("another-queue")

@publisher
@broker.subscriber("test-queue")
async def handle() -> str:
    return "Hi!"

1
2
3
4
5
6

publisher = broker.publisher("another-subject")

@publisher
@broker.subscriber("test-subject")
async def handle() -> str:
    return "Hi!"

1
2
3
4
5

publisher = broker.publisher("another-topic")

@broker.subscriber("test-topic")
async def handle():
    await publisher.publish("Hi!")

1
2
3
4
5

publisher = broker.publisher("another-queue")

@broker.subscriber("test-queue")
async def handle():
    await publisher.publish("Hi!")

1
2
3
4
5

publisher = broker.publisher("another-subject")

@broker.subscriber("test-subject")
async def handle():
    await publisher.publish("Hi!")

Testing#

To test it, you just need to patch your broker with a special TestBroker.

1
2
3
4
5
6
7
8
9

import pytest

from faststream.kafka import TestKafkaBroker


@pytest.mark.asyncio
async def test_handle():
    async with TestKafkaBroker(broker) as br:
        await br.publish("", topic="test-topic")

1
2
3
4
5
6
7
8
9

import pytest

from faststream.rabbit import TestRabbitBroker


@pytest.mark.asyncio
async def test_handle():
    async with TestRabbitBroker(broker) as br:
        await br.publish("", queue="test-queue")

1
2
3
4
5
6
7
8
9

import pytest

from faststream.nats import TestNatsBroker


@pytest.mark.asyncio
async def test_handle():
    async with TestNatsBroker(broker) as br:
        await br.publish("", subject="test-subject")

By default, it patches you broker to run In-Memory, so you can use it without any external broker. It should be extremely usefull in your CI or local development environment.

Also, it allows you to check the outgoing message body in the same way as with a subscriber.

publisher.mock.assert_called_once_with("Hi!")

Additionally, TestBroker can be used with a real external broker to make your tests end-to-end suitable. For more information, please visit the subscriber testing page.