diff options
Diffstat (limited to 'tests/unittests/sources/test_openstack.py')
| -rw-r--r-- | tests/unittests/sources/test_openstack.py | 788 |
1 files changed, 788 insertions, 0 deletions
diff --git a/tests/unittests/sources/test_openstack.py b/tests/unittests/sources/test_openstack.py new file mode 100644 index 00000000..c111bbcd --- /dev/null +++ b/tests/unittests/sources/test_openstack.py @@ -0,0 +1,788 @@ +# Copyright (C) 2014 Yahoo! Inc. +# +# Author: Joshua Harlow <harlowja@yahoo-inc.com> +# +# This file is part of cloud-init. See LICENSE file for license information. + +import copy +import json +import re +from io import StringIO +from urllib.parse import urlparse + +import httpretty as hp + +from cloudinit import helpers, settings, util +from cloudinit.sources import UNSET, BrokenMetadata +from cloudinit.sources import DataSourceOpenStack as ds +from cloudinit.sources import convert_vendordata +from cloudinit.sources.helpers import openstack +from tests.unittests import helpers as test_helpers + +BASE_URL = "http://169.254.169.254" +PUBKEY = "ssh-rsa AAAAB3NzaC1....sIkJhq8wdX+4I3A4cYbYP ubuntu@server-460\n" +EC2_META = { + "ami-id": "ami-00000001", + "ami-launch-index": "0", + "ami-manifest-path": "FIXME", + "hostname": "sm-foo-test.novalocal", + "instance-action": "none", + "instance-id": "i-00000001", + "instance-type": "m1.tiny", + "local-hostname": "sm-foo-test.novalocal", + "local-ipv4": "0.0.0.0", + "public-hostname": "sm-foo-test.novalocal", + "public-ipv4": "0.0.0.1", + "reservation-id": "r-iru5qm4m", +} +USER_DATA = b"#!/bin/sh\necho This is user data\n" +VENDOR_DATA = { + "magic": "", +} +VENDOR_DATA2 = {"static": {}} +OSTACK_META = { + "availability_zone": "nova", + "files": [ + {"content_path": "/content/0000", "path": "/etc/foo.cfg"}, + {"content_path": "/content/0001", "path": "/etc/bar/bar.cfg"}, + ], + "hostname": "sm-foo-test.novalocal", + "meta": {"dsmode": "local", "my-meta": "my-value"}, + "name": "sm-foo-test", + "public_keys": {"mykey": PUBKEY}, + "uuid": "b0fa911b-69d4-4476-bbe2-1c92bff6535c", +} +CONTENT_0 = b"This is contents of /etc/foo.cfg\n" +CONTENT_1 = b"# this is /etc/bar/bar.cfg\n" +OS_FILES = { + "openstack/content/0000": CONTENT_0, + "openstack/content/0001": CONTENT_1, + "openstack/latest/meta_data.json": json.dumps(OSTACK_META), + "openstack/latest/network_data.json": json.dumps( + {"links": [], "networks": [], "services": []} + ), + "openstack/latest/user_data": USER_DATA, + "openstack/latest/vendor_data.json": json.dumps(VENDOR_DATA), + "openstack/latest/vendor_data2.json": json.dumps(VENDOR_DATA2), +} +EC2_FILES = { + "latest/user-data": USER_DATA, +} +EC2_VERSIONS = [ + "latest", +] + +MOCK_PATH = "cloudinit.sources.DataSourceOpenStack." + + +# TODO _register_uris should leverage test_ec2.register_mock_metaserver. +def _register_uris(version, ec2_files, ec2_meta, os_files): + """Registers a set of url patterns into httpretty that will mimic the + same data returned by the openstack metadata service (and ec2 service).""" + + def match_ec2_url(uri, headers): + path = uri.path.strip("/") + if len(path) == 0: + return (200, headers, "\n".join(EC2_VERSIONS)) + path = uri.path.lstrip("/") + if path in ec2_files: + return (200, headers, ec2_files.get(path)) + if path == "latest/meta-data/": + buf = StringIO() + for (k, v) in ec2_meta.items(): + if isinstance(v, (list, tuple)): + buf.write("%s/" % (k)) + else: + buf.write("%s" % (k)) + buf.write("\n") + return (200, headers, buf.getvalue()) + if path.startswith("latest/meta-data/"): + value = None + pieces = path.split("/") + if path.endswith("/"): + pieces = pieces[2:-1] + value = util.get_cfg_by_path(ec2_meta, pieces) + else: + pieces = pieces[2:] + value = util.get_cfg_by_path(ec2_meta, pieces) + if value is not None: + return (200, headers, str(value)) + return (404, headers, "") + + def match_os_uri(uri, headers): + path = uri.path.strip("/") + if path == "openstack": + return (200, headers, "\n".join([openstack.OS_LATEST])) + path = uri.path.lstrip("/") + if path in os_files: + return (200, headers, os_files.get(path)) + return (404, headers, "") + + def get_request_callback(method, uri, headers): + uri = urlparse(uri) + path = uri.path.lstrip("/").split("/") + if path[0] == "openstack": + return match_os_uri(uri, headers) + return match_ec2_url(uri, headers) + + hp.register_uri( + hp.GET, + re.compile(r"http://169.254.169.254/.*"), + body=get_request_callback, + ) + + +def _read_metadata_service(): + return ds.read_metadata_service(BASE_URL, retries=0, timeout=0.1) + + +class TestOpenStackDataSource(test_helpers.HttprettyTestCase): + + with_logs = True + VERSION = "latest" + + def setUp(self): + super(TestOpenStackDataSource, self).setUp() + self.tmp = self.tmp_dir() + + def test_successful(self): + _register_uris(self.VERSION, EC2_FILES, EC2_META, OS_FILES) + f = _read_metadata_service() + self.assertEqual(VENDOR_DATA, f.get("vendordata")) + self.assertEqual(VENDOR_DATA2, f.get("vendordata2")) + self.assertEqual(CONTENT_0, f["files"]["/etc/foo.cfg"]) + self.assertEqual(CONTENT_1, f["files"]["/etc/bar/bar.cfg"]) + self.assertEqual(2, len(f["files"])) + self.assertEqual(USER_DATA, f.get("userdata")) + self.assertEqual(EC2_META, f.get("ec2-metadata")) + self.assertEqual(2, f.get("version")) + metadata = f["metadata"] + self.assertEqual("nova", metadata.get("availability_zone")) + self.assertEqual("sm-foo-test.novalocal", metadata.get("hostname")) + self.assertEqual( + "sm-foo-test.novalocal", metadata.get("local-hostname") + ) + self.assertEqual("sm-foo-test", metadata.get("name")) + self.assertEqual( + "b0fa911b-69d4-4476-bbe2-1c92bff6535c", metadata.get("uuid") + ) + self.assertEqual( + "b0fa911b-69d4-4476-bbe2-1c92bff6535c", metadata.get("instance-id") + ) + + def test_no_ec2(self): + _register_uris(self.VERSION, {}, {}, OS_FILES) + f = _read_metadata_service() + self.assertEqual(VENDOR_DATA, f.get("vendordata")) + self.assertEqual(VENDOR_DATA2, f.get("vendordata2")) + self.assertEqual(CONTENT_0, f["files"]["/etc/foo.cfg"]) + self.assertEqual(CONTENT_1, f["files"]["/etc/bar/bar.cfg"]) + self.assertEqual(USER_DATA, f.get("userdata")) + self.assertEqual({}, f.get("ec2-metadata")) + self.assertEqual(2, f.get("version")) + + def test_bad_metadata(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("meta_data.json"): + os_files.pop(k, None) + _register_uris(self.VERSION, {}, {}, os_files) + self.assertRaises(openstack.NonReadable, _read_metadata_service) + + def test_bad_uuid(self): + os_files = copy.deepcopy(OS_FILES) + os_meta = copy.deepcopy(OSTACK_META) + os_meta.pop("uuid") + for k in list(os_files.keys()): + if k.endswith("meta_data.json"): + os_files[k] = json.dumps(os_meta) + _register_uris(self.VERSION, {}, {}, os_files) + self.assertRaises(BrokenMetadata, _read_metadata_service) + + def test_userdata_empty(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("user_data"): + os_files.pop(k, None) + _register_uris(self.VERSION, {}, {}, os_files) + f = _read_metadata_service() + self.assertEqual(VENDOR_DATA, f.get("vendordata")) + self.assertEqual(VENDOR_DATA2, f.get("vendordata2")) + self.assertEqual(CONTENT_0, f["files"]["/etc/foo.cfg"]) + self.assertEqual(CONTENT_1, f["files"]["/etc/bar/bar.cfg"]) + self.assertFalse(f.get("userdata")) + + def test_vendordata_empty(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("vendor_data.json"): + os_files.pop(k, None) + _register_uris(self.VERSION, {}, {}, os_files) + f = _read_metadata_service() + self.assertEqual(CONTENT_0, f["files"]["/etc/foo.cfg"]) + self.assertEqual(CONTENT_1, f["files"]["/etc/bar/bar.cfg"]) + self.assertFalse(f.get("vendordata")) + + def test_vendordata2_empty(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("vendor_data2.json"): + os_files.pop(k, None) + _register_uris(self.VERSION, {}, {}, os_files) + f = _read_metadata_service() + self.assertEqual(CONTENT_0, f["files"]["/etc/foo.cfg"]) + self.assertEqual(CONTENT_1, f["files"]["/etc/bar/bar.cfg"]) + self.assertFalse(f.get("vendordata2")) + + def test_vendordata_invalid(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("vendor_data.json"): + os_files[k] = "{" # some invalid json + _register_uris(self.VERSION, {}, {}, os_files) + self.assertRaises(BrokenMetadata, _read_metadata_service) + + def test_vendordata2_invalid(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("vendor_data2.json"): + os_files[k] = "{" # some invalid json + _register_uris(self.VERSION, {}, {}, os_files) + self.assertRaises(BrokenMetadata, _read_metadata_service) + + def test_metadata_invalid(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("meta_data.json"): + os_files[k] = "{" # some invalid json + _register_uris(self.VERSION, {}, {}, os_files) + self.assertRaises(BrokenMetadata, _read_metadata_service) + + @test_helpers.mock.patch("cloudinit.net.dhcp.maybe_perform_dhcp_discovery") + def test_datasource(self, m_dhcp): + _register_uris(self.VERSION, EC2_FILES, EC2_META, OS_FILES) + ds_os = ds.DataSourceOpenStack( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + self.assertIsNone(ds_os.version) + mock_path = MOCK_PATH + "detect_openstack" + with test_helpers.mock.patch(mock_path) as m_detect_os: + m_detect_os.return_value = True + found = ds_os.get_data() + self.assertTrue(found) + self.assertEqual(2, ds_os.version) + md = dict(ds_os.metadata) + md.pop("instance-id", None) + md.pop("local-hostname", None) + self.assertEqual(OSTACK_META, md) + self.assertEqual(EC2_META, ds_os.ec2_metadata) + self.assertEqual(USER_DATA, ds_os.userdata_raw) + self.assertEqual(2, len(ds_os.files)) + self.assertEqual(VENDOR_DATA, ds_os.vendordata_pure) + self.assertEqual(VENDOR_DATA2, ds_os.vendordata2_pure) + self.assertIsNone(ds_os.vendordata_raw) + m_dhcp.assert_not_called() + + @hp.activate + @test_helpers.mock.patch("cloudinit.net.dhcp.EphemeralIPv4Network") + @test_helpers.mock.patch("cloudinit.net.dhcp.maybe_perform_dhcp_discovery") + def test_local_datasource(self, m_dhcp, m_net): + """OpenStackLocal calls EphemeralDHCPNetwork and gets instance data.""" + _register_uris(self.VERSION, EC2_FILES, EC2_META, OS_FILES) + ds_os_local = ds.DataSourceOpenStackLocal( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + ds_os_local._fallback_interface = "eth9" # Monkey patch for dhcp + m_dhcp.return_value = [ + { + "interface": "eth9", + "fixed-address": "192.168.2.9", + "routers": "192.168.2.1", + "subnet-mask": "255.255.255.0", + "broadcast-address": "192.168.2.255", + } + ] + + self.assertIsNone(ds_os_local.version) + mock_path = MOCK_PATH + "detect_openstack" + with test_helpers.mock.patch(mock_path) as m_detect_os: + m_detect_os.return_value = True + found = ds_os_local.get_data() + self.assertTrue(found) + self.assertEqual(2, ds_os_local.version) + md = dict(ds_os_local.metadata) + md.pop("instance-id", None) + md.pop("local-hostname", None) + self.assertEqual(OSTACK_META, md) + self.assertEqual(EC2_META, ds_os_local.ec2_metadata) + self.assertEqual(USER_DATA, ds_os_local.userdata_raw) + self.assertEqual(2, len(ds_os_local.files)) + self.assertEqual(VENDOR_DATA, ds_os_local.vendordata_pure) + self.assertEqual(VENDOR_DATA2, ds_os_local.vendordata2_pure) + self.assertIsNone(ds_os_local.vendordata_raw) + m_dhcp.assert_called_with("eth9", None) + + def test_bad_datasource_meta(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("meta_data.json"): + os_files[k] = "{" # some invalid json + _register_uris(self.VERSION, {}, {}, os_files) + ds_os = ds.DataSourceOpenStack( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + self.assertIsNone(ds_os.version) + mock_path = MOCK_PATH + "detect_openstack" + with test_helpers.mock.patch(mock_path) as m_detect_os: + m_detect_os.return_value = True + found = ds_os.get_data() + self.assertFalse(found) + self.assertIsNone(ds_os.version) + self.assertIn( + "InvalidMetaDataException: Broken metadata address" + " http://169.254.169.25", + self.logs.getvalue(), + ) + + def test_no_datasource(self): + os_files = copy.deepcopy(OS_FILES) + for k in list(os_files.keys()): + if k.endswith("meta_data.json"): + os_files.pop(k) + _register_uris(self.VERSION, {}, {}, os_files) + ds_os = ds.DataSourceOpenStack( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + ds_os.ds_cfg = { + "max_wait": 0, + "timeout": 0, + } + self.assertIsNone(ds_os.version) + mock_path = MOCK_PATH + "detect_openstack" + with test_helpers.mock.patch(mock_path) as m_detect_os: + m_detect_os.return_value = True + found = ds_os.get_data() + self.assertFalse(found) + self.assertIsNone(ds_os.version) + + def test_network_config_disabled_by_datasource_config(self): + """The network_config can be disabled from datasource config.""" + mock_path = MOCK_PATH + "openstack.convert_net_json" + ds_os = ds.DataSourceOpenStack( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + ds_os.ds_cfg = {"apply_network_config": False} + sample_json = { + "links": [{"ethernet_mac_address": "mymac"}], + "networks": [], + "services": [], + } + ds_os.network_json = sample_json # Ignore this content from metadata + with test_helpers.mock.patch(mock_path) as m_convert_json: + self.assertIsNone(ds_os.network_config) + m_convert_json.assert_not_called() + + def test_network_config_from_network_json(self): + """The datasource gets network_config from network_data.json.""" + mock_path = MOCK_PATH + "openstack.convert_net_json" + example_cfg = {"version": 1, "config": []} + ds_os = ds.DataSourceOpenStack( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + sample_json = { + "links": [{"ethernet_mac_address": "mymac"}], + "networks": [], + "services": [], + } + ds_os.network_json = sample_json + with test_helpers.mock.patch(mock_path) as m_convert_json: + m_convert_json.return_value = example_cfg + self.assertEqual(example_cfg, ds_os.network_config) + self.assertIn( + "network config provided via network_json", self.logs.getvalue() + ) + m_convert_json.assert_called_with(sample_json, known_macs=None) + + def test_network_config_cached(self): + """The datasource caches the network_config property.""" + mock_path = MOCK_PATH + "openstack.convert_net_json" + example_cfg = {"version": 1, "config": []} + ds_os = ds.DataSourceOpenStack( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + ds_os._network_config = example_cfg + with test_helpers.mock.patch(mock_path) as m_convert_json: + self.assertEqual(example_cfg, ds_os.network_config) + m_convert_json.assert_not_called() + + def test_disabled_datasource(self): + os_files = copy.deepcopy(OS_FILES) + os_meta = copy.deepcopy(OSTACK_META) + os_meta["meta"] = { + "dsmode": "disabled", + } + for k in list(os_files.keys()): + if k.endswith("meta_data.json"): + os_files[k] = json.dumps(os_meta) + _register_uris(self.VERSION, {}, {}, os_files) + ds_os = ds.DataSourceOpenStack( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + ds_os.ds_cfg = { + "max_wait": 0, + "timeout": 0, + } + self.assertIsNone(ds_os.version) + mock_path = MOCK_PATH + "detect_openstack" + with test_helpers.mock.patch(mock_path) as m_detect_os: + m_detect_os.return_value = True + found = ds_os.get_data() + self.assertFalse(found) + self.assertIsNone(ds_os.version) + + @hp.activate + def test_wb__crawl_metadata_does_not_persist(self): + """_crawl_metadata returns current metadata and does not cache.""" + _register_uris(self.VERSION, EC2_FILES, EC2_META, OS_FILES) + ds_os = ds.DataSourceOpenStack( + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) + ) + crawled_data = ds_os._crawl_metadata() + self.assertEqual(UNSET, ds_os.ec2_metadata) + self.assertIsNone(ds_os.userdata_raw) + self.assertEqual(0, len(ds_os.files)) + self.assertIsNone(ds_os.vendordata_raw) + self.assertEqual( + [ + "dsmode", + "ec2-metadata", + "files", + "metadata", + "networkdata", + "userdata", + "vendordata", + "vendordata2", + "version", + ], + sorted(crawled_data.keys()), + ) + self.assertEqual("local", crawled_data["dsmode"]) + self.assertEqual(EC2_META, crawled_data["ec2-metadata"]) + self.assertEqual(2, len(crawled_data["files"])) + md = copy.deepcopy(crawled_data["metadata"]) + md.pop("instance-id") + md.pop("local-hostname") + self.assertEqual(OSTACK_META, md) + self.assertEqual( + json.loads(OS_FILES["openstack/latest/network_data.json"]), + crawled_data["networkdata"], + ) + self.assertEqual(USER_DATA, crawled_data["userdata"]) + self.assertEqual(VENDOR_DATA, crawled_data["vendordata"]) + self.assertEqual(VENDOR_DATA2, crawled_data["vendordata2"]) + self.assertEqual(2, crawled_data["version"]) + + +class TestVendorDataLoading(test_helpers.TestCase): + def cvj(self, data): + return convert_vendordata(data) + + def test_vd_load_none(self): + # non-existant vendor-data should return none + self.assertIsNone(self.cvj(None)) + + def test_vd_load_string(self): + self.assertEqual(self.cvj("foobar"), "foobar") + + def test_vd_load_list(self): + data = [{"foo": "bar"}, "mystring", list(["another", "list"])] + self.assertEqual(self.cvj(data), data) + + def test_vd_load_dict_no_ci(self): + self.assertIsNone(self.cvj({"foo": "bar"})) + + def test_vd_load_dict_ci_dict(self): + self.assertRaises( + ValueError, self.cvj, {"foo": "bar", "cloud-init": {"x": 1}} + ) + + def test_vd_load_dict_ci_string(self): + data = {"foo": "bar", "cloud-init": "VENDOR_DATA"} + self.assertEqual(self.cvj(data), data["cloud-init"]) + + def test_vd_load_dict_ci_list(self): + data = {"foo": "bar", "cloud-init": ["VD_1", "VD_2"]} + self.assertEqual(self.cvj(data), data["cloud-init"]) + + +@test_helpers.mock.patch(MOCK_PATH + "util.is_x86") +class TestDetectOpenStack(test_helpers.CiTestCase): + def test_detect_openstack_non_intel_x86(self, m_is_x86): + """Return True on non-intel platforms because dmi isn't conclusive.""" + m_is_x86.return_value = False + self.assertTrue( + ds.detect_openstack(), "Expected detect_openstack == True" + ) + + @test_helpers.mock.patch(MOCK_PATH + "util.get_proc_env") + @test_helpers.mock.patch(MOCK_PATH + "dmi.read_dmi_data") + def test_not_detect_openstack_intel_x86_ec2( + self, m_dmi, m_proc_env, m_is_x86 + ): + """Return False on EC2 platforms.""" + m_is_x86.return_value = True + # No product_name in proc/1/environ + m_proc_env.return_value = {"HOME": "/"} + + def fake_dmi_read(dmi_key): + if dmi_key == "system-product-name": + return "HVM domU" # Nothing 'openstackish' on EC2 + if dmi_key == "chassis-asset-tag": + return "" # Empty string on EC2 + assert False, "Unexpected dmi read of %s" % dmi_key + + m_dmi.side_effect = fake_dmi_read + self.assertFalse( + ds.detect_openstack(), "Expected detect_openstack == False on EC2" + ) + m_proc_env.assert_called_with(1) + + @test_helpers.mock.patch(MOCK_PATH + "dmi.read_dmi_data") + def test_detect_openstack_intel_product_name_compute( + self, m_dmi, m_is_x86 + ): + """Return True on OpenStack compute and nova instances.""" + m_is_x86.return_value = True + openstack_product_names = ["OpenStack Nova", "OpenStack Compute"] + + for product_name in openstack_product_names: + m_dmi.return_value = product_name + self.assertTrue( + ds.detect_openstack(), "Failed to detect_openstack" + ) + + @test_helpers.mock.patch(MOCK_PATH + "dmi.read_dmi_data") + def test_detect_openstack_opentelekomcloud_chassis_asset_tag( + self, m_dmi, m_is_x86 + ): + """Return True on OpenStack reporting OpenTelekomCloud asset-tag.""" + m_is_x86.return_value = True + + def fake_dmi_read(dmi_key): + if dmi_key == "system-product-name": + return "HVM domU" # Nothing 'openstackish' on OpenTelekomCloud + if dmi_key == "chassis-asset-tag": + return "OpenTelekomCloud" + assert False, "Unexpected dmi read of %s" % dmi_key + + m_dmi.side_effect = fake_dmi_read + self.assertTrue( + ds.detect_openstack(), + "Expected detect_openstack == True on OpenTelekomCloud", + ) + + @test_helpers.mock.patch(MOCK_PATH + "dmi.read_dmi_data") + def test_detect_openstack_sapccloud_chassis_asset_tag( + self, m_dmi, m_is_x86 + ): + """Return True on OpenStack reporting SAP CCloud VM asset-tag.""" + m_is_x86.return_value = True + + def fake_dmi_read(dmi_key): + if dmi_key == "system-product-name": + return "VMware Virtual Platform" # SAP CCloud uses VMware + if dmi_key == "chassis-asset-tag": + return "SAP CCloud VM" + assert False, "Unexpected dmi read of %s" % dmi_key + + m_dmi.side_effect = fake_dmi_read + self.assertTrue( + ds.detect_openstack(), + "Expected detect_openstack == True on SAP CCloud VM", + ) + + @test_helpers.mock.patch(MOCK_PATH + "dmi.read_dmi_data") + def test_detect_openstack_oraclecloud_chassis_asset_tag( + self, m_dmi, m_is_x86 + ): + """Return True on OpenStack reporting Oracle cloud asset-tag.""" + m_is_x86.return_value = True + + def fake_dmi_read(dmi_key): + if dmi_key == "system-product-name": + return "Standard PC (i440FX + PIIX, 1996)" # No match + if dmi_key == "chassis-asset-tag": + return "OracleCloud.com" + assert False, "Unexpected dmi read of %s" % dmi_key + + m_dmi.side_effect = fake_dmi_read + self.assertTrue( + ds.detect_openstack(accept_oracle=True), + "Expected detect_openstack == True on OracleCloud.com", + ) + self.assertFalse( + ds.detect_openstack(accept_oracle=False), + "Expected detect_openstack == False.", + ) + + def _test_detect_openstack_nova_compute_chassis_asset_tag( + self, m_dmi, m_is_x86, chassis_tag + ): + """Return True on OpenStack reporting generic asset-tag.""" + m_is_x86.return_value = True + + def fake_dmi_read(dmi_key): + if dmi_key == "system-product-name": + return "Generic OpenStack Platform" + if dmi_key == "chassis-asset-tag": + return chassis_tag + assert False, "Unexpected dmi read of %s" % dmi_key + + m_dmi.side_effect = fake_dmi_read + self.assertTrue( + ds.detect_openstack(), + "Expected detect_openstack == True on Generic OpenStack Platform", + ) + + @test_helpers.mock.patch(MOCK_PATH + "dmi.read_dmi_data") + def test_detect_openstack_nova_chassis_asset_tag(self, m_dmi, m_is_x86): + self._test_detect_openstack_nova_compute_chassis_asset_tag( + m_dmi, m_is_x86, "OpenStack Nova" + ) + + @test_helpers.mock.patch(MOCK_PATH + "dmi.read_dmi_data") + def test_detect_openstack_compute_chassis_asset_tag(self, m_dmi, m_is_x86): + self._test_detect_openstack_nova_compute_chassis_asset_tag( + m_dmi, m_is_x86, "OpenStack Compute" + ) + + @test_helpers.mock.patch(MOCK_PATH + "util.get_proc_env") + @test_helpers.mock.patch(MOCK_PATH + "dmi.read_dmi_data") + def test_detect_openstack_by_proc_1_environ( + self, m_dmi, m_proc_env, m_is_x86 + ): + """Return True when nova product_name specified in /proc/1/environ.""" + m_is_x86.return_value = True + # Nova product_name in proc/1/environ + m_proc_env.return_value = { + "HOME": "/", + "product_name": "OpenStack Nova", + } + + def fake_dmi_read(dmi_key): + if dmi_key == "system-product-name": + return "HVM domU" # Nothing 'openstackish' + if dmi_key == "chassis-asset-tag": + return "" # Nothin 'openstackish' + assert False, "Unexpected dmi read of %s" % dmi_key + + m_dmi.side_effect = fake_dmi_read + self.assertTrue( + ds.detect_openstack(), + "Expected detect_openstack == True on OpenTelekomCloud", + ) + m_proc_env.assert_called_with(1) + + +class TestMetadataReader(test_helpers.HttprettyTestCase): + """Test the MetadataReader.""" + + burl = "http://169.254.169.254/" + md_base = { + "availability_zone": "myaz1", + "hostname": "sm-foo-test.novalocal", + "keys": [{"data": PUBKEY, "name": "brickies", "type": "ssh"}], + "launch_index": 0, + "name": "sm-foo-test", + "public_keys": {"mykey": PUBKEY}, + "project_id": "6a103f813b774b9fb15a4fcd36e1c056", + "uuid": "b0fa911b-69d4-4476-bbe2-1c92bff6535c", + } + + def register(self, path, body=None, status=200): + content = body if not isinstance(body, str) else body.encode("utf-8") + hp.register_uri( + hp.GET, self.burl + "openstack" + path, status=status, body=content + ) + + def register_versions(self, versions): + self.register("", "\n".join(versions)) + self.register("/", "\n".join(versions)) + + def register_version(self, version, data): + content = "\n".join(sorted(data.keys())) + self.register(version, content) + self.register(version + "/", content) + for path, content in data.items(): + self.register("/%s/%s" % (version, path), content) + self.register("/%s/%s" % (version, path), content) + if "user_data" not in data: + self.register("/%s/user_data" % version, "nodata", status=404) + + def test__find_working_version(self): + """Test a working version ignores unsupported.""" + unsup = "2016-11-09" + self.register_versions( + [ + openstack.OS_FOLSOM, + openstack.OS_LIBERTY, + unsup, + openstack.OS_LATEST, + ] + ) + self.assertEqual( + openstack.OS_LIBERTY, + openstack.MetadataReader(self.burl)._find_working_version(), + ) + + def test__find_working_version_uses_latest(self): + """'latest' should be used if no supported versions.""" + unsup1, unsup2 = ("2016-11-09", "2017-06-06") + self.register_versions([unsup1, unsup2, openstack.OS_LATEST]) + self.assertEqual( + openstack.OS_LATEST, + openstack.MetadataReader(self.burl)._find_working_version(), + ) + + def test_read_v2_os_ocata(self): + """Validate return value of read_v2 for os_ocata data.""" + md = copy.deepcopy(self.md_base) + md["devices"] = [] + network_data = {"links": [], "networks": [], "services": []} + vendor_data = {} + vendor_data2 = {"static": {}} + + data = { + "meta_data.json": json.dumps(md), + "network_data.json": json.dumps(network_data), + "vendor_data.json": json.dumps(vendor_data), + "vendor_data2.json": json.dumps(vendor_data2), + } + + self.register_versions([openstack.OS_OCATA, openstack.OS_LATEST]) + self.register_version(openstack.OS_OCATA, data) + + mock_read_ec2 = test_helpers.mock.MagicMock( + return_value={"instance-id": "unused-ec2"} + ) + expected_md = copy.deepcopy(md) + expected_md.update( + {"instance-id": md["uuid"], "local-hostname": md["hostname"]} + ) + expected = { + "userdata": "", # Annoying, no user-data results in empty string. + "version": 2, + "metadata": expected_md, + "vendordata": vendor_data, + "vendordata2": vendor_data2, + "networkdata": network_data, + "ec2-metadata": mock_read_ec2.return_value, + "files": {}, + } + reader = openstack.MetadataReader(self.burl) + reader._read_ec2_metadata = mock_read_ec2 + self.assertEqual(expected, reader.read_v2()) + self.assertEqual(1, mock_read_ec2.call_count) + + +# vi: ts=4 expandtab |