# Author: Antti Myyrä # # This file is part of cloud-init. See LICENSE file for license information. import json from cloudinit import helpers, settings, sources from cloudinit.sources.DataSourceUpCloud import ( DataSourceUpCloud, DataSourceUpCloudLocal, ) from tests.unittests.helpers import CiTestCase, mock 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