openstack
/
ironic
Code Issues Proposed changes

Remove deprecated, untested ipminative driver 

The ipminative driver was marked as unsupported on September 28, 2016,
and an email was sent to the list on March 9th, 2017 asking for
volunteers to run third-party CI to save the driver from deprecation.

Additionally, many ironic contributors who also deploy ironic have
reported instability while using this driver. For these reasons it's
being removed from the main ironic tree. If there are any users still
interested in this driver, they are invited to host it elsewhere.

Change-Id: I9bc9f4cbd916f040a636b967ec5556197ad3d8a8
Closes-bug: #1671532
This commit is contained in:
Jay Faulkner 2017-03-29 11:57:20 -07:00
parent a84c2e0d8e
commit 96eafdc866
14 changed files with 11 additions and 1505 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
driver-requirements.txt
View File

@ -5,7 +5,6 @@
# These are available on pypi
proliantutils>=2.2.1
pyghmi>=0.8.0
pysnmp
python-ironic-inspector-client>=1.5.0
python-oneviewclient<3.0.0,>=2.5.2

10
install-guide/source/setup-drivers.rst
View File

@ -475,11 +475,11 @@ controller in your bare metal server by using ``ipmitool``::
default is fairly conservative, as setting this timeout too low can cause
older BMCs to crash and require a hard-reset.
Bare Metal service supports sending IPMI sensor data to Telemetry with pxe_ipmitool,
pxe_ipminative, agent_ipmitool, agent_pyghmi, agent_ilo, iscsi_ilo, pxe_ilo,
and with pxe_irmc driver starting from Kilo release. By default, support for
sending IPMI sensor data to Telemetry is disabled. If you want to enable it,
you should make the following two changes in ``ironic.conf``:
Bare Metal service supports sending IPMI sensor data to Telemetry with
pxe_ipmitool, agent_ipmitool, agent_ilo, iscsi_ilo, pxe_ilo, and with pxe_irmc
driver. By default, support for sending IPMI sensor data to Telemetry is
disabled. If you want to enable it, you should make the following two changes
in ``ironic.conf``:
* ``notification_driver = messaging`` in the ``DEFAULT`` section
* ``send_sensor_data = true`` in the ``conductor`` section

2
install-guide/source/standalone.rst
View File

@ -43,7 +43,7 @@ service via hrefs.
There are however some limitations for different drivers:
* If you're using one of the drivers that use agent deploy method (namely,
``agent_ilo``, ``agent_ipmitool``, ``agent_pyghmi`` or ``agent_ssh``)
``agent_ilo``, ``agent_ipmitool``, or ``agent_ssh``)
you have to know MD5 checksum for your instance image. To
compute it, you can use the following command::

27
ironic/drivers/agent.py
View File

