T2117: Cloud-init updated to 22.1

Merged with 22.1 tag from the upstream Cloud-init repository.

Our modules were slightly modified for compatibility with the new
version.

5 files changed, 2817 insertions, 0 deletions

diff --git a/tests/unittests/net/__init__.py b/tests/unittests/net/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/tests/unittests/net/__init__.py
diff --git a/tests/unittests/net/test_dhcp.py b/tests/unittests/net/test_dhcp.py
new file mode 100644
index 00000000..876873d5
--- /dev/null
+++ b/tests/unittests/net/test_dhcp.py
@@ -0,0 +1,797 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+import os
+import signal
+from textwrap import dedent
+
+import httpretty
+
+import cloudinit.net as net
+from cloudinit.net.dhcp import (
+    InvalidDHCPLeaseFileError,
+    dhcp_discovery,
+    maybe_perform_dhcp_discovery,
+    networkd_load_leases,
+    parse_dhcp_lease_file,
+    parse_static_routes,
+)
+from cloudinit.util import ensure_file, write_file
+from tests.unittests.helpers import (
+    CiTestCase,
+    HttprettyTestCase,
+    mock,
+    populate_dir,
+    wrap_and_call,
+)
+
+
+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:
+            parse_dhcp_lease_file(empty_file)
+        error = context_manager.exception
+        self.assertIn("Cannot parse empty dhcp lease file", str(error))
+
+    def test_parse_malformed_lease_file_content_errors(self):
+        """parse_dhcp_lease_file 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:
+            parse_dhcp_lease_file(non_lease_file)
+        error = context_manager.exception
+        self.assertIn("Cannot parse dhcp lease file", str(error))
+
+    def test_parse_multiple_leases(self):
+        """parse_dhcp_lease_file returns a list of all leases within."""
+        lease_file = self.tmp_path("leases")
+        content = dedent(
+            """
+            lease {
+              interface "wlp3s0";
+              fixed-address 192.168.2.74;
+              filename "http://192.168.2.50/boot.php?mac=${netX}";
+              option subnet-mask 255.255.255.0;
+              option routers 192.168.2.1;
+              renew 4 2017/07/27 18:02:30;
+              expire 5 2017/07/28 07:08:15;
+            }
+            lease {
+              interface "wlp3s0";
+              fixed-address 192.168.2.74;
+              filename "http://192.168.2.50/boot.php?mac=${netX}";
+              option subnet-mask 255.255.255.0;
+              option routers 192.168.2.1;
+            }
+        """
+        )
+        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, parse_dhcp_lease_file(lease_file))
+
+
+class TestDHCPRFC3442(CiTestCase):
+    def test_parse_lease_finds_rfc3442_classless_static_routes(self):
+        """parse_dhcp_lease_file returns rfc3442-classless-static-routes."""
+        lease_file = self.tmp_path("leases")
+        content = dedent(
+            """
+            lease {
+              interface "wlp3s0";
+              fixed-address 192.168.2.74;
+              option subnet-mask 255.255.255.0;
+              option routers 192.168.2.1;
+              option 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;
+            }
+        """
+        )
+        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, parse_dhcp_lease_file(lease_file))
+
+    def test_parse_lease_finds_classless_static_routes(self):
+        """
+        parse_dhcp_lease_file returns classless-static-routes
+        for Centos lease format.
+        """
+        lease_file = self.tmp_path("leases")
+        content = dedent(
+            """
+            lease {
+              interface "wlp3s0";
+              fixed-address 192.168.2.74;
+              option subnet-mask 255.255.255.0;
+              option routers 192.168.2.1;
+              option classless-static-routes 0 130.56.240.1;
+              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",
+                "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, parse_dhcp_lease_file(lease_file))
+
+    @mock.patch("cloudinit.net.dhcp.EphemeralIPv4Network")
+    @mock.patch("cloudinit.net.dhcp.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
+        eph = net.dhcp.EphemeralDHCPv4()
+        eph.obtain_lease()
+        expected_kwargs = {
+            "interface": "wlp3s0",
+            "ip": "192.168.2.74",
+            "prefix_or_mask": "255.255.255.0",
+            "broadcast": "192.168.2.255",
+            "static_routes": [("0.0.0.0/0", "130.56.240.1")],
+            "router": "192.168.2.1",
+        }
+        m_ipv4.assert_called_with(**expected_kwargs)
+
+    @mock.patch("cloudinit.net.dhcp.EphemeralIPv4Network")
+    @mock.patch("cloudinit.net.dhcp.maybe_perform_dhcp_discovery")
+    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
+        eph = net.dhcp.EphemeralDHCPv4()
+        eph.obtain_lease()
+        expected_kwargs = {
+            "interface": "wlp3s0",
+            "ip": "192.168.2.74",
+            "prefix_or_mask": "255.255.255.0",
+            "broadcast": "192.168.2.255",
+            "static_routes": [("0.0.0.0/0", "130.56.240.1")],
+            "router": "192.168.2.1",
+        }
+        m_ipv4.assert_called_with(**expected_kwargs)
+
+
+class TestDHCPParseStaticRoutes(CiTestCase):
+
+    with_logs = True
+
+    def parse_static_routes_empty_string(self):
+        self.assertEqual([], parse_static_routes(""))
+
+    def test_parse_static_routes_invalid_input_returns_empty_list(self):
+        rfc3442 = "32,169,254,169,254,130,56,248"
+        self.assertEqual([], parse_static_routes(rfc3442))
+
+    def test_parse_static_routes_bogus_width_returns_empty_list(self):
+        rfc3442 = "33,169,254,169,254,130,56,248"
+        self.assertEqual([], parse_static_routes(rfc3442))
+
+    def test_parse_static_routes_single_ip(self):
+        rfc3442 = "32,169,254,169,254,130,56,248,255"
+        self.assertEqual(
+            [("169.254.169.254/32", "130.56.248.255")],
+            parse_static_routes(rfc3442),
+        )
+
+    def test_parse_static_routes_single_ip_handles_trailing_semicolon(self):
+        rfc3442 = "32,169,254,169,254,130,56,248,255;"
+        self.assertEqual(
+            [("169.254.169.254/32", "130.56.248.255")],
+            parse_static_routes(rfc3442),
+        )
+
+    def test_parse_static_routes_default_route(self):
+        rfc3442 = "0,130,56,240,1"
+        self.assertEqual(
+            [("0.0.0.0/0", "130.56.240.1")], parse_static_routes(rfc3442)
+        )
+
+    def test_unspecified_gateway(self):
+        rfc3442 = "32,169,254,169,254,0,0,0,0"
+        self.assertEqual(
+            [("169.254.169.254/32", "0.0.0.0")], parse_static_routes(rfc3442)
+        )
+
+    def test_parse_static_routes_class_c_b_a(self):
+        class_c = "24,192,168,74,192,168,0,4"
+        class_b = "16,172,16,172,16,0,4"
+        class_a = "8,10,10,0,0,4"
+        rfc3442 = ",".join([class_c, class_b, class_a])
+        self.assertEqual(
+            sorted(
+                [
+                    ("192.168.74.0/24", "192.168.0.4"),
+                    ("172.16.0.0/16", "172.16.0.4"),
+                    ("10.0.0.0/8", "10.0.0.4"),
+                ]
+            ),
+            sorted(parse_static_routes(rfc3442)),
+        )
+
+    def test_parse_static_routes_logs_error_truncated(self):
+        bad_rfc3442 = {
+            "class_c": "24,169,254,169,10",
+            "class_b": "16,172,16,10",
+            "class_a": "8,10,10",
+            "gateway": "0,0",
+            "netlen": "33,0",
+        }
+        for rfc3442 in bad_rfc3442.values():
+            self.assertEqual([], parse_static_routes(rfc3442))
+
+        logs = self.logs.getvalue()
+        self.assertEqual(len(bad_rfc3442.keys()), len(logs.splitlines()))
+
+    def test_parse_static_routes_returns_valid_routes_until_parse_err(self):
+        class_c = "24,192,168,74,192,168,0,4"
+        class_b = "16,172,16,172,16,0,4"
+        class_a_error = "8,10,10,0,0"
+        rfc3442 = ",".join([class_c, class_b, class_a_error])
+        self.assertEqual(
+            sorted(
+                [
+                    ("192.168.74.0/24", "192.168.0.4"),
+                    ("172.16.0.0/16", "172.16.0.4"),
+                ]
+            ),
+            sorted(parse_static_routes(rfc3442)),
+        )
+
+        logs = self.logs.getvalue()
+        self.assertIn(rfc3442, logs.splitlines()[0])
+
+    def test_redhat_format(self):
+        redhat_format = "24.191.168.128 192.168.128.1,0 192.168.128.1"
+        self.assertEqual(
+            sorted(
+                [
+                    ("191.168.128.0/24", "192.168.128.1"),
+                    ("0.0.0.0/0", "192.168.128.1"),
+                ]
+            ),
+            sorted(parse_static_routes(redhat_format)),
+        )
+
+    def test_redhat_format_with_a_space_too_much_after_comma(self):
+        redhat_format = "24.191.168.128 192.168.128.1, 0 192.168.128.1"
+        self.assertEqual(
+            sorted(
+                [
+                    ("191.168.128.0/24", "192.168.128.1"),
+                    ("0.0.0.0/0", "192.168.128.1"),
+                ]
+            ),
+            sorted(parse_static_routes(redhat_format)),
+        )
+
+
+class TestDHCPDiscoveryClean(CiTestCase):
+    with_logs = True
+
+    @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
+        self.assertEqual([], maybe_perform_dhcp_discovery())
+        self.assertIn(
+            "Skip dhcp_discovery: Unable to find fallback nic.",
+            self.logs.getvalue(),
+        )
+
+    def test_provided_nic_does_not_exist(self):
+        """When the provided nic doesn't exist, log a message and no-op."""
+        self.assertEqual([], maybe_perform_dhcp_discovery("idontexist"))
+        self.assertIn(
+            "Skip dhcp_discovery: nic idontexist not found in get_devicelist.",
+            self.logs.getvalue(),
+        )
+
+    @mock.patch("cloudinit.net.dhcp.subp.which")
+    @mock.patch("cloudinit.net.dhcp.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
+        self.assertEqual([], maybe_perform_dhcp_discovery())
+        self.assertIn(
+            "Skip dhclient configuration: No dhclient command found.",
+            self.logs.getvalue(),
+        )
+
+    @mock.patch("cloudinit.temp_utils.os.getuid")
+    @mock.patch("cloudinit.net.dhcp.dhcp_discovery")
+    @mock.patch("cloudinit.net.dhcp.subp.which")
+    @mock.patch("cloudinit.net.dhcp.find_fallback_nic")
+    def test_dhclient_run_with_tmpdir(self, m_fback, m_which, m_dhcp, m_uid):
+        """maybe_perform_dhcp_discovery passes tmpdir to dhcp_discovery."""
+        m_uid.return_value = 0  # Fake root user for tmpdir
+        m_fback.return_value = "eth9"
+        m_which.return_value = "/sbin/dhclient"
+        m_dhcp.return_value = {"address": "192.168.2.2"}
+        retval = wrap_and_call(
+            "cloudinit.temp_utils",
+            {"_TMPDIR": {"new": None}, "os.getuid": 0},
+            maybe_perform_dhcp_discovery,
+        )
+        self.assertEqual({"address": "192.168.2.2"}, retval)
+        self.assertEqual(
+            1, m_dhcp.call_count, "dhcp_discovery not called once"
+        )
+        call = m_dhcp.call_args_list[0]
+        self.assertEqual("/sbin/dhclient", call[0][0])
+        self.assertEqual("eth9", call[0][1])
+        self.assertIn("/var/tmp/cloud-init/cloud-init-dhcp-", call[0][2])
+
+    @mock.patch("time.sleep", mock.MagicMock())
+    @mock.patch("cloudinit.net.dhcp.os.kill")
+    @mock.patch("cloudinit.net.dhcp.subp.subp")
+    def test_dhcp_discovery_run_in_sandbox_warns_invalid_pid(
+        self, m_subp, m_kill
+    ):
+        """dhcp_discovery logs a warning when pidfile contains invalid content.
+
+        Lease processing still occurs and no proc kill is attempted.
+        """
+        m_subp.return_value = ("", "")
+        tmpdir = self.tmp_dir()
+        dhclient_script = os.path.join(tmpdir, "dhclient.orig")
+        script_content = "#!/bin/bash\necho fake-dhclient"
+        write_file(dhclient_script, script_content, mode=0o755)
+        write_file(self.tmp_path("dhclient.pid", tmpdir), "")  # Empty pid ''
+        lease_content = dedent(
+            """
+            lease {
+              interface "eth9";
+              fixed-address 192.168.2.74;
+              option subnet-mask 255.255.255.0;
+              option routers 192.168.2.1;
+            }
+        """
+        )
+        write_file(self.tmp_path("dhcp.leases", tmpdir), lease_content)
+
+        self.assertCountEqual(
+            [
+                {
+                    "interface": "eth9",
+                    "fixed-address": "192.168.2.74",
+                    "subnet-mask": "255.255.255.0",
+                    "routers": "192.168.2.1",
+                }
+            ],
+            dhcp_discovery(dhclient_script, "eth9", tmpdir),
+        )
+        self.assertIn(
+            "dhclient(pid=, parentpid=unknown) failed "
+            "to daemonize after 10.0 seconds",
+            self.logs.getvalue(),
+        )
+        m_kill.assert_not_called()
+
+    @mock.patch("cloudinit.net.dhcp.util.get_proc_ppid")
+    @mock.patch("cloudinit.net.dhcp.os.kill")
+    @mock.patch("cloudinit.net.dhcp.util.wait_for_files")
+    @mock.patch("cloudinit.net.dhcp.subp.subp")
+    def test_dhcp_discovery_run_in_sandbox_waits_on_lease_and_pid(
+        self, m_subp, m_wait, m_kill, m_getppid
+    ):
+        """dhcp_discovery waits for the presence of pidfile and dhcp.leases."""
+        m_subp.return_value = ("", "")
+        tmpdir = self.tmp_dir()
+        dhclient_script = os.path.join(tmpdir, "dhclient.orig")
+        script_content = "#!/bin/bash\necho fake-dhclient"
+        write_file(dhclient_script, script_content, mode=0o755)
+        # Don't create pid or leases file
+        pidfile = self.tmp_path("dhclient.pid", tmpdir)
+        leasefile = self.tmp_path("dhcp.leases", tmpdir)
+        m_wait.return_value = [pidfile]  # Return the missing pidfile wait for
+        m_getppid.return_value = 1  # Indicate that dhclient has daemonized
+        self.assertEqual([], dhcp_discovery(dhclient_script, "eth9", tmpdir))
+        self.assertEqual(
+            mock.call([pidfile, leasefile], maxwait=5, naplen=0.01),
+            m_wait.call_args_list[0],
+        )
+        self.assertIn(
+            "WARNING: dhclient did not produce expected files: dhclient.pid",
+            self.logs.getvalue(),
+        )
+        m_kill.assert_not_called()
+
+    @mock.patch("cloudinit.net.dhcp.util.get_proc_ppid")
+    @mock.patch("cloudinit.net.dhcp.os.kill")
+    @mock.patch("cloudinit.net.dhcp.subp.subp")
+    def test_dhcp_discovery_run_in_sandbox(self, m_subp, m_kill, m_getppid):
+        """dhcp_discovery brings up the interface and runs dhclient.
+
+        It also returns the parsed dhcp.leases file generated in the sandbox.
+        """
+        m_subp.return_value = ("", "")
+        tmpdir = self.tmp_dir()
+        dhclient_script = os.path.join(tmpdir, "dhclient.orig")
+        script_content = "#!/bin/bash\necho fake-dhclient"
+        write_file(dhclient_script, script_content, mode=0o755)
+        lease_content = dedent(
+            """
+            lease {
+              interface "eth9";
+              fixed-address 192.168.2.74;
+              option subnet-mask 255.255.255.0;
+              option routers 192.168.2.1;
+            }
+        """
+        )
+        lease_file = os.path.join(tmpdir, "dhcp.leases")
+        write_file(lease_file, lease_content)
+        pid_file = os.path.join(tmpdir, "dhclient.pid")
+        my_pid = 1
+        write_file(pid_file, "%d\n" % my_pid)
+        m_getppid.return_value = 1  # Indicate that dhclient has daemonized
+
+        self.assertCountEqual(
+            [
+                {
+                    "interface": "eth9",
+                    "fixed-address": "192.168.2.74",
+                    "subnet-mask": "255.255.255.0",
+                    "routers": "192.168.2.1",
+                }
+            ],
+            dhcp_discovery(dhclient_script, "eth9", tmpdir),
+        )
+        # dhclient script got copied
+        with open(os.path.join(tmpdir, "dhclient")) as stream:
+            self.assertEqual(script_content, stream.read())
+        # Interface was brought up before dhclient called from sandbox
+        m_subp.assert_has_calls(
+            [
+                mock.call(
+                    ["ip", "link", "set", "dev", "eth9", "up"], capture=True
+                ),
+                mock.call(
+                    [
+                        os.path.join(tmpdir, "dhclient"),
+                        "-1",
+                        "-v",
+                        "-lf",
+                        lease_file,
+                        "-pf",
+                        os.path.join(tmpdir, "dhclient.pid"),
+                        "eth9",
+                        "-sf",
+                        "/bin/true",
+                    ],
+                    capture=True,
+                ),
+            ]
+        )
+        m_kill.assert_has_calls([mock.call(my_pid, signal.SIGKILL)])
+
+    @mock.patch("cloudinit.net.dhcp.util.get_proc_ppid")
+    @mock.patch("cloudinit.net.dhcp.os.kill")
+    @mock.patch("cloudinit.net.dhcp.subp.subp")
+    def test_dhcp_discovery_outside_sandbox(self, m_subp, m_kill, m_getppid):
+        """dhcp_discovery brings up the interface and runs dhclient.
+
+        It also returns the parsed dhcp.leases file generated in the sandbox.
+        """
+        m_subp.return_value = ("", "")
+        tmpdir = self.tmp_dir()
+        dhclient_script = os.path.join(tmpdir, "dhclient.orig")
+        script_content = "#!/bin/bash\necho fake-dhclient"
+        write_file(dhclient_script, script_content, mode=0o755)
+        lease_content = dedent(
+            """
+            lease {
+              interface "eth9";
+              fixed-address 192.168.2.74;
+              option subnet-mask 255.255.255.0;
+              option routers 192.168.2.1;
+            }
+        """
+        )
+        lease_file = os.path.join(tmpdir, "dhcp.leases")
+        write_file(lease_file, lease_content)
+        pid_file = os.path.join(tmpdir, "dhclient.pid")
+        my_pid = 1
+        write_file(pid_file, "%d\n" % my_pid)
+        m_getppid.return_value = 1  # Indicate that dhclient has daemonized
+
+        with mock.patch("os.access", return_value=False):
+            self.assertCountEqual(
+                [
+                    {
+                        "interface": "eth9",
+                        "fixed-address": "192.168.2.74",
+                        "subnet-mask": "255.255.255.0",
+                        "routers": "192.168.2.1",
+                    }
+                ],
+                dhcp_discovery(dhclient_script, "eth9", tmpdir),
+            )
+        # dhclient script got copied
+        with open(os.path.join(tmpdir, "dhclient.orig")) as stream:
+            self.assertEqual(script_content, stream.read())
+        # Interface was brought up before dhclient called from sandbox
+        m_subp.assert_has_calls(
+            [
+                mock.call(
+                    ["ip", "link", "set", "dev", "eth9", "up"], capture=True
+                ),
+                mock.call(
+                    [
+                        os.path.join(tmpdir, "dhclient.orig"),
+                        "-1",
+                        "-v",
+                        "-lf",
+                        lease_file,
+                        "-pf",
+                        os.path.join(tmpdir, "dhclient.pid"),
+                        "eth9",
+                        "-sf",
+                        "/bin/true",
+                    ],
+                    capture=True,
+                ),
+            ]
+        )
+        m_kill.assert_has_calls([mock.call(my_pid, signal.SIGKILL)])
+
+    @mock.patch("cloudinit.net.dhcp.util.get_proc_ppid")
+    @mock.patch("cloudinit.net.dhcp.os.kill")
+    @mock.patch("cloudinit.net.dhcp.subp.subp")
+    def test_dhcp_output_error_stream(self, m_subp, m_kill, m_getppid):
+        """ "dhcp_log_func is called with the output and error streams of
+        dhclinet when the callable is passed."""
+        dhclient_err = "FAKE DHCLIENT ERROR"
+        dhclient_out = "FAKE DHCLIENT OUT"
+        m_subp.return_value = (dhclient_out, dhclient_err)
+        tmpdir = self.tmp_dir()
+        dhclient_script = os.path.join(tmpdir, "dhclient.orig")
+        script_content = "#!/bin/bash\necho fake-dhclient"
+        write_file(dhclient_script, script_content, mode=0o755)
+        lease_content = dedent(
+            """
+                lease {
+                  interface "eth9";
+                  fixed-address 192.168.2.74;
+                  option subnet-mask 255.255.255.0;
+                  option routers 192.168.2.1;
+                }
+            """
+        )
+        lease_file = os.path.join(tmpdir, "dhcp.leases")
+        write_file(lease_file, lease_content)
+        pid_file = os.path.join(tmpdir, "dhclient.pid")
+        my_pid = 1
+        write_file(pid_file, "%d\n" % my_pid)
+        m_getppid.return_value = 1  # Indicate that dhclient has daemonized
+
+        def dhcp_log_func(out, err):
+            self.assertEqual(out, dhclient_out)
+            self.assertEqual(err, dhclient_err)
+
+        dhcp_discovery(
+            dhclient_script, "eth9", tmpdir, dhcp_log_func=dhcp_log_func
+        )
+
+
+class TestSystemdParseLeases(CiTestCase):
+
+    lxd_lease = dedent(
+        """\
+    # This is private data. Do not parse.
+    ADDRESS=10.75.205.242
+    NETMASK=255.255.255.0
+    ROUTER=10.75.205.1
+    SERVER_ADDRESS=10.75.205.1
+    NEXT_SERVER=10.75.205.1
+    BROADCAST=10.75.205.255
+    T1=1580
+    T2=2930
+    LIFETIME=3600
+    DNS=10.75.205.1
+    DOMAINNAME=lxd
+    HOSTNAME=a1
+    CLIENTID=ffe617693400020000ab110c65a6a0866931c2
+    """
+    )
+
+    lxd_parsed = {
+        "ADDRESS": "10.75.205.242",
+        "NETMASK": "255.255.255.0",
+        "ROUTER": "10.75.205.1",
+        "SERVER_ADDRESS": "10.75.205.1",
+        "NEXT_SERVER": "10.75.205.1",
+        "BROADCAST": "10.75.205.255",
+        "T1": "1580",
+        "T2": "2930",
+        "LIFETIME": "3600",
+        "DNS": "10.75.205.1",
+        "DOMAINNAME": "lxd",
+        "HOSTNAME": "a1",
+        "CLIENTID": "ffe617693400020000ab110c65a6a0866931c2",
+    }
+
+    azure_lease = dedent(
+        """\
+    # This is private data. Do not parse.
+    ADDRESS=10.132.0.5
+    NETMASK=255.255.255.255
+    ROUTER=10.132.0.1
+    SERVER_ADDRESS=169.254.169.254
+    NEXT_SERVER=10.132.0.1
+    MTU=1460
+    T1=43200
+    T2=75600
+    LIFETIME=86400
+    DNS=169.254.169.254
+    NTP=169.254.169.254
+    DOMAINNAME=c.ubuntu-foundations.internal
+    DOMAIN_SEARCH_LIST=c.ubuntu-foundations.internal google.internal
+    HOSTNAME=tribaal-test-171002-1349.c.ubuntu-foundations.internal
+    ROUTES=10.132.0.1/32,0.0.0.0 0.0.0.0/0,10.132.0.1
+    CLIENTID=ff405663a200020000ab11332859494d7a8b4c
+    OPTION_245=624c3620
+    """
+    )
+
+    azure_parsed = {
+        "ADDRESS": "10.132.0.5",
+        "NETMASK": "255.255.255.255",
+        "ROUTER": "10.132.0.1",
+        "SERVER_ADDRESS": "169.254.169.254",
+        "NEXT_SERVER": "10.132.0.1",
+        "MTU": "1460",
+        "T1": "43200",
+        "T2": "75600",
+        "LIFETIME": "86400",
+        "DNS": "169.254.169.254",
+        "NTP": "169.254.169.254",
+        "DOMAINNAME": "c.ubuntu-foundations.internal",
+        "DOMAIN_SEARCH_LIST": "c.ubuntu-foundations.internal google.internal",
+        "HOSTNAME": "tribaal-test-171002-1349.c.ubuntu-foundations.internal",
+        "ROUTES": "10.132.0.1/32,0.0.0.0 0.0.0.0/0,10.132.0.1",
+        "CLIENTID": "ff405663a200020000ab11332859494d7a8b4c",
+        "OPTION_245": "624c3620",
+    }
+
+    def setUp(self):
+        super(TestSystemdParseLeases, self).setUp()
+        self.lease_d = self.tmp_dir()
+
+    def test_no_leases_returns_empty_dict(self):
+        """A leases dir with no lease files should return empty dictionary."""
+        self.assertEqual({}, networkd_load_leases(self.lease_d))
+
+    def test_no_leases_dir_returns_empty_dict(self):
+        """A non-existing leases dir should return empty dict."""
+        enodir = os.path.join(self.lease_d, "does-not-exist")
+        self.assertEqual({}, networkd_load_leases(enodir))
+
+    def test_single_leases_file(self):
+        """A leases dir with one leases file."""
+        populate_dir(self.lease_d, {"2": self.lxd_lease})
+        self.assertEqual(
+            {"2": self.lxd_parsed}, networkd_load_leases(self.lease_d)
+        )
+
+    def test_single_azure_leases_file(self):
+        """On Azure, option 245 should be present, verify it specifically."""
+        populate_dir(self.lease_d, {"1": self.azure_lease})
+        self.assertEqual(
+            {"1": self.azure_parsed}, networkd_load_leases(self.lease_d)
+        )
+
+    def test_multiple_files(self):
+        """Multiple leases files on azure with one found return that value."""
+        self.maxDiff = None
+        populate_dir(
+            self.lease_d, {"1": self.azure_lease, "9": self.lxd_lease}
+        )
+        self.assertEqual(
+            {"1": self.azure_parsed, "9": self.lxd_parsed},
+            networkd_load_leases(self.lease_d),
+        )
+
+
+class TestEphemeralDhcpNoNetworkSetup(HttprettyTestCase):
+    @mock.patch("cloudinit.net.dhcp.maybe_perform_dhcp_discovery")
+    def test_ephemeral_dhcp_no_network_if_url_connectivity(self, m_dhcp):
+        """No EphemeralDhcp4 network setup when connectivity_url succeeds."""
+        url = "http://example.org/index.html"
+
+        httpretty.register_uri(httpretty.GET, url)
+        with net.dhcp.EphemeralDHCPv4(
+            connectivity_url_data={"url": url},
+        ) as lease:
+            self.assertIsNone(lease)
+        # Ensure that no teardown happens:
+        m_dhcp.assert_not_called()
+
+    @mock.patch("cloudinit.net.dhcp.subp.subp")
+    @mock.patch("cloudinit.net.dhcp.maybe_perform_dhcp_discovery")
+    def test_ephemeral_dhcp_setup_network_if_url_connectivity(
+        self, m_dhcp, m_subp
+    ):
+        """No EphemeralDhcp4 network setup when connectivity_url succeeds."""
+        url = "http://example.org/index.html"
+        fake_lease = {
+            "interface": "eth9",
+            "fixed-address": "192.168.2.2",
+            "subnet-mask": "255.255.0.0",
+        }
+        m_dhcp.return_value = [fake_lease]
+        m_subp.return_value = ("", "")
+
+        httpretty.register_uri(httpretty.GET, url, body={}, status=404)
+        with net.dhcp.EphemeralDHCPv4(
+            connectivity_url_data={"url": url},
+        ) as lease:
+            self.assertEqual(fake_lease, lease)
+        # Ensure that dhcp discovery occurs
+        m_dhcp.called_once_with()
+
+
+# vi: ts=4 expandtab
diff --git a/tests/unittests/net/test_init.py b/tests/unittests/net/test_init.py
new file mode 100644
index 00000000..18b3fe59
--- /dev/null
+++ b/tests/unittests/net/test_init.py
@@ -0,0 +1,1734 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+import copy
+import errno
+import ipaddress
+import os
+import textwrap
+from unittest import mock
+
+import httpretty
+import pytest
+import requests
+
+import cloudinit.net as net
+from cloudinit import safeyaml as yaml
+from cloudinit.subp import ProcessExecutionError
+from cloudinit.util import ensure_file, write_file
+from tests.unittests.helpers import CiTestCase, HttprettyTestCase
+
+
+class TestSysDevPath(CiTestCase):
+    def test_sys_dev_path(self):
+        """sys_dev_path returns a path under SYS_CLASS_NET for a device."""
+        dev = "something"
+        path = "attribute"
+        expected = net.SYS_CLASS_NET + dev + "/" + path
+        self.assertEqual(expected, net.sys_dev_path(dev, path))
+
+    def test_sys_dev_path_without_path(self):
+        """When path param isn't provided it defaults to empty string."""
+        dev = "something"
+        expected = net.SYS_CLASS_NET + dev + "/"
+        self.assertEqual(expected, net.sys_dev_path(dev))
+
+
+class TestReadSysNet(CiTestCase):
+    with_logs = True
+
+    def setUp(self):
+        super(TestReadSysNet, self).setUp()
+        sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
+        self.m_sys_path = sys_mock.start()
+        self.sysdir = self.tmp_dir() + "/"
+        self.m_sys_path.return_value = self.sysdir
+        self.addCleanup(sys_mock.stop)
+
+    def test_read_sys_net_strips_contents_of_sys_path(self):
+        """read_sys_net strips whitespace from the contents of a sys file."""
+        content = "some stuff with trailing whitespace\t\r\n"
+        write_file(os.path.join(self.sysdir, "dev", "attr"), content)
+        self.assertEqual(content.strip(), net.read_sys_net("dev", "attr"))
+
+    def test_read_sys_net_reraises_oserror(self):
+        """read_sys_net raises OSError/IOError when file doesn't exist."""
+        # Non-specific Exception because versions of python OSError vs IOError.
+        with self.assertRaises(Exception) as context_manager:  # noqa: H202
+            net.read_sys_net("dev", "attr")
+        error = context_manager.exception
+        self.assertIn("No such file or directory", str(error))
+
+    def test_read_sys_net_handles_error_with_on_enoent(self):
+        """read_sys_net handles OSError/IOError with on_enoent if provided."""
+        handled_errors = []
+
+        def on_enoent(e):
+            handled_errors.append(e)
+
+        net.read_sys_net("dev", "attr", on_enoent=on_enoent)
+        error = handled_errors[0]
+        self.assertIsInstance(error, Exception)
+        self.assertIn("No such file or directory", str(error))
+
+    def test_read_sys_net_translates_content(self):
+        """read_sys_net translates content when translate dict is provided."""
+        content = "you're welcome\n"
+        write_file(os.path.join(self.sysdir, "dev", "attr"), content)
+        translate = {"you're welcome": "de nada"}
+        self.assertEqual(
+            "de nada", net.read_sys_net("dev", "attr", translate=translate)
+        )
+
+    def test_read_sys_net_errors_on_translation_failures(self):
+        """read_sys_net raises a KeyError and logs details on failure."""
+        content = "you're welcome\n"
+        write_file(os.path.join(self.sysdir, "dev", "attr"), content)
+        with self.assertRaises(KeyError) as context_manager:
+            net.read_sys_net("dev", "attr", translate={})
+        error = context_manager.exception
+        self.assertEqual('"you\'re welcome"', str(error))
+        self.assertIn(
+            "Found unexpected (not translatable) value 'you're welcome' in "
+            "'{0}dev/attr".format(self.sysdir),
+            self.logs.getvalue(),
+        )
+
+    def test_read_sys_net_handles_handles_with_onkeyerror(self):
+        """read_sys_net handles translation errors calling on_keyerror."""
+        content = "you're welcome\n"
+        write_file(os.path.join(self.sysdir, "dev", "attr"), content)
+        handled_errors = []
+
+        def on_keyerror(e):
+            handled_errors.append(e)
+
+        net.read_sys_net("dev", "attr", translate={}, on_keyerror=on_keyerror)
+        error = handled_errors[0]
+        self.assertIsInstance(error, KeyError)
+        self.assertEqual('"you\'re welcome"', str(error))
+
+    def test_read_sys_net_safe_false_on_translate_failure(self):
+        """read_sys_net_safe returns False on translation failures."""
+        content = "you're welcome\n"
+        write_file(os.path.join(self.sysdir, "dev", "attr"), content)
+        self.assertFalse(net.read_sys_net_safe("dev", "attr", translate={}))
+
+    def test_read_sys_net_safe_returns_false_on_noent_failure(self):
+        """read_sys_net_safe returns False on file not found failures."""
+        self.assertFalse(net.read_sys_net_safe("dev", "attr"))
+
+    def test_read_sys_net_int_returns_none_on_error(self):
+        """read_sys_net_safe returns None on failures."""
+        self.assertFalse(net.read_sys_net_int("dev", "attr"))
+
+    def test_read_sys_net_int_returns_none_on_valueerror(self):
+        """read_sys_net_safe returns None when content is not an int."""
+        write_file(os.path.join(self.sysdir, "dev", "attr"), "NOTINT\n")
+        self.assertFalse(net.read_sys_net_int("dev", "attr"))
+
+    def test_read_sys_net_int_returns_integer_from_content(self):
+        """read_sys_net_safe returns None on failures."""
+        write_file(os.path.join(self.sysdir, "dev", "attr"), "1\n")
+        self.assertEqual(1, net.read_sys_net_int("dev", "attr"))
+
+    def test_is_up_true(self):
+        """is_up is True if sys/net/devname/operstate is 'up' or 'unknown'."""
+        for state in ["up", "unknown"]:
+            write_file(os.path.join(self.sysdir, "eth0", "operstate"), state)
+            self.assertTrue(net.is_up("eth0"))
+
+    def test_is_up_false(self):
+        """is_up is False if sys/net/devname/operstate is 'down' or invalid."""
+        for state in ["down", "incomprehensible"]:
+            write_file(os.path.join(self.sysdir, "eth0", "operstate"), state)
+            self.assertFalse(net.is_up("eth0"))
+
+    def test_is_bridge(self):
+        """is_bridge is True when /sys/net/devname/bridge exists."""
+        self.assertFalse(net.is_bridge("eth0"))
+        ensure_file(os.path.join(self.sysdir, "eth0", "bridge"))
+        self.assertTrue(net.is_bridge("eth0"))
+
+    def test_is_bond(self):
+        """is_bond is True when /sys/net/devname/bonding exists."""
+        self.assertFalse(net.is_bond("eth0"))
+        ensure_file(os.path.join(self.sysdir, "eth0", "bonding"))
+        self.assertTrue(net.is_bond("eth0"))
+
+    def test_get_master(self):
+        """get_master returns the path when /sys/net/devname/master exists."""
+        self.assertIsNone(net.get_master("enP1s1"))
+        master_path = os.path.join(self.sysdir, "enP1s1", "master")
+        ensure_file(master_path)
+        self.assertEqual(master_path, net.get_master("enP1s1"))
+
+    def test_master_is_bridge_or_bond(self):
+        bridge_mac = "aa:bb:cc:aa:bb:cc"
+        bond_mac = "cc:bb:aa:cc:bb:aa"
+
+        # No master => False
+        write_file(os.path.join(self.sysdir, "eth1", "address"), bridge_mac)
+        write_file(os.path.join(self.sysdir, "eth2", "address"), bond_mac)
+
+        self.assertFalse(net.master_is_bridge_or_bond("eth1"))
+        self.assertFalse(net.master_is_bridge_or_bond("eth2"))
+
+        # masters without bridge/bonding => False
+        write_file(os.path.join(self.sysdir, "br0", "address"), bridge_mac)
+        write_file(os.path.join(self.sysdir, "bond0", "address"), bond_mac)
+
+        os.symlink("../br0", os.path.join(self.sysdir, "eth1", "master"))
+        os.symlink("../bond0", os.path.join(self.sysdir, "eth2", "master"))
+
+        self.assertFalse(net.master_is_bridge_or_bond("eth1"))
+        self.assertFalse(net.master_is_bridge_or_bond("eth2"))
+
+        # masters with bridge/bonding => True
+        write_file(os.path.join(self.sysdir, "br0", "bridge"), "")
+        write_file(os.path.join(self.sysdir, "bond0", "bonding"), "")
+
+        self.assertTrue(net.master_is_bridge_or_bond("eth1"))
+        self.assertTrue(net.master_is_bridge_or_bond("eth2"))
+
+    def test_master_is_openvswitch(self):
+        ovs_mac = "bb:cc:aa:bb:cc:aa"
+
+        # No master => False
+        write_file(os.path.join(self.sysdir, "eth1", "address"), ovs_mac)
+
+        self.assertFalse(net.master_is_bridge_or_bond("eth1"))
+
+        # masters without ovs-system => False
+        write_file(os.path.join(self.sysdir, "ovs-system", "address"), ovs_mac)
+
+        os.symlink(
+            "../ovs-system", os.path.join(self.sysdir, "eth1", "master")
+        )
+
+        self.assertFalse(net.master_is_openvswitch("eth1"))
+
+        # masters with ovs-system => True
+        os.symlink(
+            "../ovs-system",
+            os.path.join(self.sysdir, "eth1", "upper_ovs-system"),
+        )
+
+        self.assertTrue(net.master_is_openvswitch("eth1"))
+
+    def test_is_vlan(self):
+        """is_vlan is True when /sys/net/devname/uevent has DEVTYPE=vlan."""
+        ensure_file(os.path.join(self.sysdir, "eth0", "uevent"))
+        self.assertFalse(net.is_vlan("eth0"))
+        content = "junk\nDEVTYPE=vlan\njunk\n"
+        write_file(os.path.join(self.sysdir, "eth0", "uevent"), content)
+        self.assertTrue(net.is_vlan("eth0"))
+
+
+class TestGenerateFallbackConfig(CiTestCase):
+    def setUp(self):
+        super(TestGenerateFallbackConfig, self).setUp()
+        sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
+        self.m_sys_path = sys_mock.start()
+        self.sysdir = self.tmp_dir() + "/"
+        self.m_sys_path.return_value = self.sysdir
+        self.addCleanup(sys_mock.stop)
+        self.add_patch(
+            "cloudinit.net.util.is_container",
+            "m_is_container",
+            return_value=False,
+        )
+        self.add_patch("cloudinit.net.util.udevadm_settle", "m_settle")
+        self.add_patch(
+            "cloudinit.net.is_netfailover", "m_netfail", return_value=False
+        )
+        self.add_patch(
+            "cloudinit.net.is_netfail_master",
+            "m_netfail_master",
+            return_value=False,
+        )
+
+    def test_generate_fallback_finds_connected_eth_with_mac(self):
+        """generate_fallback_config finds any connected device with a mac."""
+        write_file(os.path.join(self.sysdir, "eth0", "carrier"), "1")
+        write_file(os.path.join(self.sysdir, "eth1", "carrier"), "1")
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth1", "address"), mac)
+        expected = {
+            "ethernets": {
+                "eth1": {
+                    "match": {"macaddress": mac},
+                    "dhcp4": True,
+                    "set-name": "eth1",
+                }
+            },
+            "version": 2,
+        }
+        self.assertEqual(expected, net.generate_fallback_config())
+
+    def test_generate_fallback_finds_dormant_eth_with_mac(self):
+        """generate_fallback_config finds any dormant device with a mac."""
+        write_file(os.path.join(self.sysdir, "eth0", "dormant"), "1")
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth0", "address"), mac)
+        expected = {
+            "ethernets": {
+                "eth0": {
+                    "match": {"macaddress": mac},
+                    "dhcp4": True,
+                    "set-name": "eth0",
+                }
+            },
+            "version": 2,
+        }
+        self.assertEqual(expected, net.generate_fallback_config())
+
+    def test_generate_fallback_finds_eth_by_operstate(self):
+        """generate_fallback_config finds any dormant device with a mac."""
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth0", "address"), mac)
+        expected = {
+            "ethernets": {
+                "eth0": {
+                    "dhcp4": True,
+                    "match": {"macaddress": mac},
+                    "set-name": "eth0",
+                }
+            },
+            "version": 2,
+        }
+        valid_operstates = ["dormant", "down", "lowerlayerdown", "unknown"]
+        for state in valid_operstates:
+            write_file(os.path.join(self.sysdir, "eth0", "operstate"), state)
+            self.assertEqual(expected, net.generate_fallback_config())
+        write_file(os.path.join(self.sysdir, "eth0", "operstate"), "noworky")
+        self.assertIsNone(net.generate_fallback_config())
+
+    def test_generate_fallback_config_skips_veth(self):
+        """generate_fallback_config will skip any veth interfaces."""
+        # A connected veth which gets ignored
+        write_file(os.path.join(self.sysdir, "veth0", "carrier"), "1")
+        self.assertIsNone(net.generate_fallback_config())
+
+    def test_generate_fallback_config_skips_bridges(self):
+        """generate_fallback_config will skip any bridges interfaces."""
+        # A connected veth which gets ignored
+        write_file(os.path.join(self.sysdir, "eth0", "carrier"), "1")
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth0", "address"), mac)
+        ensure_file(os.path.join(self.sysdir, "eth0", "bridge"))
+        self.assertIsNone(net.generate_fallback_config())
+
+    def test_generate_fallback_config_skips_bonds(self):
+        """generate_fallback_config will skip any bonded interfaces."""
+        # A connected veth which gets ignored
+        write_file(os.path.join(self.sysdir, "eth0", "carrier"), "1")
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth0", "address"), mac)
+        ensure_file(os.path.join(self.sysdir, "eth0", "bonding"))
+        self.assertIsNone(net.generate_fallback_config())
+
+    def test_generate_fallback_config_skips_netfail_devs(self):
+        """gen_fallback_config ignores netfail primary,sby no mac on master."""
+        mac = "aa:bb:cc:aa:bb:cc"  # netfailover devs share the same mac
+        for iface in ["ens3", "ens3sby", "enP0s1f3"]:
+            write_file(os.path.join(self.sysdir, iface, "carrier"), "1")
+            write_file(
+                os.path.join(self.sysdir, iface, "addr_assign_type"), "0"
+            )
+            write_file(os.path.join(self.sysdir, iface, "address"), mac)
+
+        def is_netfail(iface, _driver=None):
+            # ens3 is the master
+            if iface == "ens3":
+                return False
+            return True
+
+        self.m_netfail.side_effect = is_netfail
+
+        def is_netfail_master(iface, _driver=None):
+            # ens3 is the master
+            if iface == "ens3":
+                return True
+            return False
+
+        self.m_netfail_master.side_effect = is_netfail_master
+        expected = {
+            "ethernets": {
+                "ens3": {
+                    "dhcp4": True,
+                    "match": {"name": "ens3"},
+                    "set-name": "ens3",
+                }
+            },
+            "version": 2,
+        }
+        result = net.generate_fallback_config()
+        self.assertEqual(expected, result)
+
+
+class TestNetFindFallBackNic(CiTestCase):
+    def setUp(self):
+        super(TestNetFindFallBackNic, self).setUp()
+        sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
+        self.m_sys_path = sys_mock.start()
+        self.sysdir = self.tmp_dir() + "/"
+        self.m_sys_path.return_value = self.sysdir
+        self.addCleanup(sys_mock.stop)
+        self.add_patch(
+            "cloudinit.net.util.is_container",
+            "m_is_container",
+            return_value=False,
+        )
+        self.add_patch("cloudinit.net.util.udevadm_settle", "m_settle")
+
+    def test_generate_fallback_finds_first_connected_eth_with_mac(self):
+        """find_fallback_nic finds any connected device with a mac."""
+        write_file(os.path.join(self.sysdir, "eth0", "carrier"), "1")
+        write_file(os.path.join(self.sysdir, "eth1", "carrier"), "1")
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth1", "address"), mac)
+        self.assertEqual("eth1", net.find_fallback_nic())
+
+
+class TestGetDeviceList(CiTestCase):
+    def setUp(self):
+        super(TestGetDeviceList, self).setUp()
+        sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
+        self.m_sys_path = sys_mock.start()
+        self.sysdir = self.tmp_dir() + "/"
+        self.m_sys_path.return_value = self.sysdir
+        self.addCleanup(sys_mock.stop)
+
+    def test_get_devicelist_raise_oserror(self):
+        """get_devicelist raise any non-ENOENT OSerror."""
+        error = OSError("Can not do it")
+        error.errno = errno.EPERM  # Set non-ENOENT
+        self.m_sys_path.side_effect = error
+        with self.assertRaises(OSError) as context_manager:
+            net.get_devicelist()
+        exception = context_manager.exception
+        self.assertEqual("Can not do it", str(exception))
+
+    def test_get_devicelist_empty_without_sys_net(self):
+        """get_devicelist returns empty list when missing SYS_CLASS_NET."""
+        self.m_sys_path.return_value = "idontexist"
+        self.assertEqual([], net.get_devicelist())
+
+    def test_get_devicelist_empty_with_no_devices_in_sys_net(self):
+        """get_devicelist returns empty directoty listing for SYS_CLASS_NET."""
+        self.assertEqual([], net.get_devicelist())
+
+    def test_get_devicelist_lists_any_subdirectories_in_sys_net(self):
+        """get_devicelist returns a directory listing for SYS_CLASS_NET."""
+        write_file(os.path.join(self.sysdir, "eth0", "operstate"), "up")
+        write_file(os.path.join(self.sysdir, "eth1", "operstate"), "up")
+        self.assertCountEqual(["eth0", "eth1"], net.get_devicelist())
+
+
+@mock.patch(
+    "cloudinit.net.is_openvswitch_internal_interface",
+    mock.Mock(return_value=False),
+)
+class TestGetInterfaceMAC(CiTestCase):
+    def setUp(self):
+        super(TestGetInterfaceMAC, self).setUp()
+        sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
+        self.m_sys_path = sys_mock.start()
+        self.sysdir = self.tmp_dir() + "/"
+        self.m_sys_path.return_value = self.sysdir
+        self.addCleanup(sys_mock.stop)
+
+    def test_get_interface_mac_false_with_no_mac(self):
+        """get_device_list returns False when no mac is reported."""
+        ensure_file(os.path.join(self.sysdir, "eth0", "bonding"))
+        mac_path = os.path.join(self.sysdir, "eth0", "address")
+        self.assertFalse(os.path.exists(mac_path))
+        self.assertFalse(net.get_interface_mac("eth0"))
+
+    def test_get_interface_mac(self):
+        """get_interfaces returns the mac from SYS_CLASS_NET/dev/address."""
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth1", "address"), mac)
+        self.assertEqual(mac, net.get_interface_mac("eth1"))
+
+    def test_get_interface_mac_grabs_bonding_address(self):
+        """get_interfaces returns the source device mac for bonded devices."""
+        source_dev_mac = "aa:bb:cc:aa:bb:cc"
+        bonded_mac = "dd:ee:ff:dd:ee:ff"
+        write_file(os.path.join(self.sysdir, "eth1", "address"), bonded_mac)
+        write_file(
+            os.path.join(self.sysdir, "eth1", "bonding_slave", "perm_hwaddr"),
+            source_dev_mac,
+        )
+        self.assertEqual(source_dev_mac, net.get_interface_mac("eth1"))
+
+    def test_get_interfaces_empty_list_without_sys_net(self):
+        """get_interfaces returns an empty list when missing SYS_CLASS_NET."""
+        self.m_sys_path.return_value = "idontexist"
+        self.assertEqual([], net.get_interfaces())
+
+    def test_get_interfaces_by_mac_skips_empty_mac(self):
+        """Ignore 00:00:00:00:00:00 addresses from get_interfaces_by_mac."""
+        empty_mac = "00:00:00:00:00:00"
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth1", "address"), empty_mac)
+        write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "0")
+        write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0")
+        write_file(os.path.join(self.sysdir, "eth2", "address"), mac)
+        expected = [("eth2", "aa:bb:cc:aa:bb:cc", None, None)]
+        self.assertEqual(expected, net.get_interfaces())
+
+    def test_get_interfaces_by_mac_skips_missing_mac(self):
+        """Ignore interfaces without an address from get_interfaces_by_mac."""
+        write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "0")
+        address_path = os.path.join(self.sysdir, "eth1", "address")
+        self.assertFalse(os.path.exists(address_path))
+        mac = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0")
+        write_file(os.path.join(self.sysdir, "eth2", "address"), mac)
+        expected = [("eth2", "aa:bb:cc:aa:bb:cc", None, None)]
+        self.assertEqual(expected, net.get_interfaces())
+
+    def test_get_interfaces_by_mac_skips_master_devs(self):
+        """Ignore interfaces with a master device which would have dup mac."""
+        mac1 = mac2 = "aa:bb:cc:aa:bb:cc"
+        write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "0")
+        write_file(os.path.join(self.sysdir, "eth1", "address"), mac1)
+        write_file(os.path.join(self.sysdir, "eth1", "master"), "blah")
+        write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0")
+        write_file(os.path.join(self.sysdir, "eth2", "address"), mac2)
+        expected = [("eth2", mac2, None, None)]
+        self.assertEqual(expected, net.get_interfaces())
+
+    @mock.patch("cloudinit.net.is_netfailover")
+    def test_get_interfaces_by_mac_skips_netfailvoer(self, m_netfail):
+        """Ignore interfaces if netfailover primary or standby."""
+        mac = "aa:bb:cc:aa:bb:cc"  # netfailover devs share the same mac
+        for iface in ["ens3", "ens3sby", "enP0s1f3"]:
+            write_file(
+                os.path.join(self.sysdir, iface, "addr_assign_type"), "0"
+            )
+            write_file(os.path.join(self.sysdir, iface, "address"), mac)
+
+        def is_netfail(iface, _driver=None):
+            # ens3 is the master
+            if iface == "ens3":
+                return False
+            else:
+                return True
+
+        m_netfail.side_effect = is_netfail
+        expected = [("ens3", mac, None, None)]
+        self.assertEqual(expected, net.get_interfaces())
+
+    def test_get_interfaces_does_not_skip_phys_members_of_bridges_and_bonds(
+        self,
+    ):
+        bridge_mac = "aa:bb:cc:aa:bb:cc"
+        bond_mac = "cc:bb:aa:cc:bb:aa"
+        ovs_mac = "bb:cc:aa:bb:cc:aa"
+
+        write_file(os.path.join(self.sysdir, "br0", "address"), bridge_mac)
+        write_file(os.path.join(self.sysdir, "br0", "bridge"), "")
+
+        write_file(os.path.join(self.sysdir, "bond0", "address"), bond_mac)
+        write_file(os.path.join(self.sysdir, "bond0", "bonding"), "")
+
+        write_file(os.path.join(self.sysdir, "ovs-system", "address"), ovs_mac)
+
+        write_file(os.path.join(self.sysdir, "eth1", "address"), bridge_mac)
+        os.symlink("../br0", os.path.join(self.sysdir, "eth1", "master"))
+
+        write_file(os.path.join(self.sysdir, "eth2", "address"), bond_mac)
+        os.symlink("../bond0", os.path.join(self.sysdir, "eth2", "master"))
+
+        write_file(os.path.join(self.sysdir, "eth3", "address"), ovs_mac)
+        os.symlink(
+            "../ovs-system", os.path.join(self.sysdir, "eth3", "master")
+        )
+        os.symlink(
+            "../ovs-system",
+            os.path.join(self.sysdir, "eth3", "upper_ovs-system"),
+        )
+
+        interface_names = [interface[0] for interface in net.get_interfaces()]
+        self.assertEqual(
+            ["eth1", "eth2", "eth3", "ovs-system"], sorted(interface_names)
+        )
+
+
+class TestInterfaceHasOwnMAC(CiTestCase):
+    def setUp(self):
+        super(TestInterfaceHasOwnMAC, self).setUp()
+        sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
+        self.m_sys_path = sys_mock.start()
+        self.sysdir = self.tmp_dir() + "/"
+        self.m_sys_path.return_value = self.sysdir
+        self.addCleanup(sys_mock.stop)
+
+    def test_interface_has_own_mac_false_when_stolen(self):
+        """Return False from interface_has_own_mac when address is stolen."""
+        write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "2")
+        self.assertFalse(net.interface_has_own_mac("eth1"))
+
+    def test_interface_has_own_mac_true_when_not_stolen(self):
+        """Return False from interface_has_own_mac when mac isn't stolen."""
+        valid_assign_types = ["0", "1", "3"]
+        assign_path = os.path.join(self.sysdir, "eth1", "addr_assign_type")
+        for _type in valid_assign_types:
+            write_file(assign_path, _type)
+            self.assertTrue(net.interface_has_own_mac("eth1"))
+
+    def test_interface_has_own_mac_strict_errors_on_absent_assign_type(self):
+        """When addr_assign_type is absent, interface_has_own_mac errors."""
+        with self.assertRaises(ValueError):
+            net.interface_has_own_mac("eth1", strict=True)
+
+
+@mock.patch("cloudinit.net.subp.subp")
+class TestEphemeralIPV4Network(CiTestCase):
+
+    with_logs = True
+
+    def setUp(self):
+        super(TestEphemeralIPV4Network, self).setUp()
+        sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
+        self.m_sys_path = sys_mock.start()
+        self.sysdir = self.tmp_dir() + "/"
+        self.m_sys_path.return_value = self.sysdir
+        self.addCleanup(sys_mock.stop)
+
+    def test_ephemeral_ipv4_network_errors_on_missing_params(self, m_subp):
+        """No required params for EphemeralIPv4Network can be None."""
+        required_params = {
+            "interface": "eth0",
+            "ip": "192.168.2.2",
+            "prefix_or_mask": "255.255.255.0",
+            "broadcast": "192.168.2.255",
+        }
+        for key in required_params.keys():
+            params = copy.deepcopy(required_params)
+            params[key] = None
+            with self.assertRaises(ValueError) as context_manager:
+                net.EphemeralIPv4Network(**params)
+            error = context_manager.exception
+            self.assertIn("Cannot init network on", str(error))
+            self.assertEqual(0, m_subp.call_count)
+
+    def test_ephemeral_ipv4_network_errors_invalid_mask_prefix(self, m_subp):
+        """Raise an error when prefix_or_mask is not a netmask or prefix."""
+        params = {
+            "interface": "eth0",
+            "ip": "192.168.2.2",
+            "broadcast": "192.168.2.255",
+        }
+        invalid_masks = ("invalid", "invalid.", "123.123.123")
+        for error_val in invalid_masks:
+            params["prefix_or_mask"] = error_val
+            with self.assertRaises(ValueError) as context_manager:
+                with net.EphemeralIPv4Network(**params):
+                    pass
+            error = context_manager.exception
+            self.assertIn(
+                "Cannot setup network, invalid prefix or netmask: ", str(error)
+            )
+            self.assertEqual(0, m_subp.call_count)
+
+    def test_ephemeral_ipv4_network_performs_teardown(self, m_subp):
+        """EphemeralIPv4Network performs teardown on the device if setup."""
+        expected_setup_calls = [
+            mock.call(
+                [
+                    "ip",
+                    "-family",
+                    "inet",
+                    "addr",
+                    "add",
+                    "192.168.2.2/24",
+                    "broadcast",
+                    "192.168.2.255",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+                update_env={"LANG": "C"},
+            ),
+            mock.call(
+                ["ip", "-family", "inet", "link", "set", "dev", "eth0", "up"],
+                capture=True,
+            ),
+        ]
+        expected_teardown_calls = [
+            mock.call(
+                [
+                    "ip",
+                    "-family",
+                    "inet",
+                    "link",
+                    "set",
+                    "dev",
+                    "eth0",
+                    "down",
+                ],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-family",
+                    "inet",
+                    "addr",
+                    "del",
+                    "192.168.2.2/24",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+            ),
+        ]
+        params = {
+            "interface": "eth0",
+            "ip": "192.168.2.2",
+            "prefix_or_mask": "255.255.255.0",
+            "broadcast": "192.168.2.255",
+        }
+        with net.EphemeralIPv4Network(**params):
+            self.assertEqual(expected_setup_calls, m_subp.call_args_list)
+        m_subp.assert_has_calls(expected_teardown_calls)
+
+    @mock.patch("cloudinit.net.readurl")
+    def test_ephemeral_ipv4_no_network_if_url_connectivity(
+        self, m_readurl, m_subp
+    ):
+        """No network setup is performed if we can successfully connect to
+        connectivity_url."""
+        params = {
+            "interface": "eth0",
+            "ip": "192.168.2.2",
+            "prefix_or_mask": "255.255.255.0",
+            "broadcast": "192.168.2.255",
+            "connectivity_url_data": {"url": "http://example.org/index.html"},
+        }
+
+        with net.EphemeralIPv4Network(**params):
+            self.assertEqual(
+                [mock.call(url="http://example.org/index.html", timeout=5)],
+                m_readurl.call_args_list,
+            )
+        # Ensure that no teardown happens:
+        m_subp.assert_has_calls([])
+
+    def test_ephemeral_ipv4_network_noop_when_configured(self, m_subp):
+        """EphemeralIPv4Network handles exception when address is setup.
+
+        It performs no cleanup as the interface was already setup.
+        """
+        params = {
+            "interface": "eth0",
+            "ip": "192.168.2.2",
+            "prefix_or_mask": "255.255.255.0",
+            "broadcast": "192.168.2.255",
+        }
+        m_subp.side_effect = ProcessExecutionError(
+            "", "RTNETLINK answers: File exists", 2
+        )
+        expected_calls = [
+            mock.call(
+                [
+                    "ip",
+                    "-family",
+                    "inet",
+                    "addr",
+                    "add",
+                    "192.168.2.2/24",
+                    "broadcast",
+                    "192.168.2.255",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+                update_env={"LANG": "C"},
+            )
+        ]
+        with net.EphemeralIPv4Network(**params):
+            pass
+        self.assertEqual(expected_calls, m_subp.call_args_list)
+        self.assertIn(
+            "Skip ephemeral network setup, eth0 already has address",
+            self.logs.getvalue(),
+        )
+
+    def test_ephemeral_ipv4_network_with_prefix(self, m_subp):
+        """EphemeralIPv4Network takes a valid prefix to setup the network."""
+        params = {
+            "interface": "eth0",
+            "ip": "192.168.2.2",
+            "prefix_or_mask": "24",
+            "broadcast": "192.168.2.255",
+        }
+        for prefix_val in ["24", 16]:  # prefix can be int or string
+            params["prefix_or_mask"] = prefix_val
+            with net.EphemeralIPv4Network(**params):
+                pass
+        m_subp.assert_has_calls(
+            [
+                mock.call(
+                    [
+                        "ip",
+                        "-family",
+                        "inet",
+                        "addr",
+                        "add",
+                        "192.168.2.2/24",
+                        "broadcast",
+                        "192.168.2.255",
+                        "dev",
+                        "eth0",
+                    ],
+                    capture=True,
+                    update_env={"LANG": "C"},
+                )
+            ]
+        )
+        m_subp.assert_has_calls(
+            [
+                mock.call(
+                    [
+                        "ip",
+                        "-family",
+                        "inet",
+                        "addr",
+                        "add",
+                        "192.168.2.2/16",
+                        "broadcast",
+                        "192.168.2.255",
+                        "dev",
+                        "eth0",
+                    ],
+                    capture=True,
+                    update_env={"LANG": "C"},
+                )
+            ]
+        )
+
+    def test_ephemeral_ipv4_network_with_new_default_route(self, m_subp):
+        """Add the route when router is set and no default route exists."""
+        params = {
+            "interface": "eth0",
+            "ip": "192.168.2.2",
+            "prefix_or_mask": "255.255.255.0",
+            "broadcast": "192.168.2.255",
+            "router": "192.168.2.1",
+        }
+        m_subp.return_value = "", ""  # Empty response from ip route gw check
+        expected_setup_calls = [
+            mock.call(
+                [
+                    "ip",
+                    "-family",
+                    "inet",
+                    "addr",
+                    "add",
+                    "192.168.2.2/24",
+                    "broadcast",
+                    "192.168.2.255",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+                update_env={"LANG": "C"},
+            ),
+            mock.call(
+                ["ip", "-family", "inet", "link", "set", "dev", "eth0", "up"],
+                capture=True,
+            ),
+            mock.call(["ip", "route", "show", "0.0.0.0/0"], capture=True),
+            mock.call(
+                [
+                    "ip",
+                    "-4",
+                    "route",
+                    "add",
+                    "192.168.2.1",
+                    "dev",
+                    "eth0",
+                    "src",
+                    "192.168.2.2",
+                ],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-4",
+                    "route",
+                    "add",
+                    "default",
+                    "via",
+                    "192.168.2.1",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+            ),
+        ]
+        expected_teardown_calls = [
+            mock.call(
+                ["ip", "-4", "route", "del", "default", "dev", "eth0"],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-4",
+                    "route",
+                    "del",
+                    "192.168.2.1",
+                    "dev",
+                    "eth0",
+                    "src",
+                    "192.168.2.2",
+                ],
+                capture=True,
+            ),
+        ]
+
+        with net.EphemeralIPv4Network(**params):
+            self.assertEqual(expected_setup_calls, m_subp.call_args_list)
+        m_subp.assert_has_calls(expected_teardown_calls)
+
+    def test_ephemeral_ipv4_network_with_rfc3442_static_routes(self, m_subp):
+        params = {
+            "interface": "eth0",
+            "ip": "192.168.2.2",
+            "prefix_or_mask": "255.255.255.255",
+            "broadcast": "192.168.2.255",
+            "static_routes": [
+                ("192.168.2.1/32", "0.0.0.0"),
+                ("169.254.169.254/32", "192.168.2.1"),
+                ("0.0.0.0/0", "192.168.2.1"),
+            ],
+            "router": "192.168.2.1",
+        }
+        expected_setup_calls = [
+            mock.call(
+                [
+                    "ip",
+                    "-family",
+                    "inet",
+                    "addr",
+                    "add",
+                    "192.168.2.2/32",
+                    "broadcast",
+                    "192.168.2.255",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+                update_env={"LANG": "C"},
+            ),
+            mock.call(
+                ["ip", "-family", "inet", "link", "set", "dev", "eth0", "up"],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-4",
+                    "route",
+                    "append",
+                    "192.168.2.1/32",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-4",
+                    "route",
+                    "append",
+                    "169.254.169.254/32",
+                    "via",
+                    "192.168.2.1",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-4",
+                    "route",
+                    "append",
+                    "0.0.0.0/0",
+                    "via",
+                    "192.168.2.1",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+            ),
+        ]
+        expected_teardown_calls = [
+            mock.call(
+                [
+                    "ip",
+                    "-4",
+                    "route",
+                    "del",
+                    "0.0.0.0/0",
+                    "via",
+                    "192.168.2.1",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-4",
+                    "route",
+                    "del",
+                    "169.254.169.254/32",
+                    "via",
+                    "192.168.2.1",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+            ),
+            mock.call(
+                ["ip", "-4", "route", "del", "192.168.2.1/32", "dev", "eth0"],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-family",
+                    "inet",
+                    "link",
+                    "set",
+                    "dev",
+                    "eth0",
+                    "down",
+                ],
+                capture=True,
+            ),
+            mock.call(
+                [
+                    "ip",
+                    "-family",
+                    "inet",
+                    "addr",
+                    "del",
+                    "192.168.2.2/32",
+                    "dev",
+                    "eth0",
+                ],
+                capture=True,
+            ),
+        ]
+        with net.EphemeralIPv4Network(**params):
+            self.assertEqual(expected_setup_calls, m_subp.call_args_list)
+        m_subp.assert_has_calls(expected_setup_calls + expected_teardown_calls)
+
+
+class TestApplyNetworkCfgNames(CiTestCase):
+    V1_CONFIG = textwrap.dedent(
+        """\
+        version: 1
+        config:
+            - type: physical
+              name: interface0
+              mac_address: "52:54:00:12:34:00"
+              subnets:
+                  - type: static
+                    address: 10.0.2.15
+                    netmask: 255.255.255.0
+                    gateway: 10.0.2.2
+    """
+    )
+    V2_CONFIG = textwrap.dedent(
+        """\
+      version: 2
+      ethernets:
+          interface0:
+            match:
+              macaddress: "52:54:00:12:34:00"
+            addresses:
+              - 10.0.2.15/24
+            gateway4: 10.0.2.2
+            set-name: interface0
+    """
+    )
+
+    V2_CONFIG_NO_SETNAME = textwrap.dedent(
+        """\
+      version: 2
+      ethernets:
+          interface0:
+            match:
+              macaddress: "52:54:00:12:34:00"
+            addresses:
+              - 10.0.2.15/24
+            gateway4: 10.0.2.2
+    """
+    )
+
+    V2_CONFIG_NO_MAC = textwrap.dedent(
+        """\
+      version: 2
+      ethernets:
+          interface0:
+            match:
+              driver: virtio-net
+            addresses:
+              - 10.0.2.15/24
+            gateway4: 10.0.2.2
+            set-name: interface0
+    """
+    )
+
+    @mock.patch("cloudinit.net.device_devid")
+    @mock.patch("cloudinit.net.device_driver")
+    @mock.patch("cloudinit.net._rename_interfaces")
+    def test_apply_v1_renames(
+        self, m_rename_interfaces, m_device_driver, m_device_devid
+    ):
+        m_device_driver.return_value = "virtio_net"
+        m_device_devid.return_value = "0x15d8"
+
+        net.apply_network_config_names(yaml.load(self.V1_CONFIG))
+
+        call = ["52:54:00:12:34:00", "interface0", "virtio_net", "0x15d8"]
+        m_rename_interfaces.assert_called_with([call])
+
+    @mock.patch("cloudinit.net.device_devid")
+    @mock.patch("cloudinit.net.device_driver")
+    @mock.patch("cloudinit.net._rename_interfaces")
+    def test_apply_v2_renames(
+        self, m_rename_interfaces, m_device_driver, m_device_devid
+    ):
+        m_device_driver.return_value = "virtio_net"
+        m_device_devid.return_value = "0x15d8"
+
+        net.apply_network_config_names(yaml.load(self.V2_CONFIG))
+
+        call = ["52:54:00:12:34:00", "interface0", "virtio_net", "0x15d8"]
+        m_rename_interfaces.assert_called_with([call])
+
+    @mock.patch("cloudinit.net._rename_interfaces")
+    def test_apply_v2_renames_skips_without_setname(self, m_rename_interfaces):
+        net.apply_network_config_names(yaml.load(self.V2_CONFIG_NO_SETNAME))
+        m_rename_interfaces.assert_called_with([])
+
+    @mock.patch("cloudinit.net._rename_interfaces")
+    def test_apply_v2_renames_skips_without_mac(self, m_rename_interfaces):
+        net.apply_network_config_names(yaml.load(self.V2_CONFIG_NO_MAC))
+        m_rename_interfaces.assert_called_with([])
+
+    def test_apply_v2_renames_raises_runtime_error_on_unknown_version(self):
+        with self.assertRaises(RuntimeError):
+            net.apply_network_config_names(yaml.load("version: 3"))
+
+
+class TestHasURLConnectivity(HttprettyTestCase):
+    def setUp(self):
+        super(TestHasURLConnectivity, self).setUp()
+        self.url = "http://fake/"
+        self.kwargs = {"allow_redirects": True, "timeout": 5.0}
+
+    @mock.patch("cloudinit.net.readurl")
+    def test_url_timeout_on_connectivity_check(self, m_readurl):
+        """A timeout of 5 seconds is provided when reading a url."""
+        self.assertTrue(
+            net.has_url_connectivity({"url": self.url}),
+            "Expected True on url connect",
+        )
+
+    def test_true_on_url_connectivity_success(self):
+        httpretty.register_uri(httpretty.GET, self.url)
+        self.assertTrue(
+            net.has_url_connectivity({"url": self.url}),
+            "Expected True on url connect",
+        )
+
+    @mock.patch("requests.Session.request")
+    def test_true_on_url_connectivity_timeout(self, m_request):
+        """A timeout raised accessing the url will return False."""
+        m_request.side_effect = requests.Timeout("Fake Connection Timeout")
+        self.assertFalse(
+            net.has_url_connectivity({"url": self.url}),
+            "Expected False on url timeout",
+        )
+
+    def test_true_on_url_connectivity_failure(self):
+        httpretty.register_uri(httpretty.GET, self.url, body={}, status=404)
+        self.assertFalse(
+            net.has_url_connectivity({"url": self.url}),
+            "Expected False on url fail",
+        )
+
+
+def _mk_v1_phys(mac, name, driver, device_id):
+    v1_cfg = {"type": "physical", "name": name, "mac_address": mac}
+    params = {}
+    if driver:
+        params.update({"driver": driver})
+    if device_id:
+        params.update({"device_id": device_id})
+
+    if params:
+        v1_cfg.update({"params": params})
+
+    return v1_cfg
+
+
+def _mk_v2_phys(mac, name, driver=None, device_id=None):
+    v2_cfg = {"set-name": name, "match": {"macaddress": mac}}
+    if driver:
+        v2_cfg["match"].update({"driver": driver})
+    if device_id:
+        v2_cfg["match"].update({"device_id": device_id})
+
+    return v2_cfg
+
+
+class TestExtractPhysdevs(CiTestCase):
+    def setUp(self):
+        super(TestExtractPhysdevs, self).setUp()
+        self.add_patch("cloudinit.net.device_driver", "m_driver")
+        self.add_patch("cloudinit.net.device_devid", "m_devid")
+
+    def test_extract_physdevs_looks_up_driver_v1(self):
+        driver = "virtio"
+        self.m_driver.return_value = driver
+        physdevs = [
+            ["aa:bb:cc:dd:ee:ff", "eth0", None, "0x1000"],
+        ]
+        netcfg = {
+            "version": 1,
+            "config": [_mk_v1_phys(*args) for args in physdevs],
+        }
+        # insert the driver value for verification
+        physdevs[0][2] = driver
+        self.assertEqual(
+            sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+        )
+        self.m_driver.assert_called_with("eth0")
+
+    def test_extract_physdevs_looks_up_driver_v2(self):
+        driver = "virtio"
+        self.m_driver.return_value = driver
+        physdevs = [
+            ["aa:bb:cc:dd:ee:ff", "eth0", None, "0x1000"],
+        ]
+        netcfg = {
+            "version": 2,
+            "ethernets": {args[1]: _mk_v2_phys(*args) for args in physdevs},
+        }
+        # insert the driver value for verification
+        physdevs[0][2] = driver
+        self.assertEqual(
+            sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+        )
+        self.m_driver.assert_called_with("eth0")
+
+    def test_extract_physdevs_looks_up_devid_v1(self):
+        devid = "0x1000"
+        self.m_devid.return_value = devid
+        physdevs = [
+            ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", None],
+        ]
+        netcfg = {
+            "version": 1,
+            "config": [_mk_v1_phys(*args) for args in physdevs],
+        }
+        # insert the driver value for verification
+        physdevs[0][3] = devid
+        self.assertEqual(
+            sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+        )
+        self.m_devid.assert_called_with("eth0")
+
+    def test_extract_physdevs_looks_up_devid_v2(self):
+        devid = "0x1000"
+        self.m_devid.return_value = devid
+        physdevs = [
+            ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", None],
+        ]
+        netcfg = {
+            "version": 2,
+            "ethernets": {args[1]: _mk_v2_phys(*args) for args in physdevs},
+        }
+        # insert the driver value for verification
+        physdevs[0][3] = devid
+        self.assertEqual(
+            sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+        )
+        self.m_devid.assert_called_with("eth0")
+
+    def test_get_v1_type_physical(self):
+        physdevs = [
+            ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", "0x1000"],
+            ["00:11:22:33:44:55", "ens3", "e1000", "0x1643"],
+            ["09:87:65:43:21:10", "ens0p1", "mlx4_core", "0:0:1000"],
+        ]
+        netcfg = {
+            "version": 1,
+            "config": [_mk_v1_phys(*args) for args in physdevs],
+        }
+        self.assertEqual(
+            sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+        )
+
+    def test_get_v2_type_physical(self):
+        physdevs = [
+            ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", "0x1000"],
+            ["00:11:22:33:44:55", "ens3", "e1000", "0x1643"],
+            ["09:87:65:43:21:10", "ens0p1", "mlx4_core", "0:0:1000"],
+        ]
+        netcfg = {
+            "version": 2,
+            "ethernets": {args[1]: _mk_v2_phys(*args) for args in physdevs},
+        }
+        self.assertEqual(
+            sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+        )
+
+    def test_get_v2_type_physical_skips_if_no_set_name(self):
+        netcfg = {
+            "version": 2,
+            "ethernets": {
+                "ens3": {
+                    "match": {"macaddress": "00:11:22:33:44:55"},
+                }
+            },
+        }
+        self.assertEqual([], net.extract_physdevs(netcfg))
+
+    def test_runtime_error_on_unknown_netcfg_version(self):
+        with self.assertRaises(RuntimeError):
+            net.extract_physdevs({"version": 3, "awesome_config": []})
+
+
+class TestNetFailOver(CiTestCase):
+    def setUp(self):
+        super(TestNetFailOver, self).setUp()
+        self.add_patch("cloudinit.net.util", "m_util")
+        self.add_patch("cloudinit.net.read_sys_net", "m_read_sys_net")
+        self.add_patch("cloudinit.net.device_driver", "m_device_driver")
+
+    def test_get_dev_features(self):
+        devname = self.random_string()
+        features = self.random_string()
+        self.m_read_sys_net.return_value = features
+
+        self.assertEqual(features, net.get_dev_features(devname))
+        self.assertEqual(1, self.m_read_sys_net.call_count)
+        self.assertEqual(
+            mock.call(devname, "device/features"),
+            self.m_read_sys_net.call_args_list[0],
+        )
+
+    def test_get_dev_features_none_returns_empty_string(self):
+        devname = self.random_string()
+        self.m_read_sys_net.side_effect = Exception("error")
+        self.assertEqual("", net.get_dev_features(devname))
+        self.assertEqual(1, self.m_read_sys_net.call_count)
+        self.assertEqual(
+            mock.call(devname, "device/features"),
+            self.m_read_sys_net.call_args_list[0],
+        )
+
+    @mock.patch("cloudinit.net.get_dev_features")
+    def test_has_netfail_standby_feature(self, m_dev_features):
+        devname = self.random_string()
+        standby_features = ("0" * 62) + "1" + "0"
+        m_dev_features.return_value = standby_features
+        self.assertTrue(net.has_netfail_standby_feature(devname))
+
+    @mock.patch("cloudinit.net.get_dev_features")
+    def test_has_netfail_standby_feature_short_is_false(self, m_dev_features):
+        devname = self.random_string()
+        standby_features = self.random_string()
+        m_dev_features.return_value = standby_features
+        self.assertFalse(net.has_netfail_standby_feature(devname))
+
+    @mock.patch("cloudinit.net.get_dev_features")
+    def test_has_netfail_standby_feature_not_present_is_false(
+        self, m_dev_features
+    ):
+        devname = self.random_string()
+        standby_features = "0" * 64
+        m_dev_features.return_value = standby_features
+        self.assertFalse(net.has_netfail_standby_feature(devname))
+
+    @mock.patch("cloudinit.net.get_dev_features")
+    def test_has_netfail_standby_feature_no_features_is_false(
+        self, m_dev_features
+    ):
+        devname = self.random_string()
+        standby_features = None
+        m_dev_features.return_value = standby_features
+        self.assertFalse(net.has_netfail_standby_feature(devname))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    def test_is_netfail_master(self, m_exists, m_standby):
+        devname = self.random_string()
+        driver = "virtio_net"
+        m_exists.return_value = False  # no master sysfs attr
+        m_standby.return_value = True  # has standby feature flag
+        self.assertTrue(net.is_netfail_master(devname, driver))
+
+    @mock.patch("cloudinit.net.sys_dev_path")
+    def test_is_netfail_master_checks_master_attr(self, m_sysdev):
+        devname = self.random_string()
+        driver = "virtio_net"
+        m_sysdev.return_value = self.random_string()
+        self.assertFalse(net.is_netfail_master(devname, driver))
+        self.assertEqual(1, m_sysdev.call_count)
+        self.assertEqual(
+            mock.call(devname, path="master"), m_sysdev.call_args_list[0]
+        )
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    def test_is_netfail_master_wrong_driver(self, m_exists, m_standby):
+        devname = self.random_string()
+        driver = self.random_string()
+        self.assertFalse(net.is_netfail_master(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    def test_is_netfail_master_has_master_attr(self, m_exists, m_standby):
+        devname = self.random_string()
+        driver = "virtio_net"
+        m_exists.return_value = True  # has master sysfs attr
+        self.assertFalse(net.is_netfail_master(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    def test_is_netfail_master_no_standby_feat(self, m_exists, m_standby):
+        devname = self.random_string()
+        driver = "virtio_net"
+        m_exists.return_value = False  # no master sysfs attr
+        m_standby.return_value = False  # no standby feature flag
+        self.assertFalse(net.is_netfail_master(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    @mock.patch("cloudinit.net.sys_dev_path")
+    def test_is_netfail_primary(self, m_sysdev, m_exists, m_standby):
+        devname = self.random_string()
+        driver = self.random_string()  # device not virtio_net
+        master_devname = self.random_string()
+        m_sysdev.return_value = "%s/%s" % (
+            self.random_string(),
+            master_devname,
+        )
+        m_exists.return_value = True  # has master sysfs attr
+        self.m_device_driver.return_value = "virtio_net"  # master virtio_net
+        m_standby.return_value = True  # has standby feature flag
+        self.assertTrue(net.is_netfail_primary(devname, driver))
+        self.assertEqual(1, self.m_device_driver.call_count)
+        self.assertEqual(
+            mock.call(master_devname), self.m_device_driver.call_args_list[0]
+        )
+        self.assertEqual(1, m_standby.call_count)
+        self.assertEqual(
+            mock.call(master_devname), m_standby.call_args_list[0]
+        )
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    @mock.patch("cloudinit.net.sys_dev_path")
+    def test_is_netfail_primary_wrong_driver(
+        self, m_sysdev, m_exists, m_standby
+    ):
+        devname = self.random_string()
+        driver = "virtio_net"
+        self.assertFalse(net.is_netfail_primary(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    @mock.patch("cloudinit.net.sys_dev_path")
+    def test_is_netfail_primary_no_master(self, m_sysdev, m_exists, m_standby):
+        devname = self.random_string()
+        driver = self.random_string()  # device not virtio_net
+        m_exists.return_value = False  # no master sysfs attr
+        self.assertFalse(net.is_netfail_primary(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    @mock.patch("cloudinit.net.sys_dev_path")
+    def test_is_netfail_primary_bad_master(
+        self, m_sysdev, m_exists, m_standby
+    ):
+        devname = self.random_string()
+        driver = self.random_string()  # device not virtio_net
+        master_devname = self.random_string()
+        m_sysdev.return_value = "%s/%s" % (
+            self.random_string(),
+            master_devname,
+        )
+        m_exists.return_value = True  # has master sysfs attr
+        self.m_device_driver.return_value = "XXXX"  # master not virtio_net
+        self.assertFalse(net.is_netfail_primary(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    @mock.patch("cloudinit.net.sys_dev_path")
+    def test_is_netfail_primary_no_standby(
+        self, m_sysdev, m_exists, m_standby
+    ):
+        devname = self.random_string()
+        driver = self.random_string()  # device not virtio_net
+        master_devname = self.random_string()
+        m_sysdev.return_value = "%s/%s" % (
+            self.random_string(),
+            master_devname,
+        )
+        m_exists.return_value = True  # has master sysfs attr
+        self.m_device_driver.return_value = "virtio_net"  # master virtio_net
+        m_standby.return_value = False  # master has no standby feature flag
+        self.assertFalse(net.is_netfail_primary(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    def test_is_netfail_standby(self, m_exists, m_standby):
+        devname = self.random_string()
+        driver = "virtio_net"
+        m_exists.return_value = True  # has master sysfs attr
+        m_standby.return_value = True  # has standby feature flag
+        self.assertTrue(net.is_netfail_standby(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    def test_is_netfail_standby_wrong_driver(self, m_exists, m_standby):
+        devname = self.random_string()
+        driver = self.random_string()
+        self.assertFalse(net.is_netfail_standby(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    def test_is_netfail_standby_no_master(self, m_exists, m_standby):
+        devname = self.random_string()
+        driver = "virtio_net"
+        m_exists.return_value = False  # has master sysfs attr
+        self.assertFalse(net.is_netfail_standby(devname, driver))
+
+    @mock.patch("cloudinit.net.has_netfail_standby_feature")
+    @mock.patch("cloudinit.net.os.path.exists")
+    def test_is_netfail_standby_no_standby_feature(self, m_exists, m_standby):
+        devname = self.random_string()
+        driver = "virtio_net"
+        m_exists.return_value = True  # has master sysfs attr
+        m_standby.return_value = False  # has standby feature flag
+        self.assertFalse(net.is_netfail_standby(devname, driver))
+
+    @mock.patch("cloudinit.net.is_netfail_standby")
+    @mock.patch("cloudinit.net.is_netfail_primary")
+    def test_is_netfailover_primary(self, m_primary, m_standby):
+        devname = self.random_string()
+        driver = self.random_string()
+        m_primary.return_value = True
+        m_standby.return_value = False
+        self.assertTrue(net.is_netfailover(devname, driver))
+
+    @mock.patch("cloudinit.net.is_netfail_standby")
+    @mock.patch("cloudinit.net.is_netfail_primary")
+    def test_is_netfailover_standby(self, m_primary, m_standby):
+        devname = self.random_string()
+        driver = self.random_string()
+        m_primary.return_value = False
+        m_standby.return_value = True
+        self.assertTrue(net.is_netfailover(devname, driver))
+
+    @mock.patch("cloudinit.net.is_netfail_standby")
+    @mock.patch("cloudinit.net.is_netfail_primary")
+    def test_is_netfailover_returns_false(self, m_primary, m_standby):
+        devname = self.random_string()
+        driver = self.random_string()
+        m_primary.return_value = False
+        m_standby.return_value = False
+        self.assertFalse(net.is_netfailover(devname, driver))
+
+
+class TestOpenvswitchIsInstalled:
+    """Test cloudinit.net.openvswitch_is_installed.
+
+    Uses the ``clear_lru_cache`` local autouse fixture to allow us to test
+    despite the ``lru_cache`` decorator on the unit under test.
+    """
+
+    @pytest.fixture(autouse=True)
+    def clear_lru_cache(self):
+        net.openvswitch_is_installed.cache_clear()
+
+    @pytest.mark.parametrize(
+        "expected,which_return", [(True, "/some/path"), (False, None)]
+    )
+    @mock.patch("cloudinit.net.subp.which")
+    def test_mirrors_which_result(self, m_which, expected, which_return):
+        m_which.return_value = which_return
+        assert expected == net.openvswitch_is_installed()
+
+    @mock.patch("cloudinit.net.subp.which")
+    def test_only_calls_which_once(self, m_which):
+        net.openvswitch_is_installed()
+        net.openvswitch_is_installed()
+        assert 1 == m_which.call_count
+
+
+@mock.patch("cloudinit.net.subp.subp", return_value=("", ""))
+class TestGetOVSInternalInterfaces:
+    """Test cloudinit.net.get_ovs_internal_interfaces.
+
+    Uses the ``clear_lru_cache`` local autouse fixture to allow us to test
+    despite the ``lru_cache`` decorator on the unit under test.
+    """
+
+    @pytest.fixture(autouse=True)
+    def clear_lru_cache(self):
+        net.get_ovs_internal_interfaces.cache_clear()
+
+    def test_command_used(self, m_subp):
+        """Test we use the correct command when we call subp"""
+        net.get_ovs_internal_interfaces()
+
+        assert [
+            mock.call(net.OVS_INTERNAL_INTERFACE_LOOKUP_CMD)
+        ] == m_subp.call_args_list
+
+    def test_subp_contents_split_and_returned(self, m_subp):
+        """Test that the command output is appropriately mangled."""
+        stdout = "iface1\niface2\niface3\n"
+        m_subp.return_value = (stdout, "")
+
+        assert [
+            "iface1",
+            "iface2",
+            "iface3",
+        ] == net.get_ovs_internal_interfaces()
+
+    def test_database_connection_error_handled_gracefully(self, m_subp):
+        """Test that the error indicating OVS is down is handled gracefully."""
+        m_subp.side_effect = ProcessExecutionError(
+            stderr="database connection failed"
+        )
+
+        assert [] == net.get_ovs_internal_interfaces()
+
+    def test_other_errors_raised(self, m_subp):
+        """Test that only database connection errors are handled."""
+        m_subp.side_effect = ProcessExecutionError()
+
+        with pytest.raises(ProcessExecutionError):
+            net.get_ovs_internal_interfaces()
+
+    def test_only_runs_once(self, m_subp):
+        """Test that we cache the value."""
+        net.get_ovs_internal_interfaces()
+        net.get_ovs_internal_interfaces()
+
+        assert 1 == m_subp.call_count
+
+
+@mock.patch("cloudinit.net.get_ovs_internal_interfaces")
+@mock.patch("cloudinit.net.openvswitch_is_installed")
+class TestIsOpenVSwitchInternalInterface:
+    def test_false_if_ovs_not_installed(
+        self, m_openvswitch_is_installed, _m_get_ovs_internal_interfaces
+    ):
+        """Test that OVS' absence returns False."""
+        m_openvswitch_is_installed.return_value = False
+
+        assert not net.is_openvswitch_internal_interface("devname")
+
+    @pytest.mark.parametrize(
+        "detected_interfaces,devname,expected_return",
+        [
+            ([], "devname", False),
+            (["notdevname"], "devname", False),
+            (["devname"], "devname", True),
+            (["some", "other", "devices", "and", "ours"], "ours", True),
+        ],
+    )
+    def test_return_value_based_on_detected_interfaces(
+        self,
+        m_openvswitch_is_installed,
+        m_get_ovs_internal_interfaces,
+        detected_interfaces,
+        devname,
+        expected_return,
+    ):
+        """Test that the detected interfaces are used correctly."""
+        m_openvswitch_is_installed.return_value = True
+        m_get_ovs_internal_interfaces.return_value = detected_interfaces
+        assert expected_return == net.is_openvswitch_internal_interface(
+            devname
+        )
+
+
+class TestIsIpAddress:
+    """Tests for net.is_ip_address.
+
+    Instead of testing with values we rely on the ipaddress stdlib module to
+    handle all values correctly, so simply test that is_ip_address defers to
+    the ipaddress module correctly.
+    """
+
+    @pytest.mark.parametrize(
+        "ip_address_side_effect,expected_return",
+        (
+            (ValueError, False),
+            (lambda _: ipaddress.IPv4Address("192.168.0.1"), True),
+            (lambda _: ipaddress.IPv6Address("2001:db8::"), True),
+        ),
+    )
+    def test_is_ip_address(self, ip_address_side_effect, expected_return):
+        with mock.patch(
+            "cloudinit.net.ipaddress.ip_address",
+            side_effect=ip_address_side_effect,
+        ) as m_ip_address:
+            ret = net.is_ip_address(mock.sentinel.ip_address_in)
+        assert expected_return == ret
+        expected_call = mock.call(mock.sentinel.ip_address_in)
+        assert [expected_call] == m_ip_address.call_args_list
+
+
+class TestIsIpv4Address:
+    """Tests for net.is_ipv4_address.
+
+    Instead of testing with values we rely on the ipaddress stdlib module to
+    handle all values correctly, so simply test that is_ipv4_address defers to
+    the ipaddress module correctly.
+    """
+
+    @pytest.mark.parametrize(
+        "ipv4address_mock,expected_return",
+        (
+            (mock.Mock(side_effect=ValueError), False),
+            (
+                mock.Mock(return_value=ipaddress.IPv4Address("192.168.0.1")),
+                True,
+            ),
+        ),
+    )
+    def test_is_ip_address(self, ipv4address_mock, expected_return):
+        with mock.patch(
+            "cloudinit.net.ipaddress.IPv4Address", ipv4address_mock
+        ) as m_ipv4address:
+            ret = net.is_ipv4_address(mock.sentinel.ip_address_in)
+        assert expected_return == ret
+        expected_call = mock.call(mock.sentinel.ip_address_in)
+        assert [expected_call] == m_ipv4address.call_args_list
+
+
+# vi: ts=4 expandtab
diff --git a/tests/unittests/net/test_network_state.py b/tests/unittests/net/test_network_state.py
new file mode 100644
index 00000000..471d969a
--- /dev/null
+++ b/tests/unittests/net/test_network_state.py
@@ -0,0 +1,222 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+import ipaddress
+from unittest import mock
+
+import pytest
+
+from cloudinit import safeyaml
+from cloudinit.net import network_state
+from tests.unittests.helpers import CiTestCase
+
+netstate_path = "cloudinit.net.network_state"
+
+
+_V1_CONFIG_NAMESERVERS = """\
+network:
+  version: 1
+  config:
+    - type: nameserver
+      interface: {iface}
+      address:
+        - 192.168.1.1
+        - 8.8.8.8
+      search:
+        - spam.local
+    - type: nameserver
+      address:
+        - 192.168.1.0
+        - 4.4.4.4
+      search:
+        - eggs.local
+    - type: physical
+      name: eth0
+      mac_address: '00:11:22:33:44:55'
+    - type: physical
+      name: eth1
+      mac_address: '66:77:88:99:00:11'
+"""
+
+V1_CONFIG_NAMESERVERS_VALID = _V1_CONFIG_NAMESERVERS.format(iface="eth1")
+V1_CONFIG_NAMESERVERS_INVALID = _V1_CONFIG_NAMESERVERS.format(iface="eth90")
+
+V2_CONFIG_NAMESERVERS = """\
+network:
+  version: 2
+  ethernets:
+    eth0:
+      match:
+        macaddress: '00:11:22:33:44:55'
+      nameservers:
+        search: [spam.local, eggs.local]
+        addresses: [8.8.8.8]
+    eth1:
+      match:
+        macaddress: '66:77:88:99:00:11'
+      set-name: "ens92"
+      nameservers:
+        search: [foo.local, bar.local]
+        addresses: [4.4.4.4]
+"""
+
+
+class TestNetworkStateParseConfig(CiTestCase):
+    def setUp(self):
+        super(TestNetworkStateParseConfig, self).setUp()
+        nsi_path = netstate_path + ".NetworkStateInterpreter"
+        self.add_patch(nsi_path, "m_nsi")
+
+    def test_missing_version_returns_none(self):
+        ncfg = {}
+        with self.assertRaises(RuntimeError):
+            network_state.parse_net_config_data(ncfg)
+
+    def test_unknown_versions_returns_none(self):
+        ncfg = {"version": 13.2}
+        with self.assertRaises(RuntimeError):
+            network_state.parse_net_config_data(ncfg)
+
+    def test_version_2_passes_self_as_config(self):
+        ncfg = {"version": 2, "otherconfig": {}, "somemore": [1, 2, 3]}
+        network_state.parse_net_config_data(ncfg)
+        self.assertEqual(
+            [mock.call(version=2, config=ncfg)], self.m_nsi.call_args_list
+        )
+
+    def test_valid_config_gets_network_state(self):
+        ncfg = {"version": 2, "otherconfig": {}, "somemore": [1, 2, 3]}
+        result = network_state.parse_net_config_data(ncfg)
+        self.assertNotEqual(None, result)
+
+    def test_empty_v1_config_gets_network_state(self):
+        ncfg = {"version": 1, "config": []}
+        result = network_state.parse_net_config_data(ncfg)
+        self.assertNotEqual(None, result)
+
+    def test_empty_v2_config_gets_network_state(self):
+        ncfg = {"version": 2}
+        result = network_state.parse_net_config_data(ncfg)
+        self.assertNotEqual(None, result)
+
+
+class TestNetworkStateParseConfigV2(CiTestCase):
+    def test_version_2_ignores_renderer_key(self):
+        ncfg = {"version": 2, "renderer": "networkd", "ethernets": {}}
+        nsi = network_state.NetworkStateInterpreter(
+            version=ncfg["version"], config=ncfg
+        )
+        nsi.parse_config(skip_broken=False)
+        self.assertEqual(ncfg, nsi.as_dict()["config"])
+
+
+class TestNetworkStateParseNameservers:
+    def _parse_network_state_from_config(self, config):
+        yaml = safeyaml.load(config)
+        return network_state.parse_net_config_data(yaml["network"])
+
+    def test_v1_nameservers_valid(self):
+        config = self._parse_network_state_from_config(
+            V1_CONFIG_NAMESERVERS_VALID
+        )
+
+        # If an interface was specified, DNS shouldn't be in the global list
+        assert ["192.168.1.0", "4.4.4.4"] == sorted(config.dns_nameservers)
+        assert ["eggs.local"] == config.dns_searchdomains
+
+        # If an interface was specified, DNS should be part of the interface
+        for iface in config.iter_interfaces():
+            if iface["name"] == "eth1":
+                assert iface["dns"]["addresses"] == ["192.168.1.1", "8.8.8.8"]
+                assert iface["dns"]["search"] == ["spam.local"]
+            else:
+                assert "dns" not in iface
+
+    def test_v1_nameservers_invalid(self):
+        with pytest.raises(ValueError):
+            self._parse_network_state_from_config(
+                V1_CONFIG_NAMESERVERS_INVALID
+            )
+
+    def test_v2_nameservers(self):
+        config = self._parse_network_state_from_config(V2_CONFIG_NAMESERVERS)
+
+        # Ensure DNS defined on interface exists on interface
+        for iface in config.iter_interfaces():
+            if iface["name"] == "eth0":
+                assert iface["dns"] == {
+                    "nameservers": ["8.8.8.8"],
+                    "search": ["spam.local", "eggs.local"],
+                }
+            else:
+                assert iface["dns"] == {
+                    "nameservers": ["4.4.4.4"],
+                    "search": ["foo.local", "bar.local"],
+                }
+
+        # Ensure DNS defined on interface also exists globally (since there
+        # is no global DNS definitions in v2)
+        assert ["4.4.4.4", "8.8.8.8"] == sorted(config.dns_nameservers)
+        assert [
+            "bar.local",
+            "eggs.local",
+            "foo.local",
+            "spam.local",
+        ] == sorted(config.dns_searchdomains)
+
+
+class TestNetworkStateHelperFunctions(CiTestCase):
+    def test_mask_to_net_prefix_ipv4(self):
+        netmask_value = "255.255.255.0"
+        expected = 24
+        prefix_value = network_state.ipv4_mask_to_net_prefix(netmask_value)
+        assert prefix_value == expected
+
+    def test_mask_to_net_prefix_all_bits_ipv4(self):
+        netmask_value = "255.255.255.255"
+        expected = 32
+        prefix_value = network_state.ipv4_mask_to_net_prefix(netmask_value)
+        assert prefix_value == expected
+
+    def test_mask_to_net_prefix_to_many_bits_ipv4(self):
+        netmask_value = "33"
+        self.assertRaises(
+            ValueError, network_state.ipv4_mask_to_net_prefix, netmask_value
+        )
+
+    def test_mask_to_net_prefix_all_bits_ipv6(self):
+        netmask_value = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
+        expected = 128
+        prefix_value = network_state.ipv6_mask_to_net_prefix(netmask_value)
+        assert prefix_value == expected
+
+    def test_mask_to_net_prefix_ipv6(self):
+        netmask_value = "ffff:ffff:ffff:ffff::"
+        expected = 64
+        prefix_value = network_state.ipv6_mask_to_net_prefix(netmask_value)
+        assert prefix_value == expected
+
+    def test_mask_to_net_prefix_raises_value_error(self):
+        netmask_value = "ff:ff:ff:ff::"
+        self.assertRaises(
+            ValueError, network_state.ipv6_mask_to_net_prefix, netmask_value
+        )
+
+    def test_mask_to_net_prefix_to_many_bits_ipv6(self):
+        netmask_value = "129"
+        self.assertRaises(
+            ValueError, network_state.ipv6_mask_to_net_prefix, netmask_value
+        )
+
+    def test_mask_to_net_prefix_ipv4_object(self):
+        netmask_value = ipaddress.IPv4Address("255.255.255.255")
+        expected = 32
+        prefix_value = network_state.ipv4_mask_to_net_prefix(netmask_value)
+        assert prefix_value == expected
+
+    def test_mask_to_net_prefix_ipv6_object(self):
+        netmask_value = ipaddress.IPv6Address("ffff:ffff:ffff::")
+        expected = 48
+        prefix_value = network_state.ipv6_mask_to_net_prefix(netmask_value)
+        assert prefix_value == expected
+
+
+# vi: ts=4 expandtab
diff --git a/tests/unittests/net/test_networkd.py b/tests/unittests/net/test_networkd.py
new file mode 100644
index 00000000..ec1d04e9
--- /dev/null
+++ b/tests/unittests/net/test_networkd.py
@@ -0,0 +1,64 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+from cloudinit import safeyaml
+from cloudinit.net import network_state, networkd
+
+V2_CONFIG_SET_NAME = """\
+network:
+  version: 2
+  ethernets:
+    eth0:
+      match:
+        macaddress: '00:11:22:33:44:55'
+      nameservers:
+        search: [spam.local, eggs.local]
+        addresses: [8.8.8.8]
+    eth1:
+      match:
+        macaddress: '66:77:88:99:00:11'
+      set-name: "ens92"
+      nameservers:
+        search: [foo.local, bar.local]
+        addresses: [4.4.4.4]
+"""
+
+V2_CONFIG_SET_NAME_RENDERED_ETH0 = """[Match]
+MACAddress=00:11:22:33:44:55
+Name=eth0
+
+[Network]
+DHCP=no
+DNS=8.8.8.8
+Domains=spam.local eggs.local
+
+"""
+
+V2_CONFIG_SET_NAME_RENDERED_ETH1 = """[Match]
+MACAddress=66:77:88:99:00:11
+Name=ens92
+
+[Network]
+DHCP=no
+DNS=4.4.4.4
+Domains=foo.local bar.local
+
+"""
+
+
+class TestNetworkdRenderState:
+    def _parse_network_state_from_config(self, config):
+        yaml = safeyaml.load(config)
+        return network_state.parse_net_config_data(yaml["network"])
+
+    def test_networkd_render_with_set_name(self):
+        ns = self._parse_network_state_from_config(V2_CONFIG_SET_NAME)
+        renderer = networkd.Renderer()
+        rendered_content = renderer._render_content(ns)
+
+        assert "eth0" in rendered_content
+        assert rendered_content["eth0"] == V2_CONFIG_SET_NAME_RENDERED_ETH0
+        assert "ens92" in rendered_content
+        assert rendered_content["ens92"] == V2_CONFIG_SET_NAME_RENDERED_ETH1
+
+
+# vi: ts=4 expandtab