# 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