1 # ============LICENSE_START=======================================================
3 # ================================================================================
4 # Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
5 # Copyright (c) 2020 Pantheon.tech. All rights reserved.
6 # Copyright (c) 2020 Nokia. All rights reserved.
7 # ================================================================================
8 # Licensed under the Apache License, Version 2.0 (the "License");
9 # you may not use this file except in compliance with the License.
10 # You may obtain a copy of the License at
12 # http://www.apache.org/licenses/LICENSE-2.0
14 # Unless required by applicable law or agreed to in writing, software
15 # distributed under the License is distributed on an "AS IS" BASIS,
16 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 # See the License for the specific language governing permissions and
18 # limitations under the License.
19 # ============LICENSE_END=========================================================
21 # Common functions for unit testing
22 def _set_k8s_configuration():
23 ''' Set up the basic k8s configuration '''
25 "image_pull_secrets" : ["secret0", "secret1"],
27 "log_path": "/var/log/onap",
28 "data_path": "/usr/share/filebeat/data",
29 "config_path": "/usr/share/filebeat/filebeat.yml",
30 "config_subpath": "filebeat.yml",
31 "image" : "filebeat-repo/filebeat:latest",
32 "config_map" : "dcae-filebeat-configmap"
35 "cert_path": "/opt/certs",
36 "image": "tlsrepo/tls-init-container:1.2.3",
37 "component_cert_dir": "/opt/dcae/cacert"
40 "image_tag": "repo/aaf-certservice-client:1.2.3",
41 "request_url" : "https://request:1010/url",
44 "organization" : "Linux-Foundation",
45 "state" : "California",
46 "organizational_unit" : "ONAP",
47 "location" : "San-Francisco",
48 "keystore_password" : "secret1",
49 "truststore_password" : "secret2"
51 "truststore_merger": {
52 "image_tag": "repo/oom-truststore-merger:1.2.3"
55 "base_url": "https://config-binding-service:10443/service_component_all/test-component"
72 def _set_common_kwargs():
73 ''' Set kwargs common to all test cases '''
76 {"host":{"path": "/path/on/host"}, "container":{"bind":"/path/on/container","mode":"rw"}}
78 "ports": ["80:0", "443:0"],
79 "env": {"NAME0": "value0", "NAME1": "value1"},
80 "log_info": {"log_directory": "/path/to/container/log/directory"},
81 "readiness": {"type": "http", "endpoint" : "/ready"}
84 def _get_item_by_name(list, name):
85 ''' Search a list of k8s API objects with the specified name '''
91 def check_env_var(env_list, name, value):
92 e = _get_item_by_name(env_list, name)
93 assert e and e.value == value
95 def verify_common(dep, deployment_description):
96 ''' Check results common to all test cases '''
97 assert deployment_description["deployment"] == "dep-testcomponent"
98 assert deployment_description["namespace"] == "k8stest"
99 assert deployment_description["services"][0] == "testcomponent"
101 # For unit test purposes, we want to make sure that the deployment object
102 # we're passing to the k8s API is correct
103 app_container = dep.spec.template.spec.containers[0]
104 assert app_container.image == "example.com/testcomponent:1.4.3"
105 assert app_container.image_pull_policy == "IfNotPresent"
106 assert len(app_container.ports) == 2
107 assert app_container.ports[0].container_port == 80
108 assert app_container.ports[1].container_port == 443
109 assert app_container.readiness_probe.http_get.path == "/ready"
110 assert app_container.readiness_probe.http_get.scheme == "HTTP"
111 assert len(app_container.volume_mounts) == 3
112 assert app_container.volume_mounts[0].mount_path == "/path/on/container"
113 assert app_container.volume_mounts[1].mount_path == "/path/to/container/log/directory"
115 # Check environment variables
116 env = app_container.env
117 check_env_var(env, "NAME0", "value0")
118 check_env_var(env, "NAME1", "value1")
120 # Should have a log container with volume mounts
121 log_container = dep.spec.template.spec.containers[1]
122 assert log_container.image == "filebeat-repo/filebeat:latest"
123 assert log_container.volume_mounts[0].mount_path == "/var/log/onap/testcomponent"
124 assert log_container.volume_mounts[0].name == "component-log"
125 assert log_container.volume_mounts[1].mount_path == "/usr/share/filebeat/data"
126 assert log_container.volume_mounts[1].name == "filebeat-data"
127 assert log_container.volume_mounts[2].mount_path == "/usr/share/filebeat/filebeat.yml"
128 assert log_container.volume_mounts[2].name == "filebeat-conf"
130 # Needs to be correctly labeled so that the Service can find it
131 assert dep.spec.template.metadata.labels["app"] == "testcomponent"
133 def verify_external_cert(dep):
134 cert_container = dep.spec.template.spec.init_containers[1]
135 print(cert_container)
136 assert cert_container.image == "repo/aaf-certservice-client:1.2.3"
137 assert cert_container.name == "cert-service-client"
138 assert len(cert_container.volume_mounts) == 2
139 assert cert_container.volume_mounts[0].name == "tls-info"
140 assert cert_container.volume_mounts[0].mount_path == "/path/to/container/cert/directory/"
141 assert cert_container.volume_mounts[1].name == "tls-volume"
142 assert cert_container.volume_mounts[1].mount_path == "/etc/onap/aaf/certservice/certs/"
145 "REQUEST_URL": "https://request:1010/url",
146 "REQUEST_TIMEOUT": "30000",
147 "OUTPUT_PATH": "/path/to/container/cert/directory/external",
148 "OUTPUT_TYPE": "P12",
150 "COMMON_NAME": "mycommonname",
151 "ORGANIZATION": "Linux-Foundation",
152 "ORGANIZATION_UNIT": "ONAP",
153 "LOCATION": "San-Francisco",
154 "STATE": "California",
157 "KEYSTORE_PATH": "/etc/onap/aaf/certservice/certs/certServiceClient-keystore.jks",
158 "KEYSTORE_PASSWORD": "secret1",
159 "TRUSTSTORE_PATH": "/etc/onap/aaf/certservice/certs/truststore.jks",
160 "TRUSTSTORE_PASSWORD": "secret2"}
162 envs = {k.name: k.value for k in cert_container.env}
163 for k in expected_envs:
164 assert (k in envs and expected_envs[k] == envs[k])
166 def verify_truststore_merger(dep):
167 cert_container = dep.spec.template.spec.init_containers[2]
168 print(cert_container)
169 assert cert_container.image == "repo/oom-truststore-merger:1.2.3"
170 assert cert_container.name == "truststore-merger"
171 assert len(cert_container.volume_mounts) == 1
172 assert cert_container.volume_mounts[0].name == "tls-info"
173 assert cert_container.volume_mounts[0].mount_path == "/opt/dcae/cacert/"
176 "TRUSTSTORES_PATHS": "/opt/dcae/cacert/trust.jks:/opt/dcae/cacert/external/truststore.p12",
177 "TRUSTSTORES_PASSWORDS_PATHS": "/opt/dcae/cacert/trust.pass:/opt/dcae/cacert/external/truststore.pass",
178 "KEYSTORE_SOURCE_PATHS": "/opt/dcae/cacert/external/keystore.p12:/opt/dcae/cacert/external/keystore.pass",
179 "KEYSTORE_DESTINATION_PATHS": "/opt/dcae/cacert/cert.p12:/opt/dcae/cacert/p12.pass"
182 envs = {k.name: k.value for k in cert_container.env}
183 for k in expected_envs:
184 assert (k in envs and expected_envs[k] == envs[k])
187 def do_deploy(tls_info=None):
188 ''' Common deployment operations '''
189 import k8sclient.k8sclient
191 k8s_test_config = _set_k8s_configuration()
193 kwargs = _set_common_kwargs()
194 kwargs['resources'] = _set_resources()
197 kwargs["tls_info"] = tls_info
199 dep, deployment_description = k8sclient.k8sclient.deploy(k8s_ctx(), "k8stest", "testcomponent", "example.com/testcomponent:1.4.3", 1, False, k8s_test_config, **kwargs)
201 # Make sure all of the basic k8s parameters are correct
202 verify_common(dep, deployment_description)
204 return dep, deployment_description
207 def do_deploy_ext(ext_tls_info):
208 ''' Common deployment operations '''
209 import k8sclient.k8sclient
211 k8s_test_config = _set_k8s_configuration()
213 kwargs = _set_common_kwargs()
214 kwargs['resources'] = _set_resources()
215 kwargs["external_cert"] = ext_tls_info
217 dep, deployment_description = k8sclient.k8sclient.deploy(k8s_ctx(), "k8stest", "testcomponent", "example.com/testcomponent:1.4.3", 1, False, k8s_test_config, **kwargs)
219 # Make sure all of the basic k8s parameters are correct
220 verify_common(dep, deployment_description)
222 return dep, deployment_description
225 def info(self, text):
229 logger = k8s_logger()