[Lancetnik]

Stacked on #3111. Ticket 02 of the Call assertions plan.

A Kafka test wanted the record key by name. The only route was a context path into the raw message, context={"message.raw_message.key": b"..."}, which nothing completes in an editor, and which reads differently under aiokafka and confluent: an attribute on one, a method on the other.

Handlers and publishers of both Kafka packages now take key and partition in all three Call assertions, under the names their publish() uses:

await handle.assert_called_once_with(
    Data(name="John", user_id=1),
    key=b"user-1",
    partition=0,
)

Both compare exactly, as the client delivered them: key as bytes, nothing run through a serializer. A batch refuses them the way it refuses headers.

What changes

Endpoint reads the wrapper class it builds from a class attribute, _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 its class, and a Kafka field asked of a message another broker delivered raises SetupError naming Kafka.

Both Kafka LogicSubscriber and LogicPublisher set KafkaHandlerCallWrapper there and repeat their __call__ overloads with the return type narrowed to it, so mypy sees @broker.subscriber(...) return the Kafka wrapper. The class is internal and not exported.

KafkaCallAssertions in _internal/kafka declares the three methods with the two arguments and hands them to the recorder as BrokerFields: the values asked for and a reader (name, message) -> value. Each package supplies its reader from a new call_wrapper module: attributes on an aiokafka record, method calls on a confluent message. The spec placed the reader in testing.py, but that module imports the broker, which imports the subscriber, which needs the wrapper class at class-definition time. That is an import cycle, so the reader sits one module lower.

The in-memory confluent message is now a confluent_kafka.Message subclass, so the reader's type check holds under the in-memory broker as it does against a real client. Its headers() and error() signatures widened to match the base.

ADR-0010 has the reasoning for a class rather than **fields.

Tests and docs

• tests/brokers/kafka/test_test_client.py and its confluent twin: the three methods with key and partition on a handler and on a publisher, the mismatch text naming key, the batch refusal, and a handler shared with a Rabbit broker refusing a Kafka field for the Rabbit message.
• tests/mypy/kafka.py and tests/mypy/confluent.py assert_type a decorated handler and a published handler as KafkaHandlerCallWrapper, and call the three methods with the two arguments on a publisher.
• The AIOKafka and Confluent advanced_testing.py snippets gain test_kafka_fields, the docs runner calls it, and the "Validates Message Fields" section says each broker's handlers and publishers take that broker's own fields. The context example no longer reaches the key by path, since the key is a named argument now.

Tickets 03–06 (Rabbit, NATS, Redis, MQTT) branch from here.

🤖 Generated with Claude Code