Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/tests/charts/test_kerberos.py: 0%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, 

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17from __future__ import annotations 

18 

19import json 

20 

21import jmespath 

22 

23from tests.charts.helm_template_generator import render_chart 

24 

25 

26class TestKerberos: 

27 def test_kerberos_not_mentioned_in_render_if_disabled(self): 

28 # the name is deliberately shorter as we look for "kerberos" in the rendered chart 

29 k8s_objects = render_chart(name="no-krbros", values={"kerberos": {"enabled": False}}) 

30 # ignore airflow config map 

31 k8s_objects_to_consider = [ 

32 obj for obj in k8s_objects if obj["metadata"]["name"] != "no-krbros-airflow-config" 

33 ] 

34 k8s_objects_to_consider_str = json.dumps(k8s_objects_to_consider) 

35 assert k8s_objects_to_consider_str.count("kerberos") == 1 

36 

37 def test_kerberos_envs_available_in_worker_with_persistence(self): 

38 docs = render_chart( 

39 values={ 

40 "executor": "CeleryExecutor", 

41 "workers": { 

42 "kerberosSidecar": {"enabled": True}, 

43 "persistence": { 

44 "enabled": True, 

45 }, 

46 }, 

47 "kerberos": { 

48 "enabled": True, 

49 "configPath": "/etc/krb5.conf", 

50 "ccacheMountPath": "/var/kerberos-ccache", 

51 "ccacheFileName": "ccache", 

52 }, 

53 }, 

54 show_only=["templates/workers/worker-deployment.yaml"], 

55 ) 

56 

57 assert {"name": "KRB5_CONFIG", "value": "/etc/krb5.conf"} in jmespath.search( 

58 "spec.template.spec.containers[0].env", docs[0] 

59 ) 

60 assert {"name": "KRB5CCNAME", "value": "/var/kerberos-ccache/ccache"} in jmespath.search( 

61 "spec.template.spec.containers[0].env", docs[0] 

62 ) 

63 

64 def test_kerberos_sidecar_resources(self): 

65 docs = render_chart( 

66 values={ 

67 "executor": "CeleryExecutor", 

68 "workers": { 

69 "kerberosSidecar": { 

70 "enabled": True, 

71 "resources": { 

72 "requests": { 

73 "cpu": "200m", 

74 "memory": "200Mi", 

75 }, 

76 "limits": { 

77 "cpu": "201m", 

78 "memory": "201Mi", 

79 }, 

80 }, 

81 }, 

82 }, 

83 }, 

84 show_only=["templates/workers/worker-deployment.yaml"], 

85 ) 

86 

87 assert jmespath.search("spec.template.spec.containers[2].resources.requests.cpu", docs[0]) == "200m" 

88 assert ( 

89 jmespath.search("spec.template.spec.containers[2].resources.requests.memory", docs[0]) == "200Mi" 

90 ) 

91 assert jmespath.search("spec.template.spec.containers[2].resources.limits.cpu", docs[0]) == "201m" 

92 assert jmespath.search("spec.template.spec.containers[2].resources.limits.memory", docs[0]) == "201Mi" 

93 

94 def test_keberos_sidecar_resources_are_not_added_by_default(self): 

95 docs = render_chart( 

96 show_only=["templates/workers/worker-deployment.yaml"], 

97 ) 

98 assert jmespath.search("spec.template.spec.containers[0].resources", docs[0]) == {} 

99 

100 def test_kerberos_keytab_secret_available(self): 

101 docs = render_chart( 

102 values={ 

103 "executor": "CeleryExecutor", 

104 "kerberos": { 

105 "enabled": True, 

106 "keytabBase64Content": "dGVzdGtleXRhYg==", 

107 "configPath": "/etc/krb5.conf", 

108 "ccacheMountPath": "/var/kerberos-ccache", 

109 "ccacheFileName": "ccache", 

110 }, 

111 }, 

112 show_only=["templates/secrets/kerberos-keytab-secret.yaml"], 

113 ) 

114 

115 assert jmespath.search('data."kerberos.keytab"', docs[0]) == "dGVzdGtleXRhYg==" 

116 

117 def test_kerberos_keytab_secret_unavailable_when_not_specified(self): 

118 docs = render_chart( 

119 values={ 

120 "executor": "CeleryExecutor", 

121 "kerberos": { 

122 "enabled": True, 

123 "configPath": "/etc/krb5.conf", 

124 "ccacheMountPath": "/var/kerberos-ccache", 

125 "ccacheFileName": "ccache", 

126 }, 

127 }, 

128 show_only=["templates/secrets/kerberos-keytab-secret.yaml"], 

129 ) 

130 

131 assert 0 == len(docs)