cloud-init-dev team mailing list archive

[Scott Moser <smoser@xxxxxxxxxx>]

why did you have to add mock to tox's pylint entries ?
also some inline.
this does look really nice though.
thank you.


Diff comments:

> diff --git a/cloudinit/net/__init__.py b/cloudinit/net/__init__.py
> index d1740e5..5a4a232 100644
> --- a/cloudinit/net/__init__.py
> +++ b/cloudinit/net/__init__.py
> @@ -557,6 +553,27 @@ def get_interfaces():
>      return ret
>  
>  
> +def initialize_network_device(interface, ip_address, netmask, broadcast,
> +                              router=None):
> +    if not all([interface, ip_address, netmask, broadcast]):
> +        raise ValueError(
> +            "Cannot init network dev {0} with ip{1}/{2} and bcast {3}".format(
> +                interface, ip_address, netmask, broadcast))
> +    prefix = mask_to_net_prefix(netmask)

i dont like 'netmask'... if we can avoid that, or at least support 'prefix' as input that'd be nice.
raise value error if both netmask and prefix are provided.

broadcast is not needed, right ? we can derive it from address and netmaks (or prefix)

> +
> +    util.subp([
> +        'ip', '-family', 'inet', 'addr', 'add', '%s/%s' % (ip_address, prefix),
> +        'broadcast', broadcast, 'dev', interface], capture=True)
> +    util.subp(
> +        ['ip', '-family', 'inet', 'link', 'set', 'dev', interface, 'up'],
> +        capture=True)
> +    if router:
> +        util.subp(
> +            ['ip', '-4', 'route', 'add', 'default', 'via', router, 'dev',
> +             interface],
> +            capture=True)

OK, so maybe i like context handlers too much. but this seems so nice:

with ephemeral_network_config('eth0', '192.168.1.2/16'):
   x = read_metadata()

then the contect handler could also know if it had brought the nic up (link set up) or not and bring down if it idd.

> +
> +
>  class RendererNotFoundError(RuntimeError):
>      pass
>  
> diff --git a/cloudinit/net/tests/test_init.py b/cloudinit/net/tests/test_init.py
> new file mode 100644
> index 0000000..8dc06b3
> --- /dev/null
> +++ b/cloudinit/net/tests/test_init.py
> @@ -0,0 +1,406 @@
> +# This file is part of cloud-init. See LICENSE file for license information.
> +
> +import copy
> +import mock
> +import os
> +
> +import cloudinit.net as net
> +from cloudinit.util import ensure_file, write_file
> +from tests.unittests.helpers import CiTestCase
> +
> +
> +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()
> +        self.sysdir = self.tmp_dir()
> +        self.net = net
> +        self.net.SYS_CLASS_NET = self.sysdir + '/'

dont you need to put this back on tearDown ?

> +
> +    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 and 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_translation_errors_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_returns_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_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()
> +        self.sysdir = self.tmp_dir()
> +        self.net = net
> +        self.net.SYS_CLASS_NET = self.sysdir + '/'
> +
> +    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 = {
> +            'config': [{'type': 'physical', 'mac_address': mac,
> +                        'name': 'eth1', 'subnets': [{'type': 'dhcp'}]}],
> +            'version': 1}
> +        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 = {
> +            'config': [{'type': 'physical', 'mac_address': mac,
> +                        'name': 'eth0', 'subnets': [{'type': 'dhcp'}]}],
> +            'version': 1}
> +        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 = {
> +            'config': [{'type': 'physical', 'mac_address': mac,
> +                        'name': 'eth0', 'subnets': [{'type': 'dhcp'}]}],
> +            'version': 1}
> +        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())
> +
> +
> +class TestGetDeviceList(CiTestCase):
> +
> +    def setUp(self):
> +        super(TestGetDeviceList, self).setUp()
> +        self.sysdir = self.tmp_dir()
> +        self.net = net
> +        self.net.SYS_CLASS_NET = self.sysdir + '/'
> +
> +    def test_get_device_list_empty_without_sys_net(self):
> +        """get_device_list returns empty list when missing SYS_CLASS_NET."""
> +        self.net.SYS_CLASS_NET = 'idontexist'
> +        self.assertEqual([], net.get_interfaces())
> +
> +    def test_get_interfaces_empty_list_without_sys_net(self):
> +        """get_interfaces returns an empty list when missing SYS_CLASS_NET."""
> +        self.net.SYS_CLASS_NET = 'idontexist'
> +        self.assertEqual([], net.get_interfaces())
> +
> +
> +class TestGetInterfaceMAC(CiTestCase):
> +
> +    def setUp(self):
> +        super(TestGetInterfaceMAC, self).setUp()
> +        self.sysdir = self.tmp_dir()
> +        self.net = net
> +        self.net.SYS_CLASS_NET = self.sysdir + '/'
> +
> +    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.net.SYS_CLASS_NET, '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_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())
> +
> +
> +class TestInterfaceHasOwnMAC(CiTestCase):
> +
> +    def setUp(self):
> +        super(TestInterfaceHasOwnMAC, self).setUp()
> +        self.sysdir = self.tmp_dir()
> +        self.net = net
> +        self.net.SYS_CLASS_NET = self.sysdir + '/'
> +
> +    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 TestInitializeNetworkDevice(CiTestCase):
> +
> +    def test_initialize_network_devices_errors_on_missing_params(self, m_subp):
> +        """All parameters for initialize_network_device cannot be None."""
> +        required_params = {
> +            'interface': 'eth0', 'ip_address': '192.168.2.2',
> +            'netmask': '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.initialize_network_device(**params)
> +            error = context_manager.exception
> +            self.assertIn('Cannot init network dev', str(error))
> +            self.assertEqual(0, m_subp.call_count)
> +
> +    def test_initialize_network_device_without_router(self, m_subp):
> +        """When router is None, initialize_network_device adds no gateway."""
> +        params = {
> +            'interface': 'eth0', 'ip_address': '192.168.2.2',
> +            'netmask': '255.255.255.0', 'broadcast': '192.168.2.255'}
> +        net.initialize_network_device(**params)
> +        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),
> +            mock.call(
> +                ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
> +                capture=True)])
> +
> +    def test_initialize_network_device_with_router(self, m_subp):
> +        """When router is None, initialize_network_device adds a gateway."""
> +        params = {
> +            'interface': 'eth0', 'ip_address': '192.168.2.2',
> +            'netmask': '255.255.255.0', 'broadcast': '192.168.2.255',
> +            'router': '192.168.2.1'}
> +        net.initialize_network_device(**params)
> +        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),
> +            mock.call(
> +                ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
> +                capture=True),
> +            mock.call(
> +                ['ip', '-4', 'route', 'add', 'default', 'via',
> +                 params['router'], 'dev', 'eth0'], capture=True)])


-- 
https://code.launchpad.net/~chad.smith/cloud-init/+git/cloud-init/+merge/327827
Your team cloud-init commiters is requested to review the proposed merge of ~chad.smith/cloud-init:unittests-in-cloudinit-package into cloud-init:master.