openstack/ironic
Code Issues Proposed changes

Add pxe_ucs and agent_ucs drivers to manage Cisco UCS servers

Browse Source 
This commit adds pxe_ucs, agent_ucs driver changes.
Adds support for the following:

- Power driver changes
- Management interface changes
- UCS Helper code, which can be reused with other parts of the driver,
  ex: to support vendor apis.

Change-Id: I02211da5fad039dc7e6b509d547e473e9b57009b
Implements: blueprint cisco-ucs-pxe-driver
This commit is contained in:
saripurigopi 2015-02-27 12:47:03 +05:30
parent 102b85bd31
commit 8c26a8c33b
19 changed files with 1256 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
doc/source/deploy/drivers.rst
View File

@ -119,7 +119,6 @@ iRMC
The iRMC driver enables PXE Deploy to control power via ServerView Common
Command Interface (SCCI).
Software Requirements
^^^^^^^^^^^^^^^^^^^^^
@ -162,3 +161,12 @@ VirtualBox drivers
:maxdepth: 1
../drivers/vbox
Cisco UCS Driver
----------------
.. toctree::
:maxdepth: 1
../drivers/ucs

84
doc/source/drivers/ucs.rst Normal file
View File

@ -0,0 +1,84 @@
.. _UCS:
===========
UCS drivers
===========
Overview
========
The UCS driver is targeted for UCS Manager managed Cisco UCS B/C series
servers. The pxe_ucs, agent_ucs drivers enables you to take advantage of
UCS Manager by using the python SDK.
``pxe_ucs`` driver uses PXE/iSCSI (just like ``pxe_ipmitool`` driver) to
deploy the image and uses UCS to do all management operations on the
baremetal node (instead of using IPMI).
``agent_ucs`` driver uses IPA ramdisk (just like ``agent_ipmitool`` and
``agent_ipminative`` drivers.) to deploy the image and uses UCS to do all
management operations on the baremetal node (instead of using IPMI).
Prerequisites
=============
* ``UcsSdk`` is a python package version of XML API sdk available to
to manage Cisco UCS Managed B/C-series servers.
Install ``UcsSdk`` [1]_ module on the Ironic conductor node.
Required version is 0.8.1.6::
$ pip install "UcsSdk==0.8.1.6"
Tested Platforms
~~~~~~~~~~~~~~~~
This driver works on Cisco UCS Manager Managed B/C-series servers.
It has been tested with the following servers:
UCS Manager version: 2.2(1b), 2.2(3d).
* UCS B22M, B200M3
* UCS C220M3.
All the Cisco UCS B/C-series servers managed by UCSM 2.1 or later are supported
by this driver.
Configuring and Enabling the driver
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
1. Add ``pxe_ucs`` and/or ``agent_ucs`` to the list of ``enabled_drivers`` in
``/etc/ironic/ironic.conf``. For example::
enabled_drivers = pxe_ipmitool,pxe_ucs,agent_ucs
2. Restart the Ironic conductor service::
service ironic-conductor restart
Registering UCS node in Ironic
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Nodes configured for UCS driver should have the ``driver`` property set to
``pxe_ucs/agent_ucs``. The following configuration values are also required in
``driver_info``:
- ``ucs_hostname``: IP address or hostname of the UCS Manager
- ``ucs_username``: UCS Manager login user name with administrator or
server_profile privileges.
- ``ucs_password``: UCS Manager login password for the above UCS Manager user.
- ``deploy_kernel``: The Glance UUID of the deployment kernel.
- ``deploy_ramdisk``: The Glance UUID of the deployment ramdisk.
- ``ucs_service_profile``: Distinguished name(DN) of service_profile being enrolled.
The following sequence of commands can be used to enroll a UCS node.
Create Node::
ironic node-create -d <pxe_ucs/agent_ucs> -i ucs_hostname=<UCS Manager hostname/ip-address> -i ucs_username=<ucsm_username> -i ucs_password=<ucsm_password> -i ucs_service_profile=<serivce_profile_dn_being_enrolled> -i deploy_kernel=<glance_uuid_of_deploy_kernel> -i deploy_ramdisk=<glance_uuid_of_deploy_ramdisk> -p cpus=<number_of_cpus> -p memory_mb=<memory_size_in_MB> -p local_gb=<local_disk_size_in_GB> -p cpu_arch=<cpu_arch>
The above command 'ironic node-create' will return UUID of the node, which is the value of $NODE in the following command.
Associate port with the node created::
ironic port-create -n $NODE -a <MAC_address_of_Ucs_server's_NIC>
References
==========
.. [1] UcsSdk - https://pypi.python.org/pypi/UcsSdk

1
driver-requirements.txt
View File

@ -10,6 +10,7 @@ pyghmi
pysnmp
python-scciclient
python-seamicroclient>=0.4.0
UcsSdk==0.8.1.6
# The drac and amt driver import a python module called "pywsman", however,
# this does not exist on pypi.

11
etc/ironic/ironic.conf.sample
View File

@ -1600,3 +1600,14 @@
#port=18083
[cisco_ucs]
#
# Options defined in ironic.drivers.modules.ucs.power
#
# Number of times a power operation needs to be retried.
#max_retry=6
# Amount of time in seconds to wait in between power operations.
#action_interval=5

10
ironic/common/exception.py
View File