@ -22,7 +22,6 @@ 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 inspector
from ironic.drivers.modules import ipminative
from ironic.drivers.modules import pxe
from ironic.drivers.modules import ssh
from ironic.drivers.modules.ucs import management as ucs_mgmt
@ -34,32 +33,6 @@ AgentAndIPMIToolDriver = ipmi.AgentAndIPMIToolDriver
AgentAndIPMIToolAndSocatDriver = ipmi.AgentAndIPMIToolAndSocatDriver
class AgentAndIPMINativeDriver(base.BaseDriver):
"""Agent + IPMINative driver.
This driver implements the `core` functionality, combining
:class:`ironic.drivers.modules.ipminative.NativeIPMIPower` (for power
on/off and reboot) with
:class:`ironic.drivers.modules.agent.AgentDeploy` (for image
deployment).
Implementations are in those respective classes; this class is merely the
glue between them.
"""
supported = False
def __init__(self):
self.power = ipminative.NativeIPMIPower()
self.boot = pxe.PXEBoot()
self.deploy = agent.AgentDeploy()
self.management = ipminative.NativeIPMIManagement()
self.console = ipminative.NativeIPMIShellinaboxConsole()
self.vendor = ipminative.VendorPassthru()
self.raid = agent.AgentRAID()
self.inspect = inspector.Inspector.create_if_enabled(
'AgentAndIPMINativeDriver')
class AgentAndSSHDriver(base.BaseDriver):
"""Agent + SSH driver.

18
ironic/drivers/fake.py
View File

@ -36,7 +36,6 @@ from ironic.drivers.modules.ilo import inspect as ilo_inspect
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 inspector
from ironic.drivers.modules import ipminative
from ironic.drivers.modules import ipmitool
from ironic.drivers.modules.irmc import inspect as irmc_inspect
from ironic.drivers.modules.irmc import management as irmc_management
@ -125,23 +124,6 @@ class FakeSSHDriver(base.BaseDriver):
self.console = ssh.ShellinaboxConsole()
class FakeIPMINativeDriver(base.BaseDriver):
"""Fake IPMINative driver."""
supported = False
def __init__(self):
if not importutils.try_import('pyghmi'):
raise exception.DriverLoadError(
driver=self.__class__.__name__,
reason=_("Unable to import pyghmi IPMI library"))
self.power = ipminative.NativeIPMIPower()
self.console = ipminative.NativeIPMIShellinaboxConsole()
self.deploy = fake.FakeDeploy()
self.vendor = ipminative.VendorPassthru()
self.management = ipminative.NativeIPMIManagement()
class FakeAgentDriver(base.BaseDriver):
"""Example implementation of an AgentDriver."""

702
ironic/drivers/modules/ipminative.py
View File

@ -1,702 +0,0 @@
# coding=utf-8
# Copyright 2013 International Business Machines Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Ironic Native IPMI power manager.
"""
import os
from ironic_lib import metrics_utils
from ironic_lib import utils as ironic_utils
from oslo_log import log as logging
from oslo_utils import excutils
from oslo_utils import importutils
from oslo_utils import strutils
from ironic.common import boot_devices
from ironic.common import exception
from ironic.common.i18n import _, _LE, _LW
from ironic.common import states
from ironic.common import utils
from ironic.conductor import task_manager
from ironic.conf import CONF
from ironic.drivers import base
from ironic.drivers.modules import console_utils
from ironic.drivers.modules import deploy_utils
from ironic.drivers import utils as driver_utils
pyghmi = importutils.try_import('pyghmi')
if pyghmi:
from pyghmi import exceptions as pyghmi_exception
from pyghmi.ipmi import command as ipmi_command
LOG = logging.getLogger(__name__)
METRICS = metrics_utils.get_metrics_logger(__name__)
REQUIRED_PROPERTIES = {'ipmi_address': _("IP of the node's BMC. Required."),
'ipmi_password': _("IPMI password. Required."),
'ipmi_username': _("IPMI username. Required.")}
OPTIONAL_PROPERTIES = {
'ipmi_force_boot_device': _("Whether Ironic should specify the boot "
"device to the BMC each time the server "
"is turned on, eg. because the BMC is not "
"capable of remembering the selected boot "
"device across power cycles; default value "
"is False. Optional.")
}
COMMON_PROPERTIES = REQUIRED_PROPERTIES.copy()
COMMON_PROPERTIES.update(OPTIONAL_PROPERTIES)
CONSOLE_PROPERTIES = {
'ipmi_terminal_port': _("node's UDP port to connect to. Only required for "
"console access.")
}
_BOOT_DEVICES_MAP = {
boot_devices.DISK: 'hd',
boot_devices.PXE: 'network',
boot_devices.CDROM: 'cdrom',
boot_devices.BIOS: 'setup',
}
def _parse_driver_info(node):
"""Gets the bmc access info for the given node.
:raises: MissingParameterValue when required ipmi credentials
are missing.
:raises: InvalidParameterValue when the IPMI terminal port is not an
integer.
"""
info = node.driver_info or {}
missing_info = [key for key in REQUIRED_PROPERTIES if not info.get(key)]
if missing_info:
raise exception.MissingParameterValue(_(
"Missing the following IPMI credentials in node's"
" driver_info: %s.") % missing_info)
bmc_info = {}
bmc_info['address'] = info.get('ipmi_address')
bmc_info['username'] = info.get('ipmi_username')
bmc_info['password'] = info.get('ipmi_password')
bmc_info['force_boot_device'] = info.get('ipmi_force_boot_device', False)
# get additional info
bmc_info['uuid'] = node.uuid
# terminal port must be an integer
port = info.get('ipmi_terminal_port')
if port is not None:
port = utils.validate_network_port(port, 'ipmi_terminal_port')
bmc_info['port'] = port
return bmc_info
def _console_pwfile_path(uuid):
"""Return the file path for storing the ipmi password."""
file_name = "%(uuid)s.pw" % {'uuid': uuid}
return os.path.join(CONF.tempdir, file_name)
def _power_on(driver_info):
"""Turn the power on for this node.
:param driver_info: the bmc access info for a node.
:returns: power state POWER_ON, one of :class:`ironic.common.states`.
:raises: IPMIFailure when the native ipmi call fails.
:raises: PowerStateFailure when invalid power state is returned
from ipmi.
"""
msg = _("IPMI power on failed for node %(node_id)s with the "
"following error: %(error)s")
try:
ipmicmd = ipmi_command.Command(bmc=driver_info['address'],
userid=driver_info['username'],
password=driver_info['password'])
wait = CONF.ipmi.retry_timeout
ret = ipmicmd.set_power('on', wait)
except pyghmi_exception.IpmiException as e:
error = msg % {'node_id': driver_info['uuid'], 'error': e}
LOG.error(error)
raise exception.IPMIFailure(error)
state = ret.get('powerstate')
if state == 'on':
return states.POWER_ON
else:
error = _("bad response: %s") % ret
LOG.error(msg, {'node_id': driver_info['uuid'], 'error': error})
raise exception.PowerStateFailure(pstate=states.POWER_ON)
def _power_off(driver_info):
"""Turn the power off for this node.
:param driver_info: the bmc access info for a node.
:returns: power state POWER_OFF, one of :class:`ironic.common.states`.
:raises: IPMIFailure when the native ipmi call fails.
:raises: PowerStateFailure when invalid power state is returned
from ipmi.
"""
msg = _("IPMI power off failed for node %(node_id)s with the "
"following error: %(error)s")
try:
ipmicmd = ipmi_command.Command(bmc=driver_info['address'],
userid=driver_info['username'],
password=driver_info['password'])
wait = CONF.ipmi.retry_timeout
ret = ipmicmd.set_power('off', wait)
except pyghmi_exception.IpmiException as e:
error = msg % {'node_id': driver_info['uuid'], 'error': e}
LOG.error(error)
raise exception.IPMIFailure(error)
state = ret.get('powerstate')
if state == 'off':
return states.POWER_OFF
else:
error = _("bad response: %s") % ret
LOG.error(msg, {'node_id': driver_info['uuid'], 'error': error})
raise exception.PowerStateFailure(pstate=states.POWER_OFF)
def _reboot(driver_info):
"""Reboot this node.
If the power is off, turn it on. If the power is on, reset it.
:param driver_info: the bmc access info for a node.
:returns: power state POWER_ON, one of :class:`ironic.common.states`.
:raises: IPMIFailure when the native ipmi call fails.
:raises: PowerStateFailure when invalid power state is returned
from ipmi.
"""
msg = _("IPMI power reboot failed for node %(node_id)s with the "
"following error: %(error)s")
try:
ipmicmd = ipmi_command.Command(bmc=driver_info['address'],
userid=driver_info['username'],
password=driver_info['password'])
wait = CONF.ipmi.retry_timeout
ret = ipmicmd.set_power('boot', wait)
except pyghmi_exception.IpmiException as e:
error = msg % {'node_id': driver_info['uuid'], 'error': e}
LOG.error(error)
raise exception.IPMIFailure(error)
if 'error' in ret:
error = _("bad response: %s") % ret
LOG.error(msg, {'node_id': driver_info['uuid'], 'error': error})
raise exception.PowerStateFailure(pstate=states.REBOOT)
return states.POWER_ON
def _power_status(driver_info):
"""Get the power status for this node.
:param driver_info: the bmc access info for a node.
:returns: power state POWER_ON, POWER_OFF or ERROR defined in
:class:`ironic.common.states`.
:raises: IPMIFailure when the native ipmi call fails.
"""
try:
ipmicmd = ipmi_command.Command(bmc=driver_info['address'],
userid=driver_info['username'],
password=driver_info['password'])
ret = ipmicmd.get_power()
except pyghmi_exception.IpmiException as e:
msg = (_("IPMI get power state failed for node %(node_id)s "
"with the following error: %(error)s") %
{'node_id': driver_info['uuid'], 'error': e})
LOG.error(msg)
raise exception.IPMIFailure(msg)
state = ret.get('powerstate')
if state == 'on':
return states.POWER_ON
elif state == 'off':
return states.POWER_OFF
else:
# NOTE(linggao): Do not throw an exception here because it might
# return other valid values. It is up to the caller to decide
# what to do.
LOG.warning(_LW("IPMI get power state for node %(node_id)s returns the"
" following details: %(detail)s"),
{'node_id': driver_info['uuid'], 'detail': ret})
return states.ERROR
def _get_sensors_data(driver_info):
"""Get sensors data.
:param driver_info: node's driver info
:raises: FailedToGetSensorData when getting the sensor data fails.
:returns: returns a dict of sensor data group by sensor type.
"""
try:
ipmicmd = ipmi_command.Command(bmc=driver_info['address'],
userid=driver_info['username'],
password=driver_info['password'])
ret = ipmicmd.get_sensor_data()
except Exception as e:
LOG.error(_LE("IPMI get sensor data failed for node %(node_id)s "
"with the following error: %(error)s"),
{'node_id': driver_info['uuid'], 'error': e})
raise exception.FailedToGetSensorData(
node=driver_info['uuid'], error=e)
if not ret:
return {}
sensors_data = {}
for reading in ret:
# ignore the sensor data which has no sensor reading value
if not reading.value:
continue
sensors_data.setdefault(
reading.type,
{})[reading.name] = {
'Sensor Reading': '%s %s' % (reading.value, reading.units),
'Sensor ID': reading.name,
'States': str(reading.states),
'Units': reading.units,
'Health': str(reading.health)}
return sensors_data
def _parse_raw_bytes(raw_bytes):
"""Parse raw bytes string.
:param raw_bytes: a string of hexadecimal raw bytes, e.g. '0x00 0x01'.
:returns: a tuple containing the arguments for pyghmi call as integers,
(IPMI net function, IPMI command, list of command's data).
:raises: InvalidParameterValue when an invalid value is specified.
"""
try:
bytes_list = [int(x, base=16) for x in raw_bytes.split()]
return bytes_list[0], bytes_list[1], bytes_list[2:]
except ValueError:
raise exception.InvalidParameterValue(_(
"Invalid raw bytes string: '%s'") % raw_bytes)
except IndexError:
raise exception.InvalidParameterValue(_(
"Raw bytes string requires two bytes at least."))
def _send_raw(driver_info, raw_bytes):
"""Send raw bytes to the BMC."""
netfn, command, data = _parse_raw_bytes(raw_bytes)
LOG.debug("Sending raw bytes %(bytes)s to node %(node_id)s",
{'bytes': raw_bytes, 'node_id': driver_info['uuid']})
try:
ipmicmd = ipmi_command.Command(bmc=driver_info['address'],
userid=driver_info['username'],
password=driver_info['password'])
ipmicmd.xraw_command(netfn, command, data=data)
except pyghmi_exception.IpmiException as e:
msg = (_("IPMI send raw bytes '%(bytes)s' failed for node %(node_id)s"
" with the following error: %(error)s") %
{'bytes': raw_bytes, 'node_id': driver_info['uuid'],
'error': e})
LOG.error(msg)
raise exception.IPMIFailure(msg)
class NativeIPMIPower(base.PowerInterface):
"""The power driver using native python-ipmi library."""
def get_properties(self):
return COMMON_PROPERTIES
@METRICS.timer('NativeIPMIPower.validate')
def validate(self, task):
"""Check that node['driver_info'] contains IPMI credentials.
:param task: a TaskManager instance containing the node to act on.
:raises: MissingParameterValue when required ipmi credentials
are missing.
"""
_parse_driver_info(task.node)
@METRICS.timer('NativeIPMIPower.get_power_state')
def get_power_state(self, task):
"""Get the current power state of the task's node.
:param task: a TaskManager instance containing the node to act on.
:returns: power state POWER_ON, POWER_OFF or ERROR defined in
:class:`ironic.common.states`.
:raises: MissingParameterValue when required ipmi credentials
are missing.
:raises: IPMIFailure when the native ipmi call fails.
"""
driver_info = _parse_driver_info(task.node)
return _power_status(driver_info)
@METRICS.timer('NativeIPMIPower.set_power_state')
@task_manager.require_exclusive_lock
def set_power_state(self, task, pstate):
"""Turn the power on or off.
:param task: a TaskManager instance containing the node to act on.
:param pstate: a power state that will be set on the task's node.
:raises: IPMIFailure when the native ipmi call fails.
:raises: MissingParameterValue when required ipmi credentials
are missing.
:raises: InvalidParameterValue when an invalid power state
is specified
:raises: PowerStateFailure when invalid power state is returned
from ipmi.
"""
driver_info = _parse_driver_info(task.node)
if pstate == states.POWER_ON:
driver_utils.ensure_next_boot_device(task, driver_info)
_power_on(driver_info)
elif pstate == states.POWER_OFF:
_power_off(driver_info)
else:
raise exception.InvalidParameterValue(
_("set_power_state called with an invalid power state: %s."
) % pstate)
@METRICS.timer('NativeIPMIPower.reboot')
@task_manager.require_exclusive_lock
def reboot(self, task):
"""Cycles the power to the task's node.
:param task: a TaskManager instance containing the node to act on.
:raises: IPMIFailure when the native ipmi call fails.
:raises: MissingParameterValue when required ipmi credentials
are missing.
:raises: PowerStateFailure when invalid power state is returned
from ipmi.
"""
driver_info = _parse_driver_info(task.node)
driver_utils.ensure_next_boot_device(task, driver_info)
_reboot(driver_info)
class NativeIPMIManagement(base.ManagementInterface):
def get_properties(self):
return COMMON_PROPERTIES
@METRICS.timer('NativeIPMIManagement.validate')
def validate(self, task):
"""Check that 'driver_info' contains IPMI 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 when required ipmi credentials
are missing.
"""
_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(_BOOT_DEVICES_MAP.keys())
@METRICS.timer('NativeIPMIManagement.set_boot_device')
@task_manager.require_exclusive_lock
def set_boot_device(self, task, device, persistent=False):
"""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
:mod:`ironic.common.boot_devices`.
:param persistent: Boolean value. True if the boot device will
persist to all future boots, False if not.
Default: False.
:raises: InvalidParameterValue if an invalid boot device is specified
or required ipmi credentials are missing.
:raises: MissingParameterValue when required ipmi credentials
are missing.
:raises: IPMIFailure on an error from pyghmi.
"""
if device not in self.get_supported_boot_devices(task):
raise exception.InvalidParameterValue(_(
"Invalid boot device %s specified.") % device)
if task.node.driver_info.get('ipmi_force_boot_device', False):
driver_utils.force_persistent_boot(task,
device,
persistent)
# Reset persistent to False, in case of BMC does not support
# persistent or we do not have admin rights.
persistent = False
boot_mode = deploy_utils.get_boot_mode_for_deploy(task.node)
driver_info = _parse_driver_info(task.node)
try:
ipmicmd = ipmi_command.Command(bmc=driver_info['address'],
userid=driver_info['username'],
password=driver_info['password'])
bootdev = _BOOT_DEVICES_MAP[device]
uefiboot = boot_mode == 'uefi'
ipmicmd.set_bootdev(bootdev, persist=persistent, uefiboot=uefiboot)
except pyghmi_exception.IpmiException as e:
LOG.error(_LE("IPMI set boot device failed for node %(node_id)s "
"with the following error: %(error)s"),
{'node_id': driver_info['uuid'], 'error': e})
raise exception.IPMIFailure(cmd=e)
@METRICS.timer('NativeIPMIManagement.get_boot_device')
def get_boot_device(self, task):
"""Get the current boot device for the task's node.
Returns the current boot device of the node.
:param task: a task from TaskManager.
:raises: MissingParameterValue if required IPMI parameters
are missing.
:raises: IPMIFailure on an error from pyghmi.
: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.
"""
driver_info = task.node.driver_info
driver_internal_info = task.node.driver_internal_info
if (driver_info.get('ipmi_force_boot_device', False) and
driver_internal_info.get('persistent_boot_device') and
driver_internal_info.get('is_next_boot_persistent', True)):
return {
'boot_device': driver_internal_info['persistent_boot_device'],
'persistent': True
}
driver_info = _parse_driver_info(task.node)
response = {'boot_device': None}
try:
ipmicmd = ipmi_command.Command(bmc=driver_info['address'],
userid=driver_info['username'],
password=driver_info['password'])
ret = ipmicmd.get_bootdev()
# FIXME(lucasagomes): pyghmi doesn't seem to handle errors
# consistently, for some errors it raises an exception
# others it just returns a dictionary with the error.
if 'error' in ret:
raise pyghmi_exception.IpmiException(ret['error'])
except pyghmi_exception.IpmiException as e:
LOG.error(_LE("IPMI get boot device failed for node %(node_id)s "
"with the following error: %(error)s"),
{'node_id': driver_info['uuid'], 'error': e})
raise exception.IPMIFailure(cmd=e)
response['persistent'] = ret.get('persistent')
bootdev = ret.get('bootdev')
if bootdev:
response['boot_device'] = next((dev for dev, hdev in
_BOOT_DEVICES_MAP.items()
if hdev == bootdev), None)
return response
@METRICS.timer('NativeIPMIManagement.get_sensors_data')
def get_sensors_data(self, task):
"""Get sensors data.
:param task: a TaskManager instance.
:raises: FailedToGetSensorData when getting the sensor data fails.
:raises: MissingParameterValue if required ipmi parameters are missing
:returns: returns a dict of sensor data group by sensor type.
"""
driver_info = _parse_driver_info(task.node)
return _get_sensors_data(driver_info)
class NativeIPMIShellinaboxConsole(base.ConsoleInterface):
"""A ConsoleInterface that uses pyghmi and shellinabox."""
def get_properties(self):
d = COMMON_PROPERTIES.copy()
d.update(CONSOLE_PROPERTIES)
return d
@METRICS.timer('NativeIPMIShellinaboxConsole.validate')
def validate(self, task):
"""Validate the Node console info.
:param task: a TaskManager instance containing the node to act on.
:raises: MissingParameterValue when required IPMI credentials or
the IPMI terminal port are missing
:raises: InvalidParameterValue when the IPMI terminal port is not
an integer.
"""
driver_info = _parse_driver_info(task.node)
if not driver_info['port']:
raise exception.MissingParameterValue(_(
"Missing 'ipmi_terminal_port' parameter in node's"
" driver_info."))
@METRICS.timer('NativeIPMIShellinaboxConsole.start_console')
def start_console(self, task):
"""Start a remote console for the node.
:param task: a TaskManager instance containing the node to act on.
:raises: MissingParameterValue when required ipmi credentials
are missing.
:raises: InvalidParameterValue when the IPMI terminal port is not an
integer.
:raises: ConsoleError if unable to start the console process.
"""
driver_info = _parse_driver_info(task.node)
path = _console_pwfile_path(driver_info['uuid'])
pw_file = console_utils.make_persistent_password_file(
path, driver_info['password'])
console_cmd = ("/:%(uid)s:%(gid)s:HOME:pyghmicons %(bmc)s"
" %(user)s"
" %(passwd_file)s"
% {'uid': os.getuid(),
'gid': os.getgid(),
'bmc': driver_info['address'],
'user': driver_info['username'],
'passwd_file': pw_file})
try:
console_utils.start_shellinabox_console(driver_info['uuid'],
driver_info['port'],
console_cmd)
except exception.ConsoleError:
with excutils.save_and_reraise_exception():
ironic_utils.unlink_without_raise(path)
@METRICS.timer('NativeIPMIShellinaboxConsole.stop_console')
def stop_console(self, task):
"""Stop the remote console session for the node.
:param task: a TaskManager instance containing the node to act on.
:raises: ConsoleError if unable to stop the console process.
"""
try:
console_utils.stop_shellinabox_console(task.node.uuid)
finally:
password_file = _console_pwfile_path(task.node.uuid)
ironic_utils.unlink_without_raise(password_file)
@METRICS.timer('NativeIPMIShellinaboxConsole.get_console')
def get_console(self, task):
"""Get the type and connection information about the console.
:param task: a TaskManager instance containing the node to act on.
:raises: MissingParameterValue when required IPMI credentials or
the IPMI terminal port are missing
:raises: InvalidParameterValue when the IPMI terminal port is not
an integer.
"""
driver_info = _parse_driver_info(task.node)
url = console_utils.get_shellinabox_console_url(driver_info['port'])
return {'type': 'shellinabox', 'url': url}
class VendorPassthru(base.VendorInterface):
def get_properties(self):
return COMMON_PROPERTIES
@METRICS.timer('VendorPassthru.validate')
def validate(self, task, method, **kwargs):
"""Validate vendor-specific actions.
:param task: a task from TaskManager.
:param method: method to be validated
:param kwargs: info for action.
:raises: InvalidParameterValue when an invalid parameter value is
specified.
:raises: MissingParameterValue if a required parameter is missing.
"""
if method == 'send_raw':
raw_bytes = kwargs.get('raw_bytes')
if not raw_bytes:
raise exception.MissingParameterValue(_(
'Parameter raw_bytes (string of bytes) was not '
'specified.'))
_parse_raw_bytes(raw_bytes)
_parse_driver_info(task.node)
@METRICS.timer('VendorPassthru.send_raw')
@base.passthru(['POST'],
description=_("Send raw bytes to the BMC. Required "
"argument: 'raw_bytes' - a string of raw "
"bytes (e.g. '0x00 0x01')."))
@task_manager.require_exclusive_lock
def send_raw(self, task, http_method, raw_bytes):
"""Send raw bytes to the BMC. Bytes should be a string of bytes.
:param task: a TaskManager instance.
:param http_method: the HTTP method used on the request.
:param raw_bytes: a string of raw bytes to send, e.g. '0x00 0x01'
:raises: IPMIFailure on an error from native IPMI call.
:raises: MissingParameterValue if a required parameter is missing.
:raises: InvalidParameterValue when an invalid value is specified.
"""
driver_info = _parse_driver_info(task.node)
_send_raw(driver_info, raw_bytes)
@METRICS.timer('VendorPassthru.bmc_reset')
@base.passthru(['POST'],
description=_("Reset the BMC. Required argument: 'warm' "
"(Boolean) - for warm (True) or cold (False) "
"reset."))
@task_manager.require_exclusive_lock
def bmc_reset(self, task, http_method, warm=True):
"""Reset BMC via IPMI command.
:param task: a TaskManager instance.
:param http_method: the HTTP method used on the request.
:param warm: boolean parameter to decide on warm or cold reset.
:raises: IPMIFailure on an error from native IPMI call.
:raises: MissingParameterValue if a required parameter is missing.
:raises: InvalidParameterValue when an invalid value is specified
"""
driver_info = _parse_driver_info(task.node)
warm = strutils.bool_from_string(warm)
# NOTE(yuriyz): pyghmi 0.8.0 does not have a method for BMC reset
command = '0x03' if warm else '0x02'
raw_command = '0x06 ' + command
_send_raw(driver_info, raw_command)

