From 085ce6b6fcc86101b3c5cc6fe40f66451034f453 Mon Sep 17 00:00:00 2001
From: Scott Moser <smoser@ubuntu.com>
Date: Thu, 23 Aug 2012 17:04:46 -0400
Subject: initial functional unit test
---
cloudinit/sources/DataSourceConfigDrive.py | 102 +++++++++--
.../unittests/test_datasource/test_configdrive.py | 194 +++++++++++++++++++++
2 files changed, 286 insertions(+), 10 deletions(-)
create mode 100644 tests/unittests/test_datasource/test_configdrive.py
diff --git a/cloudinit/sources/DataSourceConfigDrive.py b/cloudinit/sources/DataSourceConfigDrive.py
index 850b281c..677483d0 100644
--- a/cloudinit/sources/DataSourceConfigDrive.py
+++ b/cloudinit/sources/DataSourceConfigDrive.py
@@ -30,7 +30,7 @@ LOG = logging.getLogger(__name__)
# Various defaults/constants...
DEFAULT_IID = "iid-dsconfigdrive"
DEFAULT_MODE = 'pass'
-CFG_DRIVE_FILES = [
+CFG_DRIVE_FILES_V1 = [
"etc/network/interfaces",
"root/.ssh/authorized_keys",
"meta.js",
@@ -49,9 +49,11 @@ class DataSourceConfigDrive(sources.DataSource):
self.cfg = {}
self.dsmode = 'local'
self.seed_dir = os.path.join(paths.seed_dir, 'config_drive')
+ self.version = None
def __str__(self):
- mstr = "%s [%s]" % (util.obj_name(self), self.dsmode)
+ mstr = "%s [%s,ver=%s]" % (util.obj_name(self), self.dsmode,
+ self.version)
mstr += "[seed=%s]" % (self.seed)
return mstr
@@ -62,17 +64,29 @@ class DataSourceConfigDrive(sources.DataSource):
if os.path.isdir(self.seed_dir):
try:
- (md, ud) = read_config_drive_dir(self.seed_dir)
+ (md, ud, files, ver) = read_config_drive_dir(self.seed_dir)
found = self.seed_dir
except NonConfigDriveDir:
util.logexc(LOG, "Failed reading config drive from %s",
self.seed_dir)
if not found:
+ fslist = util.find_devs_with("TYPE=vfat")
+ fslist.extend(util.find_devs_with("TYPE=iso9660"))
+
+ label_list = util.find_devs_with("LABEL=config-2")
+ devlist = list(set(fslist) & set(label_list))
+
dev = find_cfg_drive_device()
- if dev:
+ if dev not in devlist:
+ devlist.append(dev)
+
+ devlist.sort(reverse=True)
+
+ for dev in devlist:
try:
- (md, ud) = util.mount_cb(dev, read_config_drive_dir)
+ (md, ud, files, ver) = util.mount_cb(dev, read_config_drive_dir)
found = dev
+ break
except (NonConfigDriveDir, util.MountFailedError):
pass
@@ -94,6 +108,7 @@ class DataSourceConfigDrive(sources.DataSource):
self.seed = found
self.metadata = md
self.userdata_raw = ud
+ self.version = ver
if md['dsmode'] == self.dsmode:
return True
@@ -122,6 +137,9 @@ class DataSourceConfigDriveNet(DataSourceConfigDrive):
class NonConfigDriveDir(Exception):
pass
+class BrokenConfigDriveDir(Exception):
+ pass
+
def find_cfg_drive_device():
"""Get the config drive device. Return a string like '/dev/vdb'
@@ -154,17 +172,80 @@ def find_cfg_drive_device():
def read_config_drive_dir(source_dir):
+ last_e = NonConfigDriveDir("Not found")
+ for finder in (read_config_drive_dir_v2, read_config_drive_dir_v1):
+ try:
+ data = finder(source_dir)
+ return data
+ except NonConfigDriveDir as exc:
+ last_e = exc
+ raise last_e
+
+def read_config_drive_dir_v2(source_dir, version="latest"):
+ datafiles = (
+ ('metadata',
+ "openstack/%s/meta_data.json" % version, True, json.loads),
+ ('userdata', "openstack/%s/user_data" % version, False, None),
+ ('ec2-metadata', "ec2/latest/metadata.json", False, json.loads),
+ )
+
+ results = {}
+ for (name, path, required, process) in datafiles:
+ fpath = os.path.join(source_dir, path)
+ data = None
+ found = False
+ if os.path.isfile(fpath):
+ try:
+ with open(fpath) as fp:
+ data = fp.read()
+ except Exception as exc:
+ raise BrokenConfigDriveDir("failed to read: %s" % fpath)
+ found = True
+ elif required:
+ raise BrokenConfigDriveDir("missing mandatory %s" % fpath)
+
+ if found and process:
+ try:
+ data = process(data)
+ except Exception as exc:
+ raise BrokenConfigDriveDir("failed to process: %s" % fpath)
+
+ if found:
+ results[name] = data
+
+ def read_content_path(item):
+ # do not use os.path.join here, as content_path starts with /
+ cpath = os.path.sep.join((source_dir, "openstack",
+ "./%s" % item['content_path']))
+ with open(cpath) as fp:
+ return(fp.read())
+
+ files = {}
+ try:
+ for item in results['metadata'].get('files',{}):
+ files[item['path']] = read_content_path(item)
+
+ item = results['metadata'].get("network_config", None)
+ if item:
+ results['network_config'] = read_content_path(item)
+ except Exception as exc:
+ raise BrokenConfigDriveDir("failed to read file %s: %s" % (item, exc))
+
+ results['files'] = files
+ results['cfgdrive_ver'] = 2
+ return results
+
+def read_config_drive_dir_v1(source_dir):
"""
- read_config_drive_dir(source_dir):
- read source_dir, and return a tuple with metadata dict and user-data
- string populated. If not a valid dir, raise a NonConfigDriveDir
+ read source_dir, and return a tuple with metadata dict, user-data,
+ files and version (1). If not a valid dir, raise a NonConfigDriveDir
"""
# TODO(harlowja): fix this for other operating systems...
# Ie: this is where https://fedorahosted.org/netcf/ or similar should
# be hooked in... (or could be)
found = {}
- for af in CFG_DRIVE_FILES:
+ for af in CFG_DRIVE_FILES_V1:
fn = os.path.join(source_dir, af)
if os.path.isfile(fn):
found[af] = fn
@@ -211,7 +292,8 @@ def read_config_drive_dir(source_dir):
if 'user-data' in meta_js:
ud = meta_js['user-data']
- return (md, ud)
+ # metadata, user-data, 'files', 1
+ return {'metadata': md, 'userdata': ud, 'files': [], 'cfgdrive_ver': 1}
# Used to match classes to dependencies
diff --git a/tests/unittests/test_datasource/test_configdrive.py b/tests/unittests/test_datasource/test_configdrive.py
new file mode 100644
index 00000000..f73200e5
--- /dev/null
+++ b/tests/unittests/test_datasource/test_configdrive.py
@@ -0,0 +1,194 @@
+from copy import copy
+import json
+import os
+import os.path
+
+from cloudinit.sources import DataSourceConfigDrive
+from cloudinit import url_helper
+from mocker import MockerTestCase
+
+
+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',
+ 'block-device-mapping': {
+ 'ami': 'sda1',
+ 'ephemeral0': 'sda2',
+ 'root': '/dev/sda1',
+ 'swap': 'sda3'},
+ 'hostname': 'sm-foo-test.novalocal',
+ 'instance-action': 'none',
+ 'instance-id': 'i-00000001',
+ 'instance-type': 'm1.tiny',
+ 'local-hostname': 'sm-foo-test.novalocal',
+ 'local-ipv4': None,
+ 'placement': {'availability-zone': 'nova'},
+ 'public-hostname': 'sm-foo-test.novalocal',
+ 'public-ipv4': '',
+ 'public-keys': {'0': {'openssh-key': PUBKEY}},
+ 'reservation-id': 'r-iru5qm4m',
+ 'security-groups': ['default']
+}
+USER_DATA = '#!/bin/sh\necho This is user data\n'
+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'
+
+CFG_DRIVE_FILES_V2 = {
+ 'ec2/2009-04-04/meta-data.json': json.dumps(EC2_META),
+ 'ec2/2009-04-04/user-data': USER_DATA,
+ 'ec2/latest/meta-data.json': json.dumps(EC2_META),
+ 'ec2/latest/user-data': USER_DATA,
+ '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}
+
+
+class TestConfigDriveDataSource(MockerTestCase):
+
+ def setUp(self):
+ super(TestConfigDriveDataSource, self).setUp()
+ # Make a temp directoy for tests to use.
+ self.tmp = self.makeDir()
+
+ def test_dir_valid(self):
+ """Verify a dir is read as such."""
+
+ my_d = os.path.join(self.tmp, "valid")
+ populate_dir(my_d, CFG_DRIVE_FILES_V2)
+
+ found = DataSourceConfigDrive.read_config_drive_dir(my_d)
+
+ self.assertEqual(USER_DATA, found['userdata'])
+ self.assertEqual(OSTACK_META, found['metadata'])
+ self.assertEqual(found['files']['/etc/foo.cfg'], CONTENT_0)
+ self.assertEqual(found['files']['/etc/bar/bar.cfg'], CONTENT_1)
+
+ def test_seed_dir_valid_extra(self):
+ """Verify extra files do not affect datasource validity."""
+
+ my_d = os.path.join(self.tmp, "valid_extra")
+ data = copy(CFG_DRIVE_FILES_V2)
+ data["myfoofile"] = "myfoocontent"
+
+ populate_dir(my_d, data)
+
+# (userdata, metadata) = DataSourceMAAS.read_maas_seed_dir(my_d)
+#
+# self.assertEqual(userdata, data['user-data'])
+# for key in ('instance-id', 'local-hostname'):
+# self.assertEqual(data[key], metadata[key])
+#
+# # additional files should not just appear as keys in metadata atm
+# self.assertFalse(('foo' in metadata))
+#
+# def test_seed_dir_invalid(self):
+# """Verify that invalid seed_dir raises MAASSeedDirMalformed."""
+#
+# valid = {'instance-id': 'i-instanceid',
+# 'local-hostname': 'test-hostname', 'user-data': ''}
+#
+# my_based = os.path.join(self.tmp, "valid_extra")
+#
+# # missing 'userdata' file
+# my_d = "%s-01" % my_based
+# invalid_data = copy(valid)
+# del invalid_data['local-hostname']
+# populate_dir(my_d, invalid_data)
+# self.assertRaises(DataSourceMAAS.MAASSeedDirMalformed,
+# DataSourceMAAS.read_maas_seed_dir, my_d)
+#
+# # missing 'instance-id'
+# my_d = "%s-02" % my_based
+# invalid_data = copy(valid)
+# del invalid_data['instance-id']
+# populate_dir(my_d, invalid_data)
+# self.assertRaises(DataSourceMAAS.MAASSeedDirMalformed,
+# DataSourceMAAS.read_maas_seed_dir, my_d)
+#
+# def test_seed_dir_none(self):
+# """Verify that empty seed_dir raises MAASSeedDirNone."""
+#
+# my_d = os.path.join(self.tmp, "valid_empty")
+# self.assertRaises(DataSourceMAAS.MAASSeedDirNone,
+# DataSourceMAAS.read_maas_seed_dir, my_d)
+#
+# def test_seed_dir_missing(self):
+# """Verify that missing seed_dir raises MAASSeedDirNone."""
+# self.assertRaises(DataSourceMAAS.MAASSeedDirNone,
+# DataSourceMAAS.read_maas_seed_dir,
+# os.path.join(self.tmp, "nonexistantdirectory"))
+#
+# def test_seed_url_valid(self):
+# """Verify that valid seed_url is read as such."""
+# valid = {'meta-data/instance-id': 'i-instanceid',
+# 'meta-data/local-hostname': 'test-hostname',
+# 'meta-data/public-keys': 'test-hostname',
+# 'user-data': 'foodata'}
+# valid_order = [
+# 'meta-data/local-hostname',
+# 'meta-data/instance-id',
+# 'meta-data/public-keys',
+# 'user-data',
+# ]
+# my_seed = "http://example.com/xmeta"
+# my_ver = "1999-99-99"
+# my_headers = {'header1': 'value1', 'header2': 'value2'}
+#
+# def my_headers_cb(url):
+# return my_headers
+#
+# mock_request = self.mocker.replace(url_helper.readurl,
+# passthrough=False)
+#
+# for key in valid_order:
+# url = "%s/%s/%s" % (my_seed, my_ver, key)
+# mock_request(url, headers=my_headers, timeout=None)
+# resp = valid.get(key)
+# self.mocker.result(url_helper.UrlResponse(200, resp))
+# self.mocker.replay()
+#
+# (userdata, metadata) = DataSourceMAAS.read_maas_seed_url(my_seed,
+# header_cb=my_headers_cb, version=my_ver)
+#
+# self.assertEqual("foodata", userdata)
+# self.assertEqual(metadata['instance-id'],
+# valid['meta-data/instance-id'])
+# self.assertEqual(metadata['local-hostname'],
+# valid['meta-data/local-hostname'])
+#
+# def test_seed_url_invalid(self):
+# """Verify that invalid seed_url raises MAASSeedDirMalformed."""
+# pass
+#
+# def test_seed_url_missing(self):
+# """Verify seed_url with no found entries raises MAASSeedDirNone."""
+# pass
+
+
+def populate_dir(seed_dir, files):
+ os.mkdir(seed_dir)
+ for (name, content) in files.iteritems():
+ path = os.path.join(seed_dir, name)
+ dirname = os.path.dirname(path)
+ if not os.path.isdir(dirname):
+ os.makedirs(dirname)
+ with open(path, "w") as fp:
+ fp.write(content)
+ fp.close()
+
+# vi: ts=4 expandtab
--
cgit v1.2.3