openstack/ironic
Code Issues Proposed changes

Merge "Remove CIMC/UCS drivers"

Browse Source
This commit is contained in:
Zuul 2019-06-28 07:45:26 +00:00 committed by Gerrit Code Review
commit 966f832bfb
30 changed files with 0 additions and 2704 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doc/source/admin/drivers.rst
View File

@ -17,7 +17,6 @@ Hardware Types
.. toctree::
:maxdepth: 1
drivers/cimc
drivers/ibmc
drivers/idrac
drivers/ilo
@ -25,7 +24,6 @@ Hardware Types
drivers/irmc
drivers/redfish
drivers/snmp
drivers/ucs
drivers/xclarity
Changing Hardware Types and Interfaces

93
doc/source/admin/drivers/cimc.rst
View File

@ -1,93 +0,0 @@
.. _CIMC:
===========
CIMC driver
===========
Overview
========
The ``cisco-ucs-standalone`` hardware type targets standalone Cisco UCS C
series servers. It enables you to take advantage of CIMC by using
the python SDK.
The CIMC hardware type can use the Ironic Inspector service for in-band
inspection of equipment. For more information see the `Ironic Inspector
documentation <https://docs.openstack.org/ironic-inspector/latest>`_.
Prerequisites
=============
* ``ImcSdk`` is a python SDK for the CIMC HTTP/HTTPS XML API used to control
CIMC.
Install the ``ImcSdk`` module
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. note::
Install the ``ImcSdk`` module on the Ironic conductor node. Required version is
0.7.2.
#. Install it::
$ pip install "ImcSdk>=0.7.2"
Tested Platforms
~~~~~~~~~~~~~~~~
This driver works with UCS C-Series servers and has been tested with:
* UCS C240M3S
Configuring and Enabling the driver
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
1. Add ``cisco-ucs-standalone`` to ``enabled_hardware_types`` in
``/etc/ironic/ironic.conf``. For example::
enabled_hardware_types = ipmi,cisco-ucs-standalone
2. Restart the Ironic conductor service:
For Ubuntu/Debian systems::
$ sudo service ironic-conductor restart
or for RHEL/CentOS/Fedora::
$ sudo systemctl restart openstack-ironic-conductor
Registering CIMC Managed UCS node in Ironic
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Nodes configured for CIMC driver should have the ``driver`` property set to
``cisco-ucs-standalone``. The following configuration values are also required
in ``driver_info``:
- ``cimc_address``: IP address or hostname for CIMC
- ``cimc_username``: CIMC login user name
- ``cimc_password``: CIMC login password for the above CIMC user.
- ``deploy_kernel``: Identifier for the deployment kernel e.g. a Glance UUID
- ``deploy_ramdisk``: Identifier for the deployment ramdisk e.g. a Glance UUID
The following sequence of commands can be used to enroll a UCS Standalone node.
* Create Node::
openstack baremetal node create --driver cisco-ucs-standalone \
--driver-info cimc_address=<CIMC hostname OR ip-address> \
--driver-info cimc_username=<cimc_username> \
--driver-info cimc_password=<cimc_password> \
--driver-info deploy_kernel=<glance_uuid_of_deploy_kernel> \
--driver-info deploy_ramdisk=<glance_uuid_of_deploy_ramdisk> \
--property cpus=<number_of_cpus> \
--property memory_mb=<memory_size_in_MB> \
--property local_gb=<local_disk_size_in_GB> \
--property cpu_arch=<cpu_arch>
The above command 'openstack baremetal node create' will return UUID of the
node, which is the value of $NODE in the following command.
* Associate port with the node created::
openstack baremetal port create --node $NODE <MAC_address_of_Ucs_server's_NIC>
For more information about enrolling nodes see :ref:`enrollment` in the install guide.

92
doc/source/admin/drivers/ucs.rst
View File

@ -1,92 +0,0 @@
.. _UCS:
==========
UCS driver
==========
Overview
========
The UCS driver is targeted for UCS Manager managed Cisco UCS B/C series
servers. The ``cisco-ucs-managed`` hardware type enables you to take advantage
of UCS Manager by using the python SDK.
The UCS hardware type can use the Ironic Inspector service for in-band
inspection of equipment. For more information see the `Ironic Inspector
documentation <https://docs.openstack.org/ironic-inspector/latest>`_.
Prerequisites
=============
* ``UcsSdk`` is a python package version of XML API sdk available to
manage Cisco UCS Managed B/C-series servers.
Install ``UcsSdk`` [1]_ module on the Ironic conductor node.
Required version is 0.8.2.2::
$ pip install "UcsSdk==0.8.2.2"
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 ``cisco-ucs-managed`` to the ``enabled_hardware_types`` in
``/etc/ironic/ironic.conf``. For example::
enabled_hardware_types = ipmi,cisco-ucs-managed
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
``cisco-ucs-managed``. The following configuration values are also required in
``driver_info``:
- ``ucs_address``: 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::
openstack baremetal node create --driver cisco-ucs-managed \
--driver-info ucs_address=<UCS Manager hostname/ip-address> \
--driver-info ucs_username=<ucsm_username> \
--driver-info ucs_password=<ucsm_password> \
--driver-info ucs_service_profile=<service_profile_dn_being_enrolled> \
--driver-info deploy_kernel=<glance_uuid_of_deploy_kernel> \
--driver-info deploy_ramdisk=<glance_uuid_of_deploy_ramdisk> \
--property cpus=<number_of_cpus> \
--property memory_mb=<memory_size_in_MB> \
--property local_gb=<local_disk_size_in_GB> \
--property cpu_arch=<cpu_arch>
The above command 'openstack baremetal node create' will return UUID of the
node, which is the value of $NODE in the following command.
* Associate port with the node created::
openstack baremetal port create --node $NODE <MAC_address_of_Ucs_server's_NIC>
References
==========
.. [1] UcsSdk - https://pypi.org/project/UcsSdk

4
doc/source/admin/upgrade-to-hardware-types.rst
View File

@ -22,19 +22,15 @@ agent_ilo ilo ilo-virtual-media direct
agent_ipmitool ipmi pxe direct inspector ipmitool ipmitool
agent_ipmitool_socat ipmi pxe direct inspector ipmitool ipmitool
agent_irmc irmc irmc-virtual-media direct irmc irmc irmc
agent_ucs cisco-ucs-managed pxe direct inspector ucsm ucsm
iscsi_ilo ilo ilo-virtual-media iscsi ilo ilo ilo
iscsi_irmc irmc irmc-virtual-media iscsi irmc irmc irmc
pxe_agent_cimc cisco-ucs-standalone pxe direct inspector cimc cimc
pxe_drac idrac pxe iscsi idrac idrac idrac
pxe_drac_inspector idrac pxe iscsi inspector idrac idrac
pxe_ilo ilo ilo-pxe iscsi ilo ilo ilo
pxe_ipmitool ipmi pxe iscsi inspector ipmitool ipmitool
pxe_ipmitool_socat ipmi pxe iscsi inspector ipmitool ipmitool
pxe_iscsi_cimc cisco-ucs-standalone pxe iscsi inspector cimc cimc
pxe_irmc irmc irmc-pxe iscsi irmc irmc irmc
pxe_snmp snmp pxe iscsi no-inspect fake snmp
pxe_ucs cisco-ucs-managed pxe iscsi inspector ucsm ucsm
===================== ==================== ==================== ============== ========== ========== =========
.. note::

4
driver-requirements.txt
View File

