diff --git a/testcases/consistency/test_consistency.py b/testcases/consistency/test_consistency.py index aeb4982..34ae9fa 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 + smbclient = SMBClient( + mount_params["host"], + mount_params["share"], + mount_params["username"], + mount_params["password"], + ) + smbclient.write_text(test_filename, test_string) + smbclient.disconnect() + + # file read cycle + smbclient = SMBClient( + mount_params["host"], + mount_params["share"], + mount_params["username"], + mount_params["password"], + ) + retstr = smbclient.read_text(test_filename) + smbclient.unlink(test_filename) + smbclient.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..0f84ec0 --- /dev/null +++ b/testhelper/smbclient.py @@ -0,0 +1,77 @@ +from smb.SMBConnection import SMBConnection # type: ignore +from smb import smb_structs, base # type: ignore +import typing +import io + + +class SMBClient: + """Use pysmb 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.connected = False + self.connect() + + def connect(self) -> None: + if self.connected: + return + try: + self.ctx = SMBConnection( + self.username, + self.password, + "smbclient", + self.server, + use_ntlm_v2=True, + ) + self.ctx.connect(self.server) + self.connected = True + except base.SMBTimeout as error: + raise IOError(f"failed to connect: {error}") + + def disconnect(self) -> None: + self.connected = False + self.ctx.close() + + def listdir(self, path: str = "/") -> typing.List[str]: + try: + dentries = self.ctx.listPath(self.share, path) + except smb_structs.OperationFailure as error: + raise IOError(f"failed to readdir: {error}") + return [dent.filename for dent in dentries] + + def mkdir(self, dpath: str) -> None: + try: + self.ctx.createDirectory(self.share, dpath) + except smb_structs.OperationFailure as error: + raise IOError(f"failed to mkdir: {error}") + + def rmdir(self, dpath: str) -> None: + try: + self.ctx.deleteDirectory(self.share, dpath) + except smb_structs.OperationFailure as error: + raise IOError(f"failed to rmdir: {error}") + + def unlink(self, fpath: str) -> None: + try: + self.ctx.deleteFiles(self.share, fpath) + except smb_structs.OperationFailure as error: + raise IOError(f"failed to unlink: {error}") + + def write_text(self, fpath: str, teststr: str) -> None: + try: + with io.BytesIO(teststr.encode()) as writeobj: + self.ctx.storeFile(self.share, fpath, writeobj) + except smb_structs.OperationFailure as error: + raise IOError(f"failed in write_text: {error}") + + def read_text(self, fpath: str) -> str: + try: + with io.BytesIO() as readobj: + self.ctx.retrieveFile(self.share, fpath, readobj) + ret = readobj.getvalue().decode("utf8") + except smb_structs.OperationFailure as error: + raise IOError(f"failed in read_text: {error}") + return ret diff --git a/tox.ini b/tox.ini index a2646b6..4f91051 100644 --- a/tox.ini +++ b/tox.ini @@ -13,6 +13,7 @@ deps = pyyaml pytest-randomly iso8601 + pysmb commands = pytest -vrfEsxXpP testcases/ [testenv:pytest-unprivileged] @@ -21,6 +22,7 @@ deps = pyyaml pytest-randomly iso8601 + pysmb commands = pytest -vrfEsxXpP -k 'not privileged' testcases/ [testenv:sanity] @@ -29,6 +31,7 @@ deps = pyyaml pytest-randomly iso8601 + pysmb changedir = {toxinidir} commands = pytest -vrfEsxXpP testcases/consistency