diff --git a/cloudinit/distros/__init__.py b/cloudinit/distros/__init__.py
index c857a8737df0..462161905d7b 100644
--- a/cloudinit/distros/__init__.py
+++ b/cloudinit/distros/__init__.py
@@ -147,7 +147,6 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta):
     resolve_conf_fn = "/etc/resolv.conf"
 
     osfamily: str
-    dhcp_client_priority = [dhcp.IscDhclient, dhcp.Dhcpcd, dhcp.Udhcpc]
     # Directory where the distro stores their DHCP leases.
     # The children classes should override this with their dhcp leases
     # directory
@@ -162,14 +161,12 @@ def __init__(self, name, cfg, paths):
         self._cfg = cfg
         self.name = name
         self.networking: Networking = self.networking_cls()
-        self.dhcp_client_priority = [
-            dhcp.IscDhclient,
-            dhcp.Dhcpcd,
-            dhcp.Udhcpc,
-        ]
+        self.dhcp_client_priority = dhcp.ALL_DHCP_CLIENTS
         self.net_ops = iproute2.Iproute2
         self._runner = helpers.Runners(paths)
         self.package_managers: List[PackageManager] = []
+        self._dhcp_client = None
+        self._fallback_interface = None
 
     def _unpickle(self, ci_pkl_version: int) -> None:
         """Perform deserialization fixes for Distro."""
@@ -182,6 +179,10 @@ def _unpickle(self, ci_pkl_version: int) -> None:
             # either because it isn't present at all, or because it will be
             # missing expected instance state otherwise.
             self.networking = self.networking_cls()
+        if not hasattr(self, "_dhcp_client"):
+            self._dhcp_client = None
+        if not hasattr(self, "_fallback_interface"):
+            self._fallback_interface = None
 
     def _validate_entry(self, entry):
         if isinstance(entry, str):
@@ -274,6 +275,66 @@ def _write_network(self, settings):
             "_write_network_config needs implementation.\n" % self.name
         )
 
+    @property
+    def dhcp_client(self) -> dhcp.DhcpClient:
+        """access the distro's preferred dhcp client
+
+        if no client has been selected yet select one - uses
+        self.dhcp_client_priority, which may be overriden in each distro's
+        object to eliminate checking for clients which will not be provided
+        by the distro
+        """
+        if self._dhcp_client:
+            return self._dhcp_client
+
+        # no client has been selected yet, so pick one
+        #
+        # set the default priority list to the distro-defined priority list
+        dhcp_client_priority = self.dhcp_client_priority
+
+        # if the configuration includes a network.dhcp_client_priority list
+        # then attempt to use it
+        config_priority = util.get_cfg_by_path(
+            self._cfg, ("network", "dhcp_client_priority"), []
+        )
+
+        if config_priority:
+            # user or image builder configured a custom dhcp client priority
+            # list
+            found_clients = []
+            LOG.debug(
+                "Using configured dhcp client priority list: %s",
+                config_priority,
+            )
+            for client_configured in config_priority:
+                for client_class in dhcp.ALL_DHCP_CLIENTS:
+                    if client_configured == client_class.client_name:
+                        found_clients.append(client_class)
+                        break
+                else:
+                    LOG.warning(
+                        "Configured dhcp client %s is not supported, skipping",
+                        client_configured,
+                    )
+            # If dhcp_client_priority is defined in the configuration, but none
+            # of the defined clients are supported by cloud-init, then we don't
+            # override the distro default. If at least one client in the
+            # configured list exists, then we use that for our list of clients
+            # to check.
+            if found_clients:
+                dhcp_client_priority = found_clients
+
+        # iterate through our priority list and use the first client that is
+        # installed on the system
+        for client in dhcp_client_priority:
+            try:
+                self._dhcp_client = client()
+                LOG.debug("DHCP client selected: %s", client.client_name)
+                return self._dhcp_client
+            except (dhcp.NoDHCPLeaseMissingDhclientError,):
+                LOG.debug("DHCP client not found: %s", client.client_name)
+        raise dhcp.NoDHCPLeaseMissingDhclientError()
+
     @property
     def network_activator(self) -> Optional[Type[activators.NetworkActivator]]:
         """Return the configured network activator for this environment."""
@@ -1248,6 +1309,22 @@ def build_dhclient_cmd(
             "/bin/true",
         ] + (["-cf", config_file, interface] if config_file else [interface])
 
+    @property
+    def fallback_interface(self):
+        """Determine the network interface used during local network config."""
+        if self._fallback_interface is None:
+            self._fallback_interface = net.find_fallback_nic()
+            if not self._fallback_interface:
+                LOG.warning(
+                    "Did not find a fallback interface on distro: %s.",
+                    self.name,
+                )
+        return self._fallback_interface
+
+    @fallback_interface.setter
+    def fallback_interface(self, value):
+        self._fallback_interface = value
+
 
 def _apply_hostname_transformations_to_url(url: str, transformations: list):
     """
diff --git a/cloudinit/net/dhcp.py b/cloudinit/net/dhcp.py
index a5857b1b3bbd..9aead42edaea 100644
--- a/cloudinit/net/dhcp.py
+++ b/cloudinit/net/dhcp.py
@@ -13,14 +13,12 @@
 import time
 from contextlib import suppress
 from io import StringIO
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
 import configobj
 
 from cloudinit import subp, temp_utils, util
 from cloudinit.net import (
-    find_fallback_nic,
-    get_devicelist,
     get_ib_interface_hwaddr,
     get_interface_mac,
     is_ib_interface,
@@ -85,28 +83,6 @@ class NoDHCPLeaseMissingDhclientError(NoDHCPLeaseError):
     """Raised when unable to find dhclient."""
 
 
-class NoDHCPLeaseMissingUdhcpcError(NoDHCPLeaseError):
-    """Raised when unable to find udhcpc client."""
-
-
-def select_dhcp_client(distro):
-    """distros set priority list, select based on this order which to use
-
-    If the priority dhcp client isn't found, fall back to lower in list.
-    """
-    for client in distro.dhcp_client_priority:
-        try:
-            dhcp_client = client()
-            LOG.debug("DHCP client selected: %s", client.client_name)
-            return dhcp_client
-        except (
-            NoDHCPLeaseMissingDhclientError,
-            NoDHCPLeaseMissingUdhcpcError,
-        ):
-            LOG.warning("DHCP client not found: %s", client.client_name)
-    raise NoDHCPLeaseMissingDhclientError()
-
-
 def maybe_perform_dhcp_discovery(distro, nic=None, dhcp_log_func=None):
     """Perform dhcp discovery if nic valid and dhclient command exists.
 
@@ -120,18 +96,8 @@ def maybe_perform_dhcp_discovery(distro, nic=None, dhcp_log_func=None):
         from the dhclient discovery if run, otherwise an empty list is
         returned.
     """
-    if nic is None:
-        nic = find_fallback_nic()
-        if nic is None:
-            LOG.debug("Skip dhcp_discovery: Unable to find fallback nic.")
-            raise NoDHCPLeaseInterfaceError()
-    elif nic not in get_devicelist():
-        LOG.debug(
-            "Skip dhcp_discovery: nic %s not found in get_devicelist.", nic
-        )
-        raise NoDHCPLeaseInterfaceError()
-    client = select_dhcp_client(distro)
-    return client.dhcp_discovery(nic, dhcp_log_func, distro)
+    interface = nic or distro.fallback_interface
+    return distro.dhcp_client.dhcp_discovery(interface, dhcp_log_func, distro)
 
 
 def networkd_parse_lease(content):
@@ -176,6 +142,12 @@ def networkd_get_option_from_leases(keyname, leases_d=None):
 
 class DhcpClient(abc.ABC):
     client_name = ""
+    timeout = 10
+
+    def __init__(self):
+        self.dhcp_client_path = subp.which(self.client_name)
+        if not self.dhcp_client_path:
+            raise NoDHCPLeaseMissingDhclientError()
 
     @classmethod
     def kill_dhcp_client(cls):
@@ -198,67 +170,116 @@ def start_service(cls, dhcp_interface: str, distro):
     def stop_service(cls, dhcp_interface: str, distro):
         distro.manage_service("stop", cls.client_name, rcs=[0, 1])
 
+    @abc.abstractmethod
+    def get_newest_lease(self, interface: str) -> Dict[str, Any]:
+        """Get the most recent lease from the ephemeral phase as a dict.
+
+        Return a dict of dhcp options. The dict contains key value
+        pairs from the most recent lease.
+        """
+        return {}
+
+    @staticmethod
+    @abc.abstractmethod
+    def parse_static_routes(routes: str) -> List[Tuple[str, str]]:
+        """
+        parse classless static routes from string
+
+        The tuple is composed of the network_address (including net length) and
+        gateway for a parsed static route.
+
+        @param routes: string containing classless static routes
+        @returns: list of tuple(str, str) for all valid parsed routes until the
+                  first parsing error.
+        """
+        return []
+
+    @abc.abstractmethod
+    def dhcp_discovery(
+        self,
+        interface: str,
+        dhcp_log_func: Optional[Callable] = None,
+        distro=None,
+    ) -> Dict[str, Any]:
+        """Run dhcp client on the interface without scripts or filesystem
+        artifacts.
+
+        @param interface: Name of the network interface on which to send a
+            dhcp request
+        @param dhcp_log_func: A callable accepting the client output and
+            error streams.
+        @param distro: a distro object for network interface manipulation
+        @return: dict of lease options representing the most recent dhcp lease
+            parsed from the dhclient.lease file
+        """
+        return {}
+
 
 class IscDhclient(DhcpClient):
     client_name = "dhclient"
 
     def __init__(self):
-        self.dhclient_path = subp.which("dhclient")
-        if not self.dhclient_path:
-            LOG.debug(
-                "Skip dhclient configuration: No dhclient command found."
-            )
-            raise NoDHCPLeaseMissingDhclientError()
+        super().__init__()
+        self.lease_file = "/run/dhclient.lease"
 
     @staticmethod
-    def parse_dhcp_lease_file(lease_file: str) -> List[Dict[str, Any]]:
-        """Parse the given dhcp lease file returning all leases as dicts.
-
-        Return a list of dicts of dhcp options. Each dict contains key value
-        pairs a specific lease in order from oldest to newest.
+    def parse_leases(lease_content: str) -> List[Dict[str, Any]]:
+        """parse the content of a lease file
 
-        @raises: InvalidDHCPLeaseFileError on empty of unparseable leasefile
-            content.
+        @param lease_content: a string containing the contents of an
+            isc-dhclient lease
+        @return: a list of leases, most recent last
         """
         lease_regex = re.compile(r"lease {(?P<lease>.*?)}\n", re.DOTALL)
-        dhcp_leases = []
-        lease_content = util.load_file(lease_file)
+        dhcp_leases: List[Dict] = []
         if len(lease_content) == 0:
-            raise InvalidDHCPLeaseFileError(
-                "Cannot parse empty dhcp lease file {0}".format(lease_file)
-            )
+            return []
         for lease in lease_regex.findall(lease_content):
             lease_options = []
             for line in lease.split(";"):
                 # Strip newlines, double-quotes and option prefix
                 line = line.strip().replace('"', "").replace("option ", "")
-                if not line:
-                    continue
-                lease_options.append(line.split(" ", 1))
+                if line:
+                    lease_options.append(line.split(" ", 1))
             dhcp_leases.append(dict(lease_options))
-        if not dhcp_leases:
-            raise InvalidDHCPLeaseFileError(
-                "Cannot parse dhcp lease file {0}. No leases found".format(
-                    lease_file
-                )
-            )
         return dhcp_leases
 
+    def get_newest_lease(self, interface: str) -> Dict[str, Any]:
+        """Get the most recent lease from the ephemeral phase as a dict.
+
+        Return a dict of dhcp options. The dict contains key value
+        pairs from the most recent lease.
+
+        @param interface: an interface string - not used in this class, but
+            required for function signature compatibility with other classes
+            that require a distro object
+        @raises: InvalidDHCPLeaseFileError on empty or unparseable leasefile
+            content.
+        """
+        with suppress(FileNotFoundError):
+            content: str
+            content = util.load_file(self.lease_file)  # pyright: ignore
+            if content:
+                dhcp_leases = self.parse_leases(content)
+                if dhcp_leases:
+                    return dhcp_leases[-1]
+        return {}
+
     def dhcp_discovery(
         self,
-        interface,
-        dhcp_log_func=None,
+        interface: str,
+        dhcp_log_func: Optional[Callable] = None,
         distro=None,
-    ):
+    ) -> Dict[str, Any]:
         """Run dhclient on the interface without scripts/filesystem artifacts.
 
-        @param dhclient_cmd_path: Full path to the dhclient used.
-        @param interface: Name of the network interface on which to dhclient.
+        @param interface: Name of the network interface on which to send a
+            dhcp request
         @param dhcp_log_func: A callable accepting the dhclient output and
             error streams.
-
-        @return: A list of dicts of representing the dhcp leases parsed from
-            the dhclient.lease file or empty list.
+        @param distro: a distro object for network interface manipulation
+        @return: dict of lease options representing the most recent dhcp lease
+            parsed from the dhclient.lease file
         """
         LOG.debug("Performing a dhcp discovery on %s", interface)
 
