diff options
Diffstat (limited to 'tests/unittests/test_datasource/test_openstack.py')
| -rw-r--r-- | tests/unittests/test_datasource/test_openstack.py | 74 |
1 files changed, 61 insertions, 13 deletions
diff --git a/tests/unittests/test_datasource/test_openstack.py b/tests/unittests/test_datasource/test_openstack.py index f43cbec8..7b4e651a 100644 --- a/tests/unittests/test_datasource/test_openstack.py +++ b/tests/unittests/test_datasource/test_openstack.py @@ -19,6 +19,7 @@ import copy import json import re +import unittest from StringIO import StringIO @@ -67,8 +68,8 @@ OSTACK_META = { 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/latest/meta_data.json': json.dumps(OSTACK_META), + 'openstack/latest/user_data': USER_DATA, 'openstack/content/0000': CONTENT_0, 'openstack/content/0001': CONTENT_1, 'openstack/latest/meta_data.json': json.dumps(OSTACK_META), @@ -78,6 +79,9 @@ OS_FILES = { EC2_FILES = { 'latest/user-data': USER_DATA, } +EC2_VERSIONS = [ + 'latest', +] def _register_uris(version, ec2_files, ec2_meta, os_files): @@ -85,6 +89,9 @@ def _register_uris(version, ec2_files, ec2_meta, os_files): 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)) @@ -110,11 +117,20 @@ def _register_uris(version, ec2_files, ec2_meta, os_files): return (200, headers, str(value)) return (404, headers, '') - def get_request_callback(method, uri, headers): - uri = urlparse(uri) + 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/.*'), @@ -127,7 +143,7 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): @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) + f = ds.read_metadata_service(BASE_URL) 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']) @@ -149,7 +165,7 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): @hp.activate def test_no_ec2(self): _register_uris(self.VERSION, {}, {}, OS_FILES) - f = ds.read_metadata_service(BASE_URL, version=self.VERSION) + f = ds.read_metadata_service(BASE_URL) 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']) @@ -165,7 +181,7 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): os_files.pop(k, None) _register_uris(self.VERSION, {}, {}, os_files) self.assertRaises(openstack.NonReadable, ds.read_metadata_service, - BASE_URL, version=self.VERSION) + BASE_URL) @hp.activate def test_bad_uuid(self): @@ -177,7 +193,7 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): 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) + BASE_URL) @hp.activate def test_userdata_empty(self): @@ -186,7 +202,7 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): 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) + f = ds.read_metadata_service(BASE_URL) 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']) @@ -199,7 +215,7 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): 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) + f = ds.read_metadata_service(BASE_URL) self.assertEquals(CONTENT_0, f['files']['/etc/foo.cfg']) self.assertEquals(CONTENT_1, f['files']['/etc/bar/bar.cfg']) self.assertFalse(f.get('vendordata')) @@ -212,7 +228,7 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): 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) + BASE_URL) @hp.activate def test_metadata_invalid(self): @@ -222,7 +238,7 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): 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) + BASE_URL) @hp.activate def test_datasource(self): @@ -241,7 +257,8 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): 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) + self.assertEquals(VENDOR_DATA, ds_os.vendordata_pure) + self.assertEquals(ds_os.vendordata_raw, None) @hp.activate def test_bad_datasource_meta(self): @@ -299,3 +316,34 @@ class TestOpenStackDataSource(test_helpers.HttprettyTestCase): found = ds_os.get_data() self.assertFalse(found) self.assertIsNone(ds_os.version) + + +class TestVendorDataLoading(unittest.TestCase): + def cvj(self, data): + return openstack.convert_vendordata_json(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.assertEqual(self.cvj({'foo': 'bar'}), None) + + 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']) |