openstack
/
ironic
Code Issues Proposed changes

Add iRMC Driver and its iRMC Power module 

Adding new iRMC Driver, namely pxe_irmc, to the list of available
drivers in Ironic and implementing the iRMC power module to interact
with ServerView Common Command Interface (SCCI) described in FUJITSU
Software ServerView Suite, Remote Management, iRMC S4 - integrated
Remote Management Controller.

Implements: blueprint irmc-power-driver
Change-Id: I90d8027877b0873ea13bfbd7731ff7bac3e35d13
This commit is contained in:
Naohiro Tamura 2014-12-28 03:39:30 +09:00 committed by Devananda van der Veen
parent cee9594382
commit 4412b83481
14 changed files with 697 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
doc/source/deploy/drivers.rst Normal file → Executable file
View File

@ -84,3 +84,44 @@ SeaMicro driver
:maxdepth: 1
../drivers/seamicro
iRMC
----
The iRMC driver enables PXE Deploy to control power via ServerView Common
Command Interface (SCCI).
Software Requirements
^^^^^^^^^^^^^^^^^^^^^
- Install `python-scciclient package <https://pypi.python.org/pypi/python-scciclient>`_
Enabling the iRMC Driver
^^^^^^^^^^^^^^^^^^^^^^^^
- Add ``pxe_irmc`` to the list of ``enabled_drivers in``
``/etc/ironic/ironic.conf``
- Ironic Conductor must be restarted for the new driver to be loaded.
Ironic Node Configuration
^^^^^^^^^^^^^^^^^^^^^^^^^
Nodes are configured for iRMC with PXE Deploy by setting the Ironic node
object's ``driver`` property to be ``pxe_irmc``. Further configuration values
are added to ``driver_info``:
- ``irmc_address``: hostname or IP of iRMC
- ``irmc_username``: username for iRMC with administrator privileges
- ``irmc_password``: password for irmc_username
- ``irmc_port``: port number of iRMC (optional, either 80 or 443. defalut 443)
- ``irmc_auth_method``: authentication method for iRMC (optional, either
'basic' or 'digest'. default is 'basic')
Supported Platforms
^^^^^^^^^^^^^^^^^^^
This driver supports FUJITSU PRIMERGY BX S4 or RX S8 servers and above.
- PRIMERGY BX920 S4
- PRIMERGY BX924 S4
- PRIMERGY RX300 S8

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

@ -900,6 +900,24 @@
#min_command_interval=5
[irmc]
#
# Options defined in ironic.drivers.modules.irmc.common
#
# Port to be used for iRMC operations, either 80 or 443
# (integer value)
#port=443
# Authentication method to be used for iRMC operations, either
# "basic" or "digest" (string value)
#auth_method=basic
# Timeout (in seconds) for iRMC operations (integer value)
#client_timeout=60
[keystone_authtoken]
#

4
ironic/common/exception.py Normal file → Executable file
View File

@ -501,3 +501,7 @@ class SNMPFailure(IronicException):
class FileSystemNotSupported(IronicException):
message = _("Failed to create a file system. "
"File system %(fs)s is not supported.")
class IRMCOperationError(IronicException):
message = _('iRMC %(operation)s failed. Reason: %(error)s')

13
ironic/drivers/fake.py Normal file → Executable file
View File

@ -31,6 +31,7 @@ from ironic.drivers.modules.ilo import management as ilo_management
from ironic.drivers.modules.ilo import power as ilo_power
from ironic.drivers.modules import ipminative
from ironic.drivers.modules import ipmitool
from ironic.drivers.modules.irmc import power as irmc_power
from ironic.drivers.modules import pxe
from ironic.drivers.modules import seamicro
from ironic.drivers.modules import snmp
@ -171,3 +172,15 @@ class FakeSNMPDriver(base.BaseDriver):
reason=_("Unable to import pysnmp library"))
self.power = snmp.SNMPPower()
self.deploy = fake.FakeDeploy()
class FakeIRMCDriver(base.BaseDriver):
"""Fake iRMC driver."""
def __init__(self):
if not importutils.try_import('scciclient'):
raise exception.DriverLoadError(
driver=self.__class__.__name__,
reason=_("Unable to import python-scciclient library"))
self.power = irmc_power.IRMCPower()
self.deploy = fake.FakeDeploy()

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

