From facc073e9e4c3208a3f0d1652a03e7e20cc6bc47 Mon Sep 17 00:00:00 2001 From: SidneyAn Date: Wed, 26 Sep 2018 11:01:03 +0800 Subject: [PATCH] Add unit tests for nfv kubernetes api (config taint) add test cases to test add and delete node taints mock functions: kubernetes.config.load_kube_config kubernetes.client.CoreV1Api.read_node kubernetes.client.CoreV1Api.patch_node mock functions above has been tested on a real kubernetes cluster, which ensure mock functions have the same behaviors with kubernetes apis. Story: 2002843 Task: 24909 Change-Id: I12d2a15404afc38274aaa87841a63ee576aaf095 Signed-off-by: SidneyAn --- .../tests/test_plugin_kubernetes_client.py | 202 ++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100755 nfv/nfv-tests/nfv_unit_tests/tests/test_plugin_kubernetes_client.py diff --git a/nfv/nfv-tests/nfv_unit_tests/tests/test_plugin_kubernetes_client.py b/nfv/nfv-tests/nfv_unit_tests/tests/test_plugin_kubernetes_client.py new file mode 100755 index 00000000..9d498fc2 --- /dev/null +++ b/nfv/nfv-tests/nfv_unit_tests/tests/test_plugin_kubernetes_client.py @@ -0,0 +1,202 @@ +# +# Copyright +# +# SPDX-License-Identifier: Apache-2.0 +# +import mock +import kubernetes +from kubernetes.client.rest import ApiException +from . import testcase +from nfv_plugins.nfvi_plugins.clients import kubernetes_client + + +def mock_load_kube_config(path): + return + + +def exchange_json_to_V1Node(body): + node = kubernetes.client.V1Node() + + # exchange taints only. + if 'spec' not in body: + return node + + node.spec = kubernetes.client.V1NodeSpec() + if 'taints' not in body['spec']: + return node + + node.spec.taints = [] + for taint in body['spec']['taints']: + if type(taint) is kubernetes.client.models.v1_taint.V1Taint: + node.spec.taints.append(taint) + continue + + if 'key' in taint and 'effect' in taint: + taintBody = kubernetes.client.V1Taint(taint['effect'], taint['key']) + if 'value' in taint: + taintBody.value = taint['value'] + node.spec.taints.append(taintBody) + + return node + + +@mock.patch('kubernetes.config.load_kube_config', mock_load_kube_config) +class TestNFVPluginsK8SNodeTaint(testcase.NFVTestCase): + + test_node_name = 'testNode' + test_key1 = 'testKey1' + test_value1 = 'testValue1' + test_key2 = 'testKey2' + test_value2 = 'testValue2' + test_node_repo = {} + + def setUp(self): + super(TestNFVPluginsK8SNodeTaint, self).setUp() + self.setup_node_repo(self.test_node_name) + + def mock_patch_node(obj, node_name, body): + if node_name in self.test_node_repo: + self.test_node_repo[node_name] = exchange_json_to_V1Node(body) + else: + raise ApiException + return self.test_node_repo[node_name] + + self.mocked_patch_node = mock.patch( + 'kubernetes.client.CoreV1Api.patch_node', mock_patch_node) + self.mocked_patch_node.start() + + def mock_read_node(obj, node_name): + if node_name in self.test_node_repo: + return self.test_node_repo[node_name] + else: + raise ApiException + + self.mocked_read_node = mock.patch( + 'kubernetes.client.CoreV1Api.read_node', mock_read_node) + self.mocked_read_node.start() + + def tearDown(self): + super(TestNFVPluginsK8SNodeTaint, self).tearDown() + self.mocked_patch_node.stop() + self.mocked_read_node.stop() + self.node_repo_clear() + + def check_taint_exist(self, node_name, effect, key, value): + try: + kube_client = kubernetes_client.get_client() + response = kube_client.read_node(node_name) + except ApiException as e: + return False + + taints = response.spec.taints + if taints is not None: + for taint in taints: + if (taint.key == key and + taint.effect == effect and + taint.value == value): + return True + return False + + def setup_node_repo(self, node_name): + body = kubernetes.client.V1Node() + body.spec = kubernetes.client.V1NodeSpec() + body.spec.taints = [] + + self.test_node_repo[node_name] = body + + def node_repo_clear(self): + self.test_node_repo.clear() + + def test_when_add_taint_and_get_then_get_it(self): + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) is False + kubernetes_client.taint_node(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) is True + + def test_when_add_two_taints_and_get_then_get_them(self): + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) is False + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key2, + self.test_value2) is False + + kubernetes_client.taint_node(self.test_node_name, + 'NoExecute', + self.test_key2, + self.test_value2) + kubernetes_client.taint_node(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) + + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) is True + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key2, + self.test_value2) is True + + def test_when_delete_exist_taint_and_get_then_get_none(self): + kubernetes_client.taint_node(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) is True + kubernetes_client.untaint_node(self.test_node_name, + 'NoExecute', + self.test_key1) + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) is False + + def test_when_delete_no_exist_taint_and_get_then_get_none(self): + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) is False + kubernetes_client.untaint_node(self.test_node_name, + 'NoExecute', + self.test_key1) + assert self.check_taint_exist(self.test_node_name, + 'NoExecute', + self.test_key1, + self.test_value1) is False + + def test_when_add_taint_twice_and_delete_it_and_get_then_get_none(self): + kubernetes_client.taint_node(self.test_node_name, + 'NoSchedule', + self.test_key1, + self.test_value1) + kubernetes_client.taint_node(self.test_node_name, + 'NoSchedule', + self.test_key1, + self.test_value1) + assert self.check_taint_exist(self.test_node_name, + 'NoSchedule', + self.test_key1, + self.test_value1) is True + + kubernetes_client.untaint_node(self.test_node_name, + 'NoSchedule', + self.test_key1) + assert self.check_taint_exist(self.test_node_name, + 'NoSchedule', + self.test_key1, + self.test_value1) is False