30
ironic/drivers/pxe.py
View File

@ -33,7 +33,6 @@ from ironic.drivers.modules.ilo import management as ilo_management
from ironic.drivers.modules.ilo import power as ilo_power
from ironic.drivers.modules.ilo import vendor as ilo_vendor
from ironic.drivers.modules import inspector
from ironic.drivers.modules import ipminative
from ironic.drivers.modules import ipmitool
from ironic.drivers.modules.irmc import inspect as irmc_inspect
from ironic.drivers.modules.irmc import management as irmc_management
@ -77,35 +76,6 @@ class PXEAndSSHDriver(base.BaseDriver):
self.console = ssh.ShellinaboxConsole()
class PXEAndIPMINativeDriver(base.BaseDriver):
"""PXE + Native IPMI driver.
This driver implements the `core` functionality, combining
:class:`ironic.drivers.modules.ipminative.NativeIPMIPower`
for power on/off and reboot with
:class:`ironic.drivers.modules.iscsi_deploy.ISCSIDeploy`
for image deployment. Implementations are in those respective
classes; this class is merely the glue between them.
"""
supported = False
def __init__(self):
if not importutils.try_import('pyghmi'):
raise exception.DriverLoadError(
driver=self.__class__.__name__,
reason=_("Unable to import pyghmi library"))
self.power = ipminative.NativeIPMIPower()
self.console = ipminative.NativeIPMIShellinaboxConsole()
self.boot = pxe.PXEBoot()
self.deploy = iscsi_deploy.ISCSIDeploy()
self.management = ipminative.NativeIPMIManagement()
self.vendor = ipminative.VendorPassthru()
self.inspect = inspector.Inspector.create_if_enabled(
'PXEAndIPMINativeDriver')
self.raid = agent.AgentRAID()
class PXEAndIloDriver(base.BaseDriver):
"""PXE + Ilo Driver using IloClient interface.

12
ironic/tests/unit/conductor/test_manager.py
View File

@ -4758,11 +4758,6 @@ class ManagerTestProperties(mgr_utils.ServiceSetUpMixin,
]
self._check_driver_properties("fake_ipmitool", expected)
def test_driver_properties_fake_ipminative(self):
expected = ['ipmi_address', 'ipmi_password', 'ipmi_username',
'ipmi_terminal_port', 'ipmi_force_boot_device']
self._check_driver_properties("fake_ipminative", expected)
def test_driver_properties_fake_ssh(self):
expected = ['ssh_address', 'ssh_username',
'vbox_use_headless', 'ssh_virt_type',
@ -4790,13 +4785,6 @@ class ManagerTestProperties(mgr_utils.ServiceSetUpMixin,
'ipmi_force_boot_device', 'deploy_forces_oob_reboot']
self._check_driver_properties("pxe_ipmitool", expected)
def test_driver_properties_pxe_ipminative(self):
expected = ['ipmi_address', 'ipmi_password', 'ipmi_username',
'deploy_kernel', 'deploy_ramdisk',
'ipmi_terminal_port', 'ipmi_force_boot_device',
'deploy_forces_oob_reboot']
self._check_driver_properties("pxe_ipminative", expected)
def test_driver_properties_pxe_ssh(self):
expected = ['deploy_kernel', 'deploy_ramdisk',
'ssh_address', 'ssh_username',

642
ironic/tests/unit/drivers/modules/test_ipminative.py
View File

@ -1,642 +0,0 @@
# coding=utf-8
# Copyright 2013 International Business Machines Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Test class for Native IPMI power driver module.
"""
import mock
from oslo_utils import uuidutils
from pyghmi import exceptions as pyghmi_exception
from ironic.common import boot_devices
from ironic.common import driver_factory
from ironic.common import exception
from ironic.common import states
from ironic.conductor import task_manager
from ironic.drivers.modules import console_utils
from ironic.drivers.modules import deploy_utils
from ironic.drivers.modules import ipminative
from ironic.drivers import utils as driver_utils
from ironic.tests.unit.conductor import mgr_utils
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_ipmi_info()
class IPMINativePrivateMethodTestCase(db_base.DbTestCase):
"""Test cases for ipminative private methods."""
def setUp(self):
super(IPMINativePrivateMethodTestCase, self).setUp()
self.node = obj_utils.create_test_node(self.context,
driver='fake_ipminative',
driver_info=INFO_DICT)
self.info = ipminative._parse_driver_info(self.node)
def test__parse_driver_info(self):
# make sure we get back the expected things
self.assertEqual('1.2.3.4', self.info['address'])
self.assertEqual('admin', self.info['username'])
self.assertEqual('fake', self.info['password'])
self.assertEqual('1be26c0b-03f2-4d2e-ae87-c02d7f33c123',
self.info['uuid'])
self.assertEqual(False, self.info['force_boot_device'])
# make sure error is raised when info, eg. username, is missing
info = dict(INFO_DICT)
del info['ipmi_username']
node = obj_utils.get_test_node(self.context, driver_info=info)
self.assertRaises(exception.MissingParameterValue,
ipminative._parse_driver_info,
node)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__power_status_on(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_power.return_value = {'powerstate': 'on'}
state = ipminative._power_status(self.info)
ipmicmd.get_power.assert_called_once_with()
self.assertEqual(states.POWER_ON, state)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__power_status_off(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_power.return_value = {'powerstate': 'off'}
state = ipminative._power_status(self.info)
ipmicmd.get_power.assert_called_once_with()
self.assertEqual(states.POWER_OFF, state)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__power_status_error(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_power.return_value = {'powerstate': 'Error'}
state = ipminative._power_status(self.info)
ipmicmd.get_power.assert_called_once_with()
self.assertEqual(states.ERROR, state)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__power_on(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.set_power.return_value = {'powerstate': 'on'}
self.config(retry_timeout=400, group='ipmi')
state = ipminative._power_on(self.info)
ipmicmd.set_power.assert_called_once_with('on', 400)
self.assertEqual(states.POWER_ON, state)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__power_off(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.set_power.return_value = {'powerstate': 'off'}
self.config(retry_timeout=500, group='ipmi')
state = ipminative._power_off(self.info)
ipmicmd.set_power.assert_called_once_with('off', 500)
self.assertEqual(states.POWER_OFF, state)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__reboot(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.set_power.return_value = {'powerstate': 'on'}
self.config(retry_timeout=600, group='ipmi')
state = ipminative._reboot(self.info)
ipmicmd.set_power.assert_called_once_with('boot', 600)
self.assertEqual(states.POWER_ON, state)
def _create_sensor_object(self, value, type_, name, states=None,
units='fake_units', health=0):
if states is None:
states = []
return type('Reading', (object, ), {
'value': value, 'type': type_, 'name': name,
'states': states, 'units': units, 'health': health})()
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__get_sensors_data(self, ipmi_mock):
reading_1 = self._create_sensor_object('fake_value1',
'fake_type_A',
'fake_name1')
reading_2 = self._create_sensor_object('fake_value2',
'fake_type_A',
'fake_name2')
reading_3 = self._create_sensor_object('fake_value3',
'fake_type_B',
'fake_name3')
readings = [reading_1, reading_2, reading_3]
ipmicmd = ipmi_mock.return_value
ipmicmd.get_sensor_data.return_value = readings
expected = {
'fake_type_A': {
'fake_name1': {
'Health': '0',
'Sensor ID': 'fake_name1',
'Sensor Reading': 'fake_value1 fake_units',
'States': '[]',
'Units': 'fake_units'
},
'fake_name2': {
'Health': '0',
'Sensor ID': 'fake_name2',
'Sensor Reading': 'fake_value2 fake_units',
'States': '[]',
'Units': 'fake_units'
}
},
'fake_type_B': {
'fake_name3': {
'Health': '0',
'Sensor ID': 'fake_name3',
'Sensor Reading': 'fake_value3 fake_units',
'States': '[]', 'Units': 'fake_units'
}
}
}
ret = ipminative._get_sensors_data(self.info)
self.assertEqual(expected, ret)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__get_sensors_data_missing_values(self, ipmi_mock):
reading_1 = self._create_sensor_object('fake_value1',
'fake_type_A',
'fake_name1')
reading_2 = self._create_sensor_object(None,
'fake_type_A',
'fake_name2')
reading_3 = self._create_sensor_object(None,
'fake_type_B',
'fake_name3')
readings = [reading_1, reading_2, reading_3]
ipmicmd = ipmi_mock.return_value
ipmicmd.get_sensor_data.return_value = readings
expected = {
'fake_type_A': {
'fake_name1': {
'Health': '0',
'Sensor ID': 'fake_name1',
'Sensor Reading': 'fake_value1 fake_units',
'States': '[]',
'Units': 'fake_units'
}
}
}
ret = ipminative._get_sensors_data(self.info)
self.assertEqual(expected, ret)
def test__parse_raw_bytes_ok(self):
bytes_string = '0x11 0x12 0x25 0xFF'
netfn, cmd, data = ipminative._parse_raw_bytes(bytes_string)
self.assertEqual(0x11, netfn)
self.assertEqual(0x12, cmd)
self.assertEqual([0x25, 0xFF], data)
def test__parse_raw_bytes_invalid_value(self):
bytes_string = '0x11 oops'
self.assertRaises(exception.InvalidParameterValue,
ipminative._parse_raw_bytes,
bytes_string)
def test__parse_raw_bytes_missing_byte(self):
bytes_string = '0x11'
self.assertRaises(exception.InvalidParameterValue,
ipminative._parse_raw_bytes,
bytes_string)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__send_raw(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipminative._send_raw(self.info, '0x01 0x02 0x03 0x04')
ipmicmd.xraw_command.assert_called_once_with(1, 2, data=[3, 4])
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test__send_raw_fail(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.xraw_command.side_effect = pyghmi_exception.IpmiException()
self.assertRaises(exception.IPMIFailure, ipminative._send_raw,
self.info, '0x01 0x02')
class IPMINativeDriverTestCase(db_base.DbTestCase):
"""Test cases for ipminative.NativeIPMIPower class functions."""
def setUp(self):
super(IPMINativeDriverTestCase, self).setUp()
mgr_utils.mock_the_extension_manager(driver="fake_ipminative")
self.driver = driver_factory.get_driver("fake_ipminative")
self.node = obj_utils.create_test_node(self.context,
driver='fake_ipminative',
driver_info=INFO_DICT)
self.info = ipminative._parse_driver_info(self.node)
def test_get_properties(self):
expected = ipminative.COMMON_PROPERTIES
self.assertEqual(expected, self.driver.power.get_properties())
self.assertEqual(expected, self.driver.management.get_properties())
self.assertEqual(expected, self.driver.vendor.get_properties())
expected = list(ipminative.COMMON_PROPERTIES)
expected += list(ipminative.CONSOLE_PROPERTIES)
self.assertEqual(sorted(expected),
sorted(self.driver.console.get_properties().keys()))
self.assertEqual(sorted(expected),
sorted(self.driver.get_properties().keys()))
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_get_power_state(self, ipmi_mock):
# Getting the mocked command.
cmd_mock = ipmi_mock.return_value
# Getting the get power mock.
get_power_mock = cmd_mock.get_power
return_values = [{'powerstate': 'error'},
{'powerstate': 'on'},
{'powerstate': 'off'}]
get_power_mock.side_effect = lambda: return_values.pop()
with task_manager.acquire(self.context, self.node.uuid) as task:
pstate = self.driver.power.get_power_state(task)
self.assertEqual(states.POWER_OFF, pstate)
pstate = self.driver.power.get_power_state(task)
self.assertEqual(states.POWER_ON, pstate)
pstate = self.driver.power.get_power_state(task)
self.assertEqual(states.ERROR, pstate)
self.assertEqual(3, get_power_mock.call_count,
"pyghmi.ipmi.command.Command.get_power was not"
" called 3 times.")
@mock.patch.object(ipminative, '_power_on', autospec=True)
def test_set_power_on_ok(self, power_on_mock):
power_on_mock.return_value = states.POWER_ON
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.power.set_power_state(
task, states.POWER_ON)
power_on_mock.assert_called_once_with(self.info)
@mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
@mock.patch.object(ipminative, '_power_on', autospec=True)
def test_set_power_on_with_next_boot(self, power_on_mock, mock_next_boot):
power_on_mock.return_value = states.POWER_ON
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.power.set_power_state(
task, states.POWER_ON)
mock_next_boot.assert_called_once_with(task, self.info)
power_on_mock.assert_called_once_with(self.info)
@mock.patch.object(ipminative, '_power_off', autospec=True)
def test_set_power_off_ok(self, power_off_mock):
power_off_mock.return_value = states.POWER_OFF
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.power.set_power_state(
task, states.POWER_OFF)
power_off_mock.assert_called_once_with(self.info)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_set_power_on_fail(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.set_power.return_value = {'powerstate': 'error'}
self.config(retry_timeout=500, group='ipmi')
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.assertRaises(exception.PowerStateFailure,
self.driver.power.set_power_state,
task,
states.POWER_ON)
ipmicmd.set_power.assert_called_once_with('on', 500)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_set_boot_device_ok(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.set_bootdev.return_value = None
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.management.set_boot_device(task, boot_devices.PXE)
# PXE is converted to 'network' internally by ipminative
ipmicmd.set_bootdev.assert_called_once_with('network', persist=False,
uefiboot=False)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_force_set_boot_device_ok(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.set_bootdev.return_value = None
with task_manager.acquire(self.context,
self.node.uuid) as task:
task.node.driver_info['ipmi_force_boot_device'] = True
self.driver.management.set_boot_device(task, boot_devices.PXE)
task.node.refresh()
self.assertEqual(
False,
task.node.driver_internal_info['is_next_boot_persistent']
)
# PXE is converted to 'network' internally by ipminative
ipmicmd.set_bootdev.assert_called_once_with('network', persist=False,
uefiboot=False)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_set_boot_device_with_persistent(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.set_bootdev.return_value = None
with task_manager.acquire(self.context,
self.node.uuid) as task:
task.node.driver_info['ipmi_force_boot_device'] = True
self.driver.management.set_boot_device(task,
boot_devices.PXE,
True)
self.assertEqual(
boot_devices.PXE,
task.node.driver_internal_info['persistent_boot_device'])
# PXE is converted to 'network' internally by ipminative
ipmicmd.set_bootdev.assert_called_once_with('network', persist=False,
uefiboot=False)
def test_set_boot_device_bad_device(self):
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertRaises(exception.InvalidParameterValue,
self.driver.management.set_boot_device,
task,
'fake-device')
@mock.patch.object(deploy_utils, 'get_boot_mode_for_deploy')
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_set_boot_device_uefi(self, ipmi_mock, boot_mode_mock):
ipmicmd = ipmi_mock.return_value
boot_mode_mock.return_value = 'uefi'
with task_manager.acquire(self.context, self.node.uuid) as task:
self.driver.management.set_boot_device(task, boot_devices.PXE)
# PXE is converted to 'network' internally by ipminative
ipmicmd.set_bootdev.assert_called_once_with('network', persist=False,
uefiboot=True)
@mock.patch.object(deploy_utils, 'get_boot_mode_for_deploy')
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_set_boot_device_uefi_and_persistent(
self, ipmi_mock, boot_mode_mock):
ipmicmd = ipmi_mock.return_value
boot_mode_mock.return_value = 'uefi'
with task_manager.acquire(self.context, self.node.uuid) as task:
self.driver.management.set_boot_device(task, boot_devices.PXE,
persistent=True)
# PXE is converted to 'network' internally by ipminative
ipmicmd.set_bootdev.assert_called_once_with('network', persist=True,
uefiboot=True)
@mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
@mock.patch.object(ipminative, '_reboot', autospec=True)
def test_reboot_ok(self, reboot_mock, mock_next_boot):
reboot_mock.return_value = None
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.power.reboot(task)
mock_next_boot.assert_called_once_with(task, self.info)
reboot_mock.assert_called_once_with(self.info)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_reboot_fail(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.set_power.return_value = {'error': 'Some IPMI error'}
self.config(retry_timeout=500, group='ipmi')
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.assertRaises(exception.PowerStateFailure,
self.driver.power.reboot,
task)
ipmicmd.set_power.assert_called_once_with('boot', 500)
def test_management_interface_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, boot_devices.BIOS]
self.assertEqual(sorted(expected), sorted(task.driver.management.
get_supported_boot_devices(task)))
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_management_interface_get_boot_device_good(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_bootdev.return_value = {'bootdev': 'hd'}
with task_manager.acquire(self.context, self.node.uuid) as task:
bootdev = self.driver.management.get_boot_device(task)
self.assertEqual(boot_devices.DISK, bootdev['boot_device'])
self.assertIsNone(bootdev['persistent'])
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_management_interface_get_boot_device_persistent(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_bootdev.return_value = {'bootdev': 'hd',
'persistent': True}
with task_manager.acquire(self.context, self.node.uuid) as task:
bootdev = self.driver.management.get_boot_device(task)
self.assertEqual(boot_devices.DISK, bootdev['boot_device'])
self.assertTrue(bootdev['persistent'])
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_management_interface_get_boot_device_fail(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_bootdev.side_effect = pyghmi_exception.IpmiException
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertRaises(exception.IPMIFailure,
self.driver.management.get_boot_device, task)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_management_interface_get_boot_device_fail_dict(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_bootdev.return_value = {'error': 'boooom'}
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertRaises(exception.IPMIFailure,
self.driver.management.get_boot_device, task)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_management_interface_get_boot_device_unknown(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_bootdev.return_value = {'bootdev': 'unknown'}
with task_manager.acquire(self.context, self.node.uuid) as task:
expected = {'boot_device': None, 'persistent': None}
self.assertEqual(expected,
self.driver.management.get_boot_device(task))
def test_get_force_boot_device_persistent(self):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.driver_info['ipmi_force_boot_device'] = True
task.node.driver_internal_info['persistent_boot_device'] = 'pxe'
bootdev = self.driver.management.get_boot_device(task)
self.assertEqual('pxe', bootdev['boot_device'])
self.assertTrue(bootdev['persistent'])
def test_management_interface_validate_good(self):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.driver.management.validate(task)
def test_management_interface_validate_fail(self):
# Missing IPMI driver_info information
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
driver='fake_ipminative')
with task_manager.acquire(self.context, node.uuid) as task:
self.assertRaises(exception.MissingParameterValue,
task.driver.management.validate, task)
@mock.patch('pyghmi.ipmi.command.Command', autospec=True)
def test_get_sensors_data(self, ipmi_mock):
ipmicmd = ipmi_mock.return_value
ipmicmd.get_sensor_data.return_value = None
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.management.get_sensors_data(task)
ipmicmd.get_sensor_data.assert_called_once_with()
@mock.patch.object(console_utils, 'start_shellinabox_console',
autospec=True)
def test_start_console(self, mock_start):
mock_start.return_value = None
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.console.start_console(task)
mock_start.assert_called_once_with(self.info['uuid'],
self.info['port'],
mock.ANY)
self.assertTrue(mock_start.called)
@mock.patch.object(console_utils, 'start_shellinabox_console',
autospec=True)
def test_start_console_fail(self, mock_start):
mock_start.side_effect = exception.ConsoleSubprocessFailed(
error='error')
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.assertRaises(exception.ConsoleSubprocessFailed,
self.driver.console.start_console,
task)
self.assertTrue(mock_start.called)
@mock.patch.object(console_utils, 'stop_shellinabox_console',
autospec=True)
def test_stop_console(self, mock_stop):
mock_stop.return_value = None
with task_manager.acquire(self.context,
self.node['uuid']) as task:
self.driver.console.stop_console(task)
mock_stop.assert_called_once_with(self.info['uuid'])
@mock.patch.object(console_utils, 'stop_shellinabox_console',
autospec=True)
def test_stop_console_fail(self, mock_stop):
mock_stop.side_effect = exception.ConsoleError()
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.assertRaises(exception.ConsoleError,
self.driver.console.stop_console,
task)
mock_stop.assert_called_once_with(self.node.uuid)
@mock.patch.object(console_utils, 'get_shellinabox_console_url',
autospec=True)
def test_get_console(self, mock_get_url):
url = 'http://localhost:4201'
mock_get_url.return_value = url
expected = {'type': 'shellinabox', 'url': url}
with task_manager.acquire(self.context,
self.node.uuid) as task:
console_info = self.driver.console.get_console(task)
self.assertEqual(expected, console_info)
mock_get_url.assert_called_once_with(self.info['port'])
@mock.patch.object(ipminative, '_parse_driver_info', autospec=True)
@mock.patch.object(ipminative, '_parse_raw_bytes', autospec=True)
def test_vendor_passthru_validate__send_raw_bytes_good(self, mock_raw,
mock_driver):
with task_manager.acquire(self.context, self.node.uuid) as task:
self.driver.vendor.validate(task,
method='send_raw',
http_method='POST',
raw_bytes='0x00 0x01')
mock_raw.assert_called_once_with('0x00 0x01')
mock_driver.assert_called_once_with(task.node)
def test_vendor_passthru_validate__send_raw_bytes_fail(self):
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertRaises(exception.MissingParameterValue,
self.driver.vendor.validate,
task, method='send_raw')
def test_vendor_passthru_vendor_routes(self):
expected = ['send_raw', 'bmc_reset']
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
vendor_routes = task.driver.vendor.vendor_routes
self.assertIsInstance(vendor_routes, dict)
self.assertEqual(sorted(expected), sorted(vendor_routes))
@mock.patch.object(ipminative, '_send_raw', autospec=True)
def test_send_raw(self, send_raw_mock):
bytes = '0x00 0x01'
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.vendor.send_raw(task, http_method='POST',
raw_bytes=bytes)
send_raw_mock.assert_called_once_with(self.info, bytes)
@mock.patch.object(ipminative, '_send_raw', autospec=True)
def _test_bmc_reset(self, warm, expected_bytes, send_raw_mock):
with task_manager.acquire(self.context,
self.node.uuid) as task:
self.driver.vendor.bmc_reset(task, http_method='POST', warm=warm)
send_raw_mock.assert_called_once_with(self.info, expected_bytes)
def test_bmc_reset_cold(self):
for param in (False, 'false', 'off', 'n', 'no'):
self._test_bmc_reset(param, '0x06 0x02')
def test_bmc_reset_warm(self):
for param in (True, 'true', 'on', 'y', 'yes'):
self._test_bmc_reset(param, '0x06 0x03')

27
ironic/tests/unit/drivers/test_pxe.py
View File

@ -28,7 +28,6 @@ from ironic.drivers.modules.ilo import inspect as ilo_inspect
from ironic.drivers.modules.ilo import management as ilo_management
from ironic.drivers.modules.ilo import power as ilo_power
from ironic.drivers.modules.ilo import vendor as ilo_vendor
from ironic.drivers.modules import ipminative
from ironic.drivers.modules import ipmitool
from ironic.drivers.modules.irmc import management as irmc_management
from ironic.drivers.modules.irmc import power as irmc_power
@ -53,32 +52,6 @@ class PXEDriversTestCase(testtools.TestCase):
self.assertIsNone(driver.inspect)
self.assertIsInstance(driver.raid, agent.AgentRAID)
@mock.patch.object(pxe.importutils, 'try_import', spec_set=True,
autospec=True)
def test_pxe_ipminative_driver(self, try_import_mock):
try_import_mock.return_value = True
driver = pxe.PXEAndIPMINativeDriver()
self.assertIsInstance(driver.power, ipminative.NativeIPMIPower)
self.assertIsInstance(driver.console,
ipminative.NativeIPMIShellinaboxConsole)
self.assertIsInstance(driver.boot, pxe_module.PXEBoot)
self.assertIsInstance(driver.deploy, iscsi_deploy.ISCSIDeploy)
self.assertIsInstance(driver.management,
ipminative.NativeIPMIManagement)
self.assertIsInstance(driver.vendor, ipminative.VendorPassthru)
self.assertIsNone(driver.inspect)
self.assertIsInstance(driver.raid, agent.AgentRAID)
@mock.patch.object(pxe.importutils, 'try_import', spec_set=True,
autospec=True)
def test_pxe_ipminative_driver_import_error(self, try_import_mock):
try_import_mock.return_value = False
self.assertRaises(exception.DriverLoadError,
pxe.PXEAndIPMINativeDriver)
@mock.patch.object(pxe.importutils, 'try_import', spec_set=True,
autospec=True)
def test_pxe_ilo_driver(self, try_import_mock):

16
ironic/tests/unit/drivers/third_party_driver_mock_specs.py
View File

@ -57,22 +57,6 @@ PROLIANTUTILS_SPEC = (
'utils',
)
# pyghmi
PYGHMI_SPEC = (
'exceptions',
'ipmi',
)
PYGHMI_EXC_SPEC = (
'IpmiException',
)
PYGHMI_IPMI_SPEC = (
'command',
)
PYGHMI_IPMICMD_SPEC = (
'boot_devices',
'Command',
)
# pywsman
PYWSMAN_SPEC = (
'Client',

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

@ -22,7 +22,6 @@ respective external libraries' actually being present.
Any external library required by a third-party driver should be mocked here.
Current list of mocked libraries:
- ipminative
- proliantutils
- pysnmp
- scciclient
@ -49,26 +48,6 @@ ipmitool.TIMING_SUPPORT = False
ipmitool.DUAL_BRIDGE_SUPPORT = False
ipmitool.SINGLE_BRIDGE_SUPPORT = False
pyghmi = importutils.try_import("pyghmi")
if not pyghmi:
p = mock.MagicMock(spec_set=mock_specs.PYGHMI_SPEC)
p.exceptions = mock.MagicMock(spec_set=mock_specs.PYGHMI_EXC_SPEC)
p.exceptions.IpmiException = Exception
p.ipmi = mock.MagicMock(spec_set=mock_specs.PYGHMI_IPMI_SPEC)
p.ipmi.command = mock.MagicMock(spec_set=mock_specs.PYGHMI_IPMICMD_SPEC)
p.ipmi.command.Command = mock.MagicMock(spec_set=[])
sys.modules['pyghmi'] = p
sys.modules['pyghmi.exceptions'] = p.exceptions
sys.modules['pyghmi.ipmi'] = p.ipmi
sys.modules['pyghmi.ipmi.command'] = p.ipmi.command
# FIXME(deva): the next line is a hack, because several unit tests
# actually depend on this particular string being present
# in pyghmi.ipmi.command.boot_devices
p.ipmi.command.boot_devices = {'pxe': 4}
if 'ironic.drivers.modules.ipminative' in sys.modules:
six.moves.reload_module(sys.modules['ironic.drivers.modules.ipminative'])
proliantutils = importutils.try_import('proliantutils')
if not proliantutils:
proliantutils = mock.MagicMock(spec_set=mock_specs.PROLIANTUTILS_SPEC)

5
releasenotes/notes/remove-ipminative-driver-3367d25bbcc41fdc.yaml Normal file
View File

@ -0,0 +1,5 @@
upgrade:
- |
The agent_pyghmi, pxe_ipminative, and fake_ipminative drivers have all
been removed from ironic due to lack of testing. Nodes using these
drivers should be changed to the agent_ipmitool or pxe_ipmitool driver.

3
setup.cfg
View File

@ -47,7 +47,6 @@ ironic.drivers =
agent_ipmitool_socat = ironic.drivers.ipmi:AgentAndIPMIToolAndSocatDriver
agent_irmc = ironic.drivers.irmc:IRMCVirtualMediaAgentDriver
agent_pxe_oneview = ironic.drivers.oneview:AgentPXEOneViewDriver
agent_pyghmi = ironic.drivers.agent:AgentAndIPMINativeDriver
agent_ssh = ironic.drivers.agent:AgentAndSSHDriver
agent_ucs = ironic.drivers.agent:AgentAndUcsDriver
fake = ironic.drivers.fake:FakeDriver
@ -56,7 +55,6 @@ ironic.drivers =
fake_inspector = ironic.drivers.fake:FakeIPMIToolInspectorDriver
fake_ipmitool = ironic.drivers.fake:FakeIPMIToolDriver
fake_ipmitool_socat = ironic.drivers.fake:FakeIPMIToolSocatDriver
fake_ipminative = ironic.drivers.fake:FakeIPMINativeDriver
fake_ssh = ironic.drivers.fake:FakeSSHDriver
fake_pxe = ironic.drivers.fake:FakePXEDriver
fake_ilo = ironic.drivers.fake:FakeIloDriver
@ -71,7 +69,6 @@ ironic.drivers =
iscsi_pxe_oneview = ironic.drivers.oneview:ISCSIPXEOneViewDriver
pxe_ipmitool = ironic.drivers.ipmi:PXEAndIPMIToolDriver
pxe_ipmitool_socat = ironic.drivers.ipmi:PXEAndIPMIToolAndSocatDriver
pxe_ipminative = ironic.drivers.pxe:PXEAndIPMINativeDriver
pxe_ssh = ironic.drivers.pxe:PXEAndSSHDriver
pxe_ilo = ironic.drivers.pxe:PXEAndIloDriver
pxe_drac = ironic.drivers.drac:PXEDracDriver