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