Skip to content

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 propagation
  • 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-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
6
publisher = broker.publisher("another-channel")

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

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
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-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!")

1
2
3
4
5
publisher = broker.publisher("another-channel")

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

1
2
3
4
5
publisher = broker.publisher("another-topic")

@broker.subscriber("test-topic")
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
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
import pytest

from faststream.confluent 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
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
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")

1
2
3
4
5
6
7
8
import pytest

from faststream.redis import TestRedisBroker

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

1
2
3
4
5
6
7
8
import pytest

from faststream.mqtt import TestMQTTBroker

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

By default, it patches your broker to run In-Memory, so you can use it without any external broker. It should be extremely useful 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!")

In addition, the publisher has the same assert_called_once_with, assert_called_with and assert_any_call methods as a subscriber. They take the body as a dict, your model or a matcher, and the message fields the outgoing message carried. Here the publisher inherits the correlation_id of the message the handler consumed:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import pytest

from faststream.kafka import TestKafkaBroker

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

        await publisher.assert_called_once_with("Hi!", correlation_id="42")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import pytest

from faststream.confluent import TestKafkaBroker

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

        await publisher.assert_called_once_with("Hi!", correlation_id="42")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import pytest

from faststream.rabbit import TestRabbitBroker

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

        await publisher.assert_called_once_with("Hi!", correlation_id="42")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import pytest

from faststream.nats import TestNatsBroker

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

        await publisher.assert_called_once_with("Hi!", correlation_id="42")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import pytest

from faststream.redis import TestRedisBroker

@pytest.mark.asyncio
async def test_message_fields():
    async with TestRedisBroker(broker) as br:
        await br.publish("", channel="test-channel", correlation_id="42")

        await publisher.assert_called_once_with("Hi!", correlation_id="42")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import pytest

from faststream.mqtt import TestMQTTBroker

@pytest.mark.asyncio
async def test_message_fields():
    async with TestMQTTBroker(broker) as br:
        await br.publish("", topic="test-topic", correlation_id="42")

        await publisher.assert_called_once_with("Hi!", correlation_id="42")

Note

The Publisher mock contains not just a publish method input value. It sets up a virtual consumer for an outgoing topic, consumes a message, and stores this consumed one.

Note

In order for publishers to be properly patched by the test broker, you need to create them before running the test broker

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.

If a publisher is triggered from a lifespan hook rather than from a subscriber, the hooks have to run inside the test as well. TestApp does that: see Events Testing.