general: improve commands handling, pass CompletedProcess instead of dummy...

general: improve commands handling, pass CompletedProcess instead of dummy dto, finalize basic restic coverage

rwm.py

@@ -45,18 +45,17 @@ def run_command(*args, **kwargs):
         "encoding": "utf-8",
     })
     logger.debug("run_command, %s", (args, kwargs))
-    proc = subprocess.run(*args, **kwargs, check=False)
-    return (proc.returncode, proc.stdout, proc.stderr)
+    return subprocess.run(*args, **kwargs, check=False)
 
 
-def wrap_output(returncode, stdout, stderr):
+def wrap_output(process):
     """wraps command output and prints results"""
 
-    if stdout:
-        print(stdout)
-    if stderr:
-        print(stderr, file=sys.stderr)
-    return returncode
+    if process.stdout:
+        print(process.stdout)
+    if process.stderr:
+        print(process.stderr, file=sys.stderr)
+    return process.returncode
 
 
 def rclone_obscure_password(plaintext, iv=None):
@@ -81,13 +80,7 @@ class RWM:
         self.config = config
 
     def aws_cmd(self, args):
-        """
-        aws cli wrapper
-
-        :param list args: list passed to subprocess
-        :return: returncode, stdout, stderr
-        :rtype: tuple
-        """
+        """aws cli wrapper"""
 
         env = {
             "PATH": os.environ["PATH"],
@@ -103,13 +96,7 @@ class RWM:
         return run_command(["aws", "--endpoint-url", self.config["RWM_S3_ENDPOINT_URL"]] + args, env=env)
 
     def rclone_cmd(self, args):
-        """
-        rclone wrapper
-
-        :param list args: list passed to subprocess
-        :return: returncode, stdout, stderr
-        :rtype: tuple
-        """
+        """rclone wrapper"""
 
         env = {
             "RCLONE_CONFIG": "",
@@ -128,10 +115,6 @@ class RWM:
         rclone crypt wrapper
         * https://rclone.org/docs/#config-file
         * https://rclone.org/crypt/
-
-        :param list args: list passed to subprocess
-        :return: returncode, stdout, stderr
-        :rtype: tuple
         """
 
         env = {
@@ -190,13 +173,13 @@ def main(argv=None):
     rwm = RWM(config)
 
     if args.command == "aws":
-        return wrap_output(*rwm.aws_cmd(args.cmd_args))
+        return wrap_output(rwm.aws_cmd(args.cmd_args))
     if args.command == "rclone":
-        return wrap_output(*rwm.rclone_cmd(args.cmd_args))
+        return wrap_output(rwm.rclone_cmd(args.cmd_args))
     if args.command == "rclone_crypt":
-        return wrap_output(*rwm.rclone_crypt_cmd(args.cmd_args))
+        return wrap_output(rwm.rclone_crypt_cmd(args.cmd_args))
     if args.command == "restic":
-        return wrap_output(*rwm.restic_cmd(args.cmd_args))
+        return wrap_output(rwm.restic_cmd(args.cmd_args))
 
     return 0
 

tests/test_default.py

 """default tests"""
 
+import json
 from pathlib import Path
-from textwrap import dedent
+from subprocess import CompletedProcess
 from unittest.mock import Mock, patch
 
 import boto3
@@ -32,7 +33,7 @@ def test_sublist():
 def test_wrap_output():
     """test wrap_output"""
 
-    assert wrap_output(11, "dummy", "dummy") == 11
+    assert wrap_output(CompletedProcess(args='dummy', returncode=11, stdout="dummy", stderr="dummy")) == 11
 
 
 def test_main(tmpworkdir: str):  # pylint: disable=unused-argument
@@ -44,7 +45,7 @@ def test_main(tmpworkdir: str):  # pylint: disable=unused-argument
     assert rwm_main([]) == 0
 
     # command branches
-    mock = Mock(return_value=(0, "", ""))
+    mock = Mock(return_value=CompletedProcess(args='dummy', returncode=0))
     for item in ["aws", "rclone", "rclone_crypt", "restic"]:
         with patch.object(rwm.RWM, f"{item}_cmd", mock):
             assert rwm_main([item]) == 0
@@ -53,7 +54,7 @@ def test_main(tmpworkdir: str):  # pylint: disable=unused-argument
 def test_aws_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=unused-argument
     """test aws command"""
 
-    rwm = RWM({
+    trwm = RWM({
         "RWM_S3_ENDPOINT_URL": motoserver,
         "RWM_S3_ACCESS_KEY": "dummy",
         "RWM_S3_SECRET_KEY": "dummy",
@@ -63,17 +64,17 @@ def test_aws_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=unused-ar
 
     assert test_bucket not in buckets_plain_list(s3.list_buckets())
 
-    rwm.aws_cmd(["s3", "mb", f"s3://{test_bucket}"])
+    trwm.aws_cmd(["s3", "mb", f"s3://{test_bucket}"])
     assert test_bucket in buckets_plain_list(s3.list_buckets())
 
-    rwm.aws_cmd(["s3", "rb", f"s3://{test_bucket}"])
+    trwm.aws_cmd(["s3", "rb", f"s3://{test_bucket}"])
     assert test_bucket not in buckets_plain_list(s3.list_buckets())
 
 
 def test_rclone_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=unused-argument
     """test rclone command"""
 
-    rwm = RWM({
+    trwm = RWM({
         "RWM_S3_ENDPOINT_URL": motoserver,
         "RWM_S3_ACCESS_KEY": "dummy",
         "RWM_S3_SECRET_KEY": "dummy",
@@ -84,8 +85,8 @@ def test_rclone_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=unused
     test_file = "testfile.txt"
     Path(test_file).write_text('1234', encoding='utf-8')
 
-    rwm.rclone_cmd(["mkdir", f"rwmbe:/{test_bucket}/"])
-    rwm.rclone_cmd(["copy", test_file, f"rwmbe:/{test_bucket}/"])
+    trwm.rclone_cmd(["mkdir", f"rwmbe:/{test_bucket}/"])
+    trwm.rclone_cmd(["copy", test_file, f"rwmbe:/{test_bucket}/"])
     assert test_bucket in buckets_plain_list(s3.list_buckets())
     assert test_file in objects_plain_list(s3.list_objects_v2(Bucket=test_bucket))
 
@@ -93,7 +94,7 @@ def test_rclone_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=unused
 def test_rclone_crypt_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=unused-argument
     """test rclone with crypt overlay"""
 
-    rwm = RWM({
+    trwm = RWM({
         "RWM_S3_ENDPOINT_URL": motoserver,
         "RWM_S3_ACCESS_KEY": "dummy",
         "RWM_S3_SECRET_KEY": "dummy",
@@ -106,51 +107,33 @@ def test_rclone_crypt_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=
     test_file = "testfile.txt"
     Path(test_file).write_text('1234', encoding='utf-8')
 
-    rwm.rclone_crypt_cmd(["copy", test_file, f"rwmbe:/{test_bucket}/"])
-    assert len(objects_plain_list(s3.list_objects_v2(Bucket=rwm.config["RWM_RCLONE_CRYPT_BUCKET"]))) == 1
+    trwm.rclone_crypt_cmd(["copy", test_file, f"rwmbe:/{test_bucket}/"])
+    assert len(objects_plain_list(s3.list_objects_v2(Bucket=trwm.config["RWM_RCLONE_CRYPT_BUCKET"]))) == 1
 
-    rwm.rclone_crypt_cmd(["delete", f"rwmbe:/{test_bucket}/{test_file}"])
-    assert s3.list_objects_v2(Bucket=rwm.config["RWM_RCLONE_CRYPT_BUCKET"])["KeyCount"] == 0
+    trwm.rclone_crypt_cmd(["delete", f"rwmbe:/{test_bucket}/{test_file}"])
+    assert s3.list_objects_v2(Bucket=trwm.config["RWM_RCLONE_CRYPT_BUCKET"])["KeyCount"] == 0
 
     test_file1 = "testfile1.txt"
     Path(test_file1).write_text('4321', encoding='utf-8')
-    rwm.rclone_crypt_cmd(["sync", ".", f"rwmbe:/{test_bucket}/"])
-    assert s3.list_objects_v2(Bucket=rwm.config["RWM_RCLONE_CRYPT_BUCKET"])["KeyCount"] == 2
+    trwm.rclone_crypt_cmd(["sync", ".", f"rwmbe:/{test_bucket}/"])
+    assert s3.list_objects_v2(Bucket=trwm.config["RWM_RCLONE_CRYPT_BUCKET"])["KeyCount"] == 2
 
     Path(test_file1).unlink()
-    rwm.rclone_crypt_cmd(["sync", ".", f"rwmbe:/{test_bucket}/"])
-    assert s3.list_objects_v2(Bucket=rwm.config["RWM_RCLONE_CRYPT_BUCKET"])["KeyCount"] == 1
-
-
-# def test_restic_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=unused-argument
-#    """test rclone with crypt overlay"""
-#
-#    rwm_conf = {
-#        "S3_ENDPOINT_URL": motoserver,
-#        "S3_ACCESS_KEY": "dummy",
-#        "S3_SECRET_KEY": "dummy",
-#        "RES_BUCKET": "restic_test",
-#        "RES_PASSWORD": "dummydummydummydummydummydummydummydummy",
-#    }
-#    s3 = boto3.client('s3', endpoint_url=motoserver, aws_access_key_id="dummy", aws_secret_access_key="dummy")
-#
-#    test_bucket = "testbucket"
-#    test_file = "testfile.txt"
-#    Path(test_file).write_text('1234', encoding='utf-8')
-#
-#    rwm_main(["res", "init"], rwm_conf)
-#    assert len(objects_plain_list(s3.list_objects_v2(Bucket=rwm_conf["RES_BUCKET"]))) == 1
-
-
-#
-#    rwm_main(["rcc", "delete", f"rwmbe:/{test_bucket}/{test_file}"], rwm_conf)
-#    assert s3.list_objects_v2(Bucket=rwm_conf["RCC_CRYPT_BUCKET"])["KeyCount"] == 0
-#
-#    test_file1 = "testfile1.txt"
-#    Path(test_file1).write_text('4321', encoding='utf-8')
-#    rwm_main(["rcc", "sync", ".", f"rwmbe:/{test_bucket}/"], rwm_conf)
-#    assert s3.list_objects_v2(Bucket=rwm_conf["RCC_CRYPT_BUCKET"])["KeyCount"] == 2
-#
-#    Path(test_file1).unlink()
-#    rwm_main(["rcc", "sync", ".", f"rwmbe:/{test_bucket}/"], rwm_conf)
-#    assert s3.list_objects_v2(Bucket=rwm_conf["RCC_CRYPT_BUCKET"])["KeyCount"] == 1
+    trwm.rclone_crypt_cmd(["sync", ".", f"rwmbe:/{test_bucket}/"])
+    assert s3.list_objects_v2(Bucket=trwm.config["RWM_RCLONE_CRYPT_BUCKET"])["KeyCount"] == 1
+
+
+def test_restic_cmd(tmpworkdir: str, motoserver: str):  # pylint: disable=unused-argument
+    """test restic command"""
+
+    trwm = RWM({
+        "RWM_S3_ENDPOINT_URL": motoserver,
+        "RWM_S3_ACCESS_KEY": "dummy",
+        "RWM_S3_SECRET_KEY": "dummy",
+        "RWM_RESTIC_BUCKET": "restictest",
+        "RWM_RESTIC_PASSWORD": "dummydummydummydummydummydummydummydummy",
+    })
+
+    assert trwm.restic_cmd(["init"]).returncode == 0
+    proc = trwm.restic_cmd(["cat", "config"])
+    assert "id" in json.loads(proc.stdout)