Bug: Expired is missing from rabbit testing header properties

[marcm-ml]

Hello,

the expiration property is not respected when using the rabbit test broker and setting the "expiration" flag in the testbroker.publish method.

I think its simply missing here.

faststream/faststream/rabbit/testing.py

Lines 173 to 185 in 8148f2d

	header=ContentHeader(
	properties=spec.Basic.Properties(
	content_type=msg.content_type,
	headers=msg.headers,
	reply_to=msg.reply_to,
	content_encoding=msg.content_encoding,
	priority=msg.priority,
	correlation_id=msg.correlation_id,
	message_id=msg.message_id,
	timestamp=msg.timestamp,
	message_type=message_type,
	user_id=msg.user_id,
	app_id=msg.app_id,

Up until this point in the code, the expiration property is propagated but not passed to the aiormq.abc.DeliveredMessage header property

[Lancetnik]

@marcm-ml hi! Thank you for the report!
I think, the problem is a little bit dipper than just flag missing. We can set the headed, but it has no sense for testbroker. Test broker emulates message delivery by direct subscriber function execution. So, is is a synchronous call, not a real publishing.

What behavior you expect in this case?

[marcm-ml]

Hey, thanks for the reply.

I have a test suite and just encountered this when writing the following test. I have already implemented the expected behavior by mocking the faststream build_message method.

import asyncio
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Optional, Union
from unittest.mock import AsyncMock

import pytest
from aio_pika.message import encode_expiration
from faststream.rabbit import RabbitBroker
from faststream.rabbit.subscriber.usecase import RabbitSubscriber
from faststream.rabbit.testing import TestRabbitBroker
from pytest_mock import MockerFixture

if TYPE_CHECKING:
    from aio_pika.abc import DateType, HeadersType
    from fast_depends.library.serializer import SerializerProto
    from faststream.rabbit.schemas import RabbitExchange, RabbitQueue
    from faststream.rabbit.subscriber import RabbitSubscriber
    from faststream.rabbit.types import AioPikaSendableMessage


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("expiration", "expected"),
    [
        (None, None),
        (1, 1),
        (1.0, 1.0),
        (1e1, 1e1),
        (datetime(2025, 10, 1), encode_expiration(datetime(2025, 10, 1))),
        (timedelta(2), encode_expiration(timedelta(2))),
    ],
)
async def test_base_handler_expiration_value(
    mocker: MockerFixture,
    broker: RabbitBroker,
    secret: Any,
    consumer: RabbitSubscriber,
    sink_mock: AsyncMock,
    expiration,
    expected,
):
    # faststream has bug that does not propagate the expiration value correctly
    # see https://github.com/ag2ai/faststream/issues/2785
    def build_message(
        message: "AioPikaSendableMessage" = "",
        queue: Union["RabbitQueue", str] = "",
        exchange: Union["RabbitExchange", str, None] = None,
        *,
        routing_key: str = "",
        persist: bool = False,
        reply_to: str | None = None,
        headers: Optional["HeadersType"] = None,
        content_type: str | None = None,
        content_encoding: str | None = None,
        priority: int | None = None,
        correlation_id: str | None = None,
        expiration: Optional["DateType"] = None,
        message_id: str | None = None,
        timestamp: Optional["DateType"] = None,
        message_type: str | None = None,
        user_id: str | None = None,
        app_id: str | None = None,
        serializer: Optional["SerializerProto"] = None,
    ):
        import aiormq
        from faststream.message import gen_cor_id
        from faststream.rabbit.parser import AioPikaParser
        from faststream.rabbit.schemas import RabbitExchange, RabbitQueue
        from faststream.rabbit.testing import PatchedMessage
        from pamqp.header import ContentHeader

        que = RabbitQueue.validate(queue)
        exch = RabbitExchange.validate(exchange)

        routing = routing_key or que.routing()

        correlation_id = correlation_id or gen_cor_id()
        msg = AioPikaParser.encode_message(
            message=message,
            persist=persist,
            reply_to=reply_to,
            headers=headers,
            content_type=content_type,
            content_encoding=content_encoding,
            priority=priority,
            correlation_id=correlation_id,
            expiration=expiration,
            message_id=message_id or correlation_id,
            timestamp=timestamp,
            message_type=message_type,
            user_id=user_id,
            app_id=app_id,
            serializer=serializer,
        )

        return PatchedMessage(
            aiormq.abc.DeliveredMessage(
                delivery=aiormq.spec.Basic.Deliver(exchange=getattr(exch, "name", ""), routing_key=routing),
                header=ContentHeader(
                    properties=aiormq.spec.Basic.Properties(
                        content_type=msg.content_type,
                        headers=msg.headers,
                        reply_to=msg.reply_to,
                        content_encoding=msg.content_encoding,
                        priority=msg.priority,
                        correlation_id=msg.correlation_id,
                        message_id=msg.message_id,
                        timestamp=msg.timestamp,
                        expiration=encode_expiration(msg.expiration),
                        message_type=message_type,
                        user_id=msg.user_id,
                        app_id=msg.app_id,
                    )
                ),
                body=msg.body,
                channel=AsyncMock(),
            )
        )

    mocker.patch("faststream.rabbit.testing.build_message", new=build_message)

    random_data = b"random_data"
    async with TestRabbitBroker(broker) as b:
        await b.publish(random_data, secret.QUEUE, expiration=expiration)
        await asyncio.sleep(0.05)

        # has only one handler, namely base_handler
        assert len(consumer.calls) == 1
        assert consumer.calls[0].handler.mock
        assert consumer.calls[0].handler.mock.call_args_list[0][0][0].body == random_data

        assert sink_mock.put.called
        record: dict = sink_mock.put.call_args_list[0][1]
        data = my_message_model.loads(record["Data"])
        assert data.message_info.expiration == expected

[marcm-ml]

Let me elaborate further: I have a message model where i want to capture the producer info dict for debugging purposes. This includes the expiration property. Now I want to write tests to check what kind of type the Model generates when validating the expiration property for different producer values (like int, float, datetime, timedelta). Since I typed this field as "Any", my validating library (msgspec) has different behavior for different input types

[Lancetnik]

@marcm-ml so, all you need is just an option without behavior changes? It seems easy to add, I'll be happy to merge your PR if you want open it