Coverage for tests / brokers / confluent / test_security.py: 33%

1from contextlib import contextmanager (empty)

2from unittest.mock import AsyncMock, MagicMock, patch (empty)

3 

4import pytest (empty)

5 

6from faststream.exceptions import SetupError (empty)

7 

8 

9@contextmanager (empty)

10def patch_aio_consumer_and_producer() -> tuple[MagicMock, MagicMock]: (empty)

11 try: 

12 producer = MagicMock(return_value=AsyncMock()) 

13 

14 with patch( 

15 "faststream.confluent.helpers.client.AsyncConfluentProducer", 

16 new=producer, 

17 ): 

18 yield producer 

19 finally: 

20 pass 

21 

22 

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

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

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

26 import ssl (empty)

27 

28 from faststream.confluent import KafkaBroker (empty)

29 from faststream.security import BaseSecurity (empty)

30 

31 ssl_context = ssl.create_default_context() (empty)

32 security = BaseSecurity(ssl_context=ssl_context) (empty)

33 

34 with pytest.raises( (empty)

35 SetupError, (empty)

36 match=r"ssl_context is not supported by confluent-kafka-python, please use config instead.", (empty)

37 ): 

38 KafkaBroker("localhost:9092", security=security) (empty)