diff options
Diffstat (limited to 'tests/unittests')
| -rw-r--r-- | tests/unittests/helpers.py | 25 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_configdrive.py | 35 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_gce.py | 111 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_openstack.py | 301 | ||||
| -rw-r--r-- | tests/unittests/test_ec2_util.py | 35 |
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") |