diff options
Diffstat (limited to 'tests/unittests/sources/test_opennebula.py')
| -rw-r--r-- | tests/unittests/sources/test_opennebula.py | 888 |
1 files changed, 498 insertions, 390 deletions
diff --git a/tests/unittests/sources/test_opennebula.py b/tests/unittests/sources/test_opennebula.py index e5963f5a..e05c4749 100644 --- a/tests/unittests/sources/test_opennebula.py +++ b/tests/unittests/sources/test_opennebula.py @@ -1,62 +1,61 @@ # This file is part of cloud-init. See LICENSE file for license information. -from cloudinit import helpers -from cloudinit.sources import DataSourceOpenNebula as ds -from cloudinit import util -from tests.unittests.helpers import mock, populate_dir, CiTestCase - 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 $ + "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' +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'] + 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}) + {"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'}}} + 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): @@ -86,7 +85,7 @@ class TestOpenNebulaDataSource(CiTestCase): try: # dont' try to lookup for CDs util.find_devs_with = lambda n: [] - populate_dir(self.seed_dir, {'context.sh': INVALID_CONTEXT}) + 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: @@ -97,18 +96,19 @@ class TestOpenNebulaDataSource(CiTestCase): try: # generate non-existing system user name sys_cfg = self.sys_cfg - invalid_user = 'invalid' - while not sys_cfg['datasource']['OpenNebula'].get('parseuser'): + invalid_user = "invalid" + while not sys_cfg["datasource"]["OpenNebula"].get("parseuser"): try: pwd.getpwnam(invalid_user) - invalid_user += 'X' + invalid_user += "X" except KeyError: - sys_cfg['datasource']['OpenNebula']['parseuser'] = \ - invalid_user + 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'}) + 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: @@ -119,227 +119,265 @@ class TestOpenNebulaDataSource(CiTestCase): try: # dont' try to lookup for CDs util.find_devs_with = lambda n: [] - populate_context_dir(self.seed_dir, {'KEY1': 'val1'}) + 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("opennebula", dsrc.cloud_name) + self.assertEqual("opennebula", dsrc.platform_type) self.assertEqual( - 'seed-dir (%s/seed/opennebula)' % self.tmp, dsrc.subplatform) + "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()) + 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': ''}) + 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'], {}) + 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'], {}) + self.assertIsNone(results["userdata"]) + self.assertEqual(results["metadata"], {}) def test_seed_dir_broken_context(self): - populate_dir(self.seed_dir, {'context.sh': INVALID_CONTEXT}) + populate_dir(self.seed_dir, {"context.sh": INVALID_CONTEXT}) - self.assertRaises(ds.BrokenContextDiskDir, - ds.read_context_disk_dir, - self.seed_dir, mock.Mock()) + 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']) + self.assertTrue("metadata" in results) + self.assertEqual(TEST_VARS, results["metadata"]) def test_ssh_key(self): - public_keys = ['first key', 'second key'] + public_keys = ["first key", "second key"] for c in range(4): - for k in ('SSH_KEY', 'SSH_PUBLIC_KEY'): + 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)}) + 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']) + 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'): + for k in ("USER_DATA", "USERDATA"): my_d = os.path.join(self.tmp, k) - populate_context_dir(my_d, {k: USER_DATA, - 'USERDATA_ENCODING': ''}) + 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']) + 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'): + 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']) + self.assertTrue("userdata" in results) + self.assertEqual(b64userdata, results["userdata"]) def test_user_data_base64_encoding(self): - for k in ('USER_DATA', 'USERDATA'): + 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'}) + 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']) + 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'): + 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'): + 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.assertTrue("metadata" in results) + self.assertTrue("local-hostname" in results["metadata"]) self.assertEqual( - PUBLIC_IP, results['metadata']['local-hostname']) + 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'): + 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}) + 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("network-interfaces" in results) self.assertTrue( - IP_BY_MACADDR + '/' + IP4_PREFIX in - results['network-interfaces']['ethernets'][dev]['addresses']) + 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}) + 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("network-interfaces" in results) self.assertTrue( - IP_BY_MACADDR + '/' + IP4_PREFIX in - results['network-interfaces']['ethernets'][dev]['addresses']) + 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}) + 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("network-interfaces" in results) self.assertTrue( - IP_BY_MACADDR + '/' + IP4_PREFIX in - results['network-interfaces']['ethernets'][dev]['addresses']) + 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' - }) + 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("network-interfaces" in results) self.assertTrue( - IP_BY_MACADDR + '/16' in - results['network-interfaces']['ethernets'][dev]['addresses']) + 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': '' - }) + 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("network-interfaces" in results) self.assertTrue( - IP_BY_MACADDR + '/' + IP4_PREFIX in - results['network-interfaces']['ethernets'][dev]['addresses']) + 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, - }) + 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("network-interfaces" in results) self.assertTrue( - IP6_GLOBAL + '/64' in - results['network-interfaces']['ethernets'][dev]['addresses']) + 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, - }) + 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("network-interfaces" in results) self.assertTrue( - IP6_ULA + '/64' in - results['network-interfaces']['ethernets'][dev]['addresses']) + 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, - }) + 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("network-interfaces" in results) self.assertTrue( - IP6_GLOBAL + '/' + IP6_PREFIX in - results['network-interfaces']['ethernets'][dev]['addresses']) + 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, - }) + 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("network-interfaces" in results) self.assertTrue( - IP6_GLOBAL + '/64' in - results['network-interfaces']['ethernets'][dev]['addresses']) + IP6_GLOBAL + "/64" + in results["network-interfaces"]["ethernets"][dev]["addresses"] + ) def test_find_candidates(self): def my_devs_with(criteria): @@ -352,25 +390,28 @@ class TestOpenNebulaDataSource(CiTestCase): 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()) + 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={})) +@mock.patch(DS_PATH + ".net.get_interfaces_by_mac", mock.Mock(return_value={})) class TestOpenNebulaNetwork(unittest.TestCase): - system_nics = ('eth0', 'ens3') + 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', } + "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', } + "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) @@ -380,28 +421,30 @@ class TestOpenNebulaNetwork(unittest.TestCase): 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', } + "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']} + "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') + 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'} + context = {"ETH0_MTU": "1280"} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_mtu('eth0') - self.assertEqual('1280', val) + 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} + context = {"ETH0_IP": PUBLIC_IP} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_ip('eth0', MACADDR) + val = net.get_ip("eth0", MACADDR) self.assertEqual(PUBLIC_IP, val) def test_get_ip_emptystring(self): @@ -410,9 +453,9 @@ class TestOpenNebulaNetwork(unittest.TestCase): It returns IP address created by MAC address if ETH0_IP has empty string. """ - context = {'ETH0_IP': ''} + context = {"ETH0_IP": ""} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_ip('eth0', MACADDR) + val = net.get_ip("eth0", MACADDR) self.assertEqual(IP_BY_MACADDR, val) def test_get_ip6(self): @@ -421,11 +464,12 @@ class TestOpenNebulaNetwork(unittest.TestCase): In this case, IPv6 address is Given by ETH0_IP6. """ context = { - 'ETH0_IP6': IP6_GLOBAL, - 'ETH0_IP6_ULA': '', } + "ETH0_IP6": IP6_GLOBAL, + "ETH0_IP6_ULA": "", + } expected = [IP6_GLOBAL] net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_ip6('eth0') + val = net.get_ip6("eth0") self.assertEqual(expected, val) def test_get_ip6_ula(self): @@ -434,11 +478,12 @@ class TestOpenNebulaNetwork(unittest.TestCase): In this case, IPv6 address is Given by ETH0_IP6_ULA. """ context = { - 'ETH0_IP6': '', - 'ETH0_IP6_ULA': IP6_ULA, } + "ETH0_IP6": "", + "ETH0_IP6_ULA": IP6_ULA, + } expected = [IP6_ULA] net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_ip6('eth0') + val = net.get_ip6("eth0") self.assertEqual(expected, val) def test_get_ip6_dual(self): @@ -447,20 +492,21 @@ class TestOpenNebulaNetwork(unittest.TestCase): In this case, IPv6 addresses are Given by ETH0_IP6 and ETH0_IP6_ULA. """ context = { - 'ETH0_IP6': IP6_GLOBAL, - 'ETH0_IP6_ULA': IP6_ULA, } + "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') + 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} + context = {"ETH0_IP6_PREFIX_LENGTH": IP6_PREFIX} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_ip6_prefix('eth0') + val = net.get_ip6_prefix("eth0") self.assertEqual(IP6_PREFIX, val) def test_get_ip6_prefix_emptystring(self): @@ -469,59 +515,59 @@ class TestOpenNebulaNetwork(unittest.TestCase): It returns default value '64' if ETH0_IP6_PREFIX_LENGTH has empty string. """ - context = {'ETH0_IP6_PREFIX_LENGTH': ''} + context = {"ETH0_IP6_PREFIX_LENGTH": ""} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_ip6_prefix('eth0') - self.assertEqual('64', val) + 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'} + 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) + 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} + for k in ("GATEWAY6", "IP6_GATEWAY"): + context = {"ETH0_" + k: IP6_GW} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_gateway6('eth0') + 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'} + 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) + 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': ''} + context = {"ETH0_MASK": ""} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_mask('eth0') - self.assertEqual('255.255.255.0', val) + 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'} + 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) + val = net.get_network("eth0", MACADDR) + self.assertEqual("1.2.3.0", val) def test_get_network_emptystring(self): """ @@ -529,48 +575,48 @@ class TestOpenNebulaNetwork(unittest.TestCase): It returns network address created by MAC address if ETH0_NETWORK has empty string. """ - context = {'ETH0_NETWORK': ''} + context = {"ETH0_NETWORK": ""} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_network('eth0', MACADDR) - self.assertEqual('10.18.1.0', val) + 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'} + context = {"ETH9_DUMMY": "DUMMY_VALUE"} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_field('eth9', 'dummy') - self.assertEqual('DUMMY_VALUE', val) + 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'} + context = {"ETH9_DUMMY": "DUMMY_VALUE"} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_field('eth9', 'dummy', 'DEFAULT_VALUE') - self.assertEqual('DUMMY_VALUE', val) + 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': ''} + context = {"ETH9_DUMMY": ""} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_field('eth9', 'dummy', 'DEFAULT_VALUE') - self.assertEqual('DEFAULT_VALUE', val) + 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': ''} + context = {"ETH9_DUMMY": ""} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_field('eth9', 'dummy') + val = net.get_field("eth9", "dummy") self.assertEqual(None, val) def test_get_field_nonecontext(self): @@ -578,9 +624,9 @@ class TestOpenNebulaNetwork(unittest.TestCase): Verify get_field('device', 'name') returns None if context value is None. """ - context = {'ETH9_DUMMY': None} + context = {"ETH9_DUMMY": None} net = ds.OpenNebulaNetwork(context, mock.Mock()) - val = net.get_field('eth9', 'dummy') + val = net.get_field("eth9", "dummy") self.assertEqual(None, val) @mock.patch(DS_PATH + ".get_physical_nics_by_mac") @@ -589,31 +635,39 @@ class TestOpenNebulaNetwork(unittest.TestCase): self.maxDiff = None # empty ETH0_GATEWAY context = { - 'ETH0_MAC': '02:00:0a:12:01:01', - 'ETH0_GATEWAY': '', } + "ETH0_MAC": "02:00:0a:12:01:01", + "ETH0_GATEWAY": "", + } for nic in self.system_nics: expected = { - 'version': 2, - 'ethernets': { + "version": 2, + "ethernets": { nic: { - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "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', } + "ETH0_MAC": "02:00:0a:12:01:01", + "ETH0_GATEWAY": "1.2.3.5", + } for nic in self.system_nics: expected = { - 'version': 2, - 'ethernets': { + "version": 2, + "ethernets": { nic: { - 'gateway4': '1.2.3.5', - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "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) @@ -624,31 +678,39 @@ class TestOpenNebulaNetwork(unittest.TestCase): self.maxDiff = None # empty ETH0_GATEWAY6 context = { - 'ETH0_MAC': '02:00:0a:12:01:01', - 'ETH0_GATEWAY6': '', } + "ETH0_MAC": "02:00:0a:12:01:01", + "ETH0_GATEWAY6": "", + } for nic in self.system_nics: expected = { - 'version': 2, - 'ethernets': { + "version": 2, + "ethernets": { nic: { - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "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, } + "ETH0_MAC": "02:00:0a:12:01:01", + "ETH0_GATEWAY6": IP6_GW, + } for nic in self.system_nics: expected = { - 'version': 2, - 'ethernets': { + "version": 2, + "ethernets": { nic: { - 'gateway6': IP6_GW, - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "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) @@ -659,37 +721,46 @@ class TestOpenNebulaNetwork(unittest.TestCase): 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': '', } + "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': { + "version": 2, + "ethernets": { nic: { - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "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, } + "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': { + "version": 2, + "ethernets": { nic: { - 'match': {'macaddress': MACADDR}, - 'addresses': [ - IP_BY_MACADDR + '/' + IP4_PREFIX, - IP6_GLOBAL + '/' + IP6_PREFIX, - IP6_ULA + '/' + IP6_PREFIX]}}} + "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) @@ -700,37 +771,46 @@ class TestOpenNebulaNetwork(unittest.TestCase): 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': '', } + "ETH0_MAC": "02:00:0a:12:01:01", + "DNS": "", + "ETH0_DNS": "", + "ETH0_SEARCH_DOMAIN": "", + } for nic in self.system_nics: expected = { - 'version': 2, - 'ethernets': { + "version": 2, + "ethernets": { nic: { - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "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', } + "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': { + "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]}}} + "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) @@ -741,31 +821,39 @@ class TestOpenNebulaNetwork(unittest.TestCase): self.maxDiff = None # empty ETH0_MTU context = { - 'ETH0_MAC': '02:00:0a:12:01:01', - 'ETH0_MTU': '', } + "ETH0_MAC": "02:00:0a:12:01:01", + "ETH0_MTU": "", + } for nic in self.system_nics: expected = { - 'version': 2, - 'ethernets': { + "version": 2, + "ethernets": { nic: { - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "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', } + "ETH0_MAC": "02:00:0a:12:01:01", + "ETH0_MTU": "1280", + } for nic in self.system_nics: expected = { - 'version': 2, - 'ethernets': { + "version": 2, + "ethernets": { nic: { - 'mtu': '1280', - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "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) @@ -776,11 +864,14 @@ class TestOpenNebulaNetwork(unittest.TestCase): m_get_phys_by_mac.return_value = {MACADDR: nic} net = ds.OpenNebulaNetwork({}, mock.Mock()) expected = { - 'version': 2, - 'ethernets': { + "version": 2, + "ethernets": { nic: { - 'match': {'macaddress': MACADDR}, - 'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}} + "match": {"macaddress": MACADDR}, + "addresses": [IP_BY_MACADDR + "/" + IP4_PREFIX], + } + }, + } self.assertEqual(net.gen_conf(), expected) @@ -795,71 +886,82 @@ class TestOpenNebulaNetwork(unittest.TestCase): 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': '', + "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}) + net = ds.OpenNebulaNetwork( + context, mock.Mock(), system_nics_by_mac={MACADDR: nic} + ) expected = { - 'version': 2, - 'ethernets': { + "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']}}}} + "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', + "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}) + net = ds.OpenNebulaNetwork( + context, mock.Mock(), system_nics_by_mac={MACADDR: nic} + ) expected = { - 'version': 2, - 'ethernets': { + "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'}}} + "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()) @@ -869,62 +971,67 @@ class TestOpenNebulaNetwork(unittest.TestCase): 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', + "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'} + 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']}}}} + "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()) @@ -932,7 +1039,7 @@ class TestOpenNebulaNetwork(unittest.TestCase): class TestParseShellConfig: @pytest.mark.allow_subp_for("bash") def test_no_seconds(self): - cfg = '\n'.join(["foo=bar", "SECONDS=2", "xx=foo"]) + 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"} @@ -971,7 +1078,8 @@ class TestGetPhysicalNicsByMac: 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}) + data += "%s='%s'\n" % (k.upper(), v.replace(r"'", r"'\''")) + populate_dir(path, {"context.sh": data}) + # vi: ts=4 expandtab |