@ -572,3 +572,13 @@ class PathNotFound(IronicException):
class DirectoryNotWritable(IronicException):
message = _("Directory %(dir)s is not writable.")
class UcsOperationError(IronicException):
message = _("Cisco UCS client: operation %(operation)s failed for node"
" %(node)s. Reason: %(error)s")
class UcsConnectionError(IronicException):
message = _("Cisco UCS client: connection failed for node "
"%(node)s. Reason: %(error)s")

24
ironic/drivers/agent.py
View File

@ -21,6 +21,8 @@ from ironic.drivers.modules import agent
from ironic.drivers.modules import ipminative
from ironic.drivers.modules import ipmitool
from ironic.drivers.modules import ssh
from ironic.drivers.modules.ucs import management as ucs_mgmt
from ironic.drivers.modules.ucs import power as ucs_power
from ironic.drivers.modules import virtualbox
@ -105,3 +107,25 @@ class AgentAndVirtualBoxDriver(base.BaseDriver):
self.deploy = agent.AgentDeploy()
self.management = virtualbox.VirtualBoxManagement()
self.vendor = agent.AgentVendorInterface()
class AgentAndUcsDriver(base.BaseDriver):
"""Agent + Cisco UCSM driver.
This driver implements the `core` functionality, combining
:class:ironic.drivers.modules.ucs.power.Power for power
on/off and reboot with
:class:'ironic.driver.modules.agent.AgentDeploy' (for image deployment.)
Implementations are in those respective classes;
this class is merely the glue between them.
"""
def __init__(self):
if not importutils.try_import('UcsSdk'):
raise exception.DriverLoadError(
driver=self.__class__.__name__,
reason=_("Unable to import UcsSdk library"))
self.power = ucs_power.Power()
self.deploy = agent.AgentDeploy()
self.management = ucs_mgmt.UcsManagement()
self.vendor = agent.AgentVendorInterface()

15
ironic/drivers/fake.py
View File

@ -43,6 +43,8 @@ from ironic.drivers.modules import pxe
from ironic.drivers.modules import seamicro
from ironic.drivers.modules import snmp
from ironic.drivers.modules import ssh
from ironic.drivers.modules.ucs import management as ucs_mgmt
from ironic.drivers.modules.ucs import power as ucs_power
from ironic.drivers.modules import virtualbox
from ironic.drivers import utils
@ -245,3 +247,16 @@ class FakeMSFTOCSDriver(base.BaseDriver):
self.power = msftocs_power.MSFTOCSPower()
self.deploy = fake.FakeDeploy()
self.management = msftocs_management.MSFTOCSManagement()
class FakeUcsDriver(base.BaseDriver):
"""Fake UCS driver."""
def __init__(self):
if not importutils.try_import('UcsSdk'):
raise exception.DriverLoadError(
driver=self.__class__.__name__,
reason=_("Unable to import UcsSdk library"))
self.power = ucs_power.Power()
self.deploy = fake.FakeDeploy()
self.management = ucs_mgmt.UcsManagement()

0
ironic/drivers/modules/ucs/__init__.py Normal file
View File

126
ironic/drivers/modules/ucs/helper.py Normal file
View File

@ -0,0 +1,126 @@
# Copyright 2015, Cisco Systems.
# 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.
"""
Ironic Cisco UCSM helper functions
"""
import functools
from oslo_log import log as logging
from oslo_utils import importutils
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common.i18n import _LE
from ironic.drivers.modules import deploy_utils
ucs_helper = importutils.try_import('UcsSdk.utils.helper')
ucs_error = importutils.try_import('UcsSdk.utils.exception')
LOG = logging.getLogger(__name__)
REQUIRED_PROPERTIES = {
'ucs_address': _('IP or Hostname of the UCS Manager. Required.'),
'ucs_username': _('UCS Manager admin/server-profile username. Required.'),
'ucs_password': _('UCS Manager password. Required.'),
'ucs_service_profile': _('UCS Manager service-profile name. Required.')
}
COMMON_PROPERTIES = REQUIRED_PROPERTIES
def requires_ucs_client(func):
"""Creates handle to connect to UCS Manager.
This method is being used as a decorator method. It establishes connection
with UCS Manager. And creates a session. Any method that has to perform
operation on UCS Manager, requries this session, which can use this method
as decorator method. Use this method as decorator method requires having
helper keyword argument in the definition.
:param func: function using this as a decorator.
:returns: a wrapper function that performs the required tasks
mentioned above before and after calling the actual function.
"""
@functools.wraps(func)
def wrapper(self, task, *args, **kwargs):
if kwargs.get('helper') is None:
kwargs['helper'] = CiscoUcsHelper(task)
try:
kwargs['helper'].connect_ucsm()
return func(self, task, *args, **kwargs)
finally:
kwargs['helper'].logout()
return wrapper
def parse_driver_info(node):
"""Parses and creates Cisco driver info
:param node: An Ironic node object.
:returns: dictonary that contains node.driver_info parameter/values.
:raises: MissingParameterValue if any required parameters are missing.
"""
info = {}
for param in REQUIRED_PROPERTIES:
info[param] = node.driver_info.get(param)
error_msg = _("cisco driver requries these parameter to be set.")
deploy_utils.check_for_missing_params(info, error_msg)
return info
class CiscoUcsHelper(object):
"""Cisco UCS helper. Performs session managemnt."""
def __init__(self, task):
"""Initialize with UCS Manager details.
:param task: instance of `ironic.manager.task_manager.TaskManager`.
"""
info = parse_driver_info(task.node)
self.address = info['ucs_address']
self.username = info['ucs_username']
self.password = info['ucs_password']
# service_profile is used by the utilities functions in UcsSdk.utils.*.
self.service_profile = info['ucs_service_profile']
self.handle = None
self.uuid = task.node.uuid
def connect_ucsm(self):
"""Creates the UcsHandle
:raises: UcsConnectionError, if ucs helper failes to establish session
with UCS Manager.
"""
try:
success, self.handle = ucs_helper.generate_ucsm_handle(
self.address,
self.username,
self.password)
except ucs_error.UcsConnectionError as ucs_exception:
LOG.error(_LE("Cisco client: service unavailable for node "
"%(uuid)s."), {'uuid': self.uuid})
raise exception.UcsConnectionError(error=ucs_exception,
node=self.uuid)
def logout(self):
"""Logouts the current active session."""
if self.handle:
self.handle.Logout()

