Merge ~chad.smith/cloud-init:ubuntu/disco into cloud-init:ubuntu/disco

[Chad Smith]

new-upstream snapshot to get exoscale fixes before we complete current -proposed SRU

[Server Team CI bot (server-team-bot)]

FAILED: Continuous integration, rev:f0141fe08a30d606035109975696bb1bd6c1c54d
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1088/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    FAILED: Ubuntu LTS: Build

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1088//rebuild

review: Needs Fixing (continuous-integration)

[Ryan Harper (raharper)]

This matches my output.

review: Approve

[Server Team CI bot (server-team-bot)]

FAILED: Continuous integration, rev:f0141fe08a30d606035109975696bb1bd6c1c54d
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1091/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    FAILED: Ubuntu LTS: Build

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1091//rebuild

review: Needs Fixing (continuous-integration)

[Server Team CI bot (server-team-bot)]

FAILED: Continuous integration, rev:f0141fe08a30d606035109975696bb1bd6c1c54d
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1092/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    SUCCESS: Ubuntu LTS: Build
    FAILED: Ubuntu LTS: Integration

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1092//rebuild

review: Needs Fixing (continuous-integration)

[Server Team CI bot (server-team-bot)]

PASSED: Continuous integration, rev:f0141fe08a30d606035109975696bb1bd6c1c54d
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1093/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    SUCCESS: Ubuntu LTS: Build
    SUCCESS: Ubuntu LTS: Integration
    IN_PROGRESS: Declarative: Post Actions

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1093//rebuild

review: Approve (continuous-integration)

[Server Team CI bot (server-team-bot)]

PASSED: Continuous integration, rev:f0141fe08a30d606035109975696bb1bd6c1c54d
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1095/
Executed test runs:
    SUCCESS: Checkout
    SUCCESS: Unit & Style Tests
    SUCCESS: Ubuntu LTS: Build
    SUCCESS: Ubuntu LTS: Integration
    IN_PROGRESS: Declarative: Post Actions

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/cloud-init-ci/1095//rebuild

review: Approve (continuous-integration)

diff --git a/cloudinit/net/cmdline.py b/cloudinit/net/cmdline.py
index 556a10f..55166ea 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"
 
 
[e-mail]_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/cloudinit/sources/DataSourceExoscale.py b/cloudinit/sources/DataSourceExoscale.py
index 52e7f6f..fdfb4ed 100644
--- a/cloudinit/sources/DataSourceExoscale.py
+++ b/cloudinit/sources/DataSourceExoscale.py
@@ -6,6 +6,7 @@
 from cloudinit import ec2_utils as ec2
 from cloudinit import log as logging
 from cloudinit import sources
+from cloudinit import helpers
 from cloudinit import url_helper
 from cloudinit import util
 
@@ -20,13 +21,6 @@ URL_RETRIES = 6
 
 EXOSCALE_DMI_NAME = "Exoscale"
 
-BUILTIN_DS_CONFIG = {
-    # We run the set password config module on every boot in order to enable
-    # resetting the instance's password via the exoscale console (and a
-    # subsequent instance reboot).
-    'cloud_config_modules': [["set-passwords", "always"]]
-}
-
 
 class DataSourceExoscale(sources.DataSource):
 
@@ -42,8 +36,22 @@ class DataSourceExoscale(sources.DataSource):
             self.ds_cfg.get('password_server_port', PASSWORD_SERVER_PORT))
         self.url_timeout = self.ds_cfg.get('timeout', URL_TIMEOUT)
         self.url_retries = self.ds_cfg.get('retries', URL_RETRIES)
-
-        self.extra_config = BUILTIN_DS_CONFIG
+        self.extra_config = {}
+
+    def activate(self, cfg, is_new_instance):
+        """Adjust set-passwords module to run 'always' during each boot"""
+        # We run the set password config module on every boot in order to
+        # enable resetting the instance's password via the exoscale console
+        # (and a subsequent instance reboot).
+        # Exoscale password server only provides set-passwords user-data if
+        # a user has triggered a password reset. So calling that password
+        # service generally results in no additional cloud-config.
+        # TODO(Create util functions for overriding merged sys_cfg module freq)
+        mod = 'set_passwords'
+        sem_path = self.paths.get_ipath_cur('sem')
+        sem_helper = helpers.FileSemaphores(sem_path)
+        if sem_helper.clear('config_' + mod, None):
+            LOG.debug('Overriding module set-passwords with frequency always')
 
     def wait_for_metadata_service(self):
         """Wait for the metadata service to be reachable."""
