x/python-ganttclient
Code Issues Proposed changes

Fix tests/baremetal/test_driver.py

Browse Source 
Unit tests for baremetal/driver.py were not functioning and were
falsely reporting success due to inheriting from the wrong class.
This lead to drift between the tests and the code over time.
I decided it was easier to re-implement tests/baremetal/test_driver.py
instead of trying to fix the drift. This gave me the chance to make
the test class easier to extend. Additional tests will be added
in the future, when I refactor driver.py:spawn().

This patch also has some minor variable name changes for baremetal/driver.py.

blueprint general-bare-metal-provisioning-framework.

Change-Id: I36ba5ed340b722b9a6e8d02d6d57d3c8d53b1eef
This commit is contained in:
Devananda van der Veen
2012-12-22 16:35:58 -08:00
parent 80eb567d28
commit 63c728771a
1 changed files with 127 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

269
nova/tests/baremetal/test_driver.py
View File

@@ -1,3 +1,7 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# coding=utf-8
# Copyright 2012 Hewlett-Packard Development Company, L.P.
# Copyright (c) 2012 NTT DOCOMO, INC. # Copyright (c) 2012 NTT DOCOMO, INC.
# Copyright (c) 2011 University of Southern California / ISI # Copyright (c) 2011 University of Southern California / ISI
# All Rights Reserved. # All Rights Reserved.
@@ -15,180 +19,161 @@
# under the License. # under the License.
""" """
Tests for baremetal driver. Tests for the base baremetal driver class.
""" """
from nova import exception import mox
from nova.openstack.common import cfg from nova.openstack.common import cfg
from nova.tests.baremetal.db import base
from nova.tests.baremetal.db import utils from nova import exception
from nova import test
from nova.tests.baremetal.db import base as bm_db_base
from nova.tests.baremetal.db import utils as bm_db_utils
from nova.tests.image import fake as fake_image from nova.tests.image import fake as fake_image
from nova.tests import test_virt_drivers from nova.tests import utils
from nova.tests import utils as test_utils
from nova.virt.baremetal import baremetal_states from nova.virt.baremetal import baremetal_states
from nova.virt.baremetal import db from nova.virt.baremetal import db
from nova.virt.baremetal import driver as bm_driver from nova.virt.baremetal import driver as bm_driver
from nova.virt.baremetal import volume_driver from nova.virt.baremetal import fake
from nova.virt.firewall import NoopFirewallDriver from nova.virt import firewall
CONF = cfg.CONF CONF = cfg.CONF
class FakeVifDriver(object):
def plug(self, instance, vif):
pass
def unplug(self, instance, vif):
pass
FakeFirewallDriver = NoopFirewallDriver
class FakeVolumeDriver(volume_driver.VolumeDriver):
def __init__(self, virtapi):
super(FakeVolumeDriver, self).__init__(virtapi)
self._initiator = "testtesttest"
NODE = utils.new_bm_node(cpus=2, memory_mb=4096, service_host="host1")
NICS = [
{'address': '01:23:45:67:89:01', 'datapath_id': '0x1', 'port_no': 1, },
{'address': '01:23:45:67:89:02', 'datapath_id': '0x2', 'port_no': 2, },
]
def class_path(class_):
return class_.__module__ + '.' + class_.__name__
COMMON_FLAGS = dict( COMMON_FLAGS = dict(
firewall_driver=class_path(FakeFirewallDriver), firewall_driver='nova.virt.baremetal.fake.FakeFirewallDriver',
host='test_host',
) )
BAREMETAL_FLAGS = dict( BAREMETAL_FLAGS = dict(
driver='nova.virt.baremetal.fake.Fake', driver='nova.virt.baremetal.fake.FakeDriver',
host=NODE['service_host'], instance_type_extra_specs=['cpu_arch:test', 'test_spec:test_value'],
instance_type_extra_specs=['cpu_arch:test'],
power_manager='nova.virt.baremetal.fake.FakePowerManager', power_manager='nova.virt.baremetal.fake.FakePowerManager',
sql_connection='sqlite:///:memory:', vif_driver='nova.virt.baremetal.fake.FakeVifDriver',
vif_driver=class_path(FakeVifDriver), volume_driver='nova.virt.baremetal.fake.FakeVolumeDriver',
volume_driver=class_path(FakeVolumeDriver),
group='baremetal', group='baremetal',
) )
def _create_baremetal_stuff(): class BareMetalDriverNoDBTestCase(test.TestCase):
context = test_utils.get_test_admin_context()
node = db.bm_node_create(context, NODE)
for nic in NICS:
db.bm_interface_create(context,
node['id'],
nic['address'],
nic['datapath_id'],
nic['port_no'])
return node
class BaremetalDriverSpawnTestCase(base.Database):
def setUp(self): def setUp(self):
super(BaremetalDriverSpawnTestCase, self).setUp() super(BareMetalDriverNoDBTestCase, self).setUp()
self.flags(**COMMON_FLAGS) self.flags(**COMMON_FLAGS)
self.flags(**BAREMETAL_FLAGS) self.flags(**BAREMETAL_FLAGS)
fake_image.stub_out_image_service(self.stubs)
self.node = _create_baremetal_stuff()
self.node_id = self.node['id']
self.context = test_utils.get_test_admin_context()
self.instance = test_utils.get_test_instance()
self.network_info = test_utils.get_test_network_info()
self.block_device_info = None
self.image_meta = test_utils.get_test_image_info(None, self.instance)
self.driver = bm_driver.BareMetalDriver(None) self.driver = bm_driver.BareMetalDriver(None)
self.kwargs = dict(
context=self.context,
instance=self.instance,
image_meta=self.image_meta,
injected_files=[('/foo', 'bar'), ('/abc', 'xyz')],
admin_password='testpass',
network_info=self.network_info,
block_device_info=self.block_device_info)
self.addCleanup(fake_image.FakeImageService_reset)
def test_ok(self): def test_validate_driver_loading(self):
self.instance['node'] = str(self.node_id) self.assertTrue(isinstance(self.driver.driver,
self.driver.spawn(**self.kwargs) fake.FakeDriver))
node = db.bm_node_get(self.context, self.node_id) self.assertTrue(isinstance(self.driver.vif_driver,
self.assertEqual(node['instance_uuid'], self.instance['uuid']) fake.FakeVifDriver))
self.assertEqual(node['task_state'], baremetal_states.ACTIVE) self.assertTrue(isinstance(self.driver.volume_driver,
fake.FakeVolumeDriver))
def test_without_node(self): self.assertTrue(isinstance(self.driver.firewall_driver,
self.assertRaises( fake.FakeFirewallDriver))
exception.NovaException,
self.driver.spawn,
**self.kwargs)
def test_node_not_found(self):
self.instance['node'] = "123456789"
self.assertRaises(
exception.InstanceNotFound,
self.driver.spawn,
**self.kwargs)
def test_node_in_use(self):
self.instance['node'] = str(self.node_id)
db.bm_node_update(self.context, self.node_id,
{'instance_uuid': 'something'})
self.assertRaises(
exception.NovaException,
self.driver.spawn,
**self.kwargs)
class BaremetalDriverTestCase(test_virt_drivers._VirtDriverTestCase, class BareMetalDriverWithDBTestCase(bm_db_base.BMDBTestCase):
base.Database):
def setUp(self): def setUp(self):
super(BaremetalDriverTestCase, self).setUp() super(BareMetalDriverWithDBTestCase, self).setUp()
self.driver_module = 'nova.virt.baremetal.BareMetalDriver'
self.flags(**COMMON_FLAGS) self.flags(**COMMON_FLAGS)
self.node = _create_baremetal_stuff() self.flags(**BAREMETAL_FLAGS)
self.node_id = self.node['id']
fake_image.stub_out_image_service(self.stubs) fake_image.stub_out_image_service(self.stubs)
self.context = utils.get_test_admin_context()
self.driver = bm_driver.BareMetalDriver(None)
self.node_info = bm_db_utils.new_bm_node(
id=123,
service_host='test_host',
cpus=2,
memory_mb=2048,
)
self.nic_info = [
{'address': '01:23:45:67:89:01', 'datapath_id': '0x1',
'port_no': 1},
{'address': '01:23:45:67:89:02', 'datapath_id': '0x2',
'port_no': 2},
]
self.addCleanup(fake_image.FakeImageService_reset) self.addCleanup(fake_image.FakeImageService_reset)
def _get_running_instance(self): def _create_node(self):
instance_ref = test_utils.get_test_instance() self.node = db.bm_node_create(self.context, self.node_info)
instance_ref['node'] = str(self.node_id) for nic in self.nic_info:
network_info = test_utils.get_test_network_info() db.bm_interface_create(
image_info = test_utils.get_test_image_info(None, instance_ref) self.context,
self.connection.spawn(self.ctxt, instance_ref, image_info, self.node['id'],
[], 'herp', network_info=network_info) nic['address'],
return instance_ref, network_info nic['datapath_id'],
nic['port_no'],
def test_loading_baremetal_drivers(self): )
from nova.virt.baremetal import fake self.test_instance = utils.get_test_instance()
drv = bm_driver.BareMetalDriver(None) self.test_instance['node'] = self.node['id']
self.assertTrue(isinstance(drv.baremetal_nodes, fake.Fake)) self.spawn_params = dict(
self.assertTrue(isinstance(drv._vif_driver, FakeVifDriver)) admin_password='test_pass',
self.assertTrue(isinstance(drv._firewall_driver, FakeFirewallDriver)) block_device_info=None,
self.assertTrue(isinstance(drv._volume_driver, FakeVolumeDriver)) context=self.context,
image_meta=utils.get_test_image_info(None,
self.test_instance),
injected_files=[('/fake/path', 'hello world')],
instance=self.test_instance,
network_info=utils.get_test_network_info(),
)
def test_get_host_stats(self): def test_get_host_stats(self):
self.flags(instance_type_extra_specs=['cpu_arch:x86_64', self._create_node()
'x:123', stats = self.driver.get_host_stats()
'y:456', ]) self.assertTrue(isinstance(stats, list))
drv = bm_driver.BareMetalDriver(None) self.assertEqual(len(stats), 1)
cap_list = drv.get_host_stats() stats = stats[0]
self.assertTrue(isinstance(cap_list, list)) self.assertEqual(stats['cpu_arch'], 'test')
self.assertEqual(len(cap_list), 1) self.assertEqual(stats['test_spec'], 'test_value')
cap = cap_list[0] self.assertEqual(stats['hypervisor_type'], 'baremetal')
self.assertEqual(cap['cpu_arch'], 'x86_64') self.assertEqual(stats['hypervisor_hostname'], '123')
self.assertEqual(cap['x'], '123') self.assertEqual(stats['host'], 'test_host')
self.assertEqual(cap['y'], '456') self.assertEqual(stats['vcpus'], 2)
self.assertEqual(cap['hypervisor_type'], 'baremetal') self.assertEqual(stats['host_memory_total'], 2048)
self.assertEqual(cap['driver'],
'nova.virt.baremetal.fake.Fake') def test_spawn_ok(self):
self._create_node()
self.driver.spawn(**self.spawn_params)
row = db.bm_node_get(self.context, self.node['id'])
self.assertEqual(row['task_state'], baremetal_states.ACTIVE)
def test_spawn_node_in_use(self):
self._create_node()
db.bm_node_update(self.context, self.node['id'],
{'instance_uuid': '1234-5678'})
self.assertRaises(exception.NovaException,
self.driver.spawn, **self.spawn_params)
row = db.bm_node_get(self.context, self.node['id'])
self.assertEqual(row['task_state'], None)
def test_spawn_node_not_found(self):
self._create_node()
db.bm_node_update(self.context, self.node['id'],
{'id': 9876})
self.assertRaises(exception.InstanceNotFound,
self.driver.spawn, **self.spawn_params)
row = db.bm_node_get(self.context, 9876)
self.assertEqual(row['task_state'], None)
def test_spawn_fails(self):
self._create_node()
self.mox.StubOutWithMock(fake.FakePowerManager, 'activate_node')
fake.FakePowerManager.activate_node().AndRaise(test.TestingException)
self.mox.ReplayAll()
self.assertRaises(test.TestingException,
self.driver.spawn, **self.spawn_params)
row = db.bm_node_get(self.context, self.node['id'])
self.assertEqual(row['task_state'], baremetal_states.ERROR)