summaryrefslogtreecommitdiff
path: root/tests/unittests/sources/test_digitalocean.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests/sources/test_digitalocean.py')
-rw-r--r--tests/unittests/sources/test_digitalocean.py372
1 files changed, 372 insertions, 0 deletions
diff --git a/tests/unittests/sources/test_digitalocean.py b/tests/unittests/sources/test_digitalocean.py
new file mode 100644
index 00000000..351bf7ba
--- /dev/null
+++ b/tests/unittests/sources/test_digitalocean.py
@@ -0,0 +1,372 @@
+# Copyright (C) 2014 Neal Shrader
+#
+# Author: Neal Shrader <neal@digitalocean.com>
+# Author: Ben Howard <bh@digitalocean.com>
+# Author: Scott Moser <smoser@ubuntu.com>
+#
+# This file is part of cloud-init. See LICENSE file for license information.
+
+import json
+
+from cloudinit import helpers
+from cloudinit import settings
+from cloudinit.sources import DataSourceDigitalOcean
+from cloudinit.sources.helpers import digitalocean
+
+from tests.unittests.helpers import mock, CiTestCase
+
+DO_MULTIPLE_KEYS = ["ssh-rsa AAAAB3NzaC1yc2EAAAA... test1@do.co",
+ "ssh-rsa AAAAB3NzaC1yc2EAAAA... test2@do.co"]
+DO_SINGLE_KEY = "ssh-rsa AAAAB3NzaC1yc2EAAAA... test@do.co"
+
+# the following JSON was taken from droplet (that's why its a string)
+DO_META = json.loads("""
+{
+ "droplet_id": "22532410",
+ "hostname": "utl-96268",
+ "vendor_data": "vendordata goes here",
+ "user_data": "userdata goes here",
+ "public_keys": "",
+ "auth_key": "authorization_key",
+ "region": "nyc3",
+ "interfaces": {
+ "private": [
+ {
+ "ipv4": {
+ "ip_address": "10.132.6.205",
+ "netmask": "255.255.0.0",
+ "gateway": "10.132.0.1"
+ },
+ "mac": "04:01:57:d1:9e:02",
+ "type": "private"
+ }
+ ],
+ "public": [
+ {
+ "ipv4": {
+ "ip_address": "192.0.0.20",
+ "netmask": "255.255.255.0",
+ "gateway": "104.236.0.1"
+ },
+ "ipv6": {
+ "ip_address": "2604:A880:0800:0000:1000:0000:0000:0000",
+ "cidr": 64,
+ "gateway": "2604:A880:0800:0000:0000:0000:0000:0001"
+ },
+ "anchor_ipv4": {
+ "ip_address": "10.0.0.5",
+ "netmask": "255.255.0.0",
+ "gateway": "10.0.0.1"
+ },
+ "mac": "04:01:57:d1:9e:01",
+ "type": "public"
+ }
+ ]
+ },
+ "floating_ip": {
+ "ipv4": {
+ "active": false
+ }
+ },
+ "dns": {
+ "nameservers": [
+ "2001:4860:4860::8844",
+ "2001:4860:4860::8888",
+ "8.8.8.8"
+ ]
+ }
+}
+""")
+
+# This has no private interface
+DO_META_2 = {
+ "droplet_id": 27223699,
+ "hostname": "smtest1",
+ "vendor_data": "\n".join([
+ ('"Content-Type: multipart/mixed; '
+ 'boundary=\"===============8645434374073493512==\"'),
+ 'MIME-Version: 1.0',
+ '',
+ '--===============8645434374073493512==',
+ 'MIME-Version: 1.0'
+ 'Content-Type: text/cloud-config; charset="us-ascii"'
+ 'Content-Transfer-Encoding: 7bit'
+ 'Content-Disposition: attachment; filename="cloud-config"'
+ '',
+ '#cloud-config',
+ 'disable_root: false',
+ 'manage_etc_hosts: true',
+ '',
+ '',
+ '--===============8645434374073493512=='
+ ]),
+ "public_keys": [
+ "ssh-rsa AAAAB3NzaN...N3NtHw== smoser@brickies"
+ ],
+ "auth_key": "88888888888888888888888888888888",
+ "region": "nyc3",
+ "interfaces": {
+ "public": [{
+ "ipv4": {
+ "ip_address": "45.55.249.133",
+ "netmask": "255.255.192.0",
+ "gateway": "45.55.192.1"
+ },
+ "anchor_ipv4": {
+ "ip_address": "10.17.0.5",
+ "netmask": "255.255.0.0",
+ "gateway": "10.17.0.1"
+ },
+ "mac": "ae:cc:08:7c:88:00",
+ "type": "public"
+ }]
+ },
+ "floating_ip": {"ipv4": {"active": True, "ip_address": "138.197.59.92"}},
+ "dns": {"nameservers": ["8.8.8.8", "8.8.4.4"]},
+ "tags": None,
+}
+
+DO_META['public_keys'] = DO_SINGLE_KEY
+
+MD_URL = 'http://169.254.169.254/metadata/v1.json'
+
+
+def _mock_dmi():
+ return (True, DO_META.get('id'))
+
+
+class TestDataSourceDigitalOcean(CiTestCase):
+ """
+ Test reading the meta-data
+ """
+ def setUp(self):
+ super(TestDataSourceDigitalOcean, self).setUp()
+ self.tmp = self.tmp_dir()
+
+ def get_ds(self, get_sysinfo=_mock_dmi):
+ ds = DataSourceDigitalOcean.DataSourceDigitalOcean(
+ settings.CFG_BUILTIN, None, helpers.Paths({'run_dir': self.tmp}))
+ ds.use_ip4LL = False
+ if get_sysinfo is not None:
+ ds._get_sysinfo = get_sysinfo
+ return ds
+
+ @mock.patch('cloudinit.sources.helpers.digitalocean.read_sysinfo')
+ def test_returns_false_not_on_docean(self, m_read_sysinfo):
+ m_read_sysinfo.return_value = (False, None)
+ ds = self.get_ds(get_sysinfo=None)
+ self.assertEqual(False, ds.get_data())
+ self.assertTrue(m_read_sysinfo.called)
+
+ @mock.patch('cloudinit.sources.helpers.digitalocean.read_metadata')
+ def test_metadata(self, mock_readmd):
+ mock_readmd.return_value = DO_META.copy()
+
+ ds = self.get_ds()
+ ret = ds.get_data()
+ self.assertTrue(ret)
+
+ self.assertTrue(mock_readmd.called)
+
+ self.assertEqual(DO_META.get('user_data'), ds.get_userdata_raw())
+ self.assertEqual(DO_META.get('vendor_data'), ds.get_vendordata_raw())
+ self.assertEqual(DO_META.get('region'), ds.availability_zone)
+ self.assertEqual(DO_META.get('droplet_id'), ds.get_instance_id())
+ self.assertEqual(DO_META.get('hostname'), ds.get_hostname())
+
+ # Single key
+ self.assertEqual([DO_META.get('public_keys')],
+ ds.get_public_ssh_keys())
+
+ self.assertIsInstance(ds.get_public_ssh_keys(), list)
+
+ @mock.patch('cloudinit.sources.helpers.digitalocean.read_metadata')
+ def test_multiple_ssh_keys(self, mock_readmd):
+ metadata = DO_META.copy()
+ metadata['public_keys'] = DO_MULTIPLE_KEYS
+ mock_readmd.return_value = metadata.copy()
+
+ ds = self.get_ds()
+ ret = ds.get_data()
+ self.assertTrue(ret)
+
+ self.assertTrue(mock_readmd.called)
+
+ # Multiple keys
+ self.assertEqual(metadata['public_keys'], ds.get_public_ssh_keys())
+ self.assertIsInstance(ds.get_public_ssh_keys(), list)
+
+
+class TestNetworkConvert(CiTestCase):
+
+ def _get_networking(self):
+ self.m_get_by_mac.return_value = {
+ '04:01:57:d1:9e:01': 'ens1',
+ '04:01:57:d1:9e:02': 'ens2',
+ 'b8:ae:ed:75:5f:9a': 'enp0s25',
+ 'ae:cc:08:7c:88:00': 'meta2p1'}
+ netcfg = digitalocean.convert_network_configuration(
+ DO_META['interfaces'], DO_META['dns']['nameservers'])
+ self.assertIn('config', netcfg)
+ return netcfg
+
+ def setUp(self):
+ super(TestNetworkConvert, self).setUp()
+ self.add_patch('cloudinit.net.get_interfaces_by_mac', 'm_get_by_mac')
+
+ def test_networking_defined(self):
+ netcfg = self._get_networking()
+ self.assertIsNotNone(netcfg)
+ dns_defined = False
+
+ for part in netcfg.get('config'):
+ n_type = part.get('type')
+ print("testing part ", n_type, "\n", json.dumps(part, indent=3))
+
+ if n_type == 'nameserver':
+ n_address = part.get('address')
+ self.assertIsNotNone(n_address)
+ self.assertEqual(len(n_address), 3)
+
+ dns_resolvers = DO_META["dns"]["nameservers"]
+ for x in n_address:
+ self.assertIn(x, dns_resolvers)
+ dns_defined = True
+
+ else:
+ n_subnets = part.get('type')
+ n_name = part.get('name')
+ n_mac = part.get('mac_address')
+
+ self.assertIsNotNone(n_type)
+ self.assertIsNotNone(n_subnets)
+ self.assertIsNotNone(n_name)
+ self.assertIsNotNone(n_mac)
+
+ self.assertTrue(dns_defined)
+
+ def _get_nic_definition(self, int_type, expected_name):
+ """helper function to return if_type (i.e. public) and the expected
+ name used by cloud-init (i.e eth0)"""
+ netcfg = self._get_networking()
+ meta_def = (DO_META.get('interfaces')).get(int_type)[0]
+
+ self.assertEqual(int_type, meta_def.get('type'))
+
+ for nic_def in netcfg.get('config'):
+ print(nic_def)
+ if nic_def.get('name') == expected_name:
+ return nic_def, meta_def
+
+ def _get_match_subn(self, subnets, ip_addr):
+ """get the matching subnet definition based on ip address"""
+ for subn in subnets:
+ address = subn.get('address')
+ self.assertIsNotNone(address)
+
+ # equals won't work because of ipv6 addressing being in
+ # cidr notation, i.e fe00::1/64
+ if ip_addr in address:
+ print(json.dumps(subn, indent=3))
+ return subn
+
+ def test_correct_gateways_defined(self):
+ """test to make sure the eth0 ipv4 and ipv6 gateways are defined"""
+ netcfg = self._get_networking()
+ gateways = []
+ for nic_def in netcfg.get('config'):
+ if nic_def.get('type') != 'physical':
+ continue
+ for subn in nic_def.get('subnets'):
+ if 'gateway' in subn:
+ gateways.append(subn.get('gateway'))
+
+ # we should have two gateways, one ipv4 and ipv6
+ self.assertEqual(len(gateways), 2)
+
+ # make that the ipv6 gateway is there
+ (nic_def, meta_def) = self._get_nic_definition('public', 'eth0')
+ ipv4_def = meta_def.get('ipv4')
+ self.assertIn(ipv4_def.get('gateway'), gateways)
+
+ # make sure the the ipv6 gateway is there
+ ipv6_def = meta_def.get('ipv6')
+ self.assertIn(ipv6_def.get('gateway'), gateways)
+
+ def test_public_interface_defined(self):
+ """test that the public interface is defined as eth0"""
+ (nic_def, meta_def) = self._get_nic_definition('public', 'eth0')
+ self.assertEqual('eth0', nic_def.get('name'))
+ self.assertEqual(meta_def.get('mac'), nic_def.get('mac_address'))
+ self.assertEqual('physical', nic_def.get('type'))
+
+ def test_private_interface_defined(self):
+ """test that the private interface is defined as eth1"""
+ (nic_def, meta_def) = self._get_nic_definition('private', 'eth1')
+ self.assertEqual('eth1', nic_def.get('name'))
+ self.assertEqual(meta_def.get('mac'), nic_def.get('mac_address'))
+ self.assertEqual('physical', nic_def.get('type'))
+
+ def test_public_interface_ipv6(self):
+ """test public ipv6 addressing"""
+ (nic_def, meta_def) = self._get_nic_definition('public', 'eth0')
+ ipv6_def = meta_def.get('ipv6')
+ self.assertIsNotNone(ipv6_def)
+
+ subn_def = self._get_match_subn(nic_def.get('subnets'),
+ ipv6_def.get('ip_address'))
+
+ cidr_notated_address = "{0}/{1}".format(ipv6_def.get('ip_address'),
+ ipv6_def.get('cidr'))
+
+ self.assertEqual(cidr_notated_address, subn_def.get('address'))
+ self.assertEqual(ipv6_def.get('gateway'), subn_def.get('gateway'))
+
+ def test_public_interface_ipv4(self):
+ """test public ipv4 addressing"""
+ (nic_def, meta_def) = self._get_nic_definition('public', 'eth0')
+ ipv4_def = meta_def.get('ipv4')
+ self.assertIsNotNone(ipv4_def)
+
+ subn_def = self._get_match_subn(nic_def.get('subnets'),
+ ipv4_def.get('ip_address'))
+
+ self.assertEqual(ipv4_def.get('netmask'), subn_def.get('netmask'))
+ self.assertEqual(ipv4_def.get('gateway'), subn_def.get('gateway'))
+
+ def test_public_interface_anchor_ipv4(self):
+ """test public ipv4 addressing"""
+ (nic_def, meta_def) = self._get_nic_definition('public', 'eth0')
+ ipv4_def = meta_def.get('anchor_ipv4')
+ self.assertIsNotNone(ipv4_def)
+
+ subn_def = self._get_match_subn(nic_def.get('subnets'),
+ ipv4_def.get('ip_address'))
+
+ self.assertEqual(ipv4_def.get('netmask'), subn_def.get('netmask'))
+ self.assertNotIn('gateway', subn_def)
+
+ @mock.patch('cloudinit.net.get_interfaces_by_mac')
+ def test_convert_without_private(self, m_get_by_mac):
+ m_get_by_mac.return_value = {
+ 'b8:ae:ed:75:5f:9a': 'enp0s25',
+ 'ae:cc:08:7c:88:00': 'meta2p1'}
+ netcfg = digitalocean.convert_network_configuration(
+ DO_META_2['interfaces'], DO_META_2['dns']['nameservers'])
+
+ # print(netcfg)
+ byname = {}
+ for i in netcfg['config']:
+ if 'name' in i:
+ if i['name'] in byname:
+ raise ValueError("name '%s' in config twice: %s" %
+ (i['name'], netcfg))
+ byname[i['name']] = i
+ self.assertTrue('eth0' in byname)
+ self.assertTrue('subnets' in byname['eth0'])
+ eth0 = byname['eth0']
+ self.assertEqual(
+ sorted(['45.55.249.133', '10.17.0.5']),
+ sorted([i['address'] for i in eth0['subnets']]))
+
+# vi: ts=4 expandtab