148
ironic/drivers/modules/irmc/common.py Normal file
View File

@ -0,0 +1,148 @@
#
# 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.
"""
Common functionalities shared between different iRMC modules.
"""
from oslo.config import cfg
from oslo.utils import importutils
from ironic.common import exception
from ironic.common.i18n import _
from ironic.openstack.common import log as logging
scci = importutils.try_import('scciclient.irmc.scci')
opts = [
cfg.IntOpt('port',
default=443,
help='Port to be used for iRMC operations, either 80 or 443'),
cfg.StrOpt('auth_method',
default='basic',
help='Authentication method to be used for iRMC operations, ' +
'either "basic" or "digest"'),
cfg.IntOpt('client_timeout',
default=60,
help='Timeout (in seconds) for iRMC operations'),
]
CONF = cfg.CONF
CONF.register_opts(opts, group='irmc')
LOG = logging.getLogger(__name__)
REQUIRED_PROPERTIES = {
'irmc_address': _("IP address or hostname of the iRMC. Required."),
'irmc_username': _("Username for the iRMC with administrator privileges. "
"Required."),
'irmc_password': _("Password for irmc_username. Required.")
}
OPTIONAL_PROPERTIES = {
'port': _("Port to be used for irmc operations either 80 or 443. "
"Optional. The default value is 443"),
'auth_method': _("Authentication method for iRMC operations "
"either 'basic' or 'digest'. "
"Optional. The default value is 'digest'"),
'client_timeout': _("Timeout (in seconds) for iRMC operations. "
"Optional. The default value is 60.")
}
COMMON_PROPERTIES = REQUIRED_PROPERTIES.copy()
COMMON_PROPERTIES.update(OPTIONAL_PROPERTIES)
def parse_driver_info(node):
"""Gets the specific Node driver info.
This method validates whether the 'driver_info' property of the
supplied node contains the required information for this driver.
:param node: an ironic node object.
:returns: a dict containing information from driver_info
and default values.
:raises: InvalidParameterValue if invalid value is contained
in the 'driver_info' property.
:raises: MissingParameterValue if some mandatory key is missing
in the 'driver_info' property.
"""
info = node.driver_info
missing_info = [key for key in REQUIRED_PROPERTIES if not info.get(key)]
if missing_info:
raise exception.MissingParameterValue(_(
"Missing the following iRMC parameters in node's"
" driver_info: %s.") % missing_info)
req = {key: value for key, value in info.iteritems()
if key in REQUIRED_PROPERTIES}
opt = {'irmc_' + param: info.get('irmc_' + param, CONF.irmc.get(param))
for param in OPTIONAL_PROPERTIES}
d_info = dict(list(req.items()) + list(opt.items()))
error_msgs = []
if (d_info['irmc_auth_method'].lower() not in ('basic', 'digest')):
error_msgs.append(
_("'%s' has unsupported value.") % 'irmc_auth_method')
if d_info['irmc_port'] not in (80, 443):
error_msgs.append(
_("'%s' has unsupported value.") % 'irmc_port')
if not isinstance(d_info['irmc_client_timeout'], int):
error_msgs.append(
_("'%s' is not integer type.") % 'irmc_client_timeout')
if error_msgs:
msg = (_("The following type errors were encountered while parsing "
"driver_info:\n%s") % "\n".join(error_msgs))
raise exception.InvalidParameterValue(msg)
return d_info
def get_irmc_client(node):
"""Gets an iRMC SCCI client.
Given an ironic node object, this method gives back a iRMC SCCI client
to do operations on the iRMC.
:param node: an ironic node object.
:returns: scci_cmd partial function which takes a SCCI command param.
:raises: InvalidParameterValue on invalid inputs.
:raises: MissingParameterValue if some mandatory information
is missing on the node
"""
driver_info = parse_driver_info(node)
scci_client = scci.get_client(
driver_info['irmc_address'],
driver_info['irmc_username'],
driver_info['irmc_password'],
port=driver_info['irmc_port'],
auth_method=driver_info['irmc_auth_method'],
client_timeout=driver_info['irmc_client_timeout'])
return scci_client
def update_ipmi_properties(task):
"""Update ipmi properties to node driver_info
:param task: a task from TaskManager.
"""
node = task.node
info = node.driver_info
# updating ipmi credentials
info['ipmi_address'] = info.get('irmc_address')
info['ipmi_username'] = info.get('irmc_username')
info['ipmi_password'] = info.get('irmc_password')
# saving ipmi credentials to task object
task.node.driver_info = info

