summaryrefslogtreecommitdiff
path: root/nova/tests/unit/virt/disk/vfs
diff options
context:
space:
mode:
authorSean Dague <sean@dague.net>2014-11-07 14:27:03 +0100
committerSean Dague <sean@dague.net>2014-11-12 15:31:08 -0500
commit89cd6a0c493e26b5a9e017c99d731464292abbaf (patch)
treec2bf790d1684cd539b820247113492495123a163 /nova/tests/unit/virt/disk/vfs
parent5c8bbaafef590e4d346a03051a0ba55c8be26c5c (diff)
downloadnova-89cd6a0c493e26b5a9e017c99d731464292abbaf.tar.gz
move all tests to nova/tests/unit
As part of the split of functional and unit tests we need to isolate the unit tests into a separate directory for having multiple test targets in a sane way. Part of bp:functional-tests-for-nova Change-Id: Id42ba373c1bda6a312b673ab2b489ca56da8c628
Diffstat (limited to 'nova/tests/unit/virt/disk/vfs')
-rw-r--r--nova/tests/unit/virt/disk/vfs/__init__.py0
-rw-r--r--nova/tests/unit/virt/disk/vfs/fakeguestfs.py188
-rw-r--r--nova/tests/unit/virt/disk/vfs/test_guestfs.py264
-rw-r--r--nova/tests/unit/virt/disk/vfs/test_localfs.py385
4 files changed, 837 insertions, 0 deletions
diff --git a/nova/tests/unit/virt/disk/vfs/__init__.py b/nova/tests/unit/virt/disk/vfs/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/nova/tests/unit/virt/disk/vfs/__init__.py
diff --git a/nova/tests/unit/virt/disk/vfs/fakeguestfs.py b/nova/tests/unit/virt/disk/vfs/fakeguestfs.py
new file mode 100644
index 0000000000..5e5efa7a14
--- /dev/null
+++ b/nova/tests/unit/virt/disk/vfs/fakeguestfs.py
@@ -0,0 +1,188 @@
+# Copyright 2012 Red Hat, Inc
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+
+EVENT_APPLIANCE = 0x1
+EVENT_LIBRARY = 0x2
+EVENT_WARNING = 0x3
+EVENT_TRACE = 0x4
+
+
+class GuestFS(object):
+ SUPPORT_CLOSE_ON_EXIT = True
+ SUPPORT_RETURN_DICT = True
+
+ def __init__(self, **kwargs):
+ if not self.SUPPORT_CLOSE_ON_EXIT and 'close_on_exit' in kwargs:
+ raise TypeError('close_on_exit')
+ if not self.SUPPORT_RETURN_DICT and 'python_return_dict' in kwargs:
+ raise TypeError('python_return_dict')
+
+ self._python_return_dict = kwargs.get('python_return_dict', False)
+ self.kwargs = kwargs
+ self.drives = []
+ self.running = False
+ self.closed = False
+ self.mounts = []
+ self.files = {}
+ self.auginit = False
+ self.root_mounted = False
+ self.backend_settings = None
+ self.trace_enabled = False
+ self.verbose_enabled = False
+ self.event_callback = None
+
+ def launch(self):
+ self.running = True
+
+ def shutdown(self):
+ self.running = False
+ self.mounts = []
+ self.drives = []
+
+ def set_backend_settings(self, settings):
+ self.backend_settings = settings
+
+ def close(self):
+ self.closed = True
+
+ def add_drive_opts(self, file, *args, **kwargs):
+ if file == "/some/fail/file":
+ raise RuntimeError("%s: No such file or directory", file)
+
+ self.drives.append((file, kwargs['format']))
+
+ def add_drive(self, file, format=None, *args, **kwargs):
+ self.add_drive_opts(file, format=None, *args, **kwargs)
+
+ def inspect_os(self):
+ return ["/dev/guestvgf/lv_root"]
+
+ def inspect_get_mountpoints(self, dev):
+ mountpoints = [("/home", "/dev/mapper/guestvgf-lv_home"),
+ ("/", "/dev/mapper/guestvgf-lv_root"),
+ ("/boot", "/dev/vda1")]
+
+ if self.SUPPORT_RETURN_DICT and self._python_return_dict:
+ return dict(mountpoints)
+ else:
+ return mountpoints
+
+ def mount_options(self, options, device, mntpoint):
+ if mntpoint == "/":
+ self.root_mounted = True
+ else:
+ if not self.root_mounted:
+ raise RuntimeError(
+ "mount: %s: No such file or directory" % mntpoint)
+ self.mounts.append((options, device, mntpoint))
+
+ def mkdir_p(self, path):
+ if path not in self.files:
+ self.files[path] = {
+ "isdir": True,
+ "gid": 100,
+ "uid": 100,
+ "mode": 0o700
+ }
+
+ def read_file(self, path):
+ if path not in self.files:
+ self.files[path] = {
+ "isdir": False,
+ "content": "Hello World",
+ "gid": 100,
+ "uid": 100,
+ "mode": 0o700
+ }
+
+ return self.files[path]["content"]
+
+ def write(self, path, content):
+ if path not in self.files:
+ self.files[path] = {
+ "isdir": False,
+ "content": "Hello World",
+ "gid": 100,
+ "uid": 100,
+ "mode": 0o700
+ }
+
+ self.files[path]["content"] = content
+
+ def write_append(self, path, content):
+ if path not in self.files:
+ self.files[path] = {
+ "isdir": False,
+ "content": "Hello World",
+ "gid": 100,
+ "uid": 100,
+ "mode": 0o700
+ }
+
+ self.files[path]["content"] = self.files[path]["content"] + content
+
+ def stat(self, path):
+ if path not in self.files:
+ raise RuntimeError("No such file: " + path)
+
+ return self.files[path]["mode"]
+
+ def chown(self, uid, gid, path):
+ if path not in self.files:
+ raise RuntimeError("No such file: " + path)
+
+ if uid != -1:
+ self.files[path]["uid"] = uid
+ if gid != -1:
+ self.files[path]["gid"] = gid
+
+ def chmod(self, mode, path):
+ if path not in self.files:
+ raise RuntimeError("No such file: " + path)
+
+ self.files[path]["mode"] = mode
+
+ def aug_init(self, root, flags):
+ self.auginit = True
+
+ def aug_close(self):
+ self.auginit = False
+
+ def aug_get(self, cfgpath):
+ if not self.auginit:
+ raise RuntimeError("Augeus not initialized")
+
+ if cfgpath == "/files/etc/passwd/root/uid":
+ return 0
+ elif cfgpath == "/files/etc/passwd/fred/uid":
+ return 105
+ elif cfgpath == "/files/etc/passwd/joe/uid":
+ return 110
+ elif cfgpath == "/files/etc/group/root/gid":
+ return 0
+ elif cfgpath == "/files/etc/group/users/gid":
+ return 500
+ elif cfgpath == "/files/etc/group/admins/gid":
+ return 600
+ raise RuntimeError("Unknown path %s", cfgpath)
+
+ def set_trace(self, enabled):
+ self.trace_enabled = enabled
+
+ def set_verbose(self, enabled):
+ self.verbose_enabled = enabled
+
+ def set_event_callback(self, func, events):
+ self.event_callback = (func, events)
diff --git a/nova/tests/unit/virt/disk/vfs/test_guestfs.py b/nova/tests/unit/virt/disk/vfs/test_guestfs.py
new file mode 100644
index 0000000000..33dd100329
--- /dev/null
+++ b/nova/tests/unit/virt/disk/vfs/test_guestfs.py
@@ -0,0 +1,264 @@
+# Copyright (C) 2012 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import sys
+
+from nova import exception
+from nova import test
+from nova.tests.unit.virt.disk.vfs import fakeguestfs
+from nova.virt.disk.vfs import guestfs as vfsimpl
+
+
+class VirtDiskVFSGuestFSTest(test.NoDBTestCase):
+
+ def setUp(self):
+ super(VirtDiskVFSGuestFSTest, self).setUp()
+ sys.modules['guestfs'] = fakeguestfs
+ vfsimpl.guestfs = fakeguestfs
+
+ def _do_test_appliance_setup_inspect(self, forcetcg):
+ if forcetcg:
+ vfsimpl.force_tcg()
+ else:
+ vfsimpl.force_tcg(False)
+
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2",
+ imgfmt="qcow2",
+ partition=-1)
+ vfs.setup()
+
+ if forcetcg:
+ self.assertEqual("force_tcg", vfs.handle.backend_settings)
+ vfsimpl.force_tcg(False)
+ else:
+ self.assertIsNone(vfs.handle.backend_settings)
+
+ self.assertTrue(vfs.handle.running)
+ self.assertEqual(3, len(vfs.handle.mounts))
+ self.assertEqual("/dev/mapper/guestvgf-lv_root",
+ vfs.handle.mounts[0][1])
+ self.assertEqual("/dev/vda1",
+ vfs.handle.mounts[1][1])
+ self.assertEqual("/dev/mapper/guestvgf-lv_home",
+ vfs.handle.mounts[2][1])
+ self.assertEqual("/", vfs.handle.mounts[0][2])
+ self.assertEqual("/boot", vfs.handle.mounts[1][2])
+ self.assertEqual("/home", vfs.handle.mounts[2][2])
+
+ handle = vfs.handle
+ vfs.teardown()
+
+ self.assertIsNone(vfs.handle)
+ self.assertFalse(handle.running)
+ self.assertTrue(handle.closed)
+ self.assertEqual(0, len(handle.mounts))
+
+ def test_appliance_setup_inspect_auto(self):
+ self._do_test_appliance_setup_inspect(False)
+
+ def test_appliance_setup_inspect_tcg(self):
+ self._do_test_appliance_setup_inspect(True)
+
+ def test_appliance_setup_inspect_no_root_raises(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2",
+ imgfmt="qcow2",
+ partition=-1)
+ # call setup to init the handle so we can stub it
+ vfs.setup()
+
+ self.assertIsNone(vfs.handle.backend_settings)
+
+ def fake_inspect_os():
+ return []
+
+ self.stubs.Set(vfs.handle, 'inspect_os', fake_inspect_os)
+ self.assertRaises(exception.NovaException, vfs.setup_os_inspect)
+
+ def test_appliance_setup_inspect_multi_boots_raises(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2",
+ imgfmt="qcow2",
+ partition=-1)
+ # call setup to init the handle so we can stub it
+ vfs.setup()
+
+ self.assertIsNone(vfs.handle.backend_settings)
+
+ def fake_inspect_os():
+ return ['fake1', 'fake2']
+
+ self.stubs.Set(vfs.handle, 'inspect_os', fake_inspect_os)
+ self.assertRaises(exception.NovaException, vfs.setup_os_inspect)
+
+ def test_appliance_setup_static_nopart(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2",
+ imgfmt="qcow2",
+ partition=None)
+ vfs.setup()
+
+ self.assertIsNone(vfs.handle.backend_settings)
+ self.assertTrue(vfs.handle.running)
+ self.assertEqual(1, len(vfs.handle.mounts))
+ self.assertEqual("/dev/sda", vfs.handle.mounts[0][1])
+ self.assertEqual("/", vfs.handle.mounts[0][2])
+
+ handle = vfs.handle
+ vfs.teardown()
+
+ self.assertIsNone(vfs.handle)
+ self.assertFalse(handle.running)
+ self.assertTrue(handle.closed)
+ self.assertEqual(0, len(handle.mounts))
+
+ def test_appliance_setup_static_part(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2",
+ imgfmt="qcow2",
+ partition=2)
+ vfs.setup()
+
+ self.assertIsNone(vfs.handle.backend_settings)
+ self.assertTrue(vfs.handle.running)
+ self.assertEqual(1, len(vfs.handle.mounts))
+ self.assertEqual("/dev/sda2", vfs.handle.mounts[0][1])
+ self.assertEqual("/", vfs.handle.mounts[0][2])
+
+ handle = vfs.handle
+ vfs.teardown()
+
+ self.assertIsNone(vfs.handle)
+ self.assertFalse(handle.running)
+ self.assertTrue(handle.closed)
+ self.assertEqual(0, len(handle.mounts))
+
+ def test_makepath(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ vfs.make_path("/some/dir")
+ vfs.make_path("/other/dir")
+
+ self.assertIn("/some/dir", vfs.handle.files)
+ self.assertIn("/other/dir", vfs.handle.files)
+ self.assertTrue(vfs.handle.files["/some/dir"]["isdir"])
+ self.assertTrue(vfs.handle.files["/other/dir"]["isdir"])
+
+ vfs.teardown()
+
+ def test_append_file(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ vfs.append_file("/some/file", " Goodbye")
+
+ self.assertIn("/some/file", vfs.handle.files)
+ self.assertEqual("Hello World Goodbye",
+ vfs.handle.files["/some/file"]["content"])
+
+ vfs.teardown()
+
+ def test_replace_file(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ vfs.replace_file("/some/file", "Goodbye")
+
+ self.assertIn("/some/file", vfs.handle.files)
+ self.assertEqual("Goodbye",
+ vfs.handle.files["/some/file"]["content"])
+
+ vfs.teardown()
+
+ def test_read_file(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ self.assertEqual("Hello World", vfs.read_file("/some/file"))
+
+ vfs.teardown()
+
+ def test_has_file(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ vfs.read_file("/some/file")
+
+ self.assertTrue(vfs.has_file("/some/file"))
+ self.assertFalse(vfs.has_file("/other/file"))
+
+ vfs.teardown()
+
+ def test_set_permissions(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ vfs.read_file("/some/file")
+
+ self.assertEqual(0o700, vfs.handle.files["/some/file"]["mode"])
+
+ vfs.set_permissions("/some/file", 0o7777)
+ self.assertEqual(0o7777, vfs.handle.files["/some/file"]["mode"])
+
+ vfs.teardown()
+
+ def test_set_ownership(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ vfs.read_file("/some/file")
+
+ self.assertEqual(100, vfs.handle.files["/some/file"]["uid"])
+ self.assertEqual(100, vfs.handle.files["/some/file"]["gid"])
+
+ vfs.set_ownership("/some/file", "fred", None)
+ self.assertEqual(105, vfs.handle.files["/some/file"]["uid"])
+ self.assertEqual(100, vfs.handle.files["/some/file"]["gid"])
+
+ vfs.set_ownership("/some/file", None, "users")
+ self.assertEqual(105, vfs.handle.files["/some/file"]["uid"])
+ self.assertEqual(500, vfs.handle.files["/some/file"]["gid"])
+
+ vfs.set_ownership("/some/file", "joe", "admins")
+ self.assertEqual(110, vfs.handle.files["/some/file"]["uid"])
+ self.assertEqual(600, vfs.handle.files["/some/file"]["gid"])
+
+ vfs.teardown()
+
+ def test_close_on_error(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ self.assertFalse(vfs.handle.kwargs['close_on_exit'])
+ vfs.teardown()
+ self.stubs.Set(fakeguestfs.GuestFS, 'SUPPORT_CLOSE_ON_EXIT', False)
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ self.assertNotIn('close_on_exit', vfs.handle.kwargs)
+ vfs.teardown()
+
+ def test_python_return_dict(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ self.assertFalse(vfs.handle.kwargs['python_return_dict'])
+ vfs.teardown()
+ self.stubs.Set(fakeguestfs.GuestFS, 'SUPPORT_RETURN_DICT', False)
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ self.assertNotIn('python_return_dict', vfs.handle.kwargs)
+ vfs.teardown()
+
+ def test_setup_debug_disable(self):
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ self.assertFalse(vfs.handle.trace_enabled)
+ self.assertFalse(vfs.handle.verbose_enabled)
+ self.assertIsNone(vfs.handle.event_callback)
+
+ def test_setup_debug_enabled(self):
+ self.flags(debug=True, group='guestfs')
+ vfs = vfsimpl.VFSGuestFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.setup()
+ self.assertTrue(vfs.handle.trace_enabled)
+ self.assertTrue(vfs.handle.verbose_enabled)
+ self.assertIsNotNone(vfs.handle.event_callback)
diff --git a/nova/tests/unit/virt/disk/vfs/test_localfs.py b/nova/tests/unit/virt/disk/vfs/test_localfs.py
new file mode 100644
index 0000000000..6e7780e74b
--- /dev/null
+++ b/nova/tests/unit/virt/disk/vfs/test_localfs.py
@@ -0,0 +1,385 @@
+# Copyright (C) 2012 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from oslo.concurrency import processutils
+from oslo.config import cfg
+
+from nova import exception
+from nova import test
+from nova.tests.unit import utils as tests_utils
+import nova.utils
+from nova.virt.disk.vfs import localfs as vfsimpl
+
+CONF = cfg.CONF
+
+dirs = []
+files = {}
+commands = []
+
+
+def fake_execute(*args, **kwargs):
+ commands.append({"args": args, "kwargs": kwargs})
+
+ if args[0] == "readlink":
+ if args[1] == "-nm":
+ if args[2] in ["/scratch/dir/some/file",
+ "/scratch/dir/some/dir",
+ "/scratch/dir/other/dir",
+ "/scratch/dir/other/file"]:
+ return args[2], ""
+ elif args[1] == "-e":
+ if args[2] in files:
+ return args[2], ""
+
+ return "", "No such file"
+ elif args[0] == "mkdir":
+ dirs.append(args[2])
+ elif args[0] == "chown":
+ owner = args[1]
+ path = args[2]
+ if path not in files:
+ raise Exception("No such file: " + path)
+
+ sep = owner.find(':')
+ if sep != -1:
+ user = owner[0:sep]
+ group = owner[sep + 1:]
+ else:
+ user = owner
+ group = None
+
+ if user:
+ if user == "fred":
+ uid = 105
+ else:
+ uid = 110
+ files[path]["uid"] = uid
+ if group:
+ if group == "users":
+ gid = 500
+ else:
+ gid = 600
+ files[path]["gid"] = gid
+ elif args[0] == "chgrp":
+ group = args[1]
+ path = args[2]
+ if path not in files:
+ raise Exception("No such file: " + path)
+
+ if group == "users":
+ gid = 500
+ else:
+ gid = 600
+ files[path]["gid"] = gid
+ elif args[0] == "chmod":
+ mode = args[1]
+ path = args[2]
+ if path not in files:
+ raise Exception("No such file: " + path)
+
+ files[path]["mode"] = int(mode, 8)
+ elif args[0] == "cat":
+ path = args[1]
+ if path not in files:
+ files[path] = {
+ "content": "Hello World",
+ "gid": 100,
+ "uid": 100,
+ "mode": 0o700
+ }
+ return files[path]["content"], ""
+ elif args[0] == "tee":
+ if args[1] == "-a":
+ path = args[2]
+ append = True
+ else:
+ path = args[1]
+ append = False
+ if path not in files:
+ files[path] = {
+ "content": "Hello World",
+ "gid": 100,
+ "uid": 100,
+ "mode": 0o700,
+ }
+ if append:
+ files[path]["content"] += kwargs["process_input"]
+ else:
+ files[path]["content"] = kwargs["process_input"]
+
+
+class VirtDiskVFSLocalFSTestPaths(test.NoDBTestCase):
+ def setUp(self):
+ super(VirtDiskVFSLocalFSTestPaths, self).setUp()
+
+ real_execute = processutils.execute
+
+ def nonroot_execute(*cmd_parts, **kwargs):
+ kwargs.pop('run_as_root', None)
+ return real_execute(*cmd_parts, **kwargs)
+
+ self.stubs.Set(processutils, 'execute', nonroot_execute)
+
+ def test_check_safe_path(self):
+ if not tests_utils.coreutils_readlink_available():
+ self.skipTest("coreutils readlink(1) unavailable")
+ vfs = vfsimpl.VFSLocalFS("dummy.img")
+ vfs.imgdir = "/foo"
+ ret = vfs._canonical_path('etc/something.conf')
+ self.assertEqual(ret, '/foo/etc/something.conf')
+
+ def test_check_unsafe_path(self):
+ if not tests_utils.coreutils_readlink_available():
+ self.skipTest("coreutils readlink(1) unavailable")
+ vfs = vfsimpl.VFSLocalFS("dummy.img")
+ vfs.imgdir = "/foo"
+ self.assertRaises(exception.Invalid,
+ vfs._canonical_path,
+ 'etc/../../../something.conf')
+
+
+class VirtDiskVFSLocalFSTest(test.NoDBTestCase):
+ def test_makepath(self):
+ global dirs, commands
+ dirs = []
+ commands = []
+ self.stubs.Set(processutils, 'execute', fake_execute)
+
+ vfs = vfsimpl.VFSLocalFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.imgdir = "/scratch/dir"
+ vfs.make_path("/some/dir")
+ vfs.make_path("/other/dir")
+
+ self.assertEqual(dirs,
+ ["/scratch/dir/some/dir", "/scratch/dir/other/dir"]),
+
+ root_helper = nova.utils._get_root_helper()
+ self.assertEqual(commands,
+ [{'args': ('readlink', '-nm',
+ '/scratch/dir/some/dir'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('mkdir', '-p',
+ '/scratch/dir/some/dir'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-nm',
+ '/scratch/dir/other/dir'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('mkdir', '-p',
+ '/scratch/dir/other/dir'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}}])
+
+ def test_append_file(self):
+ global files, commands
+ files = {}
+ commands = []
+ self.stubs.Set(processutils, 'execute', fake_execute)
+
+ vfs = vfsimpl.VFSLocalFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.imgdir = "/scratch/dir"
+ vfs.append_file("/some/file", " Goodbye")
+
+ self.assertIn("/scratch/dir/some/file", files)
+ self.assertEqual(files["/scratch/dir/some/file"]["content"],
+ "Hello World Goodbye")
+
+ root_helper = nova.utils._get_root_helper()
+ self.assertEqual(commands,
+ [{'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('tee', '-a',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'process_input': ' Goodbye',
+ 'run_as_root': True,
+ 'root_helper': root_helper}}])
+
+ def test_replace_file(self):
+ global files, commands
+ files = {}
+ commands = []
+ self.stubs.Set(processutils, 'execute', fake_execute)
+
+ vfs = vfsimpl.VFSLocalFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.imgdir = "/scratch/dir"
+ vfs.replace_file("/some/file", "Goodbye")
+
+ self.assertIn("/scratch/dir/some/file", files)
+ self.assertEqual(files["/scratch/dir/some/file"]["content"],
+ "Goodbye")
+
+ root_helper = nova.utils._get_root_helper()
+ self.assertEqual(commands,
+ [{'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('tee', '/scratch/dir/some/file'),
+ 'kwargs': {'process_input': 'Goodbye',
+ 'run_as_root': True,
+ 'root_helper': root_helper}}])
+
+ def test_read_file(self):
+ global commands, files
+ files = {}
+ commands = []
+ self.stubs.Set(processutils, 'execute', fake_execute)
+
+ vfs = vfsimpl.VFSLocalFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.imgdir = "/scratch/dir"
+ self.assertEqual(vfs.read_file("/some/file"), "Hello World")
+
+ root_helper = nova.utils._get_root_helper()
+ self.assertEqual(commands,
+ [{'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('cat', '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}}])
+
+ def test_has_file(self):
+ global commands, files
+ files = {}
+ commands = []
+ self.stubs.Set(processutils, 'execute', fake_execute)
+
+ vfs = vfsimpl.VFSLocalFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.imgdir = "/scratch/dir"
+ vfs.read_file("/some/file")
+
+ self.assertTrue(vfs.has_file("/some/file"))
+ self.assertFalse(vfs.has_file("/other/file"))
+
+ root_helper = nova.utils._get_root_helper()
+ self.assertEqual(commands,
+ [{'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('cat', '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-e',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-nm',
+ '/scratch/dir/other/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-e',
+ '/scratch/dir/other/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ ])
+
+ def test_set_permissions(self):
+ global commands, files
+ commands = []
+ files = {}
+ self.stubs.Set(processutils, 'execute', fake_execute)
+
+ vfs = vfsimpl.VFSLocalFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.imgdir = "/scratch/dir"
+ vfs.read_file("/some/file")
+
+ vfs.set_permissions("/some/file", 0o777)
+ self.assertEqual(files["/scratch/dir/some/file"]["mode"], 0o777)
+
+ root_helper = nova.utils._get_root_helper()
+ self.assertEqual(commands,
+ [{'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('cat', '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('chmod', '777',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}}])
+
+ def test_set_ownership(self):
+ global commands, files
+ commands = []
+ files = {}
+ self.stubs.Set(processutils, 'execute', fake_execute)
+
+ vfs = vfsimpl.VFSLocalFS(imgfile="/dummy.qcow2", imgfmt="qcow2")
+ vfs.imgdir = "/scratch/dir"
+ vfs.read_file("/some/file")
+
+ self.assertEqual(files["/scratch/dir/some/file"]["uid"], 100)
+ self.assertEqual(files["/scratch/dir/some/file"]["gid"], 100)
+
+ vfs.set_ownership("/some/file", "fred", None)
+ self.assertEqual(files["/scratch/dir/some/file"]["uid"], 105)
+ self.assertEqual(files["/scratch/dir/some/file"]["gid"], 100)
+
+ vfs.set_ownership("/some/file", None, "users")
+ self.assertEqual(files["/scratch/dir/some/file"]["uid"], 105)
+ self.assertEqual(files["/scratch/dir/some/file"]["gid"], 500)
+
+ vfs.set_ownership("/some/file", "joe", "admins")
+ self.assertEqual(files["/scratch/dir/some/file"]["uid"], 110)
+ self.assertEqual(files["/scratch/dir/some/file"]["gid"], 600)
+
+ root_helper = nova.utils._get_root_helper()
+ self.assertEqual(commands,
+ [{'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('cat', '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('chown', 'fred',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('chgrp', 'users',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('readlink', '-nm',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}},
+ {'args': ('chown', 'joe:admins',
+ '/scratch/dir/some/file'),
+ 'kwargs': {'run_as_root': True,
+ 'root_helper': root_helper}}])