Increase the unit test coverage of nova_helper.py

Increase coverage from 61% to 86%

Change-Id: I639168108e8076b2cccc02902793d77f6e94c1a5
This commit is contained in:
chenke 2018-11-13 15:12:39 +08:00
parent 8636770253
commit c7800225f1
1 changed files with 330 additions and 28 deletions

View File

@ -19,9 +19,15 @@
import time
from novaclient import api_versions
import mock
import glanceclient.exc as glexceptions
import novaclient.exceptions as nvexceptions
from watcher.common import clients
from watcher.common import exception
from watcher.common import nova_helper
from watcher.common import utils
from watcher.tests import base
@ -48,6 +54,16 @@ class TestNovaHelper(base.TestCase):
return server
@staticmethod
def fake_hypervisor(*args, **kwargs):
hypervisor = mock.MagicMock()
hypervisor.id = args[0]
service_dict = {"host": args[1]}
hypervisor.service = service_dict
hypervisor.hypervisor_hostname = args[1]
return hypervisor
@staticmethod
def fake_migration(*args, **kwargs):
migration = mock.MagicMock()
@ -55,19 +71,24 @@ class TestNovaHelper(base.TestCase):
return migration
@staticmethod
def fake_nova_find_list(nova_util, find=None, list=None):
nova_util.nova.servers.get.return_value = find
def fake_nova_find_list(nova_util, fake_find=None, fake_list=None):
nova_util.nova.servers.get.return_value = fake_find
if list is None:
nova_util.nova.servers.list.return_value = []
else:
nova_util.nova.servers.list.return_value = [list]
nova_util.nova.servers.list.return_value = [fake_list]
@staticmethod
def fake_nova_migration_list(nova_util, list=None):
def fake_nova_hypervisor_list(nova_util, fake_find=None, fake_list=None):
nova_util.nova.hypervisors.get.return_value = fake_find
nova_util.nova.hypervisors.list.return_value = fake_list
@staticmethod
def fake_nova_migration_list(nova_util, fake_list=None):
if list is None:
nova_util.nova.server_migrations.list.return_value = []
nova_util.nova.server_migrations.list.return_value = None
else:
nova_util.nova.server_migration.list.return_value = [list]
nova_util.nova.server_migration.list.return_value = [fake_list]
@staticmethod
def fake_live_migrate(server, *args, **kwargs):
@ -94,6 +115,37 @@ class TestNovaHelper(base.TestCase):
server.migrate.side_effect = side_effect
def test_get_compute_node_by_hostname(
self, mock_glance, mock_cinder, mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
hypervisor_id = utils.generate_uuid()
hypervisor_name = "fake_hypervisor_1"
hypervisor = self.fake_hypervisor(hypervisor_id, hypervisor_name)
self.fake_nova_hypervisor_list(
nova_util,
fake_find=hypervisor,
fake_list=[hypervisor])
nova_util.nova.hypervisors.search.return_value = [hypervisor]
# verify that the compute node can be obtained normally by name
self.assertEqual(
nova_util.get_compute_node_by_hostname(hypervisor_name),
hypervisor)
# verify that getting the compute node with the wrong name
# will throw an exception.
self.assertRaises(
exception.ComputeNodeNotFound,
nova_util.get_compute_node_by_hostname,
"exception_hypervisor_1")
# verify that when the result of getting the compute node is empty
# will throw an exception.
nova_util.nova.hypervisors.search.return_value = []
self.assertRaises(
exception.ComputeNodeNotFound,
nova_util.get_compute_node_by_hostname,
hypervisor_name)
@mock.patch.object(time, 'sleep', mock.Mock())
def test_stop_instance(self, mock_glance, mock_cinder, mock_neutron,
mock_nova):
@ -101,7 +153,10 @@ class TestNovaHelper(base.TestCase):
instance_id = utils.generate_uuid()
server = self.fake_server(instance_id)
setattr(server, 'OS-EXT-STS:vm_state', 'stopped')
self.fake_nova_find_list(nova_util, find=server, list=server)
self.fake_nova_find_list(
nova_util,
fake_find=server,
fake_list=server)
result = nova_util.stop_instance(instance_id)
self.assertTrue(result)
@ -110,18 +165,61 @@ class TestNovaHelper(base.TestCase):
result = nova_util.stop_instance(instance_id)
self.assertFalse(result)
self.fake_nova_find_list(nova_util, find=server, list=None)
self.fake_nova_find_list(nova_util, fake_find=server, fake_list=None)
result = nova_util.stop_instance(instance_id)
self.assertFalse(result)
# verify that the method will return True when the state of instance
# is in the expected state.
setattr(server, 'OS-EXT-STS:vm_state', 'active')
with mock.patch.object(
nova_util,
'wait_for_instance_state',
return_value=True
) as mock_instance_state:
result = nova_util.stop_instance(instance_id)
self.assertTrue(result)
mock_instance_state.assert_called_once_with(
mock.ANY,
"stopped",
8,
10)
# verify that the method stop_instance will return False when the
# server is not available.
nova_util.nova.servers.get.return_value = None
result = nova_util.stop_instance(instance_id)
self.assertFalse(result)
@mock.patch.object(time, 'sleep', mock.Mock())
def test_delete_instance(self, mock_glance, mock_cinder, mock_neutron,
mock_nova):
nova_util = nova_helper.NovaHelper()
instance_id = utils.generate_uuid()
# verify that the method will return False when the instance does
# not exist.
self.fake_nova_find_list(nova_util, fake_find=None, fake_list=None)
result = nova_util.delete_instance(instance_id)
self.assertFalse(result)
# verify that the method will return True when the instance exists.
server = self.fake_server(instance_id)
self.fake_nova_find_list(nova_util, fake_find=server, fake_list=None)
result = nova_util.delete_instance(instance_id)
self.assertTrue(result)
@mock.patch.object(time, 'sleep', mock.Mock())
def test_resize_instance(self, mock_glance, mock_cinder,
mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
server = self.fake_server(self.instance_uuid)
setattr(server, 'status', 'VERIFY_RESIZE')
self.fake_nova_find_list(nova_util, find=server, list=server)
self.fake_nova_find_list(
nova_util,
fake_find=server,
fake_list=server)
is_success = nova_util.resize_instance(self.instance_uuid,
self.flavor_name)
self.assertTrue(is_success)
@ -138,7 +236,10 @@ class TestNovaHelper(base.TestCase):
server = self.fake_server(self.instance_uuid)
setattr(server, 'OS-EXT-SRV-ATTR:host',
self.destination_node)
self.fake_nova_find_list(nova_util, find=server, list=server)
self.fake_nova_find_list(
nova_util,
fake_find=server,
fake_list=server)
is_success = nova_util.live_migrate_instance(
self.instance_uuid, self.destination_node
)
@ -146,12 +247,31 @@ class TestNovaHelper(base.TestCase):
setattr(server, 'OS-EXT-SRV-ATTR:host',
self.source_node)
self.fake_nova_find_list(nova_util, find=server, list=None)
self.fake_nova_find_list(nova_util, fake_find=server, fake_list=None)
is_success = nova_util.live_migrate_instance(
self.instance_uuid, self.destination_node
)
self.assertFalse(is_success)
# verify that the method will return False when the instance does
# not exist.
setattr(server, 'OS-EXT-SRV-ATTR:host',
self.source_node)
self.fake_nova_find_list(nova_util, fake_find=None, fake_list=None)
is_success = nova_util.live_migrate_instance(
self.instance_uuid, self.destination_node
)
self.assertFalse(is_success)
# verify that the method will return False when the instance status
# is in other cases.
setattr(server, 'status', 'fake_status')
self.fake_nova_find_list(nova_util, fake_find=server, fake_list=None)
is_success = nova_util.live_migrate_instance(
self.instance_uuid, None
)
self.assertFalse(is_success)
@mock.patch.object(time, 'sleep', mock.Mock())
def test_live_migrate_instance_with_task_state(
self, mock_glance, mock_cinder, mock_neutron, mock_nova):
@ -160,14 +280,17 @@ class TestNovaHelper(base.TestCase):
setattr(server, 'OS-EXT-SRV-ATTR:host',
self.source_node)
setattr(server, 'OS-EXT-STS:task_state', '')
self.fake_nova_find_list(nova_util, find=server, list=None)
self.fake_nova_find_list(nova_util, fake_find=server, fake_list=None)
nova_util.live_migrate_instance(
self.instance_uuid, self.destination_node
)
time.sleep.assert_not_called()
setattr(server, 'OS-EXT-STS:task_state', 'migrating')
self.fake_nova_find_list(nova_util, find=server, list=server)
self.fake_nova_find_list(
nova_util,
fake_find=server,
fake_list=server)
nova_util.live_migrate_instance(
self.instance_uuid, self.destination_node
)
@ -179,7 +302,10 @@ class TestNovaHelper(base.TestCase):
nova_util = nova_helper.NovaHelper()
server = self.fake_server(self.instance_uuid)
self.destination_node = None
self.fake_nova_find_list(nova_util, find=server, list=server)
self.fake_nova_find_list(
nova_util,
fake_find=server,
fake_list=server)
self.fake_live_migrate(server)
is_success = nova_util.live_migrate_instance(
self.instance_uuid, self.destination_node
@ -189,7 +315,7 @@ class TestNovaHelper(base.TestCase):
def test_watcher_non_live_migrate_instance_not_found(
self, mock_glance, mock_cinder, mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
self.fake_nova_find_list(nova_util, find=None, list=None)
self.fake_nova_find_list(nova_util, fake_find=None, fake_list=None)
is_success = nova_util.watcher_non_live_migrate_instance(
self.instance_uuid,
@ -206,9 +332,12 @@ class TestNovaHelper(base.TestCase):
self.source_node)
setattr(server, 'OS-EXT-STS:task_state', None)
migration = self.fake_migration(2)
self.fake_nova_migration_list(nova_util, list=migration)
self.fake_nova_migration_list(nova_util, fake_list=migration)
self.fake_nova_find_list(nova_util, find=server, list=server)
self.fake_nova_find_list(
nova_util,
fake_find=server,
fake_list=server)
self.assertTrue(nova_util.abort_live_migrate(
self.instance_uuid, self.source_node, self.destination_node))
@ -219,11 +348,20 @@ class TestNovaHelper(base.TestCase):
self.instance_uuid, self.source_node, self.destination_node))
setattr(server, 'status', 'ERROR')
self.assertRaises(Exception,
nova_util.abort_live_migrate,
self.instance_uuid,
self.source_node,
self.destination_node)
self.assertRaises(
Exception,
nova_util.abort_live_migrate,
self.instance_uuid,
self.source_node,
self.destination_node)
server = self.fake_server(self.instance_uuid)
setattr(server, 'OS-EXT-STS:task_state', "fake_task_state")
setattr(server, 'OS-EXT-SRV-ATTR:host', self.destination_node)
self.fake_nova_find_list(nova_util, fake_find=server, fake_list=None)
self.fake_nova_migration_list(nova_util, fake_list=None)
self.assertFalse(nova_util.abort_live_migrate(
self.instance_uuid, self.source_node, self.destination_node))
def test_non_live_migrate_instance_no_destination_node(
self, mock_glance, mock_cinder, mock_neutron, mock_nova):
@ -232,7 +370,7 @@ class TestNovaHelper(base.TestCase):
setattr(server, 'OS-EXT-SRV-ATTR:host',
self.source_node)
self.destination_node = None
self.fake_nova_find_list(nova_util, find=server, list=server)
self.fake_nova_find_list(nova_util, fake_find=server, fake_list=server)
self.fake_cold_migrate(server)
self.fake_confirm_resize(server)
is_success = nova_util.watcher_non_live_migrate_instance(
@ -248,7 +386,10 @@ class TestNovaHelper(base.TestCase):
image = mock.MagicMock()
setattr(instance, 'OS-EXT-SRV-ATTR:host', self.source_node)
setattr(instance, 'OS-EXT-STS:vm_state', "stopped")
self.fake_nova_find_list(nova_util, find=instance, list=instance)
self.fake_nova_find_list(
nova_util,
fake_find=instance,
fake_list=instance)
image_uuid = 'fake-image-uuid'
nova_util.nova.servers.create_image.return_value = image
@ -304,11 +445,71 @@ class TestNovaHelper(base.TestCase):
mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
instance = self.fake_server(self.instance_uuid)
nova_util.nova.services.create.return_value = instance
nova_util.nova.services.get.return_value = instance
nova_util.nova.servers.create.return_value = instance
nova_util.nova.servers.get.return_value = instance
create_instance = nova_util.create_instance(self.source_node)
self.assertIsNotNone(create_instance)
self.assertEqual(create_instance, instance)
# verify that the method create_instance will return None when
# the method findall raises exception.
nova_util.nova.keypairs.findall.side_effect = nvexceptions.NotFound(
404)
instance = nova_util.create_instance(self.source_node)
self.assertIsNotNone(instance)
self.assertIsNone(instance)
nova_util.nova.keypairs.findall.side_effect = None
# verify that the method create_instance will return None when
# the method get raises exception.
nova_util.glance.images.get.side_effect = glexceptions.NotFound(404)
instance = nova_util.create_instance(self.source_node)
self.assertIsNone(instance)
nova_util.glance.images.get.side_effect = None
# verify that the method create_instance will return None when
# the method find raises exception.
nova_util.nova.flavors.find.side_effect = nvexceptions.NotFound(404)
instance = nova_util.create_instance(self.source_node)
self.assertIsNone(instance)
nova_util.nova.flavors.find.side_effect = None
# verify that the method create_instance will return None when
# the method get_security_group_id_from_name return None.
with mock.patch.object(
nova_util,
'get_security_group_id_from_name',
return_value=None
) as mock_security_group_id:
instance = nova_util.create_instance(self.source_node)
self.assertIsNone(instance)
mock_security_group_id.assert_called_once_with("default")
# verify that the method create_instance will return None when
# the method get_network_id_from_name return None.
with mock.patch.object(
nova_util,
'get_network_id_from_name',
return_value=None
) as mock_get_network_id:
instance = nova_util.create_instance(self.source_node)
self.assertIsNone(instance)
mock_get_network_id.assert_called_once_with("demo-net")
# verify that the method create_instance will not return None when
# the method wait_for_instance_status return True.
with mock.patch.object(
nova_util,
'wait_for_instance_status',
return_value=True
) as mock_instance_status:
instance = nova_util.create_instance(self.source_node)
self.assertIsNotNone(instance)
mock_instance_status.assert_called_once_with(
mock.ANY,
('ACTIVE', 'ERROR'),
5,
10)
@staticmethod
def fake_volume(**kwargs):
@ -325,7 +526,7 @@ class TestNovaHelper(base.TestCase):
mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
server = self.fake_server(self.instance_uuid)
self.fake_nova_find_list(nova_util, find=server, list=server)
self.fake_nova_find_list(nova_util, fake_find=server, fake_list=server)
old_volume = self.fake_volume(
status='in-use', attachments=[{'server_id': self.instance_uuid}])
@ -334,3 +535,104 @@ class TestNovaHelper(base.TestCase):
result = nova_util.swap_volume(old_volume, new_volume)
self.assertTrue(result)
# verify that the method will return False when the status of
# new_volume is 'fake-use'.
new_volume = self.fake_volume(
id=utils.generate_uuid(), status='fake-use')
result = nova_util.swap_volume(old_volume, new_volume)
self.assertFalse(result)
@mock.patch.object(time, 'sleep', mock.Mock())
def test_wait_for_volume_status(self, mock_glance, mock_cinder,
mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
# verify that the method will return True when the status of volume
# is in the expected status.
fake_volume_1 = self.fake_volume(status='in-use')
nova_util.cinder.volumes.get.return_value = fake_volume_1
result = nova_util.wait_for_volume_status(
fake_volume_1,
"in-use",
timeout=2)
self.assertTrue(result)
# verify that the method will raise Exception when the status of
# volume is not in the expected status.
fake_volume_2 = self.fake_volume(status='fake-use')
nova_util.cinder.volumes.get.return_value = fake_volume_2
self.assertRaises(
Exception,
nova_util.wait_for_volume_status,
fake_volume_1,
"in-use",
timeout=2)
def test_check_nova_api_version(self, mock_glance, mock_cinder,
mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
# verify that the method will return True when the version of nova_api
# is supported.
api_versions.APIVersion = mock.MagicMock()
result = nova_util._check_nova_api_version(nova_util.nova, "2.56")
self.assertTrue(result)
# verify that the method will return False when the version of nova_api
# is not supported.
side_effect = nvexceptions.UnsupportedVersion()
api_versions.discover_version = mock.MagicMock(
side_effect=side_effect)
result = nova_util._check_nova_api_version(nova_util.nova, "2.56")
self.assertFalse(result)
@mock.patch.object(time, 'sleep', mock.Mock())
def test_wait_for_instance_status(self, mock_glance, mock_cinder,
mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
instance = self.fake_server(self.instance_uuid)
# verify that the method will return True when the status of instance
# is in the expected status.
result = nova_util.wait_for_instance_status(
instance,
('ACTIVE', 'ERROR'),
5,
10)
self.assertTrue(result)
# verify that the method will return False when the instance is None.
result = nova_util.wait_for_instance_status(
None,
('ACTIVE', 'ERROR'),
5,
10)
self.assertFalse(result)
# verify that the method will return False when the status of instance
# is not in the expected status.
self.fake_nova_find_list(nova_util, fake_find=instance, fake_list=None)
result = nova_util.wait_for_instance_status(
instance,
('ERROR'),
5,
10)
self.assertFalse(result)
@mock.patch.object(time, 'sleep', mock.Mock())
def test_confirm_resize(self, mock_glance, mock_cinder,
mock_neutron, mock_nova):
nova_util = nova_helper.NovaHelper()
instance = self.fake_server(self.instance_uuid)
self.fake_nova_find_list(nova_util, fake_find=instance, fake_list=None)
# verify that the method will return True when the status of instance
# is not in the expected status.
result = nova_util.confirm_resize(instance, instance.status)
self.assertTrue(result)
# verify that the method will return False when the status of instance
# is not in the expected status.
result = nova_util.confirm_resize(instance, "fake_status")
self.assertFalse(result)