Merge "IPMI double bridging functionality"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
0cc2702538
|
|
@ -73,7 +73,22 @@ OPTIONAL_PROPERTIES = {
|
|||
'ipmi_password': _("password. Optional."),
|
||||
'ipmi_priv_level': _("privilege level; default is ADMINISTRATOR. One of "
|
||||
"%s. Optional.") % ', '.join(VALID_PRIV_LEVELS),
|
||||
'ipmi_username': _("username; default is NULL user. Optional.")
|
||||
'ipmi_username': _("username; default is NULL user. Optional."),
|
||||
'ipmi_bridging': _("bridging_type; default is \"no\". One of \"single\", "
|
||||
"\"dual\", \"no\". Optional."),
|
||||
'ipmi_transit_channel': _("transit channel for bridged request. Required "
|
||||
"only if ipmi_bridging is set to \"dual\"."),
|
||||
'ipmi_transit_address': _("transit address for bridged request. Required "
|
||||
"only if ipmi_bridging is set to \"dual\"."),
|
||||
'ipmi_target_channel': _("destination channel for bridged request. "
|
||||
"Required only if ipmi_bridging is set to "
|
||||
"\"single\" or \"dual\"."),
|
||||
'ipmi_target_address': _("destination address for bridged request. "
|
||||
"Required only if ipmi_bridging is set "
|
||||
"to \"single\" or \"dual\"."),
|
||||
'ipmi_local_address': _("local IPMB address for bridged requests. "
|
||||
"Used only if ipmi_bridging is set "
|
||||
"to \"single\" or \"dual\". Optional.")
|
||||
}
|
||||
COMMON_PROPERTIES = REQUIRED_PROPERTIES.copy()
|
||||
COMMON_PROPERTIES.update(OPTIONAL_PROPERTIES)
|
||||
|
|
@ -81,49 +96,76 @@ CONSOLE_PROPERTIES = {
|
|||
'ipmi_terminal_port': _("node's UDP port to connect to. Only required for "
|
||||
"console access.")
|
||||
}
|
||||
BRIDGING_OPTIONS = [('local_address', '-m'),
|
||||
('transit_channel', '-B'), ('transit_address', '-T'),
|
||||
('target_channel', '-b'), ('target_address', '-t')]
|
||||
|
||||
LAST_CMD_TIME = {}
|
||||
TIMING_SUPPORT = None
|
||||
SINGLE_BRIDGE_SUPPORT = None
|
||||
DUAL_BRIDGE_SUPPORT = None
|
||||
|
||||
|
||||
def _is_timing_supported(is_supported=None):
|
||||
# shim to allow module variable to be mocked in unit tests
|
||||
global TIMING_SUPPORT
|
||||
|
||||
if (TIMING_SUPPORT is None) and (is_supported is not None):
|
||||
TIMING_SUPPORT = is_supported
|
||||
return TIMING_SUPPORT
|
||||
ipmitool_command_options = {
|
||||
'timing': ['ipmitool', '-N', '0', '-R', '0', '-h'],
|
||||
'single_bridge': ['ipmitool', '-m', '0', '-b', '0', '-t', '0', '-h'],
|
||||
'dual_bridge': ['ipmitool', '-m', '0', '-b', '0', '-t', '0',
|
||||
'-B', '0', '-T', '0', '-h']}
|
||||
|
||||
|
||||
def check_timing_support():
|
||||
"""Check the installed version of ipmitool for -N -R option support.
|
||||
def _check_option_support(options):
|
||||
"""Checks if the specific ipmitool options are supported on host.
|
||||
|
||||
Support was added in 1.8.12 for the -N -R options, which enable
|
||||
more precise control over timing of ipmi packets. Prior to this,
|
||||
the default behavior was to retry each command up to 18 times at
|
||||
1 to 5 second intervals.
|
||||
http://ipmitool.cvs.sourceforge.net/viewvc/ipmitool/ipmitool/ChangeLog?revision=1.37 # noqa
|
||||
This method updates the module-level variables indicating whether
|
||||
an option is supported so that it is accessible by any driver
|
||||
interface class in this module. It is intended to be called from
|
||||
the __init__ method of such classes only.
|
||||
|
||||
This method updates the module-level TIMING_SUPPORT variable so that
|
||||
it is accessible by any driver interface class in this module. It is
|
||||
intended to be called from the __init__ method of such classes only.
|
||||
|
||||
:returns: boolean indicating whether support for -N -R is present
|
||||
:param options: list of ipmitool options to be checked
|
||||
:raises: OSError
|
||||
"""
|
||||
if _is_timing_supported() is None:
|
||||
# Directly check ipmitool for support of -N and -R options. Because
|
||||
# of the way ipmitool processes' command line options, if the local
|
||||
# ipmitool does not support setting the timing options, the command
|
||||
# below will fail.
|
||||
try:
|
||||
out, err = utils.execute(*['ipmitool', '-N', '0', '-R', '0', '-h'])
|
||||
except processutils.ProcessExecutionError:
|
||||
# the local ipmitool does not support the -N and -R options.
|
||||
_is_timing_supported(False)
|
||||
else:
|
||||
# looks like ipmitool supports timing options.
|
||||
_is_timing_supported(True)
|
||||
for opt in options:
|
||||
if _is_option_supported(opt) is None:
|
||||
try:
|
||||
cmd = ipmitool_command_options[opt]
|
||||
out, err = utils.execute(*cmd)
|
||||
except processutils.ProcessExecutionError:
|
||||
# the local ipmitool does not support the command.
|
||||
_is_option_supported(opt, False)
|
||||
else:
|
||||
# looks like ipmitool supports the command.
|
||||
_is_option_supported(opt, True)
|
||||
|
||||
|
||||
def _is_option_supported(option, is_supported=None):
|
||||
"""Indicates whether the particular ipmitool option is supported.
|
||||
|
||||
:param option: specific ipmitool option
|
||||
:param is_supported: Optional Boolean. when specified, this value
|
||||
is assigned to the module-level variable indicating
|
||||
whether the option is supported. Used only if a value
|
||||
is not already assigned.
|
||||
:returns: True, indicates the option is supported
|
||||
:returns: False, indicates the option is not supported
|
||||
:returns: None, indicates that it is not aware whether the option
|
||||
is supported
|
||||
"""
|
||||
global SINGLE_BRIDGE_SUPPORT
|
||||
global DUAL_BRIDGE_SUPPORT
|
||||
global TIMING_SUPPORT
|
||||
|
||||
if option == 'single_bridge':
|
||||
if (SINGLE_BRIDGE_SUPPORT is None) and (is_supported is not None):
|
||||
SINGLE_BRIDGE_SUPPORT = is_supported
|
||||
return SINGLE_BRIDGE_SUPPORT
|
||||
elif option == 'dual_bridge':
|
||||
if (DUAL_BRIDGE_SUPPORT is None) and (is_supported is not None):
|
||||
DUAL_BRIDGE_SUPPORT = is_supported
|
||||
return DUAL_BRIDGE_SUPPORT
|
||||
elif option == 'timing':
|
||||
if (TIMING_SUPPORT is None) and (is_supported is not None):
|
||||
TIMING_SUPPORT = is_supported
|
||||
return TIMING_SUPPORT
|
||||
|
||||
|
||||
def _console_pwfile_path(uuid):
|
||||
|
|
@ -164,6 +206,7 @@ def _parse_driver_info(node):
|
|||
|
||||
"""
|
||||
info = node.driver_info or {}
|
||||
bridging_types = ['single', 'dual']
|
||||
missing_info = [key for key in REQUIRED_PROPERTIES if not info.get(key)]
|
||||
if missing_info:
|
||||
raise exception.MissingParameterValue(_(
|
||||
|
|
@ -176,6 +219,12 @@ def _parse_driver_info(node):
|
|||
password = info.get('ipmi_password')
|
||||
port = info.get('ipmi_terminal_port')
|
||||
priv_level = info.get('ipmi_priv_level', 'ADMINISTRATOR')
|
||||
bridging_type = info.get('ipmi_bridging', 'no')
|
||||
local_address = info.get('ipmi_local_address')
|
||||
transit_channel = info.get('ipmi_transit_channel')
|
||||
transit_address = info.get('ipmi_transit_address')
|
||||
target_channel = info.get('ipmi_target_channel')
|
||||
target_address = info.get('ipmi_target_address')
|
||||
|
||||
if port:
|
||||
try:
|
||||
|
|
@ -184,6 +233,46 @@ def _parse_driver_info(node):
|
|||
raise exception.InvalidParameterValue(_(
|
||||
"IPMI terminal port is not an integer."))
|
||||
|
||||
# check if ipmi_bridging has proper value
|
||||
if bridging_type == 'no':
|
||||
# if bridging is not selected, then set all bridging params to None
|
||||
local_address = transit_channel = transit_address = \
|
||||
target_channel = target_address = None
|
||||
elif bridging_type in bridging_types:
|
||||
# check if the particular bridging option is supported on host
|
||||
if not _is_option_supported('%s_bridge' % bridging_type):
|
||||
raise exception.InvalidParameterValue(_(
|
||||
"Value for ipmi_bridging is provided as %s, but IPMI "
|
||||
"bridging is not supported by the IPMI utility installed "
|
||||
"on host. Ensure ipmitool version is > 1.8.11"
|
||||
) % bridging_type)
|
||||
|
||||
# ensure that all the required parameters are provided
|
||||
params_undefined = [param for param, value in [
|
||||
("ipmi_target_channel", target_channel),
|
||||
('ipmi_target_address', target_address)] if value is None]
|
||||
if bridging_type == 'dual':
|
||||
params_undefined2 = [param for param, value in [
|
||||
("ipmi_transit_channel", transit_channel),
|
||||
('ipmi_transit_address', transit_address)
|
||||
] if value is None]
|
||||
params_undefined.extend(params_undefined2)
|
||||
else:
|
||||
# if single bridging was selected, set dual bridge params to None
|
||||
transit_channel = transit_address = None
|
||||
|
||||
# If the required parameters were not provided,
|
||||
# raise an exception
|
||||
if params_undefined:
|
||||
raise exception.MissingParameterValue(_(
|
||||
"%(param)s not provided") % {'param': params_undefined})
|
||||
else:
|
||||
raise exception.InvalidParameterValue(_(
|
||||
"Invalid value for ipmi_bridging: %(bridging_type)s,"
|
||||
" the valid value can be one of: %(bridging_types)s"
|
||||
) % {'bridging_type': bridging_type,
|
||||
'bridging_types': bridging_types + ['no']})
|
||||
|
||||
if priv_level not in VALID_PRIV_LEVELS:
|
||||
valid_priv_lvls = ', '.join(VALID_PRIV_LEVELS)
|
||||
raise exception.InvalidParameterValue(_(
|
||||
|
|
@ -197,8 +286,13 @@ def _parse_driver_info(node):
|
|||
'password': password,
|
||||
'port': port,
|
||||
'uuid': node.uuid,
|
||||
'priv_level': priv_level
|
||||
}
|
||||
'priv_level': priv_level,
|
||||
'local_address': local_address,
|
||||
'transit_channel': transit_channel,
|
||||
'transit_address': transit_address,
|
||||
'target_channel': target_channel,
|
||||
'target_address': target_address
|
||||
}
|
||||
|
||||
|
||||
def _exec_ipmitool(driver_info, command):
|
||||
|
|
@ -221,13 +315,17 @@ def _exec_ipmitool(driver_info, command):
|
|||
driver_info['address'],
|
||||
'-L', driver_info['priv_level']
|
||||
]
|
||||
|
||||
if driver_info['username']:
|
||||
args.append('-U')
|
||||
args.append(driver_info['username'])
|
||||
|
||||
for name, option in BRIDGING_OPTIONS:
|
||||
if driver_info[name] is not None:
|
||||
args.append(option)
|
||||
args.append(driver_info[name])
|
||||
|
||||
# specify retry timing more precisely, if supported
|
||||
if _is_timing_supported():
|
||||
if _is_option_supported('timing'):
|
||||
num_tries = max(
|
||||
(CONF.ipmi.retry_timeout // CONF.ipmi.min_command_interval), 1)
|
||||
args.append('-R')
|
||||
|
|
@ -453,12 +551,12 @@ class IPMIPower(base.PowerInterface):
|
|||
|
||||
def __init__(self):
|
||||
try:
|
||||
check_timing_support()
|
||||
_check_option_support(['timing', 'single_bridge', 'dual_bridge'])
|
||||
except OSError:
|
||||
raise exception.DriverLoadError(
|
||||
driver=self.__class__.__name__,
|
||||
reason="Unable to locate usable ipmitool command in "
|
||||
"the system path when checking ipmitool version")
|
||||
reason=_("Unable to locate usable ipmitool command in "
|
||||
"the system path when checking ipmitool version"))
|
||||
|
||||
def get_properties(self):
|
||||
return COMMON_PROPERTIES
|
||||
|
|
@ -544,12 +642,12 @@ class IPMIManagement(base.ManagementInterface):
|
|||
|
||||
def __init__(self):
|
||||
try:
|
||||
check_timing_support()
|
||||
_check_option_support(['timing', 'single_bridge', 'dual_bridge'])
|
||||
except OSError:
|
||||
raise exception.DriverLoadError(
|
||||
driver=self.__class__.__name__,
|
||||
reason="Unable to locate usable ipmitool command in "
|
||||
"the system path when checking ipmitool version")
|
||||
reason=_("Unable to locate usable ipmitool command in "
|
||||
"the system path when checking ipmitool version"))
|
||||
|
||||
def validate(self, task):
|
||||
"""Check that 'driver_info' contains IPMI credentials.
|
||||
|
|
@ -685,6 +783,15 @@ class IPMIManagement(base.ManagementInterface):
|
|||
|
||||
class VendorPassthru(base.VendorInterface):
|
||||
|
||||
def __init__(self):
|
||||
try:
|
||||
_check_option_support(['single_bridge', 'dual_bridge'])
|
||||
except OSError:
|
||||
raise exception.DriverLoadError(
|
||||
driver=self.__class__.__name__,
|
||||
reason=_("Unable to locate usable ipmitool command in "
|
||||
"the system path when checking ipmitool version"))
|
||||
|
||||
@task_manager.require_exclusive_lock
|
||||
def _send_raw_bytes(self, task, raw_bytes):
|
||||
"""Send raw bytes to the BMC. Bytes should be a string of bytes.
|
||||
|
|
@ -813,12 +920,12 @@ class IPMIShellinaboxConsole(base.ConsoleInterface):
|
|||
|
||||
def __init__(self):
|
||||
try:
|
||||
check_timing_support()
|
||||
_check_option_support(['timing', 'single_bridge', 'dual_bridge'])
|
||||
except OSError:
|
||||
raise exception.DriverLoadError(
|
||||
driver=self.__class__.__name__,
|
||||
reason="Unable to locate usable ipmitool command in "
|
||||
"the system path when checking ipmitool version")
|
||||
reason=_("Unable to locate usable ipmitool command in "
|
||||
"the system path when checking ipmitool version"))
|
||||
|
||||
def get_properties(self):
|
||||
d = COMMON_PROPERTIES.copy()
|
||||
|
|
@ -862,6 +969,12 @@ class IPMIShellinaboxConsole(base.ConsoleInterface):
|
|||
'address': driver_info['address'],
|
||||
'user': driver_info['username'],
|
||||
'pwfile': pw_file}
|
||||
|
||||
for name, option in BRIDGING_OPTIONS:
|
||||
if driver_info[name] is not None:
|
||||
ipmi_cmd = " ".join([ipmi_cmd,
|
||||
option, driver_info[name]])
|
||||
|
||||
if CONF.debug:
|
||||
ipmi_cmd += " -v"
|
||||
ipmi_cmd += " sol activate"
|
||||
|
|
|
|||
|
|
@ -2109,7 +2109,11 @@ class ManagerTestProperties(tests_db_base.DbTestCase):
|
|||
def test_driver_properties_fake_ipmitool(self):
|
||||
expected = ['ipmi_address', 'ipmi_terminal_port',
|
||||
'ipmi_password', 'ipmi_priv_level',
|
||||
'ipmi_username']
|
||||
'ipmi_username', 'ipmi_bridging',
|
||||
'ipmi_transit_channel', 'ipmi_transit_address',
|
||||
'ipmi_target_channel', 'ipmi_target_address',
|
||||
'ipmi_local_address'
|
||||
]
|
||||
self._check_driver_properties("fake_ipmitool", expected)
|
||||
|
||||
def test_driver_properties_fake_ipminative(self):
|
||||
|
|
@ -2134,9 +2138,12 @@ class ManagerTestProperties(tests_db_base.DbTestCase):
|
|||
|
||||
def test_driver_properties_pxe_ipmitool(self):
|
||||
expected = ['ipmi_address', 'ipmi_terminal_port',
|
||||
'pxe_deploy_kernel', 'pxe_deploy_ramdisk',
|
||||
'ipmi_password', 'ipmi_priv_level',
|
||||
'ipmi_username']
|
||||
'ipmi_username', 'ipmi_bridging', 'ipmi_transit_channel',
|
||||
'ipmi_transit_address', 'ipmi_target_channel',
|
||||
'ipmi_target_address', 'ipmi_local_address',
|
||||
'pxe_deploy_kernel', 'pxe_deploy_ramdisk'
|
||||
]
|
||||
self._check_driver_properties("pxe_ipmitool", expected)
|
||||
|
||||
def test_driver_properties_pxe_ipminative(self):
|
||||
|
|
|
|||
|
|
@ -21,7 +21,18 @@ def get_test_ipmi_info():
|
|||
return {
|
||||
"ipmi_address": "1.2.3.4",
|
||||
"ipmi_username": "admin",
|
||||
"ipmi_password": "fake",
|
||||
"ipmi_password": "fake"
|
||||
}
|
||||
|
||||
|
||||
def get_test_ipmi_bridging_parameters():
|
||||
return {
|
||||
"ipmi_bridging": "dual",
|
||||
"ipmi_local_address": "0x20",
|
||||
"ipmi_transit_channel": "0",
|
||||
"ipmi_transit_address": "0x82",
|
||||
"ipmi_target_channel": "7",
|
||||
"ipmi_target_address": "0x72"
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -52,41 +52,164 @@ CONF.import_opt('min_command_interval',
|
|||
|
||||
INFO_DICT = db_utils.get_test_ipmi_info()
|
||||
|
||||
# BRIDGE_INFO_DICT will have all the bridging parameters appended
|
||||
BRIDGE_INFO_DICT = INFO_DICT.copy()
|
||||
BRIDGE_INFO_DICT.update(db_utils.get_test_ipmi_bridging_parameters())
|
||||
|
||||
class IPMIToolCheckTimingTestCase(base.TestCase):
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
class IPMIToolCheckOptionSupportedTestCase(base.TestCase):
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_timing_pass(self, mock_exc, mock_timing):
|
||||
def test_check_timing_pass(self, mock_exc, mock_support):
|
||||
mock_exc.return_value = (None, None)
|
||||
mock_timing.return_value = None
|
||||
expected = [mock.call(), mock.call(True)]
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('timing'),
|
||||
mock.call('timing', True)]
|
||||
|
||||
ipmi.check_timing_support()
|
||||
ipmi._check_option_support(['timing'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_timing.call_args_list)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_timing_fail(self, mock_exc, mock_timing):
|
||||
def test_check_timing_fail(self, mock_exc, mock_support):
|
||||
mock_exc.side_effect = processutils.ProcessExecutionError()
|
||||
mock_timing.return_value = None
|
||||
expected = [mock.call(), mock.call(False)]
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('timing'),
|
||||
mock.call('timing', False)]
|
||||
|
||||
ipmi.check_timing_support()
|
||||
ipmi._check_option_support(['timing'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_timing.call_args_list)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_timing_no_ipmitool(self, mock_exc, mock_timing):
|
||||
def test_check_timing_no_ipmitool(self, mock_exc, mock_support):
|
||||
mock_exc.side_effect = OSError()
|
||||
mock_timing.return_value = None
|
||||
expected = [mock.call()]
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('timing')]
|
||||
|
||||
self.assertRaises(OSError, ipmi.check_timing_support)
|
||||
self.assertRaises(OSError, ipmi._check_option_support, ['timing'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_timing.call_args_list)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_single_bridge_pass(self, mock_exc, mock_support):
|
||||
mock_exc.return_value = (None, None)
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('single_bridge'),
|
||||
mock.call('single_bridge', True)]
|
||||
|
||||
ipmi._check_option_support(['single_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_single_bridge_fail(self, mock_exc, mock_support):
|
||||
mock_exc.side_effect = processutils.ProcessExecutionError()
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('single_bridge'),
|
||||
mock.call('single_bridge', False)]
|
||||
|
||||
ipmi._check_option_support(['single_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_single_bridge_no_ipmitool(self, mock_exc,
|
||||
mock_support):
|
||||
mock_exc.side_effect = OSError()
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('single_bridge')]
|
||||
|
||||
self.assertRaises(OSError, ipmi._check_option_support,
|
||||
['single_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_dual_bridge_pass(self, mock_exc, mock_support):
|
||||
mock_exc.return_value = (None, None)
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('dual_bridge'),
|
||||
mock.call('dual_bridge', True)]
|
||||
|
||||
ipmi._check_option_support(['dual_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_dual_bridge_fail(self, mock_exc, mock_support):
|
||||
mock_exc.side_effect = processutils.ProcessExecutionError()
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('dual_bridge'),
|
||||
mock.call('dual_bridge', False)]
|
||||
|
||||
ipmi._check_option_support(['dual_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_dual_bridge_no_ipmitool(self, mock_exc, mock_support):
|
||||
mock_exc.side_effect = OSError()
|
||||
mock_support.return_value = None
|
||||
expected = [mock.call('dual_bridge')]
|
||||
|
||||
self.assertRaises(OSError, ipmi._check_option_support,
|
||||
['dual_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_all_options_pass(self, mock_exc, mock_support):
|
||||
mock_exc.return_value = (None, None)
|
||||
mock_support.return_value = None
|
||||
expected = [
|
||||
mock.call('timing'), mock.call('timing', True),
|
||||
mock.call('single_bridge'),
|
||||
mock.call('single_bridge', True),
|
||||
mock.call('dual_bridge'), mock.call('dual_bridge', True)]
|
||||
|
||||
ipmi._check_option_support(['timing', 'single_bridge', 'dual_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_all_options_fail(self, mock_exc, mock_support):
|
||||
mock_exc.side_effect = processutils.ProcessExecutionError()
|
||||
mock_support.return_value = None
|
||||
expected = [
|
||||
mock.call('timing'), mock.call('timing', False),
|
||||
mock.call('single_bridge'),
|
||||
mock.call('single_bridge', False),
|
||||
mock.call('dual_bridge'),
|
||||
mock.call('dual_bridge', False)]
|
||||
|
||||
ipmi._check_option_support(['timing', 'single_bridge', 'dual_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(utils, 'execute')
|
||||
def test_check_all_options_no_ipmitool(self, mock_exc, mock_support):
|
||||
mock_exc.side_effect = OSError()
|
||||
mock_support.return_value = None
|
||||
# exception is raised once ipmitool was not found for an command
|
||||
expected = [mock.call('timing')]
|
||||
|
||||
self.assertRaises(OSError, ipmi._check_option_support,
|
||||
['timing', 'single_bridge', 'dual_bridge'])
|
||||
self.assertTrue(mock_exc.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
|
||||
|
||||
@mock.patch.object(time, 'sleep')
|
||||
|
|
@ -113,10 +236,9 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
|
||||
def test__parse_driver_info(self, mock_sleep):
|
||||
# make sure we get back the expected things
|
||||
self.assertIsNotNone(self.info.get('address'))
|
||||
self.assertIsNotNone(self.info.get('username'))
|
||||
self.assertIsNotNone(self.info.get('password'))
|
||||
self.assertIsNotNone(self.info.get('uuid'))
|
||||
_OPTIONS = ['address', 'username', 'password', 'uuid']
|
||||
for option in _OPTIONS:
|
||||
self.assertIsNotNone(self.info.get(option))
|
||||
|
||||
info = dict(INFO_DICT)
|
||||
|
||||
|
|
@ -149,11 +271,170 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
ipmi._parse_driver_info,
|
||||
node)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
def test__parse_driver_info_with_invalid_bridging_type(self,
|
||||
mock_support, mock_sleep):
|
||||
info = BRIDGE_INFO_DICT.copy()
|
||||
# make sure error is raised when ipmi_bridging has unexpected value
|
||||
info['ipmi_bridging'] = 'junk'
|
||||
node = obj_utils.get_test_node(self.context, driver_info=info)
|
||||
self.assertRaises(exception.InvalidParameterValue,
|
||||
ipmi._parse_driver_info,
|
||||
node)
|
||||
self.assertFalse(mock_support.called)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
def test__parse_driver_info_with_no_bridging(self,
|
||||
mock_support, mock_sleep):
|
||||
_OPTIONS = ['address', 'username', 'password', 'uuid']
|
||||
_BRIDGING_OPTIONS = ['local_address', 'transit_channel',
|
||||
'transit_address',
|
||||
'target_channel', 'target_address']
|
||||
info = BRIDGE_INFO_DICT.copy()
|
||||
info['ipmi_bridging'] = 'no'
|
||||
node = obj_utils.get_test_node(self.context, driver='fake_ipmitool',
|
||||
driver_info=info)
|
||||
ret = ipmi._parse_driver_info(node)
|
||||
|
||||
# ensure that _is_option_supported was not called
|
||||
self.assertFalse(mock_support.called)
|
||||
# check if we got all the required options
|
||||
for option in _OPTIONS:
|
||||
self.assertIsNotNone(ret[option])
|
||||
# test the default value for 'priv_level'
|
||||
self.assertEqual('ADMINISTRATOR', ret['priv_level'])
|
||||
|
||||
# check if bridging parameters were set to None
|
||||
for option in _BRIDGING_OPTIONS:
|
||||
self.assertIsNone(ret[option])
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
def test__parse_driver_info_with_dual_bridging_pass(self,
|
||||
mock_support, mock_sleep):
|
||||
_OPTIONS = ['address', 'username', 'password', 'uuid',
|
||||
'local_address', 'transit_channel', 'transit_address',
|
||||
'target_channel', 'target_address']
|
||||
node = obj_utils.get_test_node(self.context, driver='fake_ipmitool',
|
||||
driver_info=BRIDGE_INFO_DICT)
|
||||
|
||||
expected = [mock.call('dual_bridge')]
|
||||
|
||||
# test double bridging and make sure we get back expected result
|
||||
mock_support.return_value = True
|
||||
ret = ipmi._parse_driver_info(node)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
for option in _OPTIONS:
|
||||
self.assertIsNotNone(ret[option])
|
||||
# test the default value for 'priv_level'
|
||||
self.assertEqual('ADMINISTRATOR', ret['priv_level'])
|
||||
|
||||
info = BRIDGE_INFO_DICT.copy()
|
||||
# ipmi_local_address / ipmi_username / ipmi_password are not mandatory
|
||||
for optional_arg in ['ipmi_local_address', 'ipmi_username',
|
||||
'ipmi_password']:
|
||||
del info[optional_arg]
|
||||
node = obj_utils.get_test_node(self.context, driver_info=info)
|
||||
ipmi._parse_driver_info(node)
|
||||
self.assertEqual(mock.call('dual_bridge'), mock_support.call_args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
def test__parse_driver_info_with_dual_bridging_not_supported(self,
|
||||
mock_support, mock_sleep):
|
||||
node = obj_utils.get_test_node(self.context, driver='fake_ipmitool',
|
||||
driver_info=BRIDGE_INFO_DICT)
|
||||
# if dual bridge is not supported then check if error is raised
|
||||
mock_support.return_value = False
|
||||
self.assertRaises(exception.InvalidParameterValue,
|
||||
ipmi._parse_driver_info, node)
|
||||
mock_support.assert_called_once_with('dual_bridge')
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
def test__parse_driver_info_with_dual_bridging_missing_parameters(self,
|
||||
mock_support, mock_sleep):
|
||||
info = BRIDGE_INFO_DICT.copy()
|
||||
mock_support.return_value = True
|
||||
# make sure error is raised when dual bridging is selected and the
|
||||
# required parameters for dual bridging are not provided
|
||||
for param in ['ipmi_transit_channel', 'ipmi_target_address',
|
||||
'ipmi_transit_address', 'ipmi_target_channel']:
|
||||
del info[param]
|
||||
node = obj_utils.get_test_node(self.context, driver_info=info)
|
||||
self.assertRaises(exception.MissingParameterValue,
|
||||
ipmi._parse_driver_info, node)
|
||||
self.assertEqual(mock.call('dual_bridge'),
|
||||
mock_support.call_args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
def test__parse_driver_info_with_single_bridging_pass(self,
|
||||
mock_support, mock_sleep):
|
||||
_OPTIONS = ['address', 'username', 'password', 'uuid',
|
||||
'local_address', 'target_channel', 'target_address']
|
||||
|
||||
info = BRIDGE_INFO_DICT.copy()
|
||||
info['ipmi_bridging'] = 'single'
|
||||
node = obj_utils.get_test_node(self.context, driver='fake_ipmitool',
|
||||
driver_info=info)
|
||||
|
||||
expected = [mock.call('single_bridge')]
|
||||
|
||||
# test single bridging and make sure we get back expected things
|
||||
mock_support.return_value = True
|
||||
ret = ipmi._parse_driver_info(node)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
for option in _OPTIONS:
|
||||
self.assertIsNotNone(ret[option])
|
||||
# test the default value for 'priv_level'
|
||||
self.assertEqual('ADMINISTRATOR', ret['priv_level'])
|
||||
|
||||
# check if dual bridge params are set to None
|
||||
self.assertIsNone(ret['transit_channel'])
|
||||
self.assertIsNone(ret['transit_address'])
|
||||
|
||||
# ipmi_local_address / ipmi_username / ipmi_password are not mandatory
|
||||
for optional_arg in ['ipmi_local_address', 'ipmi_username',
|
||||
'ipmi_password']:
|
||||
del info[optional_arg]
|
||||
node = obj_utils.get_test_node(self.context, driver_info=info)
|
||||
ipmi._parse_driver_info(node)
|
||||
self.assertEqual(mock.call('single_bridge'),
|
||||
mock_support.call_args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
def test__parse_driver_info_with_single_bridging_not_supported(self,
|
||||
mock_support, mock_sleep):
|
||||
info = BRIDGE_INFO_DICT.copy()
|
||||
info['ipmi_bridging'] = 'single'
|
||||
node = obj_utils.get_test_node(self.context, driver='fake_ipmitool',
|
||||
driver_info=info)
|
||||
|
||||
# if single bridge is not supported then check if error is raised
|
||||
mock_support.return_value = False
|
||||
self.assertRaises(exception.InvalidParameterValue,
|
||||
ipmi._parse_driver_info, node)
|
||||
mock_support.assert_called_once_with('single_bridge')
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
def test__parse_driver_info_with_single_bridging_missing_parameters(
|
||||
self, mock_support, mock_sleep):
|
||||
info = dict(BRIDGE_INFO_DICT)
|
||||
info['ipmi_bridging'] = 'single'
|
||||
mock_support.return_value = True
|
||||
# make sure error is raised when single bridging is selected and the
|
||||
# required parameters for single bridging are not provided
|
||||
for param in ['ipmi_target_channel', 'ipmi_target_address']:
|
||||
del info[param]
|
||||
node = obj_utils.get_test_node(self.context, driver_info=info)
|
||||
self.assertRaises(exception.MissingParameterValue,
|
||||
ipmi._parse_driver_info,
|
||||
node)
|
||||
self.assertEqual(mock.call('single_bridge'),
|
||||
mock_support.call_args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_first_call_to_address(self, mock_exec, mock_pwf,
|
||||
mock_timing_support, mock_sleep):
|
||||
mock_support, mock_sleep):
|
||||
ipmi.LAST_CMD_TIME = {}
|
||||
pw_file_handle = tempfile.NamedTemporaryFile()
|
||||
pw_file = pw_file_handle.name
|
||||
|
|
@ -168,21 +449,22 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
'A', 'B', 'C',
|
||||
]
|
||||
|
||||
mock_timing_support.return_value = False
|
||||
mock_support.return_value = False
|
||||
mock_pwf.return_value = file_handle
|
||||
mock_exec.return_value = (None, None)
|
||||
|
||||
ipmi._exec_ipmitool(self.info, 'A B C')
|
||||
|
||||
mock_support.assert_called_once_with('timing')
|
||||
mock_pwf.assert_called_once_with(self.info['password'])
|
||||
mock_exec.assert_called_once_with(*args)
|
||||
self.assertFalse(mock_sleep.called)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_second_call_to_address_sleep(self, mock_exec,
|
||||
mock_pwf, mock_timing_support, mock_sleep):
|
||||
mock_pwf, mock_support, mock_sleep):
|
||||
ipmi.LAST_CMD_TIME = {}
|
||||
pw_file_handle1 = tempfile.NamedTemporaryFile()
|
||||
pw_file1 = pw_file_handle1.name
|
||||
|
|
@ -209,7 +491,9 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
'D', 'E', 'F',
|
||||
]]
|
||||
|
||||
mock_timing_support.return_value = False
|
||||
expected = [mock.call('timing'),
|
||||
mock.call('timing')]
|
||||
mock_support.return_value = False
|
||||
mock_pwf.side_effect = iter([file_handle1, file_handle2])
|
||||
mock_exec.side_effect = iter([(None, None), (None, None)])
|
||||
|
||||
|
|
@ -218,13 +502,14 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
|
||||
ipmi._exec_ipmitool(self.info, 'D E F')
|
||||
self.assertTrue(mock_sleep.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
mock_exec.assert_called_with(*args[1])
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_second_call_to_address_no_sleep(self, mock_exec,
|
||||
mock_pwf, mock_timing_support, mock_sleep):
|
||||
mock_pwf, mock_support, mock_sleep):
|
||||
ipmi.LAST_CMD_TIME = {}
|
||||
pw_file_handle1 = tempfile.NamedTemporaryFile()
|
||||
pw_file1 = pw_file_handle1.name
|
||||
|
|
@ -251,7 +536,9 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
'D', 'E', 'F',
|
||||
]]
|
||||
|
||||
mock_timing_support.return_value = False
|
||||
expected = [mock.call('timing'),
|
||||
mock.call('timing')]
|
||||
mock_support.return_value = False
|
||||
mock_pwf.side_effect = iter([file_handle1, file_handle2])
|
||||
mock_exec.side_effect = iter([(None, None), (None, None)])
|
||||
|
||||
|
|
@ -262,13 +549,14 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
CONF.ipmi.min_command_interval)
|
||||
ipmi._exec_ipmitool(self.info, 'D E F')
|
||||
self.assertFalse(mock_sleep.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
mock_exec.assert_called_with(*args[1])
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_two_calls_to_diff_address(self, mock_exec,
|
||||
mock_pwf, mock_timing_support, mock_sleep):
|
||||
mock_pwf, mock_support, mock_sleep):
|
||||
ipmi.LAST_CMD_TIME = {}
|
||||
pw_file_handle1 = tempfile.NamedTemporaryFile()
|
||||
pw_file1 = pw_file_handle1.name
|
||||
|
|
@ -295,7 +583,9 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
'D', 'E', 'F',
|
||||
]]
|
||||
|
||||
mock_timing_support.return_value = False
|
||||
expected = [mock.call('timing'),
|
||||
mock.call('timing')]
|
||||
mock_support.return_value = False
|
||||
mock_pwf.side_effect = iter([file_handle1, file_handle2])
|
||||
mock_exec.side_effect = iter([(None, None), (None, None)])
|
||||
|
||||
|
|
@ -304,13 +594,14 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
self.info['address'] = '127.127.127.127'
|
||||
ipmi._exec_ipmitool(self.info, 'D E F')
|
||||
self.assertFalse(mock_sleep.called)
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
mock_exec.assert_called_with(*args[1])
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_without_timing(self, mock_exec, mock_pwf,
|
||||
mock_timing_support, mock_sleep):
|
||||
mock_support, mock_sleep):
|
||||
pw_file_handle = tempfile.NamedTemporaryFile()
|
||||
pw_file = pw_file_handle.name
|
||||
file_handle = open(pw_file, "w")
|
||||
|
|
@ -325,20 +616,21 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
'A', 'B', 'C',
|
||||
]
|
||||
|
||||
mock_timing_support.return_value = False
|
||||
mock_support.return_value = False
|
||||
mock_pwf.return_value = file_handle
|
||||
mock_exec.return_value = (None, None)
|
||||
|
||||
ipmi._exec_ipmitool(self.info, 'A B C')
|
||||
|
||||
mock_support.assert_called_once_with('timing')
|
||||
mock_pwf.assert_called_once_with(self.info['password'])
|
||||
mock_exec.assert_called_once_with(*args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_with_timing(self, mock_exec, mock_pwf,
|
||||
mock_timing_support, mock_sleep):
|
||||
mock_support, mock_sleep):
|
||||
pw_file_handle = tempfile.NamedTemporaryFile()
|
||||
pw_file = pw_file_handle.name
|
||||
file_handle = open(pw_file, "w")
|
||||
|
|
@ -354,20 +646,21 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
'A', 'B', 'C',
|
||||
]
|
||||
|
||||
mock_timing_support.return_value = True
|
||||
mock_support.return_value = True
|
||||
mock_pwf.return_value = file_handle
|
||||
mock_exec.return_value = (None, None)
|
||||
|
||||
ipmi._exec_ipmitool(self.info, 'A B C')
|
||||
|
||||
mock_support.assert_called_once_with('timing')
|
||||
mock_pwf.assert_called_once_with(self.info['password'])
|
||||
mock_exec.assert_called_once_with(*args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_without_username(self, mock_exec, mock_pwf,
|
||||
mock_timing_support, mock_sleep):
|
||||
mock_support, mock_sleep):
|
||||
self.info['username'] = None
|
||||
pw_file_handle = tempfile.NamedTemporaryFile()
|
||||
pw_file = pw_file_handle.name
|
||||
|
|
@ -381,18 +674,104 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
'A', 'B', 'C',
|
||||
]
|
||||
|
||||
mock_timing_support.return_value = False
|
||||
mock_support.return_value = False
|
||||
mock_pwf.return_value = file_handle
|
||||
mock_exec.return_value = (None, None)
|
||||
ipmi._exec_ipmitool(self.info, 'A B C')
|
||||
mock_support.assert_called_once_with('timing')
|
||||
self.assertTrue(mock_pwf.called)
|
||||
mock_exec.assert_called_once_with(*args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_timing_supported')
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_with_dual_bridging(self,
|
||||
mock_exec, mock_pwf,
|
||||
mock_support,
|
||||
mock_sleep):
|
||||
|
||||
node = obj_utils.get_test_node(self.context, driver='fake_ipmitool',
|
||||
driver_info=BRIDGE_INFO_DICT)
|
||||
# when support for dual bridge command is called returns True
|
||||
mock_support.return_value = True
|
||||
info = ipmi._parse_driver_info(node)
|
||||
pw_file_handle = tempfile.NamedTemporaryFile()
|
||||
pw_file = pw_file_handle.name
|
||||
file_handle = open(pw_file, "w")
|
||||
args = [
|
||||
'ipmitool',
|
||||
'-I', 'lanplus',
|
||||
'-H', info['address'],
|
||||
'-L', info['priv_level'],
|
||||
'-U', info['username'],
|
||||
'-m', info['local_address'],
|
||||
'-B', info['transit_channel'],
|
||||
'-T', info['transit_address'],
|
||||
'-b', info['target_channel'],
|
||||
'-t', info['target_address'],
|
||||
'-f', file_handle,
|
||||
'A', 'B', 'C',
|
||||
]
|
||||
|
||||
expected = [mock.call('dual_bridge'),
|
||||
mock.call('timing')]
|
||||
# When support for timing command is called returns False
|
||||
mock_support.return_value = False
|
||||
mock_pwf.return_value = file_handle
|
||||
mock_exec.return_value = (None, None)
|
||||
ipmi._exec_ipmitool(info, 'A B C')
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
self.assertTrue(mock_pwf.called)
|
||||
mock_exec.assert_called_once_with(*args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_with_single_bridging(self,
|
||||
mock_exec, mock_pwf,
|
||||
mock_support,
|
||||
mock_sleep):
|
||||
single_bridge_info = dict(BRIDGE_INFO_DICT)
|
||||
single_bridge_info['ipmi_bridging'] = 'single'
|
||||
node = obj_utils.get_test_node(self.context, driver='fake_ipmitool',
|
||||
driver_info=single_bridge_info)
|
||||
# when support for single bridge command is called returns True
|
||||
mock_support.return_value = True
|
||||
info = ipmi._parse_driver_info(node)
|
||||
info['transit_channel'] = info['transit_address'] = None
|
||||
|
||||
pw_file_handle = tempfile.NamedTemporaryFile()
|
||||
pw_file = pw_file_handle.name
|
||||
file_handle = open(pw_file, "w")
|
||||
args = [
|
||||
'ipmitool',
|
||||
'-I', 'lanplus',
|
||||
'-H', info['address'],
|
||||
'-L', info['priv_level'],
|
||||
'-U', info['username'],
|
||||
'-m', info['local_address'],
|
||||
'-b', info['target_channel'],
|
||||
'-t', info['target_address'],
|
||||
'-f', file_handle,
|
||||
'A', 'B', 'C',
|
||||
]
|
||||
|
||||
expected = [mock.call('single_bridge'),
|
||||
mock.call('timing')]
|
||||
# When support for timing command is called returns False
|
||||
mock_support.return_value = False
|
||||
mock_pwf.return_value = file_handle
|
||||
mock_exec.return_value = (None, None)
|
||||
ipmi._exec_ipmitool(info, 'A B C')
|
||||
self.assertEqual(expected, mock_support.call_args_list)
|
||||
self.assertTrue(mock_pwf.called)
|
||||
mock_exec.assert_called_once_with(*args)
|
||||
|
||||
@mock.patch.object(ipmi, '_is_option_supported')
|
||||
@mock.patch.object(ipmi, '_make_password_file', autospec=True)
|
||||
@mock.patch.object(utils, 'execute', autospec=True)
|
||||
def test__exec_ipmitool_exception(self, mock_exec, mock_pwf,
|
||||
mock_timing_support, mock_sleep):
|
||||
mock_support, mock_sleep):
|
||||
pw_file_handle = tempfile.NamedTemporaryFile()
|
||||
pw_file = pw_file_handle.name
|
||||
file_handle = open(pw_file, "w")
|
||||
|
|
@ -406,12 +785,13 @@ class IPMIToolPrivateMethodTestCase(base.TestCase):
|
|||
'A', 'B', 'C',
|
||||
]
|
||||
|
||||
mock_timing_support.return_value = False
|
||||
mock_support.return_value = False
|
||||
mock_pwf.return_value = file_handle
|
||||
mock_exec.side_effect = processutils.ProcessExecutionError("x")
|
||||
self.assertRaises(processutils.ProcessExecutionError,
|
||||
ipmi._exec_ipmitool,
|
||||
self.info, 'A B C')
|
||||
mock_support.assert_called_once_with('timing')
|
||||
mock_pwf.assert_called_once_with(self.info['password'])
|
||||
mock_exec.assert_called_once_with(*args)
|
||||
|
||||
|
|
|
|||
|
|
@ -55,6 +55,8 @@ if 'ironic.drivers.modules.seamicro' in sys.modules:
|
|||
# __init__. We bypass that check in order to run the unit tests, which do not
|
||||
# depend on 'ipmitool' being on the system.
|
||||
ipmitool.TIMING_SUPPORT = False
|
||||
ipmitool.DUAL_BRIDGE_SUPPORT = False
|
||||
ipmitool.SINGLE_BRIDGE_SUPPORT = False
|
||||
|
||||
pyghmi = importutils.try_import("pyghmi")
|
||||
if not pyghmi:
|
||||
|
|
|
|||
Loading…
Reference in New Issue