Coverage for tests / brokers / base / connection.py: 95%

1from typing import Any (empty)

2 

3import pytest (empty)

4 

5from faststream._internal.broker import BrokerUsecase (empty)

6 

7 

8class BrokerConnectionTestcase: (empty)

9 broker: type[BrokerUsecase] (empty)

10 

11 def get_broker_args(self, settings: Any) -> dict[str, Any]: (empty)

12 return {} 

13 

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

15 async def ping(self, broker) -> bool: (empty)

16 return await broker.ping(timeout=5.0) (empty)

17 

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

19 async def test_stop_before_start(self) -> None: (empty)

20 br = self.broker() (empty)

21 assert br._connection is None (empty)

22 await br.stop() (empty)

23 assert not br.running (empty)

24 

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

26 async def test_connect(self, settings: Any) -> None: (empty)

27 kwargs = self.get_broker_args(settings) (empty)

28 broker = self.broker(**kwargs) (empty)

29 await broker.connect() (empty)

30 assert await self.ping(broker) (empty)

31 await broker.stop() (empty)