diff options
Diffstat (limited to 'tests/unittests/test_datasource/test_opennebula.py')
| -rw-r--r-- | tests/unittests/test_datasource/test_opennebula.py | 977 |
1 files changed, 0 insertions, 977 deletions
diff --git a/tests/unittests/test_datasource/test_opennebula.py b/tests/unittests/test_datasource/test_opennebula.py deleted file mode 100644 index 283b65c2..00000000 --- a/tests/unittests/test_datasource/test_opennebula.py +++ /dev/null @@ -1,977 +0,0 @@ -# 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 cloudinit.tests.helpers import mock, populate_dir, CiTestCase - -import os -import pwd -import unittest - -import pytest - - -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 |