# Author: Antti Myyrä # # 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 import sources from cloudinit.sources.DataSourceUpCloud import DataSourceUpCloud, \ DataSourceUpCloudLocal from tests.unittests.helpers import mock, CiTestCase UC_METADATA = json.loads(""" { "cloud_name": "upcloud", "instance_id": "00322b68-0096-4042-9406-faad61922128", "hostname": "test.example.com", "platform": "servers", "subplatform": "metadata (http://169.254.169.254)", "public_keys": [ "ssh-rsa AAAAB.... test1@example.com", "ssh-rsa AAAAB.... test2@example.com" ], "region": "fi-hel2", "network": { "interfaces": [ { "index": 1, "ip_addresses": [ { "address": "94.237.105.53", "dhcp": true, "dns": [ "94.237.127.9", "94.237.40.9" ], "family": "IPv4", "floating": false, "gateway": "94.237.104.1", "network": "94.237.104.0/22" }, { "address": "94.237.105.50", "dhcp": false, "dns": null, "family": "IPv4", "floating": true, "gateway": "", "network": "94.237.105.50/32" } ], "mac": "3a:d6:ba:4a:36:e7", "network_id": "031457f4-0f8c-483c-96f2-eccede02909c", "type": "public" }, { "index": 2, "ip_addresses": [ { "address": "10.6.3.27", "dhcp": true, "dns": null, "family": "IPv4", "floating": false, "gateway": "10.6.0.1", "network": "10.6.0.0/22" } ], "mac": "3a:d6:ba:4a:84:cc", "network_id": "03d82553-5bea-4132-b29a-e1cf67ec2dd1", "type": "utility" }, { "index": 3, "ip_addresses": [ { "address": "2a04:3545:1000:720:38d6:baff:fe4a:63e7", "dhcp": true, "dns": [ "2a04:3540:53::1", "2a04:3544:53::1" ], "family": "IPv6", "floating": false, "gateway": "2a04:3545:1000:720::1", "network": "2a04:3545:1000:720::/64" } ], "mac": "3a:d6:ba:4a:63:e7", "network_id": "03000000-0000-4000-8046-000000000000", "type": "public" }, { "index": 4, "ip_addresses": [ { "address": "172.30.1.10", "dhcp": true, "dns": null, "family": "IPv4", "floating": false, "gateway": "172.30.1.1", "network": "172.30.1.0/24" } ], "mac": "3a:d6:ba:4a:8a:e1", "network_id": "035a0a4a-7704-4de5-820d-189fc8132714", "type": "private" } ], "dns": [ "94.237.127.9", "94.237.40.9" ] }, "storage": { "disks": [ { "id": "014efb65-223b-4d44-8f0a-c29535b88dcf", "serial": "014efb65223b4d448f0a", "size": 10240, "type": "disk", "tier": "maxiops" } ] }, "tags": [], "user_data": "", "vendor_data": "" } """) UC_METADATA["user_data"] = b"""#cloud-config runcmd: - [touch, /root/cloud-init-worked ] """ MD_URL = 'http://169.254.169.254/metadata/v1.json' def _mock_dmi(): return True, "00322b68-0096-4042-9406-faad61922128" class TestUpCloudMetadata(CiTestCase): """ Test reading the meta-data """ def setUp(self): super(TestUpCloudMetadata, self).setUp() self.tmp = self.tmp_dir() def get_ds(self, get_sysinfo=_mock_dmi): ds = DataSourceUpCloud( settings.CFG_BUILTIN, None, helpers.Paths({'run_dir': self.tmp})) if get_sysinfo: ds._get_sysinfo = get_sysinfo return ds @mock.patch('cloudinit.sources.helpers.upcloud.read_sysinfo') def test_returns_false_not_on_upcloud(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.upcloud.read_metadata') def test_metadata(self, mock_readmd): mock_readmd.return_value = UC_METADATA.copy() ds = self.get_ds() ds.perform_dhcp_setup = False ret = ds.get_data() self.assertTrue(ret) self.assertTrue(mock_readmd.called) self.assertEqual(UC_METADATA.get('user_data'), ds.get_userdata_raw()) self.assertEqual(UC_METADATA.get('vendor_data'), ds.get_vendordata_raw()) self.assertEqual(UC_METADATA.get('region'), ds.availability_zone) self.assertEqual(UC_METADATA.get('instance_id'), ds.get_instance_id()) self.assertEqual(UC_METADATA.get('cloud_name'), ds.cloud_name) self.assertEqual(UC_METADATA.get('public_keys'), ds.get_public_ssh_keys()) self.assertIsInstance(ds.get_public_ssh_keys(), list) class TestUpCloudNetworkSetup(CiTestCase): """ Test reading the meta-data on networked context """ def setUp(self): super(TestUpCloudNetworkSetup, self).setUp() self.tmp = self.tmp_dir() def get_ds(self, get_sysinfo=_mock_dmi): ds = DataSourceUpCloudLocal( settings.CFG_BUILTIN, None, helpers.Paths({'run_dir': self.tmp})) if get_sysinfo: ds._get_sysinfo = get_sysinfo return ds @mock.patch('cloudinit.sources.helpers.upcloud.read_metadata') @mock.patch('cloudinit.net.find_fallback_nic') @mock.patch('cloudinit.net.dhcp.maybe_perform_dhcp_discovery') @mock.patch('cloudinit.net.dhcp.EphemeralIPv4Network') def test_network_configured_metadata(self, m_net, m_dhcp, m_fallback_nic, mock_readmd): mock_readmd.return_value = UC_METADATA.copy() m_fallback_nic.return_value = 'eth1' m_dhcp.return_value = [{ 'interface': 'eth1', 'fixed-address': '10.6.3.27', 'routers': '10.6.0.1', 'subnet-mask': '22', 'broadcast-address': '10.6.3.255'} ] ds = self.get_ds() ret = ds.get_data() self.assertTrue(ret) self.assertTrue(m_dhcp.called) m_dhcp.assert_called_with('eth1', None) m_net.assert_called_once_with( broadcast='10.6.3.255', interface='eth1', ip='10.6.3.27', prefix_or_mask='22', router='10.6.0.1', static_routes=None ) self.assertTrue(mock_readmd.called) self.assertEqual(UC_METADATA.get('region'), ds.availability_zone) self.assertEqual(UC_METADATA.get('instance_id'), ds.get_instance_id()) self.assertEqual(UC_METADATA.get('cloud_name'), ds.cloud_name) @mock.patch('cloudinit.sources.helpers.upcloud.read_metadata') @mock.patch('cloudinit.net.get_interfaces_by_mac') def test_network_configuration(self, m_get_by_mac, mock_readmd): mock_readmd.return_value = UC_METADATA.copy() raw_ifaces = UC_METADATA.get('network').get('interfaces') self.assertEqual(4, len(raw_ifaces)) m_get_by_mac.return_value = { raw_ifaces[0].get('mac'): 'eth0', raw_ifaces[1].get('mac'): 'eth1', raw_ifaces[2].get('mac'): 'eth2', raw_ifaces[3].get('mac'): 'eth3', } ds = self.get_ds() ds.perform_dhcp_setup = False ret = ds.get_data() self.assertTrue(ret) self.assertTrue(mock_readmd.called) netcfg = ds.network_config self.assertEqual(1, netcfg.get('version')) config = netcfg.get('config') self.assertIsInstance(config, list) self.assertEqual(5, len(config)) self.assertEqual('physical', config[3].get('type')) self.assertEqual(raw_ifaces[2].get('mac'), config[2] .get('mac_address')) self.assertEqual(1, len(config[2].get('subnets'))) self.assertEqual('ipv6_dhcpv6-stateless', config[2].get('subnets')[0] .get('type')) self.assertEqual(2, len(config[0].get('subnets'))) self.assertEqual('static', config[0].get('subnets')[1].get('type')) dns = config[4] self.assertEqual('nameserver', dns.get('type')) self.assertEqual(2, len(dns.get('address'))) self.assertEqual( UC_METADATA.get('network').get('dns')[1], dns.get('address')[1] ) class TestUpCloudDatasourceLoading(CiTestCase): def test_get_datasource_list_returns_in_local(self): deps = (sources.DEP_FILESYSTEM, ) ds_list = sources.DataSourceUpCloud.get_datasource_list(deps) self.assertEqual(ds_list, [DataSourceUpCloudLocal]) def test_get_datasource_list_returns_in_normal(self): deps = (sources.DEP_FILESYSTEM, sources.DEP_NETWORK) ds_list = sources.DataSourceUpCloud.get_datasource_list(deps) self.assertEqual(ds_list, [DataSourceUpCloud]) def test_list_sources_finds_ds(self): found = sources.list_sources( ['UpCloud'], (sources.DEP_FILESYSTEM, sources.DEP_NETWORK), ['cloudinit.sources']) self.assertEqual([DataSourceUpCloud], found) # vi: ts=4 expandtab