146
ironic/drivers/modules/ucs/management.py Normal file
View File

@ -0,0 +1,146 @@
# Copyright 2015, Cisco Systems.
# 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.
"""
Ironic Cisco UCSM interfaces.
Provides Management interface operations of servers managed by Cisco UCSM using
PyUcs Sdk.
"""
from oslo_log import log as logging
from oslo_utils import importutils
from ironic.common import boot_devices
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common.i18n import _LE
from ironic.drivers import base
from ironic.drivers.modules.ucs import helper as ucs_helper
ucs_error = importutils.try_import('UcsSdk.utils.exception')
ucs_mgmt = importutils.try_import('UcsSdk.utils.management')
LOG = logging.getLogger(__name__)
UCS_TO_IRONIC_BOOT_DEVICE = {
'storage': boot_devices.DISK,
'pxe': boot_devices.PXE,
'read-only-vm': boot_devices.CDROM
}
class UcsManagement(base.ManagementInterface):
def get_properties(self):
return ucs_helper.COMMON_PROPERTIES
def validate(self, task):
"""Check that 'driver_info' contains UCSM login credentials.
Validates whether the 'driver_info' property of the supplied
task's node contains the required credentials information.
:param task: a task from TaskManager.
:raises: MissingParameterValue if a required parameter is missing
"""
ucs_helper.parse_driver_info(task.node)
def get_supported_boot_devices(self):
"""Get a list of the supported boot devices.
:returns: A list with the supported boot devices defined
in :mod:`ironic.common.boot_devices`.
"""
return list(UCS_TO_IRONIC_BOOT_DEVICE.values())
@ucs_helper.requires_ucs_client
def set_boot_device(self, task, device, persistent=False, helper=None):
"""Set the boot device for the task's node.
Set the boot device to use on next reboot of the node.
:param task: a task from TaskManager.
:param device: the boot device, one of 'PXE, DISK or CDROM'.
:param persistent: Boolean value. True if the boot device will
persist to all future boots, False if not.
Default: False. Ignored by this driver.
:param helper: ucs helper instance.
:raises: MissingParameterValue if required CiscoDriver parameters
are missing.
:raises: UcsOperationError on error from UCS client.
setting the boot device.
"""
try:
mgmt_handle = ucs_mgmt.BootDeviceHelper(helper)
mgmt_handle.set_boot_device(device, persistent)
except ucs_error.UcsOperationError as ucs_exception:
LOG.error(_LE("%(driver)s: client failed to set boot device "
"%(device)s for node %(uuid)s."),
{'driver': task.node.driver, 'device': device,
'uuid': task.node.uuid})
operation = _('setting boot device')
raise exception.UcsOperationError(operation=operation,
error=ucs_exception,
node=task.node.uuid)
LOG.debug("Node %(uuid)s set to boot from %(device)s.",
{'uuid': task.node.uuid, 'device': device})
@ucs_helper.requires_ucs_client
def get_boot_device(self, task, helper=None):
"""Get the current boot device for the task's node.
Provides the current boot device of the node.
:param task: a task from TaskManager.
:param helper: ucs helper instance.
:returns: a dictionary containing:
:boot_device: the boot device, one of
:mod:`ironic.common.boot_devices` [PXE, DISK, CDROM] or
None if it is unknown.
:persistent: Whether the boot device will persist to all
future boots or not, None if it is unknown.
:raises: MissingParameterValue if a required UCS parameter is missing.
:raises: UcsOperationError on error from UCS client, while setting the
boot device.
"""
try:
mgmt_handle = ucs_mgmt.BootDeviceHelper(helper)
boot_device = mgmt_handle.get_boot_device()
except ucs_error.UcsOperationError as ucs_exception:
LOG.error(_LE("%(driver)s: client failed to get boot device for "
"node %(uuid)s."),
{'driver': task.node.driver, 'uuid': task.node.uuid})
operation = _('getting boot device')
raise exception.UcsOperationError(operation=operation,
error=ucs_exception,
node=task.node.uuid)
boot_device['boot_device'] = (
UCS_TO_IRONIC_BOOT_DEVICE[boot_device['boot_device']])
return boot_device
def get_sensors_data(self, task):
"""Get sensors data.
Not implemented by this driver.
:param task: a TaskManager instance.
"""
raise NotImplementedError()

212
ironic/drivers/modules/ucs/power.py Normal file
View File

