|
| 1 | +# This test is to check writes as part of a supplementary group. |
| 2 | +# |
| 3 | +# Requirements: |
| 4 | +# 1. user1 exists |
| 5 | +# 2. user1 is part of supplementary group sg |
| 6 | +# |
| 7 | +# Steps: |
| 8 | +# 1. Create folder test_subdir/ with group "sg" and mode 0770 |
| 9 | +# 2. Upload file to test_subdir/test-cp |
| 10 | +# |
| 11 | +# Expected: |
| 12 | +# Copy passes |
| 13 | + |
| 14 | +import testhelper |
| 15 | +import pytest |
| 16 | +import shutil |
| 17 | +import pwd |
| 18 | +import grp |
| 19 | +import os |
| 20 | +from pathlib import Path |
| 21 | + |
| 22 | +test_info_file = os.getenv("TEST_INFO_FILE") |
| 23 | +test_info = testhelper.read_yaml(test_info_file) |
| 24 | +test_string = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" |
| 25 | +test_subdir = Path("supplementary_group") |
| 26 | +test_sgroup = "sg" |
| 27 | + |
| 28 | + |
| 29 | +def check_reqs(): |
| 30 | + try: |
| 31 | + pwd.getpwnam("test1") |
| 32 | + if "test1" not in grp.getgrnam(test_sgroup).gr_mem: |
| 33 | + return False |
| 34 | + except KeyError: |
| 35 | + return False |
| 36 | + return True |
| 37 | + |
| 38 | + |
| 39 | +def setup_local_testdir(root: Path, group: str) -> Path: |
| 40 | + testdir = root / test_subdir |
| 41 | + testdir.mkdir(exist_ok=True) |
| 42 | + shutil.chown(testdir, group=group) |
| 43 | + testdir.chmod(0o770) |
| 44 | + return testdir |
| 45 | + |
| 46 | + |
| 47 | +def write_file_remote( |
| 48 | + mount_point: Path, |
| 49 | + ipaddr: str, |
| 50 | + share_name: str, |
| 51 | +) -> None: |
| 52 | + mount_params = testhelper.get_mount_parameters(test_info, share_name) |
| 53 | + mount_params["host"] = ipaddr |
| 54 | + try: |
| 55 | + test_file = testhelper.get_tmp_file(mount_point) |
| 56 | + test_file_remote = test_subdir / Path("test-cp") |
| 57 | + with open(test_file, "w") as f: |
| 58 | + f.write(test_string) |
| 59 | + put_cmds = "put %s %s" % (test_file, test_file_remote) |
| 60 | + (ret, output) = testhelper.smbclient(mount_params, put_cmds) |
| 61 | + assert ret == 0, "Failed to copy file to server: " + output |
| 62 | + finally: |
| 63 | + if test_file.exists(): |
| 64 | + test_file.unlink() |
| 65 | + |
| 66 | + |
| 67 | +def gen_supplementary_group_param(test_info: dict) -> list: |
| 68 | + if not check_reqs(): |
| 69 | + return [] |
| 70 | + if not test_info: |
| 71 | + return [] |
| 72 | + arr = [] |
| 73 | + for ipaddr in test_info["public_interfaces"]: |
| 74 | + for share in testhelper.get_shares_with_directmnt(test_info): |
| 75 | + arr.append((ipaddr, share["name"])) |
| 76 | + return arr |
| 77 | + |
| 78 | + |
| 79 | +@pytest.mark.parametrize( |
| 80 | + "ipaddr,share_name", gen_supplementary_group_param(test_info) |
| 81 | +) |
| 82 | +def test_supplementary_group(ipaddr: str, share_name: str) -> None: |
| 83 | + share = testhelper.get_share(test_info, share_name) |
| 84 | + fs_path = Path(share["directmnt"]) |
| 85 | + testdir = setup_local_testdir(fs_path, test_sgroup) |
| 86 | + try: |
| 87 | + tmp_root = testhelper.get_tmp_root() |
| 88 | + mount_point = testhelper.get_tmp_mount_point(tmp_root) |
| 89 | + write_file_remote(mount_point, ipaddr, share_name) |
| 90 | + finally: |
| 91 | + shutil.rmtree(testdir) |
0 commit comments