Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/helm_tests/security/test_extra_configmaps_secrets.py: 13%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

53 statements  

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 textwrap 

20from base64 import b64encode 

21from unittest import mock 

22 

23import pytest 

24import yaml 

25 

26from tests.charts.helm_template_generator import prepare_k8s_lookup_dict, render_chart 

27 

28RELEASE_NAME = "test-extra-configmaps-secrets" 

29 

30 

31class TestExtraConfigMapsSecrets: 

32 """Tests extra configmaps and secrets.""" 

33 

34 def test_extra_configmaps(self): 

35 values_str = textwrap.dedent( 

36 """ 

37 extraConfigMaps: 

38 "{{ .Release.Name }}-airflow-variables": 

39 data: | 

40 AIRFLOW_VAR_HELLO_MESSAGE: "Hi!" 

41 AIRFLOW_VAR_KUBERNETES_NAMESPACE: "{{ .Release.Namespace }}" 

42 "{{ .Release.Name }}-other-variables": 

43 data: | 

44 HELLO_WORLD: "Hi again!" 

45 """ 

46 ) 

47 values = yaml.safe_load(values_str) 

48 k8s_objects = render_chart( 

49 RELEASE_NAME, values=values, show_only=["templates/configmaps/extra-configmaps.yaml"] 

50 ) 

51 k8s_objects_by_key = prepare_k8s_lookup_dict(k8s_objects) 

52 

53 all_expected_keys = [ 

54 ("ConfigMap", f"{RELEASE_NAME}-airflow-variables"), 

55 ("ConfigMap", f"{RELEASE_NAME}-other-variables"), 

56 ] 

57 assert set(k8s_objects_by_key.keys()) == set(all_expected_keys) 

58 

59 all_expected_data = [ 

60 {"AIRFLOW_VAR_HELLO_MESSAGE": "Hi!", "AIRFLOW_VAR_KUBERNETES_NAMESPACE": "default"}, 

61 {"HELLO_WORLD": "Hi again!"}, 

62 ] 

63 for expected_key, expected_data in zip(all_expected_keys, all_expected_data): 

64 configmap_obj = k8s_objects_by_key[expected_key] 

65 assert configmap_obj["data"] == expected_data 

66 

67 def test_extra_secrets(self): 

68 values_str = textwrap.dedent( 

69 """ 

70 extraSecrets: 

71 "{{ .Release.Name }}-airflow-connections": 

72 data: | 

73 AIRFLOW_CON_AWS: {{ printf "aws_connection_string" | b64enc }} 

74 stringData: | 

75 AIRFLOW_CON_GCP: "gcp_connection_string" 

76 "{{ .Release.Name }}-other-secrets": 

77 data: | 

78 MY_SECRET_1: {{ printf "MY_SECRET_1" | b64enc }} 

79 MY_SECRET_2: {{ printf "MY_SECRET_2" | b64enc }} 

80 stringData: | 

81 MY_SECRET_3: "MY_SECRET_3" 

82 MY_SECRET_4: "MY_SECRET_4" 

83 "{{ .Release.Name }}-other-secrets-with-type": 

84 type: kubernetes.io/dockerconfigjson 

85 data: | 

86 MY_SECRET_5: {{ printf "MY_SECRET_5" | b64enc }} 

87 MY_SECRET_6: {{ printf "MY_SECRET_6" | b64enc }} 

88 stringData: | 

89 MY_SECRET_7: "MY_SECRET_7" 

90 MY_SECRET_8: "MY_SECRET_8" 

91 """ 

92 ) 

93 values = yaml.safe_load(values_str) 

94 k8s_objects = render_chart( 

95 RELEASE_NAME, values=values, show_only=["templates/secrets/extra-secrets.yaml"] 

96 ) 

97 k8s_objects_by_key = prepare_k8s_lookup_dict(k8s_objects) 

98 

99 all_expected_keys = [ 

100 ("Secret", f"{RELEASE_NAME}-airflow-connections"), 

101 ("Secret", f"{RELEASE_NAME}-other-secrets"), 

102 ("Secret", f"{RELEASE_NAME}-other-secrets-with-type"), 

103 ] 

104 assert set(k8s_objects_by_key.keys()) == set(all_expected_keys) 

105 

106 all_expected_data = [ 

107 {"AIRFLOW_CON_AWS": b64encode(b"aws_connection_string").decode("utf-8")}, 

108 { 

109 "MY_SECRET_1": b64encode(b"MY_SECRET_1").decode("utf-8"), 

110 "MY_SECRET_2": b64encode(b"MY_SECRET_2").decode("utf-8"), 

111 }, 

112 { 

113 "MY_SECRET_5": b64encode(b"MY_SECRET_5").decode("utf-8"), 

114 "MY_SECRET_6": b64encode(b"MY_SECRET_6").decode("utf-8"), 

115 }, 

116 ] 