133
ironic/drivers/modules/irmc/power.py Normal file
View File

@ -0,0 +1,133 @@
#
# 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.
"""
iRMC Power Driver using the Base Server Profile
"""
from oslo.config import cfg
from oslo.utils import importutils
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common.i18n import _LE
from ironic.common import states
from ironic.conductor import task_manager
from ironic.drivers import base
from ironic.drivers.modules import ipmitool
from ironic.drivers.modules.irmc import common as irmc_common
from ironic.openstack.common import log as logging
scci = importutils.try_import('scciclient.irmc.scci')
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
if scci:
STATES_MAP = {states.POWER_OFF: scci.POWER_OFF,
states.POWER_ON: scci.POWER_ON,
states.REBOOT: scci.POWER_RESET}
def _set_power_state(task, target_state):
"""Turns the server power on/off or do a reboot.
:param task: a TaskManager instance containing the node to act on.
:param target_state: target state of the node.
:raises: InvalidParameterValue if an invalid power state was specified.
:raises: MissingParameterValue if some mandatory information
is missing on the node
:raises: IRMCOperationError on an error from SCCI
"""
node = task.node
irmc_client = irmc_common.get_irmc_client(node)
try:
irmc_client(STATES_MAP[target_state])
except KeyError:
msg = _("_set_power_state called with invalid power state "
"'%s'") % target_state
raise exception.InvalidParameterValue(msg)
except scci.SCCIClientError as irmc_exception:
LOG.error(_LE("iRMC set_power_state failed to set state to %(tstate)s "
" for node %(node_id)s with error: %(error)s"),
{'tstate': target_state, 'node_id': node.uuid,
'error': irmc_exception})
operation = _('iRMC set_power_state')
raise exception.IRMCOperationError(operation=operation,
error=irmc_exception)
class IRMCPower(base.PowerInterface):
"""Interface for power-related actions."""
def get_properties(self):
"""Return the properties of the interface.
:returns: dictionary of <property name>:<property description> entries.
"""
return irmc_common.COMMON_PROPERTIES
def validate(self, task):
"""Validate the driver-specific Node power info.
This method validates whether the 'driver_info' property of the
supplied node contains the required information for this driver to
manage the power state of the node.
:param task: a TaskManager instance containing the node to act on.
:raises: InvalidParameterValue if required driver_info attribute
is missing or invalid on the node.
:raises: MissingParameterValue if a required parameter is missing.
"""
irmc_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.
:returns: a power state. One of :mod:`ironic.common.states`.
:raises: InvalidParameterValue if required ipmi parameters are missing.
:raises: MissingParameterValue if a required parameter is missing.
:raises: IPMIFailure on an error from ipmitool (from _power_status
call).
"""
irmc_common.update_ipmi_properties(task)
ipmi_power = ipmitool.IPMIPower()
return ipmi_power.get_power_state(task)
@task_manager.require_exclusive_lock
def set_power_state(self, task, power_state):
"""Set the power state of the task's node.
:param task: a TaskManager instance containing the node to act on.
:param power_state: Any power state from :mod:`ironic.common.states`.
:raises: InvalidParameterValue if an invalid power state was specified.
:raises: MissingParameterValue if some mandatory information
is missing on the node
:raises: IRMCOperationError if failed to set the power state.
"""
_set_power_state(task, power_state)
@task_manager.require_exclusive_lock
def reboot(self, task):
"""Perform a hard reboot of the task's node.
:param task: a TaskManager instance containing the node to act on.
:raises: InvalidParameterValue if an invalid power state was specified.
:raises: IRMCOperationError if failed to set the power state.
"""
_set_power_state(task, states.REBOOT)

22
ironic/drivers/pxe.py
View File

@ -28,6 +28,7 @@ from ironic.drivers.modules.ilo import management as ilo_management
from ironic.drivers.modules.ilo import power as ilo_power
from ironic.drivers.modules import ipminative
from ironic.drivers.modules import ipmitool
from ironic.drivers.modules.irmc import power as irmc_power
from ironic.drivers.modules import pxe
from ironic.drivers.modules import seamicro
from ironic.drivers.modules import snmp
@ -187,3 +188,24 @@ class PXEAndSNMPDriver(base.BaseDriver):
# PDUs have no boot device management capability.
# Only PXE as a boot device is supported.
self.management = None
class PXEAndIRMCDriver(base.BaseDriver):
"""PXE + iRMC driver using SCCI.
This driver implements the `core` functionality using
:class:`ironic.drivers.modules.irmc.power.IRMCPower` for power management
:class:`ironic.drivers.modules.pxe.PXEDeploy` for image deployment.
"""
def __init__(self):
if not importutils.try_import('scciclient'):
raise exception.DriverLoadError(
driver=self.__class__.__name__,
reason=_("Unable to import python-scciclient library"))
self.power = irmc_power.IRMCPower()
self.console = ipmitool.IPMIShellinaboxConsole()
self.deploy = pxe.PXEDeploy()
self.management = ipmitool.IPMIManagement()
self.vendor = pxe.VendorPassthru()

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

@ -104,6 +104,16 @@ def get_test_drac_info():
}
def get_test_irmc_info():
return {
"irmc_address": "1.2.3.4",
"irmc_username": "admin0",
"irmc_password": "fake0",
"irmc_port": 80,
"irmc_auth_method": "digest",
}
def get_test_agent_instance_info():
return {
'image_source': 'fake-image',

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

132
ironic/tests/drivers/irmc/test_common.py Normal file
View File

@ -0,0 +1,132 @@
#
# 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 iRMC modules.
"""
import mock
from oslo.config import cfg
from ironic.common import exception
from ironic.conductor import task_manager
from ironic.drivers.modules.irmc import common as irmc_common
from ironic.tests.conductor import utils as mgr_utils
from ironic.tests.db import base as db_base
from ironic.tests.db import utils as db_utils
from ironic.tests.objects import utils as obj_utils
CONF = cfg.CONF
class IRMCValidateParametersTestCase(db_base.DbTestCase):
def setUp(self):
super(IRMCValidateParametersTestCase, self).setUp()
self.node = obj_utils.create_test_node(
self.context,
driver='fake_irmc',
driver_info=db_utils.get_test_irmc_info())
def test_parse_driver_info(self):
info = irmc_common.parse_driver_info(self.node)
self.assertIsNotNone(info.get('irmc_address'))
self.assertIsNotNone(info.get('irmc_username'))
self.assertIsNotNone(info.get('irmc_password'))
self.assertIsNotNone(info.get('irmc_client_timeout'))
self.assertIsNotNone(info.get('irmc_port'))
self.assertIsNotNone(info.get('irmc_auth_method'))
def test_parse_driver_info_missing_address(self):
del self.node.driver_info['irmc_address']
self.assertRaises(exception.MissingParameterValue,
irmc_common.parse_driver_info, self.node)
def test_parse_driver_info_missing_username(self):
del self.node.driver_info['irmc_username']
self.assertRaises(exception.MissingParameterValue,
irmc_common.parse_driver_info, self.node)
def test_parse_driver_info_missing_password(self):
del self.node.driver_info['irmc_password']
self.assertRaises(exception.MissingParameterValue,
irmc_common.parse_driver_info, self.node)
def test_parse_driver_info_invalid_timeout(self):
self.node.driver_info['irmc_client_timeout'] = 'qwe'
self.assertRaises(exception.InvalidParameterValue,
irmc_common.parse_driver_info, self.node)
def test_parse_driver_info_invalid_port(self):
self.node.driver_info['irmc_port'] = 'qwe'
self.assertRaises(exception.InvalidParameterValue,
irmc_common.parse_driver_info, self.node)
def test_parse_driver_info_invalid_auth_method(self):
self.node.driver_info['irmc_auth_method'] = 'qwe'
self.assertRaises(exception.InvalidParameterValue,
irmc_common.parse_driver_info, self.node)
def test_parse_driver_info_missing_multiple_params(self):
del self.node.driver_info['irmc_password']
del self.node.driver_info['irmc_address']
try:
irmc_common.parse_driver_info(self.node)
self.fail("parse_driver_info did not throw exception.")
except exception.MissingParameterValue as e:
self.assertIn('irmc_password', str(e))
self.assertIn('irmc_address', str(e))
class IRMCCommonMethodsTestCase(db_base.DbTestCase):
def setUp(self):
super(IRMCCommonMethodsTestCase, self).setUp()
mgr_utils.mock_the_extension_manager(driver="fake_irmc")
self.info = db_utils.get_test_irmc_info()
self.node = obj_utils.create_test_node(
self.context,
driver='fake_irmc',
driver_info=self.info)
@mock.patch.object(irmc_common, 'scci')
def test_get_irmc_client(self, mock_scci):
self.info['irmc_port'] = 80
self.info['irmc_auth_method'] = 'digest'
self.info['irmc_client_timeout'] = 60
mock_scci.get_client.return_value = 'get_client'
returned_mock_scci_get_client = irmc_common.get_irmc_client(self.node)
mock_scci.get_client.assert_called_with(
self.info['irmc_address'],
self.info['irmc_username'],
self.info['irmc_password'],
port=self.info['irmc_port'],
auth_method=self.info['irmc_auth_method'],
client_timeout=self.info['irmc_client_timeout'])
self.assertEqual('get_client', returned_mock_scci_get_client)
def test_update_ipmi_properties(self):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ipmi_info = {
"ipmi_address": "1.2.3.4",
"ipmi_username": "admin0",
"ipmi_password": "fake0",
}
task.node.driver_info = self.info
irmc_common.update_ipmi_properties(task)
actual_info = task.node.driver_info
expected_info = dict(self.info, **ipmi_info)
self.assertEqual(expected_info, actual_info)