@ -8,13 +8,9 @@ proliantutils>=2.7.0
pysnmp>=4.3.0,<5.0.0
python-ironic-inspector-client>=1.5.0
python-scciclient>=0.8.0
UcsSdk==0.8.2.2;python_version<'3'
python-dracclient>=3.0.0,<4.0.0
python-xclarityclient>=0.1.6
# The CIMC drivers use the Cisco IMC SDK version 0.7.2 or greater
ImcSdk>=0.7.2
# The Redfish hardware type uses the Sushy library
sushy>=1.6.0

14
ironic/common/exception.py
View File

@ -561,25 +561,11 @@ class DirectoryNotWritable(IronicException):
_msg_fmt = _("Directory %(dir)s is not writable.")
class UcsOperationError(DriverOperationError):
_msg_fmt = _("Cisco UCS client: operation %(operation)s failed for node"
" %(node)s. Reason: %(error)s")
class UcsConnectionError(IronicException):
_msg_fmt = _("Cisco UCS client: connection failed for node "
"%(node)s. Reason: %(error)s")
class ImageUploadFailed(IronicException):
_msg_fmt = _("Failed to upload %(image_name)s image to web server "
"%(web_server)s, reason: %(reason)s")
class CIMCException(DriverOperationError):
_msg_fmt = _("Cisco IMC exception occurred for node %(node)s: %(error)s")
class NodeTagNotFound(IronicException):
_msg_fmt = _("Node %(node_id)s doesn't have a tag '%(tag)s'")

2
ironic/conf/__init__.py
View File

@ -20,7 +20,6 @@ from ironic.conf import ansible
from ironic.conf import api
from ironic.conf import audit
from ironic.conf import cinder
from ironic.conf import cisco
from ironic.conf import conductor
from ironic.conf import console
from ironic.conf import database
@ -54,7 +53,6 @@ ansible.register_opts(CONF)
api.register_opts(CONF)
audit.register_opts(CONF)
cinder.register_opts(CONF)
cisco.register_opts(CONF)
conductor.register_opts(CONF)
console.register_opts(CONF)
database.register_opts(CONF)

49
ironic/conf/cisco.py
View File

@ -1,49 +0,0 @@
# Copyright 2016 Intel Corporation
# 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.
from oslo_config import cfg
from ironic.common.i18n import _
# NOTE: options for CIMC (Cisco Integrated Management Controller), which talks
# to UCS (Unified Computing System) in standalone mode
cimc_opts = [
cfg.IntOpt('max_retry',
default=6,
help=_('Number of times a power operation needs to be '
'retried')),
cfg.IntOpt('action_interval',
default=10,
help=_('Amount of time in seconds to wait in between power '
'operations')),
]
# NOTE: options for UCSM (UCS Manager), which talks to UCS via a centralized
# management controller
ucsm_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')),
]
def register_opts(conf):
conf.register_opts(cimc_opts, group='cimc')
conf.register_opts(ucsm_opts, group='cisco_ucs')

2
ironic/conf/opts.py
View File

