diff options
Diffstat (limited to 'tests/unittests/sources/test_digitalocean.py')
| -rw-r--r-- | tests/unittests/sources/test_digitalocean.py | 389 |
1 files changed, 389 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..f3e6224e --- /dev/null +++ b/tests/unittests/sources/test_digitalocean.py @@ -0,0 +1,389 @@ +# 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, settings +from cloudinit.sources import DataSourceDigitalOcean +from cloudinit.sources.helpers import digitalocean +from tests.unittests.helpers import CiTestCase, mock + +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 |