x/gantt
Code Issues Proposed changes

Merge "baremetal: Move tests under tests/virt/baremetal"

Browse Source
This commit is contained in:
Jenkins 2013-05-22 04:04:06 +00:00 committed by Gerrit Code Review
commit 4fb306649f
19 changed files with 1 additions and 3133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
nova/tests/baremetal/__init__.py
View File

@ -1,15 +0,0 @@
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from nova.tests.baremetal import *

15
nova/tests/baremetal/db/__init__.py
View File

@ -1,15 +0,0 @@
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from nova.tests.baremetal.db import *

51
nova/tests/baremetal/db/base.py
View File

@ -1,51 +0,0 @@
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Bare-metal DB test base class."""
from oslo.config import cfg
from nova import context as nova_context
from nova import test
from nova.virt.baremetal.db import migration as bm_migration
from nova.virt.baremetal.db.sqlalchemy import session as bm_session
_DB_CACHE = None
CONF = cfg.CONF
CONF.import_opt('sql_connection',
'nova.virt.baremetal.db.sqlalchemy.session',
group='baremetal')
class Database(test.Database):
def post_migrations(self):
pass
class BMDBTestCase(test.TestCase):
def setUp(self):
super(BMDBTestCase, self).setUp()
self.flags(sql_connection='sqlite://', group='baremetal')
global _DB_CACHE
if not _DB_CACHE:
_DB_CACHE = Database(bm_session, bm_migration,
sql_connection=CONF.baremetal.sql_connection,
sqlite_db=None,
sqlite_clean_db=None)
self.useFixture(_DB_CACHE)
self.context = nova_context.get_admin_context()

55
nova/tests/baremetal/db/test_bm_interface.py
View File

@ -1,55 +0,0 @@
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Bare-metal DB testcase for BareMetalInterface
"""
from nova import exception
from nova.openstack.common.db import exception as db_exc
from nova.tests.baremetal.db import base
from nova.virt.baremetal import db
class BareMetalInterfaceTestCase(base.BMDBTestCase):
def test_unique_address(self):
pif1_id = db.bm_interface_create(self.context, 1, '11:11:11:11:11:11',
'0x1', 1)
self.assertRaises(db_exc.DBError,
db.bm_interface_create,
self.context, 2, '11:11:11:11:11:11', '0x2', 2)
# succeed after delete pif1
db.bm_interface_destroy(self.context, pif1_id)
pif2_id = db.bm_interface_create(self.context, 2, '11:11:11:11:11:11',
'0x2', 2)
self.assertTrue(pif2_id is not None)
def test_unique_vif_uuid(self):
pif1_id = db.bm_interface_create(self.context, 1, '11:11:11:11:11:11',
'0x1', 1)
pif2_id = db.bm_interface_create(self.context, 2, '22:22:22:22:22:22',
'0x2', 2)
db.bm_interface_set_vif_uuid(self.context, pif1_id, 'AAAA')
self.assertRaises(exception.NovaException,
db.bm_interface_set_vif_uuid,
self.context, pif2_id, 'AAAA')
def test_vif_not_found(self):
pif_id = db.bm_interface_create(self.context, 1, '11:11:11:11:11:11',
'0x1', 1)
self.assertRaises(exception.NovaException,
db.bm_interface_set_vif_uuid,
self.context, pif_id + 1, 'AAAA')

191
nova/tests/baremetal/db/test_bm_node.py
View File

@ -1,191 +0,0 @@
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Bare-Metal DB testcase for BareMetalNode
"""
from nova import exception
from nova.tests.baremetal.db import base
from nova.tests.baremetal.db import utils
from nova.virt.baremetal import db
class BareMetalNodesTestCase(base.BMDBTestCase):
def _create_nodes(self):
nodes = [
utils.new_bm_node(pm_address='0', service_host="host1",
memory_mb=100000, cpus=100, local_gb=10000),
utils.new_bm_node(pm_address='1', service_host="host2",
instance_uuid='A',
memory_mb=100000, cpus=100, local_gb=10000),
utils.new_bm_node(pm_address='2', service_host="host2",
memory_mb=1000, cpus=1, local_gb=1000),
utils.new_bm_node(pm_address='3', service_host="host2",
memory_mb=1000, cpus=2, local_gb=1000),
utils.new_bm_node(pm_address='4', service_host="host2",
memory_mb=2000, cpus=1, local_gb=1000),
utils.new_bm_node(pm_address='5', service_host="host2",
memory_mb=2000, cpus=2, local_gb=1000),
]
self.ids = []
for n in nodes:
ref = db.bm_node_create(self.context, n)
self.ids.append(ref['id'])
def test_get_all(self):
r = db.bm_node_get_all(self.context)
self.assertEquals(r, [])
self._create_nodes()
r = db.bm_node_get_all(self.context)
self.assertEquals(len(r), 6)
def test_get(self):
self._create_nodes()
r = db.bm_node_get(self.context, self.ids[0])
self.assertEquals(r['pm_address'], '0')
r = db.bm_node_get(self.context, self.ids[1])
self.assertEquals(r['pm_address'], '1')
self.assertRaises(
exception.NodeNotFound,
db.bm_node_get,
self.context, -1)
def test_get_by_service_host(self):
self._create_nodes()
r = db.bm_node_get_all(self.context, service_host=None)
self.assertEquals(len(r), 6)
r = db.bm_node_get_all(self.context, service_host="host1")
self.assertEquals(len(r), 1)
self.assertEquals(r[0]['pm_address'], '0')
r = db.bm_node_get_all(self.context, service_host="host2")
self.assertEquals(len(r), 5)
pmaddrs = [x['pm_address'] for x in r]
self.assertIn('1', pmaddrs)
self.assertIn('2', pmaddrs)
self.assertIn('3', pmaddrs)
self.assertIn('4', pmaddrs)
self.assertIn('5', pmaddrs)
r = db.bm_node_get_all(self.context, service_host="host3")
self.assertEquals(r, [])
def test_get_associated(self):
self._create_nodes()
r = db.bm_node_get_associated(self.context, service_host=None)
self.assertEquals(len(r), 1)
self.assertEquals(r[0]['pm_address'], '1')
r = db.bm_node_get_unassociated(self.context, service_host=None)
self.assertEquals(len(r), 5)
pmaddrs = [x['pm_address'] for x in r]
self.assertIn('0', pmaddrs)
self.assertIn('2', pmaddrs)
self.assertIn('3', pmaddrs)
self.assertIn('4', pmaddrs)
self.assertIn('5', pmaddrs)
def test_destroy(self):
self._create_nodes()
db.bm_node_destroy(self.context, self.ids[0])
self.assertRaises(
exception.NodeNotFound,
db.bm_node_get,
self.context, self.ids[0])
r = db.bm_node_get_all(self.context)
self.assertEquals(len(r), 5)
def test_destroy_with_interfaces(self):
self._create_nodes()
if_a_id = db.bm_interface_create(self.context, self.ids[0],
'aa:aa:aa:aa:aa:aa', None, None)
if_b_id = db.bm_interface_create(self.context, self.ids[0],
'bb:bb:bb:bb:bb:bb', None, None)
if_x_id = db.bm_interface_create(self.context, self.ids[1],
'11:22:33:44:55:66', None, None)
db.bm_node_destroy(self.context, self.ids[0])
self.assertRaises(
exception.NovaException,
db.bm_interface_get,
self.context, if_a_id)
self.assertRaises(
exception.NovaException,
db.bm_interface_get,
self.context, if_b_id)
# Another node's interface is not affected
if_x = db.bm_interface_get(self.context, if_x_id)
self.assertEqual(self.ids[1], if_x['bm_node_id'])
self.assertRaises(
exception.NodeNotFound,
db.bm_node_get,
self.context, self.ids[0])
r = db.bm_node_get_all(self.context)
self.assertEquals(len(r), 5)
def test_find_free(self):
self._create_nodes()
fn = db.bm_node_find_free(self.context, 'host2')
self.assertEqual(fn['pm_address'], '2')
fn = db.bm_node_find_free(self.context, 'host2',
memory_mb=500, cpus=2, local_gb=100)
self.assertEqual(fn['pm_address'], '3')
fn = db.bm_node_find_free(self.context, 'host2',
memory_mb=1001, cpus=1, local_gb=1000)
self.assertEqual(fn['pm_address'], '4')
fn = db.bm_node_find_free(self.context, 'host2',
memory_mb=2000, cpus=1, local_gb=1000)
self.assertEqual(fn['pm_address'], '4')
fn = db.bm_node_find_free(self.context, 'host2',
memory_mb=2000, cpus=2, local_gb=1000)
self.assertEqual(fn['pm_address'], '5')
# check memory_mb
fn = db.bm_node_find_free(self.context, 'host2',
memory_mb=2001, cpus=2, local_gb=1000)
self.assertTrue(fn is None)
# check cpus
fn = db.bm_node_find_free(self.context, 'host2',
memory_mb=2000, cpus=3, local_gb=1000)
self.assertTrue(fn is None)
# check local_gb
fn = db.bm_node_find_free(self.context, 'host2',
memory_mb=2000, cpus=2, local_gb=1001)
self.assertTrue(fn is None)

94
nova/tests/baremetal/db/test_bm_pxe_ip.py
View File

@ -1,94 +0,0 @@
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Bare-metal DB testcase for BareMetalPxeIp
"""
from nova import exception
from nova.openstack.common.db import exception as db_exc
from nova.tests.baremetal.db import base
from nova.tests.baremetal.db import utils
from nova.virt.baremetal import db
class BareMetalPxeIpTestCase(base.BMDBTestCase):
def _create_pxe_ip(self):
i1 = utils.new_bm_pxe_ip(address='10.1.1.1',
server_address='10.1.1.101')
i2 = utils.new_bm_pxe_ip(address='10.1.1.2',
server_address='10.1.1.102')
i1_ref = db.bm_pxe_ip_create_direct(self.context, i1)
self.assertTrue(i1_ref['id'] is not None)
self.assertEqual(i1_ref['address'], '10.1.1.1')
self.assertEqual(i1_ref['server_address'], '10.1.1.101')
i2_ref = db.bm_pxe_ip_create_direct(self.context, i2)
self.assertTrue(i2_ref['id'] is not None)
self.assertEqual(i2_ref['address'], '10.1.1.2')
self.assertEqual(i2_ref['server_address'], '10.1.1.102')
self.i1 = i1_ref
self.i2 = i2_ref
def test_unuque_address(self):
self._create_pxe_ip()
# address duplicates
i = utils.new_bm_pxe_ip(address='10.1.1.1',
server_address='10.1.1.201')
self.assertRaises(db_exc.DBError,
db.bm_pxe_ip_create_direct,
self.context, i)
# server_address duplicates
i = utils.new_bm_pxe_ip(address='10.1.1.3',
server_address='10.1.1.101')
self.assertRaises(db_exc.DBError,
db.bm_pxe_ip_create_direct,
self.context, i)
db.bm_pxe_ip_destroy(self.context, self.i1['id'])
i = utils.new_bm_pxe_ip(address='10.1.1.1',
server_address='10.1.1.101')
ref = db.bm_pxe_ip_create_direct(self.context, i)
self.assertTrue(ref is not None)
def test_bm_pxe_ip_associate(self):
self._create_pxe_ip()
node = db.bm_node_create(self.context, utils.new_bm_node())
ip_id = db.bm_pxe_ip_associate(self.context, node['id'])
ref = db.bm_pxe_ip_get(self.context, ip_id)
self.assertEqual(ref['bm_node_id'], node['id'])
def test_bm_pxe_ip_associate_raise(self):
self._create_pxe_ip()
node_id = 123
self.assertRaises(exception.NovaException,
db.bm_pxe_ip_associate,
self.context, node_id)
def test_delete_by_address(self):
self._create_pxe_ip()
db.bm_pxe_ip_destroy_by_address(self.context, '10.1.1.1')
del_ref = db.bm_pxe_ip_get(self.context, self.i1['id'])
self.assertTrue(del_ref is None)
def test_delete_by_address_not_exist(self):
self._create_pxe_ip()
del_ref = db.bm_pxe_ip_destroy_by_address(self.context, '10.11.12.13')
self.assertTrue(del_ref is None)

65
nova/tests/baremetal/db/utils.py
View File

