mount: Add a stress test with multiple clients

This test simulates multiple clients performing file operations
using the concept of threading.

Four cases with different number of clients, file operations and
file sizes are included to ensure maximum coverage.

Signed-off-by: Shwetha K Acharya <Shwetha.K.Acharya@ibm.com>

testcases/mount/mount_stress.py

@@ -0,0 +1,57 @@
+import threading
+import testhelper
+import pathlib
+
+
+def _perform_file_operations(
+    client_id: int, root_dir: str, num_operations: int, file_size: int
+) -> None:
+    try:
+        for i in range(num_operations):
+            filename = f"testfile_{client_id}_{i}.txt"
+            file_content = testhelper.generate_random_bytes(file_size)
+            path = pathlib.Path(root_dir, filename)
+            path.write_text(file_content.decode())
+            file_content_out = path.read_text()
+
+            if file_content_out != file_content:
+                raise IOError("content mismatch")
+
+            path.unlink()
+    except Exception as ex:
+        print(f"Error while stress testing with Client {client_id}: %s", ex)
+        raise
+
+
+def _stress_test(
+    root_dir: str, num_clients: int, num_operations: int, file_size: int
+) -> None:
+    threads = []
+
+    for i in range(num_clients):
+        thread = threading.Thread(
+            target=_perform_file_operations,
+            args=(i, root_dir, num_operations, file_size),
+        )
+        threads.append(thread)
+
+    for thread in threads:
+        thread.start()
+
+    for thread in threads:
+        thread.join()
+
+    print("Stress test complete.")
+
+
+def check_mnt_stress(root_dir: str) -> None:
+    _stress_test(root_dir, num_clients=5, num_operations=20, file_size=2**22)
+    _stress_test(
+        root_dir, num_clients=10, num_operations=30, file_size=2**23
+    )
+    _stress_test(
+        root_dir, num_clients=10, num_operations=40, file_size=2**24
+    )
+    _stress_test(
+        root_dir, num_clients=10, num_operations=25, file_size=2**25
+    )

testcases/mount/test_mount.py

@@ -12,6 +12,7 @@
 
 from .mount_io import check_io_consistency
 from .mount_dbm import check_dbm_consistency
+from .mount_stress import check_mnt_stress
 
 # Use a global test_info to get a better output when running pytest
 test_info: typing.Dict[str, typing.Any] = {}
@@ -30,6 +31,7 @@ def mount_check(ipaddr: str, share_name: str) -> None:
         os.mkdir(test_dir)
         check_io_consistency(test_dir)
         check_dbm_consistency(test_dir)
+        check_mnt_stress(test_dir)
     finally:
         if flag_mounted:
             shutil.rmtree(test_dir, ignore_errors=True)