Coverage for faststream / confluent / security.py: 95%

1from typing import Any (empty)

2 

3from faststream.exceptions import SetupError (empty)

4from faststream.security import ( (empty)

5 SASLGSSAPI, 

6 BaseSecurity, 

7 SASLOAuthBearer, 

8 SASLPlaintext, 

9 SASLScram256, 

10 SASLScram512, 

11) 

12 

13 

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

15 if security is None: (empty)

16 return {} (empty)

17 

18 if security.ssl_context: (empty)

19 msg = "ssl_context is not supported by confluent-kafka-python, please use config instead." (empty)

20 raise SetupError(msg) (empty)

21 

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

23 return _parse_sasl_plaintext(security) (empty)

24 if isinstance(security, SASLScram256): (empty)

25 return _parse_sasl_scram256(security) (empty)

26 if isinstance(security, SASLScram512): (empty)

27 return _parse_sasl_scram512(security) (empty)

28 if isinstance(security, SASLOAuthBearer): (empty)

29 return _parse_sasl_oauthbearer(security) (empty)

30 if isinstance(security, SASLGSSAPI): (empty)

31 return _parse_sasl_gssapi(security) (empty)

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

33 return _parse_base_security(security) (empty)

34 

35 msg = f"KafkaBroker does not support `{type(security)}`." 

36 raise NotImplementedError(msg) 

37 

38 

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

40 return { (empty)

41 "security.protocol": "ssl" if security.use_ssl else "plaintext", 

42 } 

43 

44 

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

46 return { (empty)

47 "security.protocol": "sasl_ssl" if security.use_ssl else "sasl_plaintext", 

48 "sasl.mechanism": "PLAIN", 

49 "sasl.username": security.username, 

50 "sasl.password": security.password, 

51 } 

52 

53 

54def _parse_sasl_scram256(security: SASLScram256) -> dict[str, Any]: (empty)

55 return { (empty)

56 "security.protocol": "sasl_ssl" if security.use_ssl else "sasl_plaintext", 

57 "sasl.mechanism": "SCRAM-SHA-256", 

58 "sasl.username": security.username, 

59 "sasl.password": security.password, 

60 } 

61 

62 

63def _parse_sasl_scram512(security: SASLScram512) -> dict[str, Any]: (empty)

64 return { (empty)

65 "security.protocol": "sasl_ssl" if security.use_ssl else "sasl_plaintext", 

66 "sasl.mechanism": "SCRAM-SHA-512", 

67 "sasl.username": security.username, 

68 "sasl.password": security.password, 

69 } 

70 

71 

72def _parse_sasl_oauthbearer(security: SASLOAuthBearer) -> dict[str, Any]: (empty)

73 return { (empty)

74 "security.protocol": "sasl_ssl" if security.use_ssl else "sasl_plaintext", 

75 "sasl.mechanism": "OAUTHBEARER", 

76 } 

77 

78 

79def _parse_sasl_gssapi(security: SASLGSSAPI) -> dict[str, Any]: (empty)

80 return { (empty)

81 "security.protocol": "sasl_ssl" if security.use_ssl else "sasl_plaintext", 

82 "sasl.mechanism": "GSSAPI", 

83 }