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 <[email protected]>

testcases/consistency/test_consistency.py

@@ -7,72 +7,54 @@
 # 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)
+    del scon
+
+    # file read cycle
+    scon = SMBClient(
+        hostname,
+        share_name,
+        mount_params["username"],
+        mount_params["password"],
+    )
+    retstr = scon.simple_read(test_filename)
+    scon.unlink(test_filename)
+    del scon
+
+    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)
+    "hostname,share_name", generate_consistency_check()
 )
-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)
+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,50 @@
+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 readdir(self, path: str = "/") -> typing.List:
+        uri = self.rooturi + path
+        d = self.ctx.opendir(uri)
+        return d.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")

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