Introduce setup and teardown fixture

With this change, multiple test modules triggered from testcases/mount
can be run as individual tests complying to pytest standards.
Also any number of new tests can be added efficiently on the mount.

PS: Inorder to effectively introduce fixtures, refactoring of
code is done in test_mount_dbm.py

Fixes: #30
Signed-off-by: Shwetha K Acharya <[email protected]>

conftest.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+
+import pytest
+import os
+import shutil
+import testhelper
+import typing
+
+flag_mounted: bool = False
+tmp_root: str
+test_dir: str
+mount_point: str
+test_info_file = os.getenv("TEST_INFO_FILE")
+
+
+@pytest.fixture
+def setup_mount(ipaddr: str, sharename: str, request: typing.Any) -> str:
+    global flag_mounted, tmp_root, test_dir, mount_point
+    tmp_root = testhelper.get_tmp_root()
+    mount_point = testhelper.get_tmp_mount_point(tmp_root)
+    try:
+        mount_params = testhelper.get_mount_parameters(
+            testhelper.read_yaml(test_info_file), sharename
+        )
+        mount_params["host"] = ipaddr
+
+        # mount cifs share
+        testhelper.cifs_mount(mount_params, mount_point)
+        flag_mounted = True
+        test_dir = os.path.join(mount_point, "mount_test")
+        os.makedirs(test_dir, exist_ok=True)
+    except Exception as e:
+        raise Exception(f"Setup failed: {str(e)}")
+    return test_dir
+
+    # Define a finalizer function for teardown
+    def teardown_mount() -> typing.Generator[None, str, None]:
+        yield
+        try:
+            if flag_mounted:
+                shutil.rmtree(test_dir, ignore_errors=True)
+                testhelper.cifs_umount(mount_point)
+            os.rmdir(mount_point)
+            os.rmdir(tmp_root)
+        except Exception as e:
+            raise Exception(f"Teardown failed: {str(e)}")
+
+    request.addfinalizer(teardown_mount)

testcases/consistency/test_consistency.py

@@ -13,8 +13,8 @@
 
 
 test_string = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
-test_info = os.getenv("TEST_INFO_FILE")
-test_info_dict = testhelper.read_yaml(test_info)
+test_info_file = os.getenv("TEST_INFO_FILE")
+test_info_dict = testhelper.read_yaml(test_info_file)
 
 
 def file_content_check(f: typing.IO, comp_str: str) -> bool:
@@ -54,20 +54,8 @@ def consistency_check(mount_point: str, ipaddr: str, share_name: str) -> None:
             os.unlink(test_file_resp)
 
 
-def generate_consistency_check(
-    test_info_file: dict,
-) -> typing.List[typing.Tuple[str, str]]:
-    if not test_info_file:
-        return []
-    arr = []
-    for ipaddr in test_info_file["public_interfaces"]:
-        for share_name in test_info_file["exported_sharenames"]:
-            arr.append((ipaddr, share_name))
-    return arr
-
-
 @pytest.mark.parametrize(
-    "ipaddr,share_name", generate_consistency_check(test_info_dict)
+    "ipaddr,share_name", testhelper.generate_mount_check(test_info_file)
 )
 def test_consistency(ipaddr: str, share_name: str) -> None:
     tmp_root = testhelper.get_tmp_root()

testcases/mount/mount_dbm.py → testcases/mount/test_mount_dbm.py

@@ -1,13 +1,16 @@
 #!/usr/bin/env python3
 # Test various database operations via SMB mount-point.
 
+import pytest
 import dbm
 import hashlib
 import pickle
 import pathlib
 import shutil
 import typing
 import random
+import os
+import testhelper
 
 
 class Record:
@@ -89,12 +92,24 @@ def _check_dbm_consistency(base: pathlib.Path, nrecs: int) -> None:
         db.destroy()
 
 
