summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--cloudinit/sources/DataSourceOVF.py74
-rw-r--r--cloudinit/tests/helpers.py10
-rw-r--r--tests/unittests/test_datasource/test_ovf.py164
3 files changed, 221 insertions, 27 deletions
diff --git a/cloudinit/sources/DataSourceOVF.py b/cloudinit/sources/DataSourceOVF.py
index 24b45d55..ccebf11a 100644
--- a/cloudinit/sources/DataSourceOVF.py
+++ b/cloudinit/sources/DataSourceOVF.py
@@ -375,26 +375,56 @@ def get_ovf_env(dirname):
return (None, False)
-# Transport functions take no input and return
-# a 3 tuple of content, path, filename
-def transport_iso9660(require_iso=True):
+def maybe_cdrom_device(devname):
+ """Test if devname matches known list of devices which may contain iso9660
+ filesystems.
- # default_regex matches values in
- # /lib/udev/rules.d/60-cdrom_id.rules
- # KERNEL!="sr[0-9]*|hd[a-z]|xvd*", GOTO="cdrom_end"
- envname = "CLOUD_INIT_CDROM_DEV_REGEX"
- default_regex = "^(sr[0-9]+|hd[a-z]|xvd.*)"
+ Be helpful in accepting either knames (with no leading /dev/) or full path
+ names, but do not allow paths outside of /dev/, like /dev/foo/bar/xxx.
+ """
+ if not devname:
+ return False
+ elif not isinstance(devname, util.string_types):
+ raise ValueError("Unexpected input for devname: %s" % devname)
+
+ # resolve '..' and multi '/' elements
+ devname = os.path.normpath(devname)
- devname_regex = os.environ.get(envname, default_regex)
+ # drop leading '/dev/'
+ if devname.startswith("/dev/"):
+ # partition returns tuple (before, partition, after)
+ devname = devname.partition("/dev/")[-1]
+
+ # ignore leading slash (/sr0), else fail on / in name (foo/bar/xvdc)
+ if devname.startswith("/"):
+ devname = devname.split("/")[-1]
+ elif devname.count("/") > 0:
+ return False
+
+ # if empty string
+ if not devname:
+ return False
+
+ # default_regex matches values in /lib/udev/rules.d/60-cdrom_id.rules
+ # KERNEL!="sr[0-9]*|hd[a-z]|xvd*", GOTO="cdrom_end"
+ default_regex = r"^(sr[0-9]+|hd[a-z]|xvd.*)"
+ devname_regex = os.environ.get("CLOUD_INIT_CDROM_DEV_REGEX", default_regex)
cdmatch = re.compile(devname_regex)
+ return cdmatch.match(devname) is not None
+
+
+# Transport functions take no input and return
+# a 3 tuple of content, path, filename
+def transport_iso9660(require_iso=True):
+
# Go through mounts to see if it was already mounted
mounts = util.mounts()
for (dev, info) in mounts.items():
fstype = info['fstype']
if fstype != "iso9660" and require_iso:
continue
- if cdmatch.match(dev[5:]) is None: # take off '/dev/'
+ if not maybe_cdrom_device(dev):
continue
mp = info['mountpoint']
(fname, contents) = get_ovf_env(mp)
@@ -406,29 +436,19 @@ def transport_iso9660(require_iso=True):
else:
mtype = None
- devs = os.listdir("/dev/")
- devs.sort()
+ # generate a list of devices with mtype filesystem, filter by regex
+ devs = [dev for dev in
+ util.find_devs_with("TYPE=%s" % mtype if mtype else None)
+ if maybe_cdrom_device(dev)]
for dev in devs:
- fullp = os.path.join("/dev/", dev)
-
- if (fullp in mounts or
- not cdmatch.match(dev) or os.path.isdir(fullp)):
- continue
-
- try:
- # See if we can read anything at all...??
- util.peek_file(fullp, 512)
- except IOError:
- continue
-
try:
- (fname, contents) = util.mount_cb(fullp, get_ovf_env, mtype=mtype)
+ (fname, contents) = util.mount_cb(dev, get_ovf_env, mtype=mtype)
except util.MountFailedError:
- LOG.debug("%s not mountable as iso9660", fullp)
+ LOG.debug("%s not mountable as iso9660", dev)
continue
if contents is not False:
- return (contents, fullp, fname)
+ return (contents, dev, fname)
return (False, None, None)
diff --git a/cloudinit/tests/helpers.py b/cloudinit/tests/helpers.py
index 28e26622..6f88a5b7 100644
--- a/cloudinit/tests/helpers.py
+++ b/cloudinit/tests/helpers.py
@@ -104,6 +104,16 @@ class TestCase(unittest2.TestCase):
super(TestCase, self).setUp()
self.reset_global_state()
+ def add_patch(self, target, attr, **kwargs):
+ """Patches specified target object and sets it as attr on test
+ instance also schedules cleanup"""
+ if 'autospec' not in kwargs:
+ kwargs['autospec'] = True
+ m = mock.patch(target, **kwargs)
+ p = m.start()
+ self.addCleanup(m.stop)
+ setattr(self, attr, p)
+
class CiTestCase(TestCase):
"""This is the preferred test case base class unless user
diff --git a/tests/unittests/test_datasource/test_ovf.py b/tests/unittests/test_datasource/test_ovf.py
index 9dbf4dd9..700da86c 100644
--- a/tests/unittests/test_datasource/test_ovf.py
+++ b/tests/unittests/test_datasource/test_ovf.py
@@ -5,6 +5,7 @@
# This file is part of cloud-init. See LICENSE file for license information.
import base64
+from collections import OrderedDict
from cloudinit.tests import helpers as test_helpers
@@ -70,4 +71,167 @@ class TestReadOvfEnv(test_helpers.TestCase):
self.assertEqual({'password': "passw0rd"}, cfg)
self.assertIsNone(ud)
+
+class TestTransportIso9660(test_helpers.CiTestCase):
+
+ def setUp(self):
+ super(TestTransportIso9660, self).setUp()
+ self.add_patch('cloudinit.util.find_devs_with',
+ 'm_find_devs_with')
+ self.add_patch('cloudinit.util.mounts', 'm_mounts')
+ self.add_patch('cloudinit.util.mount_cb', 'm_mount_cb')
+ self.add_patch('cloudinit.sources.DataSourceOVF.get_ovf_env',
+ 'm_get_ovf_env')
+ self.m_get_ovf_env.return_value = ('myfile', 'mycontent')
+
+ def test_find_already_mounted(self):
+ """Check we call get_ovf_env from on matching mounted devices"""
+ mounts = {
+ '/dev/sr9': {
+ 'fstype': 'iso9660',
+ 'mountpoint': 'wark/media/sr9',
+ 'opts': 'ro',
+ }
+ }
+ self.m_mounts.return_value = mounts
+
+ (contents, fullp, fname) = dsovf.transport_iso9660()
+ self.assertEqual("mycontent", contents)
+ self.assertEqual("/dev/sr9", fullp)
+ self.assertEqual("myfile", fname)
+
+ def test_find_already_mounted_skips_non_iso9660(self):
+ """Check we call get_ovf_env ignoring non iso9660"""
+ mounts = {
+ '/dev/xvdb': {
+ 'fstype': 'vfat',
+ 'mountpoint': 'wark/foobar',
+ 'opts': 'defaults,noatime',
+ },
+ '/dev/xvdc': {
+ 'fstype': 'iso9660',
+ 'mountpoint': 'wark/media/sr9',
+ 'opts': 'ro',
+ }
+ }
+ # We use an OrderedDict here to ensure we check xvdb before xvdc
+ # as we're not mocking the regex matching, however, if we place
+ # an entry in the results then we can be reasonably sure that
+ # we're skipping an entry which fails to match.
+ self.m_mounts.return_value = (
+ OrderedDict(sorted(mounts.items(), key=lambda t: t[0])))
+
+ (contents, fullp, fname) = dsovf.transport_iso9660()
+ self.assertEqual("mycontent", contents)
+ self.assertEqual("/dev/xvdc", fullp)
+ self.assertEqual("myfile", fname)
+
+ def test_find_already_mounted_matches_kname(self):
+ """Check we dont regex match on basename of the device"""
+ mounts = {
+ '/dev/foo/bar/xvdc': {
+ 'fstype': 'iso9660',
+ 'mountpoint': 'wark/media/sr9',
+ 'opts': 'ro',
+ }
+ }
+ # we're skipping an entry which fails to match.
+ self.m_mounts.return_value = mounts
+
+ (contents, fullp, fname) = dsovf.transport_iso9660()
+ self.assertEqual(False, contents)
+ self.assertIsNone(fullp)
+ self.assertIsNone(fname)
+
+ def test_mount_cb_called_on_blkdevs_with_iso9660(self):
+ """Check we call mount_cb on blockdevs with iso9660 only"""
+ self.m_mounts.return_value = {}
+ self.m_find_devs_with.return_value = ['/dev/sr0']
+ self.m_mount_cb.return_value = ("myfile", "mycontent")
+
+ (contents, fullp, fname) = dsovf.transport_iso9660()
+
+ self.m_mount_cb.assert_called_with(
+ "/dev/sr0", dsovf.get_ovf_env, mtype="iso9660")
+ self.assertEqual("mycontent", contents)
+ self.assertEqual("/dev/sr0", fullp)
+ self.assertEqual("myfile", fname)
+
+ def test_mount_cb_called_on_blkdevs_with_iso9660_check_regex(self):
+ """Check we call mount_cb on blockdevs with iso9660 and match regex"""
+ self.m_mounts.return_value = {}
+ self.m_find_devs_with.return_value = [
+ '/dev/abc', '/dev/my-cdrom', '/dev/sr0']
+ self.m_mount_cb.return_value = ("myfile", "mycontent")
+
+ (contents, fullp, fname) = dsovf.transport_iso9660()
+
+ self.m_mount_cb.assert_called_with(
+ "/dev/sr0", dsovf.get_ovf_env, mtype="iso9660")
+ self.assertEqual("mycontent", contents)
+ self.assertEqual("/dev/sr0", fullp)
+ self.assertEqual("myfile", fname)
+
+ def test_mount_cb_not_called_no_matches(self):
+ """Check we don't call mount_cb if nothing matches"""
+ self.m_mounts.return_value = {}
+ self.m_find_devs_with.return_value = ['/dev/vg/myovf']
+
+ (contents, fullp, fname) = dsovf.transport_iso9660()
+
+ self.assertEqual(0, self.m_mount_cb.call_count)
+ self.assertEqual(False, contents)
+ self.assertIsNone(fullp)
+ self.assertIsNone(fname)
+
+ def test_mount_cb_called_require_iso_false(self):
+ """Check we call mount_cb on blockdevs with require_iso=False"""
+ self.m_mounts.return_value = {}
+ self.m_find_devs_with.return_value = ['/dev/xvdz']
+ self.m_mount_cb.return_value = ("myfile", "mycontent")
+
+ (contents, fullp, fname) = dsovf.transport_iso9660(require_iso=False)
+
+ self.m_mount_cb.assert_called_with(
+ "/dev/xvdz", dsovf.get_ovf_env, mtype=None)
+ self.assertEqual("mycontent", contents)
+ self.assertEqual("/dev/xvdz", fullp)
+ self.assertEqual("myfile", fname)
+
+ def test_maybe_cdrom_device_none(self):
+ """Test maybe_cdrom_device returns False for none/empty input"""
+ self.assertFalse(dsovf.maybe_cdrom_device(None))
+ self.assertFalse(dsovf.maybe_cdrom_device(''))
+
+ def test_maybe_cdrom_device_non_string_exception(self):
+ """Test maybe_cdrom_device raises ValueError on non-string types"""
+ with self.assertRaises(ValueError):
+ dsovf.maybe_cdrom_device({'a': 'eleven'})
+
+ def test_maybe_cdrom_device_false_on_multi_dir_paths(self):
+ """Test maybe_cdrom_device is false on /dev[/.*]/* paths"""
+ self.assertFalse(dsovf.maybe_cdrom_device('/dev/foo/sr0'))
+ self.assertFalse(dsovf.maybe_cdrom_device('foo/sr0'))
+ self.assertFalse(dsovf.maybe_cdrom_device('../foo/sr0'))
+ self.assertFalse(dsovf.maybe_cdrom_device('../foo/sr0'))
+
+ def test_maybe_cdrom_device_true_on_hd_partitions(self):
+ """Test maybe_cdrom_device is false on /dev/hd[a-z][0-9]+ paths"""
+ self.assertTrue(dsovf.maybe_cdrom_device('/dev/hda1'))
+ self.assertTrue(dsovf.maybe_cdrom_device('hdz9'))
+
+ def test_maybe_cdrom_device_true_on_valid_relative_paths(self):
+ """Test maybe_cdrom_device normalizes paths"""
+ self.assertTrue(dsovf.maybe_cdrom_device('/dev/wark/../sr9'))
+ self.assertTrue(dsovf.maybe_cdrom_device('///sr0'))
+ self.assertTrue(dsovf.maybe_cdrom_device('/sr0'))
+ self.assertTrue(dsovf.maybe_cdrom_device('//dev//hda'))
+
+ def test_maybe_cdrom_device_true_on_xvd_partitions(self):
+ """Test maybe_cdrom_device returns true on xvd*"""
+ self.assertTrue(dsovf.maybe_cdrom_device('/dev/xvda'))
+ self.assertTrue(dsovf.maybe_cdrom_device('/dev/xvda1'))
+ self.assertTrue(dsovf.maybe_cdrom_device('xvdza1'))
+
+#
# vi: ts=4 expandtab