[blivet 6/6] Work on devicelibs.btrfs methods that require that the device be mounted.

mulhern amulhern at redhat.com
Wed Dec 11 15:06:18 UTC 2013


Make a small subclass that handles mounting the device before tests
and unmounting after tests. I pondered making this a more general class
that can be shared among the different filesystem tests, but there are
some things about mounting the filesystem that are somewhat btrfs specific.

Otherwise just tests and comments.

subvolume list is pretty configurable and its output can change somewhat
depending on the version installed. The tests that use it are written
to be robust to small changes, but could probably be strengthened a bit
as btrfs settles down.

Signed-off-by: mulhern <amulhern at redhat.com>
---
 blivet/devicelibs/btrfs.py          |   1 +
 tests/devicelibs_test/btrfs_test.py | 121 ++++++++++++++++++++++++++++++++++--
 2 files changed, 118 insertions(+), 4 deletions(-)

diff --git a/blivet/devicelibs/btrfs.py b/blivet/devicelibs/btrfs.py
index 869a170..bca1242 100644
--- a/blivet/devicelibs/btrfs.py
+++ b/blivet/devicelibs/btrfs.py
@@ -83,6 +83,7 @@ def create_volume(devices, label=None, data=None, metadata=None):
 # remove device
 
 def create_subvolume(mountpoint, name):
+    """Create a subvolume named name below mountpoint mountpoint."""
     if not os.path.ismount(mountpoint):
         raise ValueError("volume not mounted")
 
diff --git a/tests/devicelibs_test/btrfs_test.py b/tests/devicelibs_test/btrfs_test.py
index a6f8a75..fdfb59b 100755
--- a/tests/devicelibs_test/btrfs_test.py
+++ b/tests/devicelibs_test/btrfs_test.py
@@ -1,11 +1,57 @@
 #!/usr/bin/python
 import baseclass
 import os
+import os.path
+import subprocess
+import tempfile
 import unittest
 
 import blivet.devicelibs.btrfs as btrfs
+import blivet.util as util
+
+class BTRFSMountDevice(baseclass.DevicelibsTestCase):
+    """A superclass that mounts and unmounts the filesystem.
+       It must create the filesystem on its chosen devices before mounting.
+       It always mounts the filesystem using self.device at self.mountpoint.
+    """
+    def __init__(self, *args, **kwargs):
+        baseclass.DevicelibsTestCase.__init__(self, *args, **kwargs)
+        self.device = None
+        self.mountpoint = None
+
+    def setUp(self):
+        """After the DevicelibsTestCase setup, creates the filesystem on both
+           devices and mounts on a tmp directory.
+
+           Chooses the device to specify to mount command arbitrarily.
+        """
+        baseclass.DevicelibsTestCase.setUp(self)
+
+        btrfs.create_volume(self._loopMap.values())
+        self.device = self._loopMap.values()[0]
+
+        self.mountpoint = tempfile.mkdtemp()
+        rc = subprocess.call(["mount", self.device, self.mountpoint])
+        if rc:
+            raise OSError, "mount failed to mount device %s" % device
 
-class BTRFSAsRootTestCase(baseclass.DevicelibsTestCase):
+    def tearDown(self):
+        """Before the DevicelibsTestCase cleanup unmount the device and
+           remove the temporary mountpoint.
+        """
+        proc = subprocess.Popen(["umount", self.device])
+        while True:
+            (out, err) = proc.communicate()
+            if proc.returncode is not None:
+                rc = proc.returncode
+                break
+        if rc:
+            raise OSError, "failed to unmount device %s" % self.device
+
+        os.rmdir(self.mountpoint)
+        baseclass.DevicelibsTestCase.tearDown(self)
+
+class BTRFSAsRootTestCase1(baseclass.DevicelibsTestCase):
 
     @unittest.skipUnless(os.geteuid() == 0, "requires root privileges")
     def testUnmountedBTRFS(self):
@@ -61,16 +107,83 @@ class BTRFSAsRootTestCase(baseclass.DevicelibsTestCase):
         self.assertEqual(btrfs.summarize_filesystem(_LOOP_DEV0)["label"], "none")
         self.assertEqual(btrfs.summarize_filesystem(_LOOP_DEV0)["num_devices"], "2")
 
