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-2021 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=========================================================
23 def _get_item_by_name(list, name):
24 """ Search a list of k8s API objects with the specified name """
31 def check_env_var(env_list, name, value):
32 e = _get_item_by_name(env_list, name)
33 assert e and e.value == value
36 def verify_label(dep):
37 assert dep.spec.template.metadata.labels["app"] == "testcomponent"
40 def verify_deployment_desc(deployment_description):
41 assert deployment_description["deployment"] == "dep-testcomponent"
42 assert deployment_description["namespace"] == "k8stest"
43 assert deployment_description["services"][0] == "testcomponent"
46 def verify_env_variables(app_container):
47 env = app_container.env
48 check_env_var(env, "NAME0", "value0")
49 check_env_var(env, "NAME1", "value1")
52 def verify_logs(log_container):
53 assert log_container.image == "filebeat-repo/filebeat:latest"
54 assert log_container.volume_mounts[0].mount_path == "/var/log/onap/testcomponent"
55 assert log_container.volume_mounts[0].name == "component-log"
56 assert log_container.volume_mounts[1].mount_path == "/usr/share/filebeat/data"
57 assert log_container.volume_mounts[1].name == "filebeat-data"
58 assert log_container.volume_mounts[2].mount_path == "/usr/share/filebeat/filebeat.yml"
59 assert log_container.volume_mounts[2].name == "filebeat-conf"
62 def verify_volumes(app_container):
63 assert len(app_container.volume_mounts) >= 2
64 assert app_container.volume_mounts[0].mount_path == "/path/on/container"
65 assert app_container.volume_mounts[-2].mount_path == "/path/to/container/log/directory"
68 def verify_rediness_probe(app_container):
69 assert app_container.readiness_probe.http_get.path == "/ready"
70 assert app_container.readiness_probe.http_get.scheme == "HTTP"
73 def verify_image(app_container):
74 assert app_container.image == "example.com/testcomponent:1.4.3"
75 assert app_container.image_pull_policy == "IfNotPresent"
78 def verify_ports(app_container):
79 assert len(app_container.ports) == 2
80 assert app_container.ports[0].container_port == 80
81 assert app_container.ports[1].container_port == 443
84 def verify_external_cert(dep):
85 cert_container = dep.spec.template.spec.init_containers[1]
87 assert cert_container.image == "repo/oom-certservice-client:2.1.0"
88 assert cert_container.name == "cert-service-client"
89 assert len(cert_container.volume_mounts) == 2
90 assert cert_container.volume_mounts[0].name == "tls-info"
91 assert cert_container.volume_mounts[0].mount_path == "/path/to/container/cert/directory/"
92 assert cert_container.volume_mounts[1].name == "tls-volume"
93 assert cert_container.volume_mounts[1].mount_path == "/etc/onap/oom/certservice/certs/"
96 "REQUEST_URL": "https://request:1010/url",
97 "REQUEST_TIMEOUT": "30000",
98 "OUTPUT_PATH": "/path/to/container/cert/directory/external",
101 "COMMON_NAME": "mycommonname",
102 "ORGANIZATION": "Linux-Foundation",
103 "ORGANIZATION_UNIT": "ONAP",
104 "LOCATION": "San-Francisco",
105 "STATE": "California",
108 "KEYSTORE_PATH": "/etc/onap/oom/certservice/certs/certServiceClient-keystore.jks",
109 "KEYSTORE_PASSWORD": "secret1",
110 "TRUSTSTORE_PATH": "/etc/onap/oom/certservice/certs/truststore.jks",
111 "TRUSTSTORE_PASSWORD": "secret2"}
113 envs = {k.name: k.value for k in cert_container.env}
114 for k in expected_envs:
115 assert (k in envs and expected_envs[k] == envs[k])
118 def verify_cert_post_processor(dep):
119 cert_container = dep.spec.template.spec.init_containers[2]
120 print(cert_container)
121 assert cert_container.image == "repo/oom-cert-post-processor:2.1.0"
122 assert cert_container.name == "cert-post-processor"
123 assert len(cert_container.volume_mounts) == 1
124 assert cert_container.volume_mounts[0].name == "tls-info"
125 assert cert_container.volume_mounts[0].mount_path == "/opt/dcae/cacert/"
128 "TRUSTSTORES_PATHS": "/opt/dcae/cacert/trust.jks:/opt/dcae/cacert/external/truststore.p12",
129 "TRUSTSTORES_PASSWORDS_PATHS": "/opt/dcae/cacert/trust.pass:/opt/dcae/cacert/external/truststore.pass",
130 "KEYSTORE_SOURCE_PATHS": "/opt/dcae/cacert/external/keystore.p12:/opt/dcae/cacert/external/keystore.pass",
131 "KEYSTORE_DESTINATION_PATHS": "/opt/dcae/cacert/cert.p12:/opt/dcae/cacert/p12.pass"
134 envs = {k.name: k.value for k in cert_container.env}
135 for k in expected_envs:
136 assert (k in envs and expected_envs[k] == envs[k])
139 def do_deploy(k8s_test_config, kwargs):
140 """ Common deployment operations """
141 import k8sclient.k8sclient
142 dep, deployment_description = k8sclient.k8sclient.deploy(k8s_ctx(), "k8stest", "testcomponent",
143 "example.com/testcomponent:1.4.3", 1, False,
144 k8s_test_config, **kwargs)
146 return dep, deployment_description
150 def info(self, text):
155 logger = k8s_logger()