Offline testing

[yann-combarnous]

Hi,

I went through the testing examples and would like an advice on how to run the tests offline.

In the below code example, there are 2 parts that need connectivity atm:

• KafkaBroker, which I would like to mock, because even though my test is using TestKafkaBroker, broker is still instantiated and requires connectivity
• save function which would be a database I save my message to, here I would like to mock the save function, with the ability to return a value, and check what it was called with

Any advice?

from pydantic import BaseModel
import pytest

from faststream import FastStream
from faststream.kafka import KafkaBroker, TestKafkaBroker
from database import save


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


class Message(BaseModel):
    id: str

@broker.subscriber("topic")
async def on_message(msg: Message) -> None:
   save(msg)
   pass



@pytest.mark.asyncio
async def test_message_consuming():
    async with TestKafkaBroker(broker):
        await broker.publish(
            Message(id="1", "topic",
        )
       # Here, I would like to test that `save` was called with a certain value, how to mock `save`

[Lancetnik]

Hi, @yann-combarnous ! Your problem is not related to FastStream itself and we have no tools to help you, but you can use any type of monkeypatching: pytest or unittest one

Take a look at the pytest doc about monkeypatching

[yann-combarnous]

Hi, @yann-combarnous ! Your problem is not related to FastStream itself and we have no tools to help you, but you can use any type of monkeypatching: pytest or unittest one

Take a look at the pytest doc about monkeypatching

@Lancetnik , thanks for the feedback. I had tried monkeypatching before without success so far (I must confess my poor Python testing skills).

I think for documentation completeness - as there are examples of testing - adding the monkeypatching bit for the broker would go a long way helping users. The TestKafkaBroker is great, and existing documentation works, but it only really makes sense in offline use cases for CI/CD.

[Lancetnik]

Why are you thinking, that TestBroker can be used only in CI? I am using it locally almost all time. Also, we have a special with_real=True flag if you want to run your tests with real broker

[yann-combarnous]

Indeed, there are more use cases with TestBroker, looks like nice coverage of common scenarios.

Answer to self for above: if you have broker.py and test.py, in test.py patch broker first and then import broker.py as broker is not created in a function.