From e0026851f176aaad885b791f1f896c856f37d1b9 Mon Sep 17 00:00:00 2001 From: LiangLu Date: Tue, 12 Jan 2021 14:22:19 +0800 Subject: [PATCH] improve UT coverage of kubernetes-driver related files improve coverage for: tacker/vnfm/infra_drivers/kubernetes/ |-kubernetes_driver.py |-translate_template.py |-k8s/ |-tosca_kube_object.py |-translate_inputs.py |-translate_outputs.py Implements: blueprint improve-ut-coverage Change-Id: Ic115b5278ae3ec1e1c4822f54b3caf70c45e7380 --- .../kubernetes/test_kubernetes_driver.py | 61 +++++++++++++ .../kubernetes/test_tosca_kube_object.py | 83 +++++++++++++++++ .../kubernetes/test_translate_inputs.py | 88 ++++++++++++++++++ .../kubernetes/test_translate_outputs.py | 64 +++++++++++-- .../kubernetes/test_translate_template.py | 89 +++++++++++++++++++ 5 files changed, 378 insertions(+), 7 deletions(-) create mode 100644 tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_tosca_kube_object.py create mode 100644 tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_inputs.py create mode 100644 tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_template.py diff --git a/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_kubernetes_driver.py b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_kubernetes_driver.py index e5b3fc5ab..4147b379d 100644 --- a/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_kubernetes_driver.py +++ b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_kubernetes_driver.py @@ -489,6 +489,67 @@ class TestKubernetes(base.TestCase): ) self.assertEqual(msg, exc.format_message()) + @mock.patch.object(translate_template.TOSCAToKubernetes, + 'deploy_kubernetes_objects') + def test_create(self, mock_deploy_kubernetes_objects): + auth_attr = fakes.fake_auth_attr() + vnf = { + 'vnfd': { + 'attributes': { + 'vnfd': { + 'tosca_definitions_version': 'tosca_simple_yaml_1_0'} + }}} + plugin = "" + mock_deploy_kubernetes_objects.return_value = \ + tosca_kube_object.ToscaKubeObject( + namespace='namespace').namespace + result = self.kubernetes.create(plugin, self.context, vnf, auth_attr) + self.assertEqual("namespace", result) + + @mock.patch.object(client.CoreV1Api, 'read_namespaced_service') + @mock.patch.object(client.CoreV1Api, 'list_namespaced_pod') + def test_create_wait( + self, mock_list_namespaced_pod, mock_read_namespaced_service): + vnf_dict = fakes.fake_vnf_dict() + fake_podlist = fakes.fake_pod_list() + mock_list_namespaced_pod.return_value = fake_podlist + mock_read_namespaced_service.return_value = fakes.fake_service() + vnf_id = vnf_dict['id'] + plugin = "" + auth_attr = utils.get_vim_auth_obj() + self.kubernetes.create_wait(plugin, + self.context, vnf_dict, vnf_id, auth_attr) + + @mock.patch.object(client.CoreV1Api, 'read_namespaced_config_map') + @mock.patch.object(client.CoreV1Api, 'patch_namespaced_config_map') + def test_update(self, mock_read_namespaced_config_map, + mock_patch_namespaced_config_map): + vnf_dict = fakes.fake_vnf_dict() + vnf = { + 'vnf': { + 'attributes': { + 'vnfd': { + 'tosca_definitions_version': 'tosca_simple_yaml_1_0'}, + 'config': 'config'}}} + mock_read_namespaced_config_map.return_value = client.V1ConfigMap( + data={'abc': 'abc', 'test': 'test'}) + mock_patch_namespaced_config_map.return_value = client.V1ConfigMap( + data={'abc': 'abc', 'test': 'test'}) + vnf_id = vnf_dict['id'] + plugin = "" + auth_attr = utils.get_vim_auth_obj() + with mock.patch('yaml.safe_load') as mock_safe_load: + mock_safe_load.return_value = { + 'config': 'test_config', 'test': { + 'test1': 'test1'}} + self.kubernetes.update( + plugin, + self.context, + vnf_id, + vnf_dict, + vnf, + auth_attr) + def test_pre_instantiation_vnf_artifacts_file_none(self): instantiate_vnf_req = objects.InstantiateVnfRequest( additional_params={'a': ["Files/kubernets/pod.yaml"]}) diff --git a/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_tosca_kube_object.py b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_tosca_kube_object.py new file mode 100644 index 000000000..e5ef2c31b --- /dev/null +++ b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_tosca_kube_object.py @@ -0,0 +1,83 @@ +# Copyright (C) 2021 FUJITSU +# 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. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from tacker.tests.unit import base +from tacker.vnfm.infra_drivers.kubernetes.k8s import tosca_kube_object + + +class TestToscaKubeObject(base.TestCase): + def setUp(self): + super(TestToscaKubeObject, self).setUp() + self.tosca_kube_object = tosca_kube_object.ToscaKubeObject( + name='name', + namespace='namespace', + mapping_ports='mappingports', + containers=[ + tosca_kube_object.Container( + name="name")], + network_name="network", + mgmt_connection_point=True, + scaling_object=[ + tosca_kube_object.ScalingObject( + scale_target_name='scalingname')], + service_type='servicetype', + labels={ + 'lable': 'lable'}, + annotations="annotations") + + def test_tosca_kube_object(self): + self.assertEqual('name', self.tosca_kube_object.name) + self.assertEqual('namespace', self.tosca_kube_object.namespace) + + +class TestContainerObject(base.TestCase): + def setUp(self): + super(TestContainerObject, self).setUp() + self.container_object = tosca_kube_object.Container( + name='container', + num_cpus=1, + mem_size="100MB", + image="ubuntu", + command='command', + args=['args'], + ports=['22'], + config='config' + ) + + def test_container_object(self): + self.assertEqual('container', self.container_object.name) + self.assertEqual(1, self.container_object.num_cpus) + self.assertEqual('100MB', self.container_object.mem_size) + self.assertEqual('ubuntu', self.container_object.image) + + +class TestScalingObject(base.TestCase): + def setUp(self): + super(TestScalingObject, self).setUp() + self.scaling_object = tosca_kube_object.ScalingObject( + scaling_name='scalingname', + min_replicas=1, + max_replicas=3, + scale_target_name="cp1", + target_cpu_utilization_percentage="40" + ) + + def test_scaling_object(self): + self.assertEqual('scalingname', self.scaling_object.scaling_name) + self.assertEqual(1, self.scaling_object.min_replicas) + self.assertEqual(3, self.scaling_object.max_replicas) + self.assertEqual("cp1", self.scaling_object.scale_target_name) + self.assertEqual( + "40", self.scaling_object.target_cpu_utilization_percentage) diff --git a/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_inputs.py b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_inputs.py new file mode 100644 index 000000000..40b5710b0 --- /dev/null +++ b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_inputs.py @@ -0,0 +1,88 @@ +# Copyright (C) 2020 FUJITSU +# 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. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import os + +from tacker.tests.unit import base +from tacker.tests.unit.vnfm.infra_drivers.kubernetes import fakes +from tacker.vnfm.infra_drivers.kubernetes.k8s import translate_inputs + + +class TestParser(base.TestCase): + def setUp(self): + super(TestParser, self).setUp() + self.k8s_client_dict = fakes.fake_k8s_client_dict() + self.vnfd_path = '../../../../etc/samples/sample_tosca_vnfc.yaml' + self.yaml_path = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + self.vnfd_path) + self.vnfd_dict = { + "tosca_definitions_version": "tosca_simple_profile_for_nfv_1_0_0", + "description": "Demo example", + "metadata": { + "template_name": "sample-tosca-vnfd"}, + "topology_template": { + "node_templates": { + "VDU1": { + "type": "tosca.nodes.nfv.VDU.Tacker", + "capabilities": { + "nfv_compute": { + "properties": { + "num_cpus": 1, + "mem_size": "512 MB", + "disk_size": "1 GB"}}}, + "properties": { + "vnfcs": { + "web_server": { + "mem_size": "100 MB", + "config": "config" + } + }, + "labels": [ + "label1:1", "label2:2" + ] + } + }, + "CP1": { + "type": "tosca.nodes.nfv.CP.Tacker", + "properties": { + "order": 0, + "management": True, + "anti_spoofing_protection": False}, + "requirements": [ + {"virtualLink": { + "node": "VL1"}}, + {"virtualBinding": { + "node": "VDU1"}}]}, + "VL1": { + "type": "tosca.nodes.nfv.VL", + "properties": { + "vendor": "Tacker", + "network_name": "net_mgmt"}} + } + } + } + self.parser = translate_inputs.Parser(self.vnfd_dict) + + def test_loader(self): + tosca_kube_object = self.parser.loader() + self.assertEqual(tosca_kube_object[0].name[:8], "svc-VDU1") + self.assertEqual(tosca_kube_object[0].containers[0].name, "web_server") + self.assertEqual( + tosca_kube_object[0].containers[0].mem_size, + 100000000) + self.assertEqual( + tosca_kube_object[0].labels, { + 'label1': '1', 'label2': '2'}) diff --git a/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_outputs.py b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_outputs.py index 982f0e087..7691b65d1 100644 --- a/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_outputs.py +++ b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_outputs.py @@ -13,6 +13,7 @@ # License for the specific language governing permissions and limitations # under the License. +from kubernetes import client import os from unittest import mock @@ -20,6 +21,7 @@ from tacker.common import exceptions from tacker.tests.unit import base from tacker.tests.unit import fake_request from tacker.tests.unit.vnfm.infra_drivers.kubernetes import fakes +from tacker.vnfm.infra_drivers.kubernetes.k8s import tosca_kube_object from tacker.vnfm.infra_drivers.kubernetes.k8s import translate_outputs @@ -31,8 +33,10 @@ class TestTransformer(base.TestCase): "kubernetes_api_resource/") self.k8s_client_dict = fakes.fake_k8s_client_dict() self.transfromer = translate_outputs.Transformer( - None, None, None, self.k8s_client_dict - ) + client.CoreV1Api, + client.AppsV1Api, + client.AutoscalingApi, + self.k8s_client_dict) def test_deploy_k8s_create_false(self): kubernetes_objects = [] @@ -45,9 +49,8 @@ class TestTransformer(base.TestCase): @mock.patch.object(translate_outputs.Transformer, "_select_k8s_client_and_api") def test_deploy_k8s(self, mock_k8s_client_and_api): - req = \ - fake_request.HTTPRequest.blank( - 'apis/apps/v1/namespaces/curryns/deployments') + req = fake_request.HTTPRequest.blank( + 'apis/apps/v1/namespaces/curryns/deployments') mock_k8s_client_and_api.return_value = req kubernetes_objects = [] k8s_obj = fakes.fake_k8s_dict() @@ -59,8 +62,7 @@ class TestTransformer(base.TestCase): def test_deployment(self): k8s_objs = self.transfromer.get_k8s_objs_from_yaml( - ['deployment.yaml'], self.yaml_path - ) + ['deployment.yaml'], self.yaml_path) self.assertIsNotNone(k8s_objs[0].get('object')) self.assertEqual(k8s_objs[0].get('namespace'), '') self.assertEqual(k8s_objs[0].get('object').kind, 'Deployment') @@ -424,3 +426,51 @@ class TestTransformer(base.TestCase): 'ControllerRevision') self.assertEqual(k8s_objs[0].get('object').api_version, 'apps/v1') + + def test_transform(self): + container_obj = tosca_kube_object.Container( + config='config:abc\nconfig2:bcd', + num_cpus=2, + mem_size=10, + name='container' + ) + tosca_kube_objects = [tosca_kube_object.ToscaKubeObject( + namespace='namespace', + name='name', + containers=[container_obj], + mapping_ports=["123"], + labels={} + )] + kubernetes_objects = self.transfromer.transform(tosca_kube_objects) + self.assertEqual(kubernetes_objects['namespace'], 'namespace') + self.assertEqual( + kubernetes_objects['objects'][0].data, { + 'config': 'abc', 'config2': 'bcd'}) + + @mock.patch.object(client.CoreV1Api, 'create_namespaced_config_map') + @mock.patch.object(client.AppsV1Api, 'create_namespaced_deployment') + @mock.patch.object(client.CoreV1Api, 'create_namespaced_service') + def test_deploy( + self, + mock_create_namespaced_config_map, + mock_create_namespaced_deployment, + mock_create_namespaced_service): + mock_create_namespaced_config_map.return_value = "" + mock_create_namespaced_deployment.return_value = "" + mock_create_namespaced_service.return_value = "" + container_obj = tosca_kube_object.Container( + config='config:abc\nconfig2:bcd', + num_cpus=2, + mem_size=10, + name='container' + ) + tosca_kube_objects = [tosca_kube_object.ToscaKubeObject( + namespace='namespace', + name='name', + containers=[container_obj], + mapping_ports=["123"], + labels={} + )] + kubernetes_objects = self.transfromer.transform(tosca_kube_objects) + result = self.transfromer.deploy(kubernetes_objects) + self.assertEqual(result, 'namespace,name') diff --git a/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_template.py b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_template.py new file mode 100644 index 000000000..a3eff9ed3 --- /dev/null +++ b/tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_template.py @@ -0,0 +1,89 @@ +# Copyright (C) 2021 FUJITSU +# 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. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from kubernetes import client +from unittest import mock + + +from tacker.tests.unit import base +from tacker.vnfm.infra_drivers.kubernetes.k8s import translate_outputs +from tacker.vnfm.infra_drivers.kubernetes import translate_template + + +class TestTOSCAToKubernetes(base.TestCase): + def setUp(self): + super(TestTOSCAToKubernetes, self).setUp() + self.vnf = { + "vnfd": { + "service_types": [ + { + "service_type": "vnfd", + "id": "ca0d8667-ce35-4f7a-9744-ac4bc7d5579d" + } + ], + "description": "Sample", + "tenant_id": "689708956a2d4ae0a27120d3aca6a560", + "created_at": "2016-10-20 07:38:54", + "updated_at": None, + "attributes": { + "vnfd": + "description: " + "Demo example\nmetadata: " + "{template_name: sample-tosca-vnfd}\n" + "topology_template:\n " + "node_templates:\n CP1:\n " + "properties: {anti_spoofing_protection: " + "false, management: true, order: 0}\n " + "requirements:\n " + "- virtualLink: {node: VL1}\n " + "- virtualBinding: {node: VDU1}\n " + "type: tosca.nodes.nfv.CP.Tacker\n " + "VDU1:\n " + "capabilities:\n " + "nfv_compute:\n " + "properties: {disk_size: 1 GB, " + "mem_size: 512 MB, num_cpus: 1}\n " + "properties: {mapping_ports: [80:80] , " + "vnfcs: {web:{mem_size: 100 MB, " + "config: param0:key1}}}\n " + "type: tosca.nodes.nfv.VDU.Tacker\n " + "VL1:\n properties: {network_name: " + "net_mgmt, vendor: Tacker}\n " + "type: tosca.nodes.nfv.VL\ntosca_definitions_version: " + "tosca_simple_profile_for_nfv_1_0_0\n" + }, + "id": "0fb827e7-32b0-4e5b-b300-e1b1dce8a831", + "name": "vnfd-sample", + "template_source": "onboarded or inline" + } + } + + self.core_v1_api_client = client.CoreV1Api + self.app_v1_api_client = client.AppsV1Api + self.scaling_api_client = client.AutoscalingApi + self.tosca_to_kubernetes_object = translate_template.TOSCAToKubernetes( + self.vnf, self.core_v1_api_client, self.app_v1_api_client, + self.scaling_api_client) + + def test_generate_tosca_kube_objects(self): + result = self.tosca_to_kubernetes_object.generate_tosca_kube_objects() + self.assertEqual(result[0].name[:8], "svc-VDU1") + self.assertEqual(result[0].containers[0].name, "web") + self.assertEqual(result[0].containers[0].mem_size, 100000000) + + @mock.patch.object(translate_outputs.Transformer, 'deploy') + def test_deploy_kuberentes_objects(self, mock_deploy): + mock_deploy.return_value = "name, namespace" + self.tosca_to_kubernetes_object.deploy_kubernetes_objects()