diff options
Diffstat (limited to 'libbtrfsutil/python/tests/test_subvolume.py')
| -rw-r--r-- | libbtrfsutil/python/tests/test_subvolume.py | 215 |
1 files changed, 161 insertions, 54 deletions
diff --git a/libbtrfsutil/python/tests/test_subvolume.py b/libbtrfsutil/python/tests/test_subvolume.py index 93396cba..99ec97bc 100644 --- a/libbtrfsutil/python/tests/test_subvolume.py +++ b/libbtrfsutil/python/tests/test_subvolume.py @@ -20,10 +20,18 @@ import errno import os import os.path from pathlib import PurePath +import subprocess import traceback import btrfsutil -from tests import BtrfsTestCase, HAVE_PATH_LIKE +from tests import ( + BtrfsTestCase, + drop_privs, + HAVE_PATH_LIKE, + NOBODY_UID, + regain_privs, + skipUnlessHaveNobody, +) class TestSubvolume(BtrfsTestCase): @@ -87,7 +95,7 @@ class TestSubvolume(BtrfsTestCase): finally: os.chdir(pwd) - def test_subvolume_info(self): + def _test_subvolume_info(self, subvol, snapshot): for arg in self.path_or_fd(self.mountpoint): with self.subTest(type=type(arg)): info = btrfsutil.subvolume_info(arg) @@ -100,7 +108,7 @@ class TestSubvolume(BtrfsTestCase): self.assertEqual(info.parent_uuid, bytes(16)) self.assertEqual(info.received_uuid, bytes(16)) self.assertNotEqual(info.generation, 0) - self.assertEqual(info.ctransid, 0) + self.assertGreaterEqual(info.ctransid, 0) self.assertEqual(info.otransid, 0) self.assertEqual(info.stransid, 0) self.assertEqual(info.rtransid, 0) @@ -109,9 +117,6 @@ class TestSubvolume(BtrfsTestCase): self.assertEqual(info.stime, 0) self.assertEqual(info.rtime, 0) - subvol = os.path.join(self.mountpoint, 'subvol') - btrfsutil.create_subvolume(subvol) - info = btrfsutil.subvolume_info(subvol) self.assertEqual(info.id, 256) self.assertEqual(info.parent_id, 5) @@ -132,19 +137,43 @@ class TestSubvolume(BtrfsTestCase): self.assertEqual(info.rtime, 0) subvol_uuid = info.uuid - snapshot = os.path.join(self.mountpoint, 'snapshot') - btrfsutil.create_snapshot(subvol, snapshot) info = btrfsutil.subvolume_info(snapshot) self.assertEqual(info.parent_uuid, subvol_uuid) # TODO: test received_uuid, stransid, rtransid, stime, and rtime + def test_subvolume_info(self): + subvol = os.path.join(self.mountpoint, 'subvol') + btrfsutil.create_subvolume(subvol) + snapshot = os.path.join(self.mountpoint, 'snapshot') + btrfsutil.create_snapshot(subvol, snapshot) + + self._test_subvolume_info(subvol, snapshot) + for arg in self.path_or_fd(self.mountpoint): with self.subTest(type=type(arg)): with self.assertRaises(btrfsutil.BtrfsUtilError) as e: # BTRFS_EXTENT_TREE_OBJECTID btrfsutil.subvolume_info(arg, 2) + self.assertEqual(e.exception.btrfsutilerror, + btrfsutil.ERROR_SUBVOLUME_NOT_FOUND) + + @skipUnlessHaveNobody + def test_subvolume_info_unprivileged(self): + subvol = os.path.join(self.mountpoint, 'subvol') + btrfsutil.create_subvolume(subvol) + snapshot = os.path.join(self.mountpoint, 'snapshot') + btrfsutil.create_snapshot(subvol, snapshot) + + with drop_privs(): + try: + self._test_subvolume_info(subvol, snapshot) + except OSError as e: + if e.errno == errno.ENOTTY: + self.skipTest('BTRFS_IOC_GET_SUBVOL_INFO is not available') + else: + raise def test_read_only(self): for arg in self.path_or_fd(self.mountpoint): @@ -202,7 +231,7 @@ class TestSubvolume(BtrfsTestCase): btrfsutil.create_subvolume(subvol + '6//') self.assertTrue(btrfsutil.is_subvolume(subvol + '6')) - transid = btrfsutil.create_subvolume(subvol + '7', async=True) + transid = btrfsutil.create_subvolume(subvol + '7', async_=True) self.assertTrue(btrfsutil.is_subvolume(subvol + '7')) self.assertGreater(transid, 0) @@ -265,7 +294,7 @@ class TestSubvolume(BtrfsTestCase): btrfsutil.create_snapshot(subvol, snapshot + '2', recursive=True) self.assertTrue(os.path.exists(os.path.join(snapshot + '2', 'nested/more_nested/nested_dir'))) - transid = btrfsutil.create_snapshot(subvol, snapshot + '3', recursive=True, async=True) + transid = btrfsutil.create_snapshot(subvol, snapshot + '3', recursive=True, async_=True) self.assertTrue(os.path.exists(os.path.join(snapshot + '3', 'nested/more_nested/nested_dir'))) self.assertGreater(transid, 0) @@ -328,58 +357,136 @@ class TestSubvolume(BtrfsTestCase): with self.subTest(type=type(arg)): self.assertEqual(btrfsutil.deleted_subvolumes(arg), [256]) - def test_subvolume_iterator(self): - pwd = os.getcwd() - try: - os.chdir(self.mountpoint) - btrfsutil.create_subvolume('foo') + def _test_subvolume_iterator(self): + btrfsutil.create_subvolume('foo') - path, subvol = next(btrfsutil.SubvolumeIterator('.', info=True)) + with btrfsutil.SubvolumeIterator('.', info=True) as it: + path, subvol = next(it) self.assertEqual(path, 'foo') self.assertIsInstance(subvol, btrfsutil.SubvolumeInfo) self.assertEqual(subvol.id, 256) self.assertEqual(subvol.parent_id, 5) - - btrfsutil.create_subvolume('foo/bar') - btrfsutil.create_subvolume('foo/bar/baz') - - subvols = [ - ('foo', 256), - ('foo/bar', 257), - ('foo/bar/baz', 258), - ] - - for arg in self.path_or_fd('.'): - with self.subTest(type=type(arg)): - self.assertEqual(list(btrfsutil.SubvolumeIterator(arg)), subvols) - self.assertEqual(list(btrfsutil.SubvolumeIterator('.', top=0)), subvols) - - self.assertEqual(list(btrfsutil.SubvolumeIterator('.', post_order=True)), + self.assertRaises(StopIteration, next, it) + + btrfsutil.create_subvolume('foo/bar') + btrfsutil.create_subvolume('foo/bar/baz') + + subvols = [ + ('foo', 256), + ('foo/bar', 257), + ('foo/bar/baz', 258), + ] + + for arg in self.path_or_fd('.'): + with self.subTest(type=type(arg)), btrfsutil.SubvolumeIterator(arg) as it: + self.assertEqual(list(it), subvols) + with btrfsutil.SubvolumeIterator('.', top=0) as it: + self.assertEqual(list(it), subvols) + if os.geteuid() == 0: + with btrfsutil.SubvolumeIterator('foo', top=5) as it: + self.assertEqual(list(it), subvols) + + with btrfsutil.SubvolumeIterator('.', post_order=True) as it: + self.assertEqual(list(it), [('foo/bar/baz', 258), ('foo/bar', 257), ('foo', 256)]) - subvols = [ - ('bar', 257), - ('bar/baz', 258), - ] - - self.assertEqual(list(btrfsutil.SubvolumeIterator('.', top=256)), subvols) - self.assertEqual(list(btrfsutil.SubvolumeIterator('foo', top=0)), subvols) - - os.rename('foo/bar/baz', 'baz') - self.assertEqual(sorted(btrfsutil.SubvolumeIterator('.')), - [('baz', 258), - ('foo', 256), - ('foo/bar', 257)]) - - with btrfsutil.SubvolumeIterator('.') as it: - self.assertGreaterEqual(it.fileno(), 0) - it.close() - with self.assertRaises(ValueError): - next(iter(it)) - with self.assertRaises(ValueError): - it.fileno() - it.close() + subvols = [ + ('bar', 257), + ('bar/baz', 258), + ] + + if os.geteuid() == 0: + with btrfsutil.SubvolumeIterator('.', top=256) as it: + self.assertEqual(list(it), subvols) + with btrfsutil.SubvolumeIterator('foo') as it: + self.assertEqual(list(it), subvols) + with btrfsutil.SubvolumeIterator('foo', top=0) as it: + self.assertEqual(list(it), subvols) + + os.rename('foo/bar/baz', 'baz') + os.mkdir('dir') + btrfsutil.create_subvolume('dir/qux') + os.mkdir('dir/qux/dir2') + btrfsutil.create_subvolume('dir/qux/dir2/quux') + + subvols = [ + ('baz', 258), + ('dir/qux', 259), + ('dir/qux/dir2/quux', 260), + ('foo', 256), + ('foo/bar', 257), + ] + + # Test various corner cases of the unprivileged implementation + # where we can't access the subvolume. + if os.geteuid() != 0: + with regain_privs(): + # We don't have permission to traverse the path. + os.mkdir('directory_perms', 0o700) + btrfsutil.create_subvolume('directory_perms/subvol') + + # We don't have permission to resolve the subvolume path. + os.mkdir('subvol_perms', 0o755) + btrfsutil.create_subvolume('subvol_perms/subvol') + os.chmod('subvol_perms/subvol', 0o700) + + # The path doesn't exist. + os.mkdir('enoent', 0o755) + btrfsutil.create_subvolume('enoent/subvol') + subprocess.check_call(['mount', '-t', 'tmpfs', 'tmpfs', 'enoent']) + + # The path exists but it's not a subvolume. + os.mkdir('notsubvol', 0o755) + btrfsutil.create_subvolume('notsubvol/subvol') + subprocess.check_call(['mount', '-t', 'tmpfs', 'tmpfs', 'notsubvol']) + os.mkdir('notsubvol/subvol') + + # The path exists and is a subvolume, but on a different + # filesystem. + os.mkdir('wrongfs', 0o755) + btrfsutil.create_subvolume('wrongfs/subvol') + other_mountpoint, _ = self.mount_btrfs() + subprocess.check_call(['mount', '--bind', '--', + other_mountpoint, 'wrongfs/subvol']) + + # The path exists and is a subvolume on the same + # filesystem, but not the right one. + os.mkdir('wrongsubvol', 0o755) + btrfsutil.create_subvolume('wrongsubvol/subvol') + subprocess.check_call(['mount', '--bind', 'baz', 'wrongsubvol/subvol']) + + + with btrfsutil.SubvolumeIterator('.') as it: + self.assertEqual(sorted(it), subvols) + with btrfsutil.SubvolumeIterator('.', post_order=True) as it: + self.assertEqual(sorted(it), subvols) + + with btrfsutil.SubvolumeIterator('.') as it: + self.assertGreaterEqual(it.fileno(), 0) + it.close() + with self.assertRaises(ValueError): + next(iter(it)) + with self.assertRaises(ValueError): + it.fileno() + it.close() + + def test_subvolume_iterator(self): + pwd = os.getcwd() + try: + os.chdir(self.mountpoint) + self._test_subvolume_iterator() + finally: + os.chdir(pwd) + + @skipUnlessHaveNobody + def test_subvolume_iterator_unprivileged(self): + os.chown(self.mountpoint, NOBODY_UID, -1) + pwd = os.getcwd() + try: + os.chdir(self.mountpoint) + with drop_privs(): + self._test_subvolume_iterator() finally: os.chdir(pwd) |