@ -0,0 +1,212 @@
# Copyright 2015, Cisco Systems.
# 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.
"""
Ironic Cisco UCSM interfaces.
Provides basic power control of servers managed by Cisco UCSM using PyUcs Sdk.
"""
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import importutils
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common.i18n import _LE
from ironic.common import states
from ironic.conductor import task_manager
from ironic.drivers import base
from ironic.drivers.modules.ucs import helper as ucs_helper
from ironic.openstack.common import loopingcall
ucs_power = importutils.try_import('UcsSdk.utils.power')
ucs_error = importutils.try_import('UcsSdk.utils.exception')
opts = [
cfg.IntOpt('max_retry',
default=6,
help='Number of times a power operation needs to be retried'),
cfg.IntOpt('action_interval',
default=5,
help='Amount of time in seconds to wait in between power '
'operations'),
]
CONF = cfg.CONF
CONF.register_opts(opts, group='cisco_ucs')
LOG = logging.getLogger(__name__)
UCS_TO_IRONIC_POWER_STATE = {
'up': states.POWER_ON,
'down': states.POWER_OFF,
}
IRONIC_TO_UCS_POWER_STATE = {
states.POWER_ON: 'up',
states.POWER_OFF: 'down',
states.REBOOT: 'hard-reset-immediate'
}
def _wait_for_state_change(target_state, ucs_power_handle):
"""Wait and check for the power state change."""
state = [None]
retries = [0]
def _wait(state, retries):
state[0] = ucs_power_handle.get_power_state()
if ((retries[0] != 0) and (
UCS_TO_IRONIC_POWER_STATE.get(state[0]) == target_state)):
raise loopingcall.LoopingCallDone()
if retries[0] > CONF.cisco_ucs.max_retry:
state[0] = states.ERROR
raise loopingcall.LoopingCallDone()
retries[0] += 1
timer = loopingcall.FixedIntervalLoopingCall(_wait, state, retries)
timer.start(interval=CONF.cisco_ucs.action_interval).wait()
return UCS_TO_IRONIC_POWER_STATE.get(state[0], states.ERROR)
class Power(base.PowerInterface):
"""Cisco Power Interface.
This PowerInterface class provides a mechanism for controlling the
power state of servers managed by Cisco UCS Manager.
"""
def get_properties(self):
"""Returns common properties of the driver."""
return ucs_helper.COMMON_PROPERTIES
def validate(self, task):
"""Check that node 'driver_info' is valid.
Check that node 'driver_info' contains the required fields.
:param task: instance of `ironic.manager.task_manager.TaskManager`.
:raises: MissingParameterValue if required CiscoDriver parameters
are missing.
"""
ucs_helper.parse_driver_info(task.node)
@ucs_helper.requires_ucs_client
def get_power_state(self, task, helper=None):
"""Get the current power state.
Poll the host for the current power state of the node.
:param task: instance of `ironic.manager.task_manager.TaskManager`.
:param helper: ucs helper instance
:raises: MissingParameterValue if required CiscoDriver parameters
are missing.
:raises: UcsOperationError on error from UCS Client.
:returns: power state. One of :class:`ironic.common.states`.
"""
try:
power_handle = ucs_power.UcsPower(helper)
power_status = power_handle.get_power_state()
except ucs_error.UcsOperationError as ucs_exception:
LOG.error(_LE("%(driver)s: get_power_status operation failed for "
"node %(uuid)s with error."),
{'driver': task.node.driver, 'uuid': task.node.uuid})
operation = _('getting power status')
raise exception.UcsOperationError(operation=operation,
error=ucs_exception,
node=task.node.uuid)
return UCS_TO_IRONIC_POWER_STATE.get(power_status, states.ERROR)
@task_manager.require_exclusive_lock
@ucs_helper.requires_ucs_client
def set_power_state(self, task, pstate, helper=None):
"""Turn the power on or off.
Set the power state of a node.
:param task: instance of `ironic.manager.task_manager.TaskManager`.
:param pstate: Either POWER_ON or POWER_OFF from :class:
`ironic.common.states`.
:param helper: ucs helper instance
:raises: InvalidParameterValue if an invalid power state was specified.
:raises: MissingParameterValue if required CiscoDriver parameters
are missing.
:raises: UcsOperationError on error from UCS Client.
:raises: PowerStateFailure if the desired power state couldn't be set.
"""
if pstate not in (states.POWER_ON, states.POWER_OFF):
msg = _("set_power_state called with invalid power state "
"'%s'") % pstate
raise exception.InvalidParameterValue(msg)
try:
ucs_power_handle = ucs_power.UcsPower(helper)
power_status = ucs_power_handle.get_power_state()
if UCS_TO_IRONIC_POWER_STATE.get(power_status) != pstate:
ucs_power_handle.set_power_state(
IRONIC_TO_UCS_POWER_STATE.get(pstate))
else:
return
except ucs_error.UcsOperationError as ucs_exception:
LOG.error(_LE("Cisco client exception %(msg)s for node %(uuid)s"),
{'msg': ucs_exception, 'uuid': task.node.uuid})
operation = _("setting power status")
raise exception.UcsOperationError(operation=operation,
error=ucs_exception,
node=task.node.uuid)
state = _wait_for_state_change(pstate, ucs_power_handle)
if state != pstate:
timeout = CONF.cisco_ucs.action_interval * CONF.cisco_ucs.max_retry
LOG.error(_LE("%(driver)s: driver failed to change node %(uuid)s "
"power state to %(state)s within %(timeout)s "
"seconds."),
{'driver': task.node.driver, 'uuid': task.node.uuid,
'state': pstate, 'timeout': timeout})
raise exception.PowerStateFailure(pstate=pstate)
@task_manager.require_exclusive_lock
@ucs_helper.requires_ucs_client
def reboot(self, task, helper=None):
"""Cycles the power to a node.
:param task: a TaskManager instance.
:param helper: ucs helper instance.
:raises: UcsOperationError on error from UCS Client.
:raises: PowerStateFailure if the final state of the node is not
POWER_ON.
"""
try:
ucs_power_handle = ucs_power.UcsPower(helper)
ucs_power_handle.reboot()
except ucs_error.UcsOperationError as ucs_exception:
LOG.error(_LE("%(driver)s: driver failed to reset node %(uuid)s "
"power state."),
{'driver': task.node.driver, 'uuid': task.node.uuid})
operation = _("rebooting")
raise exception.UcsOperationError(operation=operation,
error=ucs_exception,
node=task.node.uuid)
state = _wait_for_state_change(states.POWER_ON, ucs_power_handle)
if state != states.POWER_ON:
timeout = CONF.cisco_ucs.action_interval * CONF.cisco_ucs.max_retry
LOG.error(_LE("%(driver)s: driver failed to reboot node %(uuid)s "
"within %(timeout)s seconds."),
{'driver': task.node.driver,
'uuid': task.node.uuid, 'timeout': timeout})
raise exception.PowerStateFailure(pstate=states.POWER_ON)

