Coverage for faststream / mqtt / security.py: 52%

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 if security is None: 7 ↛ 9line 7 didn't jump to line 9 because the condition on line 7 was always true(empty)

8 return {} (empty)

9 if isinstance(security, SASLPlaintext): 

10 return _parse_sasl_plaintext(security) 

11 if isinstance(security, BaseSecurity): 

12 return _parse_base_security(security) 

13 msg = f"MQTTBroker does not support {type(security)}" 

14 raise NotImplementedError(msg) 

15 

16 

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

18 if security.use_ssl: (empty)

19 return {"tls": security.ssl_context or True} (empty)

20 return {"tls": False} (empty)

21 

22 

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

24 return { 

25 **_parse_base_security(security), 

26 "username": security.username, 

27 "password": security.password, 

28 }