Add Supplementary group tests

test-info.yml.example

@@ -1,20 +1,35 @@
-private_interfaces:
-  - "192.168.122.100"
-  - "192.168.122.101"
+# Server hostname to be tested
+server: server_name
 
-public_interfaces:
-  - "192.168.123.10"
-  - "192.168.123.11"
+# Users to use for authentication
+users:
+  user1: user1password
 
-exported_sharenames:
-  - "gluster-vol"
+# Backend filesystem of the exported shares
+backend: glusterfs
 
-test_users:
-  - {"username": "test1", "password": "x"}
-  - {"username": "test2", "password": "x"}
+# Additional information which may be used by tests here
+# some tests may be skipped if the required config option is not set
+extra:
+  # The supplementary group to be used by test_supplementary_group
+  supplementary_group: sg
 
-test_backend: glusterfs
-
-premounted_shares:
-  - "/testdir1"
-  - "/testdir2"
+# shares: List of dict of exported shares
+shares:
+  # share name export1
+  export1:
+    # If present, it means the share is pre-mounted
+    path: /mnt/share/export1-cephfs-vfs
+    backend:
+      # If present backend filesystem
+      name: cephfs.vfs
+      # If present, provides direct access to share
+      path: /mnt/backend/export1
+    # If present, username to perform the tests with on this share
+    users:
+      test2: x
+    # If present, the hostname to use to connect to the test server
+    server: hostname1
+  # share name export2
+  export2:
+    # Use default values set for this share

testcases/consistency/test_consistency.py

@@ -61,9 +61,9 @@ def generate_consistency_check(
     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))
+    for share in testhelper.get_exported_shares(test_info):
+        s = testhelper.get_share(test_info, share)
+        arr.append((s["server"], s["name"]))
     return arr
 
 

testcases/containers/test_containers.py

@@ -66,12 +66,11 @@ def containers_check(ipaddr: str, share_name: str, test: str) -> None:
 
 
 def generate_containers_test() -> typing.List[typing.Tuple[str, str, str]]:
-    # Use the first given public_interface for our tests
-    ipaddr = test_info["public_interfaces"][0]
     arr = []
-    for share_name in test_info["exported_sharenames"]:
+    for share_name in testhelper.get_exported_shares(test_info):
+        server = testhelper.get_share(test_info, share_name)["server"]
         for test in container_tests.keys():
-            arr.append((ipaddr, share_name, test))
+            arr.append((server, share_name, test))
     return arr
 
 

testcases/misc/conftest.py

@@ -46,12 +46,12 @@ def setup_mount(
 
 
 def gen_params() -> typing.List[typing.Any]:
-    ipaddr = test_info["public_interfaces"][0]
-    exported_sharenames = test_info.get("exported_sharenames", [])
+    exported_sharenames = testhelper.get_exported_shares(test_info)
     arr = []
     for share_name in exported_sharenames:
+        server = testhelper.get_share(test_info, share_name)["server"]
         arr.append(
-            pytest.param((ipaddr, share_name), id=f"{ipaddr}-{share_name}")
+            pytest.param((server, share_name), id=f"{server}-{share_name}")
         )
     return arr
 

testcases/misc/test_supplemantary_group.py

@@ -0,0 +1,100 @@
+# This test is to check writes to a folder owned by a supplementary group.
+# expected username in share["users"] and
+# share["extra"]["supplementary_group"] with supplementary group
+# also needs direct access to underlying filesystem
+#
+# Requirements:
+# 1. username in share["user"] exists
+# 2. username is part of supplementary group provided in
+#    user["extra"]["supplementary_group"]
+# 3. share["backend"]["path"] should be set
+#
+# Steps:
+# 1. Create folder test_subdir/ on direct path with
+#    group set to sgroup and mode 0770
+# 2. Upload file to test_subdir/test-cp
+#
+# Expected:
+# Copy passes
+
+import testhelper
+import pytest
+import shutil
+import pwd
+import grp
+import os
+from pathlib import Path
+
+test_info_file = os.getenv("TEST_INFO_FILE")
+test_info = testhelper.read_yaml(test_info_file)
+test_string = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+test_subdir = Path("supplementary_group")
+
+
+def check_reqs(username: str, groupname: str) -> bool:
+    try:
+        pwd.getpwnam(username)
+        if username not in grp.getgrnam(groupname).gr_mem:
+            return False
+    except KeyError:
+        return False
+    return True
+
+
+def setup_local_testdir(root: Path, group: str) -> Path:
+    testdir = root / test_subdir
+    testdir.mkdir(exist_ok=True)
+    shutil.chown(testdir, group=group)
+    testdir.chmod(0o770)
+    return testdir
+
+
+def write_file_remote(
+    mount_point: Path,
+    ipaddr: str,
+    share_name: str,
+) -> None:
+    mount_params = testhelper.get_mount_parameters(test_info, share_name)
+    mount_params["host"] = ipaddr
+    try:
+        test_file = testhelper.get_tmp_file(mount_point)
+        test_file_remote = test_subdir / Path("test-cp")
+        with open(test_file, "w") as f:
+            f.write(test_string)
+        put_cmds = "put  %s %s" % (test_file, test_file_remote)
+        (ret, output) = testhelper.smbclient(mount_params, put_cmds)
+        assert ret == 0, "Failed to copy file to server: " + output
+    finally:
+        if test_file.exists():
+            test_file.unlink()
+
+
+def gen_supplementary_group_param(test_info: dict) -> list:
+    if not test_info:
+        return []
+    sgroup = testhelper.get_conf_extra(test_info, "supplementary_group")
+    if sgroup is None:
+        return []
+    arr = []
+    for share in testhelper.get_shares_with_directmnt(test_info):
+        s = testhelper.get_share(test_info, share)
+        username = list(s["users"].keys())[0]
+        if check_reqs(username, sgroup):
+            arr.append((s["server"], share))
+    return arr
+
+
+@pytest.mark.parametrize(
+    "ipaddr,share_name", gen_supplementary_group_param(test_info)
+)
+def test_supplementary_group(ipaddr: str, share_name: str) -> None:
+    share = testhelper.get_share(test_info, share_name)
+    fs_path = Path(share["backend"]["path"])
+    sgroup = testhelper.get_conf_extra(test_info, "supplementary_group")
+    testdir = setup_local_testdir(fs_path, sgroup)
+    try:
+        tmp_root = testhelper.get_tmp_root()
+        mount_point = testhelper.get_tmp_mount_point(tmp_root)
+        write_file_remote(mount_point, ipaddr, share_name)
+    finally:
+        shutil.rmtree(testdir)

testcases/smbtorture/test_smbtorture.py

@@ -48,7 +48,8 @@ def smbtorture(share_name: str, test: str, tmp_output: Path) -> bool:
             "--expected-failures=" + script_root + "/selftest/" + filter
         )
     flapping_list = ["flapping", "flapping.d"]
-    test_backend = test_info.get("test_backend")
+    share = testhelper.get_share(test_info, share_name)
+    test_backend = share["backend"].get("name")
     if test_backend is not None:
         flapping_file = "flapping." + test_backend
         flapping_file_path = os.path.join(
@@ -114,7 +115,7 @@ def list_smbtorture_tests():
 def generate_smbtorture_tests() -> typing.List[typing.Tuple[str, str]]:
     smbtorture_info = list_smbtorture_tests()
     arr = []
-    for share_name in test_info.get("exported_sharenames", []):
+    for share_name in testhelper.get_exported_shares(test_info):
         for torture_test in smbtorture_info:
             arr.append((share_name, torture_test))
     return arr