summaryrefslogtreecommitdiff
path: root/tests/unittests/sources/test_aliyun.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests/sources/test_aliyun.py')
-rw-r--r--tests/unittests/sources/test_aliyun.py248
1 files changed, 248 insertions, 0 deletions
diff --git a/tests/unittests/sources/test_aliyun.py b/tests/unittests/sources/test_aliyun.py
new file mode 100644
index 00000000..00209913
--- /dev/null
+++ b/tests/unittests/sources/test_aliyun.py
@@ -0,0 +1,248 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+import functools
+import httpretty
+import os
+from unittest import mock
+
+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...'}}
+}
+
+DEFAULT_USERDATA = """\
+#cloud-config
+
+hostname: localhost"""
+
+
+def register_mock_metaserver(base_url, data):
+ def register_helper(register, base_url, body):
+ if isinstance(body, str):
+ register(base_url, body)
+ elif isinstance(body, list):
+ register(base_url.rstrip('/'), '\n'.join(body) + '\n')
+ elif isinstance(body, dict):
+ if not body:
+ register(base_url.rstrip('/') + '/', 'not found',
+ status_code=404)
+ vals = []
+ for k, v in body.items():
+ if isinstance(v, (str, list)):
+ suffix = k.rstrip('/')
+ else:
+ suffix = k.rstrip('/') + '/'
+ vals.append(suffix)
+ url = base_url.rstrip('/') + '/' + suffix
+ register_helper(register, url, v)
+ register(base_url, '\n'.join(vals) + '\n')
+
+ register = functools.partial(httpretty.register_uri, httpretty.GET)
+ register_helper(register, base_url, data)
+
+
+class TestAliYunDatasource(test_helpers.HttprettyTestCase):
+ def setUp(self):
+ super(TestAliYunDatasource, self).setUp()
+ cfg = {'datasource': {'AliYun': {'timeout': '1', 'max_wait': '1'}}}
+ distro = {}
+ paths = helpers.Paths({'run_dir': self.tmp_dir()})
+ self.ds = ay.DataSourceAliYun(cfg, distro, paths)
+ self.metadata_address = self.ds.metadata_urls[0]
+
+ @property
+ def default_metadata(self):
+ return DEFAULT_METADATA
+
+ @property
+ def default_userdata(self):
+ return DEFAULT_USERDATA
+
+ @property
+ def metadata_url(self):
+ 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')
+
+ # EC2 provides an instance-identity document which must return 404 here
+ # for this test to pass.
+ @property
+ def default_identity(self):
+ return {}
+
+ @property
+ def identity_url(self):
+ 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)
+ register_mock_metaserver(self.userdata_url, self.default_userdata)
+ register_mock_metaserver(self.identity_url, self.default_identity)
+
+ def _test_get_data(self):
+ self.assertEqual(self.ds.metadata, self.default_metadata)
+ 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()]
+ 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())
+
+ def _test_host_name(self):
+ 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):
+ m_is_aliyun.return_value = True
+ self.regist_default_server()
+ ret = self.ds.get_data()
+ self.assertEqual(True, ret)
+ self.assertEqual(1, m_is_aliyun.call_count)
+ self._test_get_data()
+ 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(
+ '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):
+ """If is_aliyun returns false, then get_data should return False."""
+ m_is_aliyun.return_value = False
+ self.regist_default_server()
+ ret = self.ds.get_data()
+ self.assertEqual(1, m_is_aliyun.call_count)
+ self.assertEqual(False, ret)
+
+ def test_parse_public_keys(self):
+ 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', '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': {'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',
+ '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`
+
+ `device-number` is part of EC2 metadata, but not supported on aliyun.
+ Attempting to access it will raise a KeyError.
+
+ 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",
+ }
+ }}},
+ macs_to_nics={
+ '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']
+
+ # 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')]
+
+ @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(True, ret)
+
+ @mock.patch("cloudinit.sources.DataSourceAliYun.dmi.read_dmi_data")
+ def test_false_on_empty_string(self, m_read_dmi_data):
+ """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(False, ret)
+
+ @mock.patch("cloudinit.sources.DataSourceAliYun.dmi.read_dmi_data")
+ def test_false_on_unknown_string(self, m_read_dmi_data):
+ """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(False, ret)
+
+# vi: ts=4 expandtab