testhelper: introduce class SMBClient - pysmb version

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)

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

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.
 

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

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