net/cmdline: refactor to allow multiple initramfs network config sources

This refactors read_initramfs_config to support multiple different types
of initramfs network configuration.  It introduces an
InitramfsNetworkConfigSource abstract base class.  There is currently a
single sub-class, KlibcNetworkConfigSource, which contains the logic
which previously was directly within read_initramfs_config.

2 files changed, 181 insertions, 45 deletions

diff --git a/cloudinit/net/cmdline.py b/cloudinit/net/cmdline.py
index 556a10f3..55166ea8 100755
--- a/cloudinit/net/cmdline.py
+++ b/cloudinit/net/cmdline.py
@@ -5,20 +5,95 @@
 #
 # This file is part of cloud-init. See LICENSE file for license information.
 
+import abc
 import base64
 import glob
 import gzip
 import io
 import os
 
-from . import get_devicelist
-from . import read_sys_net_safe
+import six
 
 from cloudinit import util
 
+from . import get_devicelist
+from . import read_sys_net_safe
+
 _OPEN_ISCSI_INTERFACE_FILE = "/run/initramfs/open-iscsi.interface"
 
 
+@six.add_metaclass(abc.ABCMeta)
+class InitramfsNetworkConfigSource(object):
+    """ABC for net config sources that read config written by initramfses"""
+
+    @abc.abstractmethod
+    def is_applicable(self):
+        # type: () -> bool
+        """Is this initramfs config source applicable to the current system?"""
+        pass
+
+    @abc.abstractmethod
+    def render_config(self):
+        # type: () -> dict
+        """Render a v1 network config from the initramfs configuration"""
+        pass
+
+
+class KlibcNetworkConfigSource(InitramfsNetworkConfigSource):
+    """InitramfsNetworkConfigSource for klibc initramfs (i.e. Debian/Ubuntu)
+
+    Has three parameters, but they are intended to make testing simpler, _not_
+    for use in production code.  (This is indicated by the prepended
+    underscores.)
+    """
+
+    def __init__(self, _files=None, _mac_addrs=None, _cmdline=None):
+        self._files = _files
+        self._mac_addrs = _mac_addrs
+        self._cmdline = _cmdline
+
+        # Set defaults here, as they require computation that we don't want to
+        # do at method definition time
+        if self._files is None:
+            self._files = _get_klibc_net_cfg_files()
+        if self._cmdline is None:
+            self._cmdline = util.get_cmdline()
+        if self._mac_addrs is None:
+            self._mac_addrs = {}
+            for k in get_devicelist():
+                mac_addr = read_sys_net_safe(k, 'address')
+                if mac_addr:
+                    self._mac_addrs[k] = mac_addr
+
+    def is_applicable(self):
+        # type: () -> bool
+        """
+        Return whether this system has klibc initramfs network config or not
+
+        Will return True if:
+            (a) klibc files exist in /run, AND
+            (b) either:
+                (i) ip= or ip6= are on the kernel cmdline, OR
+                (ii) an open-iscsi interface file is present in the system
+        """
+        if self._files:
+            if 'ip=' in self._cmdline or 'ip6=' in self._cmdline:
+                return True
+            if os.path.exists(_OPEN_ISCSI_INTERFACE_FILE):
+                # iBft can configure networking without ip=
+                return True
+        return False
+
+    def render_config(self):
+        # type: () -> dict
+        return config_from_klibc_net_cfg(
+            files=self._files, mac_addrs=self._mac_addrs,
+        )
+
+
+_INITRAMFS_CONFIG_SOURCES = [KlibcNetworkConfigSource]
+
+
 def _klibc_to_config_entry(content, mac_addrs=None):
     """Convert a klibc written shell content file to a 'config' entry
     When ip= is seen on the kernel command line in debian initramfs
@@ -137,6 +212,24 @@ def config_from_klibc_net_cfg(files=None, mac_addrs=None):
     return {'config': entries, 'version': 1}
 
 
+def read_initramfs_config():
+    """
+    Return v1 network config for initramfs-configured networking (or None)
+
+    This will consider each _INITRAMFS_CONFIG_SOURCES entry in turn, and return
+    v1 network configuration for the first one that is applicable.  If none are
+    applicable, return None.
+    """
+    for src_cls in _INITRAMFS_CONFIG_SOURCES:
+        cfg_source = src_cls()
+
+        if not cfg_source.is_applicable():
+            continue
+
+        return cfg_source.render_config()
+    return None
+
+
 def _decomp_gzip(blob, strict=True):
     # decompress blob. raise exception if not compressed unless strict=False.
     with io.BytesIO(blob) as iobuf:
@@ -167,36 +260,6 @@ def _b64dgz(b64str, gzipped="try"):
     return _decomp_gzip(blob, strict=gzipped != "try")
 
 
-def _is_initramfs_netconfig(files, cmdline):
-    if files:
-        if 'ip=' in cmdline or 'ip6=' in cmdline:
-            return True
-        if os.path.exists(_OPEN_ISCSI_INTERFACE_FILE):
-            # iBft can configure networking without ip=
-            return True
-    return False
-
-
-def read_initramfs_config(files=None, mac_addrs=None, cmdline=None):
-    if cmdline is None:
-        cmdline = util.get_cmdline()
-
-    if files is None:
-        files = _get_klibc_net_cfg_files()
-
-    if not _is_initramfs_netconfig(files, cmdline):
-        return None
-
-    if mac_addrs is None:
-        mac_addrs = {}
-        for k in get_devicelist():
-            mac_addr = read_sys_net_safe(k, 'address')
-            if mac_addr:
-                mac_addrs[k] = mac_addr
-
-    return config_from_klibc_net_cfg(files=files, mac_addrs=mac_addrs)
-
-
 def read_kernel_cmdline_config(cmdline=None):
     if cmdline is None:
         cmdline = util.get_cmdline()
diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py
index 4f7e4207..e5789924 100644
--- a/tests/unittests/test_net.py
+++ b/tests/unittests/test_net.py
@@ -3591,7 +3591,7 @@ class TestCmdlineConfigParsing(CiTestCase):
         self.assertEqual(found, self.simple_cfg)
 
 
-class TestCmdlineReadInitramfsConfig(FilesystemMockingTestCase):
+class TestCmdlineKlibcNetworkConfigSource(FilesystemMockingTestCase):
     macs = {
         'eth0': '14:02:ec:42:48:00',
         'eno1': '14:02:ec:42:48:01',
@@ -3607,8 +3607,11 @@ class TestCmdlineReadInitramfsConfig(FilesystemMockingTestCase):
         populate_dir(root, content)
         self.reRoot(root)
 
-        found = cmdline.read_initramfs_config(
-            cmdline='foo root=/root/bar', mac_addrs=self.macs)
+        src = cmdline.KlibcNetworkConfigSource(
+            _cmdline='foo root=/root/bar', _mac_addrs=self.macs,
+        )
+        self.assertTrue(src.is_applicable())
+        found = src.render_config()
         self.assertEqual(found['version'], 1)
         self.assertEqual(found['config'], [exp1])
 
@@ -3621,8 +3624,11 @@ class TestCmdlineReadInitramfsConfig(FilesystemMockingTestCase):
         populate_dir(root, content)
         self.reRoot(root)
 
-        found = cmdline.read_initramfs_config(
-            cmdline='foo ip=dhcp', mac_addrs=self.macs)
+        src = cmdline.KlibcNetworkConfigSource(
+            _cmdline='foo ip=dhcp', _mac_addrs=self.macs,
+        )
+        self.assertTrue(src.is_applicable())
+        found = src.render_config()
         self.assertEqual(found['version'], 1)
         self.assertEqual(found['config'], [exp1])
 
@@ -3632,9 +3638,11 @@ class TestCmdlineReadInitramfsConfig(FilesystemMockingTestCase):
         populate_dir(root, content)
         self.reRoot(root)
 
-        found = cmdline.read_initramfs_config(
-            cmdline='foo ip6=dhcp root=/dev/sda',
-            mac_addrs=self.macs)
+        src = cmdline.KlibcNetworkConfigSource(
+            _cmdline='foo ip6=dhcp root=/dev/sda', _mac_addrs=self.macs,
+        )
+        self.assertTrue(src.is_applicable())
+        found = src.render_config()
         self.assertEqual(
             found,
             {'version': 1, 'config': [
@@ -3648,9 +3656,10 @@ class TestCmdlineReadInitramfsConfig(FilesystemMockingTestCase):
         # if there is no ip= or ip6= on cmdline, return value should be None
         content = {'net6-eno1.conf': DHCP6_CONTENT_1}
         files = sorted(populate_dir(self.tmp_dir(), content))
-        found = cmdline.read_initramfs_config(
-            files=files, cmdline='foo root=/dev/sda', mac_addrs=self.macs)
-        self.assertIsNone(found)
+        src = cmdline.KlibcNetworkConfigSource(
+            _files=files, _cmdline='foo root=/dev/sda', _mac_addrs=self.macs,
+        )
+        self.assertFalse(src.is_applicable())
 
     def test_with_both_ip_ip6(self):
         content = {
@@ -3667,13 +3676,77 @@ class TestCmdlineReadInitramfsConfig(FilesystemMockingTestCase):
         populate_dir(root, content)
         self.reRoot(root)
 
-        found = cmdline.read_initramfs_config(
-            cmdline='foo ip=dhcp ip6=dhcp', mac_addrs=self.macs)
+        src = cmdline.KlibcNetworkConfigSource(
+            _cmdline='foo ip=dhcp ip6=dhcp', _mac_addrs=self.macs,
+        )
 
+        self.assertTrue(src.is_applicable())
+        found = src.render_config()
         self.assertEqual(found['version'], 1)
         self.assertEqual(found['config'], expected)
 
 
+class TestReadInitramfsConfig(CiTestCase):
+
+    def _config_source_cls_mock(self, is_applicable, render_config=None):
+        return lambda: mock.Mock(
+            is_applicable=lambda: is_applicable,
+            render_config=lambda: render_config,
+        )
+
+    def test_no_sources(self):
+        with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES', []):
+            self.assertIsNone(cmdline.read_initramfs_config())
+
+    def test_no_applicable_sources(self):
+        sources = [
+            self._config_source_cls_mock(is_applicable=False),
+            self._config_source_cls_mock(is_applicable=False),
+            self._config_source_cls_mock(is_applicable=False),
+        ]
+        with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES',
+                        sources):
+            self.assertIsNone(cmdline.read_initramfs_config())
+
+    def test_one_applicable_source(self):
+        expected_config = object()
+        sources = [
+            self._config_source_cls_mock(
+                is_applicable=True, render_config=expected_config,
+            ),
+        ]
+        with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES',
+                        sources):
+            self.assertEqual(expected_config, cmdline.read_initramfs_config())
+
+    def test_one_applicable_source_after_inapplicable_sources(self):
+        expected_config = object()
+        sources = [
+            self._config_source_cls_mock(is_applicable=False),
+            self._config_source_cls_mock(is_applicable=False),
+            self._config_source_cls_mock(
+                is_applicable=True, render_config=expected_config,
+            ),
+        ]
+        with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES',
+                        sources):
+            self.assertEqual(expected_config, cmdline.read_initramfs_config())
+
+    def test_first_applicable_source_is_used(self):
+        first_config, second_config = object(), object()
+        sources = [
+            self._config_source_cls_mock(
+                is_applicable=True, render_config=first_config,
+            ),
+            self._config_source_cls_mock(
+                is_applicable=True, render_config=second_config,
+            ),
+        ]
+        with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES',
+                        sources):
+            self.assertEqual(first_config, cmdline.read_initramfs_config())
+
+
 class TestNetplanRoundTrip(CiTestCase):
     def _render_and_read(self, network_config=None, state=None,
                          netplan_path=None, target=None):