diff options
Diffstat (limited to 'tests/unittests/sources/test_aliyun.py')
| -rw-r--r-- | tests/unittests/sources/test_aliyun.py | 217 |
1 files changed, 128 insertions, 89 deletions
diff --git a/tests/unittests/sources/test_aliyun.py b/tests/unittests/sources/test_aliyun.py index 00209913..8a61d5ee 100644 --- a/tests/unittests/sources/test_aliyun.py +++ b/tests/unittests/sources/test_aliyun.py @@ -1,38 +1,46 @@ # This file is part of cloud-init. See LICENSE file for license information. import functools -import httpretty import os from unittest import mock +import httpretty + from cloudinit import helpers from cloudinit.sources import DataSourceAliYun as ay from cloudinit.sources.DataSourceEc2 import convert_ec2_metadata_network_config from tests.unittests import helpers as test_helpers DEFAULT_METADATA = { - 'instance-id': 'aliyun-test-vm-00', - 'eipv4': '10.0.0.1', - 'hostname': 'test-hostname', - 'image-id': 'm-test', - 'launch-index': '0', - 'mac': '00:16:3e:00:00:00', - 'network-type': 'vpc', - 'private-ipv4': '192.168.0.1', - 'serial-number': 'test-string', - 'vpc-cidr-block': '192.168.0.0/16', - 'vpc-id': 'test-vpc', - 'vswitch-id': 'test-vpc', - 'vswitch-cidr-block': '192.168.0.0/16', - 'zone-id': 'test-zone-1', - 'ntp-conf': {'ntp_servers': [ - 'ntp1.aliyun.com', - 'ntp2.aliyun.com', - 'ntp3.aliyun.com']}, - 'source-address': ['http://mirrors.aliyun.com', - 'http://mirrors.aliyuncs.com'], - 'public-keys': {'key-pair-1': {'openssh-key': 'ssh-rsa AAAAB3...'}, - 'key-pair-2': {'openssh-key': 'ssh-rsa AAAAB3...'}} + "instance-id": "aliyun-test-vm-00", + "eipv4": "10.0.0.1", + "hostname": "test-hostname", + "image-id": "m-test", + "launch-index": "0", + "mac": "00:16:3e:00:00:00", + "network-type": "vpc", + "private-ipv4": "192.168.0.1", + "serial-number": "test-string", + "vpc-cidr-block": "192.168.0.0/16", + "vpc-id": "test-vpc", + "vswitch-id": "test-vpc", + "vswitch-cidr-block": "192.168.0.0/16", + "zone-id": "test-zone-1", + "ntp-conf": { + "ntp_servers": [ + "ntp1.aliyun.com", + "ntp2.aliyun.com", + "ntp3.aliyun.com", + ] + }, + "source-address": [ + "http://mirrors.aliyun.com", + "http://mirrors.aliyuncs.com", + ], + "public-keys": { + "key-pair-1": {"openssh-key": "ssh-rsa AAAAB3..."}, + "key-pair-2": {"openssh-key": "ssh-rsa AAAAB3..."}, + }, } DEFAULT_USERDATA = """\ @@ -46,21 +54,22 @@ def register_mock_metaserver(base_url, data): if isinstance(body, str): register(base_url, body) elif isinstance(body, list): - register(base_url.rstrip('/'), '\n'.join(body) + '\n') + register(base_url.rstrip("/"), "\n".join(body) + "\n") elif isinstance(body, dict): if not body: - register(base_url.rstrip('/') + '/', 'not found', - status_code=404) + register( + base_url.rstrip("/") + "/", "not found", status_code=404 + ) vals = [] for k, v in body.items(): if isinstance(v, (str, list)): - suffix = k.rstrip('/') + suffix = k.rstrip("/") else: - suffix = k.rstrip('/') + '/' + suffix = k.rstrip("/") + "/" vals.append(suffix) - url = base_url.rstrip('/') + '/' + suffix + url = base_url.rstrip("/") + "/" + suffix register_helper(register, url, v) - register(base_url, '\n'.join(vals) + '\n') + register(base_url, "\n".join(vals) + "\n") register = functools.partial(httpretty.register_uri, httpretty.GET) register_helper(register, base_url, data) @@ -69,9 +78,9 @@ def register_mock_metaserver(base_url, data): class TestAliYunDatasource(test_helpers.HttprettyTestCase): def setUp(self): super(TestAliYunDatasource, self).setUp() - cfg = {'datasource': {'AliYun': {'timeout': '1', 'max_wait': '1'}}} + cfg = {"datasource": {"AliYun": {"timeout": "1", "max_wait": "1"}}} distro = {} - paths = helpers.Paths({'run_dir': self.tmp_dir()}) + paths = helpers.Paths({"run_dir": self.tmp_dir()}) self.ds = ay.DataSourceAliYun(cfg, distro, paths) self.metadata_address = self.ds.metadata_urls[0] @@ -85,15 +94,20 @@ class TestAliYunDatasource(test_helpers.HttprettyTestCase): @property def metadata_url(self): - return os.path.join( - self.metadata_address, - self.ds.min_metadata_version, 'meta-data') + '/' + return ( + os.path.join( + self.metadata_address, + self.ds.min_metadata_version, + "meta-data", + ) + + "/" + ) @property def userdata_url(self): return os.path.join( - self.metadata_address, - self.ds.min_metadata_version, 'user-data') + self.metadata_address, self.ds.min_metadata_version, "user-data" + ) # EC2 provides an instance-identity document which must return 404 here # for this test to pass. @@ -103,9 +117,12 @@ class TestAliYunDatasource(test_helpers.HttprettyTestCase): @property def identity_url(self): - return os.path.join(self.metadata_address, - self.ds.min_metadata_version, - 'dynamic', 'instance-identity') + return os.path.join( + self.metadata_address, + self.ds.min_metadata_version, + "dynamic", + "instance-identity", + ) def regist_default_server(self): register_mock_metaserver(self.metadata_url, self.default_metadata) @@ -114,21 +131,26 @@ class TestAliYunDatasource(test_helpers.HttprettyTestCase): def _test_get_data(self): self.assertEqual(self.ds.metadata, self.default_metadata) - self.assertEqual(self.ds.userdata_raw, - self.default_userdata.encode('utf8')) + self.assertEqual( + self.ds.userdata_raw, self.default_userdata.encode("utf8") + ) def _test_get_sshkey(self): - pub_keys = [v['openssh-key'] for (_, v) in - self.default_metadata['public-keys'].items()] + pub_keys = [ + v["openssh-key"] + for (_, v) in self.default_metadata["public-keys"].items() + ] self.assertEqual(self.ds.get_public_ssh_keys(), pub_keys) def _test_get_iid(self): - self.assertEqual(self.default_metadata['instance-id'], - self.ds.get_instance_id()) + self.assertEqual( + self.default_metadata["instance-id"], self.ds.get_instance_id() + ) def _test_host_name(self): - self.assertEqual(self.default_metadata['hostname'], - self.ds.get_hostname()) + self.assertEqual( + self.default_metadata["hostname"], self.ds.get_hostname() + ) @mock.patch("cloudinit.sources.DataSourceAliYun._is_aliyun") def test_with_mock_server(self, m_is_aliyun): @@ -141,10 +163,11 @@ class TestAliYunDatasource(test_helpers.HttprettyTestCase): self._test_get_sshkey() self._test_get_iid() self._test_host_name() - self.assertEqual('aliyun', self.ds.cloud_name) - self.assertEqual('ec2', self.ds.platform) + self.assertEqual("aliyun", self.ds.cloud_name) + self.assertEqual("ec2", self.ds.platform) self.assertEqual( - 'metadata (http://100.100.100.200)', self.ds.subplatform) + "metadata (http://100.100.100.200)", self.ds.subplatform + ) @mock.patch("cloudinit.sources.DataSourceAliYun._is_aliyun") def test_returns_false_when_not_on_aliyun(self, m_is_aliyun): @@ -159,30 +182,38 @@ class TestAliYunDatasource(test_helpers.HttprettyTestCase): public_keys = {} self.assertEqual(ay.parse_public_keys(public_keys), []) - public_keys = {'key-pair-0': 'ssh-key-0'} - self.assertEqual(ay.parse_public_keys(public_keys), - [public_keys['key-pair-0']]) + public_keys = {"key-pair-0": "ssh-key-0"} + self.assertEqual( + ay.parse_public_keys(public_keys), [public_keys["key-pair-0"]] + ) - public_keys = {'key-pair-0': 'ssh-key-0', 'key-pair-1': 'ssh-key-1'} - self.assertEqual(set(ay.parse_public_keys(public_keys)), - set([public_keys['key-pair-0'], - public_keys['key-pair-1']])) + public_keys = {"key-pair-0": "ssh-key-0", "key-pair-1": "ssh-key-1"} + self.assertEqual( + set(ay.parse_public_keys(public_keys)), + set([public_keys["key-pair-0"], public_keys["key-pair-1"]]), + ) - public_keys = {'key-pair-0': ['ssh-key-0', 'ssh-key-1']} - self.assertEqual(ay.parse_public_keys(public_keys), - public_keys['key-pair-0']) + public_keys = {"key-pair-0": ["ssh-key-0", "ssh-key-1"]} + self.assertEqual( + ay.parse_public_keys(public_keys), public_keys["key-pair-0"] + ) - public_keys = {'key-pair-0': {'openssh-key': []}} + public_keys = {"key-pair-0": {"openssh-key": []}} self.assertEqual(ay.parse_public_keys(public_keys), []) - public_keys = {'key-pair-0': {'openssh-key': 'ssh-key-0'}} - self.assertEqual(ay.parse_public_keys(public_keys), - [public_keys['key-pair-0']['openssh-key']]) + public_keys = {"key-pair-0": {"openssh-key": "ssh-key-0"}} + self.assertEqual( + ay.parse_public_keys(public_keys), + [public_keys["key-pair-0"]["openssh-key"]], + ) - public_keys = {'key-pair-0': {'openssh-key': ['ssh-key-0', - 'ssh-key-1']}} - self.assertEqual(ay.parse_public_keys(public_keys), - public_keys['key-pair-0']['openssh-key']) + public_keys = { + "key-pair-0": {"openssh-key": ["ssh-key-0", "ssh-key-1"]} + } + self.assertEqual( + ay.parse_public_keys(public_keys), + public_keys["key-pair-0"]["openssh-key"], + ) def test_route_metric_calculated_without_device_number(self): """Test that route-metric code works without `device-number` @@ -193,38 +224,43 @@ class TestAliYunDatasource(test_helpers.HttprettyTestCase): LP: #1917875 """ netcfg = convert_ec2_metadata_network_config( - {"interfaces": {"macs": { - "06:17:04:d7:26:09": { - "interface-id": "eni-e44ef49e", - }, - "06:17:04:d7:26:08": { - "interface-id": "eni-e44ef49f", + { + "interfaces": { + "macs": { + "06:17:04:d7:26:09": { + "interface-id": "eni-e44ef49e", + }, + "06:17:04:d7:26:08": { + "interface-id": "eni-e44ef49f", + }, + } } - }}}, + }, macs_to_nics={ - '06:17:04:d7:26:09': 'eth0', - '06:17:04:d7:26:08': 'eth1', - } + "06:17:04:d7:26:09": "eth0", + "06:17:04:d7:26:08": "eth1", + }, ) - met0 = netcfg['ethernets']['eth0']['dhcp4-overrides']['route-metric'] - met1 = netcfg['ethernets']['eth1']['dhcp4-overrides']['route-metric'] + met0 = netcfg["ethernets"]["eth0"]["dhcp4-overrides"]["route-metric"] + met1 = netcfg["ethernets"]["eth1"]["dhcp4-overrides"]["route-metric"] # route-metric numbers should be 100 apart assert 100 == abs(met0 - met1) class TestIsAliYun(test_helpers.CiTestCase): - ALIYUN_PRODUCT = 'Alibaba Cloud ECS' - read_dmi_data_expected = [mock.call('system-product-name')] + ALIYUN_PRODUCT = "Alibaba Cloud ECS" + read_dmi_data_expected = [mock.call("system-product-name")] @mock.patch("cloudinit.sources.DataSourceAliYun.dmi.read_dmi_data") def test_true_on_aliyun_product(self, m_read_dmi_data): """Should return true if the dmi product data has expected value.""" m_read_dmi_data.return_value = self.ALIYUN_PRODUCT ret = ay._is_aliyun() - self.assertEqual(self.read_dmi_data_expected, - m_read_dmi_data.call_args_list) + self.assertEqual( + self.read_dmi_data_expected, m_read_dmi_data.call_args_list + ) self.assertEqual(True, ret) @mock.patch("cloudinit.sources.DataSourceAliYun.dmi.read_dmi_data") @@ -232,8 +268,9 @@ class TestIsAliYun(test_helpers.CiTestCase): """Should return false on empty value returned.""" m_read_dmi_data.return_value = "" ret = ay._is_aliyun() - self.assertEqual(self.read_dmi_data_expected, - m_read_dmi_data.call_args_list) + self.assertEqual( + self.read_dmi_data_expected, m_read_dmi_data.call_args_list + ) self.assertEqual(False, ret) @mock.patch("cloudinit.sources.DataSourceAliYun.dmi.read_dmi_data") @@ -241,8 +278,10 @@ class TestIsAliYun(test_helpers.CiTestCase): """Should return false on an unrelated string.""" m_read_dmi_data.return_value = "cubs win" ret = ay._is_aliyun() - self.assertEqual(self.read_dmi_data_expected, - m_read_dmi_data.call_args_list) + self.assertEqual( + self.read_dmi_data_expected, m_read_dmi_data.call_args_list + ) self.assertEqual(False, ret) + # vi: ts=4 expandtab |