154
ironic/tests/drivers/irmc/test_power.py Normal file
View File

@ -0,0 +1,154 @@
#
# 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 iRMC Power Driver
"""
import mock
from oslo.config import cfg
from ironic.common import exception
from ironic.common import states
from ironic.conductor import task_manager
from ironic.drivers.modules.irmc import common as irmc_common
from ironic.drivers.modules.irmc import power as irmc_power
from ironic.tests.conductor import utils as mgr_utils
from ironic.tests.db import base as db_base
from ironic.tests.db import utils as db_utils
from ironic.tests.objects import utils as obj_utils
INFO_DICT = db_utils.get_test_irmc_info()
CONF = cfg.CONF
@mock.patch.object(irmc_common, 'get_irmc_client')
class IRMCPowerInternalMethodsTestCase(db_base.DbTestCase):
def setUp(self):
super(IRMCPowerInternalMethodsTestCase, self).setUp()
mgr_utils.mock_the_extension_manager(driver='fake_irmc')
driver_info = INFO_DICT
self.node = db_utils.create_test_node(
driver='fake_irmc',
driver_info=driver_info,
instance_uuid='instance_uuid_123')
def test__set_power_state_power_on_ok(self,
get_irmc_client_mock):
irmc_client = get_irmc_client_mock.return_value
target_state = states.POWER_ON
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
irmc_power._set_power_state(task, target_state)
irmc_client.assert_called_once_with(irmc_power.scci.POWER_ON)
def test__set_power_state_power_off_ok(self,
get_irmc_client_mock):
irmc_client = get_irmc_client_mock.return_value
target_state = states.POWER_OFF
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
irmc_power._set_power_state(task, target_state)
irmc_client.assert_called_once_with(irmc_power.scci.POWER_OFF)
def test__set_power_state_power_reboot_ok(self,
get_irmc_client_mock):
irmc_client = get_irmc_client_mock.return_value
target_state = states.REBOOT
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
irmc_power._set_power_state(task, target_state)
irmc_client.assert_called_once_with(irmc_power.scci.POWER_RESET)
def test__set_power_state_invalid_target_state(self,
get_irmc_client_mock):
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.assertRaises(exception.InvalidParameterValue,
irmc_power._set_power_state,
task,
states.ERROR)
def test__set_power_state_scci_exception(self,
get_irmc_client_mock):
irmc_client = get_irmc_client_mock.return_value
irmc_client.side_effect = Exception()
irmc_power.scci.SCCIClientError = Exception
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.assertRaises(exception.IRMCOperationError,
irmc_power._set_power_state,
task,
states.POWER_ON)
class IRMCPowerTestCase(db_base.DbTestCase):
def setUp(self):
super(IRMCPowerTestCase, self).setUp()
driver_info = INFO_DICT
mgr_utils.mock_the_extension_manager(driver="fake_irmc")
self.node = obj_utils.create_test_node(self.context,
driver='fake_irmc',
driver_info=driver_info)
def test_get_properties(self):
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
properties = task.driver.get_properties()
for prop in irmc_common.COMMON_PROPERTIES:
self.assertIn(prop, properties)
@mock.patch.object(irmc_common, 'parse_driver_info')
def test_validate(self, mock_drvinfo):
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
task.driver.power.validate(task)
mock_drvinfo.assert_called_once_with(task.node)
@mock.patch.object(irmc_common, 'parse_driver_info')
def test_validate_fail(self, mock_drvinfo):
side_effect = exception.InvalidParameterValue("Invalid Input")
mock_drvinfo.side_effect = side_effect
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.assertRaises(exception.InvalidParameterValue,
task.driver.power.validate,
task)
@mock.patch('ironic.drivers.modules.irmc.power.ipmitool.IPMIPower')
def test_get_power_state(self, mock_IPMIPower):
ipmi_power = mock_IPMIPower.return_value
ipmi_power.get_power_state.return_value = states.POWER_ON
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
self.assertEqual(states.POWER_ON,
task.driver.power.get_power_state(task))
ipmi_power.get_power_state.assert_called_once_with(task)
@mock.patch.object(irmc_power, '_set_power_state')
def test_set_power_state(self, mock_set_power):
mock_set_power.return_value = states.POWER_ON
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
task.driver.power.set_power_state(task, states.POWER_ON)
mock_set_power.assert_called_once_with(task, states.POWER_ON)
@mock.patch.object(irmc_power, '_set_power_state')
@mock.patch.object(irmc_power.IRMCPower, 'get_power_state')
def test_reboot(self, mock_get_power, mock_set_power):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
task.driver.power.reboot(task)
mock_set_power.assert_called_once_with(task, states.REBOOT)

20
ironic/tests/drivers/third_party_driver_mocks.py Normal file → Executable file
View File

@ -26,6 +26,7 @@ Current list of mocked libraries:
- ipminative
- proliantutils
- pysnmp
- scciclient
"""
import sys
@ -141,3 +142,22 @@ if not pysnmp:
# external library has been mocked
if 'ironic.drivers.modules.snmp' in sys.modules:
reload(sys.modules['ironic.drivers.modules.snmp'])
# attempt to load the external 'scciclient' library, which is required by
# the optional drivers.modules.irmc module
scciclient = importutils.try_import('scciclient')
if not scciclient:
mock_scciclient = mock.MagicMock()
sys.modules['scciclient'] = mock_scciclient
sys.modules['scciclient.irmc'] = mock_scciclient.irmc
sys.modules['scciclient.irmc.scci'] = mock.MagicMock(
POWER_OFF=mock.sentinel.POWER_OFF,
POWER_ON=mock.sentinel.POWER_ON,
POWER_RESET=mock.sentinel.POWER_RESET)
# if anything has loaded the iRMC driver yet, reload it now that the
# external library has been mocked
if 'ironic.drivers.modules.irmc' in sys.modules:
reload(sys.modules['ironic.drivers.modules.irmc'])

2
setup.cfg Normal file → Executable file
View File

@ -49,6 +49,7 @@ ironic.drivers =
fake_ilo = ironic.drivers.fake:FakeIloDriver
fake_drac = ironic.drivers.fake:FakeDracDriver
fake_snmp = ironic.drivers.fake:FakeSNMPDriver
fake_irmc = ironic.drivers.fake:FakeIRMCDriver
iscsi_ilo = ironic.drivers.ilo:IloVirtualMediaIscsiDriver
pxe_ipmitool = ironic.drivers.pxe:PXEAndIPMIToolDriver
pxe_ipminative = ironic.drivers.pxe:PXEAndIPMINativeDriver
@ -58,6 +59,7 @@ ironic.drivers =
pxe_ilo = ironic.drivers.pxe:PXEAndIloDriver
pxe_drac = ironic.drivers.drac:PXEDracDriver
pxe_snmp = ironic.drivers.pxe:PXEAndSNMPDriver
pxe_irmc = ironic.drivers.pxe:PXEAndIRMCDriver
ironic.database.migration_backend =
sqlalchemy = ironic.db.sqlalchemy.migration