@ -37,9 +37,7 @@ _opts = [
('ansible', ironic.conf.ansible.opts),
('api', ironic.conf.api.opts),
('audit', ironic.conf.audit.opts),
('cimc', ironic.conf.cisco.cimc_opts),
('cinder', ironic.conf.cinder.list_opts()),
('cisco_ucs', ironic.conf.cisco.ucsm_opts),
('conductor', ironic.conf.conductor.opts),
('console', ironic.conf.console.opts),
('database', ironic.conf.database.opts),

61
ironic/drivers/cisco_ucs.py
View File

@ -1,61 +0,0 @@
# 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.
"""
Hardware types for Cisco UCS Servers
"""
from ironic.drivers import ipmi
from ironic.drivers.modules.cimc import management as cimc_mgmt
from ironic.drivers.modules.cimc import power as cimc_power
from ironic.drivers.modules.ucs import management as ucs_mgmt
from ironic.drivers.modules.ucs import power as ucs_power
class CiscoUCSStandalone(ipmi.IPMIHardware):
"""Cisco UCS in standalone mode"""
# NOTE(TheJulia): Deprecated due to a lack of operating third party
# CI, which stopped reporting during the Stein development cycle.
supported = False
@property
def supported_management_interfaces(self):
"""List of supported management interfaces."""
mgmt = super(CiscoUCSStandalone, self).supported_management_interfaces
return [cimc_mgmt.CIMCManagement] + mgmt
@property
def supported_power_interfaces(self):
"""List of supported power interfaces."""
power = super(CiscoUCSStandalone, self).supported_power_interfaces
return [cimc_power.Power] + power
class CiscoUCSManaged(CiscoUCSStandalone):
"""Cisco UCS under UCSM management"""
# NOTE(TheJulia): Deprecated due to a lack of operating third party
# CI, which stopped reporting during the Stein development cycle.
supported = False
@property
def supported_management_interfaces(self):
"""List of supported management interfaces."""
mgmt = super(CiscoUCSManaged, self).supported_management_interfaces
return [ucs_mgmt.UcsManagement] + mgmt
@property
def supported_power_interfaces(self):
"""List of supported power interfaces."""
power = super(CiscoUCSManaged, self).supported_power_interfaces
return [ucs_power.Power] + power

0
ironic/drivers/modules/cimc/__init__.py
View File

85
ironic/drivers/modules/cimc/common.py
View File

@ -1,85 +0,0 @@
# 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.
from contextlib import contextmanager
from oslo_utils import importutils
from ironic.common import exception
from ironic.common.i18n import _
from ironic.drivers.modules import deploy_utils
REQUIRED_PROPERTIES = {
'cimc_address': _('IP or Hostname of the CIMC. Required.'),
'cimc_username': _('CIMC Manager admin username. Required.'),
'cimc_password': _('CIMC Manager password. Required.'),
}
COMMON_PROPERTIES = REQUIRED_PROPERTIES
imcsdk = importutils.try_import('ImcSdk')
def parse_driver_info(node):
"""Parses and creates Cisco driver info.
:param node: An Ironic node object.
:returns: dictionary 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 = (_("%s driver requires these parameters to be set in the "
"node's driver_info.") %
node.driver)
deploy_utils.check_for_missing_params(info, error_msg)
return info
def handle_login(task, handle, info):
"""Login to the CIMC handle.
Run login on the CIMC handle, catching any ImcException and reraising
it as an ironic CIMCException.
:param handle: A CIMC handle.
:param info: A list of driver info as produced by parse_driver_info.
:raises: CIMCException if there error logging in.
"""
try:
handle.login(info['cimc_address'],
info['cimc_username'],
info['cimc_password'])
except imcsdk.ImcException as e:
raise exception.CIMCException(node=task.node.uuid, error=e)
@contextmanager
def cimc_handle(task):
"""Context manager for creating a CIMC handle and logging into it.
:param task: The current task object.
:raises: CIMCException if login fails
:yields: A CIMC Handle for the node in the task.
"""
info = parse_driver_info(task.node)
handle = imcsdk.ImcHandle()
handle_login(task, handle, info)
try:
yield handle
finally:
handle.logout()

168
ironic/drivers/modules/cimc/management.py
View File

@ -1,168 +0,0 @@
# 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.
from oslo_utils import importutils
from ironic.common import boot_devices
from ironic.common import exception
from ironic.drivers import base
from ironic.drivers.modules.cimc import common
imcsdk = importutils.try_import('ImcSdk')
CIMC_TO_IRONIC_BOOT_DEVICE = {
'storage-read-write': boot_devices.DISK,
'lan-read-only': boot_devices.PXE,
'vm-read-only': boot_devices.CDROM
}
IRONIC_TO_CIMC_BOOT_DEVICE = {
boot_devices.DISK: ('lsbootStorage', 'storage-read-write',
'storage', 'read-write'),
boot_devices.PXE: ('lsbootLan', 'lan-read-only',
'lan', 'read-only'),
boot_devices.CDROM: ('lsbootVirtualMedia', 'vm-read-only',
'virtual-media', 'read-only')
}
class CIMCManagement(base.ManagementInterface):
# NOTE(TheJulia): Deprecated due to a lack of operating third party
# CI, which stopped reporting during the Stein development cycle.
supported = False
def get_properties(self):
"""Return the properties of the interface.
:returns: dictionary of <property name>:<property description> entries.
"""
return common.COMMON_PROPERTIES
def validate(self, task):
"""Check if node.driver_info contains the required CIMC credentials.
:param task: a TaskManager instance.
:raises: InvalidParameterValue if required CIMC credentials are
missing.
"""
common.parse_driver_info(task.node)
def get_supported_boot_devices(self, task):
"""Get a list of the supported boot devices.
:param task: a task from TaskManager.
:returns: A list with the supported boot devices defined
in :mod:`ironic.common.boot_devices`.
"""
return list(CIMC_TO_IRONIC_BOOT_DEVICE.values())
def get_boot_device(self, task):
"""Get the current boot device for a node.
Provides the current boot device of the node. Be aware that not
all drivers support this.
:param task: a task from TaskManager.
:raises: MissingParameterValue if a required parameter is missing
:raises: CIMCException if there is an error from CIMC
:returns: a dictionary containing:
:boot_device:
the boot device, one of :mod:`ironic.common.boot_devices` or
None if it is unknown.
:persistent:
Whether the boot device will persist to all future boots or
not, None if it is unknown.
"""
with common.cimc_handle(task) as handle:
method = imcsdk.ImcCore.ExternalMethod("ConfigResolveClass")
method.Cookie = handle.cookie
method.InDn = "sys/rack-unit-1"
method.InHierarchical = "true"
method.ClassId = "lsbootDef"
try:
resp = handle.xml_query(method, imcsdk.WriteXmlOption.DIRTY)
except imcsdk.ImcException as e:
raise exception.CIMCException(node=task.node.uuid, error=e)
error = getattr(resp, 'error_code', None)
if error:
raise exception.CIMCException(node=task.node.uuid, error=error)
bootDevs = resp.OutConfigs.child[0].child
first_device = None
for dev in bootDevs:
try:
if int(dev.Order) == 1:
first_device = dev
break
except (ValueError, AttributeError):
pass
boot_device = (CIMC_TO_IRONIC_BOOT_DEVICE.get(
first_device.Rn) if first_device else None)
# Every boot device in CIMC is persistent right now
persistent = True if boot_device else None
return {'boot_device': boot_device, 'persistent': persistent}
def set_boot_device(self, task, device, persistent=True):
"""Set the boot device for a 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
:mod:`ironic.common.boot_devices`.
:param persistent: Every boot device in CIMC is persistent right now,
so this value is ignored.
:raises: InvalidParameterValue if an invalid boot device is
specified.
:raises: MissingParameterValue if a required parameter is missing
:raises: CIMCException if there is an error from CIMC
"""
with common.cimc_handle(task) as handle:
dev = IRONIC_TO_CIMC_BOOT_DEVICE[device]
method = imcsdk.ImcCore.ExternalMethod("ConfigConfMo")
method.Cookie = handle.cookie
method.Dn = "sys/rack-unit-1/boot-policy"
method.InHierarchical = "true"
config = imcsdk.Imc.ConfigConfig()
bootMode = imcsdk.ImcCore.ManagedObject(dev[0])
bootMode.set_attr("access", dev[3])
bootMode.set_attr("type", dev[2])
bootMode.set_attr("Rn", dev[1])
bootMode.set_attr("order", "1")
config.add_child(bootMode)
method.InConfig = config
try:
resp = handle.xml_query(method, imcsdk.WriteXmlOption.DIRTY)
except imcsdk.ImcException as e:
raise exception.CIMCException(node=task.node.uuid, error=e)
error = getattr(resp, 'error_code')
if error:
raise exception.CIMCException(node=task.node.uuid, error=error)
def get_sensors_data(self, task):
raise NotImplementedError()

191
ironic/drivers/modules/cimc/power.py
View File

@ -1,191 +0,0 @@
# 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.
from oslo_log import log as logging
from oslo_service import loopingcall
from oslo_utils import importutils
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common import states
from ironic.conductor import task_manager
from ironic.conf import CONF
from ironic.drivers import base
from ironic.drivers.modules.cimc import common
imcsdk = importutils.try_import('ImcSdk')
LOG = logging.getLogger(__name__)
if imcsdk:
CIMC_TO_IRONIC_POWER_STATE = {
imcsdk.ComputeRackUnit.CONST_OPER_POWER_ON: states.POWER_ON,
imcsdk.ComputeRackUnit.CONST_OPER_POWER_OFF: states.POWER_OFF,
}
IRONIC_TO_CIMC_POWER_STATE = {
states.POWER_ON: imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_UP,
states.POWER_OFF: imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_DOWN,
states.REBOOT:
imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_HARD_RESET_IMMEDIATE
}
def _wait_for_state_change(target_state, task):
"""Wait and check for the power state change
:param target_state: The target state we are waiting for.
:param task: a TaskManager instance containing the node to act on.
:raises: CIMCException if there is an error communicating with CIMC
"""
store = {'state': None, 'retries': CONF.cimc.max_retry}
def _wait(store):
current_power_state = None
with common.cimc_handle(task) as handle:
try:
rack_unit = handle.get_imc_managedobject(
None, None, params={"Dn": "sys/rack-unit-1"}
)
except imcsdk.ImcException as e:
raise exception.CIMCException(node=task.node.uuid, error=e)
else:
current_power_state = rack_unit[0].get_attr("OperPower")
store['state'] = CIMC_TO_IRONIC_POWER_STATE.get(current_power_state)
if store['state'] == target_state:
raise loopingcall.LoopingCallDone()
store['retries'] -= 1
if store['retries'] <= 0:
store['state'] = states.ERROR
raise loopingcall.LoopingCallDone()
timer = loopingcall.FixedIntervalLoopingCall(_wait, store)
timer.start(interval=CONF.cimc.action_interval).wait()
return store['state']
class Power(base.PowerInterface):
# NOTE(TheJulia): Deprecated due to a lack of operating third party
# CI, which stopped reporting during the Stein development cycle.
supported = False
def get_properties(self):
"""Return the properties of the interface.
:returns: dictionary of <property name>:<property description> entries.
"""
return common.COMMON_PROPERTIES
def validate(self, task):
"""Check if node.driver_info contains the required CIMC credentials.
:param task: a TaskManager instance.
:raises: InvalidParameterValue if required CIMC credentials are
missing.
"""
common.parse_driver_info(task.node)
def get_power_state(self, task):
"""Return the power state of the task's node.
:param task: a TaskManager instance containing the node to act on.
:raises: MissingParameterValue if a required parameter is missing.
:returns: a power state. One of :mod:`ironic.common.states`.
:raises: CIMCException if there is an error communicating with CIMC
"""
current_power_state = None
with common.cimc_handle(task) as handle:
try:
rack_unit = handle.get_imc_managedobject(
None, None, params={"Dn": "sys/rack-unit-1"}
)
except imcsdk.ImcException as e:
raise exception.CIMCException(node=task.node.uuid, error=e)
else:
current_power_state = rack_unit[0].get_attr("OperPower")
return CIMC_TO_IRONIC_POWER_STATE.get(current_power_state,
states.ERROR)
@task_manager.require_exclusive_lock
def set_power_state(self, task, pstate, timeout=None):
"""Set the power state of the task's node.
:param task: a TaskManager instance containing the node to act on.
:param pstate: Any power state from :mod:`ironic.common.states`.
:param timeout: timeout (in seconds). Unsupported by this interface.
:raises: MissingParameterValue if a required parameter is missing.
:raises: InvalidParameterValue if an invalid power state is passed
:raises: CIMCException if there is an error communicating with CIMC
"""
# TODO(rloo): Support timeouts!
if timeout is not None:
LOG.warning(
"The 'cimc' Power Interface's 'set_power_state' method "
"doesn't support the 'timeout' parameter. Ignoring "
"timeout=%(timeout)s",
{'timeout': timeout})
if pstate not in IRONIC_TO_CIMC_POWER_STATE:
msg = _("set_power_state called for %(node)s with "
"invalid state %(state)s")
raise exception.InvalidParameterValue(
msg % {"node": task.node.uuid, "state": pstate})
with common.cimc_handle(task) as handle:
try:
handle.set_imc_managedobject(
None, class_id="ComputeRackUnit",
params={
imcsdk.ComputeRackUnit.ADMIN_POWER:
IRONIC_TO_CIMC_POWER_STATE[pstate],
imcsdk.ComputeRackUnit.DN: "sys/rack-unit-1"
})
except imcsdk.ImcException as e:
raise exception.CIMCException(node=task.node.uuid, error=e)
if pstate is states.REBOOT:
pstate = states.POWER_ON
state = _wait_for_state_change(pstate, task)
if state != pstate:
raise exception.PowerStateFailure(pstate=pstate)
@task_manager.require_exclusive_lock
def reboot(self, task, timeout=None):
"""Perform a hard reboot of the task's node.
If the node is already powered on then it shall reboot the node, if
its off then the node will just be turned on.
:param task: a TaskManager instance containing the node to act on.
:param timeout: timeout (in seconds). Unsupported by this interface.
:raises: MissingParameterValue if a required parameter is missing.
:raises: CIMCException if there is an error communicating with CIMC
"""
# TODO(rloo): Support timeouts!
if timeout is not None:
LOG.warning("The 'cimc' Power Interface's 'reboot' method "
"doesn't support the 'timeout' parameter. Ignoring "
"timeout=%(timeout)s",
{'timeout': timeout})
current_power_state = self.get_power_state(task)
if current_power_state == states.POWER_ON:
self.set_power_state(task, states.REBOOT)
elif current_power_state == states.POWER_OFF:
self.set_power_state(task, states.POWER_ON)

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

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

@ -1,126 +0,0 @@
# 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
"""
from oslo_log import log as logging
from oslo_utils import importutils
import six
from ironic.common import exception
from ironic.common.i18n import _
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.
"""
@six.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 = (_("%s driver requires these parameters to be set in the "
"node's driver_info.") %
node.driver)
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 fails 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("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()

152
ironic/drivers/modules/ucs/management.py
View File

@ -1,152 +0,0 @@
# 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.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,
'disk': boot_devices.DISK,
'pxe': boot_devices.PXE,
'read-only-vm': boot_devices.CDROM,
'cdrom': boot_devices.CDROM
}
class UcsManagement(base.ManagementInterface):
# NOTE(TheJulia): Deprecated due to a lack of operating third party
# CI, which stopped reporting during the Stein development cycle.
supported = False
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, task):
"""Get a list of the supported boot devices.
:param task: a task from TaskManager.
:returns: A list with the supported boot devices defined
in :mod:`ironic.common.boot_devices`.
"""
return list(set(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("%(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("%(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()

220
ironic/drivers/modules/ucs/power.py
View File

@ -1,220 +0,0 @@
# 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_log import log as logging
from oslo_service import loopingcall
from oslo_utils import importutils
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common import states
from ironic.conductor import task_manager
from ironic.conf import CONF
from ironic.drivers import base
from ironic.drivers.modules.ucs import helper as ucs_helper
ucs_power = importutils.try_import('UcsSdk.utils.power')
ucs_error = importutils.try_import('UcsSdk.utils.exception')
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.
"""
# NOTE(TheJulia): Deprecated due to a lack of operating third party
# CI, which stopped reporting during the Stein development cycle.
supported = False
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("%(driver)s: get_power_state operation failed for "
"node %(uuid)s with error: %(msg)s.",
{'driver': task.node.driver, 'uuid': task.node.uuid,
'msg': ucs_exception})
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, timeout=None, 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 timeout: timeout (in seconds). Unsupported by this interface.
: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.
"""
# TODO(rloo): Support timeouts!
if timeout is not None:
LOG.warning(
"The 'ucsm' Power Interface's 'set_power_state' method "
"doesn't support the 'timeout' parameter. Ignoring "
"timeout=%(timeout)s",
{'timeout': timeout})
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("%(driver)s: set_power_state operation failed for "
"node %(uuid)s with error: %(msg)s.",
{'driver': task.node.driver, 'uuid': task.node.uuid,
'msg': ucs_exception})
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("%(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, timeout=None, helper=None):
"""Cycles the power to a node.
:param task: a TaskManager instance.
:param timeout: timeout (in seconds). Unsupported by this interface.
: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.
"""
# TODO(rloo): Support timeouts!
if timeout is not None:
LOG.warning("The 'ucsm' Power Interface's 'reboot' method "
"doesn't support the 'timeout' parameter. Ignoring "
"timeout=%(timeout)s",
{'timeout': timeout})
try:
ucs_power_handle = ucs_power.UcsPower(helper)
ucs_power_handle.reboot()
except ucs_error.UcsOperationError as ucs_exception:
LOG.error("%(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("%(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)

17
ironic/tests/unit/db/utils.py
View File

@ -416,23 +416,6 @@ def create_test_conductor(**kw):
return dbapi.register_conductor(conductor)
def get_test_ucs_info():
return {
"ucs_username": "admin",
"ucs_password": "password",
"ucs_service_profile": "org-root/ls-devstack",
"ucs_address": "ucs-b",
}
def get_test_cimc_info():
return {
"cimc_username": "admin",
"cimc_password": "password",
"cimc_address": "1.2.3.4",
}
def get_test_redfish_info():
return {
"redfish_address": "https://example.com",

0
ironic/tests/unit/drivers/modules/cimc/__init__.py
View File

128
ironic/tests/unit/drivers/modules/cimc/test_common.py
View File

@ -1,128 +0,0 @@
# 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.
import mock
from oslo_config import cfg
from oslo_utils import importutils
from oslo_utils import uuidutils
from ironic.common import exception
from ironic.conductor import task_manager
from ironic.drivers.modules.cimc import common as cimc_common
from ironic.tests.unit.db import base as db_base
from ironic.tests.unit.db import utils as db_utils
from ironic.tests.unit.objects import utils as obj_utils
INFO_DICT = db_utils.get_test_cimc_info()
imcsdk = importutils.try_import('ImcSdk')
CONF = cfg.CONF
class CIMCBaseTestCase(db_base.DbTestCase):
def setUp(self):
super(CIMCBaseTestCase, self).setUp()
self.config(enabled_hardware_types=['cisco-ucs-standalone'],
enabled_power_interfaces=['cimc', 'fake'],
enabled_management_interfaces=['cimc', 'fake'])
self.node = obj_utils.create_test_node(
self.context,
driver='cisco-ucs-standalone',
driver_info=INFO_DICT,
instance_uuid=uuidutils.generate_uuid())
CONF.set_override('max_retry', 2, 'cimc')
CONF.set_override('action_interval', 0, 'cimc')
class ParseDriverInfoTestCase(CIMCBaseTestCase):
def test_parse_driver_info(self):
info = cimc_common.parse_driver_info(self.node)
self.assertEqual(INFO_DICT['cimc_address'], info['cimc_address'])
self.assertEqual(INFO_DICT['cimc_username'], info['cimc_username'])
self.assertEqual(INFO_DICT['cimc_password'], info['cimc_password'])
def test_parse_driver_info_missing_address(self):
del self.node.driver_info['cimc_address']
self.assertRaises(exception.MissingParameterValue,
cimc_common.parse_driver_info, self.node)
def test_parse_driver_info_missing_username(self):
del self.node.driver_info['cimc_username']
self.assertRaises(exception.MissingParameterValue,
cimc_common.parse_driver_info, self.node)
def test_parse_driver_info_missing_password(self):
del self.node.driver_info['cimc_password']
self.assertRaises(exception.MissingParameterValue,
cimc_common.parse_driver_info, self.node)
@mock.patch.object(cimc_common, 'cimc_handle', autospec=True)
class CIMCHandleLogin(CIMCBaseTestCase):
def test_cimc_handle_login(self, mock_handle):
info = cimc_common.parse_driver_info(self.node)
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
cimc_common.handle_login(task, handle, info)
handle.login.assert_called_once_with(
self.node.driver_info['cimc_address'],
self.node.driver_info['cimc_username'],
self.node.driver_info['cimc_password'])
def test_cimc_handle_login_exception(self, mock_handle):
info = cimc_common.parse_driver_info(self.node)
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
handle.login.side_effect = imcsdk.ImcException('Boom')
self.assertRaises(exception.CIMCException,
cimc_common.handle_login,
task, handle, info)
handle.login.assert_called_once_with(
self.node.driver_info['cimc_address'],
self.node.driver_info['cimc_username'],
self.node.driver_info['cimc_password'])
class CIMCHandleTestCase(CIMCBaseTestCase):
@mock.patch.object(imcsdk, 'ImcHandle', autospec=True)
@mock.patch.object(cimc_common, 'handle_login', autospec=True)
def test_cimc_handle(self, mock_login, mock_handle):
mo_hand = mock.MagicMock()
mo_hand.username = self.node.driver_info['cimc_username']
mo_hand.password = self.node.driver_info['cimc_password']
mo_hand.name = self.node.driver_info['cimc_address']
mock_handle.return_value = mo_hand
info = cimc_common.parse_driver_info(self.node)
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with cimc_common.cimc_handle(task) as handle:
self.assertEqual(handle, mock_handle.return_value)
mock_login.assert_called_once_with(task, mock_handle.return_value,
info)
mock_handle.return_value.logout.assert_called_once_with()

127
ironic/tests/unit/drivers/modules/cimc/test_management.py
View File

@ -1,127 +0,0 @@
# 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.
import mock
from oslo_utils import importutils
from six.moves import http_client
from ironic.common import boot_devices
from ironic.common import exception
from ironic.conductor import task_manager
from ironic.drivers.modules.cimc import common
from ironic.tests.unit.drivers.modules.cimc import test_common
imcsdk = importutils.try_import('ImcSdk')
@mock.patch.object(common, 'cimc_handle', autospec=True)
class CIMCManagementTestCase(test_common.CIMCBaseTestCase):
def test_get_properties(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertEqual(common.COMMON_PROPERTIES,
task.driver.management.get_properties())
@mock.patch.object(common, "parse_driver_info", autospec=True)
def test_validate(self, mock_driver_info, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
task.driver.management.validate(task)
mock_driver_info.assert_called_once_with(task.node)
def test_get_supported_boot_devices(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
expected = [boot_devices.PXE, boot_devices.DISK,
boot_devices.CDROM]
result = task.driver.management.get_supported_boot_devices(task)
self.assertEqual(sorted(expected), sorted(result))
def test_get_boot_device(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
handle.xml_query.return_value.error_code = None
mock_dev = mock.MagicMock()
mock_dev.Order = 1
mock_dev.Rn = 'storage-read-write'
handle.xml_query().OutConfigs.child[0].child = [mock_dev]
device = task.driver.management.get_boot_device(task)
self.assertEqual(
{'boot_device': boot_devices.DISK, 'persistent': True},
device)
def test_get_boot_device_fail(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
handle.xml_query.return_value.error_code = None
mock_dev = mock.MagicMock()
mock_dev.Order = 1
mock_dev.Rn = 'storage-read-write'
handle.xml_query().OutConfigs.child[0].child = [mock_dev]
device = task.driver.management.get_boot_device(task)
self.assertEqual(
{'boot_device': boot_devices.DISK, 'persistent': True},
device)
def test_set_boot_device(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
handle.xml_query.return_value.error_code = None
task.driver.management.set_boot_device(task, boot_devices.DISK)
method = imcsdk.ImcCore.ExternalMethod("ConfigConfMo")
method.Cookie = handle.cookie
method.Dn = "sys/rack-unit-1/boot-policy"
method.InHierarchical = "true"
config = imcsdk.Imc.ConfigConfig()
bootMode = imcsdk.ImcCore.ManagedObject('lsbootStorage')
bootMode.set_attr("access", 'read-write')
bootMode.set_attr("type", 'storage')
bootMode.set_attr("Rn", 'storage-read-write')
bootMode.set_attr("order", "1")
config.add_child(bootMode)
method.InConfig = config
handle.xml_query.assert_called_once_with(
method, imcsdk.WriteXmlOption.DIRTY)
def test_set_boot_device_fail(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
method = imcsdk.ImcCore.ExternalMethod("ConfigConfMo")
handle.xml_query.return_value.error_code = (
str(http_client.NOT_FOUND))
self.assertRaises(exception.CIMCException,
task.driver.management.set_boot_device,
task, boot_devices.DISK)
handle.xml_query.assert_called_once_with(
method, imcsdk.WriteXmlOption.DIRTY)
def test_get_sensors_data(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(NotImplementedError,
task.driver.management.get_sensors_data, task)

345
ironic/tests/unit/drivers/modules/cimc/test_power.py
View File

@ -1,345 +0,0 @@
# 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.
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.cimc import common
from ironic.drivers.modules.cimc import power
from ironic.tests.unit.drivers.modules.cimc import test_common
imcsdk = importutils.try_import('ImcSdk')
CONF = cfg.CONF
@mock.patch.object(common, 'cimc_handle', autospec=True)
class WaitForStateChangeTestCase(test_common.CIMCBaseTestCase):
def setUp(self):
super(WaitForStateChangeTestCase, self).setUp()
CONF.set_override('max_retry', 2, 'cimc')
CONF.set_override('action_interval', 0, 'cimc')
def test__wait_for_state_change(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
mock_rack_unit = mock.MagicMock()
mock_rack_unit.get_attr.return_value = (
imcsdk.ComputeRackUnit.CONST_OPER_POWER_ON)
handle.get_imc_managedobject.return_value = [mock_rack_unit]
state = power._wait_for_state_change(states.POWER_ON, task)
handle.get_imc_managedobject.assert_called_once_with(
None, None, params={"Dn": "sys/rack-unit-1"})
self.assertEqual(state, states.POWER_ON)
def test__wait_for_state_change_fail(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
mock_rack_unit = mock.MagicMock()
mock_rack_unit.get_attr.return_value = (
imcsdk.ComputeRackUnit.CONST_OPER_POWER_OFF)
handle.get_imc_managedobject.return_value = [mock_rack_unit]
state = power._wait_for_state_change(states.POWER_ON, task)
calls = [
mock.call(None, None, params={"Dn": "sys/rack-unit-1"}),
mock.call(None, None, params={"Dn": "sys/rack-unit-1"})
]
handle.get_imc_managedobject.assert_has_calls(calls)
self.assertEqual(state, states.ERROR)
def test__wait_for_state_change_imc_exception(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
handle.get_imc_managedobject.side_effect = (
imcsdk.ImcException('Boom'))
self.assertRaises(
exception.CIMCException,
power._wait_for_state_change, states.POWER_ON, task)
handle.get_imc_managedobject.assert_called_once_with(
None, None, params={"Dn": "sys/rack-unit-1"})
@mock.patch.object(common, 'cimc_handle', autospec=True)
class PowerTestCase(test_common.CIMCBaseTestCase):
def test_get_properties(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertEqual(common.COMMON_PROPERTIES,
task.driver.power.get_properties())
@mock.patch.object(common, "parse_driver_info", autospec=True)
def test_validate(self, mock_driver_info, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
task.driver.power.validate(task)
mock_driver_info.assert_called_once_with(task.node)
def test_get_power_state(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
mock_rack_unit = mock.MagicMock()
mock_rack_unit.get_attr.return_value = (
imcsdk.ComputeRackUnit.CONST_OPER_POWER_ON)
handle.get_imc_managedobject.return_value = [mock_rack_unit]
state = task.driver.power.get_power_state(task)
handle.get_imc_managedobject.assert_called_once_with(
None, None, params={"Dn": "sys/rack-unit-1"})
self.assertEqual(states.POWER_ON, state)
def test_get_power_state_fail(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
mock_rack_unit = mock.MagicMock()
mock_rack_unit.get_attr.return_value = (
imcsdk.ComputeRackUnit.CONST_OPER_POWER_ON)
handle.get_imc_managedobject.side_effect = (
imcsdk.ImcException("boom"))
self.assertRaises(exception.CIMCException,
task.driver.power.get_power_state, task)
handle.get_imc_managedobject.assert_called_once_with(
None, None, params={"Dn": "sys/rack-unit-1"})
def test_set_power_state_invalid_state(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.InvalidParameterValue,
task.driver.power.set_power_state,
task, states.ERROR)
def test_set_power_state_reboot_ok(self, mock_handle):
hri = imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_HARD_RESET_IMMEDIATE
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
mock_rack_unit = mock.MagicMock()
mock_rack_unit.get_attr.side_effect = [
imcsdk.ComputeRackUnit.CONST_OPER_POWER_OFF,
imcsdk.ComputeRackUnit.CONST_OPER_POWER_ON
]
handle.get_imc_managedobject.return_value = [mock_rack_unit]
task.driver.power.set_power_state(task, states.REBOOT)
handle.set_imc_managedobject.assert_called_once_with(
None, class_id="ComputeRackUnit",
params={
imcsdk.ComputeRackUnit.ADMIN_POWER: hri,
imcsdk.ComputeRackUnit.DN: "sys/rack-unit-1"
})
handle.get_imc_managedobject.assert_called_with(
None, None, params={"Dn": "sys/rack-unit-1"})
def test_set_power_state_reboot_fail(self, mock_handle):
hri = imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_HARD_RESET_IMMEDIATE
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
handle.get_imc_managedobject.side_effect = (
imcsdk.ImcException("boom"))
self.assertRaises(exception.CIMCException,
task.driver.power.set_power_state,
task, states.REBOOT)
handle.set_imc_managedobject.assert_called_once_with(
None, class_id="ComputeRackUnit",
params={
imcsdk.ComputeRackUnit.ADMIN_POWER: hri,
imcsdk.ComputeRackUnit.DN: "sys/rack-unit-1"
})
handle.get_imc_managedobject.assert_called_with(
None, None, params={"Dn": "sys/rack-unit-1"})
@mock.patch.object(power.LOG, 'warning')
def test_set_power_state_on_ok(self, mock_log, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
mock_rack_unit = mock.MagicMock()
mock_rack_unit.get_attr.side_effect = [
imcsdk.ComputeRackUnit.CONST_OPER_POWER_OFF,
imcsdk.ComputeRackUnit.CONST_OPER_POWER_ON
]
handle.get_imc_managedobject.return_value = [mock_rack_unit]
task.driver.power.set_power_state(task, states.POWER_ON)
handle.set_imc_managedobject.assert_called_once_with(
None, class_id="ComputeRackUnit",
params={
imcsdk.ComputeRackUnit.ADMIN_POWER:
imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_UP,
imcsdk.ComputeRackUnit.DN: "sys/rack-unit-1"
})
handle.get_imc_managedobject.assert_called_with(
None, None, params={"Dn": "sys/rack-unit-1"})
self.assertFalse(mock_log.called)
def test_set_power_state_on_fail(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
handle.get_imc_managedobject.side_effect = (
imcsdk.ImcException("boom"))
self.assertRaises(exception.CIMCException,
task.driver.power.set_power_state,
task, states.POWER_ON)
handle.set_imc_managedobject.assert_called_once_with(
None, class_id="ComputeRackUnit",
params={
imcsdk.ComputeRackUnit.ADMIN_POWER:
imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_UP,
imcsdk.ComputeRackUnit.DN: "sys/rack-unit-1"
})
handle.get_imc_managedobject.assert_called_with(
None, None, params={"Dn": "sys/rack-unit-1"})
@mock.patch.object(power.LOG, 'warning')
def test_set_power_state_on_timeout(self, mock_log, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
mock_rack_unit = mock.MagicMock()
mock_rack_unit.get_attr.side_effect = [
imcsdk.ComputeRackUnit.CONST_OPER_POWER_OFF,
imcsdk.ComputeRackUnit.CONST_OPER_POWER_ON
]
handle.get_imc_managedobject.return_value = [mock_rack_unit]
task.driver.power.set_power_state(task, states.POWER_ON,
timeout=10)
handle.set_imc_managedobject.assert_called_once_with(
None, class_id="ComputeRackUnit",
params={
imcsdk.ComputeRackUnit.ADMIN_POWER:
imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_UP,
imcsdk.ComputeRackUnit.DN: "sys/rack-unit-1"
})
handle.get_imc_managedobject.assert_called_with(
None, None, params={"Dn": "sys/rack-unit-1"})
self.assertTrue(mock_log.called)
def test_set_power_state_off_ok(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
mock_rack_unit = mock.MagicMock()
mock_rack_unit.get_attr.side_effect = [
imcsdk.ComputeRackUnit.CONST_OPER_POWER_ON,
imcsdk.ComputeRackUnit.CONST_OPER_POWER_OFF
]
handle.get_imc_managedobject.return_value = [mock_rack_unit]
task.driver.power.set_power_state(task, states.POWER_OFF)
handle.set_imc_managedobject.assert_called_once_with(
None, class_id="ComputeRackUnit",
params={
imcsdk.ComputeRackUnit.ADMIN_POWER:
imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_DOWN,
imcsdk.ComputeRackUnit.DN: "sys/rack-unit-1"
})
handle.get_imc_managedobject.assert_called_with(
None, None, params={"Dn": "sys/rack-unit-1"})
def test_set_power_state_off_fail(self, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
with mock_handle(task) as handle:
handle.get_imc_managedobject.side_effect = (
imcsdk.ImcException("boom"))
self.assertRaises(exception.CIMCException,
task.driver.power.set_power_state,
task, states.POWER_OFF)
handle.set_imc_managedobject.assert_called_once_with(
None, class_id="ComputeRackUnit",
params={
imcsdk.ComputeRackUnit.ADMIN_POWER:
imcsdk.ComputeRackUnit.CONST_ADMIN_POWER_DOWN,
imcsdk.ComputeRackUnit.DN: "sys/rack-unit-1"
})
handle.get_imc_managedobject.assert_called_with(
None, None, params={"Dn": "sys/rack-unit-1"})
@mock.patch.object(power.LOG, 'warning')
@mock.patch.object(power.Power, "set_power_state", autospec=True)
@mock.patch.object(power.Power, "get_power_state", autospec=True)
def test_reboot_on(self, mock_get_state, mock_set_state, mock_log,
mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
mock_get_state.return_value = states.POWER_ON
task.driver.power.reboot(task)
mock_set_state.assert_called_with(mock.ANY, task, states.REBOOT)
self.assertFalse(mock_log.called)
@mock.patch.object(power.Power, "set_power_state", autospec=True)
@mock.patch.object(power.Power, "get_power_state", autospec=True)
def test_reboot_off(self, mock_get_state, mock_set_state, mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
mock_get_state.return_value = states.POWER_OFF
task.driver.power.reboot(task)
mock_set_state.assert_called_with(mock.ANY, task, states.POWER_ON)
@mock.patch.object(power.LOG, 'warning')
@mock.patch.object(power.Power, "set_power_state", autospec=True)
@mock.patch.object(power.Power, "get_power_state", autospec=True)
def test_reboot_on_timeout(self, mock_get_state, mock_set_state, mock_log,
mock_handle):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
mock_get_state.return_value = states.POWER_ON
task.driver.power.reboot(task, timeout=30)
mock_set_state.assert_called_with(mock.ANY, task, states.REBOOT)
self.assertTrue(mock_log.called)

0
ironic/tests/unit/drivers/modules/ucs/__init__.py
View File

163
ironic/tests/unit/drivers/modules/ucs/test_helper.py
View File

@ -1,163 +0,0 @@
# 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_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.unit.db import base as db_base
from ironic.tests.unit.db import utils as db_utils
from ironic.tests.unit.objects import utils as obj_utils
ucs_error = importutils.try_import('UcsSdk.utils.exception')
class BaseUcsTest(db_base.DbTestCase):
def setUp(self):
super(BaseUcsTest, self).setUp()
self.config(enabled_hardware_types=['cisco-ucs-managed',
'fake-hardware'],
enabled_power_interfaces=['ucsm', 'fake'],
enabled_management_interfaces=['ucsm', 'fake'])
self.info = db_utils.get_test_ucs_info()
self.node = obj_utils.create_test_node(self.context,
driver='cisco-ucs-managed',
driver_info=self.info)
class UcsValidateParametersTestCase(BaseUcsTest):
def setUp(self):
super(UcsValidateParametersTestCase, self).setUp()
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.assertEqual(self.info['ucs_address'], info['ucs_address'])
self.assertEqual(self.info['ucs_username'], info['ucs_username'])
self.assertEqual(self.info['ucs_password'], info['ucs_password'])
self.assertEqual(self.info['ucs_service_profile'],
info['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(BaseUcsTest):
def setUp(self):
super(UcsCommonMethodsTestcase, self).setUp()
self.dbapi = dbapi.get_instance()
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_value = "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()

130
ironic/tests/unit/drivers/modules/ucs/test_management.py
View File

@ -1,130 +0,0 @@
# 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_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.unit.drivers.modules.ucs import test_helper
ucs_error = importutils.try_import('UcsSdk.utils.exception')
class UcsManagementTestCase(test_helper.BaseUcsTest):
def setUp(self):
super(UcsManagementTestCase, self).setUp()
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):
with task_manager.acquire(self.context, self.node.uuid) as task:
expected = [boot_devices.PXE, boot_devices.DISK,
boot_devices.CDROM]
self.assertEqual(
sorted(expected),
sorted(self.interface.get_supported_boot_devices(task)))
@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': 'disk',
'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',
)
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)

346
ironic/tests/unit/drivers/modules/ucs/test_power.py
View File

@ -1,346 +0,0 @@
# 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 import fake
from ironic.drivers.modules.ucs import helper as ucs_helper
from ironic.drivers.modules.ucs import power as ucs_power
from ironic.tests.unit.drivers.modules.ucs import test_helper
ucs_error = importutils.try_import('UcsSdk.utils.exception')
CONF = cfg.CONF
class UcsPowerTestCase(test_helper.BaseUcsTest):
def setUp(self):
super(UcsPowerTestCase, self).setUp()
CONF.set_override('max_retry', 2, 'cisco_ucs')
CONF.set_override('action_interval', 0, '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:
# Remove properties from boot and deploy interfaces
task.driver.boot = fake.FakeBoot()
task.driver.deploy = fake.FakeDeploy()
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.object(ucs_power.LOG, 'warning')
@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,
mock_log):
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)
self.assertFalse(mock_log.called)
@mock.patch.object(ucs_power.LOG, 'warning')
@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_timeout(self, mock_power_helper, mock__wait,
mock_helper, mock_log):
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,
timeout=23))
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)
self.assertTrue(mock_log.called)
@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__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__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_set_and_wait_for_state_change_fail(
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.return_value = 'down'
mock_helper.generate_ucsm_handle.return_value = (True, mock.Mock())
mock__wait.return_value = states.POWER_OFF
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.PowerStateFailure,
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.object(ucs_power.LOG, 'warning')
@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_log):
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()
self.assertFalse(mock_log.called)
@mock.patch.object(ucs_power.LOG, 'warning')
@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_timeout(self, mock_power_helper, mock__wait, mock_helper,
mock_log):
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, timeout=88))
mock_power.reboot.assert_called_once_with()
self.assertTrue(mock_log.called)
@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_reboot_fail(self, mock_power_helper,
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'))
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()
@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__wait_state_change_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__wait.return_value = states.ERROR
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.PowerStateFailure,
self.interface.reboot,
task)
mock_power.reboot.assert_called_once_with()

151
ironic/tests/unit/drivers/test_cisco.py
View File

@ -1,151 +0,0 @@
# 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 ironic.conductor import task_manager
from ironic.drivers.modules import agent
from ironic.drivers.modules.cimc import management as cimc_mgmt
from ironic.drivers.modules.cimc import power as cimc_power
from ironic.drivers.modules import ipmitool
from ironic.drivers.modules import iscsi_deploy
from ironic.drivers.modules import noop
from ironic.drivers.modules import pxe
from ironic.drivers.modules.storage import noop as noop_storage
from ironic.drivers.modules.ucs import management as ucs_mgmt
from ironic.drivers.modules.ucs import power as ucs_power
from ironic.tests.unit.db import base as db_base
from ironic.tests.unit.objects import utils as obj_utils
class CiscoUCSStandaloneHardwareTestCase(db_base.DbTestCase):
def setUp(self):
super(CiscoUCSStandaloneHardwareTestCase, self).setUp()
self.config(enabled_hardware_types=['cisco-ucs-standalone'],
enabled_power_interfaces=['cimc', 'ipmitool'],
enabled_management_interfaces=['cimc', 'ipmitool'],
enabled_raid_interfaces=['no-raid', 'agent'],
enabled_console_interfaces=['no-console'],
enabled_vendor_interfaces=['ipmitool', 'no-vendor'])
def _validate_interfaces(self, task, **kwargs):
self.assertIsInstance(
task.driver.management,
kwargs.get('management', cimc_mgmt.CIMCManagement))
self.assertIsInstance(
task.driver.power,
kwargs.get('power', cimc_power.Power))
self.assertIsInstance(
task.driver.boot,
kwargs.get('boot', pxe.PXEBoot))
self.assertIsInstance(
task.driver.deploy,
kwargs.get('deploy', iscsi_deploy.ISCSIDeploy))
self.assertIsInstance(
task.driver.console,
kwargs.get('console', noop.NoConsole))
self.assertIsInstance(
task.driver.raid,
kwargs.get('raid', noop.NoRAID))
self.assertIsInstance(
task.driver.vendor,
kwargs.get('vendor', ipmitool.VendorPassthru))
self.assertIsInstance(
task.driver.storage,
kwargs.get('storage', noop_storage.NoopStorage))
def test_default_interfaces(self):
node = obj_utils.create_test_node(self.context,
driver='cisco-ucs-standalone')
with task_manager.acquire(self.context, node.id) as task:
self._validate_interfaces(task)
def test_override_with_ipmi_interfaces(self):
node = obj_utils.create_test_node(
self.context, driver='cisco-ucs-standalone',
power_interface='ipmitool',
management_interface='ipmitool',
deploy_interface='direct',
raid_interface='agent',
console_interface='no-console',
vendor_interface='no-vendor')
with task_manager.acquire(self.context, node.id) as task:
self._validate_interfaces(
task,
deploy=agent.AgentDeploy,
console=noop.NoConsole,
raid=agent.AgentRAID,
vendor=noop.NoVendor,
power=ipmitool.IPMIPower,
management=ipmitool.IPMIManagement)
class CiscoUCSManagedHardwareTestCase(db_base.DbTestCase):
def setUp(self):
super(CiscoUCSManagedHardwareTestCase, self).setUp()
self.config(enabled_hardware_types=['cisco-ucs-managed'],
enabled_power_interfaces=['ucsm', 'cimc'],
enabled_management_interfaces=['ucsm', 'cimc'],
enabled_raid_interfaces=['no-raid', 'agent'],
enabled_console_interfaces=['no-console'],
enabled_vendor_interfaces=['ipmitool', 'no-vendor'])
def _validate_interfaces(self, task, **kwargs):
self.assertIsInstance(
task.driver.management,
kwargs.get('management', ucs_mgmt.UcsManagement))
self.assertIsInstance(
task.driver.power,
kwargs.get('power', ucs_power.Power))
self.assertIsInstance(
task.driver.boot,
kwargs.get('boot', pxe.PXEBoot))
self.assertIsInstance(
task.driver.deploy,
kwargs.get('deploy', iscsi_deploy.ISCSIDeploy))
self.assertIsInstance(
task.driver.console,
kwargs.get('console', noop.NoConsole))
self.assertIsInstance(
task.driver.raid,
kwargs.get('raid', noop.NoRAID))
self.assertIsInstance(
task.driver.vendor,
kwargs.get('vendor', ipmitool.VendorPassthru))
self.assertIsInstance(
task.driver.storage,
kwargs.get('storage', noop_storage.NoopStorage))
def test_default_interfaces(self):
node = obj_utils.create_test_node(self.context,
driver='cisco-ucs-managed')
with task_manager.acquire(self.context, node.id) as task:
self._validate_interfaces(task)
def test_override_with_cimc_interfaces(self):
node = obj_utils.create_test_node(
self.context, driver='cisco-ucs-managed',
power_interface='cimc',
management_interface='cimc',
deploy_interface='direct',
raid_interface='agent',
console_interface='no-console',
vendor_interface='no-vendor')
with task_manager.acquire(self.context, node.id) as task:
self._validate_interfaces(
task,
deploy=agent.AgentDeploy,
console=noop.NoConsole,
raid=agent.AgentRAID,
vendor=noop.NoVendor,
power=cimc_power.Power,
management=cimc_mgmt.CIMCManagement)

26
ironic/tests/unit/drivers/third_party_driver_mocks.py
View File

@ -174,32 +174,6 @@ class MockKwargsException(Exception):
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'])
imcsdk = importutils.try_import('ImcSdk')
if not imcsdk:
imcsdk = mock.MagicMock()
imcsdk.ImcException = Exception
sys.modules['ImcSdk'] = imcsdk
if 'ironic.drivers.modules.cimc' in sys.modules:
six.moves.reload_module(
sys.modules['ironic.drivers.modules.cimc'])
sushy = importutils.try_import('sushy')
if not sushy:
sushy = mock.MagicMock(

6
setup.cfg
View File

@ -94,7 +94,6 @@ ironic.hardware.interfaces.inspect =
redfish = ironic.drivers.modules.redfish.inspect:RedfishInspect
ironic.hardware.interfaces.management =
cimc = ironic.drivers.modules.cimc.management:CIMCManagement
fake = ironic.drivers.modules.fake:FakeManagement
ibmc = ironic.drivers.modules.ibmc.management:IBMCManagement
idrac = ironic.drivers.modules.drac.management:DracManagement
@ -103,7 +102,6 @@ ironic.hardware.interfaces.management =
irmc = ironic.drivers.modules.irmc.management:IRMCManagement
noop = ironic.drivers.modules.noop_mgmt:NoopManagement
redfish = ironic.drivers.modules.redfish.management:RedfishManagement
ucsm = ironic.drivers.modules.ucs.management:UcsManagement
xclarity = ironic.drivers.modules.xclarity.management:XClarityManagement
ironic.hardware.interfaces.network =
@ -112,7 +110,6 @@ ironic.hardware.interfaces.network =
noop = ironic.drivers.modules.network.noop:NoopNetwork
ironic.hardware.interfaces.power =
cimc = ironic.drivers.modules.cimc.power:Power
fake = ironic.drivers.modules.fake:FakePower
ibmc = ironic.drivers.modules.ibmc.power:IBMCPower
idrac = ironic.drivers.modules.drac.power:DracPower
@ -121,7 +118,6 @@ ironic.hardware.interfaces.power =
irmc = ironic.drivers.modules.irmc.power:IRMCPower
redfish = ironic.drivers.modules.redfish.power:RedfishPower
snmp = ironic.drivers.modules.snmp:SNMPPower
ucsm = ironic.drivers.modules.ucs.power:Power
xclarity = ironic.drivers.modules.xclarity.power:XClarityPower
ironic.hardware.interfaces.raid =
@ -152,8 +148,6 @@ ironic.hardware.interfaces.vendor =
no-vendor = ironic.drivers.modules.noop:NoVendor
ironic.hardware.types =
cisco-ucs-managed = ironic.drivers.cisco_ucs:CiscoUCSManaged
cisco-ucs-standalone = ironic.drivers.cisco_ucs:CiscoUCSStandalone
fake-hardware = ironic.drivers.fake_hardware:FakeHardware
ibmc = ironic.drivers.ibmc:IBMCHardware
idrac = ironic.drivers.drac:IDRACHardware