summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
authorScott Moser <smoser@ubuntu.com>2014-02-13 21:18:17 -0500
committerScott Moser <smoser@ubuntu.com>2014-02-13 21:18:17 -0500
commit4ba72556193219f90c313f62d0d309761bb53c6b (patch)
tree2ad80e184d53a85a1fe5933ad106249bc522da35 /tests
parent507aeed12312af7fc8a9fcfca6b845183a5a3c51 (diff)
parent053667688d7c2ad51e569c62e00dac1942e46f62 (diff)
downloadvyos-cloud-init-4ba72556193219f90c313f62d0d309761bb53c6b.tar.gz
vyos-cloud-init-4ba72556193219f90c313f62d0d309761bb53c6b.zip
merge from trunk
Diffstat (limited to 'tests')
-rw-r--r--tests/unittests/helpers.py25
-rw-r--r--tests/unittests/test_datasource/test_configdrive.py35
-rw-r--r--tests/unittests/test_datasource/test_gce.py111
-rw-r--r--tests/unittests/test_datasource/test_openstack.py301
-rw-r--r--tests/unittests/test_ec2_util.py35
5 files changed, 473 insertions, 34 deletions
diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py
index 5b4f4208..5bed13cc 100644
--- a/tests/unittests/helpers.py
+++ b/tests/unittests/helpers.py
@@ -12,10 +12,27 @@ from cloudinit import util
import shutil
-# Handle how 2.6 doesn't have the assertIn or assertNotIn
+# Used for detecting different python versions
+PY2 = False
+PY26 = False
+PY27 = False
+PY3 = False
+
_PY_VER = sys.version_info
_PY_MAJOR, _PY_MINOR = _PY_VER[0:2]
if (_PY_MAJOR, _PY_MINOR) <= (2, 6):
+ if (_PY_MAJOR, _PY_MINOR) == (2, 6):
+ PY26 = True
+ if (_PY_MAJOR, _PY_MINOR) >= (2, 0):
+ PY2 = True
+else:
+ if (_PY_MAJOR, _PY_MINOR) == (2, 7):
+ PY27 = True
+ PY2 = True
+ if (_PY_MAJOR, _PY_MINOR) >= (3, 0):
+ PY3 = True
+
+if PY26:
# For now add these on, taken from python 2.7 + slightly adjusted
class TestCase(unittest.TestCase):
def assertIn(self, member, container, msg=None):
@@ -29,6 +46,12 @@ if (_PY_MAJOR, _PY_MINOR) <= (2, 6):
standardMsg = standardMsg % (member, container)
self.fail(self._formatMessage(msg, standardMsg))
+ def assertIsNone(self, value, msg=None):
+ if value is not None:
+ standardMsg = '%r is not None'
+ standardMsg = standardMsg % (value)
+ self.fail(self._formatMessage(msg, standardMsg))
+
else:
class TestCase(unittest.TestCase):
pass
diff --git a/tests/unittests/test_datasource/test_configdrive.py b/tests/unittests/test_datasource/test_configdrive.py
index 1f4a0a0b..937b88c1 100644
--- a/tests/unittests/test_datasource/test_configdrive.py
+++ b/tests/unittests/test_datasource/test_configdrive.py
@@ -9,6 +9,7 @@ from mocker import MockerTestCase
from cloudinit import helpers
from cloudinit import settings
from cloudinit.sources import DataSourceConfigDrive as ds
+from cloudinit.sources.helpers import openstack
from cloudinit import util
from tests.unittests import helpers as unit_helpers
@@ -71,7 +72,7 @@ class TestConfigDriveDataSource(MockerTestCase):
def test_ec2_metadata(self):
populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
- found = ds.read_config_drive_dir(self.tmp)
+ found = ds.read_config_drive(self.tmp)
self.assertTrue('ec2-metadata' in found)
ec2_md = found['ec2-metadata']
self.assertEqual(EC2_META, ec2_md)
@@ -81,7 +82,7 @@ class TestConfigDriveDataSource(MockerTestCase):
cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN,
None,
helpers.Paths({}))
- found = ds.read_config_drive_dir(self.tmp)
+ found = ds.read_config_drive(self.tmp)
cfg_ds.metadata = found['metadata']
name_tests = {
'ami': '/dev/vda1',
@@ -112,7 +113,7 @@ class TestConfigDriveDataSource(MockerTestCase):
cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN,
None,
helpers.Paths({}))
- found = ds.read_config_drive_dir(self.tmp)
+ found = ds.read_config_drive(self.tmp)
os_md = found['metadata']
cfg_ds.metadata = os_md
name_tests = {
@@ -140,7 +141,7 @@ class TestConfigDriveDataSource(MockerTestCase):
cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN,
None,
helpers.Paths({}))
- found = ds.read_config_drive_dir(self.tmp)
+ found = ds.read_config_drive(self.tmp)
ec2_md = found['ec2-metadata']
os_md = found['metadata']
cfg_ds.ec2_metadata = ec2_md
@@ -165,13 +166,13 @@ class TestConfigDriveDataSource(MockerTestCase):
my_mock.replay()
device = cfg_ds.device_name_to_device(name)
self.assertEquals(dev_name, device)
-
+
def test_dev_ec2_map(self):
populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN,
None,
helpers.Paths({}))
- found = ds.read_config_drive_dir(self.tmp)
+ found = ds.read_config_drive(self.tmp)
exists_mock = self.mocker.replace(os.path.exists,
spec=False, passthrough=False)
exists_mock(mocker.ARGS)
@@ -200,10 +201,11 @@ class TestConfigDriveDataSource(MockerTestCase):
populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
- found = ds.read_config_drive_dir(self.tmp)
+ found = ds.read_config_drive(self.tmp)
expected_md = copy(OSTACK_META)
expected_md['instance-id'] = expected_md['uuid']
+ expected_md['local-hostname'] = expected_md['hostname']
self.assertEqual(USER_DATA, found['userdata'])
self.assertEqual(expected_md, found['metadata'])
@@ -219,10 +221,11 @@ class TestConfigDriveDataSource(MockerTestCase):
populate_dir(self.tmp, data)
- found = ds.read_config_drive_dir(self.tmp)
+ found = ds.read_config_drive(self.tmp)
expected_md = copy(OSTACK_META)
expected_md['instance-id'] = expected_md['uuid']
+ expected_md['local-hostname'] = expected_md['hostname']
self.assertEqual(expected_md, found['metadata'])
@@ -235,8 +238,8 @@ class TestConfigDriveDataSource(MockerTestCase):
populate_dir(self.tmp, data)
- self.assertRaises(ds.BrokenConfigDriveDir,
- ds.read_config_drive_dir, self.tmp)
+ self.assertRaises(openstack.BrokenMetadata,
+ ds.read_config_drive, self.tmp)
def test_seed_dir_no_configdrive(self):
"""Verify that no metadata raises NonConfigDriveDir."""
@@ -247,14 +250,14 @@ class TestConfigDriveDataSource(MockerTestCase):
data["openstack/latest/random-file.txt"] = "random-content"
data["content/foo"] = "foocontent"
- self.assertRaises(ds.NonConfigDriveDir,
- ds.read_config_drive_dir, my_d)
+ self.assertRaises(openstack.NonReadable,
+ ds.read_config_drive, my_d)
def test_seed_dir_missing(self):
"""Verify that missing seed_dir raises NonConfigDriveDir."""
my_d = os.path.join(self.tmp, "nonexistantdirectory")
- self.assertRaises(ds.NonConfigDriveDir,
- ds.read_config_drive_dir, my_d)
+ self.assertRaises(openstack.NonReadable,
+ ds.read_config_drive, my_d)
def test_find_candidates(self):
devs_with_answers = {}
@@ -304,7 +307,7 @@ class TestConfigDriveDataSource(MockerTestCase):
def cfg_ds_from_dir(seed_d):
- found = ds.read_config_drive_dir(seed_d)
+ found = ds.read_config_drive(seed_d)
cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN, None,
helpers.Paths({}))
populate_ds_from_read_config(cfg_ds, seed_d, found)
@@ -319,7 +322,7 @@ def populate_ds_from_read_config(cfg_ds, source, results):
cfg_ds.metadata = results.get('metadata')
cfg_ds.ec2_metadata = results.get('ec2-metadata')
cfg_ds.userdata_raw = results.get('userdata')
- cfg_ds.version = results.get('cfgdrive_ver')
+ cfg_ds.version = results.get('version')
def populate_dir(seed_dir, files):
diff --git a/tests/unittests/test_datasource/test_gce.py b/tests/unittests/test_datasource/test_gce.py
new file mode 100644
index 00000000..d91bd531
--- /dev/null
+++ b/tests/unittests/test_datasource/test_gce.py
@@ -0,0 +1,111 @@
+#
+# Copyright (C) 2014 Vaidas Jablonskis
+#
+# Author: Vaidas Jablonskis <jablonskis@gmail.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import unittest
+import httpretty
+import re
+
+from urlparse import urlparse
+
+from cloudinit import settings
+from cloudinit import helpers
+from cloudinit.sources import DataSourceGCE
+
+GCE_META = {
+ 'instance/id': '123',
+ 'instance/zone': 'foo/bar',
+ 'project/attributes/sshKeys': 'user:ssh-rsa AA2..+aRD0fyVw== root@server',
+ 'instance/hostname': 'server.project-name.local',
+ 'instance/attributes/user-data': '/bin/echo foo\n',
+}
+
+GCE_META_PARTIAL = {
+ 'instance/id': '123',
+ 'instance/hostname': 'server.project-name.local',
+}
+
+HEADERS = {'X-Google-Metadata-Request': 'True'}
+MD_URL_RE = re.compile(r'http://metadata.google.internal./computeMetadata/v1/.*')
+
+
+def _request_callback(method, uri, headers):
+ url_path = urlparse(uri).path
+ if url_path.startswith('/computeMetadata/v1/'):
+ path = url_path.split('/computeMetadata/v1/')[1:][0]
+ else:
+ path = None
+ if path in GCE_META:
+ return (200, headers, GCE_META.get(path))
+ else:
+ return (404, headers, '')
+
+
+class TestDataSourceGCE(unittest.TestCase):
+
+ def setUp(self):
+ self.ds = DataSourceGCE.DataSourceGCE(
+ settings.CFG_BUILTIN, None,
+ helpers.Paths({}))
+
+ @httpretty.activate
+ def test_connection(self):
+ httpretty.register_uri(
+ httpretty.GET, MD_URL_RE,
+ body=_request_callback)
+
+ success = self.ds.get_data()
+ self.assertTrue(success)
+
+ req_header = httpretty.last_request().headers
+ self.assertDictContainsSubset(HEADERS, req_header)
+
+ @httpretty.activate
+ def test_metadata(self):
+ httpretty.register_uri(
+ httpretty.GET, MD_URL_RE,
+ body=_request_callback)
+ self.ds.get_data()
+
+ self.assertEqual(GCE_META.get('instance/hostname'),
+ self.ds.get_hostname())
+
+ self.assertEqual(GCE_META.get('instance/id'),
+ self.ds.get_instance_id())
+
+ self.assertEqual(GCE_META.get('instance/zone'),
+ self.ds.availability_zone)
+
+ self.assertEqual(GCE_META.get('instance/attributes/user-data'),
+ self.ds.get_userdata_raw())
+
+ # we expect a list of public ssh keys with user names stripped
+ self.assertEqual(['ssh-rsa AA2..+aRD0fyVw== root@server'],
+ self.ds.get_public_ssh_keys())
+
+ # test partial metadata (missing user-data in particular)
+ @httpretty.activate
+ def test_metadata_partial(self):
+ httpretty.register_uri(
+ httpretty.GET, MD_URL_RE,
+ body=_request_callback)
+ self.ds.get_data()
+
+ self.assertEqual(GCE_META_PARTIAL.get('instance/id'),
+ self.ds.get_instance_id())
+
+ self.assertEqual(GCE_META_PARTIAL.get('instance/hostname'),
+ self.ds.get_hostname())
diff --git a/tests/unittests/test_datasource/test_openstack.py b/tests/unittests/test_datasource/test_openstack.py
new file mode 100644
index 00000000..3a64430a
--- /dev/null
+++ b/tests/unittests/test_datasource/test_openstack.py
@@ -0,0 +1,301 @@
+# vi: ts=4 expandtab
+#
+# Copyright (C) 2014 Yahoo! Inc.
+#
+# Author: Joshua Harlow <harlowja@yahoo-inc.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import copy
+import json
+import re
+
+from StringIO import StringIO
+
+from urlparse import urlparse
+
+from tests.unittests import helpers as test_helpers
+
+from cloudinit import helpers
+from cloudinit import settings
+from cloudinit.sources import DataSourceOpenStack as ds
+from cloudinit.sources.helpers import openstack
+from cloudinit import util
+
+import httpretty as hp
+
+BASE_URL = "http://169.254.169.254"
+PUBKEY = u'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 = '#!/bin/sh\necho This is user data\n'
+VENDOR_DATA = {
+ 'magic': '',
+}
+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 = 'This is contents of /etc/foo.cfg\n'
+CONTENT_1 = '# this is /etc/bar/bar.cfg\n'
+OS_FILES = {
+ 'openstack/2012-08-10/meta_data.json': json.dumps(OSTACK_META),
+ 'openstack/2012-08-10/user_data': USER_DATA,
+ 'openstack/content/0000': CONTENT_0,
+ 'openstack/content/0001': CONTENT_1,
+ 'openstack/latest/meta_data.json': json.dumps(OSTACK_META),
+ 'openstack/latest/user_data': USER_DATA,
+ 'openstack/latest/vendor_data.json': json.dumps(VENDOR_DATA),
+}
+EC2_FILES = {
+ 'latest/user-data': USER_DATA,
+}
+
+
+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.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 get_request_callback(method, uri, headers):
+ uri = urlparse(uri)
+ path = uri.path.lstrip("/")
+ if path in os_files:
+ return (200, headers, os_files.get(path))
+ return match_ec2_url(uri, headers)
+
+ hp.register_uri(hp.GET, re.compile(r'http://169.254.169.254/.*'),
+ body=get_request_callback)
+
+
+class TestOpenStackDataSource(test_helpers.TestCase):
+ VERSION = 'latest'
+
+ @hp.activate
+ def test_successful(self):
+ _register_uris(self.VERSION, EC2_FILES, EC2_META, OS_FILES)
+ f = ds.read_metadata_service(BASE_URL, version=self.VERSION)
+ self.assertEquals(VENDOR_DATA, f.get('vendordata'))
+ self.assertEquals(CONTENT_0, f['files']['/etc/foo.cfg'])
+ self.assertEquals(CONTENT_1, f['files']['/etc/bar/bar.cfg'])
+ self.assertEquals(2, len(f['files']))
+ self.assertEquals(USER_DATA, f.get('userdata'))
+ self.assertEquals(EC2_META, f.get('ec2-metadata'))
+ self.assertEquals(2, f.get('version'))
+ metadata = f['metadata']
+ self.assertEquals('nova', metadata.get('availability_zone'))
+ self.assertEquals('sm-foo-test.novalocal', metadata.get('hostname'))
+ self.assertEquals('sm-foo-test.novalocal',
+ metadata.get('local-hostname'))
+ self.assertEquals('sm-foo-test', metadata.get('name'))
+ self.assertEquals('b0fa911b-69d4-4476-bbe2-1c92bff6535c',
+ metadata.get('uuid'))
+ self.assertEquals('b0fa911b-69d4-4476-bbe2-1c92bff6535c',
+ metadata.get('instance-id'))
+
+ @hp.activate
+ def test_no_ec2(self):
+ _register_uris(self.VERSION, {}, {}, OS_FILES)
+ f = ds.read_metadata_service(BASE_URL, version=self.VERSION)
+ self.assertEquals(VENDOR_DATA, f.get('vendordata'))
+ self.assertEquals(CONTENT_0, f['files']['/etc/foo.cfg'])
+ self.assertEquals(CONTENT_1, f['files']['/etc/bar/bar.cfg'])
+ self.assertEquals(USER_DATA, f.get('userdata'))
+ self.assertEquals({}, f.get('ec2-metadata'))
+ self.assertEquals(2, f.get('version'))
+
+ @hp.activate
+ 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, ds.read_metadata_service,
+ BASE_URL, version=self.VERSION)
+
+ @hp.activate
+ 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(openstack.BrokenMetadata, ds.read_metadata_service,
+ BASE_URL, version=self.VERSION)
+
+ @hp.activate
+ 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 = ds.read_metadata_service(BASE_URL, version=self.VERSION)
+ self.assertEquals(VENDOR_DATA, f.get('vendordata'))
+ self.assertEquals(CONTENT_0, f['files']['/etc/foo.cfg'])
+ self.assertEquals(CONTENT_1, f['files']['/etc/bar/bar.cfg'])
+ self.assertFalse(f.get('userdata'))
+
+ @hp.activate
+ 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 = ds.read_metadata_service(BASE_URL, version=self.VERSION)
+ self.assertEquals(CONTENT_0, f['files']['/etc/foo.cfg'])
+ self.assertEquals(CONTENT_1, f['files']['/etc/bar/bar.cfg'])
+ self.assertFalse(f.get('vendordata'))
+
+ @hp.activate
+ 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(openstack.BrokenMetadata, ds.read_metadata_service,
+ BASE_URL, version=self.VERSION)
+
+ @hp.activate
+ 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(openstack.BrokenMetadata, ds.read_metadata_service,
+ BASE_URL, version=self.VERSION)
+
+ @hp.activate
+ def test_datasource(self):
+ _register_uris(self.VERSION, EC2_FILES, EC2_META, OS_FILES)
+ ds_os = ds.DataSourceOpenStack(settings.CFG_BUILTIN,
+ None,
+ helpers.Paths({}))
+ self.assertIsNone(ds_os.version)
+ found = ds_os.get_data()
+ self.assertTrue(found)
+ self.assertEquals(2, ds_os.version)
+ md = dict(ds_os.metadata)
+ md.pop('instance-id', None)
+ md.pop('local-hostname', None)
+ self.assertEquals(OSTACK_META, md)
+ self.assertEquals(EC2_META, ds_os.ec2_metadata)
+ self.assertEquals(USER_DATA, ds_os.userdata_raw)
+ self.assertEquals(2, len(ds_os.files))
+ self.assertEquals(VENDOR_DATA, ds_os.vendordata_raw)
+
+ @hp.activate
+ 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({}))
+ self.assertIsNone(ds_os.version)
+ found = ds_os.get_data()
+ self.assertFalse(found)
+ self.assertIsNone(ds_os.version)
+
+ @hp.activate
+ 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({}))
+ ds_os.ds_cfg = {
+ 'max_wait': 0,
+ 'timeout': 0,
+ }
+ self.assertIsNone(ds_os.version)
+ found = ds_os.get_data()
+ self.assertFalse(found)
+ self.assertIsNone(ds_os.version)
+
+ @hp.activate
+ 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({}))
+ ds_os.ds_cfg = {
+ 'max_wait': 0,
+ 'timeout': 0,
+ }
+ self.assertIsNone(ds_os.version)
+ found = ds_os.get_data()
+ self.assertFalse(found)
+ self.assertIsNone(ds_os.version)
diff --git a/tests/unittests/test_ec2_util.py b/tests/unittests/test_ec2_util.py
index 957dc3f2..dd87665d 100644
--- a/tests/unittests/test_ec2_util.py
+++ b/tests/unittests/test_ec2_util.py
@@ -1,6 +1,7 @@
from tests.unittests import helpers
from cloudinit import ec2_utils as eu
+from cloudinit import url_helper as uh
import httpretty as hp
@@ -48,11 +49,11 @@ class TestEc2Util(helpers.TestCase):
body="\n".join(['hostname',
'instance-id',
'ami-launch-index']))
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'hostname'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'hostname'),
status=200, body='ec2.fake.host.name.com')
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'instance-id'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
status=200, body='123')
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'ami-launch-index'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'ami-launch-index'),
status=200, body='1')
md = eu.get_instance_metadata(self.VERSION, retries=0)
self.assertEquals(md['hostname'], 'ec2.fake.host.name.com')
@@ -66,14 +67,14 @@ class TestEc2Util(helpers.TestCase):
body="\n".join(['hostname',
'instance-id',
'public-keys/']))
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'hostname'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'hostname'),
status=200, body='ec2.fake.host.name.com')
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'instance-id'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
status=200, body='123')
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'public-keys/'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'public-keys/'),
status=200, body='0=my-public-key')
hp.register_uri(hp.GET,
- eu.combine_url(base_url, 'public-keys/0/openssh-key'),
+ uh.combine_url(base_url, 'public-keys/0/openssh-key'),
status=200, body='ssh-rsa AAAA.....wZEf my-public-key')
md = eu.get_instance_metadata(self.VERSION, retries=0, timeout=0.1)
self.assertEquals(md['hostname'], 'ec2.fake.host.name.com')
@@ -87,18 +88,18 @@ class TestEc2Util(helpers.TestCase):
body="\n".join(['hostname',
'instance-id',
'public-keys/']))
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'hostname'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'hostname'),
status=200, body='ec2.fake.host.name.com')
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'instance-id'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
status=200, body='123')
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'public-keys/'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'public-keys/'),
status=200,
body="\n".join(['0=my-public-key', '1=my-other-key']))
hp.register_uri(hp.GET,
- eu.combine_url(base_url, 'public-keys/0/openssh-key'),
+ uh.combine_url(base_url, 'public-keys/0/openssh-key'),
status=200, body='ssh-rsa AAAA.....wZEf my-public-key')
hp.register_uri(hp.GET,
- eu.combine_url(base_url, 'public-keys/1/openssh-key'),
+ uh.combine_url(base_url, 'public-keys/1/openssh-key'),
status=200, body='ssh-rsa AAAA.....wZEf my-other-key')
md = eu.get_instance_metadata(self.VERSION, retries=0, timeout=0.1)
self.assertEquals(md['hostname'], 'ec2.fake.host.name.com')
@@ -112,20 +113,20 @@ class TestEc2Util(helpers.TestCase):
body="\n".join(['hostname',
'instance-id',
'block-device-mapping/']))
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'hostname'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'hostname'),
status=200, body='ec2.fake.host.name.com')
- hp.register_uri(hp.GET, eu.combine_url(base_url, 'instance-id'),
+ hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
status=200, body='123')
hp.register_uri(hp.GET,
- eu.combine_url(base_url, 'block-device-mapping/'),
+ uh.combine_url(base_url, 'block-device-mapping/'),
status=200,
body="\n".join(['ami', 'ephemeral0']))
hp.register_uri(hp.GET,
- eu.combine_url(base_url, 'block-device-mapping/ami'),
+ uh.combine_url(base_url, 'block-device-mapping/ami'),
status=200,
body="sdb")
hp.register_uri(hp.GET,
- eu.combine_url(base_url,
+ uh.combine_url(base_url,
'block-device-mapping/ephemeral0'),
status=200,
body="sdc")