diff --git a/cloudinit/sources/DataSourceOracle.py b/cloudinit/sources/DataSourceOracle.py
index 6e73f56..1cb0636 100644
--- a/cloudinit/sources/DataSourceOracle.py
+++ b/cloudinit/sources/DataSourceOracle.py
@@ -51,8 +51,8 @@ def _add_network_config_from_opc_imds(network_config):
     include the secondary VNICs.
 
     :param network_config:
-        A v1 network config dict with the primary NIC already configured.  This
-        dict will be mutated.
+        A v1 or v2 network config dict with the primary NIC already configured.
+        This dict will be mutated.
 
     :raises:
         Exceptions are not handled within this function.  Likely exceptions are
@@ -88,20 +88,24 @@ def _add_network_config_from_opc_imds(network_config):
             LOG.debug('Interface with MAC %s not found; skipping', mac_address)
             continue
         name = interfaces_by_mac[mac_address]
-        subnet = {
-            'type': 'static',
-            'address': vnic_dict['privateIp'],
-            'netmask': vnic_dict['subnetCidrBlock'].split('/')[1],
-            'gateway': vnic_dict['virtualRouterIp'],
-            'control': 'manual',
-        }
-        network_config['config'].append({
-            'name': name,
-            'type': 'physical',
-            'mac_address': mac_address,
-            'mtu': MTU,
-            'subnets': [subnet],
-        })
+
+        if network_config['version'] == 1:
+            subnet = {
+                'type': 'static',
+                'address': vnic_dict['privateIp'],
+            }
+            network_config['config'].append({
+                'name': name,
+                'type': 'physical',
+                'mac_address': mac_address,
+                'mtu': MTU,
+                'subnets': [subnet],
+            })
+        elif network_config['version'] == 2:
+            network_config['ethernets'][name] = {
+                'addresses': [vnic_dict['privateIp']],
+                'mtu': MTU, 'dhcp4': False, 'dhcp6': False,
+                'match': {'macaddress': mac_address}}
 
 
 class DataSourceOracle(sources.DataSource):
diff --git a/cloudinit/sources/tests/test_oracle.py b/cloudinit/sources/tests/test_oracle.py
index 3ddf7df..2a70bbc 100644
--- a/cloudinit/sources/tests/test_oracle.py
+++ b/cloudinit/sources/tests/test_oracle.py
@@ -526,6 +526,18 @@ class TestNetworkConfigFromOpcImds(test_helpers.CiTestCase):
             'Interface with MAC 00:00:17:02:2b:b1 not found; skipping',
             self.logs.getvalue())
 
+    def test_missing_mac_skipped_v2(self):
+        self.m_readurl.return_value = OPC_VM_SECONDARY_VNIC_RESPONSE
+        self.m_get_interfaces_by_mac.return_value = {}
+
+        network_config = {'version': 2, 'ethernets': {'primary': {'nic': {}}}}
+        oracle._add_network_config_from_opc_imds(network_config)
+
+        self.assertEqual(1, len(network_config['ethernets']))
+        self.assertIn(
+            'Interface with MAC 00:00:17:02:2b:b1 not found; skipping',
+            self.logs.getvalue())
+
     def test_secondary_nic(self):
         self.m_readurl.return_value = OPC_VM_SECONDARY_VNIC_RESPONSE
         mac_addr, nic_name = '00:00:17:02:2b:b1', 'ens3'
@@ -549,8 +561,29 @@ class TestNetworkConfigFromOpcImds(test_helpers.CiTestCase):
         subnet_cfg = secondary_nic_cfg['subnets'][0]
         # These values are hard-coded in OPC_VM_SECONDARY_VNIC_RESPONSE
         self.assertEqual('10.0.0.231', subnet_cfg['address'])
-        self.assertEqual('24', subnet_cfg['netmask'])
-        self.assertEqual('10.0.0.1', subnet_cfg['gateway'])
-        self.assertEqual('manual', subnet_cfg['control'])
+
+    def test_secondary_nic_v2(self):
+        self.m_readurl.return_value = OPC_VM_SECONDARY_VNIC_RESPONSE
+        mac_addr, nic_name = '00:00:17:02:2b:b1', 'ens3'
+        self.m_get_interfaces_by_mac.return_value = {
+            mac_addr: nic_name,
+        }
+
+        network_config = {'version': 2, 'ethernets': {'primary': {'nic': {}}}}
+        oracle._add_network_config_from_opc_imds(network_config)
+
+        # The input is mutated
+        self.assertEqual(2, len(network_config['ethernets']))
+
+        secondary_nic_cfg = network_config['ethernets']['ens3']
+        self.assertFalse(secondary_nic_cfg['dhcp4'])
+        self.assertFalse(secondary_nic_cfg['dhcp6'])
+        self.assertEqual(mac_addr, secondary_nic_cfg['match']['macaddress'])
+        self.assertEqual(9000, secondary_nic_cfg['mtu'])
+
+        self.assertEqual(1, len(secondary_nic_cfg['addresses']))
+        # These values are hard-coded in OPC_VM_SECONDARY_VNIC_RESPONSE
+        self.assertEqual('10.0.0.231', secondary_nic_cfg['addresses'][0])
+
 
 # vi: ts=4 expandtab