@ -1,65 +0,0 @@
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Bare-metal test utils."""
from nova import test
from nova.virt.baremetal.db.sqlalchemy import models as bm_models
def new_bm_node(**kwargs):
h = bm_models.BareMetalNode()
h.id = kwargs.pop('id', None)
h.uuid = kwargs.pop('uuid', None)
h.service_host = kwargs.pop('service_host', None)
h.instance_uuid = kwargs.pop('instance_uuid', None)
h.cpus = kwargs.pop('cpus', 1)
h.memory_mb = kwargs.pop('memory_mb', 1024)
h.local_gb = kwargs.pop('local_gb', 64)
h.pm_address = kwargs.pop('pm_address', '192.168.1.1')
h.pm_user = kwargs.pop('pm_user', 'ipmi_user')
h.pm_password = kwargs.pop('pm_password', 'ipmi_password')
h.task_state = kwargs.pop('task_state', None)
h.terminal_port = kwargs.pop('terminal_port', 8000)
if len(kwargs) > 0:
raise test.TestingException("unknown field: %s"
% ','.join(kwargs.keys()))
return h
def new_bm_pxe_ip(**kwargs):
x = bm_models.BareMetalPxeIp()
x.id = kwargs.pop('id', None)
x.address = kwargs.pop('address', None)
x.server_address = kwargs.pop('server_address', None)
x.bm_node_id = kwargs.pop('bm_node_id', None)
if len(kwargs) > 0:
raise test.TestingException("unknown field: %s"
% ','.join(kwargs.keys()))
return x
def new_bm_interface(**kwargs):
x = bm_models.BareMetalInterface()
x.id = kwargs.pop('id', None)
x.bm_node_id = kwargs.pop('bm_node_id', None)
x.address = kwargs.pop('address', None)
x.datapath_id = kwargs.pop('datapath_id', None)
x.port_no = kwargs.pop('port_no', None)
x.vif_uuid = kwargs.pop('vif_uuid', None)
if len(kwargs) > 0:
raise test.TestingException("unknown field: %s"
% ','.join(kwargs.keys()))
return x

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

@ -1,363 +0,0 @@
# 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) 2011 University of Southern California / ISI
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for the base baremetal driver class."""
from oslo.config import cfg
from nova.compute import power_state
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 import utils
from nova.virt.baremetal import baremetal_states
from nova.virt.baremetal import db
from nova.virt.baremetal import driver as bm_driver
from nova.virt.baremetal import fake
CONF = cfg.CONF
COMMON_FLAGS = dict(
firewall_driver='nova.virt.baremetal.fake.FakeFirewallDriver',
host='test_host',
)
BAREMETAL_FLAGS = dict(
driver='nova.virt.baremetal.fake.FakeDriver',
instance_type_extra_specs=['cpu_arch:test', 'test_spec:test_value'],
power_manager='nova.virt.baremetal.fake.FakePowerManager',
vif_driver='nova.virt.baremetal.fake.FakeVifDriver',
volume_driver='nova.virt.baremetal.fake.FakeVolumeDriver',
group='baremetal',
)
class BareMetalDriverNoDBTestCase(test.TestCase):
def setUp(self):
super(BareMetalDriverNoDBTestCase, self).setUp()
self.flags(**COMMON_FLAGS)
self.flags(**BAREMETAL_FLAGS)
self.driver = bm_driver.BareMetalDriver(None)
def test_validate_driver_loading(self):
self.assertTrue(isinstance(self.driver.driver,
fake.FakeDriver))
self.assertTrue(isinstance(self.driver.vif_driver,
fake.FakeVifDriver))
self.assertTrue(isinstance(self.driver.volume_driver,
fake.FakeVolumeDriver))
self.assertTrue(isinstance(self.driver.firewall_driver,
fake.FakeFirewallDriver))
class BareMetalDriverWithDBTestCase(bm_db_base.BMDBTestCase):
def setUp(self):
super(BareMetalDriverWithDBTestCase, self).setUp()
self.flags(**COMMON_FLAGS)
self.flags(**BAREMETAL_FLAGS)
fake_image.stub_out_image_service(self.stubs)
self.context = utils.get_test_admin_context()
self.driver = bm_driver.BareMetalDriver(None)
self.addCleanup(fake_image.FakeImageService_reset)
def _create_node(self, node_info=None, nic_info=None):
result = {}
if node_info is None:
node_info = bm_db_utils.new_bm_node(
id=123,
service_host='test_host',
cpus=2,
memory_mb=2048,
)
if nic_info is None:
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},
]
result['node_info'] = node_info
result['nic_info'] = nic_info
result['node'] = db.bm_node_create(self.context, node_info)
for nic in nic_info:
db.bm_interface_create(
self.context,
result['node']['id'],
nic['address'],
nic['datapath_id'],
nic['port_no'],
)
result['instance'] = utils.get_test_instance()
result['instance']['node'] = result['node']['uuid']
result['spawn_params'] = dict(
admin_password='test_pass',
block_device_info=None,
context=self.context,
image_meta=utils.get_test_image_info(
None, result['instance']),
injected_files=[('/fake/path', 'hello world')],
instance=result['instance'],
network_info=utils.get_test_network_info(),
)
result['destroy_params'] = dict(
instance=result['instance'],
network_info=result['spawn_params']['network_info'],
block_device_info=result['spawn_params']['block_device_info'],
)
return result
def test_get_host_stats(self):
node = self._create_node()
stats = self.driver.get_host_stats()
self.assertTrue(isinstance(stats, list))
self.assertEqual(len(stats), 1)
stats = stats[0]
self.assertEqual(stats['cpu_arch'], 'test')
self.assertEqual(stats['test_spec'], 'test_value')
self.assertEqual(stats['hypervisor_type'], 'baremetal')
self.assertEqual(stats['hypervisor_hostname'], node['node']['uuid'])
self.assertEqual(stats['host'], 'test_host')
self.assertEqual(stats['vcpus'], 2)
self.assertEqual(stats['host_memory_total'], 2048)
def test_spawn_ok(self):
node = self._create_node()
self.driver.spawn(**node['spawn_params'])
row = db.bm_node_get(self.context, node['node']['id'])
self.assertEqual(row['task_state'], baremetal_states.ACTIVE)
self.assertEqual(row['instance_uuid'], node['instance']['uuid'])
self.assertEqual(row['instance_name'], node['instance']['hostname'])
def test_macs_from_nic_for_instance(self):
node = self._create_node()
expected = set([nic['address'] for nic in node['nic_info']])
self.assertEqual(
expected, self.driver.macs_for_instance(node['instance']))
def test_macs_for_instance_after_spawn(self):
node = self._create_node()
self.driver.spawn(**node['spawn_params'])
expected = set([nic['address'] for nic in node['nic_info']])
self.assertEqual(
expected, self.driver.macs_for_instance(node['instance']))
def test_macs_for_instance(self):
node = self._create_node()
expected = set(['01:23:45:67:89:01', '01:23:45:67:89:02'])
self.assertEqual(
expected, self.driver.macs_for_instance(node['instance']))
def test_macs_for_instance_no_interfaces(self):
# Nodes cannot boot with no MACs, so we raise an error if that happens.
node = self._create_node(nic_info=[])
self.assertRaises(exception.NovaException,
self.driver.macs_for_instance, node['instance'])
def test_spawn_node_already_associated(self):
node = self._create_node()
db.bm_node_update(self.context, node['node']['id'],
{'instance_uuid': '1234-5678'})
self.assertRaises(exception.NovaException,
self.driver.spawn, **node['spawn_params'])
row = db.bm_node_get(self.context, node['node']['id'])
self.assertEqual(row['task_state'], None)
def test_spawn_node_in_use(self):
node = self._create_node()
self.driver.spawn(**node['spawn_params'])
self.assertRaises(exception.NovaException,
self.driver.spawn, **node['spawn_params'])
def test_spawn_node_not_found(self):
node = self._create_node()
db.bm_node_update(self.context, node['node']['id'],
{'uuid': 'hide-this-node'})
self.assertRaises(exception.NovaException,
self.driver.spawn, **node['spawn_params'])
row = db.bm_node_get(self.context, node['node']['id'])
self.assertEqual(row['task_state'], None)
def test_spawn_fails(self):
node = 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, **node['spawn_params'])
row = db.bm_node_get(self.context, node['node']['id'])
self.assertEqual(row['task_state'], baremetal_states.DELETED)
def test_spawn_fails_to_cleanup(self):
node = self._create_node()
self.mox.StubOutWithMock(fake.FakePowerManager, 'activate_node')
self.mox.StubOutWithMock(fake.FakePowerManager, 'deactivate_node')
fake.FakePowerManager.activate_node().AndRaise(test.TestingException)
fake.FakePowerManager.deactivate_node().AndRaise(test.TestingException)
self.mox.ReplayAll()
self.assertRaises(test.TestingException,
self.driver.spawn, **node['spawn_params'])
row = db.bm_node_get(self.context, node['node']['id'])
self.assertEqual(row['task_state'], baremetal_states.ERROR)
def test_destroy_ok(self):
node = self._create_node()
self.driver.spawn(**node['spawn_params'])
self.driver.destroy(**node['destroy_params'])
row = db.bm_node_get(self.context, node['node']['id'])
self.assertEqual(row['task_state'], baremetal_states.DELETED)
self.assertEqual(row['instance_uuid'], None)
self.assertEqual(row['instance_name'], None)
def test_destroy_fails(self):
node = self._create_node()
self.mox.StubOutWithMock(fake.FakePowerManager, 'deactivate_node')
fake.FakePowerManager.deactivate_node().AndRaise(test.TestingException)
self.mox.ReplayAll()
self.driver.spawn(**node['spawn_params'])
self.assertRaises(test.TestingException,
self.driver.destroy, **node['destroy_params'])
row = db.bm_node_get(self.context, node['node']['id'])
self.assertEqual(row['task_state'], baremetal_states.ERROR)
self.assertEqual(row['instance_uuid'], node['instance']['uuid'])
def test_get_available_resources(self):
node = self._create_node()
resources = self.driver.get_available_resource(node['node']['uuid'])
self.assertEqual(resources['memory_mb'],
node['node_info']['memory_mb'])
self.assertEqual(resources['memory_mb_used'], 0)
self.driver.spawn(**node['spawn_params'])
resources = self.driver.get_available_resource(node['node']['uuid'])
self.assertEqual(resources['memory_mb_used'],
node['node_info']['memory_mb'])
self.driver.destroy(**node['destroy_params'])
resources = self.driver.get_available_resource(node['node']['uuid'])
self.assertEqual(resources['memory_mb_used'], 0)
def test_get_available_nodes(self):
self.assertEqual(0, len(self.driver.get_available_nodes()))
node1 = self._create_node()
self.assertEqual(1, len(self.driver.get_available_nodes()))
node1['instance']['hostname'] = 'test-host-1'
self.driver.spawn(**node1['spawn_params'])
self.assertEqual(1, len(self.driver.get_available_nodes()))
self.assertEqual([node1['node']['uuid']],
self.driver.get_available_nodes())
def test_list_instances(self):
self.assertEqual([], self.driver.list_instances())
node1 = self._create_node()
self.assertEqual([], self.driver.list_instances())
node_info = bm_db_utils.new_bm_node(
id=456,
service_host='test_host',
cpus=2,
memory_mb=2048,
)
nic_info = [
{'address': 'cc:cc:cc', 'datapath_id': '0x1',
'port_no': 1},
{'address': 'dd:dd:dd', 'datapath_id': '0x2',
'port_no': 2},
]
node2 = self._create_node(node_info=node_info, nic_info=nic_info)
self.assertEqual([], self.driver.list_instances())
node1['instance']['hostname'] = 'test-host-1'
node2['instance']['hostname'] = 'test-host-2'
self.driver.spawn(**node1['spawn_params'])
self.assertEqual(['test-host-1'],
self.driver.list_instances())
self.driver.spawn(**node2['spawn_params'])
self.assertEqual(['test-host-1', 'test-host-2'],
self.driver.list_instances())
self.driver.destroy(**node1['destroy_params'])
self.assertEqual(['test-host-2'],
self.driver.list_instances())
self.driver.destroy(**node2['destroy_params'])
self.assertEqual([], self.driver.list_instances())
def test_get_info_no_such_node(self):
node = self._create_node()
self.assertRaises(exception.InstanceNotFound,
self.driver.get_info,
node['instance'])
def test_get_info_ok(self):
node = self._create_node()
db.bm_node_associate_and_update(self.context, node['node']['uuid'],
{'instance_uuid': node['instance']['uuid'],
'instance_name': node['instance']['hostname'],
'task_state': baremetal_states.ACTIVE})
res = self.driver.get_info(node['instance'])
self.assertEqual(res['state'], power_state.RUNNING)
def test_get_info_with_defunct_pm(self):
# test fix for bug 1178378
node = self._create_node()
db.bm_node_associate_and_update(self.context, node['node']['uuid'],
{'instance_uuid': node['instance']['uuid'],
'instance_name': node['instance']['hostname'],
'task_state': baremetal_states.ACTIVE})
# fake the power manager and don't get a power state
self.mox.StubOutWithMock(fake.FakePowerManager, 'is_power_on')
fake.FakePowerManager.is_power_on().AndReturn(None)
self.mox.ReplayAll()
res = self.driver.get_info(node['instance'])
# prior to the fix, returned power_state was SHUTDOWN
self.assertEqual(res['state'], power_state.NOSTATE)
self.mox.VerifyAll()