24
ironic/drivers/pxe.py
View File

@ -41,6 +41,8 @@ from ironic.drivers.modules import pxe
from ironic.drivers.modules import seamicro
from ironic.drivers.modules import snmp
from ironic.drivers.modules import ssh
from ironic.drivers.modules.ucs import management as ucs_mgmt
from ironic.drivers.modules.ucs import power as ucs_power
from ironic.drivers.modules import virtualbox
from ironic.drivers import utils
@ -285,3 +287,25 @@ class PXEAndMSFTOCSDriver(base.BaseDriver):
self.deploy = pxe.PXEDeploy()
self.management = msftocs_management.MSFTOCSManagement()
self.vendor = pxe.VendorPassthru()
class PXEAndUcsDriver(base.BaseDriver):
"""PXE + Cisco UCSM driver.
This driver implements the `core` functionality, combining
:class:ironic.drivers.modules.ucs.power.Power for power
on/off and reboot with
:class:ironic.driver.modules.pxe.PXE for image deployment.
Implementations are in those respective classes;
this class is merely the glue between them.
"""
def __init__(self):
if not importutils.try_import('UcsSdk'):
raise exception.DriverLoadError(
driver=self.__class__.__name__,
reason=_("Unable to import UcsSdk library"))
self.power = ucs_power.Power()
self.deploy = pxe.PXEDeploy()
self.management = ucs_mgmt.UcsManagement()
self.vendor = pxe.VendorPassthru()

9
ironic/tests/db/utils.py
View File

@ -307,3 +307,12 @@ def get_test_conductor(**kw):
'created_at': kw.get('created_at', timeutils.utcnow()),
'updated_at': kw.get('updated_at', timeutils.utcnow()),
}
def get_test_ucs_info():
return {
"ucs_username": "admin",
"ucs_password": "password",
"ucs_service_profile": "org-root/ls-devstack",
"ucs_address": "ucs-b",
}

23
ironic/tests/drivers/third_party_driver_mocks.py
View File

@ -197,3 +197,26 @@ if not ironic_inspector:
if 'ironic.drivers.modules.inspector' in sys.modules:
six.moves.reload_module(
sys.modules['ironic.drivers.modules.inspector'])
class MockKwargsException(Exception):
def __init__(self, *args, **kwargs):
super(MockKwargsException, self).__init__(*args)
self.kwargs = kwargs
ucssdk = importutils.try_import('UcsSdk')
if not ucssdk:
ucssdk = mock.MagicMock()
sys.modules['UcsSdk'] = ucssdk
sys.modules['UcsSdk.utils'] = ucssdk.utils
sys.modules['UcsSdk.utils.power'] = ucssdk.utils.power
sys.modules['UcsSdk.utils.management'] = ucssdk.utils.management
sys.modules['UcsSdk.utils.exception'] = ucssdk.utils.exception
ucssdk.utils.exception.UcsOperationError = (
type('UcsOperationError', (MockKwargsException,), {}))
ucssdk.utils.exception.UcsConnectionError = (
type('UcsConnectionError', (MockKwargsException,), {}))
if 'ironic.drivers.modules.ucs' in sys.modules:
six.moves.reload_module(
sys.modules['ironic.drivers.modules.ucs'])

0
ironic/tests/drivers/ucs/__init__.py Normal file
View File

161
ironic/tests/drivers/ucs/test_helper.py Normal file
View File

