x/python-ganttclient
Code Issues Proposed changes
Files
0ac15713239fff81b1890731e7e3e9e85de1c7b6
python-ganttclient/nova/tests/baremetal/test_virtual_power_driver.py
Devananda van der Veen 5008d67838 VirtualPowerDriver catches ProcessExecutionError. 
Baremetal virtual_power_driver now catches ProcessExecutionError coming
from ssh_execute and logs an exception. Requested actions (eg, power_on)
return an error state instead of raising an exception.

Change-Id: I8f66ba843e12de13f9a14d736f7bef2eda4ef85c
2013-02-14 14:19:39 -08:00

368 lines
13 KiB
Python
Raw Blame History

# 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 nova import exception
from nova.openstack.common import cfg
from nova import utils as nutils
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,
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,
prov_mac_address='11:11:11:11:11:11',
)
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},
]
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(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.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(["aabbccddeeff", "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_run_command(self):
self._create_node()
self._create_pm()
self.mox.StubOutWithMock(self.pm, '_set_connection')
self.mox.StubOutWithMock(nutils, 'ssh_execute')
self.pm._set_connection().AndReturn(True)
nutils.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(nutils, 'ssh_execute')
self.pm._set_connection().AndReturn(True)
nutils.ssh_execute(None, '/usr/bin/VBoxManage test return',
check_exit_code=True).\
AndRaise(exception.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(nutils, 'ssh_execute')
self.pm._check_for_node().AndReturn(["testNode"])
self.pm._check_for_node().AndReturn(["testNode"])
nutils.ssh_execute('test', '/usr/bin/VBoxManage startvm ',
check_exit_code=True).\
AndRaise(exception.ProcessExecutionError)
nutils.ssh_execute('test', '/usr/bin/VBoxManage list runningvms',
check_exit_code=True).\
AndRaise(exception.ProcessExecutionError)
self.mox.ReplayAll()
self.pm._connection = 'test'
state = self.pm.activate_node()
self.assertEqual(state, 'error')
self.mox.VerifyAll()
View Git Blame Copy Permalink