Run a mount and also write filesystem test

Signed-off-by: mulhern <[email protected]>

stratisd_cert.py

@@ -20,6 +20,7 @@
 import argparse
 import json
 import os
+import subprocess
 import sys
 import unittest
 from tempfile import NamedTemporaryFile
@@ -32,6 +33,7 @@
 from testlib.infra import (
     DbusMonitor,
     KernelKey,
+    MountPointManager,
     PostTestCheck,
     RunPostTestChecks,
     StratisdSystemdStart,
@@ -1028,6 +1030,33 @@ def test_filesystem_destroy(self):
 
         self.assertEqual(StratisDbus.fs_list(), {})
 
+    @skip(_skip_condition(1))
+    def test_filesystem_mount_and_write(self):
+        """
+        Test mount and write to filesystem.
+        """
+        pool_name = p_n()
+        pool_path, _ = make_test_pool(pool_name, StratisCertify.DISKS[0:1])
+
+        fs_name = fs_n()
+        make_test_filesystem(pool_path, fs_name)
+
+        mountpoint_mgr = MountPointManager()
+        mountpoint = mountpoint_mgr.mount(
+            os.path.join("/", "dev", "stratis", pool_name, fs_name)
+        )
+
+        subprocess.check_call(
+            [
+                "dd",
+                "if=/dev/urandom",
+                f'of={os.path.join(mountpoint, "file1")}',
+                "bs=4096",
+                "count=32",
+                "conv=fsync",
+            ]
+        )
+
     def test_get_report(self):
         """
         Test getting a valid and invalid report.

testlib/infra.py

@@ -19,8 +19,10 @@
 import fnmatch
 import json
 import os
+import shutil
 import signal
 import subprocess
+import tempfile
 import time
 import unittest
 from enum import Enum
@@ -38,9 +40,23 @@
 DBUS_NAME_HAS_NO_OWNER_ERROR = "org.freedesktop.DBus.Error.NameHasNoOwner"
 SYS_CLASS_BLOCK = "/sys/class/block"
 DEV_MAPPER = "/dev/mapper"
+VAR_TMP = "/var/tmp"
+MOUNT_POINT_SUFFIX = "_stratisd_mounts"
+UMOUNT = "umount"
+MOUNT = "mount"
 
 
-def clean_up():
+def mountpoint_path(dir_name, fs_name):
+    """
+    Get mountpoint name.
+
+    :param str dir_name: directory containing mountpoints
+    :param str fs_name: name of the mountpoint
+    """
+    return os.path.join(VAR_TMP, dir_name, fs_name)
+
+
+def clean_up():  # pylint: disable=too-many-branches
     """
     Try to clean up after a test failure.
 
@@ -68,6 +84,21 @@ def check_result(result, format_str, format_str_args):
     for uuid in StratisDbus.stopped_pools():
         StratisDbus.pool_start(uuid, "uuid")
 
+    # Unmount FS
+    for mountpoint_dir in fnmatch.filter(
+        os.listdir(VAR_TMP), f"*_{MOUNT_POINT_SUFFIX}"
+    ):
+        for name, _ in StratisDbus.fs_list().items():
+            try:
+                subprocess.check_call(
+                    [UMOUNT, os.path.join(VAR_TMP, mountpoint_dir, name)]
+                )
+            except subprocess.CalledProcessError as err:
+                error_strings.append(
+                    "Failed to umount filesystem at "
+                    f"{os.path.join(VAR_TMP, mountpoint_dir, name)}: {err}"
+                )
+
     # Remove FS
     for name, pool_name in StratisDbus.fs_list().items():
         check_result(
@@ -114,6 +145,16 @@ def check_result(result, format_str, format_str_args):
         if remnant_keys != []:
             error_strings.append(f'remnant keys: {", ".join(remnant_keys)}')
 
+    for mountpoint_dir in fnmatch.filter(
+        os.listdir(VAR_TMP), f"*_{MOUNT_POINT_SUFFIX}"
+    ):
+        try:
+            shutil.rmtree(mountpoint_dir)
+        except Exception as err:  # pylint: disable=broad-exception-caught
+            error_strings.append(
+                f"failed to clean up temporary mountpoint dir {mountpoint_dir}: {err}"
+            )
+
     assert isinstance(error_strings, list)
     if error_strings:
         raise RuntimeError(
@@ -558,3 +599,25 @@ def set_from_post_test_check_option(post_test_check):
             PostTestCheck.FILESYSTEM_SYMLINKS in post_test_check
         )
         PoolMetadataMonitor.verify = PostTestCheck.POOL_METADATA in post_test_check
+
+
+class MountPointManager:  # pylint: disable=too-few-public-methods
+    """
+    Handle mounting Stratis filesystems in a temp directory.
+    """
+
+    def __init__(self):
+        """
+        Initalizer.
+        """
+        self.mount_root = tempfile.mkdtemp(suffix=MOUNT_POINT_SUFFIX, dir=VAR_TMP)
+
+    def mount(self, fs_path):
+        """
+        :param str fs_path: the absolute path to mount
+        :returns: the mountpoint
+        :rtype: str
+        """
+        mountpoint = os.path.join(self.mount_root, os.path.basename(fs_path))
+        subprocess.check_call([MOUNT, fs_path, mountpoint])
+        return mountpoint