234
nova/tests/baremetal/test_ipmi.py
View File

@ -1,234 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# coding=utf-8
# Copyright 2012 Hewlett-Packard Development Company, L.P.
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Test class for baremetal IPMI power manager."""
import os
import stat
import tempfile
from oslo.config import cfg
from nova import test
from nova.tests.baremetal.db import utils as bm_db_utils
from nova import utils
from nova.virt.baremetal import baremetal_states
from nova.virt.baremetal import ipmi
from nova.virt.baremetal import utils as bm_utils
CONF = cfg.CONF
class BareMetalIPMITestCase(test.TestCase):
def setUp(self):
super(BareMetalIPMITestCase, self).setUp()
self.node = bm_db_utils.new_bm_node(
id=123,
pm_address='fake-address',
pm_user='fake-user',
pm_password='fake-password')
self.ipmi = ipmi.IPMI(self.node)
def test_construct(self):
self.assertEqual(self.ipmi.node_id, 123)
self.assertEqual(self.ipmi.address, 'fake-address')
self.assertEqual(self.ipmi.user, 'fake-user')
self.assertEqual(self.ipmi.password, 'fake-password')
def test_make_password_file(self):
pw_file = ipmi._make_password_file(self.node['pm_password'])
try:
self.assertTrue(os.path.isfile(pw_file))
self.assertEqual(os.stat(pw_file)[stat.ST_MODE] & 0777, 0600)
with open(pw_file, "r") as f:
pm_password = f.read()
self.assertEqual(pm_password, self.node['pm_password'])
finally:
os.unlink(pw_file)
def test_exec_ipmitool(self):
pw_file = '/tmp/password_file'
self.mox.StubOutWithMock(ipmi, '_make_password_file')
self.mox.StubOutWithMock(utils, 'execute')
self.mox.StubOutWithMock(bm_utils, 'unlink_without_raise')
ipmi._make_password_file(self.ipmi.password).AndReturn(pw_file)
args = [
'ipmitool',
'-I', 'lanplus',
'-H', self.ipmi.address,
'-U', self.ipmi.user,
'-f', pw_file,
'A', 'B', 'C',
]
utils.execute(*args, attempts=3).AndReturn(('', ''))
bm_utils.unlink_without_raise(pw_file).AndReturn(None)
self.mox.ReplayAll()
self.ipmi._exec_ipmitool('A B C')
self.mox.VerifyAll()
def test_is_power_on_ok(self):
self.mox.StubOutWithMock(self.ipmi, '_exec_ipmitool')
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is on\n"])
self.mox.ReplayAll()
res = self.ipmi.is_power_on()
self.assertEqual(res, True)
self.mox.VerifyAll()
def test_is_power_no_answer(self):
self.mox.StubOutWithMock(self.ipmi, '_exec_ipmitool')
self.ipmi._exec_ipmitool("power status").AndReturn(
["Fake reply\n"])
self.mox.ReplayAll()
res = self.ipmi.is_power_on()
self.assertEqual(res, None)
self.mox.VerifyAll()
def test_power_already_on(self):
self.flags(ipmi_power_retry=0, group='baremetal')
self.mox.StubOutWithMock(self.ipmi, '_exec_ipmitool')
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is on\n"])
self.mox.ReplayAll()
self.ipmi.state = baremetal_states.DELETED
self.ipmi._power_on()
self.mox.VerifyAll()
self.assertEqual(self.ipmi.state, baremetal_states.ACTIVE)
def test_power_on_ok(self):
self.flags(ipmi_power_retry=0, group='baremetal')
self.mox.StubOutWithMock(self.ipmi, '_exec_ipmitool')
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is off\n"])
self.ipmi._exec_ipmitool("power on").AndReturn([])
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is on\n"])
self.mox.ReplayAll()
self.ipmi.state = baremetal_states.DELETED
self.ipmi._power_on()
self.mox.VerifyAll()
self.assertEqual(self.ipmi.state, baremetal_states.ACTIVE)
def test_power_on_fail(self):
self.flags(ipmi_power_retry=0, group='baremetal')
self.mox.StubOutWithMock(self.ipmi, '_exec_ipmitool')
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is off\n"])
self.ipmi._exec_ipmitool("power on").AndReturn([])
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is off\n"])
self.mox.ReplayAll()
self.ipmi.state = baremetal_states.DELETED
self.ipmi._power_on()
self.mox.VerifyAll()
self.assertEqual(self.ipmi.state, baremetal_states.ERROR)
def test_power_on_max_retries(self):
self.flags(ipmi_power_retry=2, group='baremetal')
self.mox.StubOutWithMock(self.ipmi, '_exec_ipmitool')
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is off\n"])
self.ipmi._exec_ipmitool("power on").AndReturn([])
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is off\n"])
self.ipmi._exec_ipmitool("power on").AndReturn([])
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is off\n"])
self.ipmi._exec_ipmitool("power on").AndReturn([])
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is off\n"])
self.mox.ReplayAll()
self.ipmi.state = baremetal_states.DELETED
self.ipmi._power_on()
self.mox.VerifyAll()
self.assertEqual(self.ipmi.state, baremetal_states.ERROR)
self.assertEqual(self.ipmi.retries, 3)
def test_power_off_ok(self):
self.flags(ipmi_power_retry=0, group='baremetal')
self.mox.StubOutWithMock(self.ipmi, '_exec_ipmitool')
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is on\n"])
self.ipmi._exec_ipmitool("power off").AndReturn([])
self.ipmi._exec_ipmitool("power status").AndReturn(
["Chassis Power is off\n"])
self.mox.ReplayAll()
self.ipmi.state = baremetal_states.ACTIVE
self.ipmi._power_off()
self.mox.VerifyAll()
self.assertEqual(self.ipmi.state, baremetal_states.DELETED)
def test_get_console_pid_path(self):
self.flags(terminal_pid_dir='/tmp', group='baremetal')
path = ipmi._get_console_pid_path(self.ipmi.node_id)
self.assertEqual(path, '/tmp/%s.pid' % self.ipmi.node_id)
def test_console_pid(self):
fd, path = tempfile.mkstemp()
with os.fdopen(fd, 'w') as f:
f.write("12345\n")
self.mox.StubOutWithMock(ipmi, '_get_console_pid_path')
ipmi._get_console_pid_path(self.ipmi.node_id).AndReturn(path)
self.mox.ReplayAll()
pid = ipmi._get_console_pid(self.ipmi.node_id)
bm_utils.unlink_without_raise(path)
self.mox.VerifyAll()
self.assertEqual(pid, 12345)
def test_console_pid_nan(self):
fd, path = tempfile.mkstemp()
with os.fdopen(fd, 'w') as f:
f.write("hello world\n")
self.mox.StubOutWithMock(ipmi, '_get_console_pid_path')
ipmi._get_console_pid_path(self.ipmi.node_id).AndReturn(path)
self.mox.ReplayAll()
pid = ipmi._get_console_pid(self.ipmi.node_id)
bm_utils.unlink_without_raise(path)
self.mox.VerifyAll()
self.assertTrue(pid is None)
def test_console_pid_file_not_found(self):
pid_path = ipmi._get_console_pid_path(self.ipmi.node_id)
self.mox.StubOutWithMock(os.path, 'exists')
os.path.exists(pid_path).AndReturn(False)
self.mox.ReplayAll()
pid = ipmi._get_console_pid(self.ipmi.node_id)
self.mox.VerifyAll()
self.assertTrue(pid is None)

266
nova/tests/baremetal/test_nova_baremetal_deploy_helper.py
View File

@ -1,266 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2012 NTT DOCOMO, INC.
# Copyright 2011 OpenStack Foundation
# Copyright 2011 Ilya Alekseyev
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import tempfile
import time
import mox
from nova.cmd import baremetal_deploy_helper as bmdh
from nova.openstack.common import log as logging
from nova import test
from nova.tests.baremetal.db import base as bm_db_base
from nova.virt.baremetal import db as bm_db
bmdh.LOG = logging.getLogger('nova.virt.baremetal.deploy_helper')
_PXECONF_DEPLOY = """
default deploy
label deploy
kernel deploy_kernel
append initrd=deploy_ramdisk
ipappend 3
label boot
kernel kernel
append initrd=ramdisk root=${ROOT}
"""
_PXECONF_BOOT = """
default boot
label deploy
kernel deploy_kernel
append initrd=deploy_ramdisk
ipappend 3
label boot
kernel kernel
append initrd=ramdisk root=UUID=12345678-1234-1234-1234-1234567890abcdef
"""
class WorkerTestCase(bm_db_base.BMDBTestCase):
def setUp(self):
super(WorkerTestCase, self).setUp()
self.worker = bmdh.Worker()
# Make tearDown() fast
self.worker.queue_timeout = 0.1
self.worker.start()
def tearDown(self):
if self.worker.isAlive():
self.worker.stop = True
self.worker.join(timeout=1)
super(WorkerTestCase, self).tearDown()
def wait_queue_empty(self, timeout):
for _ in xrange(int(timeout / 0.1)):
if bmdh.QUEUE.empty():
break
time.sleep(0.1)
def test_run_calls_deploy(self):
"""Check all queued requests are passed to deploy()."""
history = []
def fake_deploy(**params):
history.append(params)
self.stubs.Set(bmdh, 'deploy', fake_deploy)
self.mox.StubOutWithMock(bm_db, 'bm_node_update')
# update is called twice inside Worker.run
for i in range(6):
bm_db.bm_node_update(mox.IgnoreArg(), mox.IgnoreArg(),
mox.IgnoreArg())
self.mox.ReplayAll()
params_list = [{'fake1': ''}, {'fake2': ''}, {'fake3': ''}]
for (dep_id, params) in enumerate(params_list):
bmdh.QUEUE.put((dep_id, params))
self.wait_queue_empty(1)
self.assertEqual(params_list, history)
self.mox.VerifyAll()
def test_run_with_failing_deploy(self):
"""Check a worker keeps on running even if deploy() raises
an exception.
"""
history = []
def fake_deploy(**params):
history.append(params)
# always fail
raise Exception('test')
self.stubs.Set(bmdh, 'deploy', fake_deploy)
self.mox.StubOutWithMock(bm_db, 'bm_node_update')
# update is called twice inside Worker.run
for i in range(6):
bm_db.bm_node_update(mox.IgnoreArg(), mox.IgnoreArg(),
mox.IgnoreArg())
self.mox.ReplayAll()
params_list = [{'fake1': ''}, {'fake2': ''}, {'fake3': ''}]
for (dep_id, params) in enumerate(params_list):
bmdh.QUEUE.put((dep_id, params))
self.wait_queue_empty(1)
self.assertEqual(params_list, history)
self.mox.VerifyAll()
class PhysicalWorkTestCase(test.TestCase):
def setUp(self):
super(PhysicalWorkTestCase, self).setUp()
def noop(*args, **kwargs):
pass
self.stubs.Set(time, 'sleep', noop)
def test_deploy(self):
"""Check loosely all functions are called with right args."""
address = '127.0.0.1'
port = 3306
iqn = 'iqn.xyz'
lun = 1
image_path = '/tmp/xyz/image'
pxe_config_path = '/tmp/abc/pxeconfig'
root_mb = 128
swap_mb = 64
dev = '/dev/fake'
root_part = '/dev/fake-part1'
swap_part = '/dev/fake-part2'
root_uuid = '12345678-1234-1234-12345678-12345678abcdef'
self.mox.StubOutWithMock(bmdh, 'get_dev')
self.mox.StubOutWithMock(bmdh, 'get_image_mb')
self.mox.StubOutWithMock(bmdh, 'discovery')
self.mox.StubOutWithMock(bmdh, 'login_iscsi')
self.mox.StubOutWithMock(bmdh, 'logout_iscsi')
self.mox.StubOutWithMock(bmdh, 'make_partitions')
self.mox.StubOutWithMock(bmdh, 'is_block_device')
self.mox.StubOutWithMock(bmdh, 'dd')
self.mox.StubOutWithMock(bmdh, 'mkswap')
self.mox.StubOutWithMock(bmdh, 'block_uuid')
self.mox.StubOutWithMock(bmdh, 'switch_pxe_config')
self.mox.StubOutWithMock(bmdh, 'notify')
bmdh.get_dev(address, port, iqn, lun).AndReturn(dev)
bmdh.get_image_mb(image_path).AndReturn(1) # < root_mb
bmdh.discovery(address, port)
bmdh.login_iscsi(address, port, iqn)
bmdh.is_block_device(dev).AndReturn(True)
bmdh.make_partitions(dev, root_mb, swap_mb)
bmdh.is_block_device(root_part).AndReturn(True)
bmdh.is_block_device(swap_part).AndReturn(True)
bmdh.dd(image_path, root_part)
bmdh.mkswap(swap_part)
bmdh.block_uuid(root_part).AndReturn(root_uuid)
bmdh.logout_iscsi(address, port, iqn)
bmdh.switch_pxe_config(pxe_config_path, root_uuid)
bmdh.notify(address, 10000)
self.mox.ReplayAll()
bmdh.deploy(address, port, iqn, lun, image_path, pxe_config_path,
root_mb, swap_mb)
self.mox.VerifyAll()
def test_always_logout_iscsi(self):
"""logout_iscsi() must be called once login_iscsi() is called."""
address = '127.0.0.1'
port = 3306
iqn = 'iqn.xyz'
lun = 1
image_path = '/tmp/xyz/image'
pxe_config_path = '/tmp/abc/pxeconfig'
root_mb = 128
swap_mb = 64
dev = '/dev/fake'
self.mox.StubOutWithMock(bmdh, 'get_dev')
self.mox.StubOutWithMock(bmdh, 'get_image_mb')
self.mox.StubOutWithMock(bmdh, 'discovery')
self.mox.StubOutWithMock(bmdh, 'login_iscsi')
self.mox.StubOutWithMock(bmdh, 'logout_iscsi')
self.mox.StubOutWithMock(bmdh, 'work_on_disk')
class TestException(Exception):
pass
bmdh.get_dev(address, port, iqn, lun).AndReturn(dev)
bmdh.get_image_mb(image_path).AndReturn(1) # < root_mb
bmdh.discovery(address, port)
bmdh.login_iscsi(address, port, iqn)
bmdh.work_on_disk(dev, root_mb, swap_mb, image_path).\
AndRaise(TestException)
bmdh.logout_iscsi(address, port, iqn)
self.mox.ReplayAll()
self.assertRaises(TestException,
bmdh.deploy,
address, port, iqn, lun, image_path,
pxe_config_path, root_mb, swap_mb)
class SwitchPxeConfigTestCase(test.TestCase):
def setUp(self):
super(SwitchPxeConfigTestCase, self).setUp()
(fd, self.fname) = tempfile.mkstemp()
os.write(fd, _PXECONF_DEPLOY)
os.close(fd)
def tearDown(self):
os.unlink(self.fname)
super(SwitchPxeConfigTestCase, self).tearDown()
def test_switch_pxe_config(self):
bmdh.switch_pxe_config(self.fname,
'12345678-1234-1234-1234-1234567890abcdef')
with open(self.fname, 'r') as f:
pxeconf = f.read()
self.assertEqual(pxeconf, _PXECONF_BOOT)
class OtherFunctionTestCase(test.TestCase):
def test_get_dev(self):
expected = '/dev/disk/by-path/ip-1.2.3.4:5678-iscsi-iqn.fake-lun-9'
actual = bmdh.get_dev('1.2.3.4', 5678, 'iqn.fake', 9)
self.assertEqual(expected, actual)
def test_get_image_mb(self):
mb = 1024 * 1024
size = None
def fake_getsize(path):
return size
self.stubs.Set(os.path, 'getsize', fake_getsize)
size = 0
self.assertEqual(bmdh.get_image_mb('x'), 0)
size = 1
self.assertEqual(bmdh.get_image_mb('x'), 1)
size = mb
self.assertEqual(bmdh.get_image_mb('x'), 1)
size = mb + 1
self.assertEqual(bmdh.get_image_mb('x'), 2)

31
nova/tests/baremetal/test_nova_baremetal_manage.py
View File

@ -1,31 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2012 NTT DOCOMO, INC.
# Copyright 2011 OpenStack Foundation
# Copyright 2011 Ilya Alekseyev
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from nova.cmd import baremetal_manage as bm_man
from nova.tests.baremetal.db import base as bm_db_base
class BareMetalDbCommandsTestCase(bm_db_base.BMDBTestCase):
def setUp(self):
super(BareMetalDbCommandsTestCase, self).setUp()
self.commands = bm_man.BareMetalDbCommands()
def test_sync_and_version(self):
self.commands.sync()
v = self.commands.version()
self.assertTrue(v > 0)

584
nova/tests/baremetal/test_pxe.py
View File

@ -1,584 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# coding=utf-8
# Copyright 2012 Hewlett-Packard Development Company, L.P.
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for baremetal pxe driver."""
import os
import mox
from oslo.config import cfg
from testtools import matchers
from nova import exception
from nova.openstack.common.db import exception as db_exc
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 import utils
from nova.virt.baremetal import baremetal_states
from nova.virt.baremetal import db
from nova.virt.baremetal import pxe
from nova.virt.baremetal import utils as bm_utils
from nova.virt.disk import api as disk_api
from nova.virt import fake as fake_virt
CONF = cfg.CONF
COMMON_FLAGS = dict(
firewall_driver='nova.virt.baremetal.fake.FakeFirewallDriver',
host='test_host',
)
BAREMETAL_FLAGS = dict(
driver='nova.virt.baremetal.pxe.PXE',
instance_type_extra_specs=['cpu_arch:test', 'test_spec:test_value'],
power_manager='nova.virt.baremetal.fake.FakePowerManager',
vif_driver='nova.virt.baremetal.fake.FakeVifDriver',
volume_driver='nova.virt.baremetal.fake.FakeVolumeDriver',
group='baremetal',
)
class BareMetalPXETestCase(bm_db_base.BMDBTestCase):
def setUp(self):
super(BareMetalPXETestCase, self).setUp()
self.flags(**COMMON_FLAGS)
self.flags(**BAREMETAL_FLAGS)
self.driver = pxe.PXE(fake_virt.FakeVirtAPI())
fake_image.stub_out_image_service(self.stubs)
self.addCleanup(fake_image.FakeImageService_reset)
self.context = utils.get_test_admin_context()
self.test_block_device_info = None,
self.instance = utils.get_test_instance()
self.test_network_info = utils.get_test_network_info(),
self.node_info = bm_db_utils.new_bm_node(
service_host='test_host',
cpus=4,
memory_mb=2048,
)
self.nic_info = [
{'address': '22:22:22:22:22:22', 'datapath_id': '0x1',
'port_no': 1},
{'address': '33:33:33:33:33:33', 'datapath_id': '0x2',
'port_no': 2},
]
def _create_node(self):
self.node = db.bm_node_create(self.context, self.node_info)
for nic in self.nic_info:
db.bm_interface_create(
self.context,
self.node['id'],
nic['address'],
nic['datapath_id'],
nic['port_no'],
)
self.instance['node'] = self.node['id']
self.spawn_params = dict(
admin_password='test_pass',
block_device_info=self.test_block_device_info,
context=self.context,
image_meta=utils.get_test_image_info(None,
self.instance),
injected_files=[('/fake/path', 'hello world')],
instance=self.instance,
network_info=self.test_network_info,
)
class PXEClassMethodsTestCase(BareMetalPXETestCase):
def test_build_pxe_config(self):
args = {
'deployment_id': 'aaa',
'deployment_key': 'bbb',
'deployment_iscsi_iqn': 'ccc',
'deployment_aki_path': 'ddd',
'deployment_ari_path': 'eee',
'aki_path': 'fff',
'ari_path': 'ggg',
}
config = pxe.build_pxe_config(**args)
self.assertThat(config, matchers.StartsWith('default deploy'))
# deploy bits are in the deploy section
start = config.index('label deploy')
end = config.index('label boot')
self.assertThat(config[start:end], matchers.MatchesAll(
matchers.Contains('kernel ddd'),
matchers.Contains('initrd=eee'),
matchers.Contains('deployment_id=aaa'),
matchers.Contains('deployment_key=bbb'),
matchers.Contains('iscsi_target_iqn=ccc'),
matchers.Not(matchers.Contains('kernel fff')),
))
# boot bits are in the boot section
start = config.index('label boot')
self.assertThat(config[start:], matchers.MatchesAll(
matchers.Contains('kernel fff'),
matchers.Contains('initrd=ggg'),
matchers.Not(matchers.Contains('kernel ddd')),
))
def test_build_network_config(self):
net = utils.get_test_network_info(1)
config = pxe.build_network_config(net)
self.assertIn('eth0', config)
self.assertNotIn('eth1', config)
net = utils.get_test_network_info(2)
config = pxe.build_network_config(net)
self.assertIn('eth0', config)
self.assertIn('eth1', config)
def test_build_network_config_dhcp(self):
self.flags(
net_config_template='$pybasedir/nova/virt/baremetal/'
'net-dhcp.ubuntu.template',
group='baremetal',
)
net = utils.get_test_network_info()
net[0][1]['ips'][0]['ip'] = '1.2.3.4'
config = pxe.build_network_config(net)
self.assertIn('iface eth0 inet dhcp', config)
self.assertNotIn('address 1.2.3.4', config)
def test_build_network_config_static(self):
self.flags(
net_config_template='$pybasedir/nova/virt/baremetal/'
'net-static.ubuntu.template',
group='baremetal',
)
net = utils.get_test_network_info()
net[0][1]['ips'][0]['ip'] = '1.2.3.4'
config = pxe.build_network_config(net)
self.assertIn('iface eth0 inet static', config)
self.assertIn('address 1.2.3.4', config)
def test_image_dir_path(self):
self.assertEqual(
pxe.get_image_dir_path(self.instance),
os.path.join(CONF.instances_path, 'instance-00000001'))
def test_image_file_path(self):
self.assertEqual(
pxe.get_image_file_path(self.instance),
os.path.join(
CONF.instances_path, 'instance-00000001', 'disk'))
def test_pxe_config_file_path(self):
self.instance['uuid'] = 'aaaa-bbbb-cccc'
self.assertEqual(
pxe.get_pxe_config_file_path(self.instance),
os.path.join(CONF.baremetal.tftp_root,
'aaaa-bbbb-cccc', 'config'))
def test_pxe_mac_path(self):
self.assertEqual(
pxe.get_pxe_mac_path('23:45:67:89:AB'),
os.path.join(CONF.baremetal.tftp_root,
'pxelinux.cfg', '01-23-45-67-89-ab'))
def test_get_instance_deploy_ids(self):
self.instance['extra_specs'] = {
'baremetal:deploy_kernel_id': 'aaaa',
'baremetal:deploy_ramdisk_id': 'bbbb',
}
self.flags(deploy_kernel="fail", group='baremetal')
self.flags(deploy_ramdisk="fail", group='baremetal')
self.assertEqual(
pxe.get_deploy_aki_id(self.instance), 'aaaa')
self.assertEqual(
pxe.get_deploy_ari_id(self.instance), 'bbbb')
def test_get_default_deploy_ids(self):
self.instance['extra_specs'] = {}
self.flags(deploy_kernel="aaaa", group='baremetal')
self.flags(deploy_ramdisk="bbbb", group='baremetal')
self.assertEqual(
pxe.get_deploy_aki_id(self.instance), 'aaaa')
self.assertEqual(
pxe.get_deploy_ari_id(self.instance), 'bbbb')
def test_get_partition_sizes(self):
# default "kinda.big" instance
sizes = pxe.get_partition_sizes(self.instance)
self.assertEqual(sizes[0], 40960)
self.assertEqual(sizes[1], 1024)
def test_swap_not_zero(self):
# override swap to 0
instance_type = utils.get_test_instance_type(self.context)
instance_type['swap'] = 0
self.instance = utils.get_test_instance(self.context, instance_type)
sizes = pxe.get_partition_sizes(self.instance)
self.assertEqual(sizes[0], 40960)
self.assertEqual(sizes[1], 1)
def test_get_tftp_image_info(self):
instance_type = utils.get_test_instance_type()
# Raises an exception when options are neither specified
# on the instance nor in configuration file
CONF.baremetal.deploy_kernel = None
CONF.baremetal.deploy_ramdisk = None
self.assertRaises(exception.NovaException,
pxe.get_tftp_image_info,
self.instance, instance_type)
# Test that other non-true values also raise an exception
CONF.baremetal.deploy_kernel = ""
CONF.baremetal.deploy_ramdisk = ""
self.assertRaises(exception.NovaException,
pxe.get_tftp_image_info,
self.instance, instance_type)
# Even if the instance includes kernel_id and ramdisk_id,
# we still need deploy_kernel_id and deploy_ramdisk_id.
# If those aren't present in instance[], and not specified in
# config file, then we raise an exception.
self.instance['kernel_id'] = 'aaaa'
self.instance['ramdisk_id'] = 'bbbb'
self.assertRaises(exception.NovaException,
pxe.get_tftp_image_info,
self.instance, instance_type)
# If an instance doesn't specify deploy_kernel_id or deploy_ramdisk_id,
# but defaults are set in the config file, we should use those.
# Here, we confirm both that all four values were set
# and that the proper paths are getting set for all of them
CONF.baremetal.deploy_kernel = 'cccc'
CONF.baremetal.deploy_ramdisk = 'dddd'
base = os.path.join(CONF.baremetal.tftp_root, self.instance['uuid'])
res = pxe.get_tftp_image_info(self.instance, instance_type)
expected = {
'kernel': ['aaaa', os.path.join(base, 'kernel')],
'ramdisk': ['bbbb', os.path.join(base, 'ramdisk')],
'deploy_kernel': ['cccc', os.path.join(base, 'deploy_kernel')],
'deploy_ramdisk': ['dddd',
os.path.join(base, 'deploy_ramdisk')],
}
self.assertEqual(res, expected)
# If deploy_kernel_id and deploy_ramdisk_id are specified on
# image extra_specs, this should override any default configuration.
# Note that it is passed on the 'instance' object, despite being
# inherited from the instance_types_extra_specs table.
extra_specs = {
'baremetal:deploy_kernel_id': 'eeee',
'baremetal:deploy_ramdisk_id': 'ffff',
}
instance_type['extra_specs'] = extra_specs
res = pxe.get_tftp_image_info(self.instance, instance_type)
self.assertEqual(res['deploy_kernel'][0], 'eeee')
self.assertEqual(res['deploy_ramdisk'][0], 'ffff')
# However, if invalid values are passed on the image extra_specs,
# this should still raise an exception.
extra_specs = {
'baremetal:deploy_kernel_id': '',
'baremetal:deploy_ramdisk_id': '',
}
instance_type['extra_specs'] = extra_specs
self.assertRaises(exception.NovaException,
pxe.get_tftp_image_info,
self.instance, instance_type)
class PXEPrivateMethodsTestCase(BareMetalPXETestCase):
def test_collect_mac_addresses(self):
self._create_node()
address_list = [nic['address'] for nic in self.nic_info]
address_list.sort()
macs = self.driver._collect_mac_addresses(self.context, self.node)
self.assertEqual(macs, address_list)
def test_cache_tftp_images(self):
self.instance['kernel_id'] = 'aaaa'
self.instance['ramdisk_id'] = 'bbbb'
instance_type = utils.get_test_instance_type()
extra_specs = {
'baremetal:deploy_kernel_id': 'cccc',
'baremetal:deploy_ramdisk_id': 'dddd',
}
instance_type['extra_specs'] = extra_specs
image_info = pxe.get_tftp_image_info(self.instance, instance_type)
self.mox.StubOutWithMock(os, 'makedirs')
self.mox.StubOutWithMock(os.path, 'exists')
os.makedirs(os.path.join(CONF.baremetal.tftp_root,
self.instance['uuid'])).AndReturn(True)
for uuid, path in [image_info[label] for label in image_info]:
os.path.exists(path).AndReturn(True)
self.mox.ReplayAll()
self.driver._cache_tftp_images(
self.context, self.instance, image_info)
self.mox.VerifyAll()
def test_cache_image(self):
self.mox.StubOutWithMock(os, 'makedirs')
self.mox.StubOutWithMock(os.path, 'exists')
os.makedirs(pxe.get_image_dir_path(self.instance)).\
AndReturn(True)
os.path.exists(pxe.get_image_file_path(self.instance)).\
AndReturn(True)
self.mox.ReplayAll()
image_meta = utils.get_test_image_info(
self.context, self.instance)
self.driver._cache_image(
self.context, self.instance, image_meta)
self.mox.VerifyAll()
def test_inject_into_image(self):
# NOTE(deva): we could also test this method by stubbing
# nova.virt.disk.api._inject_*_into_fs
self._create_node()
files = []
self.instance['hostname'] = 'fake hostname'
files.append(('/etc/hostname', 'fake hostname'))
self.instance['key_data'] = 'fake ssh key'
net_info = utils.get_test_network_info(1)
net = pxe.build_network_config(net_info)
admin_password = 'fake password'
self.mox.StubOutWithMock(disk_api, 'inject_data')
disk_api.inject_data(
admin_password=admin_password,
image=pxe.get_image_file_path(self.instance),
key='fake ssh key',
metadata=None,
partition=None,
net=net,
files=files, # this is what we're really testing
).AndReturn(True)
self.mox.ReplayAll()
self.driver._inject_into_image(
self.context, self.node, self.instance,
network_info=net_info,
admin_password=admin_password,
injected_files=None)
self.mox.VerifyAll()
class PXEPublicMethodsTestCase(BareMetalPXETestCase):
def test_cache_images(self):
self._create_node()
self.mox.StubOutWithMock(self.driver.virtapi, 'instance_type_get')
self.mox.StubOutWithMock(pxe, "get_tftp_image_info")
self.mox.StubOutWithMock(self.driver, "_cache_tftp_images")
self.mox.StubOutWithMock(self.driver, "_cache_image")
self.mox.StubOutWithMock(self.driver, "_inject_into_image")
self.driver.virtapi.instance_type_get(
self.context, self.instance['instance_type_id']).AndReturn({})
pxe.get_tftp_image_info(self.instance, {}).AndReturn([])
self.driver._cache_tftp_images(self.context, self.instance, [])
self.driver._cache_image(self.context, self.instance, [])
self.driver._inject_into_image(self.context, self.node, self.instance,
self.test_network_info, None, '')
self.mox.ReplayAll()
self.driver.cache_images(
self.context, self.node, self.instance,
admin_password='',
image_meta=[],
injected_files=None,
network_info=self.test_network_info,
)
self.mox.VerifyAll()
def test_destroy_images(self):
self._create_node()
self.mox.StubOutWithMock(bm_utils, 'unlink_without_raise')
self.mox.StubOutWithMock(bm_utils, 'rmtree_without_raise')
bm_utils.unlink_without_raise(pxe.get_image_file_path(self.instance))
bm_utils.rmtree_without_raise(pxe.get_image_dir_path(self.instance))
self.mox.ReplayAll()
self.driver.destroy_images(self.context, self.node, self.instance)
self.mox.VerifyAll()
def test_activate_bootloader_passes_details(self):
self._create_node()
macs = [nic['address'] for nic in self.nic_info]
macs.sort()
image_info = {
'deploy_kernel': [None, 'aaaa'],
'deploy_ramdisk': [None, 'bbbb'],
'kernel': [None, 'cccc'],
'ramdisk': [None, 'dddd'],
}
self.instance['uuid'] = 'fake-uuid'
iqn = "iqn-%s" % self.instance['uuid']
pxe_config = 'this is a fake pxe config'
pxe_path = pxe.get_pxe_config_file_path(self.instance)
image_path = pxe.get_image_file_path(self.instance)
self.mox.StubOutWithMock(self.driver.virtapi, 'instance_type_get')
self.mox.StubOutWithMock(pxe, 'get_tftp_image_info')
self.mox.StubOutWithMock(pxe, 'get_partition_sizes')
self.mox.StubOutWithMock(bm_utils, 'random_alnum')
self.mox.StubOutWithMock(pxe, 'build_pxe_config')
self.mox.StubOutWithMock(bm_utils, 'write_to_file')
self.mox.StubOutWithMock(bm_utils, 'create_link_without_raise')
self.driver.virtapi.instance_type_get(
self.context, self.instance['instance_type_id']).AndReturn({})
pxe.get_tftp_image_info(self.instance, {}).AndReturn(image_info)
pxe.get_partition_sizes(self.instance).AndReturn((0, 0))
bm_utils.random_alnum(32).AndReturn('alnum')
pxe.build_pxe_config(
self.node['id'], 'alnum', iqn,
'aaaa', 'bbbb', 'cccc', 'dddd').AndReturn(pxe_config)
bm_utils.write_to_file(pxe_path, pxe_config)
for mac in macs:
bm_utils.create_link_without_raise(
pxe_path, pxe.get_pxe_mac_path(mac))
self.mox.ReplayAll()
self.driver.activate_bootloader(self.context, self.node, self.instance)
self.mox.VerifyAll()
def test_activate_and_deactivate_bootloader(self):
self._create_node()
instance_type = {
'extra_specs': {
'baremetal:deploy_kernel_id': 'eeee',
'baremetal:deploy_ramdisk_id': 'ffff',
}
}
self.instance['uuid'] = 'fake-uuid'
self.mox.StubOutWithMock(self.driver.virtapi, 'instance_type_get')
self.mox.StubOutWithMock(bm_utils, 'write_to_file')
self.mox.StubOutWithMock(bm_utils, 'create_link_without_raise')
self.mox.StubOutWithMock(bm_utils, 'unlink_without_raise')
self.mox.StubOutWithMock(bm_utils, 'rmtree_without_raise')
self.driver.virtapi.instance_type_get(
self.context, self.instance['instance_type_id']).AndReturn(
instance_type)
# create the config file
bm_utils.write_to_file(mox.StrContains('fake-uuid'),
mox.StrContains(CONF.baremetal.tftp_root))
# unlink and link the 2 interfaces
for i in range(2):
bm_utils.unlink_without_raise(mox.Or(
mox.StrContains('fake-uuid'),
mox.StrContains(CONF.baremetal.tftp_root)))
bm_utils.create_link_without_raise(
mox.StrContains('fake-uuid'),
mox.StrContains(CONF.baremetal.tftp_root))
# unlink all 2 interfaces, 4 images, and the config file
for i in range(7):
bm_utils.unlink_without_raise(mox.Or(
mox.StrContains('fake-uuid'),
mox.StrContains(CONF.baremetal.tftp_root)))
bm_utils.rmtree_without_raise(mox.StrContains('fake-uuid'))
self.mox.ReplayAll()
# activate and deactivate the bootloader
# and check the deployment task_state in the database
row = db.bm_node_get(self.context, 1)
self.assertTrue(row['deploy_key'] is None)
self.driver.activate_bootloader(self.context, self.node,
self.instance)
row = db.bm_node_get(self.context, 1)
self.assertTrue(row['deploy_key'] is not None)
self.driver.deactivate_bootloader(self.context, self.node,
self.instance)
row = db.bm_node_get(self.context, 1)
self.assertTrue(row['deploy_key'] is None)
self.mox.VerifyAll()
def test_deactivate_bootloader_for_nonexistent_instance(self):
self._create_node()
self.instance['uuid'] = 'fake-uuid'
pxe_path = pxe.get_pxe_config_file_path(self.instance)
self.mox.StubOutWithMock(bm_utils, 'unlink_without_raise')
self.mox.StubOutWithMock(bm_utils, 'rmtree_without_raise')
self.mox.StubOutWithMock(pxe, 'get_tftp_image_info')
self.mox.StubOutWithMock(self.driver, '_collect_mac_addresses')
extra_specs = dict(extra_specs={
'baremetal:deploy_ramdisk_id': 'ignore',
'baremetal:deploy_kernel_id': 'ignore'})
pxe.get_tftp_image_info(self.instance, extra_specs).\
AndRaise(exception.NovaException)
bm_utils.unlink_without_raise(pxe_path)
self.driver._collect_mac_addresses(self.context, self.node).\
AndRaise(db_exc.DBError)
bm_utils.rmtree_without_raise(
os.path.join(CONF.baremetal.tftp_root, 'fake-uuid'))
self.mox.ReplayAll()
self.driver.deactivate_bootloader(
self.context, self.node, self.instance)
self.mox.VerifyAll()
def test_activate_node(self):
self._create_node()
self.instance['uuid'] = 'fake-uuid'
self.flags(pxe_deploy_timeout=1, group='baremetal')
db.bm_node_update(self.context, 1,
{'task_state': baremetal_states.DEPLOYING,
'instance_uuid': 'fake-uuid'})
# test timeout
self.assertRaises(exception.InstanceDeployFailure,
self.driver.activate_node,
self.context, self.node, self.instance)
# test DEPLOYDONE
db.bm_node_update(self.context, 1,
{'task_state': baremetal_states.DEPLOYDONE})
self.driver.activate_node(self.context, self.node, self.instance)
# test no deploy -- state is just ACTIVE
db.bm_node_update(self.context, 1,
{'task_state': baremetal_states.ACTIVE})
self.driver.activate_node(self.context, self.node, self.instance)
# test node gone
db.bm_node_destroy(self.context, 1)
self.assertRaises(exception.InstanceDeployFailure,
self.driver.activate_node,
self.context, self.node, self.instance)

