From 93954b0c2664287d830b6da5e91490548ba76cce Mon Sep 17 00:00:00 2001 From: Sachin Prabhu Date: Sun, 24 Mar 2024 03:02:44 +0000 Subject: [PATCH] testhelper: introduce class SMBClient We use the python module pysmbc to access the SMB server. The helper class hides the complexities and helps us bypass the smbclient binary to connect to the remote samba server giving us more flexibility in the python tests. Signed-off-by: Sachin Prabhu --- testcases/consistency/test_consistency.py | 90 +++++++++-------------- testhelper/__init__.py | 1 + testhelper/cmdhelper.py | 29 -------- testhelper/smbclient.py | 53 +++++++++++++ tox.ini | 3 + 5 files changed, 92 insertions(+), 84 deletions(-) create mode 100644 testhelper/smbclient.py diff --git a/testcases/consistency/test_consistency.py b/testcases/consistency/test_consistency.py index aeb4982..a04885d 100755 --- a/testcases/consistency/test_consistency.py +++ b/testcases/consistency/test_consistency.py @@ -7,72 +7,52 @@ # ip addresses). import testhelper +from testhelper import SMBClient import os import pytest import typing -from pathlib import Path - test_string = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" test_info_file = os.getenv("TEST_INFO_FILE") test_info = testhelper.read_yaml(test_info_file) -def file_content_check(f: typing.IO, comp_str: str) -> bool: - read_data = f.read() - return read_data == comp_str - - -def consistency_check(mount_point: Path, ipaddr: str, share_name: str) -> None: +def consistency_check(hostname: 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_resp = test_file.with_suffix(".resp") - test_file_remote = "test-" + ipaddr + "." + share_name - 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" - - # The file read cycle - get_cmds = "get %s %s; rm %s" % ( - test_file_remote, - test_file_resp, - test_file_remote, - ) - (ret, output) = testhelper.smbclient(mount_params, get_cmds) - assert ret == 0, "Failed to copy file from server" - with open(test_file_resp, "r") as f: - assert file_content_check( - f, test_string - ), "File content does not match" - finally: - if test_file.exists(): - test_file.unlink() - if test_file_resp.exists(): - test_file_resp.unlink() - - -def generate_consistency_check( - test_info_file: dict, -) -> typing.List[typing.Tuple[str, str]]: - if not test_info_file: - return [] + test_filename = "/test_consistency" + + # file write cycle + scon = SMBClient( + hostname, + share_name, + mount_params["username"], + mount_params["password"], + ) + scon.simple_write(test_filename, test_string) + scon.disconnect() + + # file read cycle + scon = SMBClient( + hostname, + share_name, + mount_params["username"], + mount_params["password"], + ) + retstr = scon.simple_read(test_filename) + scon.unlink(test_filename) + scon.disconnect() + + assert retstr == test_string, "File content does not match" + + +def generate_consistency_check() -> typing.List[typing.Tuple[str, str]]: arr = [] - for share in testhelper.get_exported_shares(test_info): - s = testhelper.get_share(test_info, share) - arr.append((s["server"], s["name"])) + for sharename in testhelper.get_exported_shares(test_info): + share = testhelper.get_share(test_info, sharename) + arr.append((share["server"], share["name"])) return arr -@pytest.mark.parametrize( - "ipaddr,share_name", generate_consistency_check(test_info) -) -def test_consistency(ipaddr: str, share_name: str) -> None: - tmp_root = testhelper.get_tmp_root() - mount_point = testhelper.get_tmp_mount_point(tmp_root) - consistency_check(mount_point, ipaddr, share_name) - os.rmdir(mount_point) - os.rmdir(tmp_root) +@pytest.mark.parametrize("hostname,share_name", generate_consistency_check()) +def test_consistency(hostname: str, share_name: str) -> None: + consistency_check(hostname, share_name) diff --git a/testhelper/__init__.py b/testhelper/__init__.py index 1ca5fce..626e71e 100644 --- a/testhelper/__init__.py +++ b/testhelper/__init__.py @@ -1,3 +1,4 @@ from .testhelper import * # noqa: F401, F403 from .cmdhelper import * # noqa: F401, F403 from .fshelper import * # noqa: F401, F403 +from .smbclient import * # noqa: F401, F403 diff --git a/testhelper/cmdhelper.py b/testhelper/cmdhelper.py index b19eda0..4c77aa0 100644 --- a/testhelper/cmdhelper.py +++ b/testhelper/cmdhelper.py @@ -63,35 +63,6 @@ def cifs_umount(mount_point: Path) -> int: return ret -def smbclient( - mount_params: typing.Dict[str, str], cmds: str -) -> typing.Tuple[int, str]: - """Run the following command on smbclient and return the output. - - Parameters: - mount_params: Dict containing the mount parameters. - cmds: String containg the ';' separated commands to run. - - Returns: - int: Return value from the shell execution - string: stdout - """ - smbclient_cmd = [ - "smbclient", - "--user=%s%%%s" % (mount_params["username"], mount_params["password"]), - "//%s/%s" % (mount_params["host"], mount_params["share"]), - "-c", - cmds, - ] - ret = subprocess.run( - smbclient_cmd, - universal_newlines=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - ) - return (ret.returncode, ret.stdout) - - def check_cmds(cmds: typing.List[str]) -> Path: """Return first file path which exists. diff --git a/testhelper/smbclient.py b/testhelper/smbclient.py new file mode 100644 index 0000000..9f01c7b --- /dev/null +++ b/testhelper/smbclient.py @@ -0,0 +1,53 @@ +import smbc # type: ignore +import typing +import os + + +class SMBClient: + """Use pysmbc to access the SMB server""" + + def __init__(self, hostname: str, share: str, username: str, passwd: str): + self.server = hostname + self.share = share + self.username = username + self.password = passwd + self.rooturi = f"smb://{self.server}/{self.share}" + + def auth_cb(se, sh, w, u, p): + return (w, self.username, self.password) + + self.ctx = smbc.Context(auth_fn=auth_cb) + + def disconnect(self): + del self.ctx + + def readdir(self, path: str = "/") -> typing.List: + uri = self.rooturi + path + d_ent = self.ctx.opendir(uri) + return [d_ent.name for d_ent in d_ent.getdents()] + + def _open( + self, path: str, flags: int = os.O_RDWR, mode: int = 0o644 + ) -> typing.Any: + uri = self.rooturi + path + return self.ctx.open(uri, flags) + + def mkdir(self, dpath: str) -> None: + self.ctx.mkdir(self.rooturi + dpath) + + def rmdir(self, dpath: str) -> None: + self.ctx.rmdir(self.rooturi + dpath) + + def unlink(self, fpath: str) -> None: + self.ctx.unlink(self.rooturi + fpath) + + def simple_write(self, fpath: str, teststr: str) -> None: + f = self._open(fpath, os.O_WRONLY | os.O_CREAT) + f.write(teststr) + f.close() + + def simple_read(self, fpath: str) -> str: + f = self._open(fpath) + str = f.read() + f.close() + return str.decode("ascii") diff --git a/tox.ini b/tox.ini index a2646b6..edff27c 100644 --- a/tox.ini +++ b/tox.ini @@ -13,6 +13,7 @@ deps = pyyaml pytest-randomly iso8601 + pysmbc commands = pytest -vrfEsxXpP testcases/ [testenv:pytest-unprivileged] @@ -21,6 +22,7 @@ deps = pyyaml pytest-randomly iso8601 + pysmbc commands = pytest -vrfEsxXpP -k 'not privileged' testcases/ [testenv:sanity] @@ -29,6 +31,7 @@ deps = pyyaml pytest-randomly iso8601 + pysmbc changedir = {toxinidir} commands = pytest -vrfEsxXpP testcases/consistency