Check for the existence of an IPMI device

Check for IPMI device files before the use of the `'ipmitool lan.*'`
command, avoiding unnecessary calls on non-IPMI systems.

Closes-Bug: #2076367
Change-Id: Ib800717701e6f2828df55a0da0e999fc014c12e1
This commit is contained in:
cid 2024-08-23 00:41:25 +01:00
parent 4cea26f185
commit 2d79eae382
4 changed files with 233 additions and 41 deletions

View File

@ -17,6 +17,7 @@ import binascii
import collections import collections
import contextlib import contextlib
import functools import functools
import glob
import io import io
import ipaddress import ipaddress
import json import json
@ -446,7 +447,7 @@ def md_restart(raid_device):
def md_get_raid_devices(): def md_get_raid_devices():
"""Get all discovered Software RAID (md) devices """Get all discovered Software RAID (md) devices
:return: A python dict containing details about the discovered RAID :returns: A python dict containing details about the discovered RAID
devices devices
""" """
# Note(Boushra): mdadm output is similar to lsblk, but not # Note(Boushra): mdadm output is similar to lsblk, but not
@ -509,7 +510,7 @@ def list_all_block_devices(block_type='disk',
the short and the long serial from udevadm if the short and the long serial from udevadm if
possible. possible.
:return: A list of BlockDevices :returns: A list of BlockDevices
""" """
def _is_known_device(existing, new_device_name): def _is_known_device(existing, new_device_name):
@ -925,7 +926,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
"""List physical block devices """List physical block devices
:param include_partitions: If to include partitions :param include_partitions: If to include partitions
:return: A list of BlockDevices :returns: A list of BlockDevices
""" """
raise errors.IncompatibleHardwareMethodError raise errors.IncompatibleHardwareMethodError
@ -936,7 +937,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
:param block_devices: a list of BlockDevices :param block_devices: a list of BlockDevices
:param just_raids: a boolean to signify that only RAID devices :param just_raids: a boolean to signify that only RAID devices
are important are important
:return: A set of names of devices on the skip list :returns: A set of names of devices on the skip list
""" """
raise errors.IncompatibleHardwareMethodError raise errors.IncompatibleHardwareMethodError
@ -948,7 +949,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
:param node: A node used to check the skip list :param node: A node used to check the skip list
:param include_partitions: If to include partitions :param include_partitions: If to include partitions
:return: A list of BlockDevices :returns: A list of BlockDevices
""" """
raise errors.IncompatibleHardwareMethodError raise errors.IncompatibleHardwareMethodError
@ -981,7 +982,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
List all USB final devices, based on lshw information List all USB final devices, based on lshw information
:return: a dict, containing product, vendor, and handle information :returns: a dict, containing product, vendor, and handle information
""" """
raise errors.IncompatibleHardwareMethodError() raise errors.IncompatibleHardwareMethodError()
@ -1027,7 +1028,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
may require manual intervention due to the contents and may require manual intervention due to the contents and
operational risk which exists as it could also be a sign operational risk which exists as it could also be a sign
of an environmental misconfiguration. of an environmental misconfiguration.
:return: a dictionary in the form {device.name: erasure output} :returns: a dictionary in the form {device.name: erasure output}
""" """
erase_results = {} erase_results = {}
block_devices = self.list_block_devices_check_skip_list(node) block_devices = self.list_block_devices_check_skip_list(node)
@ -1088,7 +1089,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
This inventory is sent to Ironic on lookup and to Inspector on This inventory is sent to Ironic on lookup and to Inspector on
inspection. inspection.
:return: a dictionary representing inventory :returns: a dictionary representing inventory
""" """
start = time.time() start = time.time()
LOG.info('Collecting full inventory') LOG.info('Collecting full inventory')
@ -1161,7 +1162,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
:param node: Ironic node object :param node: Ironic node object
:param ports: list of Ironic port objects :param ports: list of Ironic port objects
:return: a list of cleaning steps, where each step is described as a :returns: a list of cleaning steps, where each step is described as a
dict as defined above dict as defined above
""" """
@ -1210,7 +1211,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
:param node: Ironic node object :param node: Ironic node object
:param ports: list of Ironic port objects :param ports: list of Ironic port objects
:return: a list of deploying steps, where each step is described as a :returns: a list of deploying steps, where each step is described as a
dict as defined above dict as defined above
""" """
@ -1264,7 +1265,7 @@ class HardwareManager(object, metaclass=abc.ABCMeta):
:param node: Ironic node object :param node: Ironic node object
:param ports: list of Ironic port objects :param ports: list of Ironic port objects
:return: a list of service steps, where each step is described as a :returns: a list of service steps, where each step is described as a
dict as defined above dict as defined above
""" """
@ -1336,7 +1337,7 @@ class GenericHardwareManager(HardwareManager):
This inventory is sent to Ironic on lookup and to Inspector on This inventory is sent to Ironic on lookup and to Inspector on
inspection. inspection.
:return: a dictionary representing inventory :returns: a dictionary representing inventory
""" """
with self._cached_lshw(): with self._cached_lshw():
return super().list_hardware_info() return super().list_hardware_info()
@ -1359,7 +1360,7 @@ class GenericHardwareManager(HardwareManager):
Retrieves a json representation of the system from lshw and converts Retrieves a json representation of the system from lshw and converts
it to a python dict it to a python dict
:return: A python dict from the lshw json output :returns: A python dict from the lshw json output
""" """
if self._lshw_cache: if self._lshw_cache:
return self._lshw_cache return self._lshw_cache
@ -1379,7 +1380,7 @@ class GenericHardwareManager(HardwareManager):
be converted for serialization purposes. be converted for serialization purposes.
:param interface_names: list of names of node's interfaces. :param interface_names: list of names of node's interfaces.
:return: a dict, containing the lldp data from every interface. :returns: a dict, containing the lldp data from every interface.
""" """
if interface_names is None: if interface_names is None:
interface_names = netutils.list_interfaces() interface_names = netutils.list_interfaces()
@ -1471,7 +1472,7 @@ class GenericHardwareManager(HardwareManager):
extra field named ``biosdevname``. extra field named ``biosdevname``.
:param interface_name: list of names of node's interfaces. :param interface_name: list of names of node's interfaces.
:return: the BIOS given NIC name of node's interfaces or default :returns: the BIOS given NIC name of node's interfaces or default
as None. as None.
""" """
global WARN_BIOSDEVNAME_NOT_FOUND global WARN_BIOSDEVNAME_NOT_FOUND
@ -1529,6 +1530,15 @@ class GenericHardwareManager(HardwareManager):
return network_interfaces_list return network_interfaces_list
def any_ipmi_device_exists(self):
'''Check for an IPMI device to confirm IPMI capability.'''
for pattern in ['/dev/ipmi*', '/dev/ipmi/*', '/dev/ipmidev/*']:
ipmi_files = glob.glob(pattern)
for device in ipmi_files:
if utils.is_char_device(device):
return True
return False
@staticmethod @staticmethod
def create_cpu_info_dict(lines): def create_cpu_info_dict(lines):
cpu_info = {k.strip().lower(): v.strip() for k, v in cpu_info = {k.strip().lower(): v.strip() for k, v in
@ -2317,7 +2327,7 @@ class GenericHardwareManager(HardwareManager):
"""Attempt to clean the NVMe using the most secure supported method """Attempt to clean the NVMe using the most secure supported method
:param block_device: a BlockDevice object :param block_device: a BlockDevice object
:return: True if cleaning operation succeeded, False if it failed :returns: True if cleaning operation succeeded, False if it failed
:raises: BlockDeviceEraseError :raises: BlockDeviceEraseError
""" """
@ -2382,9 +2392,12 @@ class GenericHardwareManager(HardwareManager):
def get_bmc_address(self): def get_bmc_address(self):
"""Attempt to detect BMC IP address """Attempt to detect BMC IP address
:return: IP address of lan channel or 0.0.0.0 in case none of them is :returns: IP address of lan channel or 0.0.0.0 in case none of them is
configured properly configured properly
""" """
if not self.any_ipmi_device_exists():
return None
try: try:
# From all the channels 0-15, only 1-11 can be assigned to # From all the channels 0-15, only 1-11 can be assigned to
# different types of communication media and protocols and # different types of communication media and protocols and
@ -2419,9 +2432,13 @@ class GenericHardwareManager(HardwareManager):
def get_bmc_mac(self): def get_bmc_mac(self):
"""Attempt to detect BMC MAC address """Attempt to detect BMC MAC address
:return: MAC address of the first LAN channel or 00:00:00:00:00:00 in :returns: MAC address of the first LAN channel or 00:00:00:00:00:00 in
case none of them has one or is configured properly case none of them has one or is configured properly
:raises: IncompatibleHardwareMethodError if no valid mac is found.
""" """
if not self.any_ipmi_device_exists():
return None
try: try:
# From all the channels 0-15, only 1-11 can be assigned to # From all the channels 0-15, only 1-11 can be assigned to
# different types of communication media and protocols and # different types of communication media and protocols and
@ -2465,10 +2482,13 @@ class GenericHardwareManager(HardwareManager):
def get_bmc_v6address(self): def get_bmc_v6address(self):
"""Attempt to detect BMC v6 address """Attempt to detect BMC v6 address
:return: IPv6 address of lan channel or ::/0 in case none of them is :returns: IPv6 address of lan channel or ::/0 in case none of them is
configured properly. May return None value if it cannot configured properly. May return None value if it cannot
interact with system tools or critical error occurs. interact with system tools or critical error occurs.
""" """
if not self.any_ipmi_device_exists():
return None
null_address_re = re.compile(r'^::(/\d{1,3})*$') null_address_re = re.compile(r'^::(/\d{1,3})*$')
def get_addr(channel, dynamic=False): def get_addr(channel, dynamic=False):

View File

@ -13,6 +13,7 @@
# limitations under the License. # limitations under the License.
import binascii import binascii
import glob
import json import json
import logging import logging
import os import os
@ -2984,37 +2985,98 @@ class TestGenericHardwareManager(base.IronicAgentTest):
mock_dev_file.side_effect = reads mock_dev_file.side_effect = reads
self.assertTrue(self.hardware._is_read_only_device(device)) self.assertTrue(self.hardware._is_read_only_device(device))
@mock.patch.object(os.path, 'exists', autospec=True)
@mock.patch.object(os, 'stat', autospec=True)
@mock.patch.object(glob, 'glob', autospec=True)
def test_ipmi_device_exists(self, mock_glob, mock_stat, mock_exists):
mock_stat_result = mock.Mock()
mock_stat_result.st_mode = stat.S_IFCHR
mock_stat.return_value = mock_stat_result
# 1. Test when no device is found in default locations,
# but found in glob
mock_exists.side_effect = [False, False, False]
mock_glob.return_value = ['/dev/ipmi1']
self.assertTrue(self.hardware.any_ipmi_device_exists())
mock_stat.assert_called_once_with('/dev/ipmi1')
mock_exists.reset_mock()
mock_exists.reset_mock()
mock_stat.reset_mock()
# 2. Test when no device is found in default locations,
# but found in glob
mock_exists.side_effect = [False, False, False]
mock_glob.return_value = ['/dev/ipmidev/11']
self.assertTrue(self.hardware.any_ipmi_device_exists())
mock_stat.assert_called_once_with('/dev/ipmidev/11')
# Reset mocks
mock_exists.reset_mock()
mock_stat.reset_mock()
mock_glob.reset_mock()
# Test when no IPMI device is found at all
mock_exists.side_effect = [False, False, False]
mock_glob.return_value = []
self.assertFalse(self.hardware.any_ipmi_device_exists())
mock_stat.assert_not_called()
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_address(self, mocked_execute): def test_get_bmc_address(self, mocked_execute, mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.return_value = '192.1.2.3\n', '' mocked_execute.return_value = '192.1.2.3\n', ''
self.assertEqual('192.1.2.3', self.hardware.get_bmc_address()) self.assertEqual('192.1.2.3', self.hardware.get_bmc_address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_address_virt(self, mocked_execute): def test_get_bmc_address_virt(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.side_effect = processutils.ProcessExecutionError() mocked_execute.side_effect = processutils.ProcessExecutionError()
self.assertIsNone(self.hardware.get_bmc_address()) self.assertIsNone(self.hardware.get_bmc_address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_address_zeroed(self, mocked_execute): def test_get_bmc_address_zeroed(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.return_value = '0.0.0.0\n', '' mocked_execute.return_value = '0.0.0.0\n', ''
self.assertEqual('0.0.0.0', self.hardware.get_bmc_address()) self.assertEqual('0.0.0.0', self.hardware.get_bmc_address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_address_invalid(self, mocked_execute): def test_get_bmc_address_invalid(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
# In case of invalid lan channel, stdout is empty and the error # In case of invalid lan channel, stdout is empty and the error
# on stderr is "Invalid channel" # on stderr is "Invalid channel"
mocked_execute.return_value = '\n', 'Invalid channel: 55' mocked_execute.return_value = '\n', 'Invalid channel: 55'
self.assertEqual('0.0.0.0', self.hardware.get_bmc_address()) self.assertEqual('0.0.0.0', self.hardware.get_bmc_address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_address_random_error(self, mocked_execute): def test_get_bmc_address_random_error(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.return_value = '192.1.2.3\n', 'Random error message' mocked_execute.return_value = '192.1.2.3\n', 'Random error message'
self.assertEqual('192.1.2.3', self.hardware.get_bmc_address()) self.assertEqual('192.1.2.3', self.hardware.get_bmc_address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_address_iterate_channels(self, mocked_execute): def test_get_bmc_address_iterate_channels(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
# For channel 1 we simulate unconfigured IP # For channel 1 we simulate unconfigured IP
# and for any other we return a correct IP address # and for any other we return a correct IP address
def side_effect(*args, **kwargs): def side_effect(*args, **kwargs):
if args[0].startswith("ipmitool lan print 1"): if args[0].startswith("ipmitool lan print 1"):
return '', 'Invalid channel 1\n' return '', 'Invalid channel 1\n'
@ -3027,51 +3089,82 @@ class TestGenericHardwareManager(base.IronicAgentTest):
mocked_execute.side_effect = side_effect mocked_execute.side_effect = side_effect
self.assertEqual('192.1.2.3', self.hardware.get_bmc_address()) self.assertEqual('192.1.2.3', self.hardware.get_bmc_address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_address_not_available(self, mocked_execute): def test_get_bmc_address_not_available(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.return_value = '', '' mocked_execute.return_value = '', ''
self.assertEqual('0.0.0.0', self.hardware.get_bmc_address()) self.assertEqual('0.0.0.0', self.hardware.get_bmc_address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_mac_not_available(self, mocked_execute): def test_get_bmc_mac_not_available(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.return_value = '', '' mocked_execute.return_value = '', ''
self.assertRaises(errors.IncompatibleHardwareMethodError, self.assertRaises(errors.IncompatibleHardwareMethodError,
self.hardware.get_bmc_mac) self.hardware.get_bmc_mac)
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_mac(self, mocked_execute): def test_get_bmc_mac(self, mocked_execute, mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.return_value = '192.1.2.3\n01:02:03:04:05:06', '' mocked_execute.return_value = '192.1.2.3\n01:02:03:04:05:06', ''
self.assertEqual('01:02:03:04:05:06', self.hardware.get_bmc_mac()) self.assertEqual('01:02:03:04:05:06', self.hardware.get_bmc_mac())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_mac_virt(self, mocked_execute): def test_get_bmc_mac_virt(self, mocked_execute, mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.side_effect = processutils.ProcessExecutionError() mocked_execute.side_effect = processutils.ProcessExecutionError()
self.assertIsNone(self.hardware.get_bmc_mac()) self.assertIsNone(self.hardware.get_bmc_mac())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_mac_zeroed(self, mocked_execute): def test_get_bmc_mac_zeroed(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.return_value = '0.0.0.0\n00:00:00:00:00:00', '' mocked_execute.return_value = '0.0.0.0\n00:00:00:00:00:00', ''
self.assertRaises(errors.IncompatibleHardwareMethodError, self.assertRaises(errors.IncompatibleHardwareMethodError,
self.hardware.get_bmc_mac) self.hardware.get_bmc_mac)
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_mac_invalid(self, mocked_execute): def test_get_bmc_mac_invalid(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
# In case of invalid lan channel, stdout is empty and the error # In case of invalid lan channel, stdout is empty and the error
# on stderr is "Invalid channel" # on stderr is "Invalid channel"
mocked_execute.return_value = '\n', 'Invalid channel: 55' mocked_execute.return_value = '\n', 'Invalid channel: 55'
self.assertRaises(errors.IncompatibleHardwareMethodError, self.assertRaises(errors.IncompatibleHardwareMethodError,
self.hardware.get_bmc_mac) self.hardware.get_bmc_mac)
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_mac_random_error(self, mocked_execute): def test_get_bmc_mac_random_error(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.return_value = ('192.1.2.3\n00:00:00:00:00:02', mocked_execute.return_value = ('192.1.2.3\n00:00:00:00:00:02',
'Random error message') 'Random error message')
self.assertEqual('00:00:00:00:00:02', self.hardware.get_bmc_mac()) self.assertEqual('00:00:00:00:00:02', self.hardware.get_bmc_mac())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_mac_iterate_channels(self, mocked_execute): def test_get_bmc_mac_iterate_channels(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
# For channel 1 we simulate unconfigured IP # For channel 1 we simulate unconfigured IP
# and for any other we return a correct IP address # and for any other we return a correct IP address
def side_effect(*args, **kwargs): def side_effect(*args, **kwargs):
if args[0].startswith("ipmitool lan print 1"): if args[0].startswith("ipmitool lan print 1"):
return '', 'Invalid channel 1\n' return '', 'Invalid channel 1\n'
@ -3087,13 +3180,21 @@ class TestGenericHardwareManager(base.IronicAgentTest):
mocked_execute.side_effect = side_effect mocked_execute.side_effect = side_effect
self.assertEqual('01:02:03:04:05:06', self.hardware.get_bmc_mac()) self.assertEqual('01:02:03:04:05:06', self.hardware.get_bmc_mac())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_not_enabled(self, mocked_execute): def test_get_bmc_v6address_not_enabled(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.side_effect = [('ipv4\n', '')] * 11 mocked_execute.side_effect = [('ipv4\n', '')] * 11
self.assertEqual('::/0', self.hardware.get_bmc_v6address()) self.assertEqual('::/0', self.hardware.get_bmc_v6address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_dynamic_address(self, mocked_execute): def test_get_bmc_v6address_dynamic_address(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.side_effect = [ mocked_execute.side_effect = [
('ipv6\n', ''), ('ipv6\n', ''),
(hws.IPMITOOL_LAN6_PRINT_DYNAMIC_ADDR, '') (hws.IPMITOOL_LAN6_PRINT_DYNAMIC_ADDR, '')
@ -3101,8 +3202,12 @@ class TestGenericHardwareManager(base.IronicAgentTest):
self.assertEqual('2001:1234:1234:1234:1234:1234:1234:1234', self.assertEqual('2001:1234:1234:1234:1234:1234:1234:1234',
self.hardware.get_bmc_v6address()) self.hardware.get_bmc_v6address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_static_address_both(self, mocked_execute): def test_get_bmc_v6address_static_address_both(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
dynamic_disabled = \ dynamic_disabled = \
hws.IPMITOOL_LAN6_PRINT_DYNAMIC_ADDR.replace('active', 'disabled') hws.IPMITOOL_LAN6_PRINT_DYNAMIC_ADDR.replace('active', 'disabled')
mocked_execute.side_effect = [ mocked_execute.side_effect = [
@ -3113,13 +3218,22 @@ class TestGenericHardwareManager(base.IronicAgentTest):
self.assertEqual('2001:5678:5678:5678:5678:5678:5678:5678', self.assertEqual('2001:5678:5678:5678:5678:5678:5678:5678',
self.hardware.get_bmc_v6address()) self.hardware.get_bmc_v6address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_virt(self, mocked_execute): def test_get_bmc_v6address_virt(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
mocked_execute.side_effect = processutils.ProcessExecutionError() mocked_execute.side_effect = processutils.ProcessExecutionError()
self.assertIsNone(self.hardware.get_bmc_v6address()) self.assertIsNone(self.hardware.get_bmc_v6address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_invalid_enables(self, mocked_execute): def test_get_bmc_v6address_invalid_enables(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
def side_effect(*args, **kwargs): def side_effect(*args, **kwargs):
if args[0].startswith('ipmitool lan6 print'): if args[0].startswith('ipmitool lan6 print'):
return '', 'Failed to get IPv6/IPv4 Addressing Enables' return '', 'Failed to get IPv6/IPv4 Addressing Enables'
@ -3127,8 +3241,13 @@ class TestGenericHardwareManager(base.IronicAgentTest):
mocked_execute.side_effect = side_effect mocked_execute.side_effect = side_effect
self.assertEqual('::/0', self.hardware.get_bmc_v6address()) self.assertEqual('::/0', self.hardware.get_bmc_v6address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_invalid_get_address(self, mocked_execute): def test_get_bmc_v6address_invalid_get_address(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
def side_effect(*args, **kwargs): def side_effect(*args, **kwargs):
if args[0].startswith('ipmitool lan6 print'): if args[0].startswith('ipmitool lan6 print'):
if args[0].endswith('dynamic_addr') \ if args[0].endswith('dynamic_addr') \
@ -3139,10 +3258,14 @@ class TestGenericHardwareManager(base.IronicAgentTest):
mocked_execute.side_effect = side_effect mocked_execute.side_effect = side_effect
self.assertEqual('::/0', self.hardware.get_bmc_v6address()) self.assertEqual('::/0', self.hardware.get_bmc_v6address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(hardware, 'LOG', autospec=True) @mock.patch.object(hardware, 'LOG', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_ipmitool_invalid_stdout_format( def test_get_bmc_v6address_ipmitool_invalid_stdout_format(
self, mocked_execute, mocked_log): self, mocked_execute, mocked_log, mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
def side_effect(*args, **kwargs): def side_effect(*args, **kwargs):
if args[0].startswith('ipmitool lan6 print'): if args[0].startswith('ipmitool lan6 print'):
if args[0].endswith('dynamic_addr') \ if args[0].endswith('dynamic_addr') \
@ -3156,8 +3279,13 @@ class TestGenericHardwareManager(base.IronicAgentTest):
'command: %(e)s', mock.ANY) 'command: %(e)s', mock.ANY)
mocked_log.warning.assert_has_calls([one_call] * 14) mocked_log.warning.assert_has_calls([one_call] * 14)
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True) @mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_channel_7(self, mocked_execute): def test_get_bmc_v6address_channel_7(self, mocked_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = True
def side_effect(*args, **kwargs): def side_effect(*args, **kwargs):
if not args[0].startswith('ipmitool lan6 print 7'): if not args[0].startswith('ipmitool lan6 print 7'):
# ipv6 is not enabled for channels 1-6 # ipv6 is not enabled for channels 1-6
@ -3175,6 +3303,33 @@ class TestGenericHardwareManager(base.IronicAgentTest):
self.assertEqual('2001:5678:5678:5678:5678:5678:5678:5678', self.assertEqual('2001:5678:5678:5678:5678:5678:5678:5678',
self.hardware.get_bmc_v6address()) self.hardware.get_bmc_v6address())
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_address_no_ipmi_device(self, mock_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = False
self.assertIsNone(self.hardware.get_bmc_address())
mock_execute.assert_not_called()
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_mac_no_ipmi_device(self, mock_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = False
self.assertIsNone(self.hardware.get_bmc_mac())
mock_execute.assert_not_called()
@mock.patch.object(hardware.GenericHardwareManager,
'any_ipmi_device_exists', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bmc_v6address_no_ipmi_device(self, mock_execute,
mock_ipmi_device_exists):
mock_ipmi_device_exists.return_value = False
self.assertIsNone(self.hardware.get_bmc_v6address())
mock_execute.assert_not_called()
@mock.patch.object(efi_utils, 'clean_boot_records', autospec=True) @mock.patch.object(efi_utils, 'clean_boot_records', autospec=True)
def test_clean_uefi_nvram_defaults(self, mock_efi_utils): def test_clean_uefi_nvram_defaults(self, mock_efi_utils):
self.hardware.clean_uefi_nvram(self.node, []) self.hardware.clean_uefi_nvram(self.node, [])

View File

@ -23,6 +23,7 @@ import json
import os import os
import re import re
import shutil import shutil
import stat
import subprocess import subprocess
import sys import sys
import tarfile import tarfile
@ -972,3 +973,13 @@ def _unmount_any_config_drives():
_early_log('Issuing an umount command for /mnt/config...') _early_log('Issuing an umount command for /mnt/config...')
execute('umount', '/mnt/config') execute('umount', '/mnt/config')
time.sleep(1) time.sleep(1)
def is_char_device(path):
'''Check if the specified path is a character device.'''
try:
return stat.S_ISCHR(os.stat(path).st_mode)
except OSError:
# Likely because of insufficient permission,
# race conditions or I/O related errors.
return False

View File

@ -0,0 +1,6 @@
---
fixes:
- |
Adds a preliminary check for IPMI device files before the BMC detection
attempts, optimizing `ipmitool` command calls on systems without IPMI
capabilities.