Help with pytest

[ChadDa3mon]

Hey guys,

I'm struggling to understand your testing/mocking examples with pytest. Everything you've shown still assumes I create a broker on my actual kafka broker, but I thought this testing was supposed to be "in memory"? I'm trying to follow your example here but I just can't understand it.

So here's my problem, I'm trying to create a new consumer to consume someone elses crazy json payload. In our dev environment, something is injecting fake data every 15 minutes or so, which makes testing my ability to develop/test things slow and painful.

A coworker reminded me that you guys have some in memory testing functions so I was hoping to simply do three things:

1. Define a test payload based on what I get from this topic
2. Publish this to a fake 'in memory' topic
3. Have my consumer function pull the payload and try to properly parse it all

I just can't figure out how to (or even if I can) do this based on your examples. Here's what I've got so far in a test.py file:

import pytest
from faststream.kafka import TestKafkaBroker, KafkaBroker
from faststream import FastStream
from models.kafka import KafkaMessage

crazy_dict = {some big crazy nested dictionary from hell}

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)
publisher1 = broker.publisher("test-topic")

@publisher1
@broker.subscriber("test-topic")
async def handle():
    return "response"

@pytest.mark.asyncio
async def test_handle():
    # How do I handle the message I get from kafka and load it into a dict like I'm doing in my normal code? 

So, assuming I've got a topic called test-topic I need a way to publish crazy_dict, then consume it, then be able to put it through my code to parse it. And I'd like this to only happen in memory.

If that's possible, then I guess my only other question is how would I run this, do I have to use pytest to execute this? It's not clear to me if you guys expect people to have a seperate "test" file for pytest (like the standard uses) or if you're somehow saying we can have the pytests/mocks in the same code file as the live code.

[davorrunje]

Please try this:

from typing import Any, Dict

import pytest
from faststream import FastStream, Logger
from faststream.kafka import KafkaBroker, TestKafkaBroker

crazy_dict = {"msg": "test", "n": 2, "d": {"a": 1, "b": 2}}

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

@broker.subscriber("test")
async def handle(msg: Dict[str, Any], logger: Logger):
    logger.info(f"handle(msg={msg})")
    return "response"


@pytest.mark.asyncio
async def test_handle():
    async with TestKafkaBroker(broker) as br:

        await br.publish(crazy_dict, "test")

        # check an incoming message body
        handle.mock.assert_called_with(crazy_dict)

If you run it with pytest -s, you get the following output:

================================================= test session starts =================================================
platform darwin -- Python 3.11.5, pytest-7.4.2, pluggy-1.3.0
rootdir: /Users/davor/Projects/tmp/fs-issue/my_faststream_app
configfile: pyproject.toml
plugins: asyncio-0.21.1, anyio-3.7.1
asyncio: mode=Mode.STRICT
collected 1 item                                                                                                      

tests/test_application.py 2023-10-25 11:50:32,285 INFO     - test | 0-16982274 - handle(msg={'msg': 'test', 'n': 2, 'd': {'a': 1, 'b': 2}})
.

================================================== 1 passed in 0.34s ==================================================