# Copyright (C) 2014 Yahoo! Inc. # # Author: Joshua Harlow # # 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