openstack/ironic-python-agent
Code Issues Proposed changes

Add network interface speed to the inventory

Browse Source 
This is another fact that Metal3's baremetal-operator is currently
consuming from extra-hardware.

Change-Id: I2ec9d5e9369f5508e7583a4e13c2083f5c8b28ba
This commit is contained in:
Dmitry Tantsur
2023-03-31 13:59:13 +02:00
parent f37ea85a27
commit 9ed232e77e
6 changed files with 614 additions and 543 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
doc/source/admin/how_it_works.rst
View File

@@ -199,7 +199,8 @@ fields:
``interfaces``
list of network interfaces with fields: ``name``, ``mac_address``,
``ipv4_address``, ``lldp``, ``vendor``, ``product``, and optionally
``biosdevname`` (BIOS given NIC name).
``biosdevname`` (BIOS given NIC name) and ``speed_mbps`` (maximum supported
speed).
.. note::
For backward compatibility, interfaces may contain ``lldp`` fields.

142
ironic_python_agent/hardware.py
View File

@@ -15,6 +15,7 @@
import abc
import binascii
import collections
import contextlib
import functools
import io
import ipaddress
@@ -58,6 +59,9 @@ WARN_BIOSDEVNAME_NOT_FOUND = False
UNIT_CONVERTER = pint.UnitRegistry(filename=None)
UNIT_CONVERTER.define('bytes = []')
UNIT_CONVERTER.define('MB = 1048576 bytes')
UNIT_CONVERTER.define('bit_s = []')
UNIT_CONVERTER.define('Mbit_s = 1000000 * bit_s')
UNIT_CONVERTER.define('Gbit_s = 1000 * Mbit_s')
_MEMORY_ID_RE = re.compile(r'^memory(:\d+)?$')
NODE = None
API_CLIENT = None
@@ -93,28 +97,11 @@ def _get_device_info(dev, devclass, field):
'r') as f:
return f.read().strip()
except IOError:
LOG.warning("Can't find field %(field)s for"
LOG.warning("Can't find field %(field)s for "
"device %(dev)s in device class %(class)s",
{'field': field, 'dev': dev, 'class': devclass})
def _get_system_lshw_dict():
"""Get a dict representation of the system from lshw
Retrieves a json representation of the system from lshw and converts
it to a python dict
:return: A python dict from the lshw json output
"""
out, _e = il_utils.execute('lshw', '-quiet', '-json', log_stdout=False)
out = json.loads(out)
# Depending on lshw version, output might be a list, starting with
# https://github.com/lyonel/lshw/commit/135a853c60582b14c5b67e5cd988a8062d9896f4 # noqa
if isinstance(out, list):
return out[0]
return out
def _udev_settle():
"""Wait for the udev event queue to settle.
@@ -787,11 +774,11 @@ class NetworkInterface(encoding.SerializableComparable):
serializable_fields = ('name', 'mac_address', 'ipv4_address',
'ipv6_address', 'has_carrier', 'lldp',
'vendor', 'product', 'client_id',
'biosdevname')
'biosdevname', 'speed_mbps')
def __init__(self, name, mac_addr, ipv4_address=None, ipv6_address=None,
has_carrier=True, lldp=None, vendor=None, product=None,
client_id=None, biosdevname=None):
client_id=None, biosdevname=None, speed_mbps=None):
self.name = name
self.mac_address = mac_addr
self.ipv4_address = ipv4_address
@@ -801,6 +788,7 @@ class NetworkInterface(encoding.SerializableComparable):
self.vendor = vendor
self.product = product
self.biosdevname = biosdevname
self.speed_mbps = speed_mbps
# client_id is used for InfiniBand only. we calculate the DHCP
# client identifier Option to allow DHCP to work over InfiniBand.
# see https://tools.ietf.org/html/rfc4390
@@ -1202,6 +1190,7 @@ class GenericHardwareManager(HardwareManager):
def __init__(self):
self.lldp_data = {}
self._lshw_cache = None
def evaluate_hardware_support(self):
# Do some initialization before we declare ourself ready
@@ -1215,6 +1204,48 @@ class GenericHardwareManager(HardwareManager):
self.wait_for_disks()
return HardwareSupport.GENERIC
def list_hardware_info(self):
"""Return full hardware inventory as a serializable dict.
This inventory is sent to Ironic on lookup and to Inspector on
inspection.
:return: a dictionary representing inventory
"""
with self._cached_lshw():
return super().list_hardware_info()
@contextlib.contextmanager
def _cached_lshw(self):
if self._lshw_cache:
yield # make this context manager reentrant without purging cache
return
self._lshw_cache = self._get_system_lshw_dict()
try:
yield
finally:
self._lshw_cache = None
def _get_system_lshw_dict(self):
"""Get a dict representation of the system from lshw
Retrieves a json representation of the system from lshw and converts
it to a python dict
:return: A python dict from the lshw json output
"""
if self._lshw_cache:
return self._lshw_cache
out, _e = il_utils.execute('lshw', '-quiet', '-json', log_stdout=False)
out = json.loads(out)
# Depending on lshw version, output might be a list, starting with
# https://github.com/lyonel/lshw/commit/135a853c60582b14c5b67e5cd988a8062d9896f4 # noqa
if isinstance(out, list):
return out[0]
return out
def collect_lldp_data(self, interface_names=None):
"""Collect and convert LLDP info from the node.
@@ -1254,8 +1285,31 @@ class GenericHardwareManager(HardwareManager):
if self.lldp_data:
return self.lldp_data.get(interface_name)
def get_interface_info(self, interface_name):
def _get_network_speed(self, interface_name):
sys_dict = self._get_system_lshw_dict()
try:
iface_dict = next(
utils.find_in_lshw(sys_dict, by_class='network',
logicalname=interface_name,
recursive=True)
)
except StopIteration:
LOG.warning('Cannot find detailed information about interface %s',
interface_name)
return None
# speed is the current speed, capacity is the maximum speed
speed = iface_dict.get('capacity') or iface_dict.get('speed')
if not speed:
LOG.debug('No speed information about in %s', iface_dict)
return None
units = iface_dict.get('units', 'bit_s').replace('/', '_')
return int(UNIT_CONVERTER(f'{speed} {units}')
.to(UNIT_CONVERTER.Mbit_s)
.magnitude)
def get_interface_info(self, interface_name):
mac_addr = netutils.get_mac_addr(interface_name)
if mac_addr is None:
raise errors.IncompatibleHardwareMethodError()
@@ -1267,7 +1321,8 @@ class GenericHardwareManager(HardwareManager):
has_carrier=netutils.interface_has_carrier(interface_name),
vendor=_get_device_info(interface_name, 'net', 'vendor'),
product=_get_device_info(interface_name, 'net', 'device'),
biosdevname=self.get_bios_given_nic_name(interface_name))
biosdevname=self.get_bios_given_nic_name(interface_name),
speed_mbps=self._get_network_speed(interface_name))
def get_ipv4_addr(self, interface_id):
return netutils.get_ipv4_addr(interface_id)
@@ -1324,27 +1379,28 @@ class GenericHardwareManager(HardwareManager):
interface_names=iface_names)
network_interfaces_list = []
for iface_name in iface_names:
try:
result = dispatch_to_managers(
'get_interface_info', interface_name=iface_name)
except errors.HardwareManagerMethodNotFound:
LOG.warning('No hardware manager was able to handle '
'interface %s', iface_name)
continue
result.lldp = self._get_lldp_data(iface_name)
network_interfaces_list.append(result)
# If configured, bring up vlan interfaces. If the actual vlans aren't
# defined they are derived from LLDP data
if CONF.enable_vlan_interfaces:
vlan_iface_names = netutils.bring_up_vlan_interfaces(
network_interfaces_list)
for vlan_iface_name in vlan_iface_names:
result = dispatch_to_managers(
'get_interface_info', interface_name=vlan_iface_name)
with self._cached_lshw():
for iface_name in iface_names:
try:
result = dispatch_to_managers(
'get_interface_info', interface_name=iface_name)
except errors.HardwareManagerMethodNotFound:
LOG.warning('No hardware manager was able to handle '
'interface %s', iface_name)
continue
result.lldp = self._get_lldp_data(iface_name)
network_interfaces_list.append(result)
# If configured, bring up vlan interfaces. If the actual vlans
# aren't defined they are derived from LLDP data
if CONF.enable_vlan_interfaces:
vlan_iface_names = netutils.bring_up_vlan_interfaces(
network_interfaces_list)
for vlan_iface_name in vlan_iface_names:
result = dispatch_to_managers(
'get_interface_info', interface_name=vlan_iface_name)
network_interfaces_list.append(result)
return network_interfaces_list
def get_cpus(self):
@@ -1388,7 +1444,7 @@ class GenericHardwareManager(HardwareManager):
LOG.exception(("Cannot fetch total memory size using psutil "
"version %s"), psutil.version_info[0])
try:
sys_dict = _get_system_lshw_dict()
sys_dict = self._get_system_lshw_dict()
except (processutils.ProcessExecutionError, OSError, ValueError) as e:
LOG.warning('Could not get real physical RAM from lshw: %s', e)
physical = None
@@ -1495,7 +1551,7 @@ class GenericHardwareManager(HardwareManager):
def get_system_vendor_info(self):
try:
sys_dict = _get_system_lshw_dict()
sys_dict = self._get_system_lshw_dict()
except (processutils.ProcessExecutionError, OSError, ValueError) as e:
LOG.warning('Could not retrieve vendor info from lshw: %s', e)
sys_dict = {}

