testhelper: introduce class SMBClient using pysmb

We use the python module pysmb 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 <[email protected]>

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