T2117: Cloud-init updated to 22.1

Merged with 22.1 tag from the upstream Cloud-init repository.

Our modules were slightly modified for compatibility with the new
version.

1 files changed, 0 insertions, 1270 deletions

diff --git a/cloudinit/net/tests/test_init.py b/cloudinit/net/tests/test_init.py
deleted file mode 100644
index 0535387a..00000000
--- a/cloudinit/net/tests/test_init.py
+++ /dev/null
@@ -1,1270 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-import copy
-import errno
-import ipaddress
-import os
-import textwrap
-from unittest import mock
-
-import httpretty
-import pytest
-import requests
-
-import cloudinit.net as net
-from cloudinit import safeyaml as yaml
-from cloudinit.tests.helpers import CiTestCase, HttprettyTestCase
-from cloudinit.subp import ProcessExecutionError
-from cloudinit.util import ensure_file, write_file
-
-
-class TestSysDevPath(CiTestCase):
-
-    def test_sys_dev_path(self):
-        """sys_dev_path returns a path under SYS_CLASS_NET for a device."""
-        dev = 'something'
-        path = 'attribute'
-        expected = net.SYS_CLASS_NET + dev + '/' + path
-        self.assertEqual(expected, net.sys_dev_path(dev, path))
-
-    def test_sys_dev_path_without_path(self):
-        """When path param isn't provided it defaults to empty string."""
-        dev = 'something'
-        expected = net.SYS_CLASS_NET + dev + '/'
-        self.assertEqual(expected, net.sys_dev_path(dev))
-
-
-class TestReadSysNet(CiTestCase):
-    with_logs = True
-
-    def setUp(self):
-        super(TestReadSysNet, self).setUp()
-        sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
-        self.m_sys_path = sys_mock.start()
-        self.sysdir = self.tmp_dir() + '/'
-        self.m_sys_path.return_value = self.sysdir
-        self.addCleanup(sys_mock.stop)
-
-    def test_read_sys_net_strips_contents_of_sys_path(self):
-        """read_sys_net strips whitespace from the contents of a sys file."""
-        content = 'some stuff with trailing whitespace\t\r\n'
-        write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
-        self.assertEqual(content.strip(), net.read_sys_net('dev', 'attr'))
-
-    def test_read_sys_net_reraises_oserror(self):
-        """read_sys_net raises OSError/IOError when file doesn't exist."""
-        # Non-specific Exception because versions of python OSError vs IOError.
-        with self.assertRaises(Exception) as context_manager:  # noqa: H202
-            net.read_sys_net('dev', 'attr')
-        error = context_manager.exception
-        self.assertIn('No such file or directory', str(error))
-
-    def test_read_sys_net_handles_error_with_on_enoent(self):
-        """read_sys_net handles OSError/IOError with on_enoent if provided."""
-        handled_errors = []
-
-        def on_enoent(e):
-            handled_errors.append(e)
-
-        net.read_sys_net('dev', 'attr', on_enoent=on_enoent)
-        error = handled_errors[0]
-        self.assertIsInstance(error, Exception)
-        self.assertIn('No such file or directory', str(error))
-
-    def test_read_sys_net_translates_content(self):
-        """read_sys_net translates content when translate dict is provided."""
-        content = "you're welcome\n"
-        write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
-        translate = {"you're welcome": 'de nada'}
-        self.assertEqual(
-            'de nada',
-            net.read_sys_net('dev', 'attr', translate=translate))
-
-    def test_read_sys_net_errors_on_translation_failures(self):
-        """read_sys_net raises a KeyError and logs details on failure."""
-        content = "you're welcome\n"
-        write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
-        with self.assertRaises(KeyError) as context_manager:
-            net.read_sys_net('dev', 'attr', translate={})
-        error = context_manager.exception
-        self.assertEqual('"you\'re welcome"', str(error))
-        self.assertIn(
-            "Found unexpected (not translatable) value 'you're welcome' in "
-            "'{0}dev/attr".format(self.sysdir),
-            self.logs.getvalue())
-
-    def test_read_sys_net_handles_handles_with_onkeyerror(self):
-        """read_sys_net handles translation errors calling on_keyerror."""
-        content = "you're welcome\n"
-        write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
-        handled_errors = []
-
-        def on_keyerror(e):
-            handled_errors.append(e)
-
-        net.read_sys_net('dev', 'attr', translate={}, on_keyerror=on_keyerror)
-        error = handled_errors[0]
-        self.assertIsInstance(error, KeyError)
-        self.assertEqual('"you\'re welcome"', str(error))
-
-    def test_read_sys_net_safe_false_on_translate_failure(self):
-        """read_sys_net_safe returns False on translation failures."""
-        content = "you're welcome\n"
-        write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
-        self.assertFalse(net.read_sys_net_safe('dev', 'attr', translate={}))
-
-    def test_read_sys_net_safe_returns_false_on_noent_failure(self):
-        """read_sys_net_safe returns False on file not found failures."""
-        self.assertFalse(net.read_sys_net_safe('dev', 'attr'))
-
-    def test_read_sys_net_int_returns_none_on_error(self):
-        """read_sys_net_safe returns None on failures."""
-        self.assertFalse(net.read_sys_net_int('dev', 'attr'))
-
-    def test_read_sys_net_int_returns_none_on_valueerror(self):
-        """read_sys_net_safe returns None when content is not an int."""
-        write_file(os.path.join(self.sysdir, 'dev', 'attr'), 'NOTINT\n')
-        self.assertFalse(net.read_sys_net_int('dev', 'attr'))
-
-    def test_read_sys_net_int_returns_integer_from_content(self):
-        """read_sys_net_safe returns None on failures."""
-        write_file(os.path.join(self.sysdir, 'dev', 'attr'), '1\n')
-        self.assertEqual(1, net.read_sys_net_int('dev', 'attr'))
-
-    def test_is_up_true(self):
-        """is_up is True if sys/net/devname/operstate is 'up' or 'unknown'."""
-        for state in ['up', 'unknown']:
-            write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
-            self.assertTrue(net.is_up('eth0'))
-
-    def test_is_up_false(self):
-        """is_up is False if sys/net/devname/operstate is 'down' or invalid."""
-        for state in ['down', 'incomprehensible']:
-            write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
-            self.assertFalse(net.is_up('eth0'))
-
-    def test_is_bridge(self):
-        """is_bridge is True when /sys/net/devname/bridge exists."""
-        self.assertFalse(net.is_bridge('eth0'))
-        ensure_file(os.path.join(self.sysdir, 'eth0', 'bridge'))
-        self.assertTrue(net.is_bridge('eth0'))
-
-    def test_is_bond(self):
-        """is_bond is True when /sys/net/devname/bonding exists."""
-        self.assertFalse(net.is_bond('eth0'))
-        ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
-        self.assertTrue(net.is_bond('eth0'))
-
-    def test_get_master(self):
-        """get_master returns the path when /sys/net/devname/master exists."""
-        self.assertIsNone(net.get_master('enP1s1'))
-        master_path = os.path.join(self.sysdir, 'enP1s1', 'master')
-        ensure_file(master_path)
-        self.assertEqual(master_path, net.get_master('enP1s1'))
-
-    def test_master_is_bridge_or_bond(self):
-        bridge_mac = 'aa:bb:cc:aa:bb:cc'
-        bond_mac = 'cc:bb:aa:cc:bb:aa'
-
-        # No master => False
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), bridge_mac)
-        write_file(os.path.join(self.sysdir, 'eth2', 'address'), bond_mac)
-
-        self.assertFalse(net.master_is_bridge_or_bond('eth1'))
-        self.assertFalse(net.master_is_bridge_or_bond('eth2'))
-
-        # masters without bridge/bonding => False
-        write_file(os.path.join(self.sysdir, 'br0', 'address'), bridge_mac)
-        write_file(os.path.join(self.sysdir, 'bond0', 'address'), bond_mac)
-
-        os.symlink('../br0', os.path.join(self.sysdir, 'eth1', 'master'))
-        os.symlink('../bond0', os.path.join(self.sysdir, 'eth2', 'master'))
-
-        self.assertFalse(net.master_is_bridge_or_bond('eth1'))
-        self.assertFalse(net.master_is_bridge_or_bond('eth2'))
-
-        # masters with bridge/bonding => True
-        write_file(os.path.join(self.sysdir, 'br0', 'bridge'), '')
-        write_file(os.path.join(self.sysdir, 'bond0', 'bonding'), '')
-
-        self.assertTrue(net.master_is_bridge_or_bond('eth1'))
-        self.assertTrue(net.master_is_bridge_or_bond('eth2'))
-
-    def test_master_is_openvswitch(self):
-        ovs_mac = 'bb:cc:aa:bb:cc:aa'
-
-        # No master => False
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), ovs_mac)
-
-        self.assertFalse(net.master_is_bridge_or_bond('eth1'))
-
-        # masters without ovs-system => False
-        write_file(os.path.join(self.sysdir, 'ovs-system', 'address'), ovs_mac)
-
-        os.symlink('../ovs-system', os.path.join(self.sysdir, 'eth1',
-                   'master'))
-
-        self.assertFalse(net.master_is_openvswitch('eth1'))
-
-        # masters with ovs-system => True
-        os.symlink('../ovs-system', os.path.join(self.sysdir, 'eth1',
-                   'upper_ovs-system'))
-
-        self.assertTrue(net.master_is_openvswitch('eth1'))
-
-    def test_is_vlan(self):
-        """is_vlan is True when /sys/net/devname/uevent has DEVTYPE=vlan."""
-        ensure_file(os.path.join(self.sysdir, 'eth0', 'uevent'))
-        self.assertFalse(net.is_vlan('eth0'))
-        content = 'junk\nDEVTYPE=vlan\njunk\n'
-        write_file(os.path.join(self.sysdir, 'eth0', 'uevent'), content)
-        self.assertTrue(net.is_vlan('eth0'))
-
-
-class TestGenerateFallbackConfig(CiTestCase):
-
-    def setUp(self):
-        super(TestGenerateFallbackConfig, self).setUp()
-        sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
-        self.m_sys_path = sys_mock.start()
-        self.sysdir = self.tmp_dir() + '/'
-        self.m_sys_path.return_value = self.sysdir
-        self.addCleanup(sys_mock.stop)
-        self.add_patch('cloudinit.net.util.is_container', 'm_is_container',
-                       return_value=False)
-        self.add_patch('cloudinit.net.util.udevadm_settle', 'm_settle')
-        self.add_patch('cloudinit.net.is_netfailover', 'm_netfail',
-                       return_value=False)
-        self.add_patch('cloudinit.net.is_netfail_master', 'm_netfail_master',
-                       return_value=False)
-
-    def test_generate_fallback_finds_connected_eth_with_mac(self):
-        """generate_fallback_config finds any connected device with a mac."""
-        write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
-        write_file(os.path.join(self.sysdir, 'eth1', 'carrier'), '1')
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
-        expected = {
-            'ethernets': {'eth1': {'match': {'macaddress': mac},
-                                   'dhcp4': True, 'set-name': 'eth1'}},
-            'version': 2}
-        self.assertEqual(expected, net.generate_fallback_config())
-
-    def test_generate_fallback_finds_dormant_eth_with_mac(self):
-        """generate_fallback_config finds any dormant device with a mac."""
-        write_file(os.path.join(self.sysdir, 'eth0', 'dormant'), '1')
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
-        expected = {
-            'ethernets': {'eth0': {'match': {'macaddress': mac}, 'dhcp4': True,
-                                   'set-name': 'eth0'}},
-            'version': 2}
-        self.assertEqual(expected, net.generate_fallback_config())
-
-    def test_generate_fallback_finds_eth_by_operstate(self):
-        """generate_fallback_config finds any dormant device with a mac."""
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
-        expected = {
-            'ethernets': {
-                'eth0': {'dhcp4': True, 'match': {'macaddress': mac},
-                         'set-name': 'eth0'}},
-            'version': 2}
-        valid_operstates = ['dormant', 'down', 'lowerlayerdown', 'unknown']
-        for state in valid_operstates:
-            write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
-            self.assertEqual(expected, net.generate_fallback_config())
-        write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), 'noworky')
-        self.assertIsNone(net.generate_fallback_config())
-
-    def test_generate_fallback_config_skips_veth(self):
-        """generate_fallback_config will skip any veth interfaces."""
-        # A connected veth which gets ignored
-        write_file(os.path.join(self.sysdir, 'veth0', 'carrier'), '1')
-        self.assertIsNone(net.generate_fallback_config())
-
-    def test_generate_fallback_config_skips_bridges(self):
-        """generate_fallback_config will skip any bridges interfaces."""
-        # A connected veth which gets ignored
-        write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
-        ensure_file(os.path.join(self.sysdir, 'eth0', 'bridge'))
-        self.assertIsNone(net.generate_fallback_config())
-
-    def test_generate_fallback_config_skips_bonds(self):
-        """generate_fallback_config will skip any bonded interfaces."""
-        # A connected veth which gets ignored
-        write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
-        ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
-        self.assertIsNone(net.generate_fallback_config())
-
-    def test_generate_fallback_config_skips_netfail_devs(self):
-        """gen_fallback_config ignores netfail primary,sby no mac on master."""
-        mac = 'aa:bb:cc:aa:bb:cc'  # netfailover devs share the same mac
-        for iface in ['ens3', 'ens3sby', 'enP0s1f3']:
-            write_file(os.path.join(self.sysdir, iface, 'carrier'), '1')
-            write_file(
-                os.path.join(self.sysdir, iface, 'addr_assign_type'), '0')
-            write_file(
-                os.path.join(self.sysdir, iface, 'address'), mac)
-
-        def is_netfail(iface, _driver=None):
-            # ens3 is the master
-            if iface == 'ens3':
-                return False
-            return True
-        self.m_netfail.side_effect = is_netfail
-
-        def is_netfail_master(iface, _driver=None):
-            # ens3 is the master
-            if iface == 'ens3':
-                return True
-            return False
-        self.m_netfail_master.side_effect = is_netfail_master
-        expected = {
-            'ethernets': {
-                'ens3': {'dhcp4': True, 'match': {'name': 'ens3'},
-                         'set-name': 'ens3'}},
-            'version': 2}
-        result = net.generate_fallback_config()
-        self.assertEqual(expected, result)
-
-
-class TestNetFindFallBackNic(CiTestCase):
-
-    def setUp(self):
-        super(TestNetFindFallBackNic, self).setUp()
-        sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
-        self.m_sys_path = sys_mock.start()
-        self.sysdir = self.tmp_dir() + '/'
-        self.m_sys_path.return_value = self.sysdir
-        self.addCleanup(sys_mock.stop)
-        self.add_patch('cloudinit.net.util.is_container', 'm_is_container',
-                       return_value=False)
-        self.add_patch('cloudinit.net.util.udevadm_settle', 'm_settle')
-
-    def test_generate_fallback_finds_first_connected_eth_with_mac(self):
-        """find_fallback_nic finds any connected device with a mac."""
-        write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
-        write_file(os.path.join(self.sysdir, 'eth1', 'carrier'), '1')
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
-        self.assertEqual('eth1', net.find_fallback_nic())
-
-
-class TestGetDeviceList(CiTestCase):
-
-    def setUp(self):
-        super(TestGetDeviceList, self).setUp()
-        sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
-        self.m_sys_path = sys_mock.start()
-        self.sysdir = self.tmp_dir() + '/'
-        self.m_sys_path.return_value = self.sysdir
-        self.addCleanup(sys_mock.stop)
-
-    def test_get_devicelist_raise_oserror(self):
-        """get_devicelist raise any non-ENOENT OSerror."""
-        error = OSError('Can not do it')
-        error.errno = errno.EPERM  # Set non-ENOENT
-        self.m_sys_path.side_effect = error
-        with self.assertRaises(OSError) as context_manager:
-            net.get_devicelist()
-        exception = context_manager.exception
-        self.assertEqual('Can not do it', str(exception))
-
-    def test_get_devicelist_empty_without_sys_net(self):
-        """get_devicelist returns empty list when missing SYS_CLASS_NET."""
-        self.m_sys_path.return_value = 'idontexist'
-        self.assertEqual([], net.get_devicelist())
-
-    def test_get_devicelist_empty_with_no_devices_in_sys_net(self):
-        """get_devicelist returns empty directoty listing for SYS_CLASS_NET."""
-        self.assertEqual([], net.get_devicelist())
-
-    def test_get_devicelist_lists_any_subdirectories_in_sys_net(self):
-        """get_devicelist returns a directory listing for SYS_CLASS_NET."""
-        write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), 'up')
-        write_file(os.path.join(self.sysdir, 'eth1', 'operstate'), 'up')
-        self.assertCountEqual(['eth0', 'eth1'], net.get_devicelist())
-
-
-class TestGetInterfaceMAC(CiTestCase):
-
-    def setUp(self):
-        super(TestGetInterfaceMAC, self).setUp()
-        sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
-        self.m_sys_path = sys_mock.start()
-        self.sysdir = self.tmp_dir() + '/'
-        self.m_sys_path.return_value = self.sysdir
-        self.addCleanup(sys_mock.stop)
-
-    def test_get_interface_mac_false_with_no_mac(self):
-        """get_device_list returns False when no mac is reported."""
-        ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
-        mac_path = os.path.join(self.sysdir, 'eth0', 'address')
-        self.assertFalse(os.path.exists(mac_path))
-        self.assertFalse(net.get_interface_mac('eth0'))
-
-    def test_get_interface_mac(self):
-        """get_interfaces returns the mac from SYS_CLASS_NET/dev/address."""
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
-        self.assertEqual(mac, net.get_interface_mac('eth1'))
-
-    def test_get_interface_mac_grabs_bonding_address(self):
-        """get_interfaces returns the source device mac for bonded devices."""
-        source_dev_mac = 'aa:bb:cc:aa:bb:cc'
-        bonded_mac = 'dd:ee:ff:dd:ee:ff'
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), bonded_mac)
-        write_file(
-            os.path.join(self.sysdir, 'eth1', 'bonding_slave', 'perm_hwaddr'),
-            source_dev_mac)
-        self.assertEqual(source_dev_mac, net.get_interface_mac('eth1'))
-
-    def test_get_interfaces_empty_list_without_sys_net(self):
-        """get_interfaces returns an empty list when missing SYS_CLASS_NET."""
-        self.m_sys_path.return_value = 'idontexist'
-        self.assertEqual([], net.get_interfaces())
-
-    def test_get_interfaces_by_mac_skips_empty_mac(self):
-        """Ignore 00:00:00:00:00:00 addresses from get_interfaces_by_mac."""
-        empty_mac = '00:00:00:00:00:00'
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), empty_mac)
-        write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
-        write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
-        write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac)
-        expected = [('eth2', 'aa:bb:cc:aa:bb:cc', None, None)]
-        self.assertEqual(expected, net.get_interfaces())
-
-    def test_get_interfaces_by_mac_skips_missing_mac(self):
-        """Ignore interfaces without an address from get_interfaces_by_mac."""
-        write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
-        address_path = os.path.join(self.sysdir, 'eth1', 'address')
-        self.assertFalse(os.path.exists(address_path))
-        mac = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
-        write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac)
-        expected = [('eth2', 'aa:bb:cc:aa:bb:cc', None, None)]
-        self.assertEqual(expected, net.get_interfaces())
-
-    def test_get_interfaces_by_mac_skips_master_devs(self):
-        """Ignore interfaces with a master device which would have dup mac."""
-        mac1 = mac2 = 'aa:bb:cc:aa:bb:cc'
-        write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac1)
-        write_file(os.path.join(self.sysdir, 'eth1', 'master'), "blah")
-        write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
-        write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac2)
-        expected = [('eth2', mac2, None, None)]
-        self.assertEqual(expected, net.get_interfaces())
-
-    @mock.patch('cloudinit.net.is_netfailover')
-    def test_get_interfaces_by_mac_skips_netfailvoer(self, m_netfail):
-        """Ignore interfaces if netfailover primary or standby."""
-        mac = 'aa:bb:cc:aa:bb:cc'  # netfailover devs share the same mac
-        for iface in ['ens3', 'ens3sby', 'enP0s1f3']:
-            write_file(
-                os.path.join(self.sysdir, iface, 'addr_assign_type'), '0')
-            write_file(
-                os.path.join(self.sysdir, iface, 'address'), mac)
-
-        def is_netfail(iface, _driver=None):
-            # ens3 is the master
-            if iface == 'ens3':
-                return False
-            else:
-                return True
-        m_netfail.side_effect = is_netfail
-        expected = [('ens3', mac, None, None)]
-        self.assertEqual(expected, net.get_interfaces())
-
-    def test_get_interfaces_does_not_skip_phys_members_of_bridges_and_bonds(
-        self
-    ):
-        bridge_mac = 'aa:bb:cc:aa:bb:cc'
-        bond_mac = 'cc:bb:aa:cc:bb:aa'
-        ovs_mac = 'bb:cc:aa:bb:cc:aa'
-
-        write_file(os.path.join(self.sysdir, 'br0', 'address'), bridge_mac)
-        write_file(os.path.join(self.sysdir, 'br0', 'bridge'), '')
-
-        write_file(os.path.join(self.sysdir, 'bond0', 'address'), bond_mac)
-        write_file(os.path.join(self.sysdir, 'bond0', 'bonding'), '')
-
-        write_file(os.path.join(self.sysdir, 'ovs-system', 'address'),
-                   ovs_mac)
-
-        write_file(os.path.join(self.sysdir, 'eth1', 'address'), bridge_mac)
-        os.symlink('../br0', os.path.join(self.sysdir, 'eth1', 'master'))
-
-        write_file(os.path.join(self.sysdir, 'eth2', 'address'), bond_mac)
-        os.symlink('../bond0', os.path.join(self.sysdir, 'eth2', 'master'))
-
-        write_file(os.path.join(self.sysdir, 'eth3', 'address'), ovs_mac)
-        os.symlink('../ovs-system', os.path.join(self.sysdir, 'eth3',
-                   'master'))
-        os.symlink('../ovs-system', os.path.join(self.sysdir, 'eth3',
-                   'upper_ovs-system'))
-
-        interface_names = [interface[0] for interface in net.get_interfaces()]
-        self.assertEqual(['eth1', 'eth2', 'eth3', 'ovs-system'],
-                         sorted(interface_names))
-
-
-class TestInterfaceHasOwnMAC(CiTestCase):
-
-    def setUp(self):
-        super(TestInterfaceHasOwnMAC, self).setUp()
-        sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
-        self.m_sys_path = sys_mock.start()
-        self.sysdir = self.tmp_dir() + '/'
-        self.m_sys_path.return_value = self.sysdir
-        self.addCleanup(sys_mock.stop)
-
-    def test_interface_has_own_mac_false_when_stolen(self):
-        """Return False from interface_has_own_mac when address is stolen."""
-        write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '2')
-        self.assertFalse(net.interface_has_own_mac('eth1'))
-
-    def test_interface_has_own_mac_true_when_not_stolen(self):
-        """Return False from interface_has_own_mac when mac isn't stolen."""
-        valid_assign_types = ['0', '1', '3']
-        assign_path = os.path.join(self.sysdir, 'eth1', 'addr_assign_type')
-        for _type in valid_assign_types:
-            write_file(assign_path, _type)
-            self.assertTrue(net.interface_has_own_mac('eth1'))
-
-    def test_interface_has_own_mac_strict_errors_on_absent_assign_type(self):
-        """When addr_assign_type is absent, interface_has_own_mac errors."""
-        with self.assertRaises(ValueError):
-            net.interface_has_own_mac('eth1', strict=True)
-
-
-@mock.patch('cloudinit.net.subp.subp')
-class TestEphemeralIPV4Network(CiTestCase):
-
-    with_logs = True
-
-    def setUp(self):
-        super(TestEphemeralIPV4Network, self).setUp()
-        sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
-        self.m_sys_path = sys_mock.start()
-        self.sysdir = self.tmp_dir() + '/'
-        self.m_sys_path.return_value = self.sysdir
-        self.addCleanup(sys_mock.stop)
-
-    def test_ephemeral_ipv4_network_errors_on_missing_params(self, m_subp):
-        """No required params for EphemeralIPv4Network can be None."""
-        required_params = {
-            'interface': 'eth0', 'ip': '192.168.2.2',
-            'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
-        for key in required_params.keys():
-            params = copy.deepcopy(required_params)
-            params[key] = None
-            with self.assertRaises(ValueError) as context_manager:
-                net.EphemeralIPv4Network(**params)
-            error = context_manager.exception
-            self.assertIn('Cannot init network on', str(error))
-            self.assertEqual(0, m_subp.call_count)
-
-    def test_ephemeral_ipv4_network_errors_invalid_mask_prefix(self, m_subp):
-        """Raise an error when prefix_or_mask is not a netmask or prefix."""
-        params = {
-            'interface': 'eth0', 'ip': '192.168.2.2',
-            'broadcast': '192.168.2.255'}
-        invalid_masks = ('invalid', 'invalid.', '123.123.123')
-        for error_val in invalid_masks:
-            params['prefix_or_mask'] = error_val
-            with self.assertRaises(ValueError) as context_manager:
-                with net.EphemeralIPv4Network(**params):
-                    pass
-            error = context_manager.exception
-            self.assertIn('Cannot setup network: netmask', str(error))
-            self.assertEqual(0, m_subp.call_count)
-
-    def test_ephemeral_ipv4_network_performs_teardown(self, m_subp):
-        """EphemeralIPv4Network performs teardown on the device if setup."""
-        expected_setup_calls = [
-            mock.call(
-                ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
-                 'broadcast', '192.168.2.255', 'dev', 'eth0'],
-                capture=True, update_env={'LANG': 'C'}),
-            mock.call(
-                ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
-                capture=True)]
-        expected_teardown_calls = [
-            mock.call(
-                ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0',
-                 'down'], capture=True),
-            mock.call(
-                ['ip', '-family', 'inet', 'addr', 'del', '192.168.2.2/24',
-                 'dev', 'eth0'], capture=True)]
-        params = {
-            'interface': 'eth0', 'ip': '192.168.2.2',
-            'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
-        with net.EphemeralIPv4Network(**params):
-            self.assertEqual(expected_setup_calls, m_subp.call_args_list)
-        m_subp.assert_has_calls(expected_teardown_calls)
-
-    @mock.patch('cloudinit.net.readurl')
-    def test_ephemeral_ipv4_no_network_if_url_connectivity(
-            self, m_readurl, m_subp):
-        """No network setup is performed if we can successfully connect to
-        connectivity_url."""
-        params = {
-            'interface': 'eth0', 'ip': '192.168.2.2',
-            'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255',
-            'connectivity_url': 'http://example.org/index.html'}
-
-        with net.EphemeralIPv4Network(**params):
-            self.assertEqual([mock.call('http://example.org/index.html',
-                                        timeout=5)], m_readurl.call_args_list)
-        # Ensure that no teardown happens:
-        m_subp.assert_has_calls([])
-
-    def test_ephemeral_ipv4_network_noop_when_configured(self, m_subp):
-        """EphemeralIPv4Network handles exception when address is setup.
-
-        It performs no cleanup as the interface was already setup.
-        """
-        params = {
-            'interface': 'eth0', 'ip': '192.168.2.2',
-            'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
-        m_subp.side_effect = ProcessExecutionError(
-            '', 'RTNETLINK answers: File exists', 2)
-        expected_calls = [
-            mock.call(
-                ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
-                 'broadcast', '192.168.2.255', 'dev', 'eth0'],
-                capture=True, update_env={'LANG': 'C'})]
-        with net.EphemeralIPv4Network(**params):
-            pass
-        self.assertEqual(expected_calls, m_subp.call_args_list)
-        self.assertIn(
-            'Skip ephemeral network setup, eth0 already has address',
-            self.logs.getvalue())
-
-    def test_ephemeral_ipv4_network_with_prefix(self, m_subp):
-        """EphemeralIPv4Network takes a valid prefix to setup the network."""
-        params = {
-            'interface': 'eth0', 'ip': '192.168.2.2',
-            'prefix_or_mask': '24', 'broadcast': '192.168.2.255'}
-        for prefix_val in ['24', 16]:  # prefix can be int or string
-            params['prefix_or_mask'] = prefix_val
-            with net.EphemeralIPv4Network(**params):
-                pass
-        m_subp.assert_has_calls([mock.call(
-            ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
-             'broadcast', '192.168.2.255', 'dev', 'eth0'],
-            capture=True, update_env={'LANG': 'C'})])
-        m_subp.assert_has_calls([mock.call(
-            ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/16',
-             'broadcast', '192.168.2.255', 'dev', 'eth0'],
-            capture=True, update_env={'LANG': 'C'})])
-
-    def test_ephemeral_ipv4_network_with_new_default_route(self, m_subp):
-        """Add the route when router is set and no default route exists."""
-        params = {
-            'interface': 'eth0', 'ip': '192.168.2.2',
-            'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255',
-            'router': '192.168.2.1'}
-        m_subp.return_value = '', ''  # Empty response from ip route gw check
-        expected_setup_calls = [
-            mock.call(
-                ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
-                 'broadcast', '192.168.2.255', 'dev', 'eth0'],
-                capture=True, update_env={'LANG': 'C'}),
-            mock.call(
-                ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
-                capture=True),
-            mock.call(
-                ['ip', 'route', 'show', '0.0.0.0/0'], capture=True),
-            mock.call(['ip', '-4', 'route', 'add', '192.168.2.1',
-                       'dev', 'eth0', 'src', '192.168.2.2'], capture=True),
-            mock.call(
-                ['ip', '-4', 'route', 'add', 'default', 'via',
-                 '192.168.2.1', 'dev', 'eth0'], capture=True)]
-        expected_teardown_calls = [
-            mock.call(['ip', '-4', 'route', 'del', 'default', 'dev', 'eth0'],
-                      capture=True),
-            mock.call(['ip', '-4', 'route', 'del', '192.168.2.1',
-                       'dev', 'eth0', 'src', '192.168.2.2'], capture=True),
-        ]
-
-        with net.EphemeralIPv4Network(**params):
-            self.assertEqual(expected_setup_calls, m_subp.call_args_list)
-        m_subp.assert_has_calls(expected_teardown_calls)
-
-    def test_ephemeral_ipv4_network_with_rfc3442_static_routes(self, m_subp):
-        params = {
-            'interface': 'eth0', 'ip': '192.168.2.2',
-            'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255',
-            'static_routes': [('169.254.169.254/32', '192.168.2.1'),
-                              ('0.0.0.0/0', '192.168.2.1')],
-            'router': '192.168.2.1'}
-        expected_setup_calls = [
-            mock.call(
-                ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
-                 'broadcast', '192.168.2.255', 'dev', 'eth0'],
-                capture=True, update_env={'LANG': 'C'}),
-            mock.call(
-                ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
-                capture=True),
-            mock.call(
-                ['ip', '-4', 'route', 'add', '169.254.169.254/32',
-                 'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
-            mock.call(
-                ['ip', '-4', 'route', 'add', '0.0.0.0/0',
-                 'via', '192.168.2.1', 'dev', 'eth0'], capture=True)]
-        expected_teardown_calls = [
-            mock.call(
-                ['ip', '-4', 'route', 'del', '0.0.0.0/0',
-                 'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
-            mock.call(
-                ['ip', '-4', 'route', 'del', '169.254.169.254/32',
-                 'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
-            mock.call(
-                ['ip', '-family', 'inet', 'link', 'set', 'dev',
-                 'eth0', 'down'], capture=True),
-            mock.call(
-                ['ip', '-family', 'inet', 'addr', 'del',
-                 '192.168.2.2/24', 'dev', 'eth0'], capture=True)
-        ]
-        with net.EphemeralIPv4Network(**params):
-            self.assertEqual(expected_setup_calls, m_subp.call_args_list)
-        m_subp.assert_has_calls(expected_setup_calls + expected_teardown_calls)
-
-
-class TestApplyNetworkCfgNames(CiTestCase):
-    V1_CONFIG = textwrap.dedent("""\
-        version: 1
-        config:
-            - type: physical
-              name: interface0
-              mac_address: "52:54:00:12:34:00"
-              subnets:
-                  - type: static
-                    address: 10.0.2.15
-                    netmask: 255.255.255.0
-                    gateway: 10.0.2.2
-    """)
-    V2_CONFIG = textwrap.dedent("""\
-      version: 2
-      ethernets:
-          interface0:
-            match:
-              macaddress: "52:54:00:12:34:00"
-            addresses:
-              - 10.0.2.15/24
-            gateway4: 10.0.2.2
-            set-name: interface0
-    """)
-
-    V2_CONFIG_NO_SETNAME = textwrap.dedent("""\
-      version: 2
-      ethernets:
-          interface0:
-            match:
-              macaddress: "52:54:00:12:34:00"
-            addresses:
-              - 10.0.2.15/24
-            gateway4: 10.0.2.2
-    """)
-
-    V2_CONFIG_NO_MAC = textwrap.dedent("""\
-      version: 2
-      ethernets:
-          interface0:
-            match:
-              driver: virtio-net
-            addresses:
-              - 10.0.2.15/24
-            gateway4: 10.0.2.2
-            set-name: interface0
-    """)
-
-    @mock.patch('cloudinit.net.device_devid')
-    @mock.patch('cloudinit.net.device_driver')
-    @mock.patch('cloudinit.net._rename_interfaces')
-    def test_apply_v1_renames(self, m_rename_interfaces, m_device_driver,
-                              m_device_devid):
-        m_device_driver.return_value = 'virtio_net'
-        m_device_devid.return_value = '0x15d8'
-
-        net.apply_network_config_names(yaml.load(self.V1_CONFIG))
-
-        call = ['52:54:00:12:34:00', 'interface0', 'virtio_net', '0x15d8']
-        m_rename_interfaces.assert_called_with([call])
-
-    @mock.patch('cloudinit.net.device_devid')
-    @mock.patch('cloudinit.net.device_driver')
-    @mock.patch('cloudinit.net._rename_interfaces')
-    def test_apply_v2_renames(self, m_rename_interfaces, m_device_driver,
-                              m_device_devid):
-        m_device_driver.return_value = 'virtio_net'
-        m_device_devid.return_value = '0x15d8'
-
-        net.apply_network_config_names(yaml.load(self.V2_CONFIG))
-
-        call = ['52:54:00:12:34:00', 'interface0', 'virtio_net', '0x15d8']
-        m_rename_interfaces.assert_called_with([call])
-
-    @mock.patch('cloudinit.net._rename_interfaces')
-    def test_apply_v2_renames_skips_without_setname(self, m_rename_interfaces):
-        net.apply_network_config_names(yaml.load(self.V2_CONFIG_NO_SETNAME))
-        m_rename_interfaces.assert_called_with([])
-
-    @mock.patch('cloudinit.net._rename_interfaces')
-    def test_apply_v2_renames_skips_without_mac(self, m_rename_interfaces):
-        net.apply_network_config_names(yaml.load(self.V2_CONFIG_NO_MAC))
-        m_rename_interfaces.assert_called_with([])
-
-    def test_apply_v2_renames_raises_runtime_error_on_unknown_version(self):
-        with self.assertRaises(RuntimeError):
-            net.apply_network_config_names(yaml.load("version: 3"))
-
-
-class TestHasURLConnectivity(HttprettyTestCase):
-
-    def setUp(self):
-        super(TestHasURLConnectivity, self).setUp()
-        self.url = 'http://fake/'
-        self.kwargs = {'allow_redirects': True, 'timeout': 5.0}
-
-    @mock.patch('cloudinit.net.readurl')
-    def test_url_timeout_on_connectivity_check(self, m_readurl):
-        """A timeout of 5 seconds is provided when reading a url."""
-        self.assertTrue(
-            net.has_url_connectivity(self.url), 'Expected True on url connect')
-
-    def test_true_on_url_connectivity_success(self):
-        httpretty.register_uri(httpretty.GET, self.url)
-        self.assertTrue(
-            net.has_url_connectivity(self.url), 'Expected True on url connect')
-
-    @mock.patch('requests.Session.request')
-    def test_true_on_url_connectivity_timeout(self, m_request):
-        """A timeout raised accessing the url will return False."""
-        m_request.side_effect = requests.Timeout('Fake Connection Timeout')
-        self.assertFalse(
-            net.has_url_connectivity(self.url),
-            'Expected False on url timeout')
-
-    def test_true_on_url_connectivity_failure(self):
-        httpretty.register_uri(httpretty.GET, self.url, body={}, status=404)
-        self.assertFalse(
-            net.has_url_connectivity(self.url), 'Expected False on url fail')
-
-
-def _mk_v1_phys(mac, name, driver, device_id):
-    v1_cfg = {'type': 'physical', 'name': name, 'mac_address': mac}
-    params = {}
-    if driver:
-        params.update({'driver': driver})
-    if device_id:
-        params.update({'device_id': device_id})
-
-    if params:
-        v1_cfg.update({'params': params})
-
-    return v1_cfg
-
-
-def _mk_v2_phys(mac, name, driver=None, device_id=None):
-    v2_cfg = {'set-name': name, 'match': {'macaddress': mac}}
-    if driver:
-        v2_cfg['match'].update({'driver': driver})
-    if device_id:
-        v2_cfg['match'].update({'device_id': device_id})
-
-    return v2_cfg
-
-
-class TestExtractPhysdevs(CiTestCase):
-
-    def setUp(self):
-        super(TestExtractPhysdevs, self).setUp()
-        self.add_patch('cloudinit.net.device_driver', 'm_driver')
-        self.add_patch('cloudinit.net.device_devid', 'm_devid')
-
-    def test_extract_physdevs_looks_up_driver_v1(self):
-        driver = 'virtio'
-        self.m_driver.return_value = driver
-        physdevs = [
-            ['aa:bb:cc:dd:ee:ff', 'eth0', None, '0x1000'],
-        ]
-        netcfg = {
-            'version': 1,
-            'config': [_mk_v1_phys(*args) for args in physdevs],
-        }
-        # insert the driver value for verification
-        physdevs[0][2] = driver
-        self.assertEqual(sorted(physdevs),
-                         sorted(net.extract_physdevs(netcfg)))
-        self.m_driver.assert_called_with('eth0')
-
-    def test_extract_physdevs_looks_up_driver_v2(self):
-        driver = 'virtio'
-        self.m_driver.return_value = driver
-        physdevs = [
-            ['aa:bb:cc:dd:ee:ff', 'eth0', None, '0x1000'],
-        ]
-        netcfg = {
-            'version': 2,
-            'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
-        }
-        # insert the driver value for verification
-        physdevs[0][2] = driver
-        self.assertEqual(sorted(physdevs),
-                         sorted(net.extract_physdevs(netcfg)))
-        self.m_driver.assert_called_with('eth0')
-
-    def test_extract_physdevs_looks_up_devid_v1(self):
-        devid = '0x1000'
-        self.m_devid.return_value = devid
-        physdevs = [
-            ['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', None],
-        ]
-        netcfg = {
-            'version': 1,
-            'config': [_mk_v1_phys(*args) for args in physdevs],
-        }
-        # insert the driver value for verification
-        physdevs[0][3] = devid
-        self.assertEqual(sorted(physdevs),
-                         sorted(net.extract_physdevs(netcfg)))
-        self.m_devid.assert_called_with('eth0')
-
-    def test_extract_physdevs_looks_up_devid_v2(self):
-        devid = '0x1000'
-        self.m_devid.return_value = devid
-        physdevs = [
-            ['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', None],
-        ]
-        netcfg = {
-            'version': 2,
-            'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
-        }
-        # insert the driver value for verification
-        physdevs[0][3] = devid
-        self.assertEqual(sorted(physdevs),
-                         sorted(net.extract_physdevs(netcfg)))
-        self.m_devid.assert_called_with('eth0')
-
-    def test_get_v1_type_physical(self):
-        physdevs = [
-            ['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', '0x1000'],
-            ['00:11:22:33:44:55', 'ens3', 'e1000', '0x1643'],
-            ['09:87:65:43:21:10', 'ens0p1', 'mlx4_core', '0:0:1000'],
-        ]
-        netcfg = {
-            'version': 1,
-            'config': [_mk_v1_phys(*args) for args in physdevs],
-        }
-        self.assertEqual(sorted(physdevs),
-                         sorted(net.extract_physdevs(netcfg)))
-
-    def test_get_v2_type_physical(self):
-        physdevs = [
-            ['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', '0x1000'],
-            ['00:11:22:33:44:55', 'ens3', 'e1000', '0x1643'],
-            ['09:87:65:43:21:10', 'ens0p1', 'mlx4_core', '0:0:1000'],
-        ]
-        netcfg = {
-            'version': 2,
-            'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
-        }
-        self.assertEqual(sorted(physdevs),
-                         sorted(net.extract_physdevs(netcfg)))
-
-    def test_get_v2_type_physical_skips_if_no_set_name(self):
-        netcfg = {
-            'version': 2,
-            'ethernets': {
-                'ens3': {
-                    'match': {'macaddress': '00:11:22:33:44:55'},
-                }
-            }
-        }
-        self.assertEqual([], net.extract_physdevs(netcfg))
-
-    def test_runtime_error_on_unknown_netcfg_version(self):
-        with self.assertRaises(RuntimeError):
-            net.extract_physdevs({'version': 3, 'awesome_config': []})
-
-
-class TestNetFailOver(CiTestCase):
-
-    def setUp(self):
-        super(TestNetFailOver, self).setUp()
-        self.add_patch('cloudinit.net.util', 'm_util')
-        self.add_patch('cloudinit.net.read_sys_net', 'm_read_sys_net')
-        self.add_patch('cloudinit.net.device_driver', 'm_device_driver')
-
-    def test_get_dev_features(self):
-        devname = self.random_string()
-        features = self.random_string()
-        self.m_read_sys_net.return_value = features
-
-        self.assertEqual(features, net.get_dev_features(devname))
-        self.assertEqual(1, self.m_read_sys_net.call_count)
-        self.assertEqual(mock.call(devname, 'device/features'),
-                         self.m_read_sys_net.call_args_list[0])
-
-    def test_get_dev_features_none_returns_empty_string(self):
-        devname = self.random_string()
-        self.m_read_sys_net.side_effect = Exception('error')
-        self.assertEqual('', net.get_dev_features(devname))
-        self.assertEqual(1, self.m_read_sys_net.call_count)
-        self.assertEqual(mock.call(devname, 'device/features'),
-                         self.m_read_sys_net.call_args_list[0])
-
-    @mock.patch('cloudinit.net.get_dev_features')
-    def test_has_netfail_standby_feature(self, m_dev_features):
-        devname = self.random_string()
-        standby_features = ('0' * 62) + '1' + '0'
-        m_dev_features.return_value = standby_features
-        self.assertTrue(net.has_netfail_standby_feature(devname))
-
-    @mock.patch('cloudinit.net.get_dev_features')
-    def test_has_netfail_standby_feature_short_is_false(self, m_dev_features):
-        devname = self.random_string()
-        standby_features = self.random_string()
-        m_dev_features.return_value = standby_features
-        self.assertFalse(net.has_netfail_standby_feature(devname))
-
-    @mock.patch('cloudinit.net.get_dev_features')
-    def test_has_netfail_standby_feature_not_present_is_false(self,
-                                                              m_dev_features):
-        devname = self.random_string()
-        standby_features = '0' * 64
-        m_dev_features.return_value = standby_features
-        self.assertFalse(net.has_netfail_standby_feature(devname))
-
-    @mock.patch('cloudinit.net.get_dev_features')
-    def test_has_netfail_standby_feature_no_features_is_false(self,
-                                                              m_dev_features):
-        devname = self.random_string()
-        standby_features = None
-        m_dev_features.return_value = standby_features
-        self.assertFalse(net.has_netfail_standby_feature(devname))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    def test_is_netfail_master(self, m_exists, m_standby):
-        devname = self.random_string()
-        driver = 'virtio_net'
-        m_exists.return_value = False  # no master sysfs attr
-        m_standby.return_value = True  # has standby feature flag
-        self.assertTrue(net.is_netfail_master(devname, driver))
-
-    @mock.patch('cloudinit.net.sys_dev_path')
-    def test_is_netfail_master_checks_master_attr(self, m_sysdev):
-        devname = self.random_string()
-        driver = 'virtio_net'
-        m_sysdev.return_value = self.random_string()
-        self.assertFalse(net.is_netfail_master(devname, driver))
-        self.assertEqual(1, m_sysdev.call_count)
-        self.assertEqual(mock.call(devname, path='master'),
-                         m_sysdev.call_args_list[0])
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    def test_is_netfail_master_wrong_driver(self, m_exists, m_standby):
-        devname = self.random_string()
-        driver = self.random_string()
-        self.assertFalse(net.is_netfail_master(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    def test_is_netfail_master_has_master_attr(self, m_exists, m_standby):
-        devname = self.random_string()
-        driver = 'virtio_net'
-        m_exists.return_value = True  # has master sysfs attr
-        self.assertFalse(net.is_netfail_master(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    def test_is_netfail_master_no_standby_feat(self, m_exists, m_standby):
-        devname = self.random_string()
-        driver = 'virtio_net'
-        m_exists.return_value = False  # no master sysfs attr
-        m_standby.return_value = False  # no standby feature flag
-        self.assertFalse(net.is_netfail_master(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    @mock.patch('cloudinit.net.sys_dev_path')
-    def test_is_netfail_primary(self, m_sysdev, m_exists, m_standby):
-        devname = self.random_string()
-        driver = self.random_string()  # device not virtio_net
-        master_devname = self.random_string()
-        m_sysdev.return_value = "%s/%s" % (self.random_string(),
-                                           master_devname)
-        m_exists.return_value = True  # has master sysfs attr
-        self.m_device_driver.return_value = 'virtio_net'  # master virtio_net
-        m_standby.return_value = True  # has standby feature flag
-        self.assertTrue(net.is_netfail_primary(devname, driver))
-        self.assertEqual(1, self.m_device_driver.call_count)
-        self.assertEqual(mock.call(master_devname),
-                         self.m_device_driver.call_args_list[0])
-        self.assertEqual(1, m_standby.call_count)
-        self.assertEqual(mock.call(master_devname),
-                         m_standby.call_args_list[0])
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    @mock.patch('cloudinit.net.sys_dev_path')
-    def test_is_netfail_primary_wrong_driver(self, m_sysdev, m_exists,
-                                             m_standby):
-        devname = self.random_string()
-        driver = 'virtio_net'
-        self.assertFalse(net.is_netfail_primary(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    @mock.patch('cloudinit.net.sys_dev_path')
-    def test_is_netfail_primary_no_master(self, m_sysdev, m_exists, m_standby):
-        devname = self.random_string()
-        driver = self.random_string()  # device not virtio_net
-        m_exists.return_value = False  # no master sysfs attr
-        self.assertFalse(net.is_netfail_primary(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    @mock.patch('cloudinit.net.sys_dev_path')
-    def test_is_netfail_primary_bad_master(self, m_sysdev, m_exists,
-                                           m_standby):
-        devname = self.random_string()
-        driver = self.random_string()  # device not virtio_net
-        master_devname = self.random_string()
-        m_sysdev.return_value = "%s/%s" % (self.random_string(),
-                                           master_devname)
-        m_exists.return_value = True  # has master sysfs attr
-        self.m_device_driver.return_value = 'XXXX'  # master not virtio_net
-        self.assertFalse(net.is_netfail_primary(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    @mock.patch('cloudinit.net.sys_dev_path')
-    def test_is_netfail_primary_no_standby(self, m_sysdev, m_exists,
-                                           m_standby):
-        devname = self.random_string()
-        driver = self.random_string()  # device not virtio_net
-        master_devname = self.random_string()
-        m_sysdev.return_value = "%s/%s" % (self.random_string(),
-                                           master_devname)
-        m_exists.return_value = True  # has master sysfs attr
-        self.m_device_driver.return_value = 'virtio_net'  # master virtio_net
-        m_standby.return_value = False  # master has no standby feature flag
-        self.assertFalse(net.is_netfail_primary(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    def test_is_netfail_standby(self, m_exists, m_standby):
-        devname = self.random_string()
-        driver = 'virtio_net'
-        m_exists.return_value = True  # has master sysfs attr
-        m_standby.return_value = True  # has standby feature flag
-        self.assertTrue(net.is_netfail_standby(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    def test_is_netfail_standby_wrong_driver(self, m_exists, m_standby):
-        devname = self.random_string()
-        driver = self.random_string()
-        self.assertFalse(net.is_netfail_standby(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    def test_is_netfail_standby_no_master(self, m_exists, m_standby):
-        devname = self.random_string()
-        driver = 'virtio_net'
-        m_exists.return_value = False  # has master sysfs attr
-        self.assertFalse(net.is_netfail_standby(devname, driver))
-
-    @mock.patch('cloudinit.net.has_netfail_standby_feature')
-    @mock.patch('cloudinit.net.os.path.exists')
-    def test_is_netfail_standby_no_standby_feature(self, m_exists, m_standby):
-        devname = self.random_string()
-        driver = 'virtio_net'
-        m_exists.return_value = True  # has master sysfs attr
-        m_standby.return_value = False  # has standby feature flag
-        self.assertFalse(net.is_netfail_standby(devname, driver))
-
-    @mock.patch('cloudinit.net.is_netfail_standby')
-    @mock.patch('cloudinit.net.is_netfail_primary')
-    def test_is_netfailover_primary(self, m_primary, m_standby):
-        devname = self.random_string()
-        driver = self.random_string()
-        m_primary.return_value = True
-        m_standby.return_value = False
-        self.assertTrue(net.is_netfailover(devname, driver))
-
-    @mock.patch('cloudinit.net.is_netfail_standby')
-    @mock.patch('cloudinit.net.is_netfail_primary')
-    def test_is_netfailover_standby(self, m_primary, m_standby):
-        devname = self.random_string()
-        driver = self.random_string()
-        m_primary.return_value = False
-        m_standby.return_value = True
-        self.assertTrue(net.is_netfailover(devname, driver))
-
-    @mock.patch('cloudinit.net.is_netfail_standby')
-    @mock.patch('cloudinit.net.is_netfail_primary')
-    def test_is_netfailover_returns_false(self, m_primary, m_standby):
-        devname = self.random_string()
-        driver = self.random_string()
-        m_primary.return_value = False
-        m_standby.return_value = False
-        self.assertFalse(net.is_netfailover(devname, driver))
-
-
-class TestIsIpAddress:
-    """Tests for net.is_ip_address.
-
-    Instead of testing with values we rely on the ipaddress stdlib module to
-    handle all values correctly, so simply test that is_ip_address defers to
-    the ipaddress module correctly.
-    """
-
-    @pytest.mark.parametrize('ip_address_side_effect,expected_return', (
-        (ValueError, False),
-        (lambda _: ipaddress.IPv4Address('192.168.0.1'), True),
-        (lambda _: ipaddress.IPv6Address('2001:db8::'), True),
-    ))
-    def test_is_ip_address(self, ip_address_side_effect, expected_return):
-        with mock.patch('cloudinit.net.ipaddress.ip_address',
-                        side_effect=ip_address_side_effect) as m_ip_address:
-            ret = net.is_ip_address(mock.sentinel.ip_address_in)
-        assert expected_return == ret
-        expected_call = mock.call(mock.sentinel.ip_address_in)
-        assert [expected_call] == m_ip_address.call_args_list
-
-
-class TestIsIpv4Address:
-    """Tests for net.is_ipv4_address.
-
-    Instead of testing with values we rely on the ipaddress stdlib module to
-    handle all values correctly, so simply test that is_ipv4_address defers to
-    the ipaddress module correctly.
-    """
-
-    @pytest.mark.parametrize('ipv4address_mock,expected_return', (
-        (mock.Mock(side_effect=ValueError), False),
-        (mock.Mock(return_value=ipaddress.IPv4Address('192.168.0.1')), True),
-    ))
-    def test_is_ip_address(self, ipv4address_mock, expected_return):
-        with mock.patch('cloudinit.net.ipaddress.IPv4Address',
-                        ipv4address_mock) as m_ipv4address:
-            ret = net.is_ipv4_address(mock.sentinel.ip_address_in)
-        assert expected_return == ret
-        expected_call = mock.call(mock.sentinel.ip_address_in)
-        assert [expected_call] == m_ipv4address.call_args_list
-
-
-# vi: ts=4 expandtab