# This file is part of cloud-init. See LICENSE file for license information.

import copy

import errno

import httpretty

import mock

import os

import requests

import textwrap

import cloudinit.net as net

from cloudinit.util import ensure_file, write_file, ProcessExecutionError

from cloudinit.tests.helpers import CiTestCase, HttprettyTestCase

from cloudinit import safeyaml as yaml

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_wireless(self):

        """is_wireless is True when /sys/net/devname/wireless exists."""

        self.assertFalse(net.is_wireless('eth0'))

        ensure_file(os.path.join(self.sysdir, 'eth0', 'wireless'))

        self.assertTrue(net.is_wireless('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_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'))

    def test_is_connected_when_physically_connected(self):

        """is_connected is True when /sys/net/devname/iflink reports 2."""

        self.assertFalse(net.is_connected('eth0'))

        write_file(os.path.join(self.sysdir, 'eth0', 'iflink'), "2")

        self.assertTrue(net.is_connected('eth0'))

    def test_is_connected_when_wireless_and_carrier_active(self):

        """is_connected is True if wireless /sys/net/devname/carrier is 1."""

        self.assertFalse(net.is_connected('eth0'))

        ensure_file(os.path.join(self.sysdir, 'eth0', 'wireless'))

        self.assertFalse(net.is_connected('eth0'))

        write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), "1")

        self.assertTrue(net.is_connected('eth0'))

    def test_is_physical(self):

        """is_physical is True when /sys/net/devname/device exists."""

        self.assertFalse(net.is_physical('eth0'))

        ensure_file(os.path.join(self.sysdir, 'eth0', 'device'))

        self.assertTrue(net.is_physical('eth0'))

    def test_is_present(self):

        """is_present is True when /sys/net/devname exists."""

        self.assertFalse(net.is_present('eth0'))

        ensure_file(os.path.join(self.sysdir, 'eth0', 'device'))

        self.assertTrue(net.is_present('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):

    with_logs = True

    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.assertItemsEqual(['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'

        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, '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'))

        interface_names = [interface[0] for interface in net.get_interfaces()]

        self.assertEqual(['eth1', 'eth2'], 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.util.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})