-def check_dbm_consistency(rootdir: str) -> None:
-    base = pathlib.Path(rootdir) / "dbm-consistency"
+def _dbm_consistency(mount_point):
+    base = pathlib.Path(mount_point) / "dbm-consistency"
     base.mkdir(parents=True, exist_ok=True)
     try:
         _check_dbm_consistency(base, 10)
         _check_dbm_consistency(base, 100)
         _check_dbm_consistency(base, 10000)
     finally:
         shutil.rmtree(base, ignore_errors=True)
+
+
+test_info_file = os.getenv("TEST_INFO_FILE")
+
+
+@pytest.mark.parametrize(
+    "ipaddr,sharename", testhelper.generate_mount_check(test_info_file)
+)
+def test_check_dbm_consistency(
+    ipaddr: str, sharename: str, setup_mount: str
+) -> None:
+    _dbm_consistency(setup_mount)

testcases/mount/mount_io.py → testcases/mount/test_mount_io.py

@@ -2,12 +2,14 @@
 
 # Test various file-system I/O operations via local SMB mount-point.
 
+import pytest
 import datetime
 import pathlib
 import shutil
 import typing
 import testhelper
 import random
+import os
 
 
 class DataPath:
@@ -101,11 +103,11 @@ def _run_checks(dsets: typing.List[DataPath]) -> None:
         dset.verify_noent()
 
 
-def _check_io_consistency(rootdir: str) -> None:
+def _check_io_consistency(test_dir: str) -> None:
     base = None
     try:
         print("\n")
-        base = pathlib.Path(rootdir) / "test_io_consistency"
+        base = pathlib.Path(test_dir) / "test_io_consistency"
         base.mkdir(parents=True)
         # Case-1: single 4K file
         _run_checks(_make_datasets(base, 4096, 1))
@@ -129,6 +131,15 @@ def _reset_random_seed() -> None:
     random.seed(seed)
 
 
-def check_io_consistency(rootdir: str) -> None:
+test_info_file = os.getenv("TEST_INFO_FILE")
+
+
+@pytest.mark.parametrize(
+    "ipaddr,sharename", testhelper.generate_mount_check(test_info_file)
+)
+def test_check_io_consistency(
+    ipaddr: str, sharename: str, setup_mount: str
+) -> None:
+    print("%s - %s" % (ipaddr, sharename))
     _reset_random_seed()
-    _check_io_consistency(rootdir)
+    _check_io_consistency(setup_mount)

testcases/mount/mount_stress.py → testcases/mount/test_mount_stress.py

@@ -1,6 +1,8 @@
+import pytest
 import threading
 import testhelper
 import pathlib
+import os
 
 
 def _perform_file_operations(
@@ -44,7 +46,7 @@ def _stress_test(
     print("Stress test complete.")
 
 
-def check_mnt_stress(root_dir: str) -> None:
+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
@@ -55,3 +57,15 @@ def check_mnt_stress(root_dir: str) -> None:
     _stress_test(
         root_dir, num_clients=15, num_operations=25, file_size=2**25
     )
+
+
+test_info_file = os.getenv("TEST_INFO_FILE")
+
+
+@pytest.mark.parametrize(
+    "ipaddr,sharename", testhelper.generate_mount_check(test_info_file)
+)
+def test_check_mnt_stress(
+    ipaddr: str, sharename: str, setup_mount: str
+) -> None:
+    _check_mnt_stress(setup_mount)

testcases/smbtorture/test_smbtorture.py

@@ -15,8 +15,8 @@
 format_subunit_exec = script_root + "/selftest/format-subunit"
 smbtorture_tests_file = script_root + "/smbtorture-tests-info.yml"
 
-test_info = os.getenv("TEST_INFO_FILE")
-test_info_dict = testhelper.read_yaml(test_info)
+test_info_file = os.getenv("TEST_INFO_FILE")
+test_info_dict = testhelper.read_yaml(test_info_file)
 
 # Temp filename containing the output of run commands.
 output = testhelper.get_tmp_file("/tmp")