-
         self.assertEqual(len(btrfs.list_devices(_LOOP_DEV0)), 2)
         self.assertIn(_LOOP_DEV1,
            [ dev["path"] for dev in btrfs.list_devices(_LOOP_DEV0) ])
         self.assertIn(_LOOP_DEV0,
            [ dev["path"] for dev in btrfs.list_devices(_LOOP_DEV1) ])
 
+class BTRFSAsRootTestCase2(BTRFSMountDevice):
+    """Tests which require mounting the device."""
+
+    @unittest.skipUnless(os.geteuid() == 0, "requires root privileges")
+    def testSubvolume(self):
+        """Tests which focus on subvolumes."""
+        _LOOP_DEV0 = self._loopMap[self._LOOP_DEVICES[0]]
+        _LOOP_DEV1 = self._loopMap[self._LOOP_DEVICES[1]]
+
+        # no subvolumes yet
+        self.assertEqual(btrfs.list_subvolumes(self.mountpoint), [])
+
+        # the default subvolume is the root subvolume
+        self.assertEqual(btrfs.get_default_subvolume(self.mountpoint),
+           btrfs.MAIN_VOLUME_ID)
+
+        # a new subvolume can be added succesfully below the mountpoint
+        self.assertEqual(btrfs.create_subvolume(self.mountpoint, "SV1"), 0)
+
+        # expect one subvolume
+        subvolumes = btrfs.list_subvolumes(self.mountpoint)
+        self.assertEqual(len(subvolumes), 1)
+        self.assertEqual(subvolumes[0]['path'], 'SV1')
+        self.assertEqual(subvolumes[0]['parent'], btrfs.MAIN_VOLUME_ID)
+
+        # the same subvolume can be deleted
+        self.assertEqual(btrfs.delete_subvolume(self.mountpoint, "SV1"), 0)
+
+        # deleted subvolume is no longer present
+        subvolumes = btrfs.list_subvolumes(self.mountpoint)
+        self.assertNotIn("SV1", [v['path'] for v in subvolumes])
+
+        # two distinct subvolumes, so both should be present
+        self.assertEqual(btrfs.create_subvolume(self.mountpoint, "SV1"), 0)
+        self.assertEqual(btrfs.create_subvolume(self.mountpoint, "SV2"), 0)
+        subvolumes = btrfs.list_subvolumes(self.mountpoint)
+        self.assertIn("SV1", [v['path'] for v in subvolumes])
+        self.assertIn("SV2", [v['path'] for v in subvolumes])
+
+        # we can remove one subvolume
+        self.assertEqual(btrfs.delete_subvolume(self.mountpoint, "SV1"), 0)
+        subvolumes = btrfs.list_subvolumes(self.mountpoint)
+        self.assertNotIn("SV1", [v['path'] for v in subvolumes])
+
+        # if the subvolume is already gone,  an error is raised by btrfs
+        self.assertRaisesRegexp(btrfs.BTRFSError,
+           "1",
+           btrfs.delete_subvolume,
+           self.mountpoint, "SV1")
+
+        # if the subvolume is already there, an error is raise by btrfs
+        self.assertRaisesRegexp(btrfs.BTRFSError,
+           "1",
+           btrfs.create_subvolume,
+           self.mountpoint, "SV2")
+
+        # if we create SV1 once again it's back
+        self.assertEqual(btrfs.create_subvolume(self.mountpoint, "SV1"), 0)
+        subvolumes = btrfs.list_subvolumes(self.mountpoint)
+        self.assertIn("SV1", [v['path'] for v in subvolumes])
+
+        # we can create an additional subvolume beneath SV1
+        self.assertEqual(btrfs.create_subvolume(os.path.join(self.mountpoint, "SV1"), "SV1.1"), 0)
+        subvolumes = btrfs.list_subvolumes(self.mountpoint)
+        self.assertEqual(len([v for v in subvolumes if v['path'].find("SV1.1") != -1]), 1)
+
+
 def suite():
-    return unittest.TestSuite(
-       unittest.TestLoader().loadTestsFromTestCase(BTRFSAsRootTestCase))
+    suite1 = unittest.TestLoader().loadTestsFromTestCase(BTRFSAsRootTestCase1)
+    suite1 = unittest.TestLoader().loadTestsFromTestCase(BTRFSAsRootTestCase2)
+    return unittest.TestSuite([suite1, suite2])
 
 
 if __name__ == "__main__":
-- 
1.8.3.1



More information about the anaconda-patches mailing list