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,31 @@ 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)
+
+        with MountPointManager(
+            [os.path.join("/", "dev", "stratis", pool_name, fs_name)]
+        ) as mountpoints:
+            subprocess.check_call(
+                [
+                    "dd",
+                    "if=/dev/urandom",
+                    f'of={os.path.join(mountpoints[0], "file1")}',
+                    "bs=4096",
+                    "count=256",
+                    "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,13 @@
 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 clean_up():  # pylint: disable=too-many-branches
     """
     Try to clean up after a test failure.
 
@@ -68,6 +74,19 @@ 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 +133,14 @@ 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(os.path.join(VAR_TMP, 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(
@@ -557,3 +584,51 @@ 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, fs_paths):
+        """
+        Initalizer.
+
+        :param str fs_paths: the absolute paths to mount
+        :rtype: None
+        """
+        self.mount_root = tempfile.mkdtemp(suffix=MOUNT_POINT_SUFFIX, dir=VAR_TMP)
+        mountpoints = []
+        for fs_path in fs_paths:
+            mountpoint = os.path.join(self.mount_root, os.path.basename(fs_path))
+            try:
+                os.mkdir(mountpoint)
+            except FileExistsError:
+                pass
+            subprocess.check_call([MOUNT, fs_path, mountpoint])
+            mountpoints.append(mountpoint)
+
+        self.mountpoints = mountpoints
+
+    def __enter__(self):
+        """
+        :returns: the mountpoints
+        :rtype: list of str
+        """
+        return self.mountpoints
+
+    def __exit__(self, exception_type, exception_value, traceback):
+        """
+        Unmount all filesystems and remove subdir in temporary directory.
+        """
+        try:
+            for mp in self.mountpoints:
+                subprocess.check_call([UMOUNT, mp])
+
+            shutil.rmtree(self.mount_root)
+
+        except Exception as rexc:
+            if exception_value is None:
+                raise rexc
+            raise rexc from exception_value