fix(testing): TestKafkaBroker should mock a real tombstone too

faststream/confluent/testing.py

@@ -279,7 +279,7 @@ async def _execute_handler(
 class MockConfluentMessage:
     def __init__(
         self,
-        raw_msg: bytes,
+        raw_msg: bytes | None,
         topic: str,
         key: bytes | str,
         headers: list[tuple[str, bytes]],
@@ -304,7 +304,7 @@ def __init__(
         self._timestamp = (timestamp_type, timestamp_ms)
 
     def len(self) -> int:
-        return len(self._raw_msg)
+        return 0 if self._raw_msg is None else len(self._raw_msg)
 
     def error(self) -> str | None:
         return self._error
@@ -327,7 +327,7 @@ def timestamp(self) -> tuple[int, int]:
     def topic(self) -> str:
         return self._topic
 
-    def value(self) -> bytes:
+    def value(self) -> bytes | None:
         return self._raw_msg
 
 
@@ -345,8 +345,12 @@ async def build_message(
     codec: Optional["CodecProto"] = None,
 ) -> MockConfluentMessage:
     """Build a mock confluent_kafka.Message for a sendable message."""
-    codec_instance = codec or DefaultCodec()
-    msg, content_type = await codec_instance.encode(message, serializer)
+    if message is None:
+        # keep a real tombstone (message.value() is None) distinct from b""
+        msg, content_type = None, None
+    else:
+        codec_instance = codec or DefaultCodec()
+        msg, content_type = await codec_instance.encode(message, serializer)
     k = key or b""
     headers = {
         "content-type": content_type or "",

faststream/kafka/testing.py

@@ -309,7 +309,12 @@ async def build_message(
     codec: Optional["CodecProto"] = None,
 ) -> "ConsumerRecord":
     """Build a Kafka ConsumerRecord for a sendable message."""
-    msg, content_type = await (codec or DefaultCodec()).encode(message, serializer)
+    if message is None and key is not None:
+        # keyed None is a real tombstone, matching publish()'s own rule
+        # (aiokafka needs a key or value, a keyless None still goes b"")
+        msg, content_type = None, None
+    else:
+        msg, content_type = await (codec or DefaultCodec()).encode(message, serializer)
 
     k = key or b""
 
@@ -328,8 +333,8 @@ async def build_message(
         partition=partition or 0,
         key=k,
         serialized_key_size=len(k),
-        serialized_value_size=len(msg),
-        checksum=sum(msg),
+        serialized_value_size=0 if msg is None else len(msg),
+        checksum=0 if msg is None else sum(msg),
         offset=0,
         headers=[(i, j.encode()) for i, j in headers.items()],
         timestamp_type=1,

tests/brokers/confluent/test_test_client.py

@@ -1,9 +1,9 @@
 import asyncio
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from faststream import AckPolicy, BaseMiddleware
+from faststream import AckPolicy, BaseMiddleware, Context
 from faststream.confluent.annotations import KafkaMessage
 from faststream.confluent.message import FAKE_CONSUMER
 from faststream.confluent.testing import FakeProducer
@@ -16,6 +16,22 @@
 @pytest.mark.confluent()
 @pytest.mark.asyncio()
 class TestTestclient(ConfluentMemoryTestcaseConfig, BrokerTestclientTestcase):
+    async def test_publish_none_tombstone(
+        self,
+        queue: str,
+        mock: MagicMock,
+    ) -> None:
+        broker = self.get_broker(apply_types=True)
+
+        @broker.subscriber(queue)
+        async def handler(msg=Context("message")) -> None:
+            mock(msg.raw_message.value())
+
+        async with self.patch_broker(broker) as br:
+            await br.publish(None, queue, key=b"tombstone-key")
+
+        mock.assert_called_once_with(None)
+
     async def test_message_nack_seek(self, queue: str) -> None:
         broker = self.get_broker(apply_types=True)
 

tests/brokers/kafka/test_test_client.py

@@ -1,9 +1,9 @@
 import asyncio
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from faststream import AckPolicy, BaseMiddleware
+from faststream import AckPolicy, BaseMiddleware, Context
 from faststream.kafka import TopicPartition
 from faststream.kafka.annotations import KafkaMessage
 from faststream.kafka.message import FAKE_CONSUMER
@@ -17,6 +17,22 @@
 @pytest.mark.kafka()
 @pytest.mark.asyncio()
 class TestTestclient(KafkaMemoryTestcaseConfig, BrokerTestclientTestcase):
+    async def test_publish_none_tombstone(
+        self,
+        queue: str,
+        mock: MagicMock,
+    ) -> None:
+        broker = self.get_broker(apply_types=True)
+
+        @broker.subscriber(queue)
+        async def handler(msg=Context("message")) -> None:
+            mock(msg.raw_message.value)
+
+        async with self.patch_broker(broker) as br:
+            await br.publish(None, queue, key=b"tombstone-key")
+
+        mock.assert_called_once_with(None)
+
     async def test_partition_match(
         self,
         queue: str,