Merge pull request #1173 from vojtechtrefny/3.9-devel_btrfs-tests

Add a btrfs storage test and example

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)

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)

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 *

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)