117 

118 all_expected_string_data = [ 

119 {"AIRFLOW_CON_GCP": "gcp_connection_string"}, 

120 {"MY_SECRET_3": "MY_SECRET_3", "MY_SECRET_4": "MY_SECRET_4"}, 

121 {"MY_SECRET_7": "MY_SECRET_7", "MY_SECRET_8": "MY_SECRET_8"}, 

122 ] 

123 all_expected_types = [None, None, "kubernetes.io/dockerconfigjson"] 

124 for expected_key, expected_data, expected_string_data, expected_type in zip( 

125 all_expected_keys, all_expected_data, all_expected_string_data, all_expected_types 

126 ): 

127 configmap_obj = k8s_objects_by_key[expected_key] 

128 if expected_type: 

129 assert configmap_obj["type"] == expected_type 

130 else: 

131 assert "type" not in configmap_obj 

132 assert configmap_obj["data"] == expected_data 

133 assert configmap_obj["stringData"] == expected_string_data 

134 

135 def test_extra_configmaps_secrets_labels(self): 

136 k8s_objects = render_chart( 

137 name=RELEASE_NAME, 

138 values={ 

139 "labels": {"label1": "value1", "label2": "value2"}, 

140 "extraSecrets": {"{{ .Release.Name }}-extra-secret-1": {"stringData": "data: secretData"}}, 

141 "extraConfigMaps": {"{{ .Release.Name }}-extra-configmap-1": {"data": "data: configData"}}, 

142 }, 

143 show_only=["templates/configmaps/extra-configmaps.yaml", "templates/secrets/extra-secrets.yaml"], 

144 ) 

145 expected_labels = { 

146 "label1": "value1", 

147 "label2": "value2", 

148 "release": RELEASE_NAME, 

149 "heritage": "Helm", 

150 "chart": mock.ANY, 

151 } 

152 for k8s_object in k8s_objects: 

153 assert k8s_object["metadata"]["labels"] == expected_labels 

154 

155 @pytest.mark.parametrize( 

156 "chart_labels, local_labels", 

157 [ 

158 ({}, {"label3": "value3", "label4": "value4"}), 

159 ({"label1": "value1", "label2": "value2"}, {}), 

160 ({"label1": "value1", "label2": "value2"}, {"label3": "value3", "label4": "value4"}), 

161 ], 

162 ) 

163 def test_extra_configmaps_secrets_additional_labels(self, chart_labels, local_labels): 

164 k8s_objects = render_chart( 

165 name=RELEASE_NAME, 

166 values={ 

167 "labels": chart_labels, 

168 "extraSecrets": { 

169 "{{ .Release.Name }}-extra-secret-1": { 

170 "labels": local_labels, 

171 "stringData": "data: secretData", 

172 } 

173 }, 

174 "extraConfigMaps": { 

175 "{{ .Release.Name }}-extra-configmap-1": { 

176 "labels": local_labels, 

177 "data": "data: configData", 

178 } 

179 }, 

180 }, 

181 show_only=["templates/configmaps/extra-configmaps.yaml", "templates/secrets/extra-secrets.yaml"], 

182 ) 

183 common_labels = { 

184 "release": RELEASE_NAME, 

185 "heritage": "Helm", 

186 "chart": mock.ANY, 

187 } 

188 for k8s_object in k8s_objects: 

189 assert k8s_object["metadata"]["labels"] == {**common_labels, **chart_labels, **local_labels} 

190 

191 def test_extra_configmaps_secrets_additional_annotations(self): 

192 k8s_objects = render_chart( 

193 name=RELEASE_NAME, 

194 values={ 

195 "extraSecrets": { 

196 "{{ .Release.Name }}-extra-secret-1": { 

197 "annotations": {"test_annotation": "test_annotation_value"}, 

198 "stringData": "data: secretData", 

199 } 

200 }, 

201 "extraConfigMaps": { 

202 "{{ .Release.Name }}-extra-configmap-1": { 

203 "annotations": {"test_annotation": "test_annotation_value"}, 

204 "data": "data: configData", 

205 } 

206 }, 

207 }, 

208 show_only=["templates/configmaps/extra-configmaps.yaml", "templates/secrets/extra-secrets.yaml"], 

209 ) 

210 

211 expected_annotations = { 

212 "helm.sh/hook": "pre-install,pre-upgrade", 

213 "helm.sh/hook-delete-policy": "before-hook-creation", 

214 "helm.sh/hook-weight": "0", 

215 "test_annotation": "test_annotation_value", 

216 } 

217 

218 for k8s_object in k8s_objects: 

219 assert k8s_object["metadata"]["annotations"] == expected_annotations