diff --git a/debian/changelog b/debian/changelog
index 0ac84a4..ba9606f 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,12 @@
+cloud-init (19.2-24-ge7881d5c-0ubuntu1~19.04.1) disco; urgency=medium
+
+  * New upstream snapshot. (LP: #1841099)
+    - Oracle: Render secondary vnic IP and MTU values only
+    - exoscale: fix sysconfig cloud_config_modules overrides
+    - net/cmdline: refactor to allow multiple initramfs network config sources
+
+ -- Chad Smith <[e-mail]>  Wed, 28 Aug 2019 15:02:21 -0600
+
 cloud-init (19.2-21-ge6383719-0ubuntu1~19.04.1) disco; urgency=medium
 
   * debian/cloud-init.templates: enable Exoscale cloud.
diff --git a/tests/unittests/test_datasource/test_exoscale.py b/tests/unittests/test_datasource/test_exoscale.py
index 350c330..f006119 100644
--- a/tests/unittests/test_datasource/test_exoscale.py
+++ b/tests/unittests/test_datasource/test_exoscale.py
@@ -11,8 +11,10 @@ from cloudinit.sources.DataSourceExoscale import (
     PASSWORD_SERVER_PORT,
     read_metadata)
 from cloudinit.tests.helpers import HttprettyTestCase, mock
+from cloudinit import util
 
 import httpretty
+import os
 import requests
 
 
@@ -63,6 +65,18 @@ class TestDatasourceExoscale(HttprettyTestCase):
         password = get_password()
         self.assertEqual(expected_password, password)
 
+    def test_activate_removes_set_passwords_semaphore(self):
+        """Allow set_passwords to run every boot by removing the semaphore."""
+        path = helpers.Paths({'cloud_dir': self.tmp})
+        sem_dir = self.tmp_path('instance/sem', dir=self.tmp)
+        util.ensure_dir(sem_dir)
+        sem_file = os.path.join(sem_dir, 'config_set_passwords')
+        with open(sem_file, 'w') as stream:
+            stream.write('')
+        ds = DataSourceExoscale({}, None, path)
+        ds.activate(None, None)
+        self.assertFalse(os.path.exists(sem_file))
+
     def test_get_data(self):
         """The datasource conforms to expected behavior when supplied
         full test data."""
@@ -95,8 +109,6 @@ class TestDatasourceExoscale(HttprettyTestCase):
         self.assertEqual(ds.get_config_obj(),
                          {'ssh_pwauth': True,
                           'password': expected_password,
-                          'cloud_config_modules': [
-                              ["set-passwords", "always"]],
                           'chpasswd': {
                               'expire': False,
                           }})
@@ -130,9 +142,7 @@ class TestDatasourceExoscale(HttprettyTestCase):
         self.assertEqual(ds.userdata_raw.decode("utf-8"), "#cloud-config")
         self.assertEqual(ds.metadata, {"instance-id": expected_id,
                                        "local-hostname": expected_hostname})
-        self.assertEqual(ds.get_config_obj(),
-                         {'cloud_config_modules': [
-                             ["set-passwords", "always"]]})
+        self.assertEqual(ds.get_config_obj(), {})
 
     def test_get_data_no_password(self):
         """The datasource conforms to expected behavior when no password is
@@ -163,9 +173,7 @@ class TestDatasourceExoscale(HttprettyTestCase):
         self.assertEqual(ds.userdata_raw.decode("utf-8"), "#cloud-config")
         self.assertEqual(ds.metadata, {"instance-id": expected_id,
                                        "local-hostname": expected_hostname})
-        self.assertEqual(ds.get_config_obj(),
-                         {'cloud_config_modules': [
-                             ["set-passwords", "always"]]})
+        self.assertEqual(ds.get_config_obj(), {})
 
     @mock.patch('cloudinit.sources.DataSourceExoscale.get_password')
     def test_read_metadata_when_password_server_unreachable(self, m_password):
diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py
index 4f7e420..e578992 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):