54
ironic_python_agent/tests/unit/samples/hardware_samples.py
View File

@@ -661,6 +661,60 @@ LSHW_JSON_OUTPUT_V2 = ("""
"id" : "memory:5",
"class" : "memory",
"physid" : "2"
},
{
"id" : "network:0",
"class" : "network",
"handle" : "PCI:0000:00:14.3",
"description" : "Wireless interface",
"product" : "ABCD",
"vendor" : "ABCD",
"physid" : "14.3",
"businfo" : "pci@0000:00:14.3",
"logicalname" : "wlp0s20f3",
"width" : 64,
"clock" : 33000000,
"capabilities" : {
"pm" : "Power Management",
"msi" : "Message Signalled Interrupts",
"pciexpress" : "PCI Express",
"msix" : "MSI-X",
"bus_master" : "bus mastering",
"cap_list" : "PCI capabilities listing",
"ethernet" : true,
"physical" : "Physical interface",
"wireless" : "Wireless-LAN"
}
},
{
"id" : "network:1",
"class" : "network",
"handle" : "PCI:0000:00:1f.6",
"description" : "Ethernet interface",
"product" : "DCBA",
"vendor" : "DCBA",
"physid" : "1f.6",
"businfo" : "pci@0000:00:1f.6",
"logicalname" : "eth0",
"units" : "bit/s",
"capacity" : 1000000000,
"width" : 32,
"clock" : 33000000,
"capabilities" : {
"pm" : "Power Management",
"msi" : "Message Signalled Interrupts",
"bus_master" : "bus mastering",
"cap_list" : "PCI capabilities listing",
"ethernet" : true,
"physical" : "Physical interface",
"tp" : "twisted pair",
"10bt" : "10Mbit/s",
"10bt-fd" : "10Mbit/s (full duplex)",
"100bt" : "100Mbit/s",
"100bt-fd" : "100Mbit/s (full duplex)",
"1000bt-fd" : "1Gbit/s (full duplex)",
"autonegotiation" : "Auto-negotiation"
}
}
]
}

922
ironic_python_agent/tests/unit/test_hardware.py
View File

@@ -269,91 +269,6 @@ class TestGenericHardwareManager(base.IronicAgentTest):
self.assertIn(if_names[0], result)
self.assertEqual(expected_lldp_data, result)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_interfaces(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
mocked_listdir.return_value = ['lo', 'eth0', 'foobar']
mocked_exists.side_effect = [False, False, True, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_has_carrier.return_value = True
mock_get_mac.side_effect = [
'00:0c:29:8c:11:b1',
None,
]
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('em0', interfaces[0].biosdevname)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_interfaces_with_biosdevname(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
mock_has_carrier.return_value = True
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('em0', interfaces[0].biosdevname)
@mock.patch.object(il_utils, 'execute', autospec=True)
def test_get_bios_given_nic_name_ok(self, mock_execute):
interface_name = 'eth0'
@@ -405,410 +320,6 @@ class TestGenericHardwareManager(base.IronicAgentTest):
mock_execute.assert_called_once_with('biosdevname', '-i',
interface_name)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('ironic_python_agent.netutils.get_lldp_info', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_interfaces_with_lldp(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mocked_lldp_info,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
CONF.set_override('collect_lldp', True)
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_lldp_info.return_value = {'eth0': [
(0, b''),
(1, b'\x04\x88Z\x92\xecTY'),
(2, b'\x05Ethernet1/18'),
(3, b'\x00x')]
}
mock_has_carrier.return_value = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
mocked_execute.return_value = ('em0\n', '')
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
expected_lldp_info = [
(0, ''),
(1, '04885a92ec5459'),
(2, '0545746865726e6574312f3138'),
(3, '0078'),
]
self.assertEqual(expected_lldp_info, interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('em0', interfaces[0].biosdevname)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('ironic_python_agent.netutils.get_lldp_info', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
def test_list_network_interfaces_with_lldp_error(
self, mocked_execute, mocked_open, mocked_exists, mocked_listdir,
mocked_ifaddresses, mocked_lldp_info, mockedget_managers,
mock_get_mac, mock_has_carrier):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
CONF.set_override('collect_lldp', True)
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_lldp_info.side_effect = Exception('Boom!')
mocked_execute.return_value = ('em0\n', '')
mock_has_carrier.return_value = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('em0', interfaces[0].biosdevname)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_interfaces_no_carrier(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = [OSError('boom')]
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_has_carrier.return_value = False
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertFalse(interfaces[0].has_carrier)
self.assertIsNone(interfaces[0].vendor)
self.assertEqual('em0', interfaces[0].biosdevname)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_interfaces_with_vendor_info(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
mac = '00:0c:29:8c:11:b1'
read_mock.side_effect = ['0x15b3\n', '0x1014\n']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_has_carrier.return_value = True
mock_get_mac.return_value = mac
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual(mac, interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('0x15b3', interfaces[0].vendor)
self.assertEqual('0x1014', interfaces[0].product)
self.assertEqual('em0', interfaces[0].biosdevname)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_interfaces_with_bond(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
mocked_listdir.return_value = ['lo', 'bond0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('\n', '')
mock_has_carrier.return_value = True
mock_get_mac.side_effect = [
'00:0c:29:8c:11:b1',
None,
]
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('bond0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('', interfaces[0].biosdevname)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_vlan_interfaces(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
CONF.set_override('enable_vlan_interfaces', 'eth0.100')
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_get_mac.mock_has_carrier = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(2, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertEqual('eth0.100', interfaces[1].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[1].mac_address)
self.assertIsNone(interfaces[1].lldp)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('ironic_python_agent.netutils.get_lldp_info', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_vlan_interfaces_using_lldp(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mocked_lldp_info,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
CONF.set_override('collect_lldp', True)
CONF.set_override('enable_vlan_interfaces', 'eth0')
mocked_listdir.return_value = ['lo', 'eth0']
mocked_execute.return_value = ('em0\n', '')
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_lldp_info.return_value = {'eth0': [
(0, b''),
(127, b'\x00\x80\xc2\x03\x00d\x08vlan-100'),
(127, b'\x00\x80\xc2\x03\x00e\x08vlan-101')]
}
mock_has_carrier.return_value = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(3, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
expected_lldp_info = [
(0, ''),
(127, "0080c203006408766c616e2d313030"),
(127, "0080c203006508766c616e2d313031")
]
self.assertEqual(expected_lldp_info, interfaces[0].lldp)
self.assertEqual('eth0.100', interfaces[1].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[1].mac_address)
self.assertIsNone(interfaces[1].lldp)
self.assertEqual('eth0.101', interfaces[2].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[2].mac_address)
self.assertIsNone(interfaces[2].lldp)
@mock.patch.object(netutils, 'LOG', autospec=True)
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
def test_list_network_vlan_invalid_int(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_log):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
CONF.set_override('collect_lldp', True)
CONF.set_override('enable_vlan_interfaces', 'enp0s1')
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_get_mac.mock_has_carrier = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
self.hardware.list_network_interfaces()
mocked_log.warning.assert_called_once_with(
'Provided interface name %s was not found', 'enp0s1')
@mock.patch('ironic_python_agent.hardware.get_managers', autospec=True)
@mock.patch('ironic_python_agent.netutils.get_lldp_info', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
def test_list_network_vlan_interfaces_using_lldp_all(self,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_lldp_info,
mockedget_managers):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
CONF.set_override('collect_lldp', True)
CONF.set_override('enable_vlan_interfaces', 'all')
mocked_listdir.return_value = ['lo', 'eth0', 'eth1']
mocked_execute.return_value = ('em0\n', '')
mocked_exists.side_effect = [False, False, True, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_lldp_info.return_value = {'eth0': [
(0, b''),
(127, b'\x00\x80\xc2\x03\x00d\x08vlan-100'),
(127, b'\x00\x80\xc2\x03\x00e\x08vlan-101')],
'eth1': [
(0, b''),
(127, b'\x00\x80\xc2\x03\x00f\x08vlan-102'),
(127, b'\x00\x80\xc2\x03\x00g\x08vlan-103')]
}
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(6, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('eth1', interfaces[1].name)
self.assertEqual('eth0.100', interfaces[2].name)
self.assertEqual('eth0.101', interfaces[3].name)
self.assertEqual('eth1.102', interfaces[4].name)
self.assertEqual('eth1.103', interfaces[5].name)
@mock.patch.object(hardware, 'get_multipath_status', autospec=True)
@mock.patch.object(os, 'readlink', autospec=True)
@mock.patch.object(os, 'listdir', autospec=True)
@@ -1480,8 +991,11 @@ class TestGenericHardwareManager(base.IronicAgentTest):
self.assertEqual(3952 * 1024 * 1024, mem.total)
self.assertEqual(65536, mem.physical_mb)
@mock.patch('ironic_python_agent.netutils.get_hostname', autospec=True)
def test_list_hardware_info(self, mocked_get_hostname):
@mock.patch.object(hardware.GenericHardwareManager,
'_get_system_lshw_dict', autospec=True,
return_value={'id': 'host'})
@mock.patch.object(netutils, 'get_hostname', autospec=True)
def test_list_hardware_info(self, mocked_get_hostname, mocked_lshw):
self.hardware.list_network_interfaces = mock.Mock()
self.hardware.list_network_interfaces.return_value = [
hardware.NetworkInterface('eth0', '00:0c:29:8c:11:b1'),
@@ -1525,6 +1039,7 @@ class TestGenericHardwareManager(base.IronicAgentTest):
self.assertEqual(self.hardware.get_boot_info(),
hardware_info['boot'])
self.assertEqual('mock_hostname', hardware_info['hostname'])
mocked_lshw.assert_called_once_with(self.hardware)
@mock.patch.object(hardware, 'list_all_block_devices', autospec=True)
def test_list_block_devices(self, list_mock):
@@ -5977,3 +5492,428 @@ class TestCollectSystemLogs(base.IronicAgentTest):
self.assertEqual(commands, expected)
self.assertGreaterEqual(len(io_dict), len(expected))
@mock.patch.object(hardware.GenericHardwareManager, '_get_system_lshw_dict',
autospec=True, return_value={'id': 'host'})
@mock.patch.object(hardware, 'get_managers', autospec=True,
return_value=[hardware.GenericHardwareManager()])
@mock.patch('netifaces.ifaddresses', autospec=True)
@mock.patch('os.listdir', autospec=True)
@mock.patch('os.path.exists', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch.object(il_utils, 'execute', autospec=True)
@mock.patch.object(netutils, 'get_mac_addr', autospec=True)
@mock.patch.object(netutils, 'interface_has_carrier', autospec=True)
class TestListNetworkInterfaces(base.IronicAgentTest):
def setUp(self):
super().setUp()
self.hardware = hardware.GenericHardwareManager()
def test_list_network_interfaces(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
mocked_lshw.return_value = json.loads(hws.LSHW_JSON_OUTPUT_V2[0])
mocked_listdir.return_value = ['lo', 'eth0', 'foobar']
mocked_exists.side_effect = [False, False, True, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_has_carrier.return_value = True
mock_get_mac.side_effect = [
'00:0c:29:8c:11:b1',
None,
]
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('em0', interfaces[0].biosdevname)
self.assertEqual(1000, interfaces[0].speed_mbps)
def test_list_network_interfaces_with_biosdevname(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
mock_has_carrier.return_value = True
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('em0', interfaces[0].biosdevname)
self.assertIsNone(interfaces[0].speed_mbps)
@mock.patch.object(netutils, 'get_lldp_info', autospec=True)
def test_list_network_interfaces_with_lldp(self,
mocked_lldp_info,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
CONF.set_override('collect_lldp', True)
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_lldp_info.return_value = {'eth0': [
(0, b''),
(1, b'\x04\x88Z\x92\xecTY'),
(2, b'\x05Ethernet1/18'),
(3, b'\x00x')]
}
mock_has_carrier.return_value = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
mocked_execute.return_value = ('em0\n', '')
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
expected_lldp_info = [
(0, ''),
(1, '04885a92ec5459'),
(2, '0545746865726e6574312f3138'),
(3, '0078'),
]
self.assertEqual(expected_lldp_info, interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('em0', interfaces[0].biosdevname)
@mock.patch.object(netutils, 'get_lldp_info', autospec=True)
def test_list_network_interfaces_with_lldp_error(
self, mocked_lldp_info, mock_has_carrier, mock_get_mac,
mocked_execute, mocked_open, mocked_exists, mocked_listdir,
mocked_ifaddresses, mockedget_managers, mocked_lshw):
CONF.set_override('collect_lldp', True)
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_lldp_info.side_effect = Exception('Boom!')
mocked_execute.return_value = ('em0\n', '')
mock_has_carrier.return_value = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('em0', interfaces[0].biosdevname)
def test_list_network_interfaces_no_carrier(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
mockedget_managers.return_value = [hardware.GenericHardwareManager()]
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = [OSError('boom')]
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_has_carrier.return_value = False
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertFalse(interfaces[0].has_carrier)
self.assertIsNone(interfaces[0].vendor)
self.assertEqual('em0', interfaces[0].biosdevname)
def test_list_network_interfaces_with_vendor_info(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
mac = '00:0c:29:8c:11:b1'
read_mock.side_effect = ['0x15b3\n', '0x1014\n']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_has_carrier.return_value = True
mock_get_mac.return_value = mac
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual(mac, interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('0x15b3', interfaces[0].vendor)
self.assertEqual('0x1014', interfaces[0].product)
self.assertEqual('em0', interfaces[0].biosdevname)
def test_list_network_interfaces_with_bond(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
mocked_listdir.return_value = ['lo', 'bond0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('\n', '')
mock_has_carrier.return_value = True
mock_get_mac.side_effect = [
'00:0c:29:8c:11:b1',
None,
]
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(1, len(interfaces))
self.assertEqual('bond0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertTrue(interfaces[0].has_carrier)
self.assertEqual('', interfaces[0].biosdevname)
def test_list_network_vlan_interfaces(self,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
CONF.set_override('enable_vlan_interfaces', 'eth0.100')
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_get_mac.mock_has_carrier = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(2, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
self.assertEqual('192.168.1.2', interfaces[0].ipv4_address)
self.assertEqual('fd00::101', interfaces[0].ipv6_address)
self.assertIsNone(interfaces[0].lldp)
self.assertEqual('eth0.100', interfaces[1].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[1].mac_address)
self.assertIsNone(interfaces[1].lldp)
@mock.patch.object(netutils, 'get_lldp_info', autospec=True)
def test_list_network_vlan_interfaces_using_lldp(self,
mocked_lldp_info,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
CONF.set_override('collect_lldp', True)
CONF.set_override('enable_vlan_interfaces', 'eth0')
mocked_listdir.return_value = ['lo', 'eth0']
mocked_execute.return_value = ('em0\n', '')
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_lldp_info.return_value = {'eth0': [
(0, b''),
(127, b'\x00\x80\xc2\x03\x00d\x08vlan-100'),
(127, b'\x00\x80\xc2\x03\x00e\x08vlan-101')]
}
mock_has_carrier.return_value = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(3, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[0].mac_address)
expected_lldp_info = [
(0, ''),
(127, "0080c203006408766c616e2d313030"),
(127, "0080c203006508766c616e2d313031")
]
self.assertEqual(expected_lldp_info, interfaces[0].lldp)
self.assertEqual('eth0.100', interfaces[1].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[1].mac_address)
self.assertIsNone(interfaces[1].lldp)
self.assertEqual('eth0.101', interfaces[2].name)
self.assertEqual('00:0c:29:8c:11:b1', interfaces[2].mac_address)
self.assertIsNone(interfaces[2].lldp)
@mock.patch.object(netutils, 'LOG', autospec=True)
def test_list_network_vlan_invalid_int(self,
mocked_log,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
CONF.set_override('collect_lldp', True)
CONF.set_override('enable_vlan_interfaces', 'enp0s1')
mocked_listdir.return_value = ['lo', 'eth0']
mocked_exists.side_effect = [False, False, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_ifaddresses.return_value = {
netifaces.AF_INET: [{'addr': '192.168.1.2'}],
netifaces.AF_INET6: [{'addr': 'fd00::101'}]
}
mocked_execute.return_value = ('em0\n', '')
mock_get_mac.mock_has_carrier = True
mock_get_mac.return_value = '00:0c:29:8c:11:b1'
self.hardware.list_network_interfaces()
mocked_log.warning.assert_called_once_with(
'Provided interface name %s was not found', 'enp0s1')
@mock.patch.object(netutils, 'get_lldp_info', autospec=True)
def test_list_network_vlan_interfaces_using_lldp_all(self,
mocked_lldp_info,
mock_has_carrier,
mock_get_mac,
mocked_execute,
mocked_open,
mocked_exists,
mocked_listdir,
mocked_ifaddresses,
mockedget_managers,
mocked_lshw):
CONF.set_override('collect_lldp', True)
CONF.set_override('enable_vlan_interfaces', 'all')
mocked_listdir.return_value = ['lo', 'eth0', 'eth1']
mocked_execute.return_value = ('em0\n', '')
mocked_exists.side_effect = [False, False, True, True]
mocked_open.return_value.__enter__ = lambda s: s
mocked_open.return_value.__exit__ = mock.Mock()
read_mock = mocked_open.return_value.read
read_mock.side_effect = ['1']
mocked_lldp_info.return_value = {'eth0': [
(0, b''),
(127, b'\x00\x80\xc2\x03\x00d\x08vlan-100'),
(127, b'\x00\x80\xc2\x03\x00e\x08vlan-101')],
'eth1': [
(0, b''),
(127, b'\x00\x80\xc2\x03\x00f\x08vlan-102'),
(127, b'\x00\x80\xc2\x03\x00g\x08vlan-103')]
}
interfaces = self.hardware.list_network_interfaces()
self.assertEqual(6, len(interfaces))
self.assertEqual('eth0', interfaces[0].name)
self.assertEqual('eth1', interfaces[1].name)
self.assertEqual('eth0.100', interfaces[2].name)
self.assertEqual('eth0.101', interfaces[3].name)
self.assertEqual('eth1.102', interfaces[4].name)
self.assertEqual('eth1.103', interfaces[5].name)

31
ironic_python_agent/utils.py
View File

@@ -919,13 +919,28 @@ def rescan_device(device):
'to settle. Error: %s', e)
def find_in_lshw(lshw, by_id):
def _lshw_matches(item, by_id, fields):
lshw_id = item.get('id', '')
if isinstance(by_id, re.Pattern):
if by_id.match(lshw_id) is None:
return False
elif by_id is not None and by_id != lshw_id:
return False
for key, value in fields.items():
if item.get(key) != value:
return False
return True
def find_in_lshw(lshw, by_id=None, by_class=None, recursive=False, **fields):
"""Yield all suitable records from lshw."""
# Cannot really pass class=... in Python
if by_class is not None:
fields['class'] = by_class
for child in lshw.get('children', ()):
lshw_id = child.get('id', '')
if isinstance(by_id, re.Pattern):
if by_id.match(lshw_id) is not None:
yield child
else:
if by_id == lshw_id:
yield child
if _lshw_matches(child, by_id, fields):
yield child
if recursive:
yield from find_in_lshw(child, by_id, recursive=True, **fields)

5
releasenotes/notes/net-speed-8854901e2051bb79.yaml Normal file
View File

@@ -0,0 +1,5 @@
---
features:
- |
The hardware inventory now contains supported network interface speed in
Mbit/s.