From a867908fc50b6cfbbeba587303797d8ae7ab3c80 Mon Sep 17 00:00:00 2001 From: Al Bailey Date: Tue, 14 Feb 2023 22:40:46 +0000 Subject: [PATCH] Adding unit tests to improve nfvi infra coverage The nfvi_infrastructure_api plugin has almost no unit test line coverage. This review adds some basic creation tests and validates some some static methods and properties. This improves overall code coverage from 22% to 23% This improves code coverage for that module from 0% to 6% This change has no impact on runtime. Test Plan: PASS: tox -e coverage Story: 2010531 Task: 47551 Signed-off-by: Al Bailey Change-Id: I24a532739e3bd62e88aab96cf6aaa3f607b5d198 --- .../tests/test_nfvi_infrastructure_api.py | 261 ++++++++++++++++++ 1 file changed, 261 insertions(+) create mode 100755 nfv/nfv-tests/nfv_unit_tests/tests/test_nfvi_infrastructure_api.py diff --git a/nfv/nfv-tests/nfv_unit_tests/tests/test_nfvi_infrastructure_api.py b/nfv/nfv-tests/nfv_unit_tests/tests/test_nfvi_infrastructure_api.py new file mode 100755 index 00000000..f58d696c --- /dev/null +++ b/nfv/nfv-tests/nfv_unit_tests/tests/test_nfvi_infrastructure_api.py @@ -0,0 +1,261 @@ +# +# Copyright (c) 2023 Wind River Systems, Inc. +# +# SPDX-License-Identifier: Apache-2.0 +# +import uuid + +from nfv_plugins.nfvi_plugins.nfvi_infrastructure_api import host_state +from nfv_plugins.nfvi_plugins.nfvi_infrastructure_api import NFVIInfrastructureAPI +from nfv_vim.nfvi.objects.v1 import HOST_AVAIL_STATUS +from nfv_vim.nfvi.objects.v1 import HOST_LABEL_KEYS +from nfv_vim.nfvi.objects.v1 import HOST_LABEL_VALUES +from nfv_vim.nfvi.objects.v1 import HOST_OPER_STATE + +from nfv_unit_tests.tests import testcase + +# todo(abailey): use already existing constants +CONTROLLER_PERSONALITY = 'controller' +WORKER_PERSONALITY = 'worker' +CONTROLLER_SUBFUNCTION = 'controller' +WORKER_SUBFUNCTION = 'worker' +ADMIN_STATE_UNLOCKED = "unlocked" + + +class TestNFVIInfrastructureAPI(testcase.NFVTestCase): + + def setUp(self): + """Setup for testing.""" + super(TestNFVIInfrastructureAPI, self).setUp() + + def tearDown(self): + """Cleanup testing setup.""" + super(TestNFVIInfrastructureAPI, self).tearDown() + + def test_create(self): + """Test that the plugin_api can be created""" + plugin_api = NFVIInfrastructureAPI() + self.assertIsNotNone(plugin_api) + # test the getter methods + self.assertEqual(plugin_api.name, NFVIInfrastructureAPI._name) + self.assertEqual(plugin_api.version, NFVIInfrastructureAPI._version) + self.assertEqual(plugin_api.provider, NFVIInfrastructureAPI._provider) + self.assertEqual(plugin_api.signature, NFVIInfrastructureAPI._signature) + + def test_static_host_supports_kubernetes(self): + """Test that the _host_supports_kubernetes static method""" + # personality is a string list. + aio_personality = ['controller', 'worker', ] + controller_personality = ['controller', ] + storage_personality = ['storage', ] + worker_personality = ['worker', ] + # aio / controller / worker support kubernetes + self.assertTrue(NFVIInfrastructureAPI._host_supports_kubernetes(aio_personality)) + self.assertTrue(NFVIInfrastructureAPI._host_supports_kubernetes(controller_personality)) + self.assertTrue(NFVIInfrastructureAPI._host_supports_kubernetes(worker_personality)) + # storage does not support kubernetes + self.assertFalse(NFVIInfrastructureAPI._host_supports_kubernetes(storage_personality)) + + def test_static_get_host_labels(self): + """Test the static _get_host_labels + + Extracts from a list of label_key:label_value + - openstack_compute + - openstack_control + - remote_storage + defaults: False, False, False + """ + OS_COMPUTE = HOST_LABEL_KEYS.OS_COMPUTE_NODE + OS_CONTROL = HOST_LABEL_KEYS.OS_CONTROL_PLANE + REMOTE_STORAGE = HOST_LABEL_KEYS.REMOTE_STORAGE + LABEL_ENABLED = HOST_LABEL_VALUES.ENABLED + LABEL_KEY = 'label_key' + LABEL_VALUE = 'label_value' + + # todo(abailey): Fix _get_host_labels KeyError caused by + # missing 'label_key' or 'label_value' + + # todo(abailey): Fix _get_host_labels KeyError caused by the commented out test + # ie: 'label_key' without 'label_value' + # test_data = [{LABEL_KEY: 'junk'}] + # self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + # test_data = [{LABEL_VALUE: 'junk'}] + # self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + + test_data = [{LABEL_KEY: 'junk', LABEL_VALUE: 'junk'}] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + test_data = [{LABEL_KEY: 'junk', LABEL_VALUE: LABEL_ENABLED}] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + + # test that OS_COMPUTE (first boolean) works when properly populated + # todo(abailey): Fix _get_host_labels KeyError caused by the commented out test + # test_data = [{LABEL_KEY: OS_COMPUTE}] + # self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + test_data = [{LABEL_KEY: OS_COMPUTE, LABEL_VALUE: 'junk'}] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + test_data = [{LABEL_KEY: OS_COMPUTE, LABEL_VALUE: LABEL_ENABLED}] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (True, False, False)) + + # test that OS_CONTROL (second boolean) works when properly populated + # todo(abailey): Fix _get_host_labels KeyError caused by the commented out test + # test_data = [{LABEL_KEY: OS_CONTROL}] + # self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + test_data = [{LABEL_KEY: OS_CONTROL, LABEL_VALUE: 'junk'}] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + test_data = [{LABEL_KEY: OS_CONTROL, LABEL_VALUE: LABEL_ENABLED}] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, True, False)) + + # test that REMOTE_STORAGE (third boolean) works when properly populated + # todo(abailey): Fix _get_host_labels KeyError caused by the commented out test + # test_data = [{LABEL_KEY: REMOTE_STORAGE}] + # self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + test_data = [{LABEL_KEY: REMOTE_STORAGE, LABEL_VALUE: 'junk'}] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, False)) + test_data = [{LABEL_KEY: REMOTE_STORAGE, LABEL_VALUE: LABEL_ENABLED}] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (False, False, True)) + + # that all three work.. + test_data = [ + {LABEL_KEY: OS_COMPUTE, LABEL_VALUE: LABEL_ENABLED}, + {LABEL_KEY: OS_CONTROL, LABEL_VALUE: LABEL_ENABLED}, + {LABEL_KEY: REMOTE_STORAGE, LABEL_VALUE: LABEL_ENABLED} + ] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (True, True, True)) + # that all three work and order not important + test_data = [ + {LABEL_KEY: REMOTE_STORAGE, LABEL_VALUE: LABEL_ENABLED}, + {LABEL_KEY: OS_COMPUTE, LABEL_VALUE: LABEL_ENABLED}, + {LABEL_KEY: OS_CONTROL, LABEL_VALUE: LABEL_ENABLED} + ] + self.assertEqual(NFVIInfrastructureAPI._get_host_labels(test_data), (True, True, True)) + + +class TestNFVIInfrastructureHostState(testcase.NFVTestCase): + + def get_host_state_args(self): + return { + "host_uuid": uuid.uuid4, + "host_name": "controller-0", + "host_personality": CONTROLLER_PERSONALITY, + "host_sub_functions": [CONTROLLER_SUBFUNCTION, + WORKER_SUBFUNCTION, ], + "host_admin_state": ADMIN_STATE_UNLOCKED, + "host_oper_state": HOST_OPER_STATE.ENABLED, + "host_avail_status": HOST_AVAIL_STATUS.AVAILABLE, + "sub_function_oper_state": HOST_OPER_STATE.ENABLED, + "sub_function_avail_status": HOST_AVAIL_STATUS.AVAILABLE, + "data_port_oper_state": HOST_OPER_STATE.ENABLED, + "data_port_avail_status": HOST_AVAIL_STATUS.AVAILABLE, + "data_port_fault_handling_enabled": False + } + + def test_host_state_aio_sx(self): + """Tests get_host_state defaults""" + # unlocked / enabled / available + + args = self.get_host_state_args() + admin_state, oper_state, avail_status, nfvi_data \ + = host_state(**args) + self.assertEqual(admin_state, ADMIN_STATE_UNLOCKED) + self.assertEqual(oper_state, HOST_OPER_STATE.ENABLED) + self.assertEqual(avail_status, HOST_AVAIL_STATUS.AVAILABLE) + + def test_host_state_aio_sx_disabled(self): + """Tests get_host_state host disabled""" + # unlocked / disabled / available + + args = self.get_host_state_args() + # override the default host oper state as disabled + args["host_oper_state"] = HOST_OPER_STATE.DISABLED + admin_state, oper_state, avail_status, nfvi_data \ + = host_state(**args) + self.assertEqual(admin_state, ADMIN_STATE_UNLOCKED) + self.assertEqual(oper_state, HOST_OPER_STATE.DISABLED) + self.assertEqual(avail_status, HOST_AVAIL_STATUS.AVAILABLE) + + def test_host_state_aio_sx_missing_fields(self): + """Tests get_host_state host missing some fields + + Set sub function operation state: None + Set sub function avail status: None + Set data port fields: None + + This may not be a valid scenario... + """ + # unlocked / None / None + + args = self.get_host_state_args() + # override the default host oper state as disabled + args["sub_function_oper_state"] = None + args["sub_function_avail_status"] = None + args["data_port_oper_state"] = None + + admin_state, oper_state, avail_status, nfvi_data \ + = host_state(**args) + self.assertEqual(admin_state, ADMIN_STATE_UNLOCKED) + self.assertIsNone(oper_state) + self.assertIsNone(avail_status) + + def test_host_state_worker(self): + """Tests get_host_state for a worker""" + # unlocked / enabled / available + + args = self.get_host_state_args() + args["host_name"] = "worker-0" + args["host_personality"] = WORKER_PERSONALITY + args["host_sub_functions"] = [WORKER_SUBFUNCTION, ] + args["data_port_fault_handling_enabled"] = True + + admin_state, oper_state, avail_status, nfvi_data \ + = host_state(**args) + self.assertEqual(admin_state, ADMIN_STATE_UNLOCKED) + self.assertEqual(oper_state, HOST_OPER_STATE.ENABLED) + self.assertEqual(avail_status, HOST_AVAIL_STATUS.AVAILABLE) + + def test_host_state_worker_unknown(self): + """Tests get_host_state for a worker with unknown states""" + # unlocked / unknown / unknown + + args = self.get_host_state_args() + args["host_name"] = "worker-0" + args["host_personality"] = WORKER_PERSONALITY + args["host_sub_functions"] = [WORKER_SUBFUNCTION, ] + args["data_port_fault_handling_enabled"] = True + args["data_port_oper_state"] = None + + admin_state, oper_state, avail_status, nfvi_data \ + = host_state(**args) + self.assertEqual(admin_state, ADMIN_STATE_UNLOCKED) + self.assertEqual(oper_state, HOST_OPER_STATE.UNKNOWN) + self.assertEqual(avail_status, HOST_AVAIL_STATUS.UNKNOWN) + + def test_host_state_worker_failed(self): + """Tests get_host_state for a failed worker""" + # unlocked / enabled / failed + + args = self.get_host_state_args() + args["host_name"] = "worker-0" + args["host_personality"] = WORKER_PERSONALITY + args["host_sub_functions"] = [WORKER_SUBFUNCTION, ] + args["data_port_fault_handling_enabled"] = True + args["data_port_avail_status"] = HOST_AVAIL_STATUS.OFFLINE + + admin_state, oper_state, avail_status, nfvi_data \ + = host_state(**args) + self.assertEqual(admin_state, ADMIN_STATE_UNLOCKED) + self.assertEqual(oper_state, HOST_OPER_STATE.ENABLED) + self.assertEqual(avail_status, HOST_AVAIL_STATUS.FAILED_COMPONENT) + + def test_host_state_controller(self): + """Tests get_host_state for a std controller""" + # unlocked / enabled / available + + args = self.get_host_state_args() + args["host_personality"] = CONTROLLER_PERSONALITY + args["host_sub_functions"] = [CONTROLLER_SUBFUNCTION, ] + + admin_state, oper_state, avail_status, nfvi_data \ + = host_state(**args) + self.assertEqual(admin_state, ADMIN_STATE_UNLOCKED) + self.assertEqual(oper_state, HOST_OPER_STATE.ENABLED) + self.assertEqual(avail_status, HOST_AVAIL_STATUS.AVAILABLE)