Add StratisCertify tests for snapshot revert

stratis_cli_cert.py

@@ -1104,6 +1104,120 @@ def test_filesystem_snapshot(self):
             True,
         )
 
+    @skip(_skip_condition(1))
+    def test_filesystem_snapshot_cancel_revert(self):
+        """
+        Test canceling a revert of a filesystem snapshot.
+        """
+        pool_name = make_test_pool(StratisCliCertify.DISKS[0:1])
+        filesystem_name = make_test_filesystem(pool_name)
+        snapshot_name = fs_n()
+        self._unittest_command(
+            [
+                _STRATIS_CLI,
+                "filesystem",
+                "snapshot",
+                pool_name,
+                filesystem_name,
+                snapshot_name,
+            ],
+            0,
+            True,
+            True,
+        )
+        self._unittest_command(
+            [
+                _STRATIS_CLI,
+                "filesystem",
+                "schedule-revert",
+                pool_name,
+                snapshot_name,
+            ],
+            0,
+            True,
+            True,
+        )
+        self._unittest_command(
+            [
+                _STRATIS_CLI,
+                "filesystem",
+                "cancel-revert",
+                pool_name,
+                snapshot_name,
+            ],
+            0,
+            True,
+            True,
+        )
+
+    @skip(_skip_condition(1))
+    def test_filesystem_snapshot_schedule_revert(self):
+        """
+        Test scheduling a revert of a filesystem snapshot.
+        """
+        pool_name = make_test_pool(StratisCliCertify.DISKS[0:1])
+        filesystem_name = make_test_filesystem(pool_name)
+        snapshot_name = fs_n()
+        self._unittest_command(
+            [
+                _STRATIS_CLI,
+                "filesystem",
+                "snapshot",
+                pool_name,
+                filesystem_name,
+                snapshot_name,
+            ],
+            0,
+            True,
+            True,
+        )
+        self._unittest_command(
+            [
+                _STRATIS_CLI,
+                "filesystem",
+                "schedule-revert",
+                pool_name,
+                snapshot_name,
+            ],
+            0,
+            True,
+            True,
+        )
+
+    @skip(_skip_condition(1))
+    def test_filesystem_snapshot_schedule_revert_noorigin_fail(self):
+        """
+        Test scheduling a revert of a filesystem with no origin, which should fail.
+        """
+        pool_name = make_test_pool(StratisCliCertify.DISKS[0:1])
+        filesystem_name = make_test_filesystem(pool_name)
+        snapshot_name = fs_n()
+        self._unittest_command(
+            [
+                _STRATIS_CLI,
+                "filesystem",
+                "snapshot",
+                pool_name,
+                filesystem_name,
+                snapshot_name,
+            ],
+            0,
+            True,
+            True,
+        )
+        self._unittest_command(
+            [
+                _STRATIS_CLI,
+                "filesystem",
+                "schedule-revert",
+                pool_name,
+                filesystem_name,
+            ],
+            1,
+            False,
+            True,
+        )
+
     @skip(_skip_condition(1))
     def test_filesystem_snapshot_destroy_filesystem(self):
         """

stratisd_cert.py

@@ -205,9 +205,7 @@ def _unittest_set_property(
         :type exception_name: NoneType or str
         """
         try:
-            StratisDbus.pool_set_property(
-                object_path, param_iface, dbus_param, dbus_value
-            )
+            StratisDbus.set_property(object_path, param_iface, dbus_param, dbus_value)
 
         except dbus.exceptions.DBusException as err:
             self.assertEqual(err.get_dbus_name(), exception_name)
@@ -948,6 +946,61 @@ def test_filesystem_snapshot(self):
             StratisDbus.fs_snapshot(pool_path, fs_path, snapshot_name), dbus.UInt16(0)
         )
 
+    @skip(_skip_condition(1))
+    def test_filesystem_snapshot_schedule_revert(self):
+        """
+        Test scheduling a revert of a filesystem snapshot.
+        """
+        pool_name = p_n()
+        pool_path, _ = make_test_pool(pool_name, StratisCertify.DISKS[0:1])
+
+        fs_name = fs_n()
+        fs_path = make_test_filesystem(pool_path, fs_name)
+
+        snapshot_name = fs_n()
+
+        ((_, snapshot_path), _, _) = StratisDbus.fs_snapshot(
+            pool_path, fs_path, snapshot_name
+        )
+
+        StratisDbus.set_property(
+            snapshot_path,
+            StratisDbus.FS_IFACE,
+            "MergeScheduled",
+            dbus.Boolean(True),
+        )
+
+    @skip(_skip_condition(1))
+    def test_filesystem_snapshot_cancel_revert(self):
+        """
+        Test canceling a revert of a filesystem snapshot.
+        """
+        pool_name = p_n()
+        pool_path, _ = make_test_pool(pool_name, StratisCertify.DISKS[0:1])
+
+        fs_name = fs_n()
+        fs_path = make_test_filesystem(pool_path, fs_name)
+
+        snapshot_name = fs_n()
+
+        ((_, snapshot_path), _, _) = StratisDbus.fs_snapshot(
+            pool_path, fs_path, snapshot_name
+        )
+
+        StratisDbus.set_property(
+            snapshot_path,
+            StratisDbus.FS_IFACE,
+            "MergeScheduled",
+            dbus.Boolean(True),
+        )
+
+        StratisDbus.set_property(
+            snapshot_path,
+            StratisDbus.FS_IFACE,
+            "MergeScheduled",
+            dbus.Boolean(False),
+        )
+
     @skip(_skip_condition(1))
     def test_filesystem_snapshot_destroy_filesystem(self):
         """

testlib/dbus.py

@@ -363,14 +363,15 @@ def pool_destroy(pool_name):
     def fs_list():
         """
         Query the file systems
