# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
-
-# 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"
- },
- "external_cert": {
- "image_tag": "repo/oom-certservice-client:2.1.0",
- "request_url": "https://request:1010/url",
- "timeout": "30000",
- "country": "US",
- "organization": "Linux-Foundation",
- "state": "California",
- "organizational_unit": "ONAP",
- "location": "San-Francisco",
- "keystore_password": "secret1",
- "truststore_password": "secret2"
- },
- "cert_post_processor": {
- "image_tag": "repo/oom-cert-post-processor:2.1.0"
- },
- "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(config_map=None):
- ''' Set kwargs common to all test cases '''
- common_kwargs = {
- "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"}
- }
- if config_map is not None:
- common_kwargs["volumes"].append(config_map)
- return common_kwargs
+import pytest
def _get_item_by_name(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) >= 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"
- # 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 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
def verify_external_cert(dep):
assert (k in envs and expected_envs[k] == envs[k])
-def do_deploy(config_map=None, tls_info=None, ext_tls_info=None):
+def do_deploy(k8s_test_config, kwargs):
""" Common deployment operations """
import k8sclient.k8sclient
-
- k8s_test_config = _set_k8s_configuration()
-
- kwargs = _set_common_kwargs(config_map)
- kwargs['resources'] = _set_resources()
-
- if tls_info:
- kwargs["tls_info"] = tls_info
- if ext_tls_info:
- kwargs["external_cert"] = ext_tls_info
-
dep, deployment_description = k8sclient.k8sclient.deploy(k8s_ctx(), "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)
-
return dep, deployment_description
# Test k8sclient deployment functions
# Verify that for a given configuration and set of inputs, k8sclient generates the proper
# Kubernetes entities
-
import pytest
-from common import do_deploy
+from common import do_deploy, verify_ports, verify_image, verify_rediness_probe, verify_volumes, \
+ verify_logs, verify_env_variables, verify_deployment_desc, verify_label
from common import verify_external_cert
from common import verify_cert_post_processor
-
-def test_deploy_full_tls(mockk8sapi):
- """ Deploy component with a full TLS configuration, to act as a server """
-
- dep, deployment_description = do_deploy(tls_info={"use_tls": True, "cert_directory": "/path/to/container/cert/directory"})
-
- app_container = dep.spec.template.spec.containers[0]
- assert app_container.volume_mounts[2].mount_path == "/path/to/container/cert/directory"
-
-
-def test_deploy_tls_off(mockk8sapi):
- """ TLS client only, but with cert directory configured """
-
- dep, deployment_description = do_deploy(tls_info={"use_tls": False, "cert_directory": "/path/to/container/cert/directory"})
-
+K8S_CONFIGURATION = {
+ "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"
+ },
+ "external_cert": {
+ "image_tag": "repo/oom-certservice-client:2.1.0",
+ "request_url": "https://request:1010/url",
+ "timeout": "30000",
+ "country": "US",
+ "organization": "Linux-Foundation",
+ "state": "California",
+ "organizational_unit": "ONAP",
+ "location": "San-Francisco",
+ "keystore_password": "secret1",
+ "truststore_password": "secret2"
+ },
+ "cert_post_processor": {
+ "image_tag": "repo/oom-cert-post-processor:2.1.0"
+ },
+ "cbs": {
+ "base_url": "https://config-binding-service:10443/service_component_all/test-component"
+ }
+}
+
+BASIC_KWARGS = {
+ "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"
+ },
+ "resources": {
+ "limits": {
+ "cpu": 0.5,
+ "memory": "2Gi"
+ },
+ "requests": {
+ "cpu": 0.5,
+ "memory": "2Gi"
+ }
+ }
+}
+
+KWARGS_WITH_FULL_TLS = {"tls_info": {"use_tls": True, "cert_directory": "/path/to/container/cert/directory"}}
+KWARGS_TLS_OFF = {"tls_info": {"use_tls": False, "cert_directory": "/path/to/container/cert/directory"}}
+KWARGS_WITH_EXTERNAL_CERT = {"external_cert": {"external_cert_directory": "/path/to/container/cert/directory/",
+ "use_external_tls": True,
+ "cert_type": "P12",
+ "ca_name": "myname",
+ "external_certificate_parameters": {
+ "common_name": "mycommonname",
+ "sans": "mysans"}
+ }}
+
+KWARGS_WITH_CONFIG_MAP = {"config_volume": {"name": "myConfigMap"},
+ "container": {"bind": "/path/to/configMap", "mode": "ro"}}
+
+
+test_data = [(KWARGS_WITH_EXTERNAL_CERT, "/opt/dcae/cacert"),
+ (BASIC_KWARGS, "/opt/dcae/cacert"),
+ (KWARGS_TLS_OFF, "/path/to/container/cert/directory"),
+ (KWARGS_WITH_FULL_TLS, "/path/to/container/cert/directory")]
+
+
+@pytest.mark.parametrize("blueprint_dict, path", test_data)
+def test_deploy(mockk8sapi, blueprint_dict, path):
+ # given
+ kwargs = dict(BASIC_KWARGS)
+ kwargs.update(blueprint_dict)
+
+ # when
+ dep, deployment_description = do_deploy(K8S_CONFIGURATION, kwargs)
app_container = dep.spec.template.spec.containers[0]
- assert app_container.volume_mounts[2].mount_path == "/path/to/container/cert/directory"
+ log_container = dep.spec.template.spec.containers[1]
-
-def test_deploy_no_tls_info(mockk8sapi):
- """ TLS client only, but with cert directory configured """
-
- dep, deployment_description = do_deploy()
-
- app_container = dep.spec.template.spec.containers[0]
- assert app_container.volume_mounts[2].mount_path == "/opt/dcae/cacert"
+ # then
+ verify_label(dep)
+ assert app_container.volume_mounts[2].mount_path == path
+ verify_ports(app_container)
+ verify_image(app_container)
+ verify_rediness_probe(app_container)
+ verify_volumes(app_container)
+ verify_logs(log_container)
+ verify_env_variables(app_container)
+ verify_deployment_desc(deployment_description)
def test_deploy_external_cert(mockk8sapi):
""" Deploy component with external TLS configuration """
+ # given
+ kwargs = dict(BASIC_KWARGS)
+ kwargs.update(KWARGS_WITH_EXTERNAL_CERT)
- dep, deployment_description = do_deploy(
- ext_tls_info={"external_cert_directory": "/path/to/container/cert/directory/",
- "use_external_tls": True,
- "cert_type": "P12",
- "ca_name": "myname",
- "external_certificate_parameters": {
- "common_name": "mycommonname",
- "sans": "mysans"}
- })
+ # when
+ dep, deployment_description = do_deploy(K8S_CONFIGURATION, kwargs)
- app_container = dep.spec.template.spec.containers[0]
- assert app_container.volume_mounts[2].mount_path == "/opt/dcae/cacert"
-
- # Make sure all of the external init container parameters are correct
+ # then
verify_external_cert(dep)
verify_cert_post_processor(dep)
+
def test_deploy_config_map(mockk8sapi):
""" Deploy component with configMap in volumes """
- config_map = {"config_volume": {"name": "myConfigMap"}, "container": {"bind": "/path/to/configMap", "mode": "ro"}}
- dep, deployment_description = do_deploy(config_map)
+ # given
+ kwargs = dict(BASIC_KWARGS)
+ kwargs['volumes'].append(KWARGS_WITH_CONFIG_MAP)
+ # when
+ dep, deployment_description = do_deploy(K8S_CONFIGURATION, kwargs)
app_container = dep.spec.template.spec.containers[0]
- assert app_container.volume_mounts[1].mount_path == "/path/to/configMap"
\ No newline at end of file
+
+ # then
+ assert app_container.volume_mounts[1].mount_path == "/path/to/configMap"
Code Review / dcaegen2 / platform / plugins.git / commitdiff