Coverage for tests / docs / confluent / publish_with_partition_key / test_app.py: 73%

1import pytest (empty)

2 

3from docs.docs_src.confluent.publish_with_partition_key.app import ( (empty)

4 Data, 

5 broker, 

6 on_input_data, 

7 to_output_data, 

8) 

9from faststream.confluent import TestKafkaBroker (empty)

10 

11 

12@pytest.mark.confluent() (empty)

13@pytest.mark.asyncio() (empty)

14async def test_app() -> None: (empty)

15 async with TestKafkaBroker(broker): (empty)

16 await broker.publish(Data(data=0.2), "input_data", key=b"my_key") (empty)

17 

18 on_input_data.mock.assert_called_once_with(dict(Data(data=0.2))) (empty)

19 to_output_data.mock.assert_called_once_with(dict(Data(data=1.2))) (empty)

20 

21 

22@pytest.mark.confluent() (empty)

23@pytest.mark.skip("we are not checking the key") (empty)

24@pytest.mark.asyncio() (empty)

25async def test_keys() -> None: (empty)

26 async with TestKafkaBroker(broker): 

27 # we should be able to publish a message with the key 

28 await broker.publish(Data(data=0.2), "input_data", key=b"my_key") 

29 

30 # we need to check the key as well 

31 on_input_data.mock.assert_called_once_with(dict(Data(data=0.2)), key=b"my_key") 

32 to_output_data.mock.assert_called_once_with(dict(Data(data=1.2)), key=b"key")