@@ -266,14 +287,16 @@ def dhcp_discovery(
         # side-effects in # /etc/resolv.conf any any other vendor specific
         # scripts in /etc/dhcp/dhclient*hooks.d.
         pid_file = "/run/dhclient.pid"
-        lease_file = "/run/dhclient.lease"
         config_file = None
+        sleep_time = 0.01
+        sleep_cycles = int(self.timeout / sleep_time)
+        maxwait = int(self.timeout / 2)
 
         # this function waits for these files to exist, clean previous runs
         # to avoid false positive in wait_for_files
         with suppress(FileNotFoundError):
             os.remove(pid_file)
-            os.remove(lease_file)
+            os.remove(self.lease_file)
 
         # ISC dhclient needs the interface up to send initial discovery packets
         # Generally dhclient relies on dhclient-script PREINIT action to bring
@@ -300,8 +323,8 @@ def dhcp_discovery(
         try:
             out, err = subp.subp(
                 distro.build_dhclient_cmd(
-                    self.dhclient_path,
-                    lease_file,
+                    self.dhcp_client_path,
+                    self.lease_file,
                     pid_file,
                     interface,
                     config_file,
@@ -324,18 +347,19 @@ def dhcp_discovery(
         # kill the correct process, thus freeing cleandir to be deleted back
         # up the callstack.
         missing = util.wait_for_files(
-            [pid_file, lease_file], maxwait=5, naplen=0.01
+            [pid_file, self.lease_file], maxwait=maxwait, naplen=0.01
         )
         if missing:
             LOG.warning(
                 "dhclient did not produce expected files: %s",
                 ", ".join(os.path.basename(f) for f in missing),
             )
-            return []
+            return {}
 
         ppid = "unknown"
         daemonized = False
-        for _ in range(1000):
+        pid_content = None
+        for _ in range(sleep_cycles):
             pid_content = util.load_file(pid_file).strip()
             try:
                 pid = int(pid_content)
@@ -348,7 +372,7 @@ def dhcp_discovery(
                     os.kill(pid, signal.SIGKILL)
                     daemonized = True
                     break
-            time.sleep(0.01)
+            time.sleep(sleep_time)
 
         if not daemonized:
             LOG.error(
@@ -360,10 +384,13 @@ def dhcp_discovery(
             )
         if dhcp_log_func is not None:
             dhcp_log_func(out, err)
-        return self.parse_dhcp_lease_file(lease_file)
+        lease = self.get_newest_lease(interface)
+        if lease:
+            return lease
+        raise InvalidDHCPLeaseFileError()
 
     @staticmethod
-    def parse_static_routes(rfc3442):
+    def parse_static_routes(routes: str) -> List[Tuple[str, str]]:
         """
         parse rfc3442 format and return a list containing tuple of strings.
 
@@ -373,7 +400,7 @@ def parse_static_routes(rfc3442):
 
         @param rfc3442: string in rfc3442 format (isc or dhcpd)
         @returns: list of tuple(str, str) for all valid parsed routes until the
-                  first parsing error.
+            first parsing error.
 
         e.g.:
 
@@ -395,9 +422,9 @@ def parse_static_routes(rfc3442):
            /etc/dhcp/dhclient-exit-hooks.d/rfc3442-classless-routes
         """
         # raw strings from dhcp lease may end in semi-colon
-        rfc3442 = rfc3442.rstrip(";")
+        rfc3442 = routes.rstrip(";")
         tokens = [tok for tok in re.split(r"[, .]", rfc3442) if tok]
-        static_routes = []
+        static_routes: List[Tuple[str, str]] = []
 
         def _trunc_error(cidr, required, remain):
             msg = (
@@ -470,12 +497,22 @@ def _trunc_error(cidr, required, remain):
         return static_routes
 
     @staticmethod
-    def get_latest_lease(lease_dir, lease_file_regex):
+    def get_newest_lease_file_from_distro(distro) -> Optional[str]:
+        """Get the latest lease file from a distro-managed dhclient
+
+        Doesn't consider the ephemeral timeframe lease.
+
+        @param distro: used for distro-specific lease location and filename
+        @return: The most recent lease file, or None
+        """
         latest_file = None
 
         # Try primary dir/regex, then the fallback ones
         for directory, regex in (
-            (lease_dir, lease_file_regex),
+            (
+                distro.dhclient_lease_directory,
+                distro.dhclient_lease_file_regex,
+            ),
             (DHCLIENT_FALLBACK_LEASE_DIR, DHCLIENT_FALLBACK_LEASE_REGEX),
         ):
             if not directory:
@@ -487,7 +524,7 @@ def get_latest_lease(lease_dir, lease_file_regex):
             except FileNotFoundError:
                 continue
 
-            latest_mtime = -1
+            latest_mtime = -1.0
             for fname in lease_files:
                 if not re.search(regex, fname):
                     continue
@@ -503,62 +540,254 @@ def get_latest_lease(lease_dir, lease_file_regex):
                 return latest_file
         return None
 
-    @staticmethod
-    def parse_dhcp_server_from_lease_file(lease_file) -> Optional[str]:
-        """Parse a lease file for the dhcp server address
+    def get_key_from_latest_lease(self, distro, key: str):
+        """Get a key from the latest lease from distro-managed dhclient
+
+        Doesn't consider the ephemeral timeframe lease.
 
-        @param lease_file: Name of a file to be parsed
-        @return: An address if found, or None
+        @param lease_dir: distro-specific lease to check
+        @param lease_file_regex: distro-specific regex to match lease name
+        @return: The most recent lease file, or None
         """
-        latest_address = None
-        with suppress(FileNotFoundError), open(lease_file, "r") as file:
-            for line in file:
-                if "dhcp-server-identifier" in line:
-                    words = line.strip(" ;\r\n").split(" ")
-                    if len(words) > 2:
-                        dhcptok = words[2]
-                        LOG.debug("Found DHCP identifier %s", dhcptok)
-                        latest_address = dhcptok
-        return latest_address
-
-
-class Dhcpcd:
+        lease_file = self.get_newest_lease_file_from_distro(distro)
+        if lease_file:
+            content: str
+            content = util.load_file(lease_file)  # pyright: ignore
+            if content:
+                for lease in reversed(self.parse_leases(content)):
+                    server = lease.get(key)
+                    if server:
+                        return server
+
+
+class Dhcpcd(DhcpClient):
     client_name = "dhcpcd"
 
-    def __init__(self):
-        raise NoDHCPLeaseMissingDhclientError("Dhcpcd not yet implemented")
+    def dhcp_discovery(
+        self,
+        interface: str,
+        dhcp_log_func: Optional[Callable] = None,
+        distro=None,
+    ) -> Dict[str, Any]:
+        """Run dhcpcd on the interface without scripts/filesystem artifacts.
+
+        @param interface: Name of the network interface on which to send a
+            dhcp request
+        @param dhcp_log_func: A callable accepting the client output and
+            error streams.
+        @param distro: a distro object for network interface manipulation
+        @return: dict of lease options representing the most recent dhcp lease
+            parsed from the dhclient.lease file
+        """
+        LOG.debug("Performing a dhcp discovery on %s", interface)
+        sleep_time = 0.01
+        sleep_cycles = int(self.timeout / sleep_time)
+
+        # ISC dhclient needs the interface up to send initial discovery packets
+        # Generally dhclient relies on dhclient-script PREINIT action to bring
+        # the link up before attempting discovery. Since we are using
+        # -sf /bin/true, we need to do that "link up" ourselves first.
+        distro.net_ops.link_up(interface)
+
+        # TODO: disabling hooks means we need to get all of the files in
+        # /lib/dhcpcd/dhcpcd-hooks/ and pass each of those with the --nohook
+        # argument to dhcpcd
+        try:
+            command = [
+                self.dhcp_client_path,  # pyright: ignore
+                "--ipv4only",  # only attempt configuring ipv4
+                "--waitip",  # wait for ipv4 to be configured
+                "--persistent",  # don't deconfigure when dhcpcd exits
+                "--noarp",  # don't be slow
+                "--script=/bin/true",  # disable hooks
+                interface,
+            ]
+            out, err = subp.subp(
+                command,
+                timeout=self.timeout,
+            )
+            if dhcp_log_func is not None:
+                dhcp_log_func(out, err)
+            lease = self.get_newest_lease(interface)
+            # Attempt cleanup and leave breadcrumbs if it fails, but return
+            # the lease regardless of failure to clean up dhcpcd.
+            if lease:
+                # Note: the pid file location depends on the arguments passed
+                # it can be discovered with the -P flag
+                pid_file = subp.subp([*command, "-P"]).stdout
+                pid_content = None
+                debug_msg = ""
+                for _ in range(sleep_cycles):
+                    try:
+                        pid_content = util.load_file(pid_file).strip()
+                        pid = int(pid_content)
+                        LOG.debug("killing dhcpcd with pid=%s", pid)
+                        os.kill(pid, signal.SIGKILL)
+                    except FileNotFoundError:
+                        debug_msg = (
+                            f"No PID file found at {pid_file}, "
+                            "dhcpcd is still running"
+                        )
+                    except ValueError:
+                        debug_msg = (
+                            f"PID file contained [{pid_content}], "
+                            "dhcpcd is still running"
+                        )
+                    else:
+                        return lease
+                    time.sleep(sleep_time)
+                LOG.debug(debug_msg)
+                return lease
+            raise NoDHCPLeaseError("No lease found")
+
+        except subp.ProcessExecutionError as error:
+            LOG.debug(
+                "dhclient exited with code: %s stderr: %r stdout: %r",
+                error.exit_code,
+                error.stderr,
+                error.stdout,
+            )
+            raise NoDHCPLeaseError from error
+
+    @staticmethod
+    def parse_dhcpcd_lease(lease_dump: str, interface: str) -> Dict:
+        """parse the output of dhcpcd --dump
+
+        map names to the datastructure we create from dhclient
+
+        example dhcpcd output:
+
+        broadcast_address='192.168.15.255'
+        dhcp_lease_time='3600'
+        dhcp_message_type='5'
+        dhcp_server_identifier='192.168.0.1'
+        domain_name='us-east-2.compute.internal'
+        domain_name_servers='192.168.0.2'
+        host_name='ip-192-168-0-212'
+        interface_mtu='9001'
+        ip_address='192.168.0.212'
+        network_number='192.168.0.0'
+        routers='192.168.0.1'
+        subnet_cidr='20'
+        subnet_mask='255.255.240.0'
+        """
+
+        # create a dict from dhcpcd dump output - remove single quotes
+        lease = dict(
+            [
+                a.split("=")
+                for a in lease_dump.strip().replace("'", "").split("\n")
+            ]
+        )
+
+        # this is expected by cloud-init's code
+        lease["interface"] = interface
+
+        # transform underscores to hyphens
+        lease = {key.replace("_", "-"): value for key, value in lease.items()}
+
+        # - isc-dhclient uses the key name "fixed-address" in place of
+        #   "ip-address", and in the codebase some code assumes that we can use
+        #   isc-dhclient's option names. Map accordingly
+        # - ephemeral.py we use an internal key name "static_routes" to map
+        #   what I think is some RHEL customization to the isc-dhclient
+        #   code, so we need to match this key for use there.
+        name_map = {
+            "ip-address": "fixed-address",
+            "classless-static-routes": "static_routes",
+        }
+        for source, destination in name_map.items():
+            if source in lease:
+                lease[destination] = lease.pop(source)
+        return lease
+
+    def get_newest_lease(self, interface: str) -> Dict[str, Any]:
+        """Return a dict of dhcp options.
+
+        @param interface: which interface to dump the lease from
+        @raises: InvalidDHCPLeaseFileError on empty or unparseable leasefile
+            content.
+        """
+        try:
+            return self.parse_dhcpcd_lease(
+                subp.subp(
+                    [
+                        "dhcpcd",
+                        "--dumplease",
+                        "--ipv4only",
+                        interface,
+                    ],
+                ).stdout,
+                interface,
+            )
+
+        except subp.ProcessExecutionError as error:
+            LOG.debug(
+                "dhcpcd exited with code: %s stderr: %r stdout: %r",
+                error.exit_code,
+                error.stderr,
+                error.stdout,
+            )
+            raise NoDHCPLeaseError from error
+
+    @staticmethod
+    def parse_static_routes(routes: str) -> List[Tuple[str, str]]:
+        """
+        classless static routes as returned from dhcpcd --dumplease and return
+        a list containing tuple of strings.
+
+        The tuple is composed of the network_address (including net length) and
+        gateway for a parsed static route.
+
+        @param routes: string containing classless static routes
+        @returns: list of tuple(str, str) for all valid parsed routes until the
+                  first parsing error.
+
+        e.g.:
+
+        sr=parse_static_routes(
+            "0.0.0.0/0 10.0.0.1 168.63.129.16/32 10.0.0.1"
+        )
+        sr=[
+            ("0.0.0.0/0", "10.0.0.1"),
+            ("169.63.129.16/32", "10.0.0.1"),
+        ]
+        """
+        static_routes = routes.split()
+        if static_routes:
+            # format: dest1/mask gw1 ... destn/mask gwn
+            return [i for i in zip(static_routes[::2], static_routes[1::2])]
+        LOG.warning("Malformed classless static routes: [%s]", routes)
+        return []
 
 
 class Udhcpc(DhcpClient):
     client_name = "udhcpc"
 
     def __init__(self):
-        self.udhcpc_path = subp.which("udhcpc")
-        if not self.udhcpc_path:
-            LOG.debug("Skip udhcpc configuration: No udhcpc command found.")
-            raise NoDHCPLeaseMissingUdhcpcError()
+        super().__init__()
+        self.lease_file = None
 
     def dhcp_discovery(
         self,
-        interface,
-        dhcp_log_func=None,
+        interface: str,
+        dhcp_log_func: Optional[Callable] = None,
         distro=None,
-    ):
+    ) -> Dict[str, Any]:
         """Run udhcpc on the interface without scripts or filesystem artifacts.
 
         @param interface: Name of the network interface on which to run udhcpc.
         @param dhcp_log_func: A callable accepting the udhcpc output and
             error streams.
-
         @return: A list of dicts of representing the dhcp leases parsed from
             the udhcpc lease file.
         """
         LOG.debug("Performing a dhcp discovery on %s", interface)
 
         tmp_dir = temp_utils.get_tmp_ancestor(needs_exe=True)
-        lease_file = os.path.join(tmp_dir, interface + ".lease.json")
+        self.lease_file = os.path.join(tmp_dir, interface + ".lease.json")
         with suppress(FileNotFoundError):
-            os.remove(lease_file)
+            os.remove(self.lease_file)
 
         # udhcpc needs the interface up to send initial discovery packets
         distro.net_ops.link_up(interface)
@@ -567,7 +796,7 @@ def dhcp_discovery(
         util.write_file(udhcpc_script, UDHCPC_SCRIPT, 0o755)
 
         cmd = [
-            self.udhcpc_path,
+            self.dhcp_client_path,
             "-O",
             "staticroutes",
             "-i",
@@ -593,7 +822,7 @@ def dhcp_discovery(
             )
         try:
             out, err = subp.subp(
-                cmd, update_env={"LEASE_FILE": lease_file}, capture=True
+                cmd, update_env={"LEASE_FILE": self.lease_file}, capture=True
             )
         except subp.ProcessExecutionError as error:
             LOG.debug(
@@ -607,11 +836,29 @@ def dhcp_discovery(
         if dhcp_log_func is not None:
             dhcp_log_func(out, err)
 
-        lease_json = util.load_json(util.load_file(lease_file))
-        static_routes = lease_json["static_routes"].split()
+        return self.get_newest_lease(interface)
+
+    def get_newest_lease(self, interface: str) -> Dict[str, Any]:
+        """Get the most recent lease from the ephemeral phase as a dict.
+
+        Return a dict of dhcp options. The dict contains key value
+        pairs from the most recent lease.
+
+        @param interface: an interface name - not used in this class, but
+            required for function signature compatibility with other classes
+            that require a distro object
+        @raises: InvalidDHCPLeaseFileError on empty or unparseable leasefile
+            content.
+        """
+        return util.load_json(util.load_file(self.lease_file))
+
+    @staticmethod
+    def parse_static_routes(routes: str) -> List[Tuple[str, str]]:
+        static_routes = routes.split()
         if static_routes:
             # format: dest1/mask gw1 ... destn/mask gwn
-            lease_json["static_routes"] = [
-                i for i in zip(static_routes[::2], static_routes[1::2])
-            ]
-        return [lease_json]
+            return [i for i in zip(static_routes[::2], static_routes[1::2])]
+        return []
+
+
+ALL_DHCP_CLIENTS = [IscDhclient, Dhcpcd, Udhcpc]
diff --git a/cloudinit/net/ephemeral.py b/cloudinit/net/ephemeral.py
index 28c851cd706e..fa601c350103 100644
--- a/cloudinit/net/ephemeral.py
+++ b/cloudinit/net/ephemeral.py
@@ -9,7 +9,7 @@
 
 import cloudinit.net as net
 from cloudinit.net.dhcp import (
-    IscDhclient,
+    Dhcpcd,
     NoDHCPLeaseError,
     maybe_perform_dhcp_discovery,
 )
@@ -285,12 +285,11 @@ def obtain_lease(self):
         """
         if self.lease:
             return self.lease
-        leases = maybe_perform_dhcp_discovery(
+        self.lease = maybe_perform_dhcp_discovery(
             self.distro, self.iface, self.dhcp_log_func
         )
-        if not leases:
+        if not self.lease:
             raise NoDHCPLeaseError()
-        self.lease = leases[-1]
         LOG.debug(
             "Received dhcp lease on %s for %s/%s",
             self.lease["interface"],
@@ -305,6 +304,7 @@ def obtain_lease(self):
             "static_routes": [
                 "rfc3442-classless-static-routes",
                 "classless-static-routes",
+                "static_routes",
             ],
             "router": "routers",
         }
@@ -314,12 +314,17 @@ def obtain_lease(self):
                 kwargs["prefix_or_mask"], kwargs["ip"]
             )
         if kwargs["static_routes"]:
-            kwargs["static_routes"] = IscDhclient.parse_static_routes(
+            kwargs[
+                "static_routes"
+            ] = self.distro.dhcp_client.parse_static_routes(
                 kwargs["static_routes"]
             )
         if self.connectivity_url_data:
             kwargs["connectivity_url_data"] = self.connectivity_url_data
-        ephipv4 = EphemeralIPv4Network(self.distro, **kwargs)
+        if isinstance(self.distro.dhcp_client, Dhcpcd):
+            ephipv4 = DhcpcdEphemeralIPv4Network(self.distro, **kwargs)
+        else:
+            ephipv4 = EphemeralIPv4Network(self.distro, **kwargs)
         ephipv4.__enter__()
         self._ephipv4 = ephipv4
         return self.lease
@@ -343,6 +348,16 @@ def get_first_option_value(
                 result[internal_mapping] = self.lease.get(different_names)
 
 
+class DhcpcdEphemeralIPv4Network(EphemeralIPv4Network):
+    """dhcpcd sets up its own ephemeral network and routes"""
+
+    def __enter__(self):
+        return
+
+    def __exit__(self, excp_type, excp_value, excp_traceback):
+        return
+
+
 class EphemeralIPNetwork:
     """Combined ephemeral context manager for IPv4 and IPv6
 
diff --git a/cloudinit/sources/DataSourceCloudStack.py b/cloudinit/sources/DataSourceCloudStack.py
index c4c1667e2f3f..6fd428ddf9a8 100644
--- a/cloudinit/sources/DataSourceCloudStack.py
+++ b/cloudinit/sources/DataSourceCloudStack.py
@@ -15,6 +15,7 @@
 import logging
 import os
 import time
+from contextlib import suppress
 from socket import gaierror, getaddrinfo, inet_ntoa
 from struct import pack
 
@@ -110,17 +111,32 @@ def _get_domainname(self):
             "Falling back to ISC dhclient"
         )
 
-        lease_file = dhcp.IscDhclient.get_latest_lease(
-            self.distro.dhclient_lease_directory,
-            self.distro.dhclient_lease_file_regex,
+        # some distros might use isc-dhclient for network setup via their
+        # network manager. If this happens, the lease is more recent than the
+        # ephemeral lease, so use it first.
+        with suppress(dhcp.NoDHCPLeaseMissingDhclientError):
+            domain_name = dhcp.IscDhclient().get_key_from_latest_lease(
+                self.distro, "domain-name"
+            )
+            if domain_name:
+                return domain_name
+
+        LOG.debug(
+            "Could not obtain FQDN from ISC dhclient leases. "
+            "Falling back to %s",
+            self.distro.dhcp_client.client_name,
         )
-        if not lease_file:
-            LOG.debug("Dhclient lease file wasn't found")
-            return None
 
-        latest_lease = dhcp.IscDhclient.parse_dhcp_lease_file(lease_file)[-1]
-        domainname = latest_lease.get("domain-name", None)
-        return domainname if domainname else None
+        # If no distro leases were found, check the ephemeral lease that
+        # cloud-init set up.
+        with suppress(FileNotFoundError):
+            latest_lease = self.distro.dhcp_client.get_newest_lease(
+                self.distro.fallback_interface
+            )
+            domain_name = latest_lease.get("domain-name") or None
+            return domain_name
+        LOG.debug("No dhcp leases found")
+        return None
 
     def get_hostname(
         self,
@@ -287,19 +303,26 @@ def get_vr_address(distro):
         return latest_address
 
     # Try dhcp lease files next
-    # get_latest_lease() needs a Distro object to know which directory
+    # get_key_from_latest_lease() needs a Distro object to know which directory
     # stores lease files
-    lease_file = dhcp.IscDhclient.get_latest_lease(
-        distro.dhclient_lease_directory, distro.dhclient_lease_file_regex
-    )
-
-    if lease_file:
-        latest_address = dhcp.IscDhclient.parse_dhcp_server_from_lease_file(
-            lease_file
+    with suppress(dhcp.NoDHCPLeaseMissingDhclientError):
+        latest_address = dhcp.IscDhclient().get_key_from_latest_lease(
+            distro, "dhcp-server-identifier"
         )
         if latest_address:
+            LOG.debug("Found SERVER_ADDRESS '%s' via dhclient", latest_address)
             return latest_address
 
+    with suppress(FileNotFoundError):
+        latest_lease = distro.dhcp_client.get_newest_lease(distro)
+        if latest_lease:
+            LOG.debug(
+                "Found SERVER_ADDRESS '%s' via ephemeral %s lease ",
+                latest_lease,
+                distro.dhcp_client.client_name,
+            )
+            return latest_lease
+
     # No virtual router found, fallback to default gateway
     LOG.debug("No DHCP found, using default gateway")
     return get_default_gateway()
diff --git a/cloudinit/sources/DataSourceEc2.py b/cloudinit/sources/DataSourceEc2.py
index 9e6bfbd10d6e..8904f9fb37ff 100644
--- a/cloudinit/sources/DataSourceEc2.py
+++ b/cloudinit/sources/DataSourceEc2.py
@@ -131,7 +131,7 @@ def _get_data(self):
             try:
                 with EphemeralIPNetwork(
                     self.distro,
-                    self.fallback_interface,
+                    self.distro.fallback_interface,
                     ipv4=True,
                     ipv6=True,
                 ) as netw:
@@ -500,7 +500,7 @@ def network_config(self):
                 func=self.get_data,
             )
 
-        iface = self.fallback_interface
+        iface = self.distro.fallback_interface
         net_md = self.metadata.get("network")
         if isinstance(net_md, dict):
             # SRU_BLOCKER: xenial, bionic and eoan should default
@@ -532,19 +532,6 @@ def network_config(self):
 
         return self._network_config
 
-    @property
-    def fallback_interface(self):
-        if self._fallback_interface is None:
-            # fallback_nic was used at one point, so restored objects may
-            # have an attribute there. respect that if found.
-            _legacy_fbnic = getattr(self, "fallback_nic", None)
-            if _legacy_fbnic:
-                self._fallback_interface = _legacy_fbnic
-                self.fallback_nic = None
-            else:
-                return super(DataSourceEc2, self).fallback_interface
-        return self._fallback_interface
-
     def crawl_metadata(self):
         """Crawl metadata service when available.
 
diff --git a/cloudinit/sources/DataSourceGCE.py b/cloudinit/sources/DataSourceGCE.py
index c730ae8674fc..1c92cacec7fa 100644
--- a/cloudinit/sources/DataSourceGCE.py
+++ b/cloudinit/sources/DataSourceGCE.py
@@ -124,10 +124,10 @@ def _get_data(self):
                 except NoDHCPLeaseError:
                     continue
                 if ret["success"]:
-                    self._fallback_interface = candidate_nic
+                    self.distro.fallback_interface = candidate_nic
                     LOG.debug("Primary NIC found: %s.", candidate_nic)
                     break
-            if self._fallback_interface is None:
+            if self.distro.fallback_interface is None:
                 LOG.warning(
                     "Did not find a fallback interface on %s.", self.cloud_name
                 )
diff --git a/cloudinit/sources/DataSourceOpenStack.py b/cloudinit/sources/DataSourceOpenStack.py
index 7ffa90902633..ef407bd31da6 100644
--- a/cloudinit/sources/DataSourceOpenStack.py
+++ b/cloudinit/sources/DataSourceOpenStack.py
@@ -153,7 +153,9 @@ def _get_data(self):
         if self.perform_dhcp_setup:  # Setup networking in init-local stage.
             try:
 
-                with EphemeralDHCPv4(self.distro, self.fallback_interface):
+                with EphemeralDHCPv4(
+                    self.distro, self.distro.fallback_interface
+                ):
                     results = util.log_time(
                         logfunc=LOG.debug,
                         msg="Crawl of metadata service",
diff --git a/cloudinit/sources/DataSourceScaleway.py b/cloudinit/sources/DataSourceScaleway.py
index fbf45df072f7..bd6b00020b75 100644
--- a/cloudinit/sources/DataSourceScaleway.py
+++ b/cloudinit/sources/DataSourceScaleway.py
@@ -19,7 +19,7 @@
 from urllib3.connection import HTTPConnection
 from urllib3.poolmanager import PoolManager
 
-from cloudinit import dmi, net, sources, url_helper, util
+from cloudinit import dmi, sources, url_helper, util
 from cloudinit.event import EventScope, EventType
 from cloudinit.net.dhcp import NoDHCPLeaseError
 from cloudinit.net.ephemeral import EphemeralDHCPv4, EphemeralIPv6Network
@@ -171,7 +171,6 @@ def __init__(self, sys_cfg, distro, paths):
         self.retries = int(self.ds_cfg.get("retries", DEF_MD_RETRIES))
         self.timeout = int(self.ds_cfg.get("timeout", DEF_MD_TIMEOUT))
         self.max_wait = int(self.ds_cfg.get("max_wait", DEF_MD_MAX_WAIT))
-        self._fallback_interface = None
         self._network_config = sources.UNSET
         self.metadata_urls = DS_BASE_URLS
         self.userdata_url = None
@@ -267,9 +266,6 @@ def _set_urls_on_ip_version(self, proto, urls):
 
     def _get_data(self):
 
-        if self._fallback_interface is None:
-            self._fallback_interface = net.find_fallback_nic()
-
         # The DataSource uses EventType.BOOT so we are called more than once.
         # Try to crawl metadata on IPv4 first and set has_ipv4 to False if we
         # timeout so we do not try to crawl on IPv4 more than once.
@@ -280,7 +276,7 @@ def _get_data(self):
                 # it will only reach timeout on VMs with only IPv6 addresses.
                 with EphemeralDHCPv4(
                     self.distro,
-                    self._fallback_interface,
+                    self.distro.fallback_interface,
                 ) as ipv4:
                     util.log_time(
                         logfunc=LOG.debug,
@@ -311,7 +307,7 @@ def _get_data(self):
             try:
                 with EphemeralIPv6Network(
                     self.distro,
-                    self._fallback_interface,
+                    self.distro.fallback_interface,
                 ):
                     util.log_time(
                         logfunc=LOG.debug,
@@ -346,9 +342,6 @@ def network_config(self):
         if self._network_config != sources.UNSET:
             return self._network_config
 
-        if self._fallback_interface is None:
-            self._fallback_interface = net.find_fallback_nic()
-
         if self.metadata["private_ip"] is None:
             # New method of network configuration
 
@@ -377,13 +370,13 @@ def network_config(self):
                             ip_cfg["routes"] += [route]
                         else:
                             ip_cfg["routes"] = [route]
-            netcfg[self._fallback_interface] = ip_cfg
+            netcfg[self.distro.fallback_interface] = ip_cfg
             self._network_config = {"version": 2, "ethernets": netcfg}
         else:
             # Kept for backward compatibility
             netcfg = {
                 "type": "physical",
-                "name": "%s" % self._fallback_interface,
+                "name": "%s" % self.distro.fallback_interface,
             }
             subnets = [{"type": "dhcp4"}]
             if self.metadata["ipv6"]:
diff --git a/cloudinit/sources/__init__.py b/cloudinit/sources/__init__.py
index c207b5ed6df0..1b2ead2c31b2 100644
--- a/cloudinit/sources/__init__.py
+++ b/cloudinit/sources/__init__.py
@@ -195,9 +195,6 @@ class DataSource(CloudInitPickleMixin, metaclass=abc.ABCMeta):
     #  - seed-dir (<dirname>)
     _subplatform = None
 
-    # Track the discovered fallback nic for use in configuration generation.
-    _fallback_interface = None
-
     # The network configuration sources that should be considered for this data
     # source.  (The first source in this list that provides network
     # configuration will be used without considering any that follow.)  This
@@ -607,17 +604,6 @@ def get_vendordata2(self):
             self.vendordata2 = self.ud_proc.process(self.get_vendordata2_raw())
         return self.vendordata2
 
-    @property
-    def fallback_interface(self):
-        """Determine the network interface used during local network config."""
-        if self._fallback_interface is None:
-            self._fallback_interface = net.find_fallback_nic()
-            if self._fallback_interface is None:
-                LOG.warning(
-                    "Did not find a fallback interface on %s.", self.cloud_name
-                )
-        return self._fallback_interface
-
     @property
     def platform_type(self):
         if not hasattr(self, "_platform_type"):
diff --git a/cloudinit/subp.py b/cloudinit/subp.py
index 3841a2f2dbe0..4c01d743f0fa 100644
--- a/cloudinit/subp.py
+++ b/cloudinit/subp.py
@@ -155,6 +155,7 @@ def subp(
     decode="replace",
     update_env=None,
     cwd=None,
+    timeout=None,
 ) -> SubpResult:
     """Run a subprocess.
 
@@ -181,6 +182,8 @@ def subp(
         this will not affect the current processes os.environ.
     :param cwd:
         change the working directory to cwd before executing the command.
+    :param timeout: maximum time for the subprocess to run, passed directly to
+        the timeout parameter of Popen.communicate()
 
     :return
         if not capturing, return is (None, None)
@@ -252,7 +255,7 @@ def subp(
             shell=shell,
             cwd=cwd,
         )
-        out, err = sp.communicate(data)
+        out, err = sp.communicate(data, timeout=timeout)
         total = time.time() - before
         if total > 0.1:
             LOG.debug("command %s took %.3ss to run", args, total)
diff --git a/cloudinit/util.py b/cloudinit/util.py
index 6401a196f7a7..dda653789406 100644
--- a/cloudinit/util.py
+++ b/cloudinit/util.py
@@ -124,14 +124,14 @@ def lsb_release():
     return data
 
 
-def decode_binary(blob, encoding="utf-8"):
+def decode_binary(blob, encoding="utf-8") -> str:
     # Converts a binary type into a text type using given encoding.
     if isinstance(blob, str):
         return blob
     return blob.decode(encoding)
 
 
-def encode_text(text, encoding="utf-8"):
+def encode_text(text, encoding="utf-8") -> bytes:
     # Converts a text string into a binary type using given encoding.
     if isinstance(text, bytes):
         return text
diff --git a/config/cloud.cfg.tmpl b/config/cloud.cfg.tmpl
index 45838754bcbb..d84627fc5d0e 100644
--- a/config/cloud.cfg.tmpl
+++ b/config/cloud.cfg.tmpl
@@ -302,6 +302,7 @@ system_info:
 {% elif variant in ["ubuntu", "unknown"] %}
 {# SRU_BLOCKER: do not ship network renderers on Xenial, Bionic or Eoan #}
   network:
+    dhcp_client_priority: [dhclient, dhcpcd, udhcpc]
     renderers: ['netplan', 'eni', 'sysconfig']
     activators: ['netplan', 'eni', 'network-manager', 'networkd']
 {% elif is_rhel %}
diff --git a/tests/integration_tests/net/test_dhcp.py b/tests/integration_tests/net/test_dhcp.py
new file mode 100644
index 000000000000..65b6bdfc44a9
--- /dev/null
+++ b/tests/integration_tests/net/test_dhcp.py
@@ -0,0 +1,74 @@
+"""Integration tests related to cloud-init dhcp."""
+
+import pytest
+
+from tests.integration_tests.integration_settings import PLATFORM
+from tests.integration_tests.releases import CURRENT_RELEASE, IS_UBUNTU, NOBLE
+from tests.integration_tests.util import verify_clean_log
+
+
+@pytest.mark.skipif(not IS_UBUNTU, reason="ubuntu-specific tests")
+@pytest.mark.skipif(
+    PLATFORM not in ["azure", "ec2", "gce", "openstack"],
+    reason="not all platforms require dhcp",
+)
+class TestDHCP:
+    """Integration tests relating to dhcp"""
+
+    @pytest.mark.skipif(
+        CURRENT_RELEASE >= NOBLE, reason="noble and later use dhcpcd"
+    )
+    def test_old_ubuntu_uses_isc_dhclient_by_default(self, client):
+        """verify that old releases use dhclient"""
+        log = client.read_from_file("/var/log/cloud-init.log")
+        assert "DHCP client selected: dhclient" in log
+        verify_clean_log(log)
+
+    @pytest.mark.xfail(
+        reason=(
+            "Noble images have dhclient installed and ordered first in their"
+            "configuration. Until this changes, dhcpcd will not be used"
+        )
+    )
+    @pytest.mark.skipif(
+        CURRENT_RELEASE < NOBLE, reason="pre-noble uses dhclient"
+    )
+    def test_noble_and_newer_uses_dhcpcd_by_default(self, client):
+        """verify that noble will use dhcpcd"""
+        log = client.read_from_file("/var/log/cloud-init.log")
+        assert "DHCP client selected: dhcpcd" in log
+        assert (
+            ", DHCP is still running" not in log
+        ), "cloud-init leaked a dhcp daemon that is still running"
+        verify_clean_log(log)
+
+    @pytest.mark.skipif(
+        CURRENT_RELEASE < NOBLE,
+        reason="earlier Ubuntu releases have a package named dhcpcd5",
+    )
+    def test_noble_and_newer_force_dhcp(self, client):
+        """force noble to use dhcpcd and test that it worked"""
+        client.execute(
+            "sed -i 's|"
+            "dhcp_client_priority.*$"
+            "|dhcp_client_priority: [dhcpcd, dhclient, udhcpc]"
+            "|' /etc/cloud/cloud.cfg"
+        )
+        client.execute("cloud-init clean --logs")
+        client.restart()
+        log = client.read_from_file("/var/log/cloud-init.log")
+        for line in log.split("\n"):
+            if "DHCP client selected" in line:
+                assert (
+                    "DHCP client selected: dhcpcd" in line
+                ), f"Selected incorrect dhcp client: {line}"
+                break
+        else:
+            assert False, "No dhcp client selected"
+        assert "Received dhcp lease on" in log, "No lease received"
+        assert (
+            ", DHCP is still running" not in log
+        ), "cloud-init leaked a dhcp daemon that is still running"
+        if not "ec2" == PLATFORM:
+            assert "Received dhcp lease on " in log, "EphemeralDHCPv4 failed"
+        verify_clean_log(log)
diff --git a/tests/unittests/conftest.py b/tests/unittests/conftest.py
index 91dbd06f16c7..fe83fa25a221 100644
--- a/tests/unittests/conftest.py
+++ b/tests/unittests/conftest.py
@@ -76,6 +76,17 @@ def side_effect(args, *other_args, **kwargs):
         yield
 
 
+# @pytest.fixture(autouse=True)
+@pytest.fixture()
+def dhclient_exists():
+    with mock.patch(
+        "cloudinit.net.dhcp.subp.which",
+        return_value="/sbin/dhclient",
+        autospec=True,
+    ):
+        yield
+
+
 log.configure_root_logger()
 
 
diff --git a/tests/unittests/distros/test__init__.py b/tests/unittests/distros/test__init__.py
index eabd551e5c7c..2f2636e56ddd 100644
--- a/tests/unittests/distros/test__init__.py
+++ b/tests/unittests/distros/test__init__.py
@@ -8,6 +8,8 @@
 import pytest
 
 from cloudinit import distros, util
+from cloudinit.distros.ubuntu import Distro
+from cloudinit.net.dhcp import Dhcpcd, IscDhclient, Udhcpc
 from tests.unittests import helpers
 
 M_PATH = "cloudinit.distros."
@@ -501,3 +503,72 @@ def test_get_tmp_exec_path(
             assert "/tmp" == tmp_path
         else:
             assert "/usr_lib_exec/cloud-init/clouddir" == tmp_path
+
+
+@pytest.mark.parametrize(
+    "chosen_client, config, which_override",
+    [
+        pytest.param(
+            IscDhclient,
+            {"network": {"dhcp_client_priority": ["dhclient"]}},
+            None,
+            id="single_client_is_found_from_config_dhclient",
+        ),
+        pytest.param(
+            Udhcpc,
+            {"network": {"dhcp_client_priority": ["udhcpc"]}},
+            None,
+            id="single_client_is_found_from_config_udhcpc",
+        ),
+        pytest.param(
+            Dhcpcd,
+            {"network": {"dhcp_client_priority": ["dhcpcd"]}},
+            None,
+            id="single_client_is_found_from_config_dhcpcd",
+        ),
+        pytest.param(
+            Dhcpcd,
+            {"network": {"dhcp_client_priority": ["dhcpcd", "dhclient"]}},
+            None,
+            id="first_client_is_found_from_config_dhcpcd",
+        ),
+        pytest.param(
+            Udhcpc,
+            {
+                "network": {
+                    "dhcp_client_priority": ["udhcpc", "dhcpcd", "dhclient"]
+                }
+            },
+            None,
+            id="first_client_is_found_from_config_udhcpc",
+        ),
+        pytest.param(
+            IscDhclient,
+            {"network": {"dhcp_client_priority": []}},
+            None,
+            id="first_client_is_found_no_config_dhclient",
+        ),
+        pytest.param(
+            Dhcpcd,
+            {
+                "network": {
+                    "dhcp_client_priority": ["udhcpc", "dhcpcd", "dhclient"]
+                }
+            },
+            [False, False, True, True],
+            id="second_client_is_found_from_config_dhcpcd",
+        ),
+    ],
+)
+class TestDHCP:
+    @mock.patch("cloudinit.net.dhcp.subp.which")
+    def test_dhcp_configuration(
+        self, m_which, chosen_client, config, which_override
+    ):
+        """check that, when a user provides a configuration at
+        network.dhcp_client_priority, the correct client is chosen
+        """
+        m_which.side_effect = which_override
+        distro = Distro("", {}, {})
+        distro._cfg = config
+        assert isinstance(distro.dhcp_client, chosen_client)
diff --git a/tests/unittests/net/test_dhcp.py b/tests/unittests/net/test_dhcp.py
index b618e7fad675..21b8733986a6 100644
--- a/tests/unittests/net/test_dhcp.py
+++ b/tests/unittests/net/test_dhcp.py
@@ -8,8 +8,10 @@
 import responses
 
 from cloudinit.distros import alpine, amazon, centos, debian, freebsd, rhel
+from cloudinit.distros.ubuntu import Distro
 from cloudinit.net.dhcp import (
     DHCLIENT_FALLBACK_LEASE_DIR,
+    Dhcpcd,
     InvalidDHCPLeaseFileError,
     IscDhclient,
     NoDHCPLeaseError,
@@ -54,52 +56,51 @@
     ),
 )
 class TestParseDHCPServerFromLeaseFile:
+    @pytest.mark.usefixtures("dhclient_exists")
     def test_find_server_address_when_present(
         self, server_address, lease_file_content, tmp_path
     ):
         """Test that we return None in the case of no file or file contains no
         server address, otherwise return the address.
         """
-        lease_file = tmp_path / "dhcp.leases"
-        if server_address:
-            if lease_file_content:
-                lease_file.write_text(lease_file_content)
-            assert (
-                server_address
-                == IscDhclient.parse_dhcp_server_from_lease_file(lease_file)
-            )
+        dhclient = IscDhclient()
+        dhclient.lease_file = tmp_path / "dhcp.leases"
+        if lease_file_content:
+            dhclient.lease_file.write_text(lease_file_content)
+            if server_address:
+                assert server_address == dhclient.get_newest_lease("eth0").get(
+                    "dhcp-server-identifier"
+                )
         else:
-            assert not IscDhclient.parse_dhcp_server_from_lease_file(
-                lease_file
+            assert None is dhclient.get_newest_lease("eth0").get(
+                "dhcp-server-identifier"
             )
 
 
+@pytest.mark.usefixtures("dhclient_exists")
 class TestParseDHCPLeasesFile(CiTestCase):
     def test_parse_empty_lease_file_errors(self):
-        """parse_dhcp_lease_file errors when file content is empty."""
-        empty_file = self.tmp_path("leases")
-        ensure_file(empty_file)
-        with self.assertRaises(InvalidDHCPLeaseFileError) as context_manager:
-            IscDhclient.parse_dhcp_lease_file(empty_file)
-        error = context_manager.exception
-        self.assertIn("Cannot parse empty dhcp lease file", str(error))
+        """get_newest_lease errors when file content is empty."""
+        client = IscDhclient()
+        client.lease_file = self.tmp_path("leases")
+        ensure_file(client.lease_file)
+        assert not client.get_newest_lease("eth0")
 
     def test_parse_malformed_lease_file_content_errors(self):
-        """IscDhclient.parse_dhcp_lease_file errors when file content isn't
+        """IscDhclient.get_newest_lease errors when file content isn't
         dhcp leases.
         """
-        non_lease_file = self.tmp_path("leases")
-        write_file(non_lease_file, "hi mom.")
-        with self.assertRaises(InvalidDHCPLeaseFileError) as context_manager:
-            IscDhclient.parse_dhcp_lease_file(non_lease_file)
-        error = context_manager.exception
-        self.assertIn("Cannot parse dhcp lease file", str(error))
+        client = IscDhclient()
+        client.lease_file = self.tmp_path("leases")
+        write_file(client.lease_file, "hi mom.")
+        assert not client.get_newest_lease("eth0")
 
     def test_parse_multiple_leases(self):
-        """IscDhclient.parse_dhcp_lease_file returns a list of all leases
+        """IscDhclient().get_newest_lease returns the latest lease
         within.
         """
-        lease_file = self.tmp_path("leases")
+        client = IscDhclient()
+        client.lease_file = self.tmp_path("leases")
         content = dedent(
             """
             lease {
@@ -120,38 +121,30 @@ def test_parse_multiple_leases(self):
             }
         """
         )
-        expected = [
-            {
-                "interface": "wlp3s0",
-                "fixed-address": "192.168.2.74",
-                "subnet-mask": "255.255.255.0",
-                "routers": "192.168.2.1",
-                "renew": "4 2017/07/27 18:02:30",
-                "expire": "5 2017/07/28 07:08:15",
-                "filename": "http://192.168.2.50/boot.php?mac=${netX}",
-            },
-            {
-                "interface": "wlp3s0",
-                "fixed-address": "192.168.2.74",
-                "filename": "http://192.168.2.50/boot.php?mac=${netX}",
-                "subnet-mask": "255.255.255.0",
-                "routers": "192.168.2.1",
-            },
-        ]
-        write_file(lease_file, content)
-        self.assertCountEqual(
-            expected, IscDhclient.parse_dhcp_lease_file(lease_file)
-        )
+        expected = {
+            "interface": "wlp3s0",
+            "fixed-address": "192.168.2.74",
+            "filename": "http://192.168.2.50/boot.php?mac=${netX}",
+            "subnet-mask": "255.255.255.0",
+            "routers": "192.168.2.1",
+        }
+        write_file(client.lease_file, content)
+        got = client.get_newest_lease("eth0")
+        self.assertCountEqual(got, expected)
 
 
+@pytest.mark.usefixtures("dhclient_exists")
 class TestDHCPRFC3442(CiTestCase):
     def test_parse_lease_finds_rfc3442_classless_static_routes(self):
-        """IscDhclient.parse_dhcp_lease_file returns
+        """IscDhclient().get_newest_lease() returns
         rfc3442-classless-static-routes.
         """
-        lease_file = self.tmp_path("leases")
-        content = dedent(
-            """
+        client = IscDhclient()
+        client.lease_file = self.tmp_path("leases")
+        write_file(
+            client.lease_file,
+            dedent(
+                """
             lease {
               interface "wlp3s0";
               fixed-address 192.168.2.74;
@@ -161,30 +154,27 @@ def test_parse_lease_finds_rfc3442_classless_static_routes(self):
               renew 4 2017/07/27 18:02:30;
               expire 5 2017/07/28 07:08:15;
             }
-        """
-        )
-        expected = [
-            {
-                "interface": "wlp3s0",
-                "fixed-address": "192.168.2.74",
-                "subnet-mask": "255.255.255.0",
-                "routers": "192.168.2.1",
-                "rfc3442-classless-static-routes": "0,130,56,240,1",
-                "renew": "4 2017/07/27 18:02:30",
-                "expire": "5 2017/07/28 07:08:15",
-            }
-        ]
-        write_file(lease_file, content)
-        self.assertCountEqual(
-            expected, IscDhclient.parse_dhcp_lease_file(lease_file)
+            """
+            ),
         )
+        expected = {
+            "interface": "wlp3s0",
+            "fixed-address": "192.168.2.74",
+            "subnet-mask": "255.255.255.0",
+            "routers": "192.168.2.1",
+            "rfc3442-classless-static-routes": "0,130,56,240,1",
+            "renew": "4 2017/07/27 18:02:30",
+            "expire": "5 2017/07/28 07:08:15",
+        }
+        self.assertCountEqual(expected, client.get_newest_lease("eth0"))
 
     def test_parse_lease_finds_classless_static_routes(self):
         """
-        IscDhclient.parse_dhcp_lease_file returns classless-static-routes
+        IscDhclient().get_newest_lease returns classless-static-routes
         for Centos lease format.
         """
-        lease_file = self.tmp_path("leases")
+        client = IscDhclient()
+        client.lease_file = self.tmp_path("leases")
         content = dedent(
             """
             lease {
@@ -198,38 +188,31 @@ def test_parse_lease_finds_classless_static_routes(self):
             }
         """
         )
-        expected = [
-            {
-                "interface": "wlp3s0",
-                "fixed-address": "192.168.2.74",
-                "subnet-mask": "255.255.255.0",
-                "routers": "192.168.2.1",
-                "classless-static-routes": "0 130.56.240.1",
-                "renew": "4 2017/07/27 18:02:30",
-                "expire": "5 2017/07/28 07:08:15",
-            }
-        ]
-        write_file(lease_file, content)
-        self.assertCountEqual(
-            expected, IscDhclient.parse_dhcp_lease_file(lease_file)
-        )
+        expected = {
+            "interface": "wlp3s0",
+            "fixed-address": "192.168.2.74",
+            "subnet-mask": "255.255.255.0",
+            "routers": "192.168.2.1",
+            "classless-static-routes": "0 130.56.240.1",
+            "renew": "4 2017/07/27 18:02:30",
+            "expire": "5 2017/07/28 07:08:15",
+        }
+        write_file(client.lease_file, content)
+        self.assertCountEqual(expected, client.get_newest_lease("eth0"))
 
     @mock.patch("cloudinit.net.ephemeral.EphemeralIPv4Network")
     @mock.patch("cloudinit.net.ephemeral.maybe_perform_dhcp_discovery")
     def test_obtain_lease_parses_static_routes(self, m_maybe, m_ipv4):
         """EphemeralDHPCv4 parses rfc3442 routes for EphemeralIPv4Network"""
-        lease = [
-            {
-                "interface": "wlp3s0",
-                "fixed-address": "192.168.2.74",
-                "subnet-mask": "255.255.255.0",
-                "routers": "192.168.2.1",
-                "rfc3442-classless-static-routes": "0,130,56,240,1",
-                "renew": "4 2017/07/27 18:02:30",
-                "expire": "5 2017/07/28 07:08:15",
-            }
-        ]
-        m_maybe.return_value = lease
+        m_maybe.return_value = {
+            "interface": "wlp3s0",
+            "fixed-address": "192.168.2.74",
+            "subnet-mask": "255.255.255.0",
+            "routers": "192.168.2.1",
+            "rfc3442-classless-static-routes": "0,130,56,240,1",
+            "renew": "4 2017/07/27 18:02:30",
+            "expire": "5 2017/07/28 07:08:15",
+        }
         distro = MockDistro()
         eph = EphemeralDHCPv4(distro)
         eph.obtain_lease()
@@ -250,18 +233,15 @@ def test_obtain_centos_lease_parses_static_routes(self, m_maybe, m_ipv4):
         EphemeralDHPCv4 parses rfc3442 routes for EphemeralIPv4Network
         for Centos Lease format
         """
-        lease = [
-            {
-                "interface": "wlp3s0",
-                "fixed-address": "192.168.2.74",
-                "subnet-mask": "255.255.255.0",
-                "routers": "192.168.2.1",
-                "classless-static-routes": "0 130.56.240.1",
-                "renew": "4 2017/07/27 18:02:30",
-                "expire": "5 2017/07/28 07:08:15",
-            }
-        ]
-        m_maybe.return_value = lease
+        m_maybe.return_value = {
+            "interface": "wlp3s0",
+            "fixed-address": "192.168.2.74",
+            "subnet-mask": "255.255.255.0",
+            "routers": "192.168.2.1",
+            "classless-static-routes": "0 130.56.240.1",
+            "renew": "4 2017/07/27 18:02:30",
+            "expire": "5 2017/07/28 07:08:15",
+        }
         distro = MockDistro()
         eph = EphemeralDHCPv4(distro)
         eph.obtain_lease()
@@ -395,20 +375,7 @@ class TestDHCPDiscoveryClean(CiTestCase):
     with_logs = True
     ib_address_prefix = "00:00:00:00:00:00:00:00:00:00:00:00"
 
-    @mock.patch("cloudinit.net.dhcp.find_fallback_nic")
-    def test_no_fallback_nic_found(self, m_fallback_nic):
-        """Log and do nothing when nic is absent and no fallback is found."""
-        m_fallback_nic.return_value = None  # No fallback nic found
-
-        with pytest.raises(NoDHCPLeaseInterfaceError):
-            maybe_perform_dhcp_discovery(MockDistro())
-
-        self.assertIn(
-            "Skip dhcp_discovery: Unable to find fallback nic.",
-            self.logs.getvalue(),
-        )
-
-    @mock.patch("cloudinit.net.dhcp.find_fallback_nic", return_value="eth9")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic", return_value="eth9")
     @mock.patch("cloudinit.net.dhcp.os.remove")
     @mock.patch("cloudinit.net.dhcp.subp.subp")
     @mock.patch("cloudinit.net.dhcp.subp.which")
@@ -422,27 +389,28 @@ def test_dhclient_exits_with_error(
         ]
 
         with pytest.raises(NoDHCPLeaseError):
-            maybe_perform_dhcp_discovery(MockDistro())
+            maybe_perform_dhcp_discovery(Distro("fake but not", {}, None))
 
         self.assertIn(
             "DHCP client selected: dhclient",
             self.logs.getvalue(),
         )
 
-    @mock.patch("cloudinit.net.dhcp.find_fallback_nic", return_value="eth9")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic", return_value="eth9")
     @mock.patch("cloudinit.net.dhcp.os.remove")
     @mock.patch("cloudinit.net.dhcp.subp.subp")
     @mock.patch("cloudinit.net.dhcp.subp.which")
     def test_dhcp_client_failover(self, m_which, m_subp, m_remove, m_fallback):
-        """Log and do nothing when nic is absent and no fallback is found."""
+        """Log and do nothing when nic is absent and no fallback client is
+        found."""
         m_subp.side_effect = [
             ("", ""),
             subp.ProcessExecutionError(exit_code=-5),
         ]
 
-        m_which.side_effect = [False, True]
+        m_which.side_effect = [False, False, False, False]
         with pytest.raises(NoDHCPLeaseError):
-            maybe_perform_dhcp_discovery(MockDistro())
+            maybe_perform_dhcp_discovery(Distro("somename", {}, None))
 
         self.assertIn(
             "DHCP client not found: dhclient",
@@ -452,30 +420,30 @@ def test_dhcp_client_failover(self, m_which, m_subp, m_remove, m_fallback):
             "DHCP client not found: dhcpcd",
             self.logs.getvalue(),
         )
-
-    @mock.patch("cloudinit.net.dhcp.find_fallback_nic", return_value=None)
-    def test_provided_nic_does_not_exist(self, m_fallback_nic):
-        """When the provided nic doesn't exist, log a message and no-op."""
-        with pytest.raises(NoDHCPLeaseInterfaceError):
-            maybe_perform_dhcp_discovery(MockDistro(), "idontexist")
-
         self.assertIn(
-            "Skip dhcp_discovery: nic idontexist not found in get_devicelist.",
+            "DHCP client not found: udhcpc",
             self.logs.getvalue(),
         )
 
     @mock.patch("cloudinit.net.dhcp.subp.which")
-    @mock.patch("cloudinit.net.dhcp.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     def test_absent_dhclient_command(self, m_fallback, m_which):
         """When dhclient doesn't exist in the OS, log the issue and no-op."""
         m_fallback.return_value = "eth9"
         m_which.return_value = None  # dhclient isn't found
-
         with pytest.raises(NoDHCPLeaseMissingDhclientError):
-            maybe_perform_dhcp_discovery(MockDistro())
+            maybe_perform_dhcp_discovery(Distro("whoa", {}, None))
 
         self.assertIn(
-            "Skip dhclient configuration: No dhclient command found.",
+            "DHCP client not found: dhclient",
+            self.logs.getvalue(),
+        )
+        self.assertIn(
+            "DHCP client not found: dhcpcd",
+            self.logs.getvalue(),
+        )
+        self.assertIn(
+            "DHCP client not found: udhcpc",
             self.logs.getvalue(),
         )
 
@@ -509,15 +477,13 @@ def test_dhcp_discovery_warns_invalid_pid(
             "cloudinit.util.load_file", return_value=lease_content
         ):
             self.assertCountEqual(
-                [
-                    {
-                        "interface": "eth9",
-                        "fixed-address": "192.168.2.74",
-                        "subnet-mask": "255.255.255.0",
-                        "routers": "192.168.2.1",
-                    }
-                ],
-                IscDhclient.parse_dhcp_lease_file("lease"),
+                {
+                    "interface": "eth9",
+                    "fixed-address": "192.168.2.74",
+                    "subnet-mask": "255.255.255.0",
+                    "routers": "192.168.2.1",
+                },
+                IscDhclient().get_newest_lease("eth0"),
             )
         with self.assertRaises(InvalidDHCPLeaseFileError):
             with mock.patch("cloudinit.util.load_file", return_value=""):
@@ -545,7 +511,7 @@ def test_dhcp_discovery_waits_on_lease_and_pid(
         m_wait.return_value = [PID_F]  # Return the missing pidfile wait for
         m_getppid.return_value = 1  # Indicate that dhclient has daemonized
         self.assertEqual(
-            [], IscDhclient().dhcp_discovery("eth9", distro=MockDistro())
+            {}, IscDhclient().dhcp_discovery("eth9", distro=MockDistro())
         )
         self.assertEqual(
             mock.call([PID_F, LEASE_F], maxwait=5, naplen=0.01),
@@ -596,14 +562,12 @@ def test_dhcp_discovery(
             "cloudinit.util.load_file", side_effect=["1", lease_content]
         ):
             self.assertCountEqual(
-                [
-                    {
-                        "interface": "eth9",
-                        "fixed-address": "192.168.2.74",
-                        "subnet-mask": "255.255.255.0",
-                        "routers": "192.168.2.1",
-                    }
-                ],
+                {
+                    "interface": "eth9",
+                    "fixed-address": "192.168.2.74",
+                    "subnet-mask": "255.255.255.0",
+                    "routers": "192.168.2.1",
+                },
                 IscDhclient().dhcp_discovery("eth9", distro=MockDistro()),
             )
         # Interface was brought up before dhclient called
@@ -676,14 +640,12 @@ def test_dhcp_discovery_ib(
             "cloudinit.util.load_file", side_effect=["1", lease_content]
         ):
             self.assertCountEqual(
-                [
-                    {
-                        "interface": "ib0",
-                        "fixed-address": "192.168.2.74",
-                        "subnet-mask": "255.255.255.0",
-                        "routers": "192.168.2.1",
-                    }
-                ],
+                {
+                    "interface": "ib0",
+                    "fixed-address": "192.168.2.74",
+                    "subnet-mask": "255.255.255.0",
+                    "routers": "192.168.2.1",
+                },
                 IscDhclient().dhcp_discovery("ib0", distro=MockDistro()),
             )
         # Interface was brought up before dhclient called
@@ -901,12 +863,11 @@ def test_ephemeral_dhcp_setup_network_if_url_connectivity(
     ):
         """No EphemeralDhcp4 network setup when connectivity_url succeeds."""
         url = "http://example.org/index.html"
-        fake_lease = {
+        m_dhcp.return_value = {
             "interface": "eth9",
             "fixed-address": "192.168.2.2",
             "subnet-mask": "255.255.0.0",
         }
-        m_dhcp.return_value = [fake_lease]
         m_subp.return_value = ("", "")
 
         self.responses.add(responses.GET, url, body=b"", status=404)
@@ -914,7 +875,7 @@ def test_ephemeral_dhcp_setup_network_if_url_connectivity(
             MockDistro(),
             connectivity_url_data={"url": url},
         ) as lease:
-            self.assertEqual(fake_lease, lease)
+            self.assertEqual(m_dhcp.return_value, lease)
         # Ensure that dhcp discovery occurs
         m_dhcp.assert_called_once()
 
@@ -977,24 +938,6 @@ class TestUDHCPCDiscoveryClean(CiTestCase):
     with_logs = True
     maxDiff = None
 
-    @mock.patch("cloudinit.net.dhcp.subp.which")
-    @mock.patch("cloudinit.net.dhcp.find_fallback_nic")
-    def test_absent_udhcpc_command(self, m_fallback, m_which):
-        """When dhclient doesn't exist in the OS, log the issue and no-op."""
-        m_fallback.return_value = "eth9"
-        m_which.return_value = None  # udhcpc isn't found
-
-        distro = MockDistro()
-        distro.dhcp_client_priority = [Udhcpc]
-
-        with pytest.raises(NoDHCPLeaseMissingDhclientError):
-            maybe_perform_dhcp_discovery(distro)
-
-        self.assertIn(
-            "Skip udhcpc configuration: No udhcpc command found.",
-            self.logs.getvalue(),
-        )
-
     @mock.patch("cloudinit.net.dhcp.is_ib_interface", return_value=False)
     @mock.patch("cloudinit.net.dhcp.subp.which", return_value="/sbin/udhcpc")
     @mock.patch("cloudinit.net.dhcp.os.remove")
@@ -1022,18 +965,13 @@ def test_udhcpc_discovery(
             "static_routes": "10.240.0.1/32 0.0.0.0 0.0.0.0/0 10.240.0.1",
         }
         self.assertEqual(
-            [
-                {
-                    "fixed-address": "192.168.2.74",
-                    "interface": "eth9",
-                    "routers": "192.168.2.1",
-                    "static_routes": [
-                        ("10.240.0.1/32", "0.0.0.0"),
-                        ("0.0.0.0/0", "10.240.0.1"),
-                    ],
-                    "subnet-mask": "255.255.255.0",
-                }
-            ],
+            {
+                "fixed-address": "192.168.2.74",
+                "interface": "eth9",
+                "routers": "192.168.2.1",
+                "static_routes": "10.240.0.1/32 0.0.0.0 0.0.0.0/0 10.240.0.1",
+                "subnet-mask": "255.255.255.0",
+            },
             Udhcpc().dhcp_discovery("eth9", distro=MockDistro()),
         )
         # Interface was brought up before dhclient called
@@ -1094,18 +1032,13 @@ def test_udhcpc_discovery_ib(
         }
         m_get_ib_interface_hwaddr.return_value = "00:21:28:00:01:cf:4b:01"
         self.assertEqual(
-            [
-                {
-                    "fixed-address": "192.168.2.74",
-                    "interface": "ib0",
-                    "routers": "192.168.2.1",
-                    "static_routes": [
-                        ("10.240.0.1/32", "0.0.0.0"),
-                        ("0.0.0.0/0", "10.240.0.1"),
-                    ],
-                    "subnet-mask": "255.255.255.0",
-                }
-            ],
+            {
+                "fixed-address": "192.168.2.74",
+                "interface": "ib0",
+                "routers": "192.168.2.1",
+                "static_routes": "10.240.0.1/32 0.0.0.0 0.0.0.0/0 10.240.0.1",
+                "subnet-mask": "255.255.255.0",
+            },
             Udhcpc().dhcp_discovery("ib0", distro=MockDistro()),
         )
         # Interface was brought up before dhclient called
@@ -1150,16 +1083,13 @@ class TestISCDHClient(CiTestCase):
         ),
     )
     @mock.patch("os.path.getmtime", return_value=123.45)
-    def test_get_latest_lease_rhel(self, *_):
+    def test_get_newest_lease_file_from_distro_rhel(self, *_):
         """
         Test that an rhel style lease has been found
         """
         self.assertEqual(
             "/var/lib/NetworkManager/dhclient-0-u-u-i-d-enp2s0f0.lease",
-            IscDhclient.get_latest_lease(
-                rhel.Distro.dhclient_lease_directory,
-                rhel.Distro.dhclient_lease_file_regex,
-            ),
+            IscDhclient.get_newest_lease_file_from_distro(rhel.Distro),
         )
 
     @mock.patch(
@@ -1172,16 +1102,13 @@ def test_get_latest_lease_rhel(self, *_):
         ),
     )
     @mock.patch("os.path.getmtime", return_value=123.45)
-    def test_get_latest_lease_amazonlinux(self, *_):
+    def test_get_newest_lease_file_from_distro_amazonlinux(self, *_):
         """
         Test that an amazon style lease has been found
         """
         self.assertEqual(
             "/var/lib/dhcp/dhclient--eth0.leases",
-            IscDhclient.get_latest_lease(
-                amazon.Distro.dhclient_lease_directory,
-                amazon.Distro.dhclient_lease_file_regex,
-            ),
+            IscDhclient.get_newest_lease_file_from_distro(amazon.Distro),
         )
 
     @mock.patch(
@@ -1194,16 +1121,13 @@ def test_get_latest_lease_amazonlinux(self, *_):
         ),
     )
     @mock.patch("os.path.getmtime", return_value=123.45)
-    def test_get_latest_lease_freebsd(self, *_):
+    def test_get_newest_lease_file_from_distro_freebsd(self, *_):
         """
         Test that an freebsd style lease has been found
         """
         self.assertEqual(
             "/var/db/dhclient.leases.vtynet0",
-            IscDhclient.get_latest_lease(
-                freebsd.Distro.dhclient_lease_directory,
-                freebsd.Distro.dhclient_lease_file_regex,
-            ),
+            IscDhclient.get_newest_lease_file_from_distro(freebsd.Distro),
         )
 
     @mock.patch(
@@ -1216,16 +1140,13 @@ def test_get_latest_lease_freebsd(self, *_):
         ),
     )
     @mock.patch("os.path.getmtime", return_value=123.45)
-    def test_get_latest_lease_alpine(self, *_):
+    def test_get_newest_lease_file_from_distro_alpine(self, *_):
         """
         Test that an alpine style lease has been found
         """
         self.assertEqual(
             "/var/lib/dhcp/dhclient.leases",
-            IscDhclient.get_latest_lease(
-                alpine.Distro.dhclient_lease_directory,
-                alpine.Distro.dhclient_lease_file_regex,
-            ),
+            IscDhclient.get_newest_lease_file_from_distro(alpine.Distro),
         )
 
     @mock.patch(
@@ -1238,40 +1159,13 @@ def test_get_latest_lease_alpine(self, *_):
         ),
     )
     @mock.patch("os.path.getmtime", return_value=123.45)
-    def test_get_latest_lease_debian(self, *_):
+    def test_get_newest_lease_file_from_distro_debian(self, *_):
         """
         Test that an debian style lease has been found
         """
         self.assertEqual(
             "/var/lib/dhcp/dhclient.eth0.leases",
-            IscDhclient.get_latest_lease(
-                debian.Distro.dhclient_lease_directory,
-                debian.Distro.dhclient_lease_file_regex,
-            ),
-        )
-
-    @mock.patch(
-        "os.listdir",
-        return_value=(
-            "some_file",
-            "!@#$-eth0.lease",
-            "some_other_file",
-        ),
-    )
-    @mock.patch("os.path.getmtime", return_value=123.45)
-    def test_no_distro_hints_fallback(self, *_):
-        """
-        This tests a situation where Distro doesn't provide
-        hints for dhclient leases.
-        The code should resort to hardcoded lease location
-        """
-        # Provide lease_dir and regex as None
-        self.assertEqual(
-            os.path.join(DHCLIENT_FALLBACK_LEASE_DIR, "!@#$-eth0.lease"),
-            IscDhclient.get_latest_lease(
-                None,
-                None,
-            ),
+            IscDhclient.get_newest_lease_file_from_distro(debian.Distro),
         )
 
     # If argument to listdir is '/var/lib/NetworkManager'
@@ -1291,9 +1185,8 @@ def test_fallback_when_nothing_found(self, *_):
         """
         self.assertEqual(
             os.path.join(DHCLIENT_FALLBACK_LEASE_DIR, "!@#$-eth0.lease"),
-            IscDhclient.get_latest_lease(
-                rhel.Distro.dhclient_lease_directory,
-                rhel.Distro.dhclient_lease_file_regex,
+            IscDhclient.get_newest_lease_file_from_distro(
+                rhel.Distro("", {}, {})
             ),
         )
 
@@ -1306,15 +1199,67 @@ def test_fallback_when_nothing_found(self, *_):
         ),
     )
     @mock.patch("os.path.getmtime", return_value=123.45)
-    def test_get_latest_lease_notfound(self, *_):
+    def test_get_newest_lease_file_from_distro_notfound(self, *_):
         """
         Test the case when no leases were found
         """
         # Any Distro would suffice for the absense test, choose Centos then.
         self.assertEqual(
             None,
-            IscDhclient.get_latest_lease(
-                centos.Distro.dhclient_lease_directory,
-                centos.Distro.dhclient_lease_file_regex,
-            ),
+            IscDhclient.get_newest_lease_file_from_distro(centos.Distro),
+        )
+
+
+class TestDhcpcd:
+    def test_parse_lease(self):
+        lease = dedent(
+            """
+            broadcast_address='192.168.15.255'
+            dhcp_lease_time='3600'
+            dhcp_message_type='5'
+            dhcp_server_identifier='192.168.0.1'
+            domain_name='us-east-2.compute.internal'
+            domain_name_servers='192.168.0.2'
+            host_name='ip-192-168-0-212'
+            interface_mtu='9001'
+            ip_address='192.168.0.212'
+            network_number='192.168.0.0'
+            routers='192.168.0.1'
+            subnet_cidr='20'
+            subnet_mask='255.255.240.0'
+            """
+        )
+        parsed_lease = Dhcpcd.parse_dhcpcd_lease(lease, "eth0")
+        assert "eth0" == parsed_lease["interface"]
+        assert "192.168.15.255" == parsed_lease["broadcast-address"]
+        assert "192.168.0.212" == parsed_lease["fixed-address"]
+        assert "255.255.240.0" == parsed_lease["subnet-mask"]
+        assert "192.168.0.1" == parsed_lease["routers"]
+
+    def test_parse_classless_static_routes(self):
+        lease = dedent(
+            """
+            broadcast_address='10.0.0.255'
+            classless_static_routes='0.0.0.0/0 10.0.0.1 168.63.129.16/32"""
+            """ 10.0.0.1 169.254.169.254/32 10.0.0.1'
+            dhcp_lease_time='4294967295'
+            dhcp_message_type='5'
+            dhcp_rebinding_time='4294967295'
+            dhcp_renewal_time='4294967295'
+            dhcp_server_identifier='168.63.129.16'
+            domain_name='ilo2tr0xng2exgucxg20yx0tjb.gx.internal.cloudapp.net'
+            domain_name_servers='168.63.129.16'
+            ip_address='10.0.0.5'
+            network_number='10.0.0.0'
+            routers='10.0.0.1'
+            server_name='DSM111070915004'
+            subnet_cidr='24'
+            subnet_mask='255.255.255.0'
+            """
         )
+        parsed_lease = Dhcpcd.parse_dhcpcd_lease(lease, "eth0")
+        assert [
+            ("0.0.0.0/0", "10.0.0.1"),
+            ("168.63.129.16/32", "10.0.0.1"),
+            ("169.254.169.254/32", "10.0.0.1"),
+        ] == Dhcpcd.parse_static_routes(parsed_lease["static_routes"])
diff --git a/tests/unittests/sources/test_aliyun.py b/tests/unittests/sources/test_aliyun.py
index 3d65462cbb19..00da2ee88e4a 100644
--- a/tests/unittests/sources/test_aliyun.py
+++ b/tests/unittests/sources/test_aliyun.py
@@ -198,15 +198,13 @@ def test_aliyun_local_with_mock_server(
     ):
         m_is_aliyun.return_value = True
         m_fallback_nic.return_value = "eth9"
-        m_dhcp.return_value = [
-            {
-                "interface": "eth9",
-                "fixed-address": "192.168.2.9",
-                "routers": "192.168.2.1",
-                "subnet-mask": "255.255.255.0",
-                "broadcast-address": "192.168.2.255",
-            }
-        ]
+        m_dhcp.return_value = {
+            "interface": "eth9",
+            "fixed-address": "192.168.2.9",
+            "routers": "192.168.2.1",
+            "subnet-mask": "255.255.255.0",
+            "broadcast-address": "192.168.2.255",
+        }
         m_is_bsd.return_value = False
         cfg = {"datasource": {"AliYun": {"timeout": "1", "max_wait": "1"}}}
         distro = mock.MagicMock()
diff --git a/tests/unittests/sources/test_azure.py b/tests/unittests/sources/test_azure.py
index 2a477f80239e..2d85912b0ab2 100644
--- a/tests/unittests/sources/test_azure.py
+++ b/tests/unittests/sources/test_azure.py
@@ -193,15 +193,13 @@ def mock_kvp_report_success_to_host():
 def mock_net_dhcp_maybe_perform_dhcp_discovery():
     with mock.patch(
         "cloudinit.net.ephemeral.maybe_perform_dhcp_discovery",
-        return_value=[
-            {
-                "unknown-245": "0a:0b:0c:0d",
-                "interface": "ethBoot0",
-                "fixed-address": "192.168.2.9",
-                "routers": "192.168.2.1",
-                "subnet-mask": "255.255.255.0",
-            }
-        ],
+        return_value={
+            "unknown-245": "0a:0b:0c:0d",
+            "interface": "ethBoot0",
+            "fixed-address": "192.168.2.9",
+            "routers": "192.168.2.1",
+            "subnet-mask": "255.255.255.0",
+        },
         autospec=True,
     ) as m:
         yield m
diff --git a/tests/unittests/sources/test_cloudstack.py b/tests/unittests/sources/test_cloudstack.py
index cc511da1c7e9..c3767dea5616 100644
--- a/tests/unittests/sources/test_cloudstack.py
+++ b/tests/unittests/sources/test_cloudstack.py
@@ -1,9 +1,14 @@
 # This file is part of cloud-init. See LICENSE file for license information.
+from textwrap import dedent
+
+import pytest
+
 from cloudinit import helpers
 from cloudinit.distros import rhel, ubuntu
 from cloudinit.sources import DataSourceHostname
 from cloudinit.sources.DataSourceCloudStack import DataSourceCloudStack
 from tests.unittests.helpers import CiTestCase, ExitStack, mock
+from tests.unittests.util import MockDistro
 
 SOURCES_PATH = "cloudinit.sources"
 MOD_PATH = SOURCES_PATH + ".DataSourceCloudStack"
@@ -11,6 +16,7 @@
 DHCP_MOD_PATH = "cloudinit.net.dhcp"
 
 
+@pytest.mark.usefixtures("dhclient_exists")
 class TestCloudStackHostname(CiTestCase):
     def setUp(self):
         super(TestCloudStackHostname, self).setUp()
@@ -30,6 +36,36 @@ def setUp(self):
                 SOURCES_PATH + ".DataSource.get_hostname", get_hostname_parent
             )
         )
+        self.patches.enter_context(
+            mock.patch(
+                DHCP_MOD_PATH + ".util.load_file",
+                return_value=dedent(
+                    """
+                    lease {
+                      interface "eth0";
+                      fixed-address 10.0.0.5;
+                      server-name "DSM111070915004";
+                      option subnet-mask 255.255.255.0;
+                      option dhcp-lease-time 4294967295;
+                      option routers 10.0.0.1;
+                      option dhcp-message-type 5;
+                      option dhcp-server-identifier 168.63.129.16;
+                      option domain-name-servers 168.63.129.16;
+                      option dhcp-renewal-time 4294967295;
+                      option rfc3442-classless-static-routes """
+                    """0,10,0,0,1,32,168,63,129,16,10,0,0,1,32,169,254,"""
+                    """169,254,10,0,0,1;
+                      option unknown-245 a8:3f:81:10;
+                      option dhcp-rebinding-time 4294967295;
+                    """
+                    """renew 0 2160/02/17 02:22:33;
+                      rebind 0 2160/02/17 02:22:33;
+                      expire 0 2160/02/17 02:22:33;
+                    }
+                    """
+                ),
+            )
+        )
 
         # Mock cloudinit.net.dhcp.networkd_get_option_from_leases() method \
         # result since we don't have a DHCP client running
@@ -43,38 +79,43 @@ def setUp(self):
             )
         )
 
-        # Mock cloudinit.net.dhcp.get_latest_lease() method \
+        # Mock cloudinit.net.dhcp.get_newest_lease_file_from_distro() method \
         # result since we don't have a DHCP client running
-        isc_dhclient_get_latest_lease = mock.MagicMock(
+        isc_dhclient_get_newest_lease_file_from_distro = mock.MagicMock(
             return_value="/var/lib/NetworkManager/dhclient-u-u-i-d-eth0.lease"
         )
         self.patches.enter_context(
             mock.patch(
-                DHCP_MOD_PATH + ".IscDhclient.get_latest_lease",
-                isc_dhclient_get_latest_lease,
+                DHCP_MOD_PATH
+                + ".IscDhclient.get_newest_lease_file_from_distro",
+                isc_dhclient_get_newest_lease_file_from_distro,
             )
         )
 
         # Mock cloudinit.net.dhcp.networkd_get_option_from_leases() method \
         # result since we don't have a DHCP client running
-        parse_dhcp_lease_file = mock.MagicMock(
-            return_value=[
-                {
-                    "interface": "eth0",
-                    "fixed-address": "192.168.0.1",
-                    "subnet-mask": "255.255.255.0",
-                    "routers": "192.168.0.1",
-                    "domain-name": self.isc_dhclient_domainname,
-                    "renew": "4 2017/07/27 18:02:30",
-                    "expire": "5 2017/07/28 07:08:15",
-                }
-            ]
+        lease = {
+            "interface": "eth0",
+            "fixed-address": "192.168.0.1",
+            "subnet-mask": "255.255.255.0",
+            "routers": "192.168.0.1",
+            "domain-name": self.isc_dhclient_domainname,
+            "renew": "4 2017/07/27 18:02:30",
+            "expire": "5 2017/07/28 07:08:15",
+        }
+        get_newest_lease = mock.MagicMock(return_value=lease)
+
+        self.patches.enter_context(
+            mock.patch(
+                DHCP_MOD_PATH + ".IscDhclient.get_newest_lease",
+                get_newest_lease,
+            )
         )
 
         self.patches.enter_context(
             mock.patch(
-                DHCP_MOD_PATH + ".IscDhclient.parse_dhcp_lease_file",
-                parse_dhcp_lease_file,
+                DHCP_MOD_PATH + ".IscDhclient.parse_leases",
+                mock.MagicMock(return_value=[lease]),
             )
         )
 
@@ -119,7 +160,36 @@ def test_get_domainname_isc_dhclient(self):
         ds = DataSourceCloudStack(
             {}, rhel.Distro, helpers.Paths({"run_dir": self.tmp})
         )
-        result = ds._get_domainname()
+        with mock.patch(
+            MOD_PATH + ".util.load_file",
+            return_value=dedent(
+                """
+                lease {
+                  interface "eth0";
+                  fixed-address 10.0.0.5;
+                  server-name "DSM111070915004";
+                  option subnet-mask 255.255.255.0;
+                  option dhcp-lease-time 4294967295;
+                  option routers 10.0.0.1;
+                  option dhcp-message-type 5;
+                  option dhcp-server-identifier 168.63.129.16;
+                  option domain-name-servers 168.63.129.16;
+                  option dhcp-renewal-time 4294967295;
+                  option rfc3442-classless-static-routes """
+                """0,10,0,0,1,32,168,63,129,16,10,0,0,1,32,169,254,"""
+                """169,254,10,0,0,1;
+                  option unknown-245 a8:3f:81:10;
+                  option dhcp-rebinding-time 4294967295;
+                """
+                f"option domain-name {self.isc_dhclient_domainname};"
+                """renew 0 2160/02/17 02:22:33;
+                  rebind 0 2160/02/17 02:22:33;
+                  expire 0 2160/02/17 02:22:33;
+                }
+                """
+            ),
+        ):
+            result = ds._get_domainname()
         self.assertEqual(self.isc_dhclient_domainname, result)
 
     def test_get_hostname_non_fqdn(self):
@@ -177,35 +247,82 @@ def test_get_hostname_fqdn_fallback(self):
             )
         )
 
-        # Override IscDhclient.parse_dhcp_lease_file()
-        # to return a lease without domain-name option.
-        parse_dhcp_lease_file = mock.MagicMock(
-            return_value=[
-                {
+        self.patches.enter_context(
+            mock.patch(
+                "cloudinit.distros.net.find_fallback_nic",
+                return_value="eth0",
+            )
+        )
+
+        self.patches.enter_context(
+            mock.patch(
+                MOD_PATH
+                + ".dhcp.IscDhclient.get_newest_lease_file_from_distro",
+                return_value=True,
+            )
+        )
+
+        self.patches.enter_context(
+            mock.patch(
+                MOD_PATH + ".dhcp.IscDhclient.parse_leases", return_value=[]
+            )
+        )
+
+        self.patches.enter_context(
+            mock.patch(
+                DHCP_MOD_PATH + ".IscDhclient.get_newest_lease",
+                return_value={
                     "interface": "eth0",
                     "fixed-address": "192.168.0.1",
                     "subnet-mask": "255.255.255.0",
                     "routers": "192.168.0.1",
                     "renew": "4 2017/07/27 18:02:30",
                     "expire": "5 2017/07/28 07:08:15",
-                }
-            ]
+                },
+            )
         )
 
         self.patches.enter_context(
             mock.patch(
-                DHCP_MOD_PATH + ".IscDhclient.parse_dhcp_lease_file",
-                parse_dhcp_lease_file,
+                DHCP_MOD_PATH + ".util.load_file",
+                return_value=dedent(
+                    """
+                    lease {
+                      interface "eth0";
+                      fixed-address 10.0.0.5;
+                      server-name "DSM111070915004";
+                      option subnet-mask 255.255.255.0;
+                      option dhcp-lease-time 4294967295;
+                      option routers 10.0.0.1;
+                      option dhcp-message-type 5;
+                      option dhcp-server-identifier 168.63.129.16;
+                      option domain-name-servers 168.63.129.16;
+                      option dhcp-renewal-time 4294967295;
+                      option rfc3442-classless-static-routes """
+                    """0,10,0,0,1,32,168,63,129,16,10,0,0,1,32,169,254,"""
+                    """169,254,10,0,0,1;
+                      option unknown-245 a8:3f:81:10;
+                      option dhcp-rebinding-time 4294967295;
+                    """
+                    """renew 0 2160/02/17 02:22:33;
+                      rebind 0 2160/02/17 02:22:33;
+                      expire 0 2160/02/17 02:22:33;
+                    }
+                    """
+                ),
             )
         )
 
         ds = DataSourceCloudStack(
-            {}, ubuntu.Distro, helpers.Paths({"run_dir": self.tmp})
+            {}, ubuntu.Distro("", {}, {}), helpers.Paths({"run_dir": self.tmp})
         )
-        result = ds.get_hostname(fqdn=True)
-        self.assertTupleEqual(expected, result)
+        ds.distro.fallback_interface = "eth0"
+        with mock.patch(MOD_PATH + ".util.load_file"):
+            result = ds.get_hostname(fqdn=True)
+            self.assertTupleEqual(expected, result)
 
 
+@pytest.mark.usefixtures("dhclient_exists")
 class TestCloudStackPasswordFetching(CiTestCase):
     def setUp(self):
         super(TestCloudStackPasswordFetching, self).setUp()
@@ -216,11 +333,25 @@ def setUp(self):
         self.patches.enter_context(mock.patch("{0}.uhelp".format(mod_name)))
         default_gw = "192.201.20.0"
 
-        get_latest_lease = mock.MagicMock(return_value=None)
+        get_newest_lease_file_from_distro = mock.MagicMock(return_value=None)
+        self.patches.enter_context(
+            mock.patch(
+                DHCP_MOD_PATH + ".IscDhclient.get_newest_lease",
+                return_value={
+                    "interface": "eth0",
+                    "fixed-address": "192.168.0.1",
+                    "subnet-mask": "255.255.255.0",
+                    "routers": "192.168.0.1",
+                    "renew": "4 2017/07/27 18:02:30",
+                    "expire": "5 2017/07/28 07:08:15",
+                },
+            )
+        )
         self.patches.enter_context(
             mock.patch(
-                DHCP_MOD_PATH + ".IscDhclient.get_latest_lease",
-                get_latest_lease,
+                DHCP_MOD_PATH
+                + ".IscDhclient.get_newest_lease_file_from_distro",
+                get_newest_lease_file_from_distro,
             )
         )
 
@@ -255,7 +386,7 @@ def _set_password_server_response(self, response_string):
     def test_empty_password_doesnt_create_config(self):
         self._set_password_server_response("")
         ds = DataSourceCloudStack(
-            {}, ubuntu.Distro, helpers.Paths({"run_dir": self.tmp})
+            {}, MockDistro(), helpers.Paths({"run_dir": self.tmp})
         )
         ds.get_data()
         self.assertEqual({}, ds.get_config_obj())
@@ -263,7 +394,7 @@ def test_empty_password_doesnt_create_config(self):
     def test_saved_password_doesnt_create_config(self):
         self._set_password_server_response("saved_password")
         ds = DataSourceCloudStack(
-            {}, ubuntu.Distro, helpers.Paths({"run_dir": self.tmp})
+            {}, MockDistro(), helpers.Paths({"run_dir": self.tmp})
         )
         ds.get_data()
         self.assertEqual({}, ds.get_config_obj())
@@ -274,7 +405,7 @@ def test_password_sets_password(self, m_wait):
         password = "SekritSquirrel"
         self._set_password_server_response(password)
         ds = DataSourceCloudStack(
-            {}, ubuntu.Distro, helpers.Paths({"run_dir": self.tmp})
+            {}, MockDistro(), helpers.Paths({"run_dir": self.tmp})
         )
         ds.get_data()
         self.assertEqual(password, ds.get_config_obj()["password"])
@@ -283,8 +414,9 @@ def test_password_sets_password(self, m_wait):
     def test_bad_request_doesnt_stop_ds_from_working(self, m_wait):
         m_wait.return_value = True
         self._set_password_server_response("bad_request")
+        # with mock.patch(DHCP_MOD_PATH + ".util.load_file"):
         ds = DataSourceCloudStack(
-            {}, ubuntu.Distro, helpers.Paths({"run_dir": self.tmp})
+            {}, MockDistro(), helpers.Paths({"run_dir": self.tmp})
         )
         self.assertTrue(ds.get_data())
 
@@ -303,7 +435,7 @@ def test_valid_response_means_password_marked_as_saved(self, m_wait):
         password = "SekritSquirrel"
         subp = self._set_password_server_response(password)
         ds = DataSourceCloudStack(
-            {}, ubuntu.Distro, helpers.Paths({"run_dir": self.tmp})
+            {}, MockDistro(), helpers.Paths({"run_dir": self.tmp})
         )
         ds.get_data()
         self.assertRequestTypesSent(
@@ -313,7 +445,7 @@ def test_valid_response_means_password_marked_as_saved(self, m_wait):
     def _check_password_not_saved_for(self, response_string):
         subp = self._set_password_server_response(response_string)
         ds = DataSourceCloudStack(
-            {}, ubuntu.Distro, helpers.Paths({"run_dir": self.tmp})
+            {}, MockDistro(), helpers.Paths({"run_dir": self.tmp})
         )
         with mock.patch(DS_PATH + ".wait_for_metadata_service") as m_wait:
             m_wait.return_value = True
diff --git a/tests/unittests/sources/test_ec2.py b/tests/unittests/sources/test_ec2.py
index ea8621a6523f..6f1597cc92b2 100644
--- a/tests/unittests/sources/test_ec2.py
+++ b/tests/unittests/sources/test_ec2.py
@@ -10,6 +10,7 @@
 import responses
 
 from cloudinit import helpers
+from cloudinit.distros import ubuntu
 from cloudinit.sources import DataSourceEc2 as ec2
 from tests.unittests import helpers as test_helpers
 
@@ -342,9 +343,11 @@ def _patch_add_cleanup(self, mpath, *args, **kwargs):
         p.start()
         self.addCleanup(p.stop)
 
-    def _setup_ds(self, sys_cfg, platform_data, md, md_version=None):
+    def _setup_ds(
+        self, sys_cfg, platform_data, md, md_version=None, distro=None
+    ):
         self.uris = []
-        distro = mock.MagicMock()
+        distro = distro or mock.MagicMock()
         distro.get_tmp_exec_path = self.tmp_dir
         paths = helpers.Paths({"run_dir": self.tmp})
         if sys_cfg is None:
@@ -577,7 +580,7 @@ def test_network_config_cached_property_refreshed_on_upgrade(self, m_dhcp):
         )
         mac1 = "06:17:04:d7:26:09"  # Defined in DEFAULT_METADATA
         get_interface_mac_path = M_PATH_NET + "get_interfaces_by_mac"
-        ds.fallback_nic = "eth9"
+        ds.distro.fallback_nic = "eth9"
         with mock.patch(get_interface_mac_path) as m_get_interfaces_by_mac:
             m_get_interfaces_by_mac.return_value = {mac1: "eth9"}
             nc = ds.network_config  # Will re-crawl network metadata
@@ -846,7 +849,7 @@ def test_ec2_local_returns_false_on_bsd(self, m_is_freebsd):
 
     @mock.patch("cloudinit.net.ephemeral.EphemeralIPv6Network")
     @mock.patch("cloudinit.net.ephemeral.EphemeralIPv4Network")
-    @mock.patch("cloudinit.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.net.ephemeral.maybe_perform_dhcp_discovery")
     @mock.patch("cloudinit.sources.DataSourceEc2.util.is_FreeBSD")
     def test_ec2_local_performs_dhcp_on_non_bsd(
@@ -861,20 +864,19 @@ def test_ec2_local_performs_dhcp_on_non_bsd(
 
         m_fallback_nic.return_value = "eth9"
         m_is_bsd.return_value = False
-        m_dhcp.return_value = [
-            {
-                "interface": "eth9",
-                "fixed-address": "192.168.2.9",
-                "routers": "192.168.2.1",
-                "subnet-mask": "255.255.255.0",
-                "broadcast-address": "192.168.2.255",
-            }
-        ]
+        m_dhcp.return_value = {
+            "interface": "eth9",
+            "fixed-address": "192.168.2.9",
+            "routers": "192.168.2.1",
+            "subnet-mask": "255.255.255.0",
+            "broadcast-address": "192.168.2.255",
+        }
         self.datasource = ec2.DataSourceEc2Local
         ds = self._setup_ds(
             platform_data=self.valid_platform_data,
             sys_cfg={"datasource": {"Ec2": {"strict_id": False}}},
             md={"md": DEFAULT_METADATA},
+            distro=ubuntu.Distro("", {}, {}),
         )
 
         ret = ds.get_data()
diff --git a/tests/unittests/sources/test_gce.py b/tests/unittests/sources/test_gce.py
index c0b19d3cd68a..6fc31ddc76f8 100644
--- a/tests/unittests/sources/test_gce.py
+++ b/tests/unittests/sources/test_gce.py
@@ -405,9 +405,8 @@ def test_publish_host_keys(self, m_readurl):
         autospec=True,
     )
     @mock.patch(M_PATH + "net.find_candidate_nics", return_value=["ens4"])
-    @mock.patch(M_PATH + "DataSourceGCELocal.fallback_interface")
     def test_local_datasource_uses_ephemeral_dhcp(
-        self, _m_fallback, _m_find_candidate_nics, m_dhcp
+        self, _m_find_candidate_nics, m_dhcp
     ):
         self._set_mock_metadata()
         distro = mock.MagicMock()
@@ -424,9 +423,8 @@ def test_local_datasource_uses_ephemeral_dhcp(
         autospec=True,
     )
     @mock.patch(M_PATH + "net.find_candidate_nics")
-    @mock.patch(M_PATH + "DataSourceGCELocal.fallback_interface")
     def test_local_datasource_tries_on_multi_nic(
-        self, _m_fallback, m_find_candidate_nics, m_dhcp, m_read_md
+        self, m_find_candidate_nics, m_dhcp, m_read_md
     ):
         self._set_mock_metadata()
         distro = mock.MagicMock()
@@ -464,7 +462,7 @@ def test_local_datasource_tries_on_multi_nic(
             mock.call(distro, iface="ens0p5"),
             mock.call(distro, iface="ens0p6"),
         ]
-        assert ds._fallback_interface == "ens0p6"
+        assert ds.distro.fallback_interface == "ens0p6"
         assert ds.metadata == "md"
         assert ds.userdata_raw == "ud"
 
diff --git a/tests/unittests/sources/test_init.py b/tests/unittests/sources/test_init.py
index 44d63b8164ea..54ffebb53da4 100644
--- a/tests/unittests/sources/test_init.py
+++ b/tests/unittests/sources/test_init.py
@@ -6,6 +6,7 @@
 import stat
 
 from cloudinit import importer, util
+from cloudinit.distros import ubuntu
 from cloudinit.event import EventScope, EventType
 from cloudinit.helpers import Paths
 from cloudinit.sources import (
@@ -73,7 +74,7 @@ class TestDataSource(CiTestCase):
     def setUp(self):
         super(TestDataSource, self).setUp()
         self.sys_cfg = {"datasource": {"_undef": {"key1": False}}}
-        self.distro = "distrotest"  # generally should be a Distro object
+        self.distro = ubuntu.Distro("somedistro", {}, {})
         self.paths = Paths({})
         self.datasource = DataSource(self.sys_cfg, self.distro, self.paths)
 
@@ -201,28 +202,28 @@ def test_datasource_get_url_uses_defaults_on_errors(self):
         for log in expected_logs:
             self.assertIn(log, logs)
 
-    @mock.patch("cloudinit.sources.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     def test_fallback_interface_is_discovered(self, m_get_fallback_nic):
         """The fallback_interface is discovered via find_fallback_nic."""
         m_get_fallback_nic.return_value = "nic9"
-        self.assertEqual("nic9", self.datasource.fallback_interface)
+        self.assertEqual("nic9", self.datasource.distro.fallback_interface)
 
     @mock.patch("cloudinit.sources.net.find_fallback_nic")
     def test_fallback_interface_logs_undiscovered(self, m_get_fallback_nic):
         """Log a warning when fallback_interface can not discover the nic."""
-        self.datasource._cloud_name = "MySupahCloud"
         m_get_fallback_nic.return_value = None  # Couldn't discover nic
-        self.assertIsNone(self.datasource.fallback_interface)
+        self.assertIsNone(self.datasource.distro.fallback_interface)
         self.assertEqual(
-            "WARNING: Did not find a fallback interface on MySupahCloud.\n",
+            "WARNING: Did not find a fallback interface on distro: "
+            "somedistro.\n",
             self.logs.getvalue(),
         )
 
     @mock.patch("cloudinit.sources.net.find_fallback_nic")
     def test_wb_fallback_interface_is_cached(self, m_get_fallback_nic):
         """The fallback_interface is cached and won't be rediscovered."""
-        self.datasource._fallback_interface = "nic10"
-        self.assertEqual("nic10", self.datasource.fallback_interface)
+        self.datasource.distro.fallback_interface = "nic10"
+        self.assertEqual("nic10", self.datasource.distro.fallback_interface)
         m_get_fallback_nic.assert_not_called()
 
     def test__get_data_unimplemented(self):
diff --git a/tests/unittests/sources/test_openstack.py b/tests/unittests/sources/test_openstack.py
index 6f588122bc06..97cc8c94e6ab 100644
--- a/tests/unittests/sources/test_openstack.py
+++ b/tests/unittests/sources/test_openstack.py
@@ -338,16 +338,14 @@ def test_local_datasource(self, m_dhcp, m_net):
         ds_os_local = ds.DataSourceOpenStackLocal(
             settings.CFG_BUILTIN, distro, helpers.Paths({"run_dir": self.tmp})
         )
-        ds_os_local._fallback_interface = "eth9"  # Monkey patch for dhcp
-        m_dhcp.return_value = [
-            {
-                "interface": "eth9",
-                "fixed-address": "192.168.2.9",
-                "routers": "192.168.2.1",
-                "subnet-mask": "255.255.255.0",
-                "broadcast-address": "192.168.2.255",
-            }
-        ]
+        distro.fallback_interface = "eth9"  # Monkey patch for dhcp
+        m_dhcp.return_value = {
+            "interface": "eth9",
+            "fixed-address": "192.168.2.9",
+            "routers": "192.168.2.1",
+            "subnet-mask": "255.255.255.0",
+            "broadcast-address": "192.168.2.255",
+        }
 
         self.assertIsNone(ds_os_local.version)
         with test_helpers.mock.patch.object(
diff --git a/tests/unittests/sources/test_scaleway.py b/tests/unittests/sources/test_scaleway.py
index 6fb49802b02c..9ac73deddd48 100644
--- a/tests/unittests/sources/test_scaleway.py
+++ b/tests/unittests/sources/test_scaleway.py
@@ -10,6 +10,7 @@
 from requests.exceptions import ConnectionError, ConnectTimeout
 
 from cloudinit import helpers, settings, sources
+from cloudinit.distros import ubuntu
 from cloudinit.sources import DataSourceScaleway
 from tests.unittests.helpers import CiTestCase, ResponsesTestCase, mock
 
@@ -190,7 +191,7 @@ def _fix_mocking_url(url: str) -> str:
 class TestDataSourceScaleway(ResponsesTestCase):
     def setUp(self):
         tmp = self.tmp_dir()
-        distro = mock.MagicMock()
+        distro = ubuntu.Distro("", {}, {})
         distro.get_tmp_exec_path = self.tmp_dir
         self.datasource = DataSourceScaleway.DataSourceScaleway(
             settings.CFG_BUILTIN, distro, helpers.Paths({"run_dir": tmp})
@@ -217,7 +218,7 @@ def setUp(self):
             return_value=True,
         )
         self.add_patch(
-            "cloudinit.sources.DataSourceScaleway.net.find_fallback_nic",
+            "cloudinit.distros.net.find_fallback_nic",
             "_m_find_fallback_nic",
             return_value="scalewaynic0",
         )
@@ -701,7 +702,7 @@ def test_ssh_keys_both(self):
             ],
         )
 
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_legacy_network_config_ok(self, m_get_cmdline, fallback_nic):
         """
@@ -726,7 +727,7 @@ def test_legacy_network_config_ok(self, m_get_cmdline, fallback_nic):
         }
         self.assertEqual(netcfg, resp)
 
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_legacy_network_config_ipv6_ok(self, m_get_cmdline, fallback_nic):
         """
@@ -769,7 +770,7 @@ def test_legacy_network_config_ipv6_ok(self, m_get_cmdline, fallback_nic):
         }
         self.assertEqual(netcfg, resp)
 
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_legacy_network_config_existing(self, m_get_cmdline, fallback_nic):
         """
@@ -782,7 +783,7 @@ def test_legacy_network_config_existing(self, m_get_cmdline, fallback_nic):
         netcfg = self.datasource.network_config
         self.assertEqual(netcfg, "0xdeadbeef")
 
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_legacy_network_config_unset(self, m_get_cmdline, fallback_nic):
         """
@@ -810,7 +811,7 @@ def test_legacy_network_config_unset(self, m_get_cmdline, fallback_nic):
         self.assertEqual(netcfg, resp)
 
     @mock.patch("cloudinit.sources.DataSourceScaleway.LOG.warning")
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_legacy_network_config_cached_none(
         self, m_get_cmdline, fallback_nic, logwarning
@@ -843,7 +844,7 @@ def test_legacy_network_config_cached_none(
             sources.UNSET,
         )
 
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_ipmob_primary_ipv4_config_ok(self, m_get_cmdline, fallback_nic):
         """
@@ -872,7 +873,7 @@ def test_ipmob_primary_ipv4_config_ok(self, m_get_cmdline, fallback_nic):
 
         self.assertEqual(netcfg, resp)
 
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_ipmob_additional_ipv4_config_ok(
         self, m_get_cmdline, fallback_nic
@@ -914,7 +915,7 @@ def test_ipmob_additional_ipv4_config_ok(
         }
         self.assertEqual(netcfg, resp)
 
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_ipmob_primary_ipv6_config_ok(self, m_get_cmdline, fallback_nic):
         """
@@ -952,7 +953,7 @@ def test_ipmob_primary_ipv6_config_ok(self, m_get_cmdline, fallback_nic):
 
         self.assertEqual(netcfg, resp)
 
-    @mock.patch("cloudinit.sources.DataSourceScaleway.net.find_fallback_nic")
+    @mock.patch("cloudinit.distros.net.find_fallback_nic")
     @mock.patch("cloudinit.util.get_cmdline")
     def test_ipmob_primary_ipv4_v6_config_ok(
         self, m_get_cmdline, fallback_nic
diff --git a/tests/unittests/sources/test_upcloud.py b/tests/unittests/sources/test_upcloud.py
index e945ae5d8892..e16733ad526b 100644
--- a/tests/unittests/sources/test_upcloud.py
+++ b/tests/unittests/sources/test_upcloud.py
@@ -226,15 +226,13 @@ def test_network_configured_metadata(
         mock_readmd.return_value = UC_METADATA.copy()
 
         m_fallback_nic.return_value = "eth1"
-        m_dhcp.return_value = [
-            {
-                "interface": "eth1",
-                "fixed-address": "10.6.3.27",
-                "routers": "10.6.0.1",
-                "subnet-mask": "22",
-                "broadcast-address": "10.6.3.255",
-            }
-        ]
+        m_dhcp.return_value = {
+            "interface": "eth1",
+            "fixed-address": "10.6.3.27",
+            "routers": "10.6.0.1",
+            "subnet-mask": "22",
+            "broadcast-address": "10.6.3.255",
+        }
 
         ds = self.get_ds()
 
diff --git a/tests/unittests/util.py b/tests/unittests/util.py
index 8ae8d82e3460..e3d430694026 100644
--- a/tests/unittests/util.py
+++ b/tests/unittests/util.py
@@ -2,6 +2,7 @@
 from unittest import mock
 
 from cloudinit import cloud, distros, helpers
+from cloudinit.net.dhcp import IscDhclient
 from cloudinit.sources import DataSource, DataSourceHostname
 from cloudinit.sources.DataSourceNone import DataSourceNone
 
@@ -52,10 +53,6 @@ def get_hostname(self, fqdn=False, resolve_ip=False, metadata_only=False):
     def persist_instance_data(self):
         return True
 
-    @property
-    def fallback_interface(self):
-        return None
-
     @property
     def cloud_name(self):
         return "testing"
@@ -64,12 +61,19 @@ def cloud_name(self):
 class MockDistro(distros.Distro):
     # MockDistro is here to test base Distro class implementations
     def __init__(self, name="testingdistro", cfg=None, paths=None):
+        self._client = None
         if not cfg:
             cfg = {}
         if not paths:
             paths = {}
         super(MockDistro, self).__init__(name, cfg, paths)
 
+    @property
+    def dhcp_client(self):
+        if not self._client:
+            self._client = IscDhclient()
+        return self._client
+
     def install_packages(self, pkglist):
         pass