diff --git a/docs/docs/en/confluent/message.md b/docs/docs/en/confluent/message.md index 23e95f132e4..a5899730e20 100644 --- a/docs/docs/en/confluent/message.md +++ b/docs/docs/en/confluent/message.md @@ -26,6 +26,11 @@ This object serves as a unified **FastStream** wrapper around the native broker * `#!python topic(): str` * `#!python value(): Optional[Union[str, bytes]]` +!!! note + A `None` value() is a Kafka tombstone, the delete marker on a compacted topic. FastStream represents it as `faststream.message.TOMBSTONE`, kept distinct from an empty payload (`b""`). + + To publish a tombstone, pass `faststream.message.TOMBSTONE` as the message body, along with a key. `broker.publish(TOMBSTONE, key=b"...")` + For example, if you would like to access the headers of an incoming message, you would do so like this: ```python hl_lines="1 6" diff --git a/docs/docs/en/kafka/message.md b/docs/docs/en/kafka/message.md index 537903ecb4f..45476066744 100644 --- a/docs/docs/en/kafka/message.md +++ b/docs/docs/en/kafka/message.md @@ -31,6 +31,11 @@ This object serves as a unified **FastStream** wrapper around the native broker * `#!python topic: str` * `#!python value: Optional[aiokafka.structs.VT]` +!!! note + A `None` value is a Kafka tombstone, the delete marker on a compacted topic. FastStream represents it as `faststream.message.TOMBSTONE`, kept distinct from an empty payload (`b""`). + + To publish a tombstone, pass `faststream.message.TOMBSTONE` as the message body and a key. `broker.publish(TOMBSTONE, key=b"...")` + For example, if you would like to access the headers of an incoming message, you would do so like this: ```python hl_lines="1 6" diff --git a/faststream/_internal/fastapi/route.py b/faststream/_internal/fastapi/route.py index 61d985853ec..0e720ae917a 100644 --- a/faststream/_internal/fastapi/route.py +++ b/faststream/_internal/fastapi/route.py @@ -18,6 +18,7 @@ from faststream._internal.context import Context, ContextRepo from faststream._internal.types import P_HandlerParams, T_HandlerReturn from faststream.exceptions import SetupError +from faststream.message import TOMBSTONE from faststream.response import Response, ensure_response from ._compat import ( @@ -51,13 +52,13 @@ class StreamMessage(Request): scope: "dict[str, Any]" _cookies: "dict[str, Any]" _headers: "dict[str, Any]" # type: ignore[assignment] - _body: Union["dict[str, Any]", list[Any]] # type: ignore[assignment] + _body: Union["dict[str, Any]", list[Any], None] # type: ignore[assignment] _query_params: "dict[str, Any]" # type: ignore[assignment] def __init__( self, *, - body: Union["dict[str, Any]", list[Any]], + body: Union["dict[str, Any]", list[Any], None], headers: "dict[str, Any]", path: "dict[str, Any]", ) -> None: @@ -172,9 +173,15 @@ async def parsed_consumer(message: "NativeMessage[Any]") -> Any: """Wrapper, that parser FastStream message to FastAPI compatible one.""" body = await message.decode() - fastapi_body: dict[str, Any] | list[Any] + fastapi_body: dict[str, Any] | list[Any] | None if first_arg is not None: - if isinstance(body, dict): + # NOTE: checks the raw pre-decode body, not the already-decoded + # `body` local (also None here) - a real b"null" payload decodes + # to None too but must still fall through to {first_arg: body} + # below, unchanged from before this sentinel existed. + if message.body is TOMBSTONE: + fastapi_body, path = None, {} + elif isinstance(body, dict): path = fastapi_body = body or {} elif isinstance(body, list): fastapi_body, path = body, {} diff --git a/faststream/confluent/opentelemetry/provider.py b/faststream/confluent/opentelemetry/provider.py index 2d76606539a..fc9b748259b 100644 --- a/faststream/confluent/opentelemetry/provider.py +++ b/faststream/confluent/opentelemetry/provider.py @@ -5,6 +5,7 @@ from faststream._internal.types import MsgType from faststream.confluent.response import KafkaPublishCommand +from faststream.message import batch_body_size, body_size from faststream.opentelemetry import TelemetrySettingsProvider from faststream.opentelemetry.consts import MESSAGING_DESTINATION_PUBLISH_NAME @@ -53,7 +54,7 @@ def get_consume_attrs_from_message( SpanAttributes.MESSAGING_SYSTEM: self.messaging_system, SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id, SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id, - SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len(msg.body), + SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: body_size(msg.body), SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: msg.raw_message.partition(), SpanAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET: msg.raw_message.offset(), MESSAGING_DESTINATION_PUBLISH_NAME: msg.raw_message.topic(), @@ -84,8 +85,8 @@ def get_consume_attrs_from_message( SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id, SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id, SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT: len(msg.raw_message), - SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len( - bytearray().join(cast("Sequence[bytes]", msg.body)), + SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: batch_body_size( + msg.body ), SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: raw_message.partition(), MESSAGING_DESTINATION_PUBLISH_NAME: raw_message.topic(), diff --git a/faststream/confluent/parser.py b/faststream/confluent/parser.py index 35645046923..dfe78bf9153 100644 --- a/faststream/confluent/parser.py +++ b/faststream/confluent/parser.py @@ -1,6 +1,6 @@ from typing import TYPE_CHECKING, Any, cast -from faststream.message import StreamMessage, decode_message +from faststream.message import StreamMessage, decode_message, value_or_tombstone from .message import FAKE_CONSUMER, KafkaMessage @@ -38,7 +38,7 @@ async def parse_message( """Parses a Kafka message.""" headers = _parse_msg_headers(cast("_HeadersInput", message.headers() or ())) - body = message.value() or b"" + body = value_or_tombstone(message.value()) offset = message.offset() _, timestamp = message.timestamp() @@ -66,7 +66,7 @@ async def parse_batch( last = message[-1] for m in message: - body.append(m.value() or b"") + body.append(value_or_tombstone(m.value())) batch_headers.append( _parse_msg_headers(cast("_HeadersInput", m.headers() or ())) ) diff --git a/faststream/confluent/prometheus/provider.py b/faststream/confluent/prometheus/provider.py index bacde5e7665..f13d88a7d06 100644 --- a/faststream/confluent/prometheus/provider.py +++ b/faststream/confluent/prometheus/provider.py @@ -1,6 +1,7 @@ from collections.abc import Sequence from typing import TYPE_CHECKING, Union, cast +from faststream.message import batch_body_size, body_size from faststream.message.message import MsgType, StreamMessage from faststream.prometheus import ( ConsumeAttrs, @@ -33,7 +34,7 @@ def get_consume_attrs_from_message( ) -> ConsumeAttrs: return { "destination_name": cast("str", msg.raw_message.topic()), - "message_size": len(msg.body), + "message_size": body_size(msg.body), "messages_count": 1, } @@ -48,7 +49,7 @@ def get_consume_attrs_from_message( raw_message = msg.raw_message[0] return { "destination_name": cast("str", raw_message.topic()), - "message_size": len(bytearray().join(cast("Sequence[bytes]", msg.body))), + "message_size": batch_body_size(msg.body), "messages_count": len(msg.raw_message), } diff --git a/faststream/confluent/publisher/producer.py b/faststream/confluent/publisher/producer.py index 61cbb392813..1151f5709c3 100644 --- a/faststream/confluent/publisher/producer.py +++ b/faststream/confluent/publisher/producer.py @@ -9,11 +9,14 @@ from faststream.confluent.parser import AsyncConfluentParser from faststream.confluent.response import KafkaPublishCommand from faststream.exceptions import FeatureNotSupportedException +from faststream.message import TOMBSTONE, Tombstone +from faststream.message.utils import encode_or_tombstone, ensure_tombstone_key from .state import EmptyProducerState, ProducerState, RealProducer if TYPE_CHECKING: import asyncio + from collections.abc import Sequence from confluent_kafka import Message from fast_depends.library.serializer import SerializerProto @@ -139,10 +142,11 @@ async def publish( cmd: "KafkaPublishCommand", ) -> "asyncio.Future[Message | None] | Message | None": """Publish a message to a topic.""" - if cmd.body is None: - message, content_type = None, None - else: - message, content_type = await self.codec.encode(cmd.body, self.serializer) + if cmd.body is TOMBSTONE: + ensure_tombstone_key(cmd.key) + message, content_type = await encode_or_tombstone( + cmd.body, self.codec, self.serializer + ) headers_to_send = { "content-type": content_type or "", @@ -166,13 +170,23 @@ async def publish_batch(self, cmd: "KafkaPublishCommand") -> None: headers_to_send = cmd.headers_to_publish() + encoded_batch: Sequence[tuple[bytes | None, str | None]] if isinstance(self.codec, BatchCodecProto): + if any(isinstance(body, Tombstone) for body in cmd.batch_bodies): + msg = ( + "a tombstone in a batch isn't supported with a custom BatchCodecProto" + ) + raise ValueError(msg) encoded_batch = await self.codec.encode_batch( cmd.batch_bodies, self.serializer ) else: + for message_position, body in enumerate(cmd.batch_bodies): + if body is TOMBSTONE: + ensure_tombstone_key(cmd.key_for(message_position)) encoded_batch = [ - await self.codec.encode(msg, self.serializer) for msg in cmd.batch_bodies + await encode_or_tombstone(msg, self.codec, self.serializer) + for msg in cmd.batch_bodies ] for message_position, (message, content_type) in enumerate(encoded_batch): diff --git a/faststream/confluent/testing.py b/faststream/confluent/testing.py index f84f86e8e00..6e2bfa95ddd 100644 --- a/faststream/confluent/testing.py +++ b/faststream/confluent/testing.py @@ -21,7 +21,8 @@ from faststream.confluent.schemas import TopicPartition from faststream.confluent.subscriber.usecase import BatchSubscriber from faststream.exceptions import SubscriberNotFound -from faststream.message import gen_cor_id +from faststream.message import TOMBSTONE, Tombstone, gen_cor_id +from faststream.message.utils import encode_or_tombstone, ensure_tombstone_key if TYPE_CHECKING: from fast_depends.library.serializer import SerializerProto @@ -194,11 +195,21 @@ async def publish_batch(self, cmd: "KafkaPublishCommand") -> None: """Publish a batch of messages to the Kafka broker.""" serializer = self.broker.config.fd_config._serializer + encoded: Sequence[tuple[bytes | None, str | None]] if isinstance(self.codec, BatchCodecProto): + if any(isinstance(body, Tombstone) for body in cmd.batch_bodies): + msg = ( + "a tombstone in a batch isn't supported with a custom BatchCodecProto" + ) + raise ValueError(msg) encoded = await self.codec.encode_batch(cmd.batch_bodies, serializer) else: + for message_position, body in enumerate(cmd.batch_bodies): + if body is TOMBSTONE: + ensure_tombstone_key(cmd.key_for(message_position)) encoded = [ - await self.codec.encode(body, serializer) for body in cmd.batch_bodies + await encode_or_tombstone(body, self.codec, serializer) + for body in cmd.batch_bodies ] for handler in _find_handler( @@ -332,7 +343,7 @@ def value(self) -> bytes | None: async def build_message( - message: "SendableMessage", + message: "SendableMessage | Tombstone", topic: str, *, correlation_id: str | None = None, @@ -345,12 +356,11 @@ async def build_message( codec: Optional["CodecProto"] = None, ) -> MockConfluentMessage: """Build a mock confluent_kafka.Message for a sendable message.""" - 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) + if message is TOMBSTONE: + ensure_tombstone_key(key) + msg, content_type = await encode_or_tombstone( + message, codec or DefaultCodec(), serializer + ) k = key or b"" headers = { "content-type": content_type or "", @@ -373,7 +383,7 @@ async def build_message( def _build_mock_message( - body: bytes, + body: bytes | None, content_type: str | None, topic: str, partition: int | None = None, diff --git a/faststream/kafka/opentelemetry/provider.py b/faststream/kafka/opentelemetry/provider.py index 52435016256..c773d48541a 100644 --- a/faststream/kafka/opentelemetry/provider.py +++ b/faststream/kafka/opentelemetry/provider.py @@ -5,6 +5,7 @@ from faststream._internal.types import MsgType from faststream.kafka.response import KafkaPublishCommand +from faststream.message import batch_body_size, body_size from faststream.opentelemetry import TelemetrySettingsProvider from faststream.opentelemetry.consts import MESSAGING_DESTINATION_PUBLISH_NAME @@ -59,7 +60,7 @@ def get_consume_attrs_from_message( SpanAttributes.MESSAGING_SYSTEM: self.messaging_system, SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id, SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id, - SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len(msg.body), + SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: body_size(msg.body), SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: msg.raw_message.partition, SpanAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET: msg.raw_message.offset, MESSAGING_DESTINATION_PUBLISH_NAME: msg.raw_message.topic, @@ -90,8 +91,8 @@ def get_consume_attrs_from_message( SpanAttributes.MESSAGING_SYSTEM: self.messaging_system, SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id, SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id, - SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len( - bytearray().join(cast("Sequence[bytes]", msg.body)), + SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: batch_body_size( + msg.body ), SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT: len(msg.raw_message), SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: raw_message.partition, diff --git a/faststream/kafka/parser.py b/faststream/kafka/parser.py index 018104a8907..1ae869686fb 100644 --- a/faststream/kafka/parser.py +++ b/faststream/kafka/parser.py @@ -7,7 +7,7 @@ KafkaMessage, KafkaRawMessage, ) -from faststream.message import decode_message +from faststream.message import decode_message, value_or_tombstone if TYPE_CHECKING: from re import Pattern @@ -42,7 +42,7 @@ async def parse_message( headers = {i: j.decode() for i, j in message.headers} return self.msg_class( - body=message.value or b"", + body=value_or_tombstone(message.value), headers=headers, reply_to=headers.get("reply_to", ""), content_type=headers.get("content-type"), @@ -74,7 +74,7 @@ async def parse_batch( last = message[-1] for m in message: - body.append(m.value or b"") + body.append(value_or_tombstone(m.value)) batch_headers.append({i: j.decode() for i, j in m.headers}) headers = next(iter(batch_headers), {}) diff --git a/faststream/kafka/prometheus/provider.py b/faststream/kafka/prometheus/provider.py index 903d9d4154a..4e7be0b71c5 100644 --- a/faststream/kafka/prometheus/provider.py +++ b/faststream/kafka/prometheus/provider.py @@ -1,6 +1,7 @@ from collections.abc import Sequence -from typing import TYPE_CHECKING, Union, cast +from typing import TYPE_CHECKING, Union +from faststream.message import batch_body_size, body_size from faststream.message.message import MsgType, StreamMessage from faststream.prometheus import MetricsSettingsProvider @@ -31,7 +32,7 @@ def get_consume_attrs_from_message( ) -> "ConsumeAttrs": return { "destination_name": msg.raw_message.topic, - "message_size": len(msg.body), + "message_size": body_size(msg.body), "messages_count": 1, } @@ -46,7 +47,7 @@ def get_consume_attrs_from_message( raw_message = msg.raw_message[0] return { "destination_name": raw_message.topic, - "message_size": len(bytearray().join(cast("Sequence[bytes]", msg.body))), + "message_size": batch_body_size(msg.body), "messages_count": len(msg.raw_message), } diff --git a/faststream/kafka/publisher/producer.py b/faststream/kafka/publisher/producer.py index ef7c5aa6fe0..8dcc83c859c 100644 --- a/faststream/kafka/publisher/producer.py +++ b/faststream/kafka/publisher/producer.py @@ -11,11 +11,14 @@ from faststream.kafka.message import KafkaMessage from faststream.kafka.parser import AioKafkaParser from faststream.kafka.response import KafkaPublishCommand +from faststream.message import TOMBSTONE, Tombstone +from faststream.message.utils import encode_or_tombstone, ensure_tombstone_key from .state import EmptyProducerState, ProducerState, RealProducer if TYPE_CHECKING: import asyncio + from collections.abc import Sequence from aiokafka import AIOKafkaProducer from aiokafka.structs import RecordMetadata @@ -110,12 +113,11 @@ async def publish( cmd: "KafkaPublishCommand", ) -> Union["asyncio.Future[RecordMetadata]", "RecordMetadata"]: """Publish a message to a topic.""" - if cmd.body is None and cmd.key is not None: - # keyed None is a tombstone: aiokafka requires at least key or value, - # so a keyless None still goes through the codec as b"" - message, content_type = None, None - else: - message, content_type = await self.codec.encode(cmd.body, self.serializer) + if cmd.body is TOMBSTONE: + ensure_tombstone_key(cmd.key) + message, content_type = await encode_or_tombstone( + cmd.body, self.codec, self.serializer + ) headers_to_send = { "content-type": content_type or "", @@ -145,13 +147,22 @@ async def publish_batch( headers_to_send = cmd.headers_to_publish() + encoded_batch: Sequence[tuple[bytes | None, str | None]] if isinstance(self.codec, BatchCodecProto): + if any(isinstance(body, Tombstone) for body in cmd.batch_bodies): + msg = ( + "a tombstone in a batch isn't supported with a custom BatchCodecProto" + ) + raise ValueError(msg) encoded_batch = await self.codec.encode_batch( cmd.batch_bodies, self.serializer ) else: + for message_position, body in enumerate(cmd.batch_bodies): + if body is TOMBSTONE: + ensure_tombstone_key(cmd.key_for(message_position)) encoded_batch = [ - await self.codec.encode(body, self.serializer) + await encode_or_tombstone(body, self.codec, self.serializer) for body in cmd.batch_bodies ] diff --git a/faststream/kafka/testing.py b/faststream/kafka/testing.py index c1c902fb229..33c0fefd9bb 100755 --- a/faststream/kafka/testing.py +++ b/faststream/kafka/testing.py @@ -24,7 +24,8 @@ from faststream.kafka.publisher.producer import AioKafkaFastProducer from faststream.kafka.publisher.usecase import BatchPublisher from faststream.kafka.subscriber.usecase import BatchSubscriber -from faststream.message import gen_cor_id +from faststream.message import TOMBSTONE, Tombstone, gen_cor_id +from faststream.message.utils import encode_or_tombstone, ensure_tombstone_key if TYPE_CHECKING: from fast_depends.library.serializer import SerializerProto @@ -243,11 +244,21 @@ async def publish_batch( """Publish a batch of messages to the Kafka broker.""" serializer = self.broker.config.fd_config._serializer + encoded: Sequence[tuple[bytes | None, str | None]] if isinstance(self.codec, BatchCodecProto): + if any(isinstance(body, Tombstone) for body in cmd.batch_bodies): + msg = ( + "a tombstone in a batch isn't supported with a custom BatchCodecProto" + ) + raise ValueError(msg) encoded = await self.codec.encode_batch(cmd.batch_bodies, serializer) else: + for message_position, body in enumerate(cmd.batch_bodies): + if body is TOMBSTONE: + ensure_tombstone_key(cmd.key_for(message_position)) encoded = [ - await self.codec.encode(body, serializer) for body in cmd.batch_bodies + await encode_or_tombstone(body, self.codec, serializer) + for body in cmd.batch_bodies ] for handler in _find_handler( @@ -296,7 +307,7 @@ async def _execute_handler( async def build_message( - message: "SendableMessage", + message: "SendableMessage | Tombstone", topic: str, partition: int | None = None, timestamp_ms: int | None = None, @@ -309,12 +320,11 @@ async def build_message( codec: Optional["CodecProto"] = None, ) -> "ConsumerRecord": """Build a Kafka ConsumerRecord for a sendable message.""" - 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) + if message is TOMBSTONE: + ensure_tombstone_key(key) + msg, content_type = await encode_or_tombstone( + message, codec or DefaultCodec(), serializer + ) k = key or b"" @@ -343,7 +353,7 @@ async def build_message( def _build_record( - body: bytes, + body: bytes | None, content_type: str | None, topic: str, partition: int | None = None, @@ -367,8 +377,8 @@ def _build_record( partition=partition or 0, key=k, serialized_key_size=len(k), - serialized_value_size=len(body), - checksum=sum(body), + serialized_value_size=0 if body is None else len(body), + checksum=0 if body is None else sum(body), offset=0, headers=[(i, j.encode()) for i, j in h.items()], timestamp_type=1, diff --git a/faststream/message/__init__.py b/faststream/message/__init__.py index 2dd53d6c4e8..497cb8a980b 100644 --- a/faststream/message/__init__.py +++ b/faststream/message/__init__.py @@ -1,12 +1,26 @@ from .message import AckStatus, StreamMessage from .source_type import SourceType -from .utils import decode_message, encode_message, gen_cor_id +from .utils import ( + TOMBSTONE, + Tombstone, + batch_body_size, + body_size, + decode_message, + encode_message, + gen_cor_id, + value_or_tombstone, +) __all__ = ( + "TOMBSTONE", "AckStatus", "SourceType", "StreamMessage", + "Tombstone", + "batch_body_size", + "body_size", "decode_message", "encode_message", "gen_cor_id", + "value_or_tombstone", ) diff --git a/faststream/message/utils.py b/faststream/message/utils.py index 13d162756ce..e494b424046 100644 --- a/faststream/message/utils.py +++ b/faststream/message/utils.py @@ -2,7 +2,7 @@ import json from collections.abc import Sequence from contextlib import suppress -from typing import TYPE_CHECKING, Any, Optional, Union, cast +from typing import TYPE_CHECKING, Any, Final, Optional, Union, cast from uuid import uuid4 from faststream._internal._compat import json_dumps, json_loads @@ -12,6 +12,7 @@ from fast_depends.library.serializer import SerializerProto from faststream._internal.basic_types import DecodedMessage, SendableMessage + from faststream._internal.parser import CodecProto from .message import StreamMessage @@ -21,9 +22,63 @@ def gen_cor_id() -> str: return str(uuid4()) +# NOTE: sentinel for a genuine null message body, distinct from b"" or b"null" +class Tombstone: + __slots__ = () + + def __repr__(self) -> str: + return "TOMBSTONE" + + def __bool__(self) -> bool: + return False + + +TOMBSTONE: Final = Tombstone() + + +# NOTE: body/bodies stay Any - StreamMessage.body is `bytes | Any` even for +# a batch message (a Sequence[bytes] at runtime), so a precise element type +# here doesn't match real call sites. isinstance (not `is TOMBSTONE`) is +# still required so len(body) narrows correctly in the non-tombstone case. +def body_size(body: Any) -> int: + return 0 if isinstance(body, Tombstone) else len(body) + + +def batch_body_size(bodies: Sequence[Any]) -> int: + return sum(body_size(b) for b in bodies) + + +def value_or_tombstone(value: bytes | None) -> "bytes | Tombstone": + return TOMBSTONE if value is None else value + + +def ensure_tombstone_key(key: "bytes | str | None") -> None: + if key is None: + msg = "a Kafka tombstone requires a key" + raise ValueError(msg) + + +async def encode_or_tombstone( + message: "SendableMessage | Tombstone", + codec: "CodecProto", + serializer: Optional["SerializerProto"], +) -> tuple[bytes | None, str | None]: + # NOTE: isinstance, not `is` - see body_size above for why. + if isinstance(message, Tombstone): + return None, None + return await codec.encode(message, serializer) + + def decode_message(message: "StreamMessage[Any]") -> "DecodedMessage": """Decodes a message.""" body: Any = getattr(message, "body", message) + + # NOTE: message.body is TOMBSTONE only for a genuine wire-level null + # value (see parser value_or_tombstone) - a real b"null" payload never + # hits this branch and decodes normally below. + if body is TOMBSTONE: + return None + m: DecodedMessage = body if content_type := getattr(message, "content_type", False): diff --git a/tests/brokers/base/fastapi.py b/tests/brokers/base/fastapi.py index d2b550bec84..cf95c50b365 100644 --- a/tests/brokers/base/fastapi.py +++ b/tests/brokers/base/fastapi.py @@ -7,6 +7,7 @@ from fastapi import BackgroundTasks, Depends, FastAPI, Header from fastapi.exceptions import RequestValidationError from fastapi.testclient import TestClient +from pydantic import BaseModel from faststream import ( Context as FSContext, @@ -19,12 +20,17 @@ from faststream._internal.fastapi.route import StreamMessage from faststream._internal.fastapi.router import StreamRouter from faststream.exceptions import SetupError +from faststream.message import TOMBSTONE from .basic import BaseTestcaseConfig Broker = TypeVar("Broker", bound=BrokerUsecase) +class _Foo(BaseModel): + x: int + + @pytest.mark.asyncio() class FastAPITestcase(BaseTestcaseConfig): router_class: type[StreamRouter[BrokerUsecase]] @@ -274,6 +280,44 @@ async def subscriber(msg: StreamMessage) -> None: mock.assert_called_once_with(True) +# NOTE: fake-broker only (uses the in-memory TestKafkaBroker/TestConfluentBroker +# so publish() drives the handler synchronously) - rabbit/nats/redis/mqtt +# inherit FastAPILocalTestcase too and don't support a real tombstone or a +# publish `key`, and a real connected broker doesn't propagate the handler's +# validation error back to the awaiting publish() call the way the fake one does. +@pytest.mark.asyncio() +class KafkaTombstoneFastAPILocalTestcase(BaseTestcaseConfig): + router_class: type[StreamRouter[BrokerUsecase]] + + async def test_optional_body_resolves_to_none_for_tombstone( + self, + queue: str, + ) -> None: + router = self.router_class() + received: list[_Foo | None] = [] + + args, kwargs = self.get_subscriber_params(queue) + + @router.subscriber(*args, **kwargs) + async def handler(msg: _Foo | None = None) -> None: + received.append(msg) + + app = FastAPI() + app.include_router(router) + + async with self.patch_broker(router.broker) as br: + with TestClient(app): + await br.publish(b'{"x": 5}', queue, key=b"k1") + await br.publish(TOMBSTONE, queue, key=b"k2") + + # a genuinely empty (non-null) body must still fail + # validation, not silently resolve like a tombstone. + with pytest.raises(RequestValidationError): + await br.publish(b"", queue, key=b"k3") + + assert received == [_Foo(x=5), None] + + @pytest.mark.asyncio() class FastAPILocalTestcase(BaseTestcaseConfig): router_class: type[StreamRouter[BrokerUsecase]] diff --git a/tests/brokers/confluent/test_consume.py b/tests/brokers/confluent/test_consume.py index 2418aef3d17..e3cbbb4f498 100644 --- a/tests/brokers/confluent/test_consume.py +++ b/tests/brokers/confluent/test_consume.py @@ -348,3 +348,65 @@ async def handler(msg: Any) -> None: ) assert mock.call_count == 2, mock.call_count + + @pytest.mark.asyncio() + async def test_consume_without_value( + self, + mock: MagicMock, + queue: str, + event: asyncio.Event, + ) -> None: + consume_broker = self.get_broker() + + args, kwargs = self.get_subscriber_params(queue) + + @consume_broker.subscriber(*args, **kwargs) + async def handler(msg: bytes | None) -> None: + event.set() + mock(msg) + + async with self.patch_broker(consume_broker) as br: + await br.start() + + await asyncio.wait( + ( + asyncio.create_task( + br._producer._producer.producer.send(queue, key=b""), + ), + asyncio.create_task(event.wait()), + ), + timeout=self.timeout, + ) + + mock.assert_called_once_with(None) + + @pytest.mark.asyncio() + async def test_consume_batch_without_value( + self, + mock: MagicMock, + queue: str, + event: asyncio.Event, + ) -> None: + consume_broker = self.get_broker() + + args, kwargs = self.get_subscriber_params(queue, batch=True) + + @consume_broker.subscriber(*args, **kwargs) + async def handler(msg: list[bytes | None]) -> None: + event.set() + mock(msg) + + async with self.patch_broker(consume_broker) as br: + await br.start() + + await asyncio.wait( + ( + asyncio.create_task( + br._producer._producer.producer.send(queue, key=b""), + ), + asyncio.create_task(event.wait()), + ), + timeout=self.timeout, + ) + + mock.assert_called_once_with([None]) diff --git a/tests/brokers/confluent/test_fastapi.py b/tests/brokers/confluent/test_fastapi.py index 6979feebf78..87bed6eded8 100644 --- a/tests/brokers/confluent/test_fastapi.py +++ b/tests/brokers/confluent/test_fastapi.py @@ -3,9 +3,18 @@ import pytest +from faststream import Context from faststream.confluent import KafkaRouter -from faststream.confluent.fastapi import KafkaRouter as StreamRouter -from tests.brokers.base.fastapi import FastAPILocalTestcase, FastAPITestcase +from faststream.confluent.fastapi import ( + KafkaMessage, + KafkaRouter as StreamRouter, +) +from tests.brokers.base.fastapi import ( + FastAPILocalTestcase, + FastAPITestcase, + KafkaTombstoneFastAPILocalTestcase, + _Foo, +) from .basic import ConfluentMemoryTestcaseConfig, ConfluentTestcaseConfig @@ -41,9 +50,46 @@ async def hello(msg: list[str]): assert event.is_set() mock.assert_called_with(["hi"]) + async def test_external_tombstone_resolves_to_none( + self, + queue: str, + event: asyncio.Event, + ) -> None: + router = self.router_class() + received: list[tuple[object, bytes | None]] = [] + + args, kwargs = self.get_subscriber_params(queue) + + @router.subscriber(*args, **kwargs) + async def handler( + msg: _Foo | None = None, + raw: KafkaMessage = Context("message"), + ) -> None: + received.append((msg, raw.raw_message.value())) + if len(received) == 2: + event.set() + + async with self.patch_broker(router.broker) as br: + await br.start() + + await br.publish(b'{"x": 5}', queue, key=b"k1") + + raw_producer = br._producer._producer.producer + await raw_producer.send(topic=queue, key=b"k2", value=None) + + await asyncio.wait_for(event.wait(), timeout=self.timeout) + + assert len(received) == 2 + assert (_Foo(x=5), b'{"x": 5}') in received + assert (None, None) in received + @pytest.mark.confluent() -class TestRouterLocal(ConfluentMemoryTestcaseConfig, FastAPILocalTestcase): +class TestRouterLocal( + ConfluentMemoryTestcaseConfig, + FastAPILocalTestcase, + KafkaTombstoneFastAPILocalTestcase, +): router_class = StreamRouter broker_router_class = KafkaRouter diff --git a/tests/brokers/confluent/test_parser.py b/tests/brokers/confluent/test_parser.py index 2600cf1970c..53f9d75775e 100644 --- a/tests/brokers/confluent/test_parser.py +++ b/tests/brokers/confluent/test_parser.py @@ -1,5 +1,9 @@ +from unittest.mock import MagicMock + import pytest +from faststream.confluent.parser import AsyncConfluentParser +from faststream.message import TOMBSTONE from tests.brokers.base.parser import CustomParserTestcase from .basic import ConfluentTestcaseConfig @@ -9,3 +13,28 @@ @pytest.mark.confluent() class TestCustomParser(ConfluentTestcaseConfig, CustomParserTestcase): pass + + +def _fake_message(value: bytes | None) -> MagicMock: + message = MagicMock() + message.value.return_value = value + message.headers.return_value = None + message.offset.return_value = 0 + message.timestamp.return_value = (0, 0) + return message + + +@pytest.mark.asyncio() +@pytest.mark.confluent() +async def test_parse_message_maps_null_value_to_tombstone() -> None: + parsed = await AsyncConfluentParser().parse_message(_fake_message(None)) + + assert parsed.body is TOMBSTONE + + +@pytest.mark.asyncio() +@pytest.mark.confluent() +async def test_parse_message_keeps_genuine_empty_value_as_bytes() -> None: + parsed = await AsyncConfluentParser().parse_message(_fake_message(b"")) + + assert parsed.body == b"" diff --git a/tests/brokers/confluent/test_publish.py b/tests/brokers/confluent/test_publish.py index 5c58f8a1acb..df3a6b8c73a 100644 --- a/tests/brokers/confluent/test_publish.py +++ b/tests/brokers/confluent/test_publish.py @@ -6,6 +6,7 @@ from faststream import Context from faststream.confluent import KafkaPublishMessage, KafkaResponse +from faststream.message import TOMBSTONE from tests.brokers.base.publish import BrokerPublishTestcase from .basic import ConfluentTestcaseConfig @@ -297,7 +298,7 @@ async def handler(msg: Any, raw_msg=Context("message")) -> None: assert messages_queue.empty() @pytest.mark.asyncio() - async def test_publish_none_sends_a_real_tombstone(self, queue: str) -> None: + async def test_publish_none_encodes_normally(self, queue: str) -> None: pub_broker = self.get_broker(apply_types=True) values: asyncio.Queue[bytes | None] = asyncio.Queue() @@ -310,7 +311,63 @@ async def handler(msg: Any = Context("message")) -> None: async with self.patch_broker(pub_broker) as br: await br.start() - await br.publish(None, queue, key=b"tombstone-key") + await br.publish(None, queue) + value = await asyncio.wait_for(values.get(), timeout=self.timeout) + + assert value == b"" + + @pytest.mark.asyncio() + async def test_publish_tombstone_sends_a_real_tombstone(self, queue: str) -> None: + pub_broker = self.get_broker(apply_types=True) + + values: asyncio.Queue[bytes | None] = asyncio.Queue() + + args, kwargs = self.get_subscriber_params(queue) + + @pub_broker.subscriber(*args, **kwargs) + async def handler(msg: Any = Context("message")) -> None: + await values.put(msg.raw_message.value()) + + async with self.patch_broker(pub_broker) as br: + await br.start() + await br.publish(TOMBSTONE, queue, key=b"tombstone-key") value = await asyncio.wait_for(values.get(), timeout=self.timeout) assert value is None + + @pytest.mark.asyncio() + async def test_publish_tombstone_without_key_raises(self, queue: str) -> None: + pub_broker = self.get_broker(apply_types=True) + + async with self.patch_broker(pub_broker) as br: + await br.start() + with pytest.raises(ValueError, match="requires a key"): + await br.publish(TOMBSTONE, queue) + + @pytest.mark.asyncio() + async def test_publish_batch_with_tombstone(self, queue: str) -> None: + pub_broker = self.get_broker(apply_types=True) + + values: asyncio.Queue[bytes | None] = asyncio.Queue() + + args, kwargs = self.get_subscriber_params(queue) + + @pub_broker.subscriber(*args, **kwargs) + async def handler(msg: Any = Context("message")) -> None: + await values.put(msg.raw_message.value()) + + async with self.patch_broker(pub_broker) as br: + await br.start() + await br.publish_batch( + "hi", + KafkaResponse(TOMBSTONE, key=b"batch-tombstone-key"), + topic=queue, + ) + + received = [ + await asyncio.wait_for(values.get(), timeout=self.timeout) + for _ in range(2) + ] + + assert b"hi" in received + assert None in received diff --git a/tests/brokers/confluent/test_test_client.py b/tests/brokers/confluent/test_test_client.py index 5d2c510a201..bb0f8b40a11 100644 --- a/tests/brokers/confluent/test_test_client.py +++ b/tests/brokers/confluent/test_test_client.py @@ -1,12 +1,14 @@ import asyncio -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, patch import pytest from faststream import AckPolicy, BaseMiddleware, Context +from faststream.confluent import KafkaResponse from faststream.confluent.annotations import KafkaMessage from faststream.confluent.message import FAKE_CONSUMER from faststream.confluent.testing import FakeProducer +from faststream.message import TOMBSTONE from tests.brokers.base.testclient import BrokerTestclientTestcase from tests.tools import spy_decorator @@ -16,22 +18,6 @@ @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) @@ -302,3 +288,65 @@ async def test_publisher_without_destination(self) -> None: await another_publisher.publish(None, topic="new-key") another_publisher.mock.assert_called_once() + + async def test_publish_none_encodes_normally(self, queue: str) -> None: + broker = self.get_broker(apply_types=True) + + values: asyncio.Queue[bytes | None] = asyncio.Queue() + + args, kwargs = self.get_subscriber_params(queue) + + @broker.subscriber(*args, **kwargs) + async def handler(msg=Context("message")) -> None: + await values.put(msg.raw_message.value()) + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + value = await asyncio.wait_for(values.get(), timeout=3) + + assert value == b"" + + async def test_publish_tombstone_sends_a_real_tombstone(self, queue: str) -> None: + broker = self.get_broker(apply_types=True) + + values: asyncio.Queue[bytes | None] = asyncio.Queue() + + args, kwargs = self.get_subscriber_params(queue) + + @broker.subscriber(*args, **kwargs) + async def handler(msg=Context("message")) -> None: + await values.put(msg.raw_message.value()) + + async with self.patch_broker(broker) as br: + await br.publish(TOMBSTONE, queue, key=b"tombstone-key") + value = await asyncio.wait_for(values.get(), timeout=3) + + assert value is None + + async def test_publish_tombstone_without_key_raises(self, queue: str) -> None: + broker = self.get_broker(apply_types=True) + + async with self.patch_broker(broker) as br: + with pytest.raises(ValueError, match="requires a key"): + await br.publish(TOMBSTONE, queue) + + async def test_publish_batch_with_tombstone(self, queue: str) -> None: + broker = self.get_broker(apply_types=True) + + values: list[bytes | None] = [] + + args, kwargs = self.get_subscriber_params(queue, batch=True) + + @broker.subscriber(*args, **kwargs) + async def handler(msg: list[bytes | None]) -> None: + values.extend(msg) + + async with self.patch_broker(broker) as br: + await br.publish_batch( + b"hi", + KafkaResponse(TOMBSTONE, key=b"batch-tombstone-key"), + topic=queue, + ) + + assert b"hi" in values + assert None in values diff --git a/tests/brokers/kafka/test_consume.py b/tests/brokers/kafka/test_consume.py index 436175e883e..356cdc262be 100644 --- a/tests/brokers/kafka/test_consume.py +++ b/tests/brokers/kafka/test_consume.py @@ -323,7 +323,7 @@ async def test_consume_without_value( consume_broker = self.get_broker() @consume_broker.subscriber(queue) - async def handler(msg: bytes) -> None: + async def handler(msg: bytes | None) -> None: event.set() mock(msg) @@ -340,7 +340,7 @@ async def handler(msg: bytes) -> None: timeout=3, ) - mock.assert_called_once_with(b"") + mock.assert_called_once_with(None) @pytest.mark.asyncio() async def test_consume_batch_without_value( @@ -352,7 +352,7 @@ async def test_consume_batch_without_value( consume_broker = self.get_broker() @consume_broker.subscriber(queue, batch=True) - async def handler(msg: list[bytes]) -> None: + async def handler(msg: list[bytes | None]) -> None: event.set() mock(msg) @@ -369,7 +369,7 @@ async def handler(msg: list[bytes]) -> None: timeout=3, ) - mock.assert_called_once_with([b""]) + mock.assert_called_once_with([None]) @pytest.mark.asyncio() @pytest.mark.slow() diff --git a/tests/brokers/kafka/test_fastapi.py b/tests/brokers/kafka/test_fastapi.py index 29768d49b62..d0c2af20a87 100644 --- a/tests/brokers/kafka/test_fastapi.py +++ b/tests/brokers/kafka/test_fastapi.py @@ -5,7 +5,11 @@ from faststream.kafka import KafkaRouter from faststream.kafka.fastapi import KafkaRouter as StreamRouter -from tests.brokers.base.fastapi import FastAPILocalTestcase, FastAPITestcase +from tests.brokers.base.fastapi import ( + FastAPILocalTestcase, + FastAPITestcase, + KafkaTombstoneFastAPILocalTestcase, +) from .basic import KafkaMemoryTestcaseConfig @@ -41,7 +45,9 @@ async def hello(msg: list[str]): @pytest.mark.kafka() -class TestRouterLocal(KafkaMemoryTestcaseConfig, FastAPILocalTestcase): +class TestRouterLocal( + KafkaMemoryTestcaseConfig, FastAPILocalTestcase, KafkaTombstoneFastAPILocalTestcase +): router_class = StreamRouter broker_router_class = KafkaRouter diff --git a/tests/brokers/kafka/test_publish.py b/tests/brokers/kafka/test_publish.py index 4aca375d329..6ad5bc41d5a 100644 --- a/tests/brokers/kafka/test_publish.py +++ b/tests/brokers/kafka/test_publish.py @@ -8,6 +8,7 @@ from faststream import Context from faststream.kafka import KafkaPublishMessage, KafkaResponse from faststream.kafka.exceptions import BatchBufferOverflowException +from faststream.message import TOMBSTONE from tests.brokers.base.publish import BrokerPublishTestcase from .basic import KafkaTestcaseConfig @@ -349,7 +350,7 @@ async def handler(msg: Any, raw_msg=Context("message")) -> None: assert messages_queue.empty() @pytest.mark.asyncio() - async def test_publish_none_sends_a_real_tombstone(self, queue: str) -> None: + async def test_publish_none_encodes_normally(self, queue: str) -> None: pub_broker = self.get_broker(apply_types=True) values: asyncio.Queue[bytes | None] = asyncio.Queue() @@ -360,7 +361,59 @@ async def handler(msg: Any = Context("message")) -> None: async with self.patch_broker(pub_broker) as br: await br.start() - await br.publish(None, queue, key=b"tombstone-key") + await br.publish(None, queue) + value = await asyncio.wait_for(values.get(), timeout=self.timeout) + + assert value == b"" + + @pytest.mark.asyncio() + async def test_publish_tombstone_sends_a_real_tombstone(self, queue: str) -> None: + pub_broker = self.get_broker(apply_types=True) + + values: asyncio.Queue[bytes | None] = asyncio.Queue() + + @pub_broker.subscriber(queue) + async def handler(msg: Any = Context("message")) -> None: + await values.put(msg.raw_message.value) + + async with self.patch_broker(pub_broker) as br: + await br.start() + await br.publish(TOMBSTONE, queue, key=b"tombstone-key") value = await asyncio.wait_for(values.get(), timeout=self.timeout) assert value is None + + @pytest.mark.asyncio() + async def test_publish_tombstone_without_key_raises(self, queue: str) -> None: + pub_broker = self.get_broker(apply_types=True) + + async with self.patch_broker(pub_broker) as br: + await br.start() + with pytest.raises(ValueError, match="requires a key"): + await br.publish(TOMBSTONE, queue) + + @pytest.mark.asyncio() + async def test_publish_batch_with_tombstone(self, queue: str) -> None: + pub_broker = self.get_broker(apply_types=True) + + values: asyncio.Queue[bytes | None] = asyncio.Queue() + + @pub_broker.subscriber(queue) + async def handler(msg: Any = Context("message")) -> None: + await values.put(msg.raw_message.value) + + async with self.patch_broker(pub_broker) as br: + await br.start() + await br.publish_batch( + "hi", + KafkaResponse(TOMBSTONE, key=b"batch-tombstone-key"), + topic=queue, + ) + + received = [ + await asyncio.wait_for(values.get(), timeout=self.timeout) + for _ in range(2) + ] + + assert b"hi" in received + assert None in received diff --git a/tests/brokers/kafka/test_test_client.py b/tests/brokers/kafka/test_test_client.py index b27279648f7..1cb89400607 100644 --- a/tests/brokers/kafka/test_test_client.py +++ b/tests/brokers/kafka/test_test_client.py @@ -1,13 +1,14 @@ import asyncio -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, patch import pytest from faststream import AckPolicy, BaseMiddleware, Context -from faststream.kafka import TopicPartition +from faststream.kafka import KafkaResponse, TopicPartition from faststream.kafka.annotations import KafkaMessage from faststream.kafka.message import FAKE_CONSUMER from faststream.kafka.testing import FakeProducer +from faststream.message import TOMBSTONE from tests.brokers.base.testclient import BrokerTestclientTestcase from tests.tools import spy_decorator @@ -17,22 +18,6 @@ @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, @@ -361,3 +346,59 @@ async def test_publisher_without_destination(self) -> None: await another_publisher.publish(None, topic="new-key") another_publisher.mock.assert_called_once() + + async def test_publish_none_encodes_normally(self, queue: str) -> None: + broker = self.get_broker(apply_types=True) + + values: asyncio.Queue[bytes | None] = asyncio.Queue() + + @broker.subscriber(queue) + async def handler(msg=Context("message")) -> None: + await values.put(msg.raw_message.value) + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + value = await asyncio.wait_for(values.get(), timeout=3) + + assert value == b"" + + async def test_publish_tombstone_sends_a_real_tombstone(self, queue: str) -> None: + broker = self.get_broker(apply_types=True) + + values: asyncio.Queue[bytes | None] = asyncio.Queue() + + @broker.subscriber(queue) + async def handler(msg=Context("message")) -> None: + await values.put(msg.raw_message.value) + + async with self.patch_broker(broker) as br: + await br.publish(TOMBSTONE, queue, key=b"tombstone-key") + value = await asyncio.wait_for(values.get(), timeout=3) + + assert value is None + + async def test_publish_tombstone_without_key_raises(self, queue: str) -> None: + broker = self.get_broker(apply_types=True) + + async with self.patch_broker(broker) as br: + with pytest.raises(ValueError, match="requires a key"): + await br.publish(TOMBSTONE, queue) + + async def test_publish_batch_with_tombstone(self, queue: str) -> None: + broker = self.get_broker(apply_types=True) + + values: list[bytes | None] = [] + + @broker.subscriber(queue, batch=True) + async def handler(msg: list[bytes | None]) -> None: + values.extend(msg) + + async with self.patch_broker(broker) as br: + await br.publish_batch( + b"hi", + KafkaResponse(TOMBSTONE, key=b"batch-tombstone-key"), + topic=queue, + ) + + assert b"hi" in values + assert None in values diff --git a/tests/message/test_utils.py b/tests/message/test_utils.py index 6dd7ac5d662..2cb5357083e 100644 --- a/tests/message/test_utils.py +++ b/tests/message/test_utils.py @@ -3,7 +3,7 @@ import pytest -from faststream.message.utils import decode_message +from faststream.message.utils import TOMBSTONE, decode_message @dataclass @@ -60,3 +60,15 @@ def test_unknown_content_type_ok(content_type: str) -> None: ) def test_no_content_type_ok(body: Any, expected: bytes | dict[str, str]) -> None: assert decode_message(body) == expected + + +def test_tombstone_decodes_to_none() -> None: + msg: Any = _MessageStub(TOMBSTONE) # type: ignore[arg-type] + + assert decode_message(msg) is None + + +def test_genuine_empty_body_is_not_a_tombstone() -> None: + msg: Any = _MessageStub(b"") + + assert decode_message(msg) == b""