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 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
  21 # Common functions for unit testing
  22 def _set_k8s_configuration():
  23     ''' Set up the basic k8s configuration '''
  24     return  {
  25         "image_pull_secrets" : ["secret0", "secret1"],
  26         "filebeat" : {
  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"
  33         },
  34         "tls" : {
  35             "cert_path": "/opt/certs",
  36             "image": "tlsrepo/tls-init-container:1.2.3",
  37             "component_cert_dir": "/opt/dcae/cacert"
  38         },
  39         "external_cert": {
  40             "image_tag": "repo/oom-certservice-client:2.1.0",
  41             "request_url" : "https://request:1010/url",
  42             "timeout" : "30000",
  43             "country" : "US",
  44             "organization" : "Linux-Foundation",
  45             "state" : "California",
  46             "organizational_unit" : "ONAP",
  47             "location" : "San-Francisco",
  48             "keystore_password" : "secret1",
  49             "truststore_password" : "secret2"
  50         },
  51         "cert_post_processor": {
  52             "image_tag": "repo/oom-cert-post-processor:2.1.0"
  53         },
  54         "cbs": {
  55             "base_url": "https://config-binding-service:10443/service_component_all/test-component"
  56         }
  57     }
  58
  59 def _set_resources():
  60     ''' Set resources '''
  61     return {
  62         "limits": {
  63             "cpu" : 0.5,
  64             "memory" : "2Gi"
  65         },
  66         "requests": {
  67             "cpu" : 0.5,
  68             "memory" : "2Gi"
  69         }
  70     }
  71
  72 def _set_common_kwargs():
  73     ''' Set kwargs common to all test cases '''
  74     return {
  75         "volumes": [
  76             {"host":{"path": "/path/on/host"}, "container":{"bind":"/path/on/container","mode":"rw"}}
  77         ],
  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"}
  82     }
  83
  84 def _get_item_by_name(list, name):
  85     ''' Search a list of k8s API objects with the specified name '''
  86     for item in list:
  87         if item.name == name:
  88             return item
  89     return None
  90
  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
  94
  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"
 100
 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"
 114
 115     # Check environment variables
 116     env = app_container.env
 117     check_env_var(env, "NAME0", "value0")
 118     check_env_var(env, "NAME1", "value1")
 119
 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"
 129
 130     # Needs to be correctly labeled so that the Service can find it
 131     assert dep.spec.template.metadata.labels["app"] == "testcomponent"
 132
 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/oom-certservice-client:2.1.0"
 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/oom/certservice/certs/"
 143
 144     expected_envs = {
 145             "REQUEST_URL": "https://request:1010/url",
 146             "REQUEST_TIMEOUT": "30000",
 147             "OUTPUT_PATH": "/path/to/container/cert/directory/external",
 148             "OUTPUT_TYPE": "P12",
 149             "CA_NAME": "myname",
 150             "COMMON_NAME": "mycommonname",
 151             "ORGANIZATION": "Linux-Foundation",
 152             "ORGANIZATION_UNIT": "ONAP",
 153             "LOCATION": "San-Francisco",
 154             "STATE": "California",
 155             "COUNTRY": "US",
 156             "SANS": "mysans",
 157             "KEYSTORE_PATH": "/etc/onap/oom/certservice/certs/certServiceClient-keystore.jks",
 158             "KEYSTORE_PASSWORD": "secret1",
 159             "TRUSTSTORE_PATH": "/etc/onap/oom/certservice/certs/truststore.jks",
 160             "TRUSTSTORE_PASSWORD": "secret2"}
 161
 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])
 165
 166 def verify_cert_post_processor(dep):
 167     cert_container = dep.spec.template.spec.init_containers[2]
 168     print(cert_container)
 169     assert cert_container.image == "repo/oom-cert-post-processor:2.1.0"
 170     assert cert_container.name == "cert-post-processor"
 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/"
 174
 175     expected_envs = {
 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"
 180     }
 181
 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])
 185
 186
 187 def do_deploy(tls_info=None):
 188     ''' Common deployment operations '''
 189     import k8sclient.k8sclient
 190
 191     k8s_test_config = _set_k8s_configuration()
 192
 193     kwargs = _set_common_kwargs()
 194     kwargs['resources'] = _set_resources()
 195
 196     if tls_info:
 197         kwargs["tls_info"] = tls_info
 198
 199     dep, deployment_description = k8sclient.k8sclient.deploy(k8s_ctx(), "k8stest", "testcomponent", "example.com/testcomponent:1.4.3", 1, False, k8s_test_config, **kwargs)
 200
 201     # Make sure all of the basic k8s parameters are correct
 202     verify_common(dep, deployment_description)
 203
 204     return dep, deployment_description
 205
 206
 207 def do_deploy_ext(ext_tls_info):
 208     ''' Common deployment operations '''
 209     import k8sclient.k8sclient
 210
 211     k8s_test_config = _set_k8s_configuration()
 212
 213     kwargs = _set_common_kwargs()
 214     kwargs['resources'] = _set_resources()
 215     kwargs["external_cert"] = ext_tls_info
 216
 217     dep, deployment_description = k8sclient.k8sclient.deploy(k8s_ctx(), "k8stest", "testcomponent", "example.com/testcomponent:1.4.3", 1, False, k8s_test_config, **kwargs)
 218
 219     # Make sure all of the basic k8s parameters are correct
 220     verify_common(dep, deployment_description)
 221
 222     return dep, deployment_description
 223
 224 class k8s_logger:
 225     def info(self, text):
 226         print(text)
 227
 228 class k8s_ctx:
 229     logger = k8s_logger()
 230
 231