Skip to content

feat(testing): RabbitMQ, NATS, Redis and MQTT handlers and publishers assert on their own fields - #3120

Draft
Lancetnik wants to merge 11 commits into
mainfrom
feat/557-broker-fields
Draft

Lancetnik wants to merge 11 commits into
mainfrom
feat/557-broker-fields

Conversation

@Lancetnik

@Lancetnik Lancetnik commented Sep 12, 2026 •

Copy link
Copy Markdown
Member

Stacked on #3113. Tickets 03–06 of the Call assertions plan, one broker per commit.

#3113 gave Kafka handlers and publishers key and partition as named arguments of the three Call assertions. The other four brokers now take their own fields the same way, under the names their publish() uses:

Broker Fields
RabbitMQ exchange, routing_key, message_id, priority
NATS subject
Redis channel, list, stream
MQTT topic, qos, retain
await handle.assert_called_once_with(
    Data(name="John", user_id=1),
    routing_key="test-queue",
)

Every field compares exactly, as the client delivered it. A batch refuses them the way it refuses headers, and a field asked of a message another broker delivered raises SetupError naming the broker.

What changes

A preparatory commit moves field_reader from _internal/kafka to _internal/testing/calls.py, beside BrokerFields, and gives it the broker's name as a parameter. The guard and its refusal text were Kafka's; now every wrapper shares them, and a broker is a wrapper class and a read callable.

Each broker commit adds faststream/<broker>/call_wrapper.py with the <Broker>CallAssertions mixin and the <Broker>HandlerCallWrapper, sets _call_wrapper_class on the subscriber base and the publisher, and narrows their __call__ to the wrapper, as Kafka does. The classes are internal and not exported.

  • RabbitMQ reads the four fields off the aio_pika IncomingMessage with getattr; the in-memory PatchedMessage is a subclass, so the guard holds under TestRabbitBroker.
  • NATS answers subject for a core or JetStream message; a key-value or object-store entry has none, so subject= on one raises SetupError naming the kind. The attribute and the overloads sit once on LogicSubscriber, inherited by all eight variants.
  • Redis keeps the user's three names and maps them inside the reader: the raw dict stores every address under channel, and its type says the kind. Naming the wrong kind (list= on a pub/sub message) raises SetupError saying which kind the message is and which name answers for it.
  • MQTT reads topic, qos and retain off the zmqtt message. The max_workers overloads on MQTTRegistrator.subscriber this needs landed in fix(mqtt): subscriber() is typed by max_workers, so a decorated handler stays typed #3123.

Neither RabbitMQ nor MQTT has batch consumption, so those two have no batch refusal test; the generic guard stays covered by Kafka and Redis.

Tests and docs

  • Each broker's test_test_client.py: the three methods with the fields on a handler and on a publisher, the mismatch text naming the field, the batch refusal where the broker has batches, the wrong-kind refusal (NATS stores, Redis kinds), and a handler shared with another broker refusing the field for that broker's message.
  • Each broker's tests/mypy/<broker>.py pins a decorated handler and a published handler as the broker's wrapper and calls the three methods on a publisher.
  • Each broker's advanced_testing.py snippet gains a test_<broker>_fields, the docs runner calls it, and the "Validates Message Fields" section names every broker's fields with a tab per broker.

🤖 Generated with Claude Code

@Lancetnik
Lancetnik added this pull request to stack #3112 September 12, 2026 12:26
@github-actions github-actions Bot added documentation Improvements or additions to documentation Confluent Issues related to `faststream.confluent` module AioKafka Issues related to `faststream.kafka` module NATS Issues related to `faststream.nats` module and NATS broker features Redis Issues related to `faststream.redis` module and Redis features MQTT Issues related to `faststream.mqtt` module labels Sep 12, 2026
@Lancetnik
Lancetnik marked this pull request as draft September 12, 2026 12:44
@Lancetnik
Lancetnik force-pushed the feat/557-broker-fields branch from 60b0e7f to 43ead48 Compare September 12, 2026 14:49
@Lancetnik
Lancetnik marked this pull request as ready for review September 12, 2026 14:50
@Lancetnik
Lancetnik marked this pull request as draft September 12, 2026 14:55
Lancetnik and others added 11 commits September 15, 2026 00:35
…tition`

`Endpoint` reads the wrapper class it builds from `_call_wrapper_class`, and
`ensure_call_wrapper` takes that class while still returning an already
wrapped handler untouched, so the first decorator on a shared handler decides.
Both Kafka packages set `KafkaHandlerCallWrapper` on their `LogicSubscriber`
and `LogicPublisher`, and repeat the `__call__` overloads with the return type
narrowed to it.

`KafkaCallAssertions` in `_internal/kafka` declares the three Call assertions
with `key=` and `partition=` and forwards them as Broker fields. `ExpectedCall`
carries the fields with a reader `(name, message) -> value` that each package
supplies from its `call_wrapper` module: attributes on an aiokafka record,
method calls on a confluent message, a `SetupError` naming Kafka for a raw
message of another broker. A batch refuses the two fields as it refuses
`headers`. The in-memory confluent message is now a `confluent_kafka.Message`
subclass, so the reader takes it for the client's own.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The AIOKafka and Confluent tabs gain a `test_kafka_fields` snippet, the
context example no longer reaches for the key by path, and the testing page
says each broker's handlers and publishers take that broker's own fields.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…eir reader

`ExpectedCall.broker_fields` is one `BrokerFields`: the values asked for, already
stripped of the ones left EMPTY, and the reader that answers them. The recorder
no longer filters EMPTY twice or guards a reader that may be missing, and the
Kafka mixin's helper is named for what it returns.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The confluent reader's `isinstance` guard had no test going red on its own.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
pyright spells an `async def`'s return as `CoroutineType[...]` where mypy says
`Coroutine[...]`, so an `assert_type` over an async handler cannot satisfy both
checkers; a sync handler reads the same in both.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…der`

