diff options
author | Sean Dague <sean@dague.net> | 2014-11-07 14:27:03 +0100 |
---|---|---|
committer | Sean Dague <sean@dague.net> | 2014-11-12 15:31:08 -0500 |
commit | 89cd6a0c493e26b5a9e017c99d731464292abbaf (patch) | |
tree | c2bf790d1684cd539b820247113492495123a163 /nova/tests/unit/virt/disk/vfs | |
parent | 5c8bbaafef590e4d346a03051a0ba55c8be26c5c (diff) | |
download | nova-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__.py | 0 | ||||
-rw-r--r-- | nova/tests/unit/virt/disk/vfs/fakeguestfs.py | 188 | ||||
-rw-r--r-- | nova/tests/unit/virt/disk/vfs/test_guestfs.py | 264 | ||||
-rw-r--r-- | nova/tests/unit/virt/disk/vfs/test_localfs.py | 385 |
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}}]) |