networking: refactor is_physical from cloudinit.net

cloudinit/distros/networking.py

@@ -1,4 +1,5 @@
 import abc
+import os
 
 from cloudinit import net
 
@@ -79,8 +80,15 @@ def is_bond(self, devname: DeviceName) -> bool:
     def is_bridge(self, devname: DeviceName) -> bool:
         return net.is_bridge(devname)
 
+    @abc.abstractmethod
     def is_physical(self, devname: DeviceName) -> bool:
-        return net.is_physical(devname)
+        """
+        Is ``devname`` a physical network device?
+
+        Examples of non-physical network devices: bonds, bridges, tunnels,
+        loopback devices.
+        """
+        pass
 
     def is_renamed(self, devname: DeviceName) -> bool:
         return net.is_renamed(devname)
@@ -103,7 +111,8 @@ def wait_for_physdevs(
 class BSDNetworking(Networking):
     """Implementation of networking functionality shared across BSDs."""
 
-    pass
+    def is_physical(self, devname: DeviceName) -> bool:
+        raise NotImplementedError()
 
 
 class LinuxNetworking(Networking):
@@ -126,3 +135,6 @@ def is_netfail_primary(self, devname: DeviceName) -> bool:
 
     def is_netfail_standby(self, devname: DeviceName) -> bool:
         return net.is_netfail_standby(devname)
+
+    def is_physical(self, devname: DeviceName) -> bool:
+        return os.path.exists(net.sys_dev_path(devname, "device"))

cloudinit/distros/tests/test_networking.py

@@ -0,0 +1,42 @@
+from unittest import mock
+
+import pytest
+
+from cloudinit.distros.networking import BSDNetworking, LinuxNetworking
+
+
+@pytest.yield_fixture
+def sys_class_net(tmpdir):
+    sys_class_net_path = tmpdir.join("sys/class/net")
+    sys_class_net_path.ensure_dir()
+    with mock.patch(
+        "cloudinit.net.get_sys_class_path",
+        return_value=sys_class_net_path.strpath + "/",
+    ):
+        yield sys_class_net_path
+
+
+class TestBSDNetworkingIsPhysical:
+    def test_raises_notimplementederror(self):
+        with pytest.raises(NotImplementedError):
+            BSDNetworking().is_physical("eth0")
+
+
+class TestLinuxNetworkingIsPhysical:
+    def test_returns_false_by_default(self, sys_class_net):
+        assert not LinuxNetworking().is_physical("eth0")
+
+    def test_returns_false_if_devname_exists_but_not_physical(
+        self, sys_class_net
+    ):
+        devname = "eth0"
+        sys_class_net.join(devname).mkdir()
+        assert not LinuxNetworking().is_physical(devname)
+
+    def test_returns_true_if_device_is_physical(self, sys_class_net):
+        devname = "eth0"
+        device_dir = sys_class_net.join(devname)
+        device_dir.mkdir()
+        device_dir.join("device").write("")
+
+        assert LinuxNetworking().is_physical(devname)

cloudinit/net/__init__.py

@@ -262,10 +262,6 @@ def is_vlan(devname):
     return 'DEVTYPE=vlan' in uevent.splitlines()
 
 
-def is_physical(devname):
-    return os.path.exists(sys_dev_path(devname, "device"))
-
-
 def device_driver(devname):
     """Return the device driver for net device named 'devname'."""
     driver = None

cloudinit/net/tests/test_init.py

@@ -198,12 +198,6 @@ def test_is_vlan(self):
         write_file(os.path.join(self.sysdir, 'eth0', 'uevent'), content)
         self.assertTrue(net.is_vlan('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'))
-
 
 class TestGenerateFallbackConfig(CiTestCase):
 

cloudinit/sources/DataSourceDigitalOcean.py

@@ -58,7 +58,7 @@ def _get_data(self):
 
         ipv4LL_nic = None
         if self.use_ip4LL:
-            ipv4LL_nic = do_helper.assign_ipv4_link_local()
+            ipv4LL_nic = do_helper.assign_ipv4_link_local(self.distro)
 
         md = do_helper.read_metadata(
             self.metadata_address, timeout=self.timeout,

cloudinit/sources/DataSourceOpenNebula.py

@@ -13,6 +13,7 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
 import collections
+import functools
 import os
 import pwd
 import re
@@ -60,10 +61,19 @@ def _get_data(self):
         for cdev in candidates:
             try:
                 if os.path.isdir(self.seed_dir):
-                    results = read_context_disk_dir(cdev, asuser=parseuser)
+                    results = read_context_disk_dir(
+                        cdev, self.distro, asuser=parseuser
+                    )
                 elif cdev.startswith("/dev"):
-                    results = util.mount_cb(cdev, read_context_disk_dir,
-                                            data=parseuser)
+                    # util.mount_cb only handles passing a single argument
+                    # through to the wrapped function, so we have to partially
+                    # apply the function to pass in `distro`.  See LP: #1884979
+                    partially_applied_func = functools.partial(
+                        read_context_disk_dir,
+                        asuser=parseuser,
+                        distro=self.distro,
+                    )
+                    results = util.mount_cb(cdev, partially_applied_func)
             except NonContextDiskDir:
                 continue
             except BrokenContextDiskDir as exc:
@@ -129,10 +139,10 @@ class BrokenContextDiskDir(Exception):
 
 
 class OpenNebulaNetwork(object):
-    def __init__(self, context, system_nics_by_mac=None):
+    def __init__(self, context, distro, system_nics_by_mac=None):
         self.context = context
         if system_nics_by_mac is None:
-            system_nics_by_mac = get_physical_nics_by_mac()
+            system_nics_by_mac = get_physical_nics_by_mac(distro)
         self.ifaces = collections.OrderedDict(
             [k for k in sorted(system_nics_by_mac.items(),
                                key=lambda k: net.natural_sort_key(k[1]))])
@@ -367,7 +377,7 @@ def varprinter(vlist):
     return ret
 
 
-def read_context_disk_dir(source_dir, asuser=None):
+def read_context_disk_dir(source_dir, distro, asuser=None):
     """
     read_context_disk_dir(source_dir):
     read source_dir and return a tuple with metadata dict and user-data
@@ -450,15 +460,17 @@ def read_context_disk_dir(source_dir, asuser=None):
     # http://docs.opennebula.org/5.4/operation/references/template.html#context-section
     ipaddr_keys = [k for k in context if re.match(r'^ETH\d+_IP.*$', k)]
     if ipaddr_keys:
-        onet = OpenNebulaNetwork(context)
+        onet = OpenNebulaNetwork(context, distro)
         results['network-interfaces'] = onet.gen_conf()
 
     return results
 
 
-def get_physical_nics_by_mac():
+def get_physical_nics_by_mac(distro):
     devs = net.get_interfaces_by_mac()
-    return dict([(m, n) for m, n in devs.items() if net.is_physical(n)])
+    return dict(
+        [(m, n) for m, n in devs.items() if distro.networking.is_physical(n)]
+    )
 
 
 # Legacy: Must be present in case we load an old pkl object

cloudinit/sources/helpers/digitalocean.py

@@ -16,15 +16,15 @@
 LOG = logging.getLogger(__name__)
 
 
-def assign_ipv4_link_local(nic=None):
+def assign_ipv4_link_local(distro, nic=None):
     """Bring up NIC using an address using link-local (ip4LL) IPs. On
        DigitalOcean, the link-local domain is per-droplet routed, so there
        is no risk of collisions. However, to be more safe, the ip4LL
        address is random.
     """
 
     if not nic:
-        nic = get_link_local_nic()
+        nic = get_link_local_nic(distro)
         LOG.debug("selected interface '%s' for reading metadata", nic)
 
     if not nic:
@@ -54,8 +54,12 @@ def assign_ipv4_link_local(nic=None):
     return nic
 
 
-def get_link_local_nic():
-    nics = [f for f in cloudnet.get_devicelist() if cloudnet.is_physical(f)]
+def get_link_local_nic(distro):
+    nics = [
+        f
+        for f in cloudnet.get_devicelist()
+        if distro.networking.is_physical(f)
+    ]
     if not nics:
         return None
     return min(nics, key=lambda d: cloudnet.read_sys_net_int(d, 'ifindex'))