How to use mock.assert_called_with inside tests properly, when we have datetime attribute

[rjambrecic]

Example:

from pydantic import BaseModel
from datetime import datetime
import pytest
import json

from faststream._compat import dump_json
from faststream import FastStream
from faststream.kafka import KafkaBroker, TestKafkaBroker


broker = KafkaBroker("localhost:9092")
app = FastStream(broker)


class Problem(BaseModel):
    time: datetime

@broker.subscriber("problem")
async def on_problem(msg: Problem) -> None:
    pass



@pytest.mark.asyncio
async def test_problem_consuming():
    async with TestKafkaBroker(broker):
        await broker.publish(
            Problem(time=datetime(2023,2,2)), "problem",
        )
        # This works but it's not pretty :)
        on_problem.mock.assert_called_with(
            json.loads(
                dump_json(
                    Problem(time=datetime(2023,2,2))
                )
            )
        )

        # This will fail
        on_problem.mock.assert_called_with(dict(Problem(time=datetime(2023,2,2))))
        """
        Error msg:
        Expected: mock({'time': datetime.datetime(2023, 2, 2, 0, 0)})
        Actual: mock({'time': '2023-02-02T00:00:00'})
        """

[Lancetnik]

@rjambrecic mock body designed to represent inner broker json message, not python serialized one
So, to test it you need to make your pydantic object JSONable

As an example - here is an app

from datetime import datetime
from pydantic import BaseModel
from faststream.rabbit import RabbitBroker, TestRabbitBroker

broker = RabbitBroker()

class Date(BaseModel):
    time: datetime

@broker.subscriber("in")
async def rabbit_handler(msg: Date):
    pass

To test it we can make our object JSONable

import pytest
from pydantic_core import to_jsonable_python
# or (for PydanticV1 and PydanticV2)
# from faststream._compat import model_to_jsonable

@pytest.mark.asyncio
async def test_pub():
    async with TestRabbitBroker(broker) as br:
        now = datetime.now()
        await br.publish(Date(time=now), "in")

        rabbit_handler.mock.assert_called_once_with(
            to_jsonable_python(Date(time=now))
        )

Or you can use perfect testing library dirty_equals (preffered way for me)

import pytest
from dirty_equals import IsDatetime
from faststream.rabbit import RabbitBroker, TestRabbitBroker

@pytest.mark.asyncio
async def test_pub():
    async with TestRabbitBroker(broker) as br:
        now = datetime.now()
        await br.publish(Date(time=now), "in")

        rabbit_handler.mock.assert_called_once_with({
            "time": IsDatetime(iso_string=True)
        })