Coverage for faststream / redis / security.py: 93%

1from typing import Any (empty)

2 

3from redis.asyncio.connection import Connection (empty)

4 

5from faststream.security import BaseSecurity, SASLPlaintext (empty)

6 

7 

8def parse_security(security: BaseSecurity | None) -> dict[str, Any]: (empty)

9 if security is None: (empty)

10 return {} (empty)

11 

12 if isinstance(security, SASLPlaintext): (empty)

13 return _parse_sasl_plaintext(security) (empty)

14 

15 if isinstance(security, BaseSecurity): 15 ↛ 18line 15 didn't jump to line 18 because the condition on line 15 was always true(empty)

16 return _parse_base_security(security) (empty)

17 

18 msg = f"RedisBroker does not support {type(security)}" 

19 raise NotImplementedError(msg) 

20 

21 

22def _parse_base_security(security: BaseSecurity) -> dict[str, Any]: (empty)

23 if security.use_ssl: (empty)

24 

25 class SSLConnection(Connection): (empty)

26 def __init__( (empty)

27 self, 

28 _security: BaseSecurity = security, 

29 **kwargs: Any, 

30 ) -> None: 

31 self._security = _security (empty)

32 super().__init__(**kwargs) (empty)

33 

34 def _connection_arguments(self) -> Any: (empty)

35 return { (empty)

36 **super()._connection_arguments(), # type: ignore[misc] 

37 "ssl": self._security.ssl_context, 

38 } 

39 

40 return {"connection_class": SSLConnection} (empty)

41 return {} (empty)

42 

43 

44def _parse_sasl_plaintext(security: SASLPlaintext) -> dict[str, Any]: (empty)

45 return { (empty)

46 **_parse_base_security(security), 

47 "username": security.username, 

48 "password": security.password, 

49 }