# ================================================================================
# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
# Copyright (c) 2020 Pantheon.tech. All rights reserved.
+# Copyright (c) 2020-2021 Nokia. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
+import pytest
-# Common functions for unit testing
-def _set_k8s_configuration():
- ''' Set up the basic k8s configuration '''
- return {
- "image_pull_secrets" : ["secret0", "secret1"],
- "filebeat" : {
- "log_path": "/var/log/onap",
- "data_path": "/usr/share/filebeat/data",
- "config_path": "/usr/share/filebeat/filebeat.yml",
- "config_subpath": "filebeat.yml",
- "image" : "filebeat-repo/filebeat:latest",
- "config_map" : "dcae-filebeat-configmap"
- },
- "tls" : {
- "cert_path": "/opt/certs",
- "image": "tlsrepo/tls-init-container:1.2.3",
- "component_cert_dir": "/opt/dcae/cacert"
- },
- "cbs": {
- "base_url": "https://config-binding-service:10443/service_component_all/test-component"
- }
- }
-
-def _set_resources():
- ''' Set resources '''
- return {
- "limits": {
- "cpu" : 0.5,
- "memory" : "2Gi"
- },
- "requests": {
- "cpu" : 0.5,
- "memory" : "2Gi"
- }
- }
-
-def _set_common_kwargs():
- ''' Set kwargs common to all test cases '''
- return {
- "volumes": [
- {"host":{"path": "/path/on/host"}, "container":{"bind":"/path/on/container","mode":"rw"}}
- ],
- "ports": ["80:0", "443:0"],
- "env": {"NAME0": "value0", "NAME1": "value1"},
- "log_info": {"log_directory": "/path/to/container/log/directory"},
- "readiness": {"type": "http", "endpoint" : "/ready"}
- }
def _get_item_by_name(list, name):
- ''' Search a list of k8s API objects with the specified name '''
+ """ Search a list of k8s API objects with the specified name """
for item in list:
if item.name == name:
return item
return None
+
def check_env_var(env_list, name, value):
e = _get_item_by_name(env_list, name)
assert e and e.value == value
-def verify_common(dep, deployment_description):
- ''' Check results common to all test cases '''
+
+def verify_label(dep):
+ assert dep.spec.template.metadata.labels["app"] == "testcomponent"
+
+
+def verify_deployment_desc(deployment_description):
assert deployment_description["deployment"] == "dep-testcomponent"
assert deployment_description["namespace"] == "k8stest"
assert deployment_description["services"][0] == "testcomponent"
- # For unit test purposes, we want to make sure that the deployment object
- # we're passing to the k8s API is correct
- app_container = dep.spec.template.spec.containers[0]
- assert app_container.image == "example.com/testcomponent:1.4.3"
- assert app_container.image_pull_policy == "IfNotPresent"
- assert len(app_container.ports) == 2
- assert app_container.ports[0].container_port == 80
- assert app_container.ports[1].container_port == 443
- assert app_container.readiness_probe.http_get.path == "/ready"
- assert app_container.readiness_probe.http_get.scheme == "HTTP"
- assert len(app_container.volume_mounts) == 3
- assert app_container.volume_mounts[0].mount_path == "/path/on/container"
- assert app_container.volume_mounts[1].mount_path == "/path/to/container/log/directory"
- # Check environment variables
+def verify_env_variables(app_container):
env = app_container.env
check_env_var(env, "NAME0", "value0")
check_env_var(env, "NAME1", "value1")
- # Should have a log container with volume mounts
- log_container = dep.spec.template.spec.containers[1]
+
+def verify_logs(log_container):
assert log_container.image == "filebeat-repo/filebeat:latest"
assert log_container.volume_mounts[0].mount_path == "/var/log/onap/testcomponent"
assert log_container.volume_mounts[0].name == "component-log"
assert log_container.volume_mounts[2].mount_path == "/usr/share/filebeat/filebeat.yml"
assert log_container.volume_mounts[2].name == "filebeat-conf"
- # Needs to be correctly labeled so that the Service can find it
- assert dep.spec.template.metadata.labels["app"] == "testcomponent"
+def verify_volumes(app_container):
+ assert len(app_container.volume_mounts) >= 2
+ assert app_container.volume_mounts[0].mount_path == "/path/on/container"
+ assert app_container.volume_mounts[-2].mount_path == "/path/to/container/log/directory"
-def do_deploy(tls_info=None):
- ''' Common deployment operations '''
- import k8sclient.k8sclient
- k8s_test_config = _set_k8s_configuration()
+def verify_rediness_probe(app_container):
+ assert app_container.readiness_probe.http_get.path == "/ready"
+ assert app_container.readiness_probe.http_get.scheme == "HTTP"
+
+
+def verify_image(app_container):
+ assert app_container.image == "example.com/testcomponent:1.4.3"
+ assert app_container.image_pull_policy == "IfNotPresent"
+
+
+def verify_ports(app_container):
+ assert len(app_container.ports) == 2
+ assert app_container.ports[0].container_port == 80
+ assert app_container.ports[1].container_port == 443
+
- kwargs = _set_common_kwargs()
- kwargs['resources'] = _set_resources()
+def verify_external_cert(dep):
+ cert_container = dep.spec.template.spec.init_containers[1]
+ print(cert_container)
+ assert cert_container.image == "repo/oom-certservice-client:2.1.0"
+ assert cert_container.name == "cert-service-client"
+ assert len(cert_container.volume_mounts) == 2
+ assert cert_container.volume_mounts[0].name == "tls-info"
+ assert cert_container.volume_mounts[0].mount_path == "/path/to/container/cert/directory/"
+ assert cert_container.volume_mounts[1].name == "tls-volume"
+ assert cert_container.volume_mounts[1].mount_path == "/etc/onap/oom/certservice/certs/"
+
+ expected_envs = {
+ "REQUEST_URL": "https://request:1010/url",
+ "REQUEST_TIMEOUT": "30000",
+ "OUTPUT_PATH": "/path/to/container/cert/directory/external",
+ "OUTPUT_TYPE": "P12",
+ "CA_NAME": "myname",
+ "COMMON_NAME": "mycommonname",
+ "ORGANIZATION": "Linux-Foundation",
+ "ORGANIZATION_UNIT": "ONAP",
+ "LOCATION": "San-Francisco",
+ "STATE": "California",
+ "COUNTRY": "US",
+ "SANS": "mysans",
+ "KEYSTORE_PATH": "/etc/onap/oom/certservice/certs/keystore.jks",
+ "TRUSTSTORE_PATH": "/etc/onap/oom/certservice/certs/truststore.jks"}
+
+
+ envs = {k.name: k.value for k in cert_container.env}
+ for k in expected_envs:
+ assert (k in envs and expected_envs[k] == envs[k])
+
+ envs_from_source = {k.name: k.value_from for k in cert_container.env}
+ expected_secret_key_ref = {
+ "KEYSTORE_PASSWORD": "oom-cert-service-client-tls-secret-password",
+ "TRUSTSTORE_PASSWORD": "oom-cert-service-client-tls-secret-password"
+ }
+ for key, value in expected_secret_key_ref.items():
+ assert (key in envs_from_source and str(envs_from_source[key]).__contains__(value))
+
+
+def verify_cert_post_processor(dep):
+ cert_container = dep.spec.template.spec.init_containers[2]
+ print(cert_container)
+ assert cert_container.image == "repo/oom-cert-post-processor:2.1.0"
+ assert cert_container.name == "cert-post-processor"
+ assert len(cert_container.volume_mounts) == 1
+ assert cert_container.volume_mounts[0].name == "tls-info"
+ assert cert_container.volume_mounts[0].mount_path == "/opt/dcae/cacert/"
+
+ expected_envs = {
+ "TRUSTSTORES_PATHS": "/opt/dcae/cacert/trust.jks:/opt/dcae/cacert/external/truststore.p12",
+ "TRUSTSTORES_PASSWORDS_PATHS": "/opt/dcae/cacert/trust.pass:/opt/dcae/cacert/external/truststore.pass",
+ "KEYSTORE_SOURCE_PATHS": "/opt/dcae/cacert/external/keystore.p12:/opt/dcae/cacert/external/keystore.pass",
+ "KEYSTORE_DESTINATION_PATHS": "/opt/dcae/cacert/cert.p12:/opt/dcae/cacert/p12.pass"
+ }
- if tls_info:
- kwargs["tls_info"] = tls_info
+ envs = {k.name: k.value for k in cert_container.env}
+ for k in expected_envs:
+ assert (k in envs and expected_envs[k] == envs[k])
- dep, deployment_description = k8sclient.k8sclient.deploy("k8stest", "testcomponent", "example.com/testcomponent:1.4.3", 1, False, k8s_test_config, **kwargs)
- # Make sure all of the basic k8s parameters are correct
- verify_common(dep, deployment_description)
+def do_deploy(k8s_test_config, kwargs):
+ """ Common deployment operations """
+ import k8sclient.k8sclient
+ dep, deployment_description = k8sclient.k8sclient.deploy(k8s_ctx(), "k8stest", "testcomponent",
+ "example.com/testcomponent:1.4.3", 1, False,
+ k8s_test_config, **kwargs)
return dep, deployment_description
+
+
+class k8s_logger:
+ def info(self, text):
+ print(text)
+
+
+class k8s_ctx:
+ logger = k8s_logger()
Code Review / dcaegen2 / platform / plugins.git / blobdiff