Coverage for tests / brokers / base / basic.py: 75%

1from abc import abstractmethod (empty)

2from typing import Any (empty)

3 

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

5from faststream._internal.broker.router import BrokerRouter (empty)

6 

7 

8class BaseTestcaseConfig: (empty)

9 timeout: float = 3.0 (empty)

10 

11 @abstractmethod (empty)

12 def get_broker( (empty)

13 self, 

14 apply_types: bool = False, (empty)

15 **kwargs: Any, (empty)

16 ) -> BrokerUsecase[Any, Any]: (empty)

17 raise NotImplementedError 

18 

19 def patch_broker( (empty)

20 self, 

21 broker: BrokerUsecase, 

22 **kwargs: Any, 

23 ) -> BrokerUsecase: 

24 return broker 

25 

26 def get_subscriber_params( (empty)

27 self, 

28 *args: Any, (empty)

29 **kwargs: Any, (empty)

30 ) -> tuple[ (empty)

31 tuple[Any, ...], (empty)

32 dict[str, Any], (empty)

33 ]: 

34 return args, kwargs (empty)

35 

36 @abstractmethod (empty)

37 def get_router(self, **kwargs: Any) -> BrokerRouter: (empty)

38 raise NotImplementedError