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.

1 files changed, 1085 insertions, 0 deletions

diff --git a/tests/unittests/sources/test_opennebula.py b/tests/unittests/sources/test_opennebula.py
new file mode 100644
index 00000000..e05c4749
--- /dev/null
+++ b/tests/unittests/sources/test_opennebula.py
@@ -0,0 +1,1085 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+import os
+import pwd
+import unittest
+
+import pytest
+
+from cloudinit import helpers, util
+from cloudinit.sources import DataSourceOpenNebula as ds
+from tests.unittests.helpers import CiTestCase, mock, populate_dir
+
+TEST_VARS = {
+    "VAR1": "single",
+    "VAR2": "double word",
+    "VAR3": "multi\nline\n",
+    "VAR4": "'single'",
+    "VAR5": "'double word'",
+    "VAR6": "'multi\nline\n'",
+    "VAR7": "single\\t",
+    "VAR8": "double\\tword",
+    "VAR9": "multi\\t\nline\n",
+    "VAR10": "\\",  # expect '\'
+    "VAR11": "'",  # expect '
+    "VAR12": "$",  # expect $
+}
+
+INVALID_CONTEXT = ";"
+USER_DATA = "#cloud-config\napt_upgrade: true"
+SSH_KEY = "ssh-rsa AAAAB3NzaC1....sIkJhq8wdX+4I3A4cYbYP ubuntu@server-460-%i"
+HOSTNAME = "foo.example.com"
+PUBLIC_IP = "10.0.0.3"
+MACADDR = "02:00:0a:12:01:01"
+IP_BY_MACADDR = "10.18.1.1"
+IP4_PREFIX = "24"
+IP6_GLOBAL = "2001:db8:1:0:400:c0ff:fea8:1ba"
+IP6_ULA = "fd01:dead:beaf:0:400:c0ff:fea8:1ba"
+IP6_GW = "2001:db8:1::ffff"
+IP6_PREFIX = "48"
+
+DS_PATH = "cloudinit.sources.DataSourceOpenNebula"
+
+
+class TestOpenNebulaDataSource(CiTestCase):
+    parsed_user = None
+    allowed_subp = ["bash"]
+
+    def setUp(self):
+        super(TestOpenNebulaDataSource, self).setUp()
+        self.tmp = self.tmp_dir()
+        self.paths = helpers.Paths(
+            {"cloud_dir": self.tmp, "run_dir": self.tmp}
+        )
+
+        # defaults for few tests
+        self.ds = ds.DataSourceOpenNebula
+        self.seed_dir = os.path.join(self.paths.seed_dir, "opennebula")
+        self.sys_cfg = {"datasource": {"OpenNebula": {"dsmode": "local"}}}
+
+        # we don't want 'sudo' called in tests. so we patch switch_user_cmd
+        def my_switch_user_cmd(user):
+            self.parsed_user = user
+            return []
+
+        self.switch_user_cmd_real = ds.switch_user_cmd
+        ds.switch_user_cmd = my_switch_user_cmd
+
+    def tearDown(self):
+        ds.switch_user_cmd = self.switch_user_cmd_real
+        super(TestOpenNebulaDataSource, self).tearDown()
+
+    def test_get_data_non_contextdisk(self):
+        orig_find_devs_with = util.find_devs_with
+        try:
+            # dont' try to lookup for CDs
+            util.find_devs_with = lambda n: []
+            dsrc = self.ds(sys_cfg=self.sys_cfg, distro=None, paths=self.paths)
+            ret = dsrc.get_data()
+            self.assertFalse(ret)
+        finally:
+            util.find_devs_with = orig_find_devs_with
+
+    def test_get_data_broken_contextdisk(self):
+        orig_find_devs_with = util.find_devs_with
+        try:
+            # dont' try to lookup for CDs
+            util.find_devs_with = lambda n: []
+            populate_dir(self.seed_dir, {"context.sh": INVALID_CONTEXT})
+            dsrc = self.ds(sys_cfg=self.sys_cfg, distro=None, paths=self.paths)
+            self.assertRaises(ds.BrokenContextDiskDir, dsrc.get_data)
+        finally:
+            util.find_devs_with = orig_find_devs_with
+
+    def test_get_data_invalid_identity(self):
+        orig_find_devs_with = util.find_devs_with
+        try:
+            # generate non-existing system user name
+            sys_cfg = self.sys_cfg
+            invalid_user = "invalid"
+            while not sys_cfg["datasource"]["OpenNebula"].get("parseuser"):
+                try:
+                    pwd.getpwnam(invalid_user)
+                    invalid_user += "X"
+                except KeyError:
+                    sys_cfg["datasource"]["OpenNebula"][
+                        "parseuser"
+                    ] = invalid_user
+
+            # dont' try to lookup for CDs
+            util.find_devs_with = lambda n: []
+            populate_context_dir(self.seed_dir, {"KEY1": "val1"})
+            dsrc = self.ds(sys_cfg=sys_cfg, distro=None, paths=self.paths)
+            self.assertRaises(ds.BrokenContextDiskDir, dsrc.get_data)
+        finally:
+            util.find_devs_with = orig_find_devs_with
+
+    def test_get_data(self):
+        orig_find_devs_with = util.find_devs_with
+        try:
+            # dont' try to lookup for CDs
+            util.find_devs_with = lambda n: []
+            populate_context_dir(self.seed_dir, {"KEY1": "val1"})
+            dsrc = self.ds(sys_cfg=self.sys_cfg, distro=None, paths=self.paths)
+            ret = dsrc.get_data()
+            self.assertTrue(ret)
+        finally:
+            util.find_devs_with = orig_find_devs_with
+        self.assertEqual("opennebula", dsrc.cloud_name)
+        self.assertEqual("opennebula", dsrc.platform_type)
+        self.assertEqual(
+            "seed-dir (%s/seed/opennebula)" % self.tmp, dsrc.subplatform
+        )
+
+    def test_seed_dir_non_contextdisk(self):
+        self.assertRaises(
+            ds.NonContextDiskDir,
+            ds.read_context_disk_dir,
+            self.seed_dir,
+            mock.Mock(),
+        )
+
+    def test_seed_dir_empty1_context(self):
+        populate_dir(self.seed_dir, {"context.sh": ""})
+        results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+        self.assertIsNone(results["userdata"])
+        self.assertEqual(results["metadata"], {})
+
+    def test_seed_dir_empty2_context(self):
+        populate_context_dir(self.seed_dir, {})
+        results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+        self.assertIsNone(results["userdata"])
+        self.assertEqual(results["metadata"], {})
+
+    def test_seed_dir_broken_context(self):
+        populate_dir(self.seed_dir, {"context.sh": INVALID_CONTEXT})
+
+        self.assertRaises(
+            ds.BrokenContextDiskDir,
+            ds.read_context_disk_dir,
+            self.seed_dir,
+            mock.Mock(),
+        )
+
+    def test_context_parser(self):
+        populate_context_dir(self.seed_dir, TEST_VARS)
+        results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+        self.assertTrue("metadata" in results)
+        self.assertEqual(TEST_VARS, results["metadata"])
+
+    def test_ssh_key(self):
+        public_keys = ["first key", "second key"]
+        for c in range(4):
+            for k in ("SSH_KEY", "SSH_PUBLIC_KEY"):
+                my_d = os.path.join(self.tmp, "%s-%i" % (k, c))
+                populate_context_dir(my_d, {k: "\n".join(public_keys)})
+                results = ds.read_context_disk_dir(my_d, mock.Mock())
+
+                self.assertTrue("metadata" in results)
+                self.assertTrue("public-keys" in results["metadata"])
+                self.assertEqual(
+                    public_keys, results["metadata"]["public-keys"]
+                )
+
+            public_keys.append(SSH_KEY % (c + 1,))
+
+    def test_user_data_plain(self):
+        for k in ("USER_DATA", "USERDATA"):
+            my_d = os.path.join(self.tmp, k)
+            populate_context_dir(my_d, {k: USER_DATA, "USERDATA_ENCODING": ""})
+            results = ds.read_context_disk_dir(my_d, mock.Mock())
+
+            self.assertTrue("userdata" in results)
+            self.assertEqual(USER_DATA, results["userdata"])
+
+    def test_user_data_encoding_required_for_decode(self):
+        b64userdata = util.b64e(USER_DATA)
+        for k in ("USER_DATA", "USERDATA"):
+            my_d = os.path.join(self.tmp, k)
+            populate_context_dir(my_d, {k: b64userdata})
+            results = ds.read_context_disk_dir(my_d, mock.Mock())
+
+            self.assertTrue("userdata" in results)
+            self.assertEqual(b64userdata, results["userdata"])
+
+    def test_user_data_base64_encoding(self):
+        for k in ("USER_DATA", "USERDATA"):
+            my_d = os.path.join(self.tmp, k)
+            populate_context_dir(
+                my_d, {k: util.b64e(USER_DATA), "USERDATA_ENCODING": "base64"}
+            )
+            results = ds.read_context_disk_dir(my_d, mock.Mock())
+
+            self.assertTrue("userdata" in results)
+            self.assertEqual(USER_DATA, results["userdata"])
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_hostname(self, m_get_phys_by_mac):
+        for dev in ("eth0", "ens3"):
+            m_get_phys_by_mac.return_value = {MACADDR: dev}
+            for k in (
+                "SET_HOSTNAME",
+                "HOSTNAME",
+                "PUBLIC_IP",
+                "IP_PUBLIC",
+                "ETH0_IP",
+            ):
+                my_d = os.path.join(self.tmp, k)
+                populate_context_dir(my_d, {k: PUBLIC_IP})
+                results = ds.read_context_disk_dir(my_d, mock.Mock())
+
+                self.assertTrue("metadata" in results)
+                self.assertTrue("local-hostname" in results["metadata"])
+                self.assertEqual(
+                    PUBLIC_IP, results["metadata"]["local-hostname"]
+                )
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_network_interfaces(self, m_get_phys_by_mac):
+        for dev in ("eth0", "ens3"):
+            m_get_phys_by_mac.return_value = {MACADDR: dev}
+
+            # without ETH0_MAC
+            # for Older OpenNebula?
+            populate_context_dir(self.seed_dir, {"ETH0_IP": IP_BY_MACADDR})
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP_BY_MACADDR + "/" + IP4_PREFIX
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+            # ETH0_IP and ETH0_MAC
+            populate_context_dir(
+                self.seed_dir, {"ETH0_IP": IP_BY_MACADDR, "ETH0_MAC": MACADDR}
+            )
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP_BY_MACADDR + "/" + IP4_PREFIX
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+            # ETH0_IP with empty string and ETH0_MAC
+            # in the case of using Virtual Network contains
+            # "AR = [ TYPE = ETHER ]"
+            populate_context_dir(
+                self.seed_dir, {"ETH0_IP": "", "ETH0_MAC": MACADDR}
+            )
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP_BY_MACADDR + "/" + IP4_PREFIX
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+            # ETH0_MASK
+            populate_context_dir(
+                self.seed_dir,
+                {
+                    "ETH0_IP": IP_BY_MACADDR,
+                    "ETH0_MAC": MACADDR,
+                    "ETH0_MASK": "255.255.0.0",
+                },
+            )
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP_BY_MACADDR + "/16"
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+            # ETH0_MASK with empty string
+            populate_context_dir(
+                self.seed_dir,
+                {
+                    "ETH0_IP": IP_BY_MACADDR,
+                    "ETH0_MAC": MACADDR,
+                    "ETH0_MASK": "",
+                },
+            )
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP_BY_MACADDR + "/" + IP4_PREFIX
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+            # ETH0_IP6
+            populate_context_dir(
+                self.seed_dir,
+                {
+                    "ETH0_IP6": IP6_GLOBAL,
+                    "ETH0_MAC": MACADDR,
+                },
+            )
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP6_GLOBAL + "/64"
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+            # ETH0_IP6_ULA
+            populate_context_dir(
+                self.seed_dir,
+                {
+                    "ETH0_IP6_ULA": IP6_ULA,
+                    "ETH0_MAC": MACADDR,
+                },
+            )
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP6_ULA + "/64"
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+            # ETH0_IP6 and ETH0_IP6_PREFIX_LENGTH
+            populate_context_dir(
+                self.seed_dir,
+                {
+                    "ETH0_IP6": IP6_GLOBAL,
+                    "ETH0_IP6_PREFIX_LENGTH": IP6_PREFIX,
+                    "ETH0_MAC": MACADDR,
+                },
+            )
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP6_GLOBAL + "/" + IP6_PREFIX
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+            # ETH0_IP6 and ETH0_IP6_PREFIX_LENGTH with empty string
+            populate_context_dir(
+                self.seed_dir,
+                {
+                    "ETH0_IP6": IP6_GLOBAL,
+                    "ETH0_IP6_PREFIX_LENGTH": "",
+                    "ETH0_MAC": MACADDR,
+                },
+            )
+            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())
+
+            self.assertTrue("network-interfaces" in results)
+            self.assertTrue(
+                IP6_GLOBAL + "/64"
+                in results["network-interfaces"]["ethernets"][dev]["addresses"]
+            )
+
+    def test_find_candidates(self):
+        def my_devs_with(criteria):
+            return {
+                "LABEL=CONTEXT": ["/dev/sdb"],
+                "LABEL=CDROM": ["/dev/sr0"],
+                "TYPE=iso9660": ["/dev/vdb"],
+            }.get(criteria, [])
+
+        orig_find_devs_with = util.find_devs_with
+        try:
+            util.find_devs_with = my_devs_with
+            self.assertEqual(
+                ["/dev/sdb", "/dev/sr0", "/dev/vdb"], ds.find_candidate_devs()
+            )
+        finally:
+            util.find_devs_with = orig_find_devs_with
+
+
+@mock.patch(DS_PATH + ".net.get_interfaces_by_mac", mock.Mock(return_value={}))
+class TestOpenNebulaNetwork(unittest.TestCase):
+
+    system_nics = ("eth0", "ens3")
+
+    def test_context_devname(self):
+        """Verify context_devname correctly returns mac and name."""
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH1_MAC": "02:00:0a:12:0f:0f",
+        }
+        expected = {
+            "02:00:0a:12:01:01": "ETH0",
+            "02:00:0a:12:0f:0f": "ETH1",
+        }
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        self.assertEqual(expected, net.context_devname)
+
+    def test_get_nameservers(self):
+        """
+        Verify get_nameservers('device') correctly returns DNS server addresses
+        and search domains.
+        """
+        context = {
+            "DNS": "1.2.3.8",
+            "ETH0_DNS": "1.2.3.6 1.2.3.7",
+            "ETH0_SEARCH_DOMAIN": "example.com example.org",
+        }
+        expected = {
+            "addresses": ["1.2.3.6", "1.2.3.7", "1.2.3.8"],
+            "search": ["example.com", "example.org"],
+        }
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_nameservers("eth0")
+        self.assertEqual(expected, val)
+
+    def test_get_mtu(self):
+        """Verify get_mtu('device') correctly returns MTU size."""
+        context = {"ETH0_MTU": "1280"}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_mtu("eth0")
+        self.assertEqual("1280", val)
+
+    def test_get_ip(self):
+        """Verify get_ip('device') correctly returns IPv4 address."""
+        context = {"ETH0_IP": PUBLIC_IP}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_ip("eth0", MACADDR)
+        self.assertEqual(PUBLIC_IP, val)
+
+    def test_get_ip_emptystring(self):
+        """
+        Verify get_ip('device') correctly returns IPv4 address.
+        It returns IP address created by MAC address if ETH0_IP has empty
+        string.
+        """
+        context = {"ETH0_IP": ""}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_ip("eth0", MACADDR)
+        self.assertEqual(IP_BY_MACADDR, val)
+
+    def test_get_ip6(self):
+        """
+        Verify get_ip6('device') correctly returns IPv6 address.
+        In this case, IPv6 address is Given by ETH0_IP6.
+        """
+        context = {
+            "ETH0_IP6": IP6_GLOBAL,
+            "ETH0_IP6_ULA": "",
+        }
+        expected = [IP6_GLOBAL]
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_ip6("eth0")
+        self.assertEqual(expected, val)
+
+    def test_get_ip6_ula(self):
+        """
+        Verify get_ip6('device') correctly returns IPv6 address.
+        In this case, IPv6 address is Given by ETH0_IP6_ULA.
+        """
+        context = {
+            "ETH0_IP6": "",
+            "ETH0_IP6_ULA": IP6_ULA,
+        }
+        expected = [IP6_ULA]
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_ip6("eth0")
+        self.assertEqual(expected, val)
+
+    def test_get_ip6_dual(self):
+        """
+        Verify get_ip6('device') correctly returns IPv6 address.
+        In this case, IPv6 addresses are Given by ETH0_IP6 and ETH0_IP6_ULA.
+        """
+        context = {
+            "ETH0_IP6": IP6_GLOBAL,
+            "ETH0_IP6_ULA": IP6_ULA,
+        }
+        expected = [IP6_GLOBAL, IP6_ULA]
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_ip6("eth0")
+        self.assertEqual(expected, val)
+
+    def test_get_ip6_prefix(self):
+        """
+        Verify get_ip6_prefix('device') correctly returns IPv6 prefix.
+        """
+        context = {"ETH0_IP6_PREFIX_LENGTH": IP6_PREFIX}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_ip6_prefix("eth0")
+        self.assertEqual(IP6_PREFIX, val)
+
+    def test_get_ip6_prefix_emptystring(self):
+        """
+        Verify get_ip6_prefix('device') correctly returns IPv6 prefix.
+        It returns default value '64' if ETH0_IP6_PREFIX_LENGTH has empty
+        string.
+        """
+        context = {"ETH0_IP6_PREFIX_LENGTH": ""}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_ip6_prefix("eth0")
+        self.assertEqual("64", val)
+
+    def test_get_gateway(self):
+        """
+        Verify get_gateway('device') correctly returns IPv4 default gateway
+        address.
+        """
+        context = {"ETH0_GATEWAY": "1.2.3.5"}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_gateway("eth0")
+        self.assertEqual("1.2.3.5", val)
+
+    def test_get_gateway6(self):
+        """
+        Verify get_gateway6('device') correctly returns IPv6 default gateway
+        address.
+        """
+        for k in ("GATEWAY6", "IP6_GATEWAY"):
+            context = {"ETH0_" + k: IP6_GW}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            val = net.get_gateway6("eth0")
+            self.assertEqual(IP6_GW, val)
+
+    def test_get_mask(self):
+        """
+        Verify get_mask('device') correctly returns IPv4 subnet mask.
+        """
+        context = {"ETH0_MASK": "255.255.0.0"}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_mask("eth0")
+        self.assertEqual("255.255.0.0", val)
+
+    def test_get_mask_emptystring(self):
+        """
+        Verify get_mask('device') correctly returns IPv4 subnet mask.
+        It returns default value '255.255.255.0' if ETH0_MASK has empty string.
+        """
+        context = {"ETH0_MASK": ""}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_mask("eth0")
+        self.assertEqual("255.255.255.0", val)
+
+    def test_get_network(self):
+        """
+        Verify get_network('device') correctly returns IPv4 network address.
+        """
+        context = {"ETH0_NETWORK": "1.2.3.0"}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_network("eth0", MACADDR)
+        self.assertEqual("1.2.3.0", val)
+
+    def test_get_network_emptystring(self):
+        """
+        Verify get_network('device') correctly returns IPv4 network address.
+        It returns network address created by MAC address if ETH0_NETWORK has
+        empty string.
+        """
+        context = {"ETH0_NETWORK": ""}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_network("eth0", MACADDR)
+        self.assertEqual("10.18.1.0", val)
+
+    def test_get_field(self):
+        """
+        Verify get_field('device', 'name') returns *context* value.
+        """
+        context = {"ETH9_DUMMY": "DUMMY_VALUE"}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_field("eth9", "dummy")
+        self.assertEqual("DUMMY_VALUE", val)
+
+    def test_get_field_withdefaultvalue(self):
+        """
+        Verify get_field('device', 'name', 'default value') returns *context*
+        value.
+        """
+        context = {"ETH9_DUMMY": "DUMMY_VALUE"}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_field("eth9", "dummy", "DEFAULT_VALUE")
+        self.assertEqual("DUMMY_VALUE", val)
+
+    def test_get_field_withdefaultvalue_emptycontext(self):
+        """
+        Verify get_field('device', 'name', 'default value') returns *default*
+        value if context value is empty string.
+        """
+        context = {"ETH9_DUMMY": ""}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_field("eth9", "dummy", "DEFAULT_VALUE")
+        self.assertEqual("DEFAULT_VALUE", val)
+
+    def test_get_field_emptycontext(self):
+        """
+        Verify get_field('device', 'name') returns None if context value is
+        empty string.
+        """
+        context = {"ETH9_DUMMY": ""}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_field("eth9", "dummy")
+        self.assertEqual(None, val)
+
+    def test_get_field_nonecontext(self):
+        """
+        Verify get_field('device', 'name') returns None if context value is
+        None.
+        """
+        context = {"ETH9_DUMMY": None}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        val = net.get_field("eth9", "dummy")
+        self.assertEqual(None, val)
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_gateway(self, m_get_phys_by_mac):
+        """Test rendering with/without IPv4 gateway"""
+        self.maxDiff = None
+        # empty ETH0_GATEWAY
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_GATEWAY": "",
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+        # set ETH0_GATEWAY
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_GATEWAY": "1.2.3.5",
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "gateway4": "1.2.3.5",
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_gateway6(self, m_get_phys_by_mac):
+        """Test rendering with/without IPv6 gateway"""
+        self.maxDiff = None
+        # empty ETH0_GATEWAY6
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_GATEWAY6": "",
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+        # set ETH0_GATEWAY6
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_GATEWAY6": IP6_GW,
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "gateway6": IP6_GW,
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_ipv6address(self, m_get_phys_by_mac):
+        """Test rendering with/without IPv6 address"""
+        self.maxDiff = None
+        # empty ETH0_IP6, ETH0_IP6_ULA, ETH0_IP6_PREFIX_LENGTH
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_IP6": "",
+            "ETH0_IP6_ULA": "",
+            "ETH0_IP6_PREFIX_LENGTH": "",
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+        # set ETH0_IP6, ETH0_IP6_ULA, ETH0_IP6_PREFIX_LENGTH
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_IP6": IP6_GLOBAL,
+            "ETH0_IP6_PREFIX_LENGTH": IP6_PREFIX,
+            "ETH0_IP6_ULA": IP6_ULA,
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [
+                            IP_BY_MACADDR + "/" + IP4_PREFIX,
+                            IP6_GLOBAL + "/" + IP6_PREFIX,
+                            IP6_ULA + "/" + IP6_PREFIX,
+                        ],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_dns(self, m_get_phys_by_mac):
+        """Test rendering with/without DNS server, search domain"""
+        self.maxDiff = None
+        # empty DNS, ETH0_DNS, ETH0_SEARCH_DOMAIN
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "DNS": "",
+            "ETH0_DNS": "",
+            "ETH0_SEARCH_DOMAIN": "",
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+        # set DNS, ETH0_DNS, ETH0_SEARCH_DOMAIN
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "DNS": "1.2.3.8",
+            "ETH0_DNS": "1.2.3.6 1.2.3.7",
+            "ETH0_SEARCH_DOMAIN": "example.com example.org",
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "nameservers": {
+                            "addresses": ["1.2.3.6", "1.2.3.7", "1.2.3.8"],
+                            "search": ["example.com", "example.org"],
+                        },
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_mtu(self, m_get_phys_by_mac):
+        """Test rendering with/without MTU"""
+        self.maxDiff = None
+        # empty ETH0_MTU
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_MTU": "",
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+        # set ETH0_MTU
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_MTU": "1280",
+        }
+        for nic in self.system_nics:
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "mtu": "1280",
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            self.assertEqual(net.gen_conf(), expected)
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_eth0(self, m_get_phys_by_mac):
+        for nic in self.system_nics:
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork({}, mock.Mock())
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX],
+                    }
+                },
+            }
+
+            self.assertEqual(net.gen_conf(), expected)
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_distro_passed_through(self, m_get_physical_nics_by_mac):
+        ds.OpenNebulaNetwork({}, mock.sentinel.distro)
+        self.assertEqual(
+            [mock.call(mock.sentinel.distro)],
+            m_get_physical_nics_by_mac.call_args_list,
+        )
+
+    def test_eth0_override(self):
+        self.maxDiff = None
+        context = {
+            "DNS": "1.2.3.8",
+            "ETH0_DNS": "1.2.3.6 1.2.3.7",
+            "ETH0_GATEWAY": "1.2.3.5",
+            "ETH0_GATEWAY6": "",
+            "ETH0_IP": IP_BY_MACADDR,
+            "ETH0_IP6": "",
+            "ETH0_IP6_PREFIX_LENGTH": "",
+            "ETH0_IP6_ULA": "",
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_MASK": "255.255.0.0",
+            "ETH0_MTU": "",
+            "ETH0_NETWORK": "10.18.0.0",
+            "ETH0_SEARCH_DOMAIN": "",
+        }
+        for nic in self.system_nics:
+            net = ds.OpenNebulaNetwork(
+                context, mock.Mock(), system_nics_by_mac={MACADDR: nic}
+            )
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [IP_BY_MACADDR + "/16"],
+                        "gateway4": "1.2.3.5",
+                        "nameservers": {
+                            "addresses": ["1.2.3.6", "1.2.3.7", "1.2.3.8"]
+                        },
+                    }
+                },
+            }
+
+            self.assertEqual(expected, net.gen_conf())
+
+    def test_eth0_v4v6_override(self):
+        self.maxDiff = None
+        context = {
+            "DNS": "1.2.3.8",
+            "ETH0_DNS": "1.2.3.6 1.2.3.7",
+            "ETH0_GATEWAY": "1.2.3.5",
+            "ETH0_GATEWAY6": IP6_GW,
+            "ETH0_IP": IP_BY_MACADDR,
+            "ETH0_IP6": IP6_GLOBAL,
+            "ETH0_IP6_PREFIX_LENGTH": IP6_PREFIX,
+            "ETH0_IP6_ULA": IP6_ULA,
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_MASK": "255.255.0.0",
+            "ETH0_MTU": "1280",
+            "ETH0_NETWORK": "10.18.0.0",
+            "ETH0_SEARCH_DOMAIN": "example.com example.org",
+        }
+        for nic in self.system_nics:
+            net = ds.OpenNebulaNetwork(
+                context, mock.Mock(), system_nics_by_mac={MACADDR: nic}
+            )
+
+            expected = {
+                "version": 2,
+                "ethernets": {
+                    nic: {
+                        "match": {"macaddress": MACADDR},
+                        "addresses": [
+                            IP_BY_MACADDR + "/16",
+                            IP6_GLOBAL + "/" + IP6_PREFIX,
+                            IP6_ULA + "/" + IP6_PREFIX,
+                        ],
+                        "gateway4": "1.2.3.5",
+                        "gateway6": IP6_GW,
+                        "nameservers": {
+                            "addresses": ["1.2.3.6", "1.2.3.7", "1.2.3.8"],
+                            "search": ["example.com", "example.org"],
+                        },
+                        "mtu": "1280",
+                    }
+                },
+            }
+
+            self.assertEqual(expected, net.gen_conf())
+
+    def test_multiple_nics(self):
+        """Test rendering multiple nics with names that differ from context."""
+        self.maxDiff = None
+        MAC_1 = "02:00:0a:12:01:01"
+        MAC_2 = "02:00:0a:12:01:02"
+        context = {
+            "DNS": "1.2.3.8",
+            "ETH0_DNS": "1.2.3.6 1.2.3.7",
+            "ETH0_GATEWAY": "1.2.3.5",
+            "ETH0_GATEWAY6": IP6_GW,
+            "ETH0_IP": "10.18.1.1",
+            "ETH0_IP6": IP6_GLOBAL,
+            "ETH0_IP6_PREFIX_LENGTH": "",
+            "ETH0_IP6_ULA": IP6_ULA,
+            "ETH0_MAC": MAC_2,
+            "ETH0_MASK": "255.255.0.0",
+            "ETH0_MTU": "1280",
+            "ETH0_NETWORK": "10.18.0.0",
+            "ETH0_SEARCH_DOMAIN": "example.com",
+            "ETH3_DNS": "10.3.1.2",
+            "ETH3_GATEWAY": "10.3.0.1",
+            "ETH3_GATEWAY6": "",
+            "ETH3_IP": "10.3.1.3",
+            "ETH3_IP6": "",
+            "ETH3_IP6_PREFIX_LENGTH": "",
+            "ETH3_IP6_ULA": "",
+            "ETH3_MAC": MAC_1,
+            "ETH3_MASK": "255.255.0.0",
+            "ETH3_MTU": "",
+            "ETH3_NETWORK": "10.3.0.0",
+            "ETH3_SEARCH_DOMAIN": "third.example.com third.example.org",
+        }
+        net = ds.OpenNebulaNetwork(
+            context,
+            mock.Mock(),
+            system_nics_by_mac={MAC_1: "enp0s25", MAC_2: "enp1s2"},
+        )
+
+        expected = {
+            "version": 2,
+            "ethernets": {
+                "enp1s2": {
+                    "match": {"macaddress": MAC_2},
+                    "addresses": [
+                        "10.18.1.1/16",
+                        IP6_GLOBAL + "/64",
+                        IP6_ULA + "/64",
+                    ],
+                    "gateway4": "1.2.3.5",
+                    "gateway6": IP6_GW,
+                    "nameservers": {
+                        "addresses": ["1.2.3.6", "1.2.3.7", "1.2.3.8"],
+                        "search": ["example.com"],
+                    },
+                    "mtu": "1280",
+                },
+                "enp0s25": {
+                    "match": {"macaddress": MAC_1},
+                    "addresses": ["10.3.1.3/16"],
+                    "gateway4": "10.3.0.1",
+                    "nameservers": {
+                        "addresses": ["10.3.1.2", "1.2.3.8"],
+                        "search": ["third.example.com", "third.example.org"],
+                    },
+                },
+            },
+        }
+
+        self.assertEqual(expected, net.gen_conf())
+
+
+class TestParseShellConfig:
+    @pytest.mark.allow_subp_for("bash")
+    def test_no_seconds(self):
+        cfg = "\n".join(["foo=bar", "SECONDS=2", "xx=foo"])
+        # we could test 'sleep 2', but that would make the test run slower.
+        ret = ds.parse_shell_config(cfg)
+        assert ret == {"foo": "bar", "xx": "foo"}
+
+
+class TestGetPhysicalNicsByMac:
+    @pytest.mark.parametrize(
+        "interfaces_by_mac,physical_devs,expected_return",
+        [
+            # No interfaces => empty return
+            ({}, [], {}),
+            # Only virtual interface => empty return
+            ({"mac1": "virtual0"}, [], {}),
+            # Only physical interface => it is returned
+            ({"mac2": "physical0"}, ["physical0"], {"mac2": "physical0"}),
+            # Combination of physical and virtual => only physical returned
+            (
+                {"mac3": "physical1", "mac4": "virtual1"},
+                ["physical1"],
+                {"mac3": "physical1"},
+            ),
+        ],
+    )
+    def test(self, interfaces_by_mac, physical_devs, expected_return):
+        distro = mock.Mock()
+        distro.networking.is_physical.side_effect = (
+            lambda devname: devname in physical_devs
+        )
+        with mock.patch(
+            DS_PATH + ".net.get_interfaces_by_mac",
+            return_value=interfaces_by_mac,
+        ):
+            assert expected_return == ds.get_physical_nics_by_mac(distro)
+
+
+def populate_context_dir(path, variables):
+    data = "# Context variables generated by OpenNebula\n"
+    for k, v in variables.items():
+        data += "%s='%s'\n" % (k.upper(), v.replace(r"'", r"'\''"))
+    populate_dir(path, {"context.sh": data})
+
+
+# vi: ts=4 expandtab