diff --git a/examples/btrfs.py b/examples/btrfs.py new file mode 100644 index 000000000..0cd2e8c62 --- /dev/null +++ b/examples/btrfs.py @@ -0,0 +1,40 @@ +import os + +import blivet +from blivet.size import Size +from blivet.util import set_up_logging, create_sparse_tempfile + +set_up_logging() +b = blivet.Blivet() # create an instance of Blivet (don't add system devices) + +# create a disk image file on which to create new devices +disk1_file = create_sparse_tempfile("disk1", Size("100GiB")) +b.disk_images["disk1"] = disk1_file + +b.reset() + +try: + disk1 = b.devicetree.get_device_by_name("disk1") + + b.initialize_disk(disk1) + + part = b.new_partition(size=Size("50GiB"), fmt_type="btrfs") + b.create_device(part) + + # allocate the partitions (decide where and on which disks they'll reside) + blivet.partitioning.do_partitioning(b) + + # new btrfs volume + vol = b.new_btrfs(parents=[part]) + b.create_device(vol) + + # new btrfs subvolume + dev = b.new_btrfs_sub_volume(name="subvol1", parents=[vol]) + b.create_device(dev) + + # write the new partitions to disk and format them as specified + b.do_it() + print(b.devicetree) +finally: + b.devicetree.teardown_disk_images() + os.unlink(disk1_file) diff --git a/tests/run_tests.py b/tests/run_tests.py index a49f6736e..e52105c29 100644 --- a/tests/run_tests.py +++ b/tests/run_tests.py @@ -5,10 +5,12 @@ import argparse import dbus import os +import pdb import re import six import subprocess import sys +import traceback import unittest import yaml @@ -136,6 +138,19 @@ def _get_tests_from_suite(test_suite, extracted_tests): return extracted_tests +class DebugTestResult(unittest.TextTestResult): + + def addError(self, test, err): # pylint: disable=redefined-outer-name + traceback.print_exception(*err) + pdb.post_mortem(err[2]) + super(DebugTestResult, self).addError(test, err) + + def addFailure(self, test, err): # pylint: disable=redefined-outer-name + traceback.print_exception(*err) + pdb.post_mortem(err[2]) + super(DebugTestResult, self).addFailure(test, err) + + if __name__ == '__main__': testdir = os.path.abspath(os.path.dirname(__file__)) projdir = os.path.abspath(os.path.normpath(os.path.join(testdir, '..'))) @@ -156,6 +171,8 @@ def _get_tests_from_suite(test_suite, extracted_tests): argparser = argparse.ArgumentParser(description="Blivet test suite") argparser.add_argument("testname", nargs="*", help="name of test class or method (e. g. 'devices_test' or 'formats_test.fs_test.Ext2FSTestCase'") + argparser.add_argument("-p", "--pdb", dest="pdb", help="run pdb after a failed test", action="store_true") + argparser.add_argument("-s", "--stop", dest="stop", help="stop executing after first failed test", action="store_true") args = argparser.parse_args() testdir = os.path.abspath(os.path.dirname(__file__)) @@ -196,7 +213,12 @@ def _get_tests_from_suite(test_suite, extracted_tests): suite.addTest(test) - result = unittest.TextTestRunner(verbosity=2).run(suite) + if args.pdb: + runner = unittest.TextTestRunner(verbosity=2, failfast=args.stop, resultclass=DebugTestResult) + else: + runner = unittest.TextTestRunner(verbosity=2, failfast=args.stop) + + result = runner.run(suite) if result.wasSuccessful(): sys.exit(0) diff --git a/tests/storage_tests/devices_test/__init__.py b/tests/storage_tests/devices_test/__init__.py index bc03efa44..8ef97488f 100644 --- a/tests/storage_tests/devices_test/__init__.py +++ b/tests/storage_tests/devices_test/__init__.py @@ -1,3 +1,4 @@ +from .btrfs_test import * from .lvm_test import * from .partition_test import * from .stratis_test import * diff --git a/tests/storage_tests/devices_test/btrfs_test.py b/tests/storage_tests/devices_test/btrfs_test.py new file mode 100644 index 000000000..f7adbd782 --- /dev/null +++ b/tests/storage_tests/devices_test/btrfs_test.py @@ -0,0 +1,150 @@ +import os +import unittest + +from ..storagetestcase import StorageTestCase + +import blivet +from blivet.devices.btrfs import BTRFSVolumeDevice + + +class BtrfsTestCase(StorageTestCase): + + volname = "blivetTestBtrfsVolume" + + @classmethod + def setUpClass(cls): + unavailable_deps = BTRFSVolumeDevice.unavailable_type_dependencies() + if unavailable_deps: + dep_str = ", ".join([d.name for d in unavailable_deps]) + raise unittest.SkipTest("some unavailable dependencies required for this test: %s" % dep_str) + + def setUp(self): + super().setUp() + + # allow automounting to get btrfs information + blivet.flags.flags.auto_dev_updates = True + + disks = [os.path.basename(vdev) for vdev in self.vdevs] + self.storage = blivet.Blivet() + self.storage.exclusive_disks = disks + self.storage.reset() + + # make sure only the targetcli disks are in the devicetree + for disk in self.storage.disks: + self.assertTrue(disk.path in self.vdevs) + self.assertIsNone(disk.format.type) + self.assertFalse(disk.children) + + def _clean_up(self): + self.storage.reset() + for disk in self.storage.disks: + if disk.path not in self.vdevs: + raise RuntimeError("Disk %s found in devicetree but not in disks created for tests" % disk.name) + self.storage.recursive_remove(disk) + + self.storage.do_it() + + # reset back to default + blivet.flags.flags.auto_dev_updates = False + + return super()._clean_up() + + def test_btrfs_basic(self): + disk = self.storage.devicetree.get_device_by_path(self.vdevs[0]) + self.assertIsNotNone(disk) + + self.storage.initialize_disk(disk) + + part = self.storage.new_partition(size=blivet.size.Size("1 GiB"), fmt_type="btrfs", + parents=[disk]) + self.storage.create_device(part) + + blivet.partitioning.do_partitioning(self.storage) + + vol = self.storage.new_btrfs(name=self.volname, parents=[part]) + self.storage.create_device(vol) + + sub = self.storage.new_btrfs_sub_volume(parents=[vol], name="blivetTestSubVol") + self.storage.create_device(sub) + + self.storage.do_it() + self.storage.reset() + self.storage.reset() + + disk = self.storage.devicetree.get_device_by_path(self.vdevs[0]) + self.assertIsNotNone(disk) + + part = self.storage.devicetree.get_device_by_name(os.path.basename(self.vdevs[0]) + "1") + self.assertIsNotNone(part) + self.assertIsInstance(part, blivet.devices.PartitionDevice) + self.assertIsNotNone(part.format) + self.assertEqual(part.format.type, "btrfs") + + vol = self.storage.devicetree.get_device_by_name(self.volname) + self.assertIsNotNone(vol) + self.assertIsInstance(vol, blivet.devices.BTRFSVolumeDevice) + self.assertIsNotNone(vol.format) + self.assertEqual(vol.format.type, "btrfs") + self.assertEqual(vol.format.container_uuid, vol.uuid) + self.assertEqual(len(vol.parents), 1) + self.assertEqual(vol.parents[0].name, part.name) + + sub = self.storage.devicetree.get_device_by_name("blivetTestSubVol") + self.assertIsNotNone(sub) + self.assertIsInstance(sub, blivet.devices.BTRFSSubVolumeDevice) + self.assertIsNotNone(sub.format) + self.assertEqual(sub.format.type, "btrfs") + self.assertEqual(sub.volume, vol) + self.assertEqual(len(sub.parents), 1) + self.assertEqual(sub.parents[0], vol) + self.assertEqual(sub.size, vol.size) + self.assertEqual(sub.format.container_uuid, vol.uuid) + self.assertEqual(sub.format.subvolspec, sub.name) + + def _test_btrfs_raid(self, raid_level): + disk1 = self.storage.devicetree.get_device_by_path(self.vdevs[0]) + self.assertIsNotNone(disk1) + self.storage.initialize_disk(disk1) + + disk2 = self.storage.devicetree.get_device_by_path(self.vdevs[1]) + self.assertIsNotNone(disk2) + self.storage.initialize_disk(disk2) + + part1 = self.storage.new_partition(size=blivet.size.Size("1 GiB"), fmt_type="btrfs", + parents=[disk1]) + self.storage.create_device(part1) + + part2 = self.storage.new_partition(size=blivet.size.Size("1 GiB"), fmt_type="btrfs", + parents=[disk2]) + self.storage.create_device(part2) + + blivet.partitioning.do_partitioning(self.storage) + + vol = self.storage.new_btrfs(name=self.volname, parents=[part1, part2], + data_level=raid_level, metadata_level=raid_level) + self.storage.create_device(vol) + + sub = self.storage.new_btrfs_sub_volume(parents=[vol], name="blivetTestSubVol") + self.storage.create_device(sub) + + self.storage.do_it() + self.storage.reset() + self.storage.reset() + + vol = self.storage.devicetree.get_device_by_name(self.volname) + self.assertIsNotNone(vol) + self.assertIsInstance(vol, blivet.devices.BTRFSVolumeDevice) + self.assertIsNotNone(vol.format) + self.assertEqual(vol.format.type, "btrfs") + self.assertEqual(vol.format.container_uuid, vol.uuid) + self.assertEqual(len(vol.parents), 2) + self.assertCountEqual([p.name for p in vol.parents], [part1.name, part2.name]) + + def test_btrfs_raid_single(self): + self._test_btrfs_raid(blivet.devicelibs.raid.Single) + + def test_btrfs_raid_raid0(self): + self._test_btrfs_raid(blivet.devicelibs.raid.RAID0) + + def test_btrfs_raid_raid1(self): + self._test_btrfs_raid(blivet.devicelibs.raid.RAID1)