@ -0,0 +1,161 @@
# Copyright 2015, Cisco Systems.
# 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 common methods used by UCS modules."""
import mock
from oslo_config import cfg
from oslo_utils import importutils
from ironic.common import exception
from ironic.conductor import task_manager
from ironic.db import api as dbapi
from ironic.drivers.modules.ucs import helper as ucs_helper
from ironic.tests.conductor import utils as mgr_utils
from ironic.tests.db import base as db_base
from ironic.tests.db import utils as db_utils
from ironic.tests.objects import utils as obj_utils
ucs_error = importutils.try_import('UcsSdk.utils.exception')
INFO_DICT = db_utils.get_test_ucs_info()
CONF = cfg.CONF
class UcsValidateParametersTestCase(db_base.DbTestCase):
def setUp(self):
super(UcsValidateParametersTestCase, self).setUp()
mgr_utils.mock_the_extension_manager(driver="fake_ucs")
self.node = obj_utils.create_test_node(self.context,
driver='fake_ucs',
driver_info=INFO_DICT)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.helper = ucs_helper.CiscoUcsHelper(task)
def test_parse_driver_info(self):
info = ucs_helper.parse_driver_info(self.node)
self.assertIsNotNone(info.get('ucs_address'))
self.assertIsNotNone(info.get('ucs_username'))
self.assertIsNotNone(info.get('ucs_password'))
self.assertIsNotNone(info.get('ucs_service_profile'))
def test_parse_driver_info_missing_address(self):
del self.node.driver_info['ucs_address']
self.assertRaises(exception.MissingParameterValue,
ucs_helper.parse_driver_info, self.node)
def test_parse_driver_info_missing_username(self):
del self.node.driver_info['ucs_username']
self.assertRaises(exception.MissingParameterValue,
ucs_helper.parse_driver_info, self.node)
def test_parse_driver_info_missing_password(self):
del self.node.driver_info['ucs_password']
self.assertRaises(exception.MissingParameterValue,
ucs_helper.parse_driver_info, self.node)
def test_parse_driver_info_missing_service_profile(self):
del self.node.driver_info['ucs_service_profile']
self.assertRaises(exception.MissingParameterValue,
ucs_helper.parse_driver_info, self.node)
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
def test_connect_ucsm(self, mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.helper.connect_ucsm()
mock_helper.generate_ucsm_handle.assert_called_once_with(
task.node.driver_info['ucs_address'],
task.node.driver_info['ucs_username'],
task.node.driver_info['ucs_password']
)
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
def test_connect_ucsm_fail(self, mock_helper):
side_effect = ucs_error.UcsConnectionError(
message='connecting to ucsm',
error='failed')
mock_helper.generate_ucsm_handle.side_effect = side_effect
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.assertRaises(exception.UcsConnectionError,
self.helper.connect_ucsm
)
mock_helper.generate_ucsm_handle.assert_called_once_with(
task.node.driver_info['ucs_address'],
task.node.driver_info['ucs_username'],
task.node.driver_info['ucs_password']
)
@mock.patch('ironic.drivers.modules.ucs.helper',
autospec=True)
def test_logout(self, mock_helper):
self.helper.logout()
class UcsCommonMethodsTestcase(db_base.DbTestCase):
def setUp(self):
super(UcsCommonMethodsTestcase, self).setUp()
self.dbapi = dbapi.get_instance()
mgr_utils.mock_the_extension_manager(driver="fake_ucs")
self.node = obj_utils.create_test_node(self.context,
driver='fake_ucs',
driver_info=INFO_DICT.copy())
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.helper = ucs_helper.CiscoUcsHelper(task)
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper', autospec=True)
@mock.patch('ironic.drivers.modules.ucs.helper.CiscoUcsHelper',
autospec=True)
def test_requires_ucs_client_ok_logout(self, mc_helper, mock_ucs_helper):
mock_helper = mc_helper.return_value
mock_helper.logout.return_value = None
mock_working_function = mock.Mock()
mock_working_function.__name__ = "Working"
mock_working_function.return_valure = "Success"
mock_ucs_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
wont_error = ucs_helper.requires_ucs_client(
mock_working_function)
wont_error(wont_error, task)
mock_helper.logout.assert_called_once_with()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper', autospec=True)
@mock.patch('ironic.drivers.modules.ucs.helper.CiscoUcsHelper',
autospec=True)
def test_requires_ucs_client_fail_logout(self, mc_helper, mock_ucs_helper):
mock_helper = mc_helper.return_value
mock_helper.logout.return_value = None
mock_broken_function = mock.Mock()
mock_broken_function.__name__ = "Broken"
mock_broken_function.side_effect = exception.IronicException()
mock_ucs_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
will_error = ucs_helper.requires_ucs_client(mock_broken_function)
self.assertRaises(exception.IronicException,
will_error, will_error, task)
mock_helper.logout.assert_called_once_with()

139
ironic/tests/drivers/ucs/test_management.py Normal file
View File

@ -0,0 +1,139 @@
# Copyright 2015, Cisco Systems.
# 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 UCS ManagementInterface
"""
import mock
from oslo_config import cfg
from oslo_utils import importutils
from ironic.common import boot_devices
from ironic.common import exception
from ironic.conductor import task_manager
from ironic.drivers.modules.ucs import helper as ucs_helper
from ironic.drivers.modules.ucs import management as ucs_mgmt
from ironic.tests.conductor import utils as mgr_utils
from ironic.tests.db import base as db_base
from ironic.tests.db import utils as db_utils
from ironic.tests.objects import utils as obj_utils
ucs_error = importutils.try_import('UcsSdk.utils.exception')
INFO_DICT = db_utils.get_test_ucs_info()
CONF = cfg.CONF
class UcsManagementTestCase(db_base.DbTestCase):
def setUp(self):
super(UcsManagementTestCase, self).setUp()
mgr_utils.mock_the_extension_manager(driver='fake_ucs')
self.node = obj_utils.create_test_node(self.context,
driver='fake_ucs',
driver_info=INFO_DICT)
self.interface = ucs_mgmt.UcsManagement()
self.task = mock.Mock()
self.task.node = self.node
def test_get_properties(self):
expected = ucs_helper.COMMON_PROPERTIES
self.assertEqual(expected, self.interface.get_properties())
def test_get_supported_boot_devices(self):
expected = [boot_devices.PXE, boot_devices.DISK, boot_devices.CDROM]
self.assertEqual(sorted(expected),
sorted(self.interface.get_supported_boot_devices()))
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch(
'ironic.drivers.modules.ucs.management.ucs_mgmt.BootDeviceHelper',
spec_set=True, autospec=True)
def test_get_boot_device(self, mock_ucs_mgmt, mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock_mgmt = mock_ucs_mgmt.return_value
mock_mgmt.get_boot_device.return_value = {
'boot_device': 'storage',
'persistent': False
}
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
expected_device = boot_devices.DISK
expected_response = {'boot_device': expected_device,
'persistent': False}
self.assertEqual(expected_response,
self.interface.get_boot_device(task))
mock_mgmt.get_boot_device.assert_called_once_with()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch(
'ironic.drivers.modules.ucs.management.ucs_mgmt.BootDeviceHelper',
spec_set=True, autospec=True)
def test_get_boot_device_fail(self, mock_ucs_mgmt, mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock_mgmt = mock_ucs_mgmt.return_value
side_effect = ucs_error.UcsOperationError(
operation='getting boot device',
error='failed',
node=self.node.uuid
)
mock_mgmt.get_boot_device.side_effect = side_effect
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.UcsOperationError,
self.interface.get_boot_device,
task)
mock_mgmt.get_boot_device.assert_called_once_with()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch(
'ironic.drivers.modules.ucs.management.ucs_mgmt.BootDeviceHelper',
spec_set=True, autospec=True)
def test_set_boot_device(self, mock_mgmt, mock_helper):
mc_mgmt = mock_mgmt.return_value
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.interface.set_boot_device(task, boot_devices.CDROM)
mc_mgmt.set_boot_device.assert_called_once_with('cdrom', False)
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch(
'ironic.drivers.modules.ucs.management.ucs_mgmt.BootDeviceHelper',
spec_set=True, autospec=True)
def test_set_boot_device_fail(self, mock_mgmt, mock_helper):
mc_mgmt = mock_mgmt.return_value
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
side_effect = exception.UcsOperationError(
operation='setting boot device',
error='failed',
node=self.node.uuid)
mc_mgmt.set_boot_device.side_effect = side_effect
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.IronicException,
self.interface.set_boot_device,
task, boot_devices.PXE)
mc_mgmt.set_boot_device.assert_called_once_with(
boot_devices.PXE, False)
def test_get_sensors_data(self):
self.assertRaises(NotImplementedError,
self.interface.get_sensors_data, self.task)

259
ironic/tests/drivers/ucs/test_power.py Normal file
View File

@ -0,0 +1,259 @@
# Copyright 2015, Cisco Systems.
# 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 UcsPower module."""
import mock
from oslo_config import cfg
from oslo_utils import importutils
from ironic.common import exception
from ironic.common import states
from ironic.conductor import task_manager
from ironic.drivers.modules.ucs import helper as ucs_helper
from ironic.drivers.modules.ucs import power as ucs_power
from ironic.tests.conductor import utils as mgr_utils
from ironic.tests.db import base as db_base
from ironic.tests.db import utils as db_utils
from ironic.tests.objects import utils as obj_utils
ucs_error = importutils.try_import('UcsSdk.utils.exception')
INFO_DICT = db_utils.get_test_ucs_info()
CONF = cfg.CONF
class UcsPowerTestCase(db_base.DbTestCase):
def setUp(self):
super(UcsPowerTestCase, self).setUp()
driver_info = INFO_DICT
mgr_utils.mock_the_extension_manager(driver="fake_ucs")
self.node = obj_utils.create_test_node(self.context,
driver='fake_ucs',
driver_info=driver_info)
CONF.set_override('max_retry', 2, 'cisco_ucs')
CONF.set_override('action_interval', 1, 'cisco_ucs')
self.interface = ucs_power.Power()
def test_get_properties(self):
expected = ucs_helper.COMMON_PROPERTIES
expected.update(ucs_helper.COMMON_PROPERTIES)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.assertEqual(expected, task.driver.get_properties())
@mock.patch.object(ucs_helper, 'parse_driver_info',
spec_set=True, autospec=True)
def test_validate(self, mock_parse_driver_info):
mock_parse_driver_info.return_value = {}
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.interface.validate(task)
mock_parse_driver_info.assert_called_once_with(task.node)
@mock.patch.object(ucs_helper, 'parse_driver_info',
spec_set=True, autospec=True)
def test_validate_fail(self, mock_parse_driver_info):
side_effect = exception.InvalidParameterValue('Invalid Input')
mock_parse_driver_info.side_effect = side_effect
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.assertRaises(exception.InvalidParameterValue,
self.interface.validate,
task)
mock_parse_driver_info.assert_called_once_with(task.node)
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test_get_power_state_up(self, mock_power_helper, mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock_power = mock_power_helper.return_value
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
mock_power.get_power_state.return_value = 'up'
self.assertEqual(states.POWER_ON,
self.interface.get_power_state(task))
mock_power.get_power_state.assert_called_once_with()
mock_power.get_power_state.reset_mock()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test_get_power_state_down(self, mock_power_helper, mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock_power = mock_power_helper.return_value
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
mock_power.get_power_state.return_value = 'down'
self.assertEqual(states.POWER_OFF,
self.interface.get_power_state(task))
mock_power.get_power_state.assert_called_once_with()
mock_power.get_power_state.reset_mock()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test_get_power_state_error(self, mock_power_helper, mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock_power = mock_power_helper.return_value
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
mock_power.get_power_state.return_value = states.ERROR
self.assertEqual(states.ERROR,
self.interface.get_power_state(task))
mock_power.get_power_state.assert_called_once_with()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test_get_power_state_fail(self,
mock_ucs_power,
mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
power = mock_ucs_power.return_value
power.get_power_state.side_effect = (
ucs_error.UcsOperationError(operation='getting power state',
error='failed'))
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.assertRaises(exception.UcsOperationError,
self.interface.get_power_state,
task)
power.get_power_state.assert_called_with()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power._wait_for_state_change',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test_set_power_state(self, mock_power_helper, mock__wait, mock_helper):
target_state = states.POWER_ON
mock_power = mock_power_helper.return_value
mock_power.get_power_state.side_effect = ['down', 'up']
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock__wait.return_value = target_state
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertIsNone(self.interface.set_power_state(task,
target_state))
mock_power.set_power_state.assert_called_once_with('up')
mock_power.get_power_state.assert_called_once_with()
mock__wait.assert_called_once_with(target_state, mock_power)
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test_set_power_state_fail(self, mock_power_helper, mock_helper):
mock_power = mock_power_helper.return_value
mock_power.set_power_state.side_effect = (
ucs_error.UcsOperationError(operation='setting power state',
error='failed'))
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.UcsOperationError,
self.interface.set_power_state,
task, states.POWER_OFF)
mock_power.set_power_state.assert_called_once_with('down')
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
def test_set_power_state_invalid_state(self, mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.InvalidParameterValue,
self.interface.set_power_state,
task, states.ERROR)
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test__set_and_wait_for_state_change_already_target_state(
self,
mock_ucs_power,
mock_helper):
mock_power = mock_ucs_power.return_value
target_state = states.POWER_ON
mock_power.get_power_state.return_value = 'up'
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
self.assertEqual(states.POWER_ON,
ucs_power._wait_for_state_change(
target_state, mock_power))
mock_power.get_power_state.assert_called_with()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test__set_and_wait_for_state_change_exceed_iterations(
self,
mock_power_helper,
mock_helper):
mock_power = mock_power_helper.return_value
target_state = states.POWER_ON
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock_power.get_power_state.side_effect = (
['down', 'down', 'down', 'down'])
self.assertEqual(states.ERROR,
ucs_power._wait_for_state_change(
target_state, mock_power)
)
mock_power.get_power_state.assert_called_with()
self.assertEqual(4, mock_power.get_power_state.call_count)
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power._wait_for_state_change',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test_reboot(self, mock_power_helper, mock__wait, mock_helper):
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock_power = mock_power_helper.return_value
mock__wait.return_value = states.POWER_ON
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertIsNone(self.interface.reboot(task))
mock_power.reboot.assert_called_once_with()
@mock.patch('ironic.drivers.modules.ucs.helper.ucs_helper',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power._wait_for_state_change',
spec_set=True, autospec=True)
@mock.patch('ironic.drivers.modules.ucs.power.ucs_power.UcsPower',
spec_set=True, autospec=True)
def test_reboot_fail(self, mock_power_helper, mock__wait,
mock_ucs_helper):
mock_ucs_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock_power = mock_power_helper.return_value
mock_power.reboot.side_effect = (
ucs_error.UcsOperationError(operation='rebooting', error='failed'))
mock__wait.return_value = states.ERROR
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.UcsOperationError,
self.interface.reboot,
task
)
mock_power.reboot.assert_called_once_with()

3
setup.cfg
View File

@ -38,6 +38,7 @@ ironic.drivers =
agent_pyghmi = ironic.drivers.agent:AgentAndIPMINativeDriver
agent_ssh = ironic.drivers.agent:AgentAndSSHDriver
agent_vbox = ironic.drivers.agent:AgentAndVirtualBoxDriver
agent_ucs = ironic.drivers.agent:AgentAndUcsDriver
fake = ironic.drivers.fake:FakeDriver
fake_agent = ironic.drivers.fake:FakeAgentDriver
fake_inspector = ironic.drivers.fake:FakeIPMIToolInspectorDriver
@ -54,6 +55,7 @@ ironic.drivers =
fake_vbox = ironic.drivers.fake:FakeVirtualBoxDriver
fake_amt = ironic.drivers.fake:FakeAMTDriver
fake_msftocs = ironic.drivers.fake:FakeMSFTOCSDriver
fake_ucs = ironic.drivers.fake:FakeUcsDriver
iscsi_ilo = ironic.drivers.ilo:IloVirtualMediaIscsiDriver
pxe_ipmitool = ironic.drivers.pxe:PXEAndIPMIToolDriver
pxe_ipminative = ironic.drivers.pxe:PXEAndIPMINativeDriver
@ -67,6 +69,7 @@ ironic.drivers =
pxe_irmc = ironic.drivers.pxe:PXEAndIRMCDriver
pxe_amt = ironic.drivers.pxe:PXEAndAMTDriver
pxe_msftocs = ironic.drivers.pxe:PXEAndMSFTOCSDriver
pxe_ucs = ironic.drivers.pxe:PXEAndUcsDriver
ironic.database.migration_backend =
sqlalchemy = ironic.db.sqlalchemy.migration