392
nova/tests/baremetal/test_tilera.py
View File

@ -1,392 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# coding=utf-8
# Copyright (c) 2011-2013 University of Southern California / ISI
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for baremetal tilera driver."""
import os
from oslo.config import cfg
from nova import exception
from nova.openstack.common.db import exception as db_exc
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 import utils
from nova.virt.baremetal import baremetal_states
from nova.virt.baremetal import db
from nova.virt.baremetal import tilera
from nova.virt.baremetal import utils as bm_utils
from nova.virt.disk import api as disk_api
from nova.virt import fake as fake_virt
CONF = cfg.CONF
COMMON_FLAGS = dict(
firewall_driver='nova.virt.baremetal.fake.FakeFirewallDriver',
host='test_host',
)
BAREMETAL_FLAGS = dict(
driver='nova.virt.baremetal.tilera.Tilera',
instance_type_extra_specs=['cpu_arch:test', 'test_spec:test_value'],
power_manager='nova.virt.baremetal.fake.FakePowerManager',
vif_driver='nova.virt.baremetal.fake.FakeVifDriver',
volume_driver='nova.virt.baremetal.fake.FakeVolumeDriver',
group='baremetal',
)
class BareMetalTileraTestCase(bm_db_base.BMDBTestCase):
def setUp(self):
super(BareMetalTileraTestCase, self).setUp()
self.flags(**COMMON_FLAGS)
self.flags(**BAREMETAL_FLAGS)
self.driver = tilera.Tilera(fake_virt.FakeVirtAPI())
fake_image.stub_out_image_service(self.stubs)
self.addCleanup(fake_image.FakeImageService_reset)
self.context = utils.get_test_admin_context()
self.test_block_device_info = None,
self.instance = utils.get_test_instance()
self.test_network_info = utils.get_test_network_info(),
self.node_info = bm_db_utils.new_bm_node(
service_host='test_host',
cpus=4,
memory_mb=2048,
)
self.nic_info = [
{'address': '22:22:22:22:22:22', 'datapath_id': '0x1',
'port_no': 1},
{'address': '33:33:33:33:33:33', 'datapath_id': '0x2',
'port_no': 2},
]
def _create_node(self):
self.node = db.bm_node_create(self.context, self.node_info)
for nic in self.nic_info:
db.bm_interface_create(
self.context,
self.node['id'],
nic['address'],
nic['datapath_id'],
nic['port_no'],
)
self.instance['node'] = self.node['id']
self.spawn_params = dict(
admin_password='test_pass',
block_device_info=self.test_block_device_info,
context=self.context,
image_meta=utils.get_test_image_info(None,
self.instance),
injected_files=[('/fake/path', 'hello world')],
instance=self.instance,
network_info=self.test_network_info,
)
class TileraClassMethodsTestCase(BareMetalTileraTestCase):
def test_build_network_config(self):
net = utils.get_test_network_info(1)
config = tilera.build_network_config(net)
self.assertIn('eth0', config)
self.assertNotIn('eth1', config)
net = utils.get_test_network_info(2)
config = tilera.build_network_config(net)
self.assertIn('eth0', config)
self.assertIn('eth1', config)
def test_build_network_config_dhcp(self):
self.flags(
net_config_template='$pybasedir/nova/virt/baremetal/'
'net-dhcp.ubuntu.template',
group='baremetal',
)
net = utils.get_test_network_info()
net[0][1]['ips'][0]['ip'] = '1.2.3.4'
config = tilera.build_network_config(net)
self.assertIn('iface eth0 inet dhcp', config)
self.assertNotIn('address 1.2.3.4', config)
def test_build_network_config_static(self):
self.flags(
net_config_template='$pybasedir/nova/virt/baremetal/'
'net-static.ubuntu.template',
group='baremetal',
)
net = utils.get_test_network_info()
net[0][1]['ips'][0]['ip'] = '1.2.3.4'
config = tilera.build_network_config(net)
self.assertIn('iface eth0 inet static', config)
self.assertIn('address 1.2.3.4', config)
def test_image_dir_path(self):
self.assertEqual(
tilera.get_image_dir_path(self.instance),
os.path.join(CONF.instances_path, 'instance-00000001'))
def test_image_file_path(self):
self.assertEqual(
tilera.get_image_file_path(self.instance),
os.path.join(
CONF.instances_path, 'instance-00000001', 'disk'))
def test_tilera_nfs_path(self):
self._create_node()
self.node['id'] = '123'
tilera_nfs_dir = "fs_" + self.node['id']
self.assertEqual(
tilera.get_tilera_nfs_path(self.node['id']),
os.path.join(CONF.baremetal.tftp_root,
tilera_nfs_dir))
def test_get_partition_sizes(self):
# default "kinda.big" instance
sizes = tilera.get_partition_sizes(self.instance)
self.assertEqual(sizes[0], 40960)
self.assertEqual(sizes[1], 1024)
def test_swap_not_zero(self):
# override swap to 0
instance_type = utils.get_test_instance_type(self.context)
instance_type['swap'] = 0
self.instance = utils.get_test_instance(self.context, instance_type)
sizes = tilera.get_partition_sizes(self.instance)
self.assertEqual(sizes[0], 40960)
self.assertEqual(sizes[1], 1)
def test_get_tftp_image_info(self):
# Tilera case needs only kernel_id.
self.instance['kernel_id'] = 'aaaa'
self.instance['uuid'] = 'fake-uuid'
# Here, we confirm both that kernel_id was set
# and that the proper paths are getting set for all of them
base = os.path.join(CONF.baremetal.tftp_root, self.instance['uuid'])
res = tilera.get_tftp_image_info(self.instance)
expected = {
'kernel': ['aaaa', os.path.join(base, 'kernel')],
}
self.assertEqual(res, expected)
class TileraPrivateMethodsTestCase(BareMetalTileraTestCase):
def test_collect_mac_addresses(self):
self._create_node()
address_list = [nic['address'] for nic in self.nic_info]
address_list.sort()
macs = self.driver._collect_mac_addresses(self.context, self.node)
self.assertEqual(macs, address_list)
def test_cache_tftp_images(self):
self.instance['kernel_id'] = 'aaaa'
image_info = tilera.get_tftp_image_info(self.instance)
self.mox.StubOutWithMock(os, 'makedirs')
self.mox.StubOutWithMock(os.path, 'exists')
os.makedirs(os.path.join(CONF.baremetal.tftp_root,
self.instance['uuid'])).AndReturn(True)
for uuid, path in [image_info[label] for label in image_info]:
os.path.exists(path).AndReturn(True)
self.mox.ReplayAll()
self.driver._cache_tftp_images(
self.context, self.instance, image_info)
self.mox.VerifyAll()
def test_cache_image(self):
self.mox.StubOutWithMock(os, 'makedirs')
self.mox.StubOutWithMock(os.path, 'exists')
os.makedirs(tilera.get_image_dir_path(self.instance)).\
AndReturn(True)
os.path.exists(tilera.get_image_file_path(self.instance)).\
AndReturn(True)
self.mox.ReplayAll()
image_meta = utils.get_test_image_info(
self.context, self.instance)
self.driver._cache_image(
self.context, self.instance, image_meta)
self.mox.VerifyAll()
def test_inject_into_image(self):
self._create_node()
files = []
self.instance['hostname'] = 'fake hostname'
files.append(('/etc/hostname', 'fake hostname'))
self.instance['key_data'] = 'fake ssh key'
net_info = utils.get_test_network_info(1)
net = tilera.build_network_config(net_info)
admin_password = 'fake password'
self.mox.StubOutWithMock(disk_api, 'inject_data')
disk_api.inject_data(
admin_password=admin_password,
image=tilera.get_image_file_path(self.instance),
key='fake ssh key',
metadata=None,
partition=None,
net=net,
files=files,
).AndReturn(True)
self.mox.ReplayAll()
self.driver._inject_into_image(
self.context, self.node, self.instance,
network_info=net_info,
admin_password=admin_password,
injected_files=None)
self.mox.VerifyAll()
class TileraPublicMethodsTestCase(BareMetalTileraTestCase):
def test_cache_images(self):
self._create_node()
self.mox.StubOutWithMock(tilera, "get_tftp_image_info")
self.mox.StubOutWithMock(self.driver, "_cache_tftp_images")
self.mox.StubOutWithMock(self.driver, "_cache_image")
self.mox.StubOutWithMock(self.driver, "_inject_into_image")
tilera.get_tftp_image_info(self.instance).AndReturn([])
self.driver._cache_tftp_images(self.context, self.instance, [])
self.driver._cache_image(self.context, self.instance, [])
self.driver._inject_into_image(self.context, self.node, self.instance,
self.test_network_info, None, '')
self.mox.ReplayAll()
self.driver.cache_images(
self.context, self.node, self.instance,
admin_password='',
image_meta=[],
injected_files=None,
network_info=self.test_network_info,
)
self.mox.VerifyAll()
def test_destroy_images(self):
self._create_node()
self.mox.StubOutWithMock(bm_utils, 'unlink_without_raise')
self.mox.StubOutWithMock(bm_utils, 'rmtree_without_raise')
bm_utils.unlink_without_raise(tilera.get_image_file_path(
self.instance))
bm_utils.rmtree_without_raise(tilera.get_image_dir_path(self.instance))
self.mox.ReplayAll()
self.driver.destroy_images(self.context, self.node, self.instance)
self.mox.VerifyAll()
def test_activate_bootloader_passes_details(self):
self._create_node()
image_info = {
'kernel': [None, 'cccc'],
}
self.instance['uuid'] = 'fake-uuid'
iqn = "iqn-%s" % self.instance['uuid']
tilera_config = 'this is a fake tilera config'
self.instance['uuid'] = 'fake-uuid'
tilera_path = tilera.get_tilera_nfs_path(self.instance)
image_path = tilera.get_image_file_path(self.instance)
self.mox.StubOutWithMock(tilera, 'get_tftp_image_info')
self.mox.StubOutWithMock(tilera, 'get_partition_sizes')
tilera.get_tftp_image_info(self.instance).AndReturn(image_info)
tilera.get_partition_sizes(self.instance).AndReturn((0, 0))
self.mox.ReplayAll()
self.driver.activate_bootloader(self.context, self.node, self.instance)
self.mox.VerifyAll()
def test_activate_and_deactivate_bootloader(self):
self._create_node()
self.instance['uuid'] = 'fake-uuid'
tilera_path = tilera.get_tilera_nfs_path(self.instance)
image_path = tilera.get_image_file_path(self.instance)
self.mox.ReplayAll()
# activate and deactivate the bootloader
# and check the deployment task_state in the database
row = db.bm_node_get(self.context, 1)
self.assertTrue(row['deploy_key'] is None)
self.driver.activate_bootloader(self.context, self.node,
self.instance)
row = db.bm_node_get(self.context, 1)
self.assertTrue(row['deploy_key'] is not None)
self.driver.deactivate_bootloader(self.context, self.node,
self.instance)
row = db.bm_node_get(self.context, 1)
self.assertTrue(row['deploy_key'] is None)
self.mox.VerifyAll()
def test_deactivate_bootloader_for_nonexistent_instance(self):
self._create_node()
self.node['id'] = 'fake-node-id'
self.mox.StubOutWithMock(bm_utils, 'unlink_without_raise')
self.mox.StubOutWithMock(bm_utils, 'rmtree_without_raise')
self.mox.StubOutWithMock(tilera, 'get_tftp_image_info')
self.mox.StubOutWithMock(self.driver, '_collect_mac_addresses')
tilera_path = tilera.get_tilera_nfs_path(self.node['id'])
tilera.get_tftp_image_info(self.instance).\
AndRaise(exception.NovaException)
self.driver._collect_mac_addresses(self.context, self.node).\
AndRaise(db_exc.DBError)
self.mox.ReplayAll()
self.driver.deactivate_bootloader(
self.context, self.node, self.instance)
self.mox.VerifyAll()
def test_activate_node(self):
self._create_node()
self.instance['uuid'] = 'fake-uuid'
db.bm_node_update(self.context, 1,
{'task_state': baremetal_states.DEPLOYING,
'instance_uuid': 'fake-uuid'})
# test DEPLOYDONE
db.bm_node_update(self.context, 1,
{'task_state': baremetal_states.DEPLOYDONE})
self.driver.activate_node(self.context, self.node, self.instance)
# test no deploy -- state is just ACTIVE
db.bm_node_update(self.context, 1,
{'task_state': baremetal_states.ACTIVE})
self.driver.activate_node(self.context, self.node, self.instance)
# test node gone
db.bm_node_destroy(self.context, 1)
self.assertRaises(exception.InstanceDeployFailure,
self.driver.activate_node,
self.context, self.node, self.instance)

141
nova/tests/baremetal/test_tilera_pdu.py
View File

@ -1,141 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# coding=utf-8
# Copyright (c) 2011-2013 University of Southern California / ISI
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Test class for baremetal PDU power manager."""
from oslo.config import cfg
from nova import test
from nova.tests.baremetal.db import utils as bm_db_utils
from nova import utils
from nova.virt.baremetal import baremetal_states
from nova.virt.baremetal import tilera_pdu
from nova.virt.baremetal import utils as bm_utils
CONF = cfg.CONF
class BareMetalPduTestCase(test.TestCase):
def setUp(self):
super(BareMetalPduTestCase, self).setUp()
self.node = bm_db_utils.new_bm_node(
id=123,
pm_address='fake-address',
pm_user='fake-user',
pm_password='fake-password')
self.tilera_pdu = tilera_pdu.Pdu(self.node)
self.tile_pdu_on = 1
self.tile_pdu_off = 2
self.tile_pdu_status = 9
def test_construct(self):
self.assertEqual(self.tilera_pdu.node_id, 123)
self.assertEqual(self.tilera_pdu.address, 'fake-address')
self.assertEqual(self.tilera_pdu.user, 'fake-user')
self.assertEqual(self.tilera_pdu.password, 'fake-password')
def test_exec_pdutool(self):
self.flags(tile_pdu_mgr='fake-pdu-mgr', group='baremetal')
self.flags(tile_pdu_ip='fake-address', group='baremetal')
self.mox.StubOutWithMock(utils, 'execute')
self.mox.StubOutWithMock(bm_utils, 'unlink_without_raise')
args = [
'fake-pdu-mgr',
'fake-address',
self.tile_pdu_on,
]
utils.execute(*args).AndReturn('')
self.mox.ReplayAll()
self.tilera_pdu._exec_pdutool(self.tile_pdu_on)
self.mox.VerifyAll()
def test_is_power(self):
self.mox.StubOutWithMock(self.tilera_pdu, '_exec_pdutool')
self.tilera_pdu._exec_pdutool(self.tile_pdu_status).AndReturn(
self.tile_pdu_on)
self.mox.ReplayAll()
self.tilera_pdu._is_power(self.tile_pdu_on)
self.mox.VerifyAll()
def test_power_already_on(self):
self.mox.StubOutWithMock(self.tilera_pdu, '_exec_pdutool')
self.tilera_pdu._exec_pdutool(self.tile_pdu_on).AndReturn(None)
self.tilera_pdu._exec_pdutool(self.tile_pdu_status).AndReturn(
self.tile_pdu_on)
self.mox.ReplayAll()
self.tilera_pdu.state = baremetal_states.DELETED
self.tilera_pdu._power_on()
self.mox.VerifyAll()
self.assertEqual(self.tilera_pdu.state, baremetal_states.ACTIVE)
def test_power_on_ok(self):
self.mox.StubOutWithMock(self.tilera_pdu, '_exec_pdutool')
self.tilera_pdu._exec_pdutool(self.tile_pdu_on).AndReturn(None)
self.tilera_pdu._exec_pdutool(self.tile_pdu_status).AndReturn(
self.tile_pdu_on)
self.mox.ReplayAll()
self.tilera_pdu.state = baremetal_states.DELETED
self.tilera_pdu._power_on()
self.mox.VerifyAll()
self.assertEqual(self.tilera_pdu.state, baremetal_states.ACTIVE)
def test_power_on_fail(self):
self.mox.StubOutWithMock(self.tilera_pdu, '_exec_pdutool')
self.tilera_pdu._exec_pdutool(self.tile_pdu_on).AndReturn(None)
self.tilera_pdu._exec_pdutool(self.tile_pdu_status).AndReturn(
self.tile_pdu_off)
self.mox.ReplayAll()
self.tilera_pdu.state = baremetal_states.DELETED
self.tilera_pdu._power_on()
self.mox.VerifyAll()
self.assertEqual(self.tilera_pdu.state, baremetal_states.ERROR)
def test_power_on_max_retries(self):
self.mox.StubOutWithMock(self.tilera_pdu, '_exec_pdutool')
self.tilera_pdu._exec_pdutool(self.tile_pdu_on).AndReturn(None)
self.tilera_pdu._exec_pdutool(self.tile_pdu_status).AndReturn(
self.tile_pdu_off)
self.mox.ReplayAll()
self.tilera_pdu.state = baremetal_states.DELETED
self.tilera_pdu._power_on()
self.mox.VerifyAll()
self.assertEqual(self.tilera_pdu.state, baremetal_states.ERROR)
def test_power_off_ok(self):
self.mox.StubOutWithMock(self.tilera_pdu, '_exec_pdutool')
self.tilera_pdu._exec_pdutool(self.tile_pdu_off).AndReturn(None)
self.tilera_pdu._exec_pdutool(self.tile_pdu_status).AndReturn(
self.tile_pdu_off)
self.mox.ReplayAll()
self.tilera_pdu.state = baremetal_states.ACTIVE
self.tilera_pdu._power_off()
self.mox.VerifyAll()
self.assertEqual(self.tilera_pdu.state, baremetal_states.DELETED)

