Coverage for faststream / rabbit / security.py: 88%

1from typing import Any (empty)

2 

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

4 

5 

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

7 """Convert security object to connection arguments.""" 

8 if security is None: (empty)

9 return {} (empty)

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

11 return _parse_sasl_plaintext(security) (empty)

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

13 return _parse_base_security(security) (empty)

14 msg = f"RabbitBroker does not support {type(security)}" 

15 raise NotImplementedError(msg) 

16 

17 

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

19 return { (empty)

20 "ssl": security.use_ssl, 

21 "ssl_context": security.ssl_context, 

22 } 

23 

24 

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

26 return { (empty)

27 "ssl": security.use_ssl, 

28 "ssl_context": security.ssl_context, 

29 "login": security.username, 

30 "password": security.password, 

31 }