diff options
Diffstat (limited to 'tests/unittests/sources/test_opennebula.py')
| -rw-r--r-- | tests/unittests/sources/test_opennebula.py | 1085 |
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 |