67
nova/tests/baremetal/test_utils.py
View File

@ -1,67 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# coding=utf-8
# Copyright 2012 Hewlett-Packard Development Company, L.P.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for baremetal utils."""
import errno
import os
from nova import test
from nova.virt.baremetal import utils
class BareMetalUtilsTestCase(test.TestCase):
def test_random_alnum(self):
s = utils.random_alnum(10)
self.assertEqual(len(s), 10)
s = utils.random_alnum(100)
self.assertEqual(len(s), 100)
def test_unlink(self):
self.mox.StubOutWithMock(os, "unlink")
os.unlink("/fake/path")
self.mox.ReplayAll()
utils.unlink_without_raise("/fake/path")
self.mox.VerifyAll()
def test_unlink_ENOENT(self):
self.mox.StubOutWithMock(os, "unlink")
os.unlink("/fake/path").AndRaise(OSError(errno.ENOENT))
self.mox.ReplayAll()
utils.unlink_without_raise("/fake/path")
self.mox.VerifyAll()
def test_create_link(self):
self.mox.StubOutWithMock(os, "symlink")
os.symlink("/fake/source", "/fake/link")
self.mox.ReplayAll()
utils.create_link_without_raise("/fake/source", "/fake/link")
self.mox.VerifyAll()
def test_create_link_EEXIST(self):
self.mox.StubOutWithMock(os, "symlink")
os.symlink("/fake/source", "/fake/link").AndRaise(
OSError(errno.EEXIST))
self.mox.ReplayAll()
utils.create_link_without_raise("/fake/source", "/fake/link")
self.mox.VerifyAll()