-        :return: A dict,  Key being the fs name, the value being the pool name
-        :rtype: dict mapping str to str
+        :return: A dict; key being a tuple of the object path, the fs name,
+                 and the origin D-Bus; the value being the pool name
+        :rtype: A dict of str * str * tuple -> str
         """
         objects = StratisDbus.get_managed_objects().items()
 
         fs_objects = [
-            obj_data[StratisDbus._FS_IFACE]
-            for _, obj_data in objects
+            (obj_path, obj_data[StratisDbus._FS_IFACE])
+            for obj_path, obj_data in objects
             if StratisDbus._FS_IFACE in obj_data
             and obj_data[StratisDbus._FS_IFACE]["Name"].startswith(_TEST_PREF)
         ]
@@ -383,8 +384,10 @@ def fs_list():
         }
 
         return {
-            fs_object["Name"]: pool_path_to_name[fs_object["Pool"]]
-            for fs_object in fs_objects
+            (obj_path, fs_object["Name"], fs_object["Origin"]): pool_path_to_name[
+                fs_object["Pool"]
+            ]
+            for obj_path, fs_object in fs_objects
         }
 
     @staticmethod
@@ -465,17 +468,17 @@ def pool_rename(pool_name, pool_name_rename):
         return iface.SetName(pool_name_rename, timeout=StratisDbus._TIMEOUT)
 
     @staticmethod
-    def pool_set_property(pool_path, param_iface, dbus_param, dbus_value):
+    def set_property(object_path, param_iface, dbus_param, dbus_value):
         """
         Set D-Bus parameter on a pool
-        :param str pool_path: The object path of the pool
+        :param str object_path: The path of the object
         :param str dbus_param: The parameter to be set
         :param str dbus_value: The value
         :return: None
         :raises dbus.exceptions.DBusException:
         """
         iface = dbus.Interface(
-            StratisDbus._BUS.get_object(StratisDbus._BUS_NAME, pool_path),
+            StratisDbus._BUS.get_object(StratisDbus._BUS_NAME, object_path),
             dbus.PROPERTIES_IFACE,
         )
 

testlib/infra.py

@@ -48,7 +48,7 @@
 STRATIS_METADATA_LEN = Range(8192, 512)
 
 
-def clean_up():  # pylint: disable=too-many-branches
+def clean_up():  # pylint: disable=too-many-branches,too-many-locals
     """
     Try to clean up after a test failure.
 
@@ -78,7 +78,7 @@ def check_result(result, format_str, format_str_args):
 
     # Unmount FS
     for mountpoint_dir in fnmatch.filter(os.listdir(VAR_TMP), f"*{MOUNT_POINT_SUFFIX}"):
-        for name, _ in StratisDbus.fs_list().items():
+        for (_, name, _), _ in StratisDbus.fs_list().items():
             try:
                 subprocess.check_call(
                     [UMOUNT, os.path.join(VAR_TMP, mountpoint_dir, name)]
@@ -89,8 +89,22 @@ def check_result(result, format_str, format_str_args):
                     f"{os.path.join(VAR_TMP, mountpoint_dir, name)}: {err}"
                 )
 
+    # Unset MergeScheduled
+    for (fs_path, name, (origin_set, _)), pool_name in StratisDbus.fs_list().items():
+        if origin_set:
+            check_result(
+                StratisDbus.set_property(
+                    fs_path,
+                    StratisDbus.FS_IFACE,
+                    "MergeScheduled",
+                    dbus.Boolean(False),
+                ),
+                "failed to set MergeScheduled to False",
+                (name, pool_name),
+            )
+
     # Remove FS
-    for name, pool_name in StratisDbus.fs_list().items():
+    for (_, name, _), pool_name in StratisDbus.fs_list().items():
         check_result(
             StratisDbus.fs_destroy(pool_name, name),
             "failed to destroy filesystem %s in pool %s",