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
+    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)

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,82 @@
+from smb.SMBConnection import SMBConnection  # type: ignore
+import smb  # 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.error = False
+        self.connected = False
+        self.connect()
+
+    def connect(self) -> None:
+        try:
+            self.ctx = SMBConnection(
+                self.username,
+                self.password,
+                "smbclient",
+                self.server,
+                use_ntlm_v2=True,
+            )
+            self.ctx.connect(self.server)
+            self.connected = True
+        except smb.base.SMBTimeout:
+            self.error = False
+
+    def disconnect(self) -> None:
+        self.connected = False
+        self.ctx.close()
+
+    def readdir(self, path: str = "/") -> typing.List:
+        ret = []
+        try:
+            dentries = self.ctx.listPath(self.share, path)
+        except smb.smb_structs.OperationFailure:
+            self.error = True
+            raise
+        else:
+            ret = [dent.filename for dent in dentries]
+        return ret
+
+    def mkdir(self, dpath: str) -> None:
+        try:
+            self.ctx.createDirectory(self.share, dpath)
+        except smb.smb_structs.OperationFailure:
+            self.error = True
+            raise
+
+    def rmdir(self, dpath: str) -> None:
+        self.ctx.deleteDirectory(self.share, dpath)
+
+    def unlink(self, fpath: str) -> None:
+        self.ctx.deleteFiles(self.share, fpath)
+
+    def simple_write(self, fpath: str, teststr: str) -> None:
+        file_obj = io.BytesIO(teststr.encode())
+        try:
+            self.ctx.storeFile(self.share, fpath, file_obj)
+        except smb.smb_structs.OperationFailure:
+            self.error = True
+            raise
+        finally:
+            file_obj.close()
+
+    def simple_read(self, fpath: str) -> str:
+        readobj = io.BytesIO()
+        try:
+            self.ctx.retrieveFile(self.share, fpath, readobj)
+        except smb.smb_structs.OperationFailure:
+            self.error = True
+            raise
+        else:
+            ret = readobj.getvalue().decode("ascii")
+        finally:
+            readobj.close()
+        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