398
nova/tests/baremetal/test_virtual_power_driver.py
View File

@ -1,398 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# coding=utf-8
# Copyright 2012 Hewlett-Packard Development Company, L.P.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for baremetal virtual power driver."""
import mox
from oslo.config import cfg
from nova import exception
from nova.openstack.common import processutils
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 import utils
from nova.virt.baremetal import db
from nova.virt.baremetal import virtual_power_driver
import nova.virt.powervm.common as connection
CONF = cfg.CONF
COMMON_FLAGS = dict(
firewall_driver='nova.virt.baremetal.fake.FakeFirewallDriver',
host='test_host',
)
BAREMETAL_FLAGS = dict(
driver='nova.virt.baremetal.pxe.PXE',
instance_type_extra_specs=['cpu_arch:test', 'test_spec:test_value'],
power_manager=
'nova.virt.baremetal.virtual_power_driver.VirtualPowerManager',
vif_driver='nova.virt.baremetal.fake.FakeVifDriver',
volume_driver='nova.virt.baremetal.fake.FakeVolumeDriver',
virtual_power_ssh_host=None,
virtual_power_type='vbox',
virtual_power_host_user=None,
virtual_power_host_pass=None,
virtual_power_host_key=None,
group='baremetal',
)
class BareMetalVPDTestCase(bm_db_base.BMDBTestCase):
def setUp(self):
super(BareMetalVPDTestCase, self).setUp()
self.flags(**COMMON_FLAGS)
self.flags(**BAREMETAL_FLAGS)
fake_image.stub_out_image_service(self.stubs)
self.context = utils.get_test_admin_context()
self.test_block_device_info = None,
self.instance = utils.get_test_instance()
self.test_network_info = utils.get_test_network_info(),
self.node_info = bm_db_utils.new_bm_node(
id=123,
service_host='test_host',
cpus=2,
memory_mb=2048,
)
self.nic_info = [
{'address': '11:11:11:11:11:11', 'datapath_id': '0x1',
'port_no': 1},
{'address': '22:22:22:22:22:22', 'datapath_id': '0x2',
'port_no': 2},
]
self.addCleanup(fake_image.FakeImageService_reset)
def _create_node(self):
self.node = db.bm_node_create(self.context, self.node_info)
for nic in self.nic_info:
db.bm_interface_create(
self.context,
self.node['id'],
nic['address'],
nic['datapath_id'],
nic['port_no'],
)
self.instance['node'] = self.node['id']
def _create_pm(self):
self.pm = virtual_power_driver.VirtualPowerManager(
node=self.node,
instance=self.instance)
return self.pm
class VPDMissingOptionsTestCase(BareMetalVPDTestCase):
def test_get_conn_missing_options(self):
self.flags(virtual_power_ssh_host=None, group="baremetal")
self.flags(virtual_power_host_user=None, group="baremetal")
self.flags(virtual_power_host_pass=None, group="baremetal")
self._create_node()
self._create_pm()
self._conn = None
self.assertRaises(exception.NovaException,
self.pm._get_conn)
self._conn = None
self.flags(virtual_power_ssh_host='127.0.0.1', group="baremetal")
self.assertRaises(exception.NovaException,
self.pm._get_conn)
self._conn = None
self.flags(virtual_power_host_user='user', group="baremetal")
self.assertRaises(exception.NovaException,
self.pm._get_conn)
class VPDClassMethodsTestCase(BareMetalVPDTestCase):
def setUp(self):
super(VPDClassMethodsTestCase, self).setUp()
self.flags(virtual_power_ssh_host='127.0.0.1', group="baremetal")
self.flags(virtual_power_host_user='user', group="baremetal")
self.flags(virtual_power_host_pass='password', group="baremetal")
def test_get_conn_success_pass(self):
self._create_node()
self._create_pm()
self._conn = self.pm._get_conn()
self.mox.StubOutWithMock(connection, 'ssh_connect')
connection.ssh_connect(mox.IsA(self._conn)).AndReturn(True)
self.mox.ReplayAll()
self.pm._set_connection()
self.assertEqual(self.pm.connection_data.host, '127.0.0.1')
self.assertEqual(self.pm.connection_data.username, 'user')
self.assertEqual(self.pm.connection_data.password, 'password')
self.assertEqual(self.pm.connection_data.keyfile, None)
self.mox.VerifyAll()
def test_get_conn_success_key(self):
self.flags(virtual_power_host_pass='', group="baremetal")
self.flags(virtual_power_host_key='/id_rsa_file.txt',
group="baremetal")
self._create_node()
self._create_pm()
self._conn = self.pm._get_conn()
self.mox.StubOutWithMock(connection, 'ssh_connect')
connection.ssh_connect(mox.IsA(self._conn)).AndReturn(True)
self.mox.ReplayAll()
self.pm._set_connection()
self.assertEqual(self.pm.connection_data.host, '127.0.0.1')
self.assertEqual(self.pm.connection_data.username, 'user')
self.assertEqual(self.pm.connection_data.password, '')
self.assertEqual(self.pm.connection_data.keyfile, '/id_rsa_file.txt')
self.mox.VerifyAll()
def test_get_full_node_list(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_run_command')
cmd = self.pm._vp_cmd.list_cmd
self.pm._run_command(cmd).AndReturn("testNode")
self.mox.ReplayAll()
name = self.pm._get_full_node_list()
self.assertEqual(name, 'testNode')
self.mox.VerifyAll()
def test_check_for_node(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_get_full_node_list')
self.pm._get_full_node_list().\
AndReturn(["testNode"])
self.mox.StubOutWithMock(self.pm, '_run_command')
cmd = self.pm._vp_cmd.get_node_macs.replace('{_NodeName_}', 'testNode')
self.pm._run_command(cmd).\
AndReturn(["111111111111", "ffeeddccbbaa"])
self.mox.ReplayAll()
name = self.pm._check_for_node()
self.assertEqual(name, '"testNode"')
self.mox.VerifyAll()
def test_check_for_node_not_found(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_get_full_node_list')
self.pm._get_full_node_list().AndReturn(["testNode"])
self.mox.StubOutWithMock(self.pm, '_run_command')
cmd = self.pm._vp_cmd.get_node_macs.replace('{_NodeName_}', 'testNode')
self.pm._run_command(cmd).AndReturn(["ffeeddccbbaa"])
self.mox.ReplayAll()
name = self.pm._check_for_node()
self.assertEqual(name, '')
self.mox.VerifyAll()
def test_activate_node(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.mox.StubOutWithMock(self.pm, 'is_power_on')
self.pm._check_for_node().AndReturn('"testNode"')
self.pm._run_command(self.pm._vp_cmd.start_cmd).AndReturn("Started")
self.pm.is_power_on().AndReturn(True)
self.mox.ReplayAll()
state = self.pm.activate_node()
self.assertEqual(state, 'active')
self.mox.VerifyAll()
def test_activate_node_fail(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.mox.StubOutWithMock(self.pm, 'is_power_on')
self.pm._check_for_node().AndReturn('"testNode"')
self.pm._run_command(self.pm._vp_cmd.start_cmd).AndReturn("Started")
self.pm.is_power_on().AndReturn(False)
self.mox.ReplayAll()
state = self.pm.activate_node()
self.assertEqual(state, 'error')
self.mox.VerifyAll()
def test_deactivate_node(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.mox.StubOutWithMock(self.pm, 'is_power_on')
self.pm._check_for_node().AndReturn('"testNode"')
self.pm.is_power_on().AndReturn(True)
self.pm._run_command(self.pm._vp_cmd.stop_cmd).AndReturn("Stopped")
self.pm.is_power_on().AndReturn(False)
self.mox.ReplayAll()
state = self.pm.deactivate_node()
self.assertEqual(state, 'deleted')
self.mox.VerifyAll()
def test_deactivate_node_fail(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.mox.StubOutWithMock(self.pm, 'is_power_on')
self.pm._check_for_node().AndReturn('"testNode"')
self.pm.is_power_on().AndReturn(True)
self.pm._run_command(self.pm._vp_cmd.stop_cmd).AndReturn("Stopped")
self.pm.is_power_on().AndReturn(True)
self.mox.ReplayAll()
state = self.pm.deactivate_node()
self.assertEqual(state, 'error')
self.mox.VerifyAll()
def test_reboot_node(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.mox.StubOutWithMock(self.pm, 'is_power_on')
self.pm._check_for_node().AndReturn(['"testNode"'])
self.pm._run_command(self.pm._vp_cmd.reboot_cmd).AndReturn("Restarted")
self.pm.is_power_on().AndReturn(True)
self.mox.ReplayAll()
state = self.pm.reboot_node()
self.assertEqual(state, 'active')
self.mox.VerifyAll()
def test_reboot_node_fail(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.mox.StubOutWithMock(self.pm, 'is_power_on')
self.pm._check_for_node().AndReturn(['"testNode"'])
self.pm._run_command(self.pm._vp_cmd.reboot_cmd).AndReturn("Restarted")
self.pm.is_power_on().AndReturn(False)
self.mox.ReplayAll()
state = self.pm.reboot_node()
self.assertEqual(state, 'error')
self.mox.VerifyAll()
def test_is_power_on(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.pm._check_for_node().AndReturn(['"testNode"'])
self.pm._run_command(self.pm._vp_cmd.list_running_cmd).\
AndReturn(['"testNode"'])
self.pm._matched_name = 'testNode'
self.mox.ReplayAll()
state = self.pm.is_power_on()
self.assertEqual(state, True)
self.mox.VerifyAll()
def test_is_power_on_fail(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.pm._check_for_node().AndReturn(['"NotFoundNode"'])
self.pm._run_command(self.pm._vp_cmd.list_running_cmd).\
AndReturn(['"NotFoundNode"'])
self.pm._matched_name = 'testNode'
self.mox.ReplayAll()
state = self.pm.is_power_on()
self.assertEqual(state, False)
self.mox.VerifyAll()
def test_is_power_on_match_subname(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(self.pm, '_run_command')
self.pm._check_for_node().AndReturn(['"testNode"'])
self.pm._run_command(self.pm._vp_cmd.list_running_cmd).\
AndReturn(['"testNode01"'])
self.pm._matched_name = '"testNode"'
self.mox.ReplayAll()
state = self.pm.is_power_on()
self.assertEqual(state, False)
self.mox.VerifyAll()
def test_run_command(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_set_connection')
self.mox.StubOutWithMock(processutils, 'ssh_execute')
self.pm._set_connection().AndReturn(True)
processutils.ssh_execute(None, '/usr/bin/VBoxManage test return',
check_exit_code=True).AndReturn(("test\nreturn", ""))
self.pm._matched_name = 'testNode'
self.mox.ReplayAll()
result = self.pm._run_command("test return")
self.assertEqual(result, ['test', 'return'])
self.mox.VerifyAll()
def test_run_command_raises_exception(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_set_connection')
self.mox.StubOutWithMock(processutils, 'ssh_execute')
self.pm._set_connection().AndReturn(True)
processutils.ssh_execute(None, '/usr/bin/VBoxManage test return',
check_exit_code=True).\
AndRaise(processutils.ProcessExecutionError)
self.mox.ReplayAll()
result = self.pm._run_command("test return")
self.assertEqual(result, [])
self.mox.VerifyAll()
def test_activate_node_with_exception(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_check_for_node')
self.mox.StubOutWithMock(processutils, 'ssh_execute')
self.pm._check_for_node().AndReturn(['"testNode"'])
self.pm._check_for_node().AndReturn(['"testNode"'])
processutils.ssh_execute('test', '/usr/bin/VBoxManage startvm ',
check_exit_code=True).\
AndRaise(processutils.ProcessExecutionError)
processutils.ssh_execute('test', '/usr/bin/VBoxManage list runningvms',
check_exit_code=True).\
AndRaise(processutils.ProcessExecutionError)
self.mox.ReplayAll()
self.pm._connection = 'test'
state = self.pm.activate_node()
self.assertEqual(state, 'error')
self.mox.VerifyAll()

161
nova/tests/baremetal/test_volume_driver.py
View File

@ -1,161 +0,0 @@
# Copyright (c) 2012 NTT DOCOMO, INC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Tests for baremetal volume driver."""
from oslo.config import cfg
from nova import test
from nova.virt.baremetal import volume_driver
CONF = cfg.CONF
SHOW_OUTPUT = """Target 1: iqn.2010-10.org.openstack:volume-00000001
System information:
Driver: iscsi
State: ready
I_T nexus information:
I_T nexus: 8
Initiator: iqn.1993-08.org.debian:01:7780c6a16b4
Connection: 0
IP Address: 172.17.12.10
LUN information:
LUN: 0
Type: controller
SCSI ID: IET 00010000
SCSI SN: beaf10
Size: 0 MB, Block size: 1
Online: Yes
Removable media: No
Readonly: No
Backing store type: null
Backing store path: None
Backing store flags:
LUN: 1
Type: disk
SCSI ID: IET 00010001
SCSI SN: beaf11
Size: 1074 MB, Block size: 512
Online: Yes
Removable media: No
Readonly: No
Backing store type: rdwr
Backing store path: /dev/nova-volumes/volume-00000001
Backing store flags:
Account information:
ACL information:
ALL
Target 2: iqn.2010-10.org.openstack:volume-00000002
System information:
Driver: iscsi
State: ready
I_T nexus information:
LUN information:
LUN: 0
Type: controller
SCSI ID: IET 00020000
SCSI SN: beaf20
Size: 0 MB, Block size: 1
Online: Yes
Removable media: No
Readonly: No
Backing store type: null
Backing store path: None
Backing store flags:
LUN: 1
Type: disk
SCSI ID: IET 00020001
SCSI SN: beaf21
Size: 2147 MB, Block size: 512
Online: Yes
Removable media: No
Readonly: No
Backing store type: rdwr
Backing store path: /dev/nova-volumes/volume-00000002
Backing store flags:
Account information:
ACL information:
ALL
Target 1000001: iqn.2010-10.org.openstack.baremetal:1000001-dev.vdc
System information:
Driver: iscsi
State: ready
I_T nexus information:
LUN information:
LUN: 0
Type: controller
SCSI ID: IET f42410000
SCSI SN: beaf10000010
Size: 0 MB, Block size: 1
Online: Yes
Removable media: No
Readonly: No
Backing store type: null
Backing store path: None
Backing store flags:
LUN: 1
Type: disk
SCSI ID: IET f42410001
SCSI SN: beaf10000011
Size: 1074 MB, Block size: 512
Online: Yes
Removable media: No
Readonly: No
Backing store type: rdwr
Backing store path: /dev/disk/by-path/ip-172.17.12.10:3260-iscsi-\
iqn.2010-10.org.openstack:volume-00000001-lun-1
Backing store flags:
Account information:
ACL information:
ALL
"""
def fake_show_tgtadm():
return SHOW_OUTPUT
class BareMetalVolumeTestCase(test.TestCase):
def setUp(self):
super(BareMetalVolumeTestCase, self).setUp()
self.stubs.Set(volume_driver, '_show_tgtadm', fake_show_tgtadm)
def test_list_backingstore_path(self):
l = volume_driver._list_backingstore_path()
self.assertEqual(len(l), 3)
self.assertIn('/dev/nova-volumes/volume-00000001', l)
self.assertIn('/dev/nova-volumes/volume-00000002', l)
self.assertIn('/dev/disk/by-path/ip-172.17.12.10:3260-iscsi-'
'iqn.2010-10.org.openstack:volume-00000001-lun-1', l)
def test_get_next_tid(self):
tid = volume_driver._get_next_tid()
self.assertEqual(1000002, tid)
def test_find_tid_found(self):
tid = volume_driver._find_tid(
'iqn.2010-10.org.openstack.baremetal:1000001-dev.vdc')
self.assertEqual(1000001, tid)
def test_find_tid_not_found(self):
tid = volume_driver._find_tid(
'iqn.2010-10.org.openstack.baremetal:1000002-dev.vdc')
self.assertTrue(tid is None)
def test_get_iqn(self):
self.flags(iscsi_iqn_prefix='iqn.2012-12.a.b', group='baremetal')
iqn = volume_driver._get_iqn('instname', '/dev/vdx')
self.assertEquals('iqn.2012-12.a.b:instname-dev-vdx', iqn)

9
nova/tests/test_baremetal_migrations.conf
View File

@ -1,9 +0,0 @@
[DEFAULT]
# Set up any number of migration data stores you want, one
# The "name" used in the test is the config variable key.
#sqlite=sqlite:///test_migrations.db
sqlite=sqlite://
#mysql=mysql://root:@localhost/test_migrations
#postgresql=postgresql://user:pass@localhost/test_migrations
[walk_style]
snake_walk=yes

2
nova/tests/test_migrations.py
View File

@ -1339,7 +1339,7 @@ class TestBaremetalMigrations(BaseMigrationTestCase, CommonTestsMixIn):
super(TestBaremetalMigrations, self).__init__(*args, **kwargs)
self.DEFAULT_CONFIG_FILE = os.path.join(os.path.dirname(__file__),
'test_baremetal_migrations.conf')
'virt/baremetal/test_baremetal_migrations.conf')
# Test machines can set the NOVA_TEST_MIGRATIONS_CONF variable
# to override the location of the config file for migration testing
self.CONFIG_FILE_PATH = os.environ.get(