262 lines
12 KiB
Python
Executable File
262 lines
12 KiB
Python
Executable File
#
|
|
# 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)
|