summaryrefslogtreecommitdiff
path: root/tests/unittests/test_net.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests/test_net.py')
-rw-r--r--tests/unittests/test_net.py478
1 files changed, 463 insertions, 15 deletions
diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py
index 8edc0b89..06e8f094 100644
--- a/tests/unittests/test_net.py
+++ b/tests/unittests/test_net.py
@@ -836,38 +836,176 @@ CONFIG_V1_EXPLICIT_LOOPBACK = {
'subnets': [{'control': 'auto', 'type': 'loopback'}]},
]}
+DEFAULT_DEV_ATTRS = {
+ 'eth1000': {
+ "bridge": False,
+ "carrier": False,
+ "dormant": False,
+ "operstate": "down",
+ "address": "07-1C-C6-75-A4-BE",
+ "device/driver": None,
+ "device/device": None,
+ }
+}
+
def _setup_test(tmp_dir, mock_get_devicelist, mock_read_sys_net,
- mock_sys_dev_path):
- mock_get_devicelist.return_value = ['eth1000']
- dev_characteristics = {
- 'eth1000': {
- "bridge": False,
- "carrier": False,
- "dormant": False,
- "operstate": "down",
- "address": "07-1C-C6-75-A4-BE",
- }
- }
+ mock_sys_dev_path, dev_attrs=None):
+ if not dev_attrs:
+ dev_attrs = DEFAULT_DEV_ATTRS
+
+ mock_get_devicelist.return_value = dev_attrs.keys()
def fake_read(devname, path, translate=None,
on_enoent=None, on_keyerror=None,
on_einval=None):
- return dev_characteristics[devname][path]
+ return dev_attrs[devname][path]
mock_read_sys_net.side_effect = fake_read
def sys_dev_path(devname, path=""):
- return tmp_dir + devname + "/" + path
+ return tmp_dir + "/" + devname + "/" + path
- for dev in dev_characteristics:
+ for dev in dev_attrs:
os.makedirs(os.path.join(tmp_dir, dev))
with open(os.path.join(tmp_dir, dev, 'operstate'), 'w') as fh:
- fh.write("down")
+ fh.write(dev_attrs[dev]['operstate'])
+ os.makedirs(os.path.join(tmp_dir, dev, "device"))
+ for key in ['device/driver']:
+ if key in dev_attrs[dev] and dev_attrs[dev][key]:
+ target = dev_attrs[dev][key]
+ link = os.path.join(tmp_dir, dev, key)
+ print('symlink %s -> %s' % (link, target))
+ os.symlink(target, link)
mock_sys_dev_path.side_effect = sys_dev_path
+class TestGenerateFallbackConfig(CiTestCase):
+
+ @mock.patch("cloudinit.net.sys_dev_path")
+ @mock.patch("cloudinit.net.read_sys_net")
+ @mock.patch("cloudinit.net.get_devicelist")
+ def test_device_driver(self, mock_get_devicelist, mock_read_sys_net,
+ mock_sys_dev_path):
+ devices = {
+ 'eth0': {
+ 'bridge': False, 'carrier': False, 'dormant': False,
+ 'operstate': 'down', 'address': '00:11:22:33:44:55',
+ 'device/driver': 'hv_netsvc', 'device/device': '0x3'},
+ 'eth1': {
+ 'bridge': False, 'carrier': False, 'dormant': False,
+ 'operstate': 'down', 'address': '00:11:22:33:44:55',
+ 'device/driver': 'mlx4_core', 'device/device': '0x7'},
+ }
+
+ tmp_dir = self.tmp_dir()
+ _setup_test(tmp_dir, mock_get_devicelist,
+ mock_read_sys_net, mock_sys_dev_path,
+ dev_attrs=devices)
+
+ network_cfg = net.generate_fallback_config(config_driver=True)
+ ns = network_state.parse_net_config_data(network_cfg,
+ skip_broken=False)
+
+ render_dir = os.path.join(tmp_dir, "render")
+ os.makedirs(render_dir)
+
+ # don't set rulepath so eni writes them
+ renderer = eni.Renderer(
+ {'eni_path': 'interfaces', 'netrules_path': 'netrules'})
+ renderer.render_network_state(ns, render_dir)
+
+ self.assertTrue(os.path.exists(os.path.join(render_dir,
+ 'interfaces')))
+ with open(os.path.join(render_dir, 'interfaces')) as fh:
+ contents = fh.read()
+ print(contents)
+ expected = """
+auto lo
+iface lo inet loopback
+
+auto eth0
+iface eth0 inet dhcp
+"""
+ self.assertEqual(expected.lstrip(), contents.lstrip())
+
+ self.assertTrue(os.path.exists(os.path.join(render_dir, 'netrules')))
+ with open(os.path.join(render_dir, 'netrules')) as fh:
+ contents = fh.read()
+ print(contents)
+ expected_rule = [
+ 'SUBSYSTEM=="net"',
+ 'ACTION=="add"',
+ 'DRIVERS=="hv_netsvc"',
+ 'ATTR{address}=="00:11:22:33:44:55"',
+ 'NAME="eth0"',
+ ]
+ self.assertEqual(", ".join(expected_rule) + '\n', contents.lstrip())
+
+ @mock.patch("cloudinit.net.sys_dev_path")
+ @mock.patch("cloudinit.net.read_sys_net")
+ @mock.patch("cloudinit.net.get_devicelist")
+ def test_device_driver_blacklist(self, mock_get_devicelist,
+ mock_read_sys_net, mock_sys_dev_path):
+ devices = {
+ 'eth1': {
+ 'bridge': False, 'carrier': False, 'dormant': False,
+ 'operstate': 'down', 'address': '00:11:22:33:44:55',
+ 'device/driver': 'hv_netsvc', 'device/device': '0x3'},
+ 'eth0': {
+ 'bridge': False, 'carrier': False, 'dormant': False,
+ 'operstate': 'down', 'address': '00:11:22:33:44:55',
+ 'device/driver': 'mlx4_core', 'device/device': '0x7'},
+ }
+
+ tmp_dir = self.tmp_dir()
+ _setup_test(tmp_dir, mock_get_devicelist,
+ mock_read_sys_net, mock_sys_dev_path,
+ dev_attrs=devices)
+
+ blacklist = ['mlx4_core']
+ network_cfg = net.generate_fallback_config(blacklist_drivers=blacklist,
+ config_driver=True)
+ ns = network_state.parse_net_config_data(network_cfg,
+ skip_broken=False)
+
+ render_dir = os.path.join(tmp_dir, "render")
+ os.makedirs(render_dir)
+
+ # don't set rulepath so eni writes them
+ renderer = eni.Renderer(
+ {'eni_path': 'interfaces', 'netrules_path': 'netrules'})
+ renderer.render_network_state(ns, render_dir)
+
+ self.assertTrue(os.path.exists(os.path.join(render_dir,
+ 'interfaces')))
+ with open(os.path.join(render_dir, 'interfaces')) as fh:
+ contents = fh.read()
+ print(contents)
+ expected = """
+auto lo
+iface lo inet loopback
+
+auto eth1
+iface eth1 inet dhcp
+"""
+ self.assertEqual(expected.lstrip(), contents.lstrip())
+
+ self.assertTrue(os.path.exists(os.path.join(render_dir, 'netrules')))
+ with open(os.path.join(render_dir, 'netrules')) as fh:
+ contents = fh.read()
+ print(contents)
+ expected_rule = [
+ 'SUBSYSTEM=="net"',
+ 'ACTION=="add"',
+ 'DRIVERS=="hv_netsvc"',
+ 'ATTR{address}=="00:11:22:33:44:55"',
+ 'NAME="eth1"',
+ ]
+ self.assertEqual(", ".join(expected_rule) + '\n', contents.lstrip())
+
+
class TestSysConfigRendering(CiTestCase):
@mock.patch("cloudinit.net.sys_dev_path")
@@ -1560,6 +1698,118 @@ class TestNetRenderers(CiTestCase):
priority=['sysconfig', 'eni'])
+class TestGetInterfaces(CiTestCase):
+ _data = {'bonds': ['bond1'],
+ 'bridges': ['bridge1'],
+ 'vlans': ['bond1.101'],
+ 'own_macs': ['enp0s1', 'enp0s2', 'bridge1-nic', 'bridge1',
+ 'bond1.101', 'lo', 'eth1'],
+ 'macs': {'enp0s1': 'aa:aa:aa:aa:aa:01',
+ 'enp0s2': 'aa:aa:aa:aa:aa:02',
+ 'bond1': 'aa:aa:aa:aa:aa:01',
+ 'bond1.101': 'aa:aa:aa:aa:aa:01',
+ 'bridge1': 'aa:aa:aa:aa:aa:03',
+ 'bridge1-nic': 'aa:aa:aa:aa:aa:03',
+ 'lo': '00:00:00:00:00:00',
+ 'greptap0': '00:00:00:00:00:00',
+ 'eth1': 'aa:aa:aa:aa:aa:01',
+ 'tun0': None},
+ 'drivers': {'enp0s1': 'virtio_net',
+ 'enp0s2': 'e1000',
+ 'bond1': None,
+ 'bond1.101': None,
+ 'bridge1': None,
+ 'bridge1-nic': None,
+ 'lo': None,
+ 'greptap0': None,
+ 'eth1': 'mlx4_core',
+ 'tun0': None}}
+ data = {}
+
+ def _se_get_devicelist(self):
+ return list(self.data['devices'])
+
+ def _se_device_driver(self, name):
+ return self.data['drivers'][name]
+
+ def _se_device_devid(self, name):
+ return '0x%s' % sorted(list(self.data['drivers'].keys())).index(name)
+
+ def _se_get_interface_mac(self, name):
+ return self.data['macs'][name]
+
+ def _se_is_bridge(self, name):
+ return name in self.data['bridges']
+
+ def _se_is_vlan(self, name):
+ return name in self.data['vlans']
+
+ def _se_interface_has_own_mac(self, name):
+ return name in self.data['own_macs']
+
+ def _mock_setup(self):
+ self.data = copy.deepcopy(self._data)
+ self.data['devices'] = set(list(self.data['macs'].keys()))
+ mocks = ('get_devicelist', 'get_interface_mac', 'is_bridge',
+ 'interface_has_own_mac', 'is_vlan', 'device_driver',
+ 'device_devid')
+ self.mocks = {}
+ for n in mocks:
+ m = mock.patch('cloudinit.net.' + n,
+ side_effect=getattr(self, '_se_' + n))
+ self.addCleanup(m.stop)
+ self.mocks[n] = m.start()
+
+ def test_gi_includes_duplicate_macs(self):
+ self._mock_setup()
+ ret = net.get_interfaces()
+
+ self.assertIn('enp0s1', self._se_get_devicelist())
+ self.assertIn('eth1', self._se_get_devicelist())
+ found = [ent for ent in ret if 'aa:aa:aa:aa:aa:01' in ent]
+ self.assertEqual(len(found), 2)
+
+ def test_gi_excludes_any_without_mac_address(self):
+ self._mock_setup()
+ ret = net.get_interfaces()
+
+ self.assertIn('tun0', self._se_get_devicelist())
+ found = [ent for ent in ret if 'tun0' in ent]
+ self.assertEqual(len(found), 0)
+
+ def test_gi_excludes_stolen_macs(self):
+ self._mock_setup()
+ ret = net.get_interfaces()
+ self.mocks['interface_has_own_mac'].assert_has_calls(
+ [mock.call('enp0s1'), mock.call('bond1')], any_order=True)
+ expected = [
+ ('enp0s2', 'aa:aa:aa:aa:aa:02', 'e1000', '0x5'),
+ ('enp0s1', 'aa:aa:aa:aa:aa:01', 'virtio_net', '0x4'),
+ ('eth1', 'aa:aa:aa:aa:aa:01', 'mlx4_core', '0x6'),
+ ('lo', '00:00:00:00:00:00', None, '0x8'),
+ ('bridge1-nic', 'aa:aa:aa:aa:aa:03', None, '0x3'),
+ ]
+ self.assertEqual(sorted(expected), sorted(ret))
+
+ def test_gi_excludes_bridges(self):
+ self._mock_setup()
+ # add a device 'b1', make all return they have their "own mac",
+ # set everything other than 'b1' to be a bridge.
+ # then expect b1 is the only thing left.
+ self.data['macs']['b1'] = 'aa:aa:aa:aa:aa:b1'
+ self.data['drivers']['b1'] = None
+ self.data['devices'].add('b1')
+ self.data['bonds'] = []
+ self.data['own_macs'] = self.data['devices']
+ self.data['bridges'] = [f for f in self.data['devices'] if f != "b1"]
+ ret = net.get_interfaces()
+ self.assertEqual([('b1', 'aa:aa:aa:aa:aa:b1', None, '0x0')], ret)
+ self.mocks['is_bridge'].assert_has_calls(
+ [mock.call('bridge1'), mock.call('enp0s1'), mock.call('bond1'),
+ mock.call('b1')],
+ any_order=True)
+
+
class TestGetInterfacesByMac(CiTestCase):
_data = {'bonds': ['bond1'],
'bridges': ['bridge1'],
@@ -1691,4 +1941,202 @@ def _gzip_data(data):
gzfp.close()
return iobuf.getvalue()
+
+class TestRenameInterfaces(CiTestCase):
+
+ @mock.patch('cloudinit.util.subp')
+ def test_rename_all(self, mock_subp):
+ renames = [
+ ('00:11:22:33:44:55', 'interface0', 'virtio_net', '0x3'),
+ ('00:11:22:33:44:aa', 'interface2', 'virtio_net', '0x5'),
+ ]
+ current_info = {
+ 'ens3': {
+ 'downable': True,
+ 'device_id': '0x3',
+ 'driver': 'virtio_net',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'ens3',
+ 'up': False},
+ 'ens5': {
+ 'downable': True,
+ 'device_id': '0x5',
+ 'driver': 'virtio_net',
+ 'mac': '00:11:22:33:44:aa',
+ 'name': 'ens5',
+ 'up': False},
+ }
+ net._rename_interfaces(renames, current_info=current_info)
+ print(mock_subp.call_args_list)
+ mock_subp.assert_has_calls([
+ mock.call(['ip', 'link', 'set', 'ens3', 'name', 'interface0'],
+ capture=True),
+ mock.call(['ip', 'link', 'set', 'ens5', 'name', 'interface2'],
+ capture=True),
+ ])
+
+ @mock.patch('cloudinit.util.subp')
+ def test_rename_no_driver_no_device_id(self, mock_subp):
+ renames = [
+ ('00:11:22:33:44:55', 'interface0', None, None),
+ ('00:11:22:33:44:aa', 'interface1', None, None),
+ ]
+ current_info = {
+ 'eth0': {
+ 'downable': True,
+ 'device_id': None,
+ 'driver': None,
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'eth0',
+ 'up': False},
+ 'eth1': {
+ 'downable': True,
+ 'device_id': None,
+ 'driver': None,
+ 'mac': '00:11:22:33:44:aa',
+ 'name': 'eth1',
+ 'up': False},
+ }
+ net._rename_interfaces(renames, current_info=current_info)
+ print(mock_subp.call_args_list)
+ mock_subp.assert_has_calls([
+ mock.call(['ip', 'link', 'set', 'eth0', 'name', 'interface0'],
+ capture=True),
+ mock.call(['ip', 'link', 'set', 'eth1', 'name', 'interface1'],
+ capture=True),
+ ])
+
+ @mock.patch('cloudinit.util.subp')
+ def test_rename_all_bounce(self, mock_subp):
+ renames = [
+ ('00:11:22:33:44:55', 'interface0', 'virtio_net', '0x3'),
+ ('00:11:22:33:44:aa', 'interface2', 'virtio_net', '0x5'),
+ ]
+ current_info = {
+ 'ens3': {
+ 'downable': True,
+ 'device_id': '0x3',
+ 'driver': 'virtio_net',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'ens3',
+ 'up': True},
+ 'ens5': {
+ 'downable': True,
+ 'device_id': '0x5',
+ 'driver': 'virtio_net',
+ 'mac': '00:11:22:33:44:aa',
+ 'name': 'ens5',
+ 'up': True},
+ }
+ net._rename_interfaces(renames, current_info=current_info)
+ print(mock_subp.call_args_list)
+ mock_subp.assert_has_calls([
+ mock.call(['ip', 'link', 'set', 'ens3', 'down'], capture=True),
+ mock.call(['ip', 'link', 'set', 'ens3', 'name', 'interface0'],
+ capture=True),
+ mock.call(['ip', 'link', 'set', 'ens5', 'down'], capture=True),
+ mock.call(['ip', 'link', 'set', 'ens5', 'name', 'interface2'],
+ capture=True),
+ mock.call(['ip', 'link', 'set', 'interface0', 'up'], capture=True),
+ mock.call(['ip', 'link', 'set', 'interface2', 'up'], capture=True)
+ ])
+
+ @mock.patch('cloudinit.util.subp')
+ def test_rename_duplicate_macs(self, mock_subp):
+ renames = [
+ ('00:11:22:33:44:55', 'eth0', 'hv_netsvc', '0x3'),
+ ('00:11:22:33:44:55', 'vf1', 'mlx4_core', '0x5'),
+ ]
+ current_info = {
+ 'eth0': {
+ 'downable': True,
+ 'device_id': '0x3',
+ 'driver': 'hv_netsvc',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'eth0',
+ 'up': False},
+ 'eth1': {
+ 'downable': True,
+ 'device_id': '0x5',
+ 'driver': 'mlx4_core',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'eth1',
+ 'up': False},
+ }
+ net._rename_interfaces(renames, current_info=current_info)
+ print(mock_subp.call_args_list)
+ mock_subp.assert_has_calls([
+ mock.call(['ip', 'link', 'set', 'eth1', 'name', 'vf1'],
+ capture=True),
+ ])
+
+ @mock.patch('cloudinit.util.subp')
+ def test_rename_duplicate_macs_driver_no_devid(self, mock_subp):
+ renames = [
+ ('00:11:22:33:44:55', 'eth0', 'hv_netsvc', None),
+ ('00:11:22:33:44:55', 'vf1', 'mlx4_core', None),
+ ]
+ current_info = {
+ 'eth0': {
+ 'downable': True,
+ 'device_id': '0x3',
+ 'driver': 'hv_netsvc',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'eth0',
+ 'up': False},
+ 'eth1': {
+ 'downable': True,
+ 'device_id': '0x5',
+ 'driver': 'mlx4_core',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'eth1',
+ 'up': False},
+ }
+ net._rename_interfaces(renames, current_info=current_info)
+ print(mock_subp.call_args_list)
+ mock_subp.assert_has_calls([
+ mock.call(['ip', 'link', 'set', 'eth1', 'name', 'vf1'],
+ capture=True),
+ ])
+
+ @mock.patch('cloudinit.util.subp')
+ def test_rename_multi_mac_dups(self, mock_subp):
+ renames = [
+ ('00:11:22:33:44:55', 'eth0', 'hv_netsvc', '0x3'),
+ ('00:11:22:33:44:55', 'vf1', 'mlx4_core', '0x5'),
+ ('00:11:22:33:44:55', 'vf2', 'mlx4_core', '0x7'),
+ ]
+ current_info = {
+ 'eth0': {
+ 'downable': True,
+ 'device_id': '0x3',
+ 'driver': 'hv_netsvc',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'eth0',
+ 'up': False},
+ 'eth1': {
+ 'downable': True,
+ 'device_id': '0x5',
+ 'driver': 'mlx4_core',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'eth1',
+ 'up': False},
+ 'eth2': {
+ 'downable': True,
+ 'device_id': '0x7',
+ 'driver': 'mlx4_core',
+ 'mac': '00:11:22:33:44:55',
+ 'name': 'eth2',
+ 'up': False},
+ }
+ net._rename_interfaces(renames, current_info=current_info)
+ print(mock_subp.call_args_list)
+ mock_subp.assert_has_calls([
+ mock.call(['ip', 'link', 'set', 'eth1', 'name', 'vf1'],
+ capture=True),
+ mock.call(['ip', 'link', 'set', 'eth2', 'name', 'vf2'],
+ capture=True),
+ ])
+
+
# vi: ts=4 expandtab
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-15 20:35:20 +0000