Both Kafka packages built the same reader by hand: an isinstance guard, the same
refusal text, and one line that differs (an attribute on aiokafka, a method on
confluent). `field_reader(record, read)` in `_internal/kafka` builds it once, so
each `call_wrapper` module opens with its class and binds the reader inline. The
mixin declares the reader as a `ClassVar` and looks it up on the class, replacing
a staticmethod stub that only raised; a `LogicPublisher` takes the reader from its
wrapper class instead of restating the import.

The refusal names the message class a shared handler got (`RabbitMessage`), not
the raw client type. `ExpectedCall.message_fields()` introspects the dataclass
again instead of listing its fields by hand. The shared-handler tests carry
`require_aiopika`, as the other cross-broker tests do.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…he recorder

Every broker's wrapper guards its reader the same way: the raw message
must be the client's own class, or the assertion refuses it naming the
broker. The guard lived in `_internal/kafka` with "Kafka" spelled into
its text; it now takes the broker's name and lives with `BrokerFields`,
so the next broker is a wrapper class and a `read` callable.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
… `routing_key`, `message_id`, `priority`

`RabbitCallAssertions` in `faststream/rabbit/call_wrapper.py` declares the
three Call assertions with the four fields, under the names `publish()`
takes, and forwards them as Broker fields; `RabbitHandlerCallWrapper` beside
it binds the reader over the aio_pika `IncomingMessage`, which the in-memory
`PatchedMessage` subclasses, so a raw message of another broker is refused
with a `SetupError` naming RabbitMQ. `RabbitSubscriber` and `RabbitPublisher`
set the wrapper class and narrow their `__call__` return types to it, and the
publisher inherits the mixin with the same reader.

The Rabbit `advanced_testing.py` snippet publishes with `routing_key=` and
asserts it; the testing page gains the RabbitMQ tab and says which fields a
RabbitMQ handler or publisher takes.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
`NatsHandlerCallWrapper` and the NATS `LogicPublisher` take `subject=`
on the three Call assertions: the whole address a core or JetStream
message arrived on, which `path` does not carry. Every subscriber
variant returns the wrapper through the overloads on `LogicSubscriber`.
A key-value or object-store entry has no subject: asking for it raises
`SetupError` naming the kind; another broker's message is refused as a
non-NATS one.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…st`, `stream`

`RedisCallAssertions` gives the three Call assertions the names Redis's
`publish()` takes; `RedisHandlerCallWrapper` binds a reader that answers
the one matching the raw message's `type`, so `list=` on a pub/sub
message is refused naming the kind, and another broker's message is
refused naming Redis. Every subscriber variant inherits the wrapper class
and the narrowed `__call__` from `LogicSubscriber`; `LogicPublisher`
inherits the mixin, so every publisher variant takes the same names.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
… `retain`

`MqttCallAssertions` declares the three Call assertions with `topic=`, `qos=`
and `retain=`, and `MqttHandlerCallWrapper` binds their reader to the zmqtt
message: the concrete topic the message arrived on (`path` holds only the
template's captures), the `QoS` enum and the retain flag, as delivered. Both
live in `faststream/mqtt/call_wrapper.py`, since MQTT is one package. The
subscriber base and `MQTTPublisher` hand out that wrapper and narrow their
`__call__` to it; a raw message of another broker is refused naming MQTT.

The MQTT `advanced_testing.py` snippet asserts `topic=` and `qos=`, and the
testing page shows it beside the Kafka tabs.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Base automatically changed from feat/557-kafka-fields to main October 7, 2026 21:47

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

AioKafka Issues related to `faststream.kafka` module Confluent Issues related to `faststream.confluent` module documentation Improvements or additions to documentation MQTT Issues related to `faststream.mqtt` module NATS Issues related to `faststream.nats` module and NATS broker features Redis Issues related to `faststream.redis` module and Redis features

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants