summaryrefslogtreecommitdiff
path: root/cloudinit
diff options
context:
space:
mode:
Diffstat (limited to 'cloudinit')
-rw-r--r--cloudinit/distros/networking.py16
-rw-r--r--cloudinit/distros/tests/test_networking.py42
-rw-r--r--cloudinit/net/__init__.py4
-rw-r--r--cloudinit/net/tests/test_init.py6
-rw-r--r--cloudinit/sources/DataSourceDigitalOcean.py2
-rw-r--r--cloudinit/sources/DataSourceOpenNebula.py30
-rw-r--r--cloudinit/sources/helpers/digitalocean.py12
7 files changed, 86 insertions, 26 deletions
diff --git a/cloudinit/distros/networking.py b/cloudinit/distros/networking.py
index eecdccc6..e421a2ce 100644
--- a/cloudinit/distros/networking.py
+++ b/cloudinit/distros/networking.py
@@ -1,4 +1,5 @@
import abc
+import os
from cloudinit import net
@@ -79,8 +80,15 @@ class Networking(metaclass=abc.ABCMeta):
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 @@ class Networking(metaclass=abc.ABCMeta):
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 @@ class LinuxNetworking(Networking):
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"))
diff --git a/cloudinit/distros/tests/test_networking.py b/cloudinit/distros/tests/test_networking.py
new file mode 100644
index 00000000..2acb12f4
--- /dev/null
+++ b/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)
diff --git a/cloudinit/net/__init__.py b/cloudinit/net/__init__.py
index fd5a1b3e..9d8c7ba9 100644
--- a/cloudinit/net/__init__.py
+++ b/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
diff --git a/cloudinit/net/tests/test_init.py b/cloudinit/net/tests/test_init.py
index e36d4387..eb458c39 100644
--- a/cloudinit/net/tests/test_init.py
+++ b/cloudinit/net/tests/test_init.py
@@ -198,12 +198,6 @@ class TestReadSysNet(CiTestCase):
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):
diff --git a/cloudinit/sources/DataSourceDigitalOcean.py b/cloudinit/sources/DataSourceDigitalOcean.py
index e0ef665e..5040ce5b 100644
--- a/cloudinit/sources/DataSourceDigitalOcean.py
+++ b/cloudinit/sources/DataSourceDigitalOcean.py
@@ -58,7 +58,7 @@ class DataSourceDigitalOcean(sources.DataSource):
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,
diff --git a/cloudinit/sources/DataSourceOpenNebula.py b/cloudinit/sources/DataSourceOpenNebula.py
index c7fdc2a5..12b1f94f 100644
--- a/cloudinit/sources/DataSourceOpenNebula.py
+++ b/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 @@ class DataSourceOpenNebula(sources.DataSource):
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 parse_shell_config(content, keylist=None, bash=None, asuser=None,
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
diff --git a/cloudinit/sources/helpers/digitalocean.py b/cloudinit/sources/helpers/digitalocean.py
index f5bbe46a..b545c4d6 100644
--- a/cloudinit/sources/helpers/digitalocean.py
+++ b/cloudinit/sources/helpers/digitalocean.py
@@ -16,7 +16,7 @@ NIC_MAP = {'public': 'eth0', 'private': 'eth1'}
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
@@ -24,7 +24,7 @@ def assign_ipv4_link_local(nic=None):
"""
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'))
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-13 05:12:15 +0000