k8s/tests/common.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   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
  11 #
  12 #      http://www.apache.org/licenses/LICENSE-2.0
  13 #
  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=========================================================
  20 import pytest
  21
  22
  23 def _get_item_by_name(list, name):
  24     """ Search a list of k8s API objects with the specified name """
  25     for item in list:
  26         if item.name == name:
  27             return item
  28     return None
  29
  30
  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
  34
  35
  36 def verify_label(dep):
  37     assert dep.spec.template.metadata.labels["app"] == "testcomponent"
  38
  39
  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"
  44
  45
  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")
  50
  51
  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"
  60
  61
  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"
  66
  67
  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"
  71
  72
  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"
  76
  77
  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
  82
  83
  84 def verify_external_cert(dep):
  85     cert_container = dep.spec.template.spec.init_containers[1]
  86     print(cert_container)
  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/"
  94
  95     expected_envs = {
  96         "REQUEST_URL": "https://request:1010/url",
  97         "REQUEST_TIMEOUT": "30000",
  98         "OUTPUT_PATH": "/path/to/container/cert/directory/external",
  99         "OUTPUT_TYPE": "P12",
 100         "CA_NAME": "myname",
 101         "COMMON_NAME": "mycommonname",
 102         "ORGANIZATION": "Linux-Foundation",
 103         "ORGANIZATION_UNIT": "ONAP",
 104         "LOCATION": "San-Francisco",
 105         "STATE": "California",
 106         "COUNTRY": "US",
 107         "SANS": "mysans",
 108         "KEYSTORE_PATH": "/etc/onap/oom/certservice/certs/keystore.jks",
 109         "TRUSTSTORE_PATH": "/etc/onap/oom/certservice/certs/truststore.jks"}
 110
 111
 112     envs = {k.name: k.value for k in cert_container.env}
 113     for k in expected_envs:
 114         assert (k in envs and expected_envs[k] == envs[k])
 115
 116     envs_from_source = {k.name: k.value_from for k in cert_container.env}
 117     expected_secret_key_ref = {
 118         "KEYSTORE_PASSWORD": "oom-cert-service-client-tls-secret-password",
 119         "TRUSTSTORE_PASSWORD": "oom-cert-service-client-tls-secret-password"
 120     }
 121     for key, value in expected_secret_key_ref.items():
 122         assert (key in envs_from_source and str(envs_from_source[key]).__contains__(value))
 123
 124
 125 def verify_cert_post_processor(dep):
 126     cert_container = dep.spec.template.spec.init_containers[2]
 127     print(cert_container)
 128     assert cert_container.image == "repo/oom-cert-post-processor:2.1.0"
 129     assert cert_container.name == "cert-post-processor"
 130     assert len(cert_container.volume_mounts) == 1
 131     assert cert_container.volume_mounts[0].name == "tls-info"
 132     assert cert_container.volume_mounts[0].mount_path == "/opt/dcae/cacert/"
 133
 134     expected_envs = {
 135         "TRUSTSTORES_PATHS": "/opt/dcae/cacert/trust.jks:/opt/dcae/cacert/external/truststore.p12",
 136         "TRUSTSTORES_PASSWORDS_PATHS": "/opt/dcae/cacert/trust.pass:/opt/dcae/cacert/external/truststore.pass",
 137         "KEYSTORE_SOURCE_PATHS": "/opt/dcae/cacert/external/keystore.p12:/opt/dcae/cacert/external/keystore.pass",
 138         "KEYSTORE_DESTINATION_PATHS": "/opt/dcae/cacert/cert.p12:/opt/dcae/cacert/p12.pass"
 139     }
 140
 141     envs = {k.name: k.value for k in cert_container.env}
 142     for k in expected_envs:
 143         assert (k in envs and expected_envs[k] == envs[k])
 144
 145
 146 def do_deploy(k8s_test_config, kwargs):
 147     """ Common deployment operations """
 148     import k8sclient.k8sclient
 149     dep, deployment_description = k8sclient.k8sclient.deploy(k8s_ctx(), "k8stest", "testcomponent",
 150                                                              "example.com/testcomponent:1.4.3", 1, False,
 151                                                              k8s_test_config, **kwargs)
 152
 153     return dep, deployment_description
 154
 155
 156 class k8s_logger:
 157     def info(self, text):
 158         print(text)
 159
 160
 161 class k8s_ctx:
 162     logger = k8s_logger()