import os import time import tempfile import re from packaging.version import Version from .fs_test import FSTestCase, FSNoDevTestCase, mounted, check_output import overrides_hack import utils from utils import TestTags, tag_test from gi.repository import BlockDev, GLib class GenericNoDevTestCase(FSNoDevTestCase): pass class GenericTestCase(FSTestCase): loop_size = 500 * 1024**2 def setUp(self): super(GenericTestCase, self).setUp() self.mount_dir = tempfile.mkdtemp(prefix="libblockdev.", suffix="generic_test") self._vfat_version = self._get_dosfstools_version() def _get_dosfstools_version(self): _ret, out, _err = utils.run_command("mkfs.vfat --help") # mkfs.fat 4.1 (2017-01-24) m = re.search(r"mkfs\.fat ([\d\.]+)", out) if not m or len(m.groups()) != 1: raise RuntimeError("Failed to determine dosfstools version from: %s" % out) return Version(m.groups()[0]) class TestGenericWipe(GenericTestCase): @tag_test(TestTags.CORE) def test_generic_wipe(self): """Verify that generic signature wipe works as expected""" with self.assertRaises(GLib.GError): BlockDev.fs_wipe("/non/existing/device", True) ret = utils.run("pvcreate -ff -y %s --config \"devices {use_devicesfile = 0}\" >/dev/null 2>&1" % self.loop_dev) self.assertEqual(ret, 0) succ = BlockDev.fs_wipe(self.loop_dev, True) self.assertTrue(succ) # now test the same multiple times in a row for i in range(10): ret = utils.run("pvcreate -ff -y %s --config \"devices {use_devicesfile = 0}\" >/dev/null 2>&1" % self.loop_dev) self.assertEqual(ret, 0) succ = BlockDev.fs_wipe(self.loop_dev, True) self.assertTrue(succ) # vfat has multiple signatures on the device so it allows us to test the # 'all' argument of fs_wipe() if self._vfat_version >= Version("4.2"): ret = utils.run("mkfs.vfat -I %s >/dev/null 2>&1 --mbr=n" % self.loop_dev) else: ret = utils.run("mkfs.vfat -I %s >/dev/null 2>&1" % self.loop_dev) self.assertEqual(ret, 0) time.sleep(0.5) succ = BlockDev.fs_wipe(self.loop_dev, False) self.assertTrue(succ) # the second signature should still be there # XXX: lsblk uses the udev db so it we need to make sure it is up to date utils.run("udevadm settle") fs_type = check_output(["blkid", "-ovalue", "-sTYPE", "-p", self.loop_dev]).strip() self.assertEqual(fs_type, b"vfat") # get rid of all the remaining signatures (there could be vfat + PMBR for some reason) succ = BlockDev.fs_wipe(self.loop_dev, True) self.assertTrue(succ) utils.run("udevadm settle") fs_type = check_output(["blkid", "-ovalue", "-sTYPE", "-p", self.loop_dev]).strip() self.assertEqual(fs_type, b"") # now do the wipe all in a one step if self._vfat_version >= Version("4.2"): ret = utils.run("mkfs.vfat -I %s >/dev/null 2>&1 --mbr=n" % self.loop_dev) else: ret = utils.run("mkfs.vfat -I %s >/dev/null 2>&1" % self.loop_dev) self.assertEqual(ret, 0) succ = BlockDev.fs_wipe(self.loop_dev, True) self.assertTrue(succ) utils.run("udevadm settle") fs_type = check_output(["blkid", "-ovalue", "-sTYPE", "-p", self.loop_dev]).strip() self.assertEqual(fs_type, b"") # try to wipe empty device with self.assertRaisesRegex(GLib.GError, "No signature detected on the device"): BlockDev.fs_wipe(self.loop_dev, True) @tag_test(TestTags.CORE) def test_generic_wipe_force(self): ret = utils.run("mkfs.ext2 %s >/dev/null 2>&1" % self.loop_dev) self.assertEqual(ret, 0) with mounted(self.loop_dev, self.mount_dir): # default should be force=False with self.assertRaisesRegex(GLib.GError, "Failed to open the device"): BlockDev.fs_wipe(self.loop_dev) succ = BlockDev.fs_wipe(self.loop_dev, force=True) self.assertTrue(succ) fs_type = check_output(["blkid", "-ovalue", "-sTYPE", "-p", self.loop_dev]).strip() self.assertEqual(fs_type, b"") class TestClean(GenericTestCase): def test_clean(self): """Verify that device clean works as expected""" with self.assertRaises(GLib.GError): BlockDev.fs_clean("/non/existing/device") # empty device shouldn't fail succ = BlockDev.fs_clean(self.loop_dev) self.assertTrue(succ) ret = utils.run("pvcreate -ff -y %s --config \"devices {use_devicesfile = 0}\" >/dev/null 2>&1" % self.loop_dev) self.assertEqual(ret, 0) succ = BlockDev.fs_clean(self.loop_dev) self.assertTrue(succ) # XXX: lsblk uses the udev db so it we need to make sure it is up to date utils.run("udevadm settle") fs_type = check_output(["blkid", "-ovalue", "-sTYPE", "-p", self.loop_dev]).strip() self.assertEqual(fs_type, b"") # vfat has multiple signatures on the device so it allows us to test # that clean removes all signatures if self._vfat_version >= Version("4.2"): ret = utils.run("mkfs.vfat -I %s >/dev/null 2>&1 --mbr=n" % self.loop_dev) else: ret = utils.run("mkfs.vfat -I %s >/dev/null 2>&1" % self.loop_dev) self.assertEqual(ret, 0) time.sleep(0.5) succ = BlockDev.fs_clean(self.loop_dev) self.assertTrue(succ) utils.run("udevadm settle") fs_type = check_output(["blkid", "-ovalue", "-sTYPE", "-p", self.loop_dev]).strip() self.assertEqual(fs_type, b"") @tag_test(TestTags.CORE) def test_generic_clean_force(self): ret = utils.run("mkfs.ext2 %s >/dev/null 2>&1" % self.loop_dev) self.assertEqual(ret, 0) with mounted(self.loop_dev, self.mount_dir): # default should be force=False with self.assertRaisesRegex(GLib.GError, "Failed to open the device"): BlockDev.fs_clean(self.loop_dev) succ = BlockDev.fs_clean(self.loop_dev, force=True) self.assertTrue(succ) fs_type = check_output(["blkid", "-ovalue", "-sTYPE", "-p", self.loop_dev]).strip() self.assertEqual(fs_type, b"") class CanResizeRepairCheckLabel(GenericNoDevTestCase): def test_can_resize(self): """Verify that tooling query works for resize""" avail, mode, util = BlockDev.fs_can_resize("ext4") self.assertTrue(avail) self.assertEqual(util, None) old_path = os.environ.get("PATH", "") os.environ["PATH"] = "" avail, mode, util = BlockDev.fs_can_resize("ext4") os.environ["PATH"] = old_path self.assertFalse(avail) self.assertEqual(util, "resize2fs") self.assertEqual(mode, BlockDev.FSResizeFlags.ONLINE_GROW | BlockDev.FSResizeFlags.OFFLINE_GROW | BlockDev.FSResizeFlags.OFFLINE_SHRINK) avail = BlockDev.fs_can_resize("vfat") self.assertTrue(avail) with self.assertRaises(GLib.GError): BlockDev.fs_can_resize("non-existing-fs") with self.assertRaises(GLib.GError): BlockDev.fs_can_resize("udf") def test_can_repair(self): """Verify that tooling query works for repair""" avail, util = BlockDev.fs_can_repair("xfs") self.assertTrue(avail) self.assertEqual(util, None) old_path = os.environ.get("PATH", "") os.environ["PATH"] = "" avail, util = BlockDev.fs_can_repair("xfs") os.environ["PATH"] = old_path self.assertFalse(avail) self.assertEqual(util, "xfs_repair") avail = BlockDev.fs_can_repair("vfat") self.assertTrue(avail) with self.assertRaises(GLib.GError): BlockDev.fs_can_repair("nilfs2") with self.assertRaises(GLib.GError): BlockDev.fs_can_repair("udf") def test_can_check(self): """Verify that tooling query works for consistency check""" avail, util = BlockDev.fs_can_check("xfs") self.assertTrue(avail) self.assertEqual(util, None) old_path = os.environ.get("PATH", "") os.environ["PATH"] = "" avail, util = BlockDev.fs_can_check("xfs") os.environ["PATH"] = old_path self.assertFalse(avail) self.assertEqual(util, "xfs_db") avail = BlockDev.fs_can_check("vfat") self.assertTrue(avail) with self.assertRaises(GLib.GError): BlockDev.fs_can_check("nilfs2") with self.assertRaises(GLib.GError): BlockDev.fs_can_check("udf") def test_can_set_label(self): """Verify that tooling query works for setting the label""" avail, util = BlockDev.fs_can_set_label("xfs") self.assertTrue(avail) self.assertEqual(util, None) old_path = os.environ.get("PATH", "") os.environ["PATH"] = "" avail, util = BlockDev.fs_can_set_label("xfs") os.environ["PATH"] = old_path self.assertFalse(avail) self.assertEqual(util, "xfs_admin") with self.assertRaises(GLib.GError): BlockDev.fs_can_set_label("non-existing-fs") def test_can_get_size(self): """Verify that tooling query works for getting size""" avail, util = BlockDev.fs_can_get_size("xfs") self.assertTrue(avail) self.assertEqual(util, None) old_path = os.environ.get("PATH", "") os.environ["PATH"] = "" avail, util = BlockDev.fs_can_get_size("xfs") os.environ["PATH"] = old_path self.assertFalse(avail) self.assertEqual(util, "xfs_admin") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_size("non-existing-fs") def test_can_get_free_space(self): """Verify that tooling query works for getting free space""" avail, util = BlockDev.fs_can_get_free_space("ext4") self.assertTrue(avail) self.assertEqual(util, None) old_path = os.environ.get("PATH", "") os.environ["PATH"] = "" avail, util = BlockDev.fs_can_get_free_space("ext4") os.environ["PATH"] = old_path self.assertFalse(avail) self.assertEqual(util, "dumpe2fs") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_free_space("xfs") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_free_space("non-existing-fs") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_free_space("exfat") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_free_space("udf") def test_can_get_info(self): """Verify that tooling query works for getting info""" avail, util = BlockDev.fs_can_get_free_space("ext4") self.assertTrue(avail) self.assertEqual(util, None) old_path = os.environ.get("PATH", "") os.environ["PATH"] = "" avail, util = BlockDev.fs_can_get_free_space("ext4") os.environ["PATH"] = old_path self.assertFalse(avail) self.assertEqual(util, "dumpe2fs") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_free_space("non-existing-fs") def test_can_get_min_size(self): """Verify that tooling query works for getting min size""" avail, util = BlockDev.fs_can_get_min_size("ext4") self.assertTrue(avail) self.assertEqual(util, None) old_path = os.environ.get("PATH", "") os.environ["PATH"] = "" avail, util = BlockDev.fs_can_get_min_size("ext4") os.environ["PATH"] = old_path self.assertFalse(avail) self.assertEqual(util, "resize2fs") avail, util = BlockDev.fs_can_get_min_size("ntfs") if self.ntfs_avail: self.assertTrue(avail) self.assertEqual(util, None) else: self.assertFalse(avail) self.assertEqual(util, "ntfsresize") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_min_size("xfs") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_min_size("non-existing-fs") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_min_size("exfat") with self.assertRaises(GLib.GError): BlockDev.fs_can_get_min_size("udf") class GenericMkfs(GenericTestCase): def _test_ext_generic_mkfs(self, fsname, info_fn, label=None, uuid=None, force=False, extra=None, default_label=None): # clean the device succ = BlockDev.fs_clean(self.loop_dev) self.assertTrue(succ) supported, flags, _util = BlockDev.fs_can_mkfs(fsname) if not supported: self.skipTest("skipping %s: not available" % fsname) if flags & BlockDev.FSMkfsOptionsFlags.DRY_RUN: # try dry run first options = BlockDev.FSMkfsOptions(None, None, True, False) succ = BlockDev.fs_mkfs(self.loop_dev, fsname, options) self.assertTrue(succ) fstype = BlockDev.fs_get_fstype (self.loop_dev) self.assertIsNone(fstype) options = BlockDev.FSMkfsOptions(label, uuid, False, False) succ = BlockDev.fs_mkfs(self.loop_dev, fsname, options, extra) self.assertTrue(succ) fstype = BlockDev.fs_get_fstype (self.loop_dev) self.assertEqual(fstype, fsname) info = info_fn(self.loop_dev) self.assertIsNotNone(info) if label is not None: if label == "": if default_label: self.assertEqual(info.label, default_label) else: self.assertEqual(info.label, "") else: self.assertEqual(info.label, label) if uuid: self.assertEqual(info.uuid, uuid) if force: # try overwriting the existing filesystem, should fail without force with self.assertRaises(GLib.GError): BlockDev.fs_mkfs(self.loop_dev, fsname) # now add the option options = BlockDev.FSMkfsOptions(force=True) succ = BlockDev.fs_mkfs(self.loop_dev, fsname, options, extra) self.assertTrue(succ) def test_exfat_generic_mkfs(self): """ Test generic mkfs with exFAT """ if not self.exfat_avail: self.skipTest("skipping exFAT: not available") label = "label" self._test_ext_generic_mkfs("exfat", BlockDev.fs_exfat_get_info, label, None) def test_ext2_generic_mkfs(self): """ Test generic mkfs with ext2 """ label = "label" uuid = "8802574c-587b-43b9-a6be-9de77759d2c5" self._test_ext_generic_mkfs("ext2", BlockDev.fs_ext2_get_info, label, uuid, False) # clean the device succ = BlockDev.fs_clean(self.loop_dev) self.assertTrue(succ) options = BlockDev.FSMkfsOptions(label, uuid, False, False) # and try with a custom extra arg (we can get block size from the info) extra = BlockDev.ExtraArg("-b", "4096") succ = BlockDev.fs_mkfs(self.loop_dev, "ext2", options, [extra]) self.assertTrue(succ) fstype = BlockDev.fs_get_fstype (self.loop_dev) self.assertEqual(fstype, "ext2") info = BlockDev.fs_ext2_get_info(self.loop_dev) self.assertEqual(info.label, label) self.assertEqual(info.uuid, uuid) self.assertEqual(info.block_size, 4096) # try with -F, it doesn't really do anything with stdin closed so just a sanity check options = BlockDev.FSMkfsOptions(force=True) succ = BlockDev.fs_mkfs(self.loop_dev, "ext2", options) self.assertTrue(succ) def test_ext3_generic_mkfs(self): """ Test generic mkfs with ext3 """ label = "label" uuid = "8802574c-587b-43b9-a6be-9de77759d2c5" self._test_ext_generic_mkfs("ext3", BlockDev.fs_ext3_get_info, label, uuid, False) # now try with empty label and UUID label = "" uuid = "" self._test_ext_generic_mkfs("ext3", BlockDev.fs_ext3_get_info, label, uuid, False) def test_ext4_generic_mkfs(self): """ Test generic mkfs with ext4 """ label = "label" uuid = "8802574c-587b-43b9-a6be-9de77759d2c5" self._test_ext_generic_mkfs("ext4", BlockDev.fs_ext4_get_info, label, uuid, False) # now try with empty label and UUID label = "" uuid = "" self._test_ext_generic_mkfs("ext4", BlockDev.fs_ext4_get_info, label, uuid, False) def test_f2fs_generic_mkfs(self): """ Test generic mkfs with F2FS """ label = "label" self._test_ext_generic_mkfs("f2fs", BlockDev.fs_f2fs_get_info, label, None, True) # now try with empty label label = "" self._test_ext_generic_mkfs("f2fs", BlockDev.fs_f2fs_get_info, label, None, True) def test_nilfs2_generic_mkfs(self): """ Test generic mkfs with nilfs2 """ if not self.nilfs2_avail: self.skipTest("skipping NILFS2: not available") label = "label" self._test_ext_generic_mkfs("nilfs2", BlockDev.fs_nilfs2_get_info, label, None, True) # now try with empty label label = "" self._test_ext_generic_mkfs("nilfs2", BlockDev.fs_nilfs2_get_info, label, None, True) def test_ntfs_generic_mkfs(self): """ Test generic mkfs with NTFS """ if not self.ntfs_avail: self.skipTest("skipping NTFS: not available") label = "label" self._test_ext_generic_mkfs("ntfs", BlockDev.fs_ntfs_get_info, label, None) # now try with empty label label = "" self._test_ext_generic_mkfs("ntfs", BlockDev.fs_ntfs_get_info, label, None) def test_vfat_generic_mkfs(self): """ Test generic mkfs with vfat """ label = "LABEL" if self._vfat_version >= Version("4.2"): extra = [BlockDev.ExtraArg.new("--mbr=n", "")] else: extra = None self._test_ext_generic_mkfs("vfat", BlockDev.fs_vfat_get_info, label, None, False, extra) # now try with empty label label = "" self._test_ext_generic_mkfs("vfat", BlockDev.fs_vfat_get_info, label, None, False, extra) # now try with UUID self._test_ext_generic_mkfs("vfat", BlockDev.fs_vfat_get_info, None, "2E24-EC82", False, extra) def test_vfat_generic_mkfs_no_pt(self): """ Test generic mkfs with vfat and mbr=no """ options = BlockDev.FSMkfsOptions(no_pt=True) succ = BlockDev.fs_mkfs(self.loop_dev, "vfat", options, None) self.assertTrue(succ) fstype = BlockDev.fs_get_fstype (self.loop_dev) self.assertEqual(fstype, "vfat") # there should be no partition self.assertFalse(os.path.exists(self.loop_dev + "1")) def test_xfs_generic_mkfs(self): """ Test generic mkfs with XFS """ label = "label" uuid = "8802574c-587b-43b9-a6be-9de77759d2c5" def _xfs_info(device): with mounted(device, self.mount_dir): info = BlockDev.fs_xfs_get_info(device) return info self._test_ext_generic_mkfs("xfs", _xfs_info, label, uuid, True) # now try with empty label and UUID label = "" uuid = "" self._test_ext_generic_mkfs("xfs", _xfs_info, label, uuid, True) def test_btrfs_generic_mkfs(self): """ Test generic mkfs with Btrfs """ if not self.btrfs_avail: self.skipTest("skipping Btrfs: not available") label = "label" uuid = "8802574c-587b-43b9-a6be-9de77759d2c5" def _btrfs_info(device): with mounted(device, self.mount_dir): info = BlockDev.fs_btrfs_get_info(self.mount_dir) return info self._test_ext_generic_mkfs("btrfs", _btrfs_info, label, uuid, True) # now try with empty label and UUID label = "" uuid = "" self._test_ext_generic_mkfs("btrfs", _btrfs_info, label, uuid, True) def test_udf_generic_mkfs(self): """ Test generic mkfs with udf """ if not self.udf_avail: self.skipTest("skipping UDF: not available") label = "LABEL" self._test_ext_generic_mkfs("udf", BlockDev.fs_udf_get_info, label, None) # now try with empty label label = "" self._test_ext_generic_mkfs("udf", BlockDev.fs_udf_get_info, label, None, default_label="LinuxUDF") def test_generic_mkfs_no_options(self): """ Test that fs_mkfs works without options specified """ succ = BlockDev.fs_mkfs(self.loop_dev, "ext2") self.assertTrue(succ) info = BlockDev.fs_ext2_get_info(self.loop_dev) self.assertIsNotNone(info) self.assertFalse(info.label) # label should be empty by default def test_fail_generic_mkfs(self): """ Test that generic mkfs fails correctly with unknown/unsupported filesystem """ with self.assertRaisesRegex(GLib.GError, "Filesystem 'non-existing-fs' is not supported"): BlockDev.fs_mkfs(self.loop_dev, "non-existing-fs") def test_can_mkfs(self): """ Test checking whether mkfs is supported """ # lets pick a filesystem that supports everything and is always available # in the CI supported, flags, _util = BlockDev.fs_can_mkfs("ext2") self.assertTrue(supported) self.assertTrue(flags & BlockDev.FSMkfsOptionsFlags.DRY_RUN) self.assertTrue(flags & BlockDev.FSMkfsOptionsFlags.LABEL) self.assertTrue(flags & BlockDev.FSMkfsOptionsFlags.UUID) self.assertTrue(flags & BlockDev.FSMkfsOptionsFlags.NODISCARD) self.assertTrue(flags & BlockDev.FSMkfsOptionsFlags.FORCE) class GenericCheck(GenericTestCase): log = [] def _test_generic_check(self, mkfs_function, fstype=None): # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = mkfs_function(self.loop_dev, None) self.assertTrue(succ) self.log = [] # check for consistency (expected to be ok) succ = BlockDev.fs_check(self.loop_dev) self.assertTrue(succ) if fstype: succ = BlockDev.fs_check(self.loop_dev, fstype) self.assertTrue(succ) def _my_progress_func(self, task, status, completion, msg): self.assertTrue(isinstance(completion, int)) self.log.append(completion) def _verify_progress(self, log): # at least 2 members self.assertLessEqual(2, len(log)) # non-decreasing members self.assertTrue(all(x<=y for x, y in zip(log, log[1:]))) def test_ext4_generic_check(self): """Test generic check function with an ext4 file system""" self._test_generic_check(mkfs_function=BlockDev.fs_ext4_mkfs, fstype="ext4") def test_ext4_progress_check(self): """Test check function with an ext4 file system and progress reporting""" succ = BlockDev.utils_init_prog_reporting(self._my_progress_func) self.assertTrue(succ) self._test_generic_check(mkfs_function=BlockDev.fs_ext4_mkfs) self._verify_progress(self.log) succ = BlockDev.utils_init_prog_reporting(None) def test_xfs_generic_check(self): """Test generic check function with an ext4 file system""" self._test_generic_check(mkfs_function=BlockDev.fs_xfs_mkfs, fstype="xfs") def test_ntfs_generic_check(self): """Test generic check function with an ntfs file system""" if not self.ntfs_avail: self.skipTest("skipping NTFS: not available") self._test_generic_check(mkfs_function=BlockDev.fs_ntfs_mkfs, fstype="ntfs") def _check_fsck_f2fs_version(self): # if it can run -V to get version it can do the check ret, _out, _err = utils.run_command("fsck.f2fs -V") return ret == 0 def test_f2fs_generic_check(self): """Test generic check function with an f2fs file system""" if not self.f2fs_avail: self.skipTest("skipping F2FS: not available") if not self._check_fsck_f2fs_version(): with self.assertRaisesRegex(GLib.GError, "Too low version of fsck.f2fs. At least 1.11.0 required."): self._test_generic_check(mkfs_function=BlockDev.fs_f2fs_mkfs, fstype="f2fs") else: self._test_generic_check(mkfs_function=BlockDev.fs_f2fs_mkfs, fstype="f2fs") def test_nilfs2_generic_check(self): """Test generic check function with an nilfs2 file system""" if not self.nilfs2_avail: self.skipTest("skipping NILFS2: not available") with self.assertRaises(GLib.GError): # nilfs2 doesn't support check self._test_generic_check(mkfs_function=BlockDev.fs_nilfs2_mkfs, fstype="nilfs2") def test_exfat_generic_check(self): """Test generic check function with an exfat file system""" if not self.exfat_avail: self.skipTest("skipping exFAT: not available") self._test_generic_check(mkfs_function=BlockDev.fs_exfat_mkfs, fstype="exfat") def test_btrfs_generic_check(self): """Test generic check function with an btrfs file system""" if not self.btrfs_avail: self.skipTest("skipping Btrfs: not available") self._test_generic_check(mkfs_function=BlockDev.fs_btrfs_mkfs, fstype="btrfs") class GenericRepair(GenericTestCase): def _test_generic_repair(self, mkfs_function, fstype): # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = mkfs_function(self.loop_dev, None) self.assertTrue(succ) # repair (expected to succeed) succ = BlockDev.fs_repair(self.loop_dev) self.assertTrue(succ) # repair (expected to succeed) succ = BlockDev.fs_repair(self.loop_dev, fstype) self.assertTrue(succ) def test_ext4_generic_repair(self): """Test generic repair function with an ext4 file system""" self._test_generic_repair(mkfs_function=BlockDev.fs_ext4_mkfs, fstype="ext4") def test_xfs_generic_repair(self): """Test generic repair function with an xfs file system""" self._test_generic_repair(mkfs_function=BlockDev.fs_xfs_mkfs, fstype="xfs") def test_ntfs_generic_repair(self): """Test generic repair function with an ntfs file system""" if not self.ntfs_avail: self.skipTest("skipping NTFS: not available") self._test_generic_repair(mkfs_function=BlockDev.fs_ntfs_mkfs, fstype="ntfs") def test_f2fs_generic_repair(self): """Test generic repair function with an f2fs file system""" if not self.f2fs_avail: self.skipTest("skipping F2FS: not available") self._test_generic_repair(mkfs_function=BlockDev.fs_f2fs_mkfs, fstype="f2fs") def test_nilfs2_generic_repair(self): """Test generic repair function with an nilfs2 file system""" if not self.nilfs2_avail: self.skipTest("skipping NILFS2: not available") with self.assertRaises(GLib.GError): # nilfs2 doesn't support repair self._test_generic_repair(mkfs_function=BlockDev.fs_nilfs2_mkfs, fstype="nilfs2") def test_exfat_generic_repair(self): """Test generic repair function with an exfat file system""" if not self.exfat_avail: self.skipTest("skipping exFAT: not available") self._test_generic_repair(mkfs_function=BlockDev.fs_exfat_mkfs, fstype="exfat") def test_btrfs_generic_repair(self): """Test generic repair function with an btrfs file system""" if not self.btrfs_avail: self.skipTest("skipping Btrfs: not available") self._test_generic_repair(mkfs_function=BlockDev.fs_btrfs_mkfs, fstype="btrfs") class GenericSetLabel(GenericTestCase): def _test_generic_set_label(self, mkfs_function, fstype): # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = mkfs_function(self.loop_dev, None) self.assertTrue(succ) succ = BlockDev.fs_check_label(fstype, "new_label") self.assertTrue(succ) # set label (expected to succeed) succ = BlockDev.fs_set_label(self.loop_dev, "new_label") self.assertTrue(succ) # set label (expected to succeed) succ = BlockDev.fs_set_label(self.loop_dev, "new_label", fstype) self.assertTrue(succ) def test_ext4_generic_set_label(self): """Test generic set_label function with an ext4 file system""" self._test_generic_set_label(mkfs_function=BlockDev.fs_ext4_mkfs, fstype="ext4") def test_xfs_generic_set_label(self): """Test generic set_label function with a xfs file system""" self._test_generic_set_label(mkfs_function=BlockDev.fs_xfs_mkfs, fstype="xfs") def test_ntfs_generic_set_label(self): """Test generic set_label function with a ntfs file system""" if not self.ntfs_avail: self.skipTest("skipping NTFS: not available") self._test_generic_set_label(mkfs_function=BlockDev.fs_ntfs_mkfs, fstype="ntfs") def test_f2fs_generic_set_label(self): """Test generic set_label function with a f2fs file system""" if not self.f2fs_avail: self.skipTest("skipping F2FS: not available") with self.assertRaises(GLib.GError): # f2fs doesn't support relabeling self._test_generic_set_label(mkfs_function=BlockDev.fs_f2fs_mkfs, fstype="f2fs") def test_nilfs2_generic_set_label(self): """Test generic set_label function with a nilfs2 file system""" if not self.nilfs2_avail: self.skipTest("skipping NILFS2: not available") self._test_generic_set_label(mkfs_function=BlockDev.fs_nilfs2_mkfs, fstype="nilfs2") def test_exfat_generic_set_label(self): """Test generic set_label function with a exfat file system""" if not self.exfat_avail: self.skipTest("skipping exFAT: not available") self._test_generic_set_label(mkfs_function=BlockDev.fs_exfat_mkfs, fstype="exfat") def test_btrfs_generic_set_label(self): """Test generic set_label function with a btrfs file system""" if not self.btrfs_avail: self.skipTest("skipping btrfs: not available") self._test_generic_set_label(mkfs_function=BlockDev.fs_btrfs_mkfs, fstype="btrfs") def test_udf_generic_set_label(self): """Test generic set_label function with a udf file system""" if not self.udf_avail: self.skipTest("skipping udf: not available") self._test_generic_set_label(mkfs_function=BlockDev.fs_udf_mkfs, fstype="udf") class GenericSetUUID(GenericTestCase): def _test_generic_set_uuid(self, mkfs_function, fstype, test_uuid="4d7086c4-a4d3-432f-819e-73da03870df9", expected_uuid=None): # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = mkfs_function(self.loop_dev, None) self.assertTrue(succ) # set uuid (expected to succeed) succ = BlockDev.fs_set_uuid(self.loop_dev, test_uuid) self.assertTrue(succ) fs_uuid = check_output(["blkid", "-ovalue", "-sUUID", "-p", self.loop_dev]).decode().strip() if expected_uuid: self.assertEqual(fs_uuid, expected_uuid) else: self.assertEqual(fs_uuid, test_uuid) # set empty/random UUID succ = BlockDev.fs_set_uuid(self.loop_dev, None, fstype) self.assertTrue(succ) # check uuid format succ = BlockDev.fs_check_uuid(fstype, test_uuid) self.assertTrue(succ) def test_ext4_generic_set_uuid(self): """Test generic set_uuid function with an ext4 file system""" self._test_generic_set_uuid(mkfs_function=BlockDev.fs_ext4_mkfs, fstype="ext4") def test_xfs_generic_set_uuid(self): """Test generic set_uuid function with a xfs file system""" self._test_generic_set_uuid(mkfs_function=BlockDev.fs_xfs_mkfs, fstype="xfs") def test_ntfs_generic_set_uuid(self): """Test generic set_uuid function with a ntfs file system""" if not self.ntfs_avail: self.skipTest("skipping NTFS: not available") self._test_generic_set_uuid(mkfs_function=BlockDev.fs_ntfs_mkfs, fstype="ntfs", test_uuid="1C2716ED53F63962") def test_vfat_generic_set_uuid(self): """Test generic set_uuid function with a vfat file system""" if self._vfat_version < Version("4.2"): self.skipTest("dosfstools >= 4.2 needed to set UUID") def mkfs_vfat(device, options=None): if self._vfat_version >= Version("4.2"): return BlockDev.fs_vfat_mkfs(device, [BlockDev.ExtraArg.new("--mbr=n", "")]) else: return BlockDev.fs_vfat_mkfs(device, options) self._test_generic_set_uuid(mkfs_function=mkfs_vfat, fstype="vfat", test_uuid="2E24EC82", expected_uuid="2E24-EC82") def test_f2fs_generic_set_uuid(self): """Test generic set_uuid function with a f2fs file system""" if not self.f2fs_avail: self.skipTest("skipping F2FS: not available") with self.assertRaises(GLib.GError): # f2fs doesn't support setting UUID self._test_generic_set_uuid(mkfs_function=BlockDev.fs_f2fs_mkfs, fstype="f2fs") def test_nilfs2_generic_set_uuid(self): """Test generic set_uuid function with a nilfs2 file system""" if not self.nilfs2_avail: self.skipTest("skipping NILFS2: not available") self._test_generic_set_uuid(mkfs_function=BlockDev.fs_nilfs2_mkfs, fstype="nilfs2") def test_exfat_generic_set_uuid(self): """Test generic set_uuid function with a exfat file system""" if not self.exfat_avail: self.skipTest("skipping exFAT: not available") self._test_generic_set_uuid(mkfs_function=BlockDev.fs_exfat_mkfs, fstype="exfat", test_uuid="2E24EC82", expected_uuid="2E24-EC82") def test_btrfs_generic_set_uuid(self): """Test generic set_uuid function with a btrfs file system""" if not self.btrfs_avail: self.skipTest("skipping Btrfs: not available") self._test_generic_set_uuid(mkfs_function=BlockDev.fs_btrfs_mkfs, fstype="btrfs") def test_udf_generic_set_uuid(self): """Test generic set_uuid function with a udf file system""" if not self.udf_avail: self.skipTest("skipping UDF: not available") self._test_generic_set_uuid(mkfs_function=BlockDev.fs_udf_mkfs, fstype="udf", test_uuid="5fae9ade7938dfc8") class GenericResize(GenericTestCase): def _test_generic_resize(self, mkfs_function, fstype, size_delta=0, min_size=130*1024**2): # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = mkfs_function(self.loop_dev, None) self.assertTrue(succ) size = BlockDev.fs_get_size(self.loop_dev) # shrink succ = BlockDev.fs_resize(self.loop_dev, min_size) self.assertTrue(succ) new_size = BlockDev.fs_get_size(self.loop_dev) self.assertAlmostEqual(new_size, min_size, delta=size_delta) # resize to maximum size succ = BlockDev.fs_resize(self.loop_dev, 0, fstype) self.assertTrue(succ) new_size = BlockDev.fs_get_size(self.loop_dev) # should be back to original size self.assertAlmostEqual(new_size, size, delta=size_delta) def test_ext2_generic_resize(self): """Test generic resize function with an ext2 file system""" self._test_generic_resize(mkfs_function=BlockDev.fs_ext2_mkfs, fstype="ext2") def test_ext3_check_generic_resize(self): """Test generic resize function with an ext3 file system""" self._test_generic_resize(mkfs_function=BlockDev.fs_ext3_mkfs, fstype="ext3") def test_ext4_generic_resize(self): """Test generic resize function with an ext4 file system""" self._test_generic_resize(mkfs_function=BlockDev.fs_ext4_mkfs, fstype="ext4") def test_ntfs_generic_resize(self): """Test generic resize function with an ntfs file system""" if not self.ntfs_avail: self.skipTest("skipping NTFS: not available") def mkfs_prepare(drive): return BlockDev.fs_ntfs_mkfs(drive, None) and BlockDev.fs_repair(drive) def ntfs_size(drive): # man ntfsresize says "The filesystem size is set to be at least one # sector smaller" (maybe depending on alignment as well?), thus on a # loop device as in this test it is 4096 bytes smaller than requested BlockDev.fs_repair(drive) fi = BlockDev.fs_ntfs_get_info(drive) return fi.size + 4096 # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = mkfs_prepare(self.loop_dev) self.assertTrue(succ) size = ntfs_size(self.loop_dev) # shrink succ = BlockDev.fs_resize(self.loop_dev, 80 * 1024**2) self.assertTrue(succ) new_size = ntfs_size(self.loop_dev) self.assertEqual(new_size, 80 * 1024**2) # resize to maximum size succ = BlockDev.fs_resize(self.loop_dev, 0, "ntfs") self.assertTrue(succ) new_size = ntfs_size(self.loop_dev) # should be back to original size self.assertEqual(new_size, size) @tag_test(TestTags.UNSTABLE) @utils.required_plugins(("tools",)) def test_vfat_generic_resize(self): """Test generic resize function with a vfat file system""" def mkfs_vfat(device, options=None): if self._vfat_version >= Version("4.2"): if options: return BlockDev.fs_vfat_mkfs(device, options + [BlockDev.ExtraArg.new("--mbr=n", "")]) else: return BlockDev.fs_vfat_mkfs(device, [BlockDev.ExtraArg.new("--mbr=n", "")]) else: return BlockDev.fs_vfat_mkfs(device, options) self._test_generic_resize(mkfs_function=mkfs_vfat, size_delta=1024**2, fstype="vfat") def test_xfs_generic_resize(self): """Test generic resize function with an xfs file system""" lv = self._setup_lvm(vgname="libbd_fs_tests", lvname="generic_test", lvsize="350M") succ = BlockDev.fs_xfs_mkfs(lv, None) self.assertTrue(succ) with mounted(lv, self.mount_dir): fi = BlockDev.fs_xfs_get_info(lv) self.assertTrue(fi) self.assertEqual(fi.block_size * fi.block_count, 350 * 1024**2) # no change, nothing should happen with mounted(lv, self.mount_dir): succ = BlockDev.fs_resize(lv, 0) self.assertTrue(succ) with mounted(lv, self.mount_dir): fi = BlockDev.fs_xfs_get_info(lv) self.assertTrue(fi) self.assertEqual(fi.block_size * fi.block_count, 350 * 1024**2) # (still) impossible to shrink an XFS file system xfs_version = self._get_xfs_version() if xfs_version < Version("5.12"): with mounted(lv, self.mount_dir): with self.assertRaises(GLib.GError): succ = BlockDev.fs_resize(lv, 40 * 1024**2) self._lvresize("libbd_fs_tests", "generic_test", "400M") # should grow to 400 MiB (full size of the LV) with mounted(lv, self.mount_dir): succ = BlockDev.fs_resize(lv, 0) self.assertTrue(succ) with mounted(lv, self.mount_dir): fi = BlockDev.fs_xfs_get_info(lv) self.assertTrue(fi) self.assertEqual(fi.block_size * fi.block_count, 400 * 1024**2) self._lvresize("libbd_fs_tests", "generic_test", "450M") # grow just to 430 MiB with mounted(lv, self.mount_dir): succ = BlockDev.fs_resize(lv, 430 * 1024**2) self.assertTrue(succ) with mounted(lv, self.mount_dir): fi = BlockDev.fs_xfs_get_info(lv) self.assertTrue(fi) self.assertEqual(fi.block_size * fi.block_count, 430 * 1024**2) # should grow to 450 MiB (full size of the LV) with mounted(lv, self.mount_dir): succ = BlockDev.fs_resize(lv, 0, "xfs") self.assertTrue(succ) with mounted(lv, self.mount_dir): fi = BlockDev.fs_xfs_get_info(lv) self.assertTrue(fi) self.assertEqual(fi.block_size * fi.block_count, 450 * 1024**2) def _can_resize_f2fs(self): ret, out, _err = utils.run_command("resize.f2fs -V") if ret != 0: # we can't even check the version return False m = re.search(r"resize.f2fs ([\d\.]+)", out) if not m or len(m.groups()) != 1: raise RuntimeError("Failed to determine f2fs version from: %s" % out) version = Version(m.groups()[0]) # XXX resize works with f2fs-tools 1.15 but dump doesn't return version >= Version("1.12.0") and version < Version("1.15.0") def test_f2fs_generic_resize(self): """Verify that it is possible to resize an f2fs file system""" if not self.f2fs_avail: self.skipTest("skipping F2FS: not available") if not self._can_resize_f2fs(): self.skipTest("skipping F2FS: f2fs-tools version doesn't support resizing") else: self._test_generic_resize(mkfs_function=BlockDev.fs_f2fs_mkfs, fstype="f2fs") def test_nilfs2_generic_resize(self): """Test generic resize function with an nilfs2 file system""" if not self.nilfs2_avail: self.skipTest("skipping NILFS2: not available") self._test_generic_resize(mkfs_function=BlockDev.fs_nilfs2_mkfs, fstype="nilfs2") def test_exfat_generic_resize(self): """Test generic resize function with an exfat file system""" if not self.exfat_avail: self.skipTest("skipping exFAT: not available") # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = BlockDev.fs_exfat_mkfs(self.loop_dev, None) self.assertTrue(succ) # no resize support for exFAT with self.assertRaisesRegex(GLib.GError, "Resizing filesystem 'exfat' is not supported."): BlockDev.fs_resize(self.loop_dev, 80 * 1024**2) def test_btrfs_generic_resize(self): """Test generic resize function with an btrfs file system""" if not self.btrfs_avail: self.skipTest("skipping Btrfs: not available") temps1 = set([temp for temp in os.listdir("/tmp") if temp.startswith("blockdev.")]) self._test_generic_resize(mkfs_function=BlockDev.fs_btrfs_mkfs, min_size=300*1024**2, fstype="btrfs") temps2 = set([temp for temp in os.listdir("/tmp") if temp.startswith("blockdev.")]) # make sure we didn't leak a temporary mount directory self.assertFalse(temps2 - temps1) def test_btrfs_generic_resize_mounted(self): """Test generic resize function with a mounted btrfs file system""" if not self.btrfs_avail: self.skipTest("skipping Btrfs: not available") # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = BlockDev.fs_btrfs_mkfs(self.loop_dev, None) self.assertTrue(succ) size = BlockDev.fs_get_size(self.loop_dev) with mounted(self.loop_dev, self.mount_dir): # shrink succ = BlockDev.fs_resize(self.loop_dev, 300*1024**2) self.assertTrue(succ) new_size = BlockDev.fs_get_size(self.loop_dev) self.assertAlmostEqual(new_size, 300*1024**2) # resize to maximum size succ = BlockDev.fs_resize(self.loop_dev, 0, "btrfs") self.assertTrue(succ) new_size = BlockDev.fs_get_size(self.loop_dev) # should be back to original size self.assertAlmostEqual(new_size, size) def test_udf_generic_resize(self): """Test generic resize function with an udf file system""" if not self.udf_avail: self.skipTest("skipping UDF: not available") # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = BlockDev.fs_udf_mkfs(self.loop_dev, None) self.assertTrue(succ) # no resize support for UDF with self.assertRaisesRegex(GLib.GError, "Resizing filesystem 'udf' is not supported."): BlockDev.fs_resize(self.loop_dev, 80 * 1024**2) class GenericGetFreeSpace(GenericTestCase): def _test_get_free_space(self, mkfs_function, fstype, size_delta=0): # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = mkfs_function(self.loop_dev, None) self.assertTrue(succ) size = BlockDev.fs_get_size(self.loop_dev) free = BlockDev.fs_get_free_space(self.loop_dev) self.assertNotEqual(free, 0) self.assertLessEqual(free, size) size = BlockDev.fs_get_size(self.loop_dev, fstype) free = BlockDev.fs_get_free_space(self.loop_dev, fstype) self.assertNotEqual(free, 0) self.assertLessEqual(free, size) def test_ext2_get_free_space(self): """Test generic resize function with an ext2 file system""" self._test_get_free_space(mkfs_function=BlockDev.fs_ext2_mkfs, fstype="ext2") def test_ext3_check_get_free_space(self): """Test generic resize function with an ext3 file system""" self._test_get_free_space(mkfs_function=BlockDev.fs_ext3_mkfs, fstype="ext3") def test_ext4_get_free_space(self): """Test generic resize function with an ext4 file system""" self._test_get_free_space(mkfs_function=BlockDev.fs_ext4_mkfs, fstype="ext4") def test_ntfs_get_free_space(self): """Test generic resize function with an ntfs file system""" if not self.ntfs_avail: self.skipTest("skipping NTFS: not available") self._test_get_free_space(mkfs_function=BlockDev.fs_ntfs_mkfs, fstype="ntfs") def test_vfat_get_free_space(self): """Test generic resize function with a vfat file system""" def mkfs_vfat(device, options=None): if self._vfat_version >= Version("4.2"): if options: return BlockDev.fs_vfat_mkfs(device, options + [BlockDev.ExtraArg.new("--mbr=n", "")]) else: return BlockDev.fs_vfat_mkfs(device, [BlockDev.ExtraArg.new("--mbr=n", "")]) else: return BlockDev.fs_vfat_mkfs(device, options) self._test_get_free_space(mkfs_function=mkfs_vfat, fstype="vfat") def test_nilfs2_get_free_space(self): """Test generic resize function with an nilfs2 file system""" if not self.nilfs2_avail: self.skipTest("skipping NILFS2: not available") self._test_get_free_space(mkfs_function=BlockDev.fs_nilfs2_mkfs, fstype="nilfs2") def test_btrfs_get_free_space(self): """Test generic resize function with an btrfs file system""" if not self.btrfs_avail: self.skipTest("skipping Btrfs: not available") self._test_get_free_space(mkfs_function=BlockDev.fs_btrfs_mkfs, fstype="btrfs") def test_udf_get_free_space(self): """Test generic resize function with an udf file system""" if not self.udf_avail: self.skipTest("skipping UDF: not available") # clean the device succ = BlockDev.fs_clean(self.loop_dev) self.assertTrue(succ) succ = BlockDev.fs_udf_mkfs(self.loop_dev) self.assertTrue(succ) with self.assertRaisesRegex(GLib.GError, "Getting free space on filesystem 'udf' is not supported."): BlockDev.fs_get_free_space(self.loop_dev) class GenericGetMinSize(GenericTestCase): def _test_get_min_size(self, mkfs_function, fstype): # clean the device succ = BlockDev.fs_clean(self.loop_dev) succ = mkfs_function(self.loop_dev, None) self.assertTrue(succ) size = BlockDev.fs_get_min_size(self.loop_dev) self.assertNotEqual(size, 0) def test_ext2_test_get_min_size(self): """Test generic min size function with an ext2 file system""" self._test_get_min_size(mkfs_function=BlockDev.fs_ext2_mkfs, fstype="ext2") def test_ext3_check_test_get_min_size(self): """Test generic min size function with an ext3 file system""" self._test_get_min_size(mkfs_function=BlockDev.fs_ext3_mkfs, fstype="ext3") def test_ext4_test_get_min_size(self): """Test generic min size function with an ext4 file system""" self._test_get_min_size(mkfs_function=BlockDev.fs_ext4_mkfs, fstype="ext4") def test_ntfs_test_get_min_size(self): """Test generic min size function with an ntfs file system""" if not self.ntfs_avail: self.skipTest("skipping NTFS: not available") self._test_get_min_size(mkfs_function=BlockDev.fs_ntfs_mkfs, fstype="ntfs") def test_xfs_get_get_min_size(self): """Test generic min size function with a xfs file system""" # clean the device succ = BlockDev.fs_clean(self.loop_dev) self.assertTrue(succ) succ = BlockDev.fs_xfs_mkfs(self.loop_dev) self.assertTrue(succ) with self.assertRaisesRegex(GLib.GError, "Getting minimum size of filesystem 'xfs' is not supported."): BlockDev.fs_get_min_size(self.loop_dev) class FSFreezeTest(GenericTestCase): def _clean_up(self): try: BlockDev.fs_unfreeze(self.loop_dev) except: pass BlockDev.fs_wipe(self.loop_dev, True) super(FSFreezeTest, self)._clean_up() def test_freeze_xfs(self): """ Test basic freezing and un-freezing with XFS """ succ = BlockDev.fs_xfs_mkfs(self.loop_dev, None) self.assertTrue(succ) # try to freeze with non-existing mountpoint with self.assertRaises(GLib.GError): BlockDev.fs_freeze("/not/a/mountpoint") tmp = tempfile.mkdtemp(prefix="libblockdev.", suffix="freeze_test") self.addCleanup(os.rmdir, tmp) self.addCleanup(utils.umount, self.loop_dev) succ = BlockDev.fs_mount(self.loop_dev, tmp, "xfs", None) self.assertTrue(succ) self.assertTrue(os.path.ismount(tmp)) succ = BlockDev.fs_freeze(tmp) self.assertTrue(succ) # try to freeze again (should fail) with self.assertRaises(GLib.GError): BlockDev.fs_freeze(tmp) # and unfreeze succ = BlockDev.fs_unfreeze(tmp) self.assertTrue(succ) def test_freeze_vfat(self): """ Test basic freezing and un-freezing with FAT """ succ = BlockDev.fs_vfat_mkfs(self.loop_dev, None) self.assertTrue(succ) tmp = tempfile.mkdtemp(prefix="libblockdev.", suffix="freeze_test") self.addCleanup(os.rmdir, tmp) self.addCleanup(utils.umount, self.loop_dev) succ = BlockDev.fs_mount(self.loop_dev, tmp, "vfat", None) self.assertTrue(succ) self.assertTrue(os.path.ismount(tmp)) # FAT doesn't support freezing with self.assertRaises(GLib.GError): BlockDev.fs_freeze(tmp) class SupportedFilesystemsTest(GenericNoDevTestCase): def test_supported_filesystems(self): filesystems = BlockDev.fs_supported_filesystems() self.assertListEqual(filesystems, ["ext2", "ext3", "ext4", "xfs", "vfat", "ntfs", "f2fs", "nilfs2", "exfat", "btrfs", "udf"])