diff --git a/rwm.py b/rwm.py
index 874b241d7b46511028eb63de76a998c41c1b2147..9ff8ffd21d2df996b5d6e9cc77ad1e1f34f428ac 100755
--- a/rwm.py
+++ b/rwm.py
@@ -4,16 +4,21 @@
import base64
import logging
import os
+import subprocess
import sys
from argparse import ArgumentParser
from pathlib import Path
-from subprocess import run as subrun
import yaml
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
+__version__ = "0.1"
+logger = logging.getLogger("rwm")
+logger.setLevel(logging.DEBUG)
+
+
def is_sublist(needle, haystack):
"""Check if needle is a sublist of haystack using list slicing and equality comparison"""
@@ -31,6 +36,29 @@ def get_config(path):
return {}
+def run_command(*args, **kwargs):
+ """output capturing command executor"""
+
+ kwargs.update({
+ "capture_output": True,
+ "text": True,
+ "encoding": "utf-8",
+ })
+ logger.debug("run_command, %s", (args, kwargs))
+ proc = subprocess.run(*args, **kwargs, check=False)
+ return (proc.returncode, proc.stdout, proc.stderr)
+
+
+def wrap_output(returncode, stdout, stderr):
+ """wraps command output and prints results"""
+
+ if stdout:
+ print(stdout)
+ if stderr:
+ print(stderr, file=sys.stderr)
+ return returncode
+
+
def rclone_obscure_password(plaintext, iv=None):
"""rclone obscure password algorithm"""
@@ -53,7 +81,13 @@ class RWM:
self.config = config
def aws_cmd(self, args):
- """aws cli wrapper"""
+ """
+ aws cli wrapper
+
+ :param list args: list passed to subprocess
+ :return: returncode, stdout, stderr
+ :rtype: tuple
+ """
env = {
"PATH": os.environ["PATH"],
@@ -66,10 +100,16 @@ class RWM:
env.update({"AWS_DEFAULT_REGION": ""})
# aws cli does not have endpoint-url as env config option
- return subrun(["aws", "--endpoint-url", self.config["S3_ENDPOINT_URL"]] + args, env=env, check=False).returncode
+ return run_command(["aws", "--endpoint-url", self.config["S3_ENDPOINT_URL"]] + args, env=env)
def rclone_cmd(self, args):
- """rclone wrapper"""
+ """
+ rclone wrapper
+
+ :param list args: list passed to subprocess
+ :return: returncode, stdout, stderr
+ :rtype: tuple
+ """
env = {
"RCLONE_CONFIG": "",
@@ -81,33 +121,39 @@ class RWM:
"RCLONE_CONFIG_RWMBE_ENV_AUTH": "false",
"RCLONE_CONFIG_RWMBE_REGION": "",
}
- return subrun(["rclone"] + args, env=env, check=False).returncode
+ return run_command(["rclone"] + args, env=env)
def rclone_crypt_cmd(self, args):
"""
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 = {
"RCLONE_CONFIG": "",
"RCLONE_CONFIG_RWMBE_TYPE": "crypt",
- "RCLONE_CONFIG_RWMBE_REMOTE": f"rwms3be:/{self.config['RCC_CRYPT_BUCKET']}",
+ "RCLONE_CONFIG_RWMBE_REMOTE": f"rwmbes3:/{self.config['RCC_CRYPT_BUCKET']}",
"RCLONE_CONFIG_RWMBE_PASSWORD": rclone_obscure_password(self.config["RCC_CRYPT_PASSWORD"]),
"RCLONE_CONFIG_RWMBE_PASSWORD2": rclone_obscure_password(self.config["RCC_CRYPT_PASSWORD"]),
- "RCLONE_CONFIG_RWMS3BE_TYPE": "s3",
- "RCLONE_CONFIG_RWMS3BE_ENDPOINT": self.config["S3_ENDPOINT_URL"],
- "RCLONE_CONFIG_RWMS3BE_ACCESS_KEY_ID": self.config["S3_ACCESS_KEY"],
- "RCLONE_CONFIG_RWMS3BE_SECRET_ACCESS_KEY": self.config["S3_SECRET_KEY"],
- "RCLONE_CONFIG_RWMS3BE_PROVIDER": "Ceph",
- "RCLONE_CONFIG_RWMS3BE_ENV_AUTH": "false",
- "RCLONE_CONFIG_RWMS3BE_REGION": "",
+ "RCLONE_CONFIG_RWMBES3_TYPE": "s3",
+ "RCLONE_CONFIG_RWMBES3_ENDPOINT": self.config["S3_ENDPOINT_URL"],
+ "RCLONE_CONFIG_RWMBES3_ACCESS_KEY_ID": self.config["S3_ACCESS_KEY"],
+ "RCLONE_CONFIG_RWMBES3_SECRET_ACCESS_KEY": self.config["S3_SECRET_KEY"],
+ "RCLONE_CONFIG_RWMBES3_PROVIDER": "Ceph",
+ "RCLONE_CONFIG_RWMBES3_ENV_AUTH": "false",
+ "RCLONE_CONFIG_RWMBES3_REGION": "",
}
- return subrun(["rclone"] + args, env=env, check=False).returncode
+ return run_command(["rclone"] + args, env=env)
def restic_cmd(self, args):
+ """restic command wrapper"""
+
env = {
"HOME": os.environ["HOME"],
"PATH": os.environ["PATH"],
@@ -116,9 +162,10 @@ class RWM:
"RESTIC_PASSWORD": self.config["RES_PASSWORD"],
"RESTIC_REPOSITORY": f"s3:{self.config['S3_ENDPOINT_URL']}/{self.config['RES_BUCKET']}",
}
- return subrun(["restic"] + args, env=env, check=False).returncode
+ return run_command(["restic"] + args, env=env)
+
-def main(argv=None, dict_config=None):
+def main(argv=None):
"""main"""
parser = ArgumentParser(description="restics3 worm manager")
@@ -127,11 +174,11 @@ def main(argv=None, dict_config=None):
subparsers = parser.add_subparsers(title="commands", dest="command", required=False)
aws_cmd_parser = subparsers.add_parser("aws", help="aws command")
aws_cmd_parser.add_argument("cmd_args", nargs="*")
- rc_cmd_parser = subparsers.add_parser("rc", help="rclone command")
+ rc_cmd_parser = subparsers.add_parser("rclone", help="rclone command")
rc_cmd_parser.add_argument("cmd_args", nargs="*")
- rcc_cmd_parser = subparsers.add_parser("rcc", help="rclone command with crypt overlay")
+ rcc_cmd_parser = subparsers.add_parser("rclone_crypt", help="rclone command with crypt overlay")
rcc_cmd_parser.add_argument("cmd_args", nargs="*")
- res_cmd_parser = subparsers.add_parser("res", help="restic command")
+ res_cmd_parser = subparsers.add_parser("restic", help="restic command")
res_cmd_parser.add_argument("cmd_args", nargs="*")
args = parser.parse_args(argv)
@@ -139,19 +186,17 @@ def main(argv=None, dict_config=None):
config = {}
if args.config:
config.update(get_config(args.config))
- if dict_config:
- config.update(dict_config)
# assert config ?
rwm = RWM(config)
if args.command == "aws":
- return rwm.aws_cmd(args.cmd_args)
- if args.command == "rc":
- return rwm.rclone_cmd(args.cmd_args)
- if args.command == "rcc":
- return rwm.rclone_crypt_cmd(args.cmd_args)
- if args.command == "res":
- return rwm.restic_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))
+ if args.command == "rclone_crypt":
+ return wrap_output(*rwm.rclone_crypt_cmd(args.cmd_args))
+ if args.command == "restic":
+ return wrap_output(*rwm.restic_cmd(args.cmd_args))
return 0
diff --git a/tests/test_default.py b/tests/test_default.py
index e0a0249898ab2ce0ba3a35793e3ea5dd9b1b5074..3fa46a95e74147eee13fbfc249ce266de60083c6 100644
--- a/tests/test_default.py
+++ b/tests/test_default.py
@@ -1,9 +1,22 @@
"""default tests"""
from pathlib import Path
+from textwrap import dedent
import boto3
-from rwm import is_sublist, main as rwm_main, rclone_obscure_password
+from rwm import is_sublist, main as rwm_main, rclone_obscure_password, RWM
+
+
+def buckets_plain_list(full_response):
+ """boto3 helper"""
+
+ return [x["Name"] for x in full_response["Buckets"]]
+
+
+def objects_plain_list(full_response):
+ """boto3 helper"""
+
+ return [x["Key"] for x in full_response["Contents"]]
def test_sublist():
@@ -14,104 +27,138 @@ def test_sublist():
assert not is_sublist([1, 3], [5, 4, 1, 2, 3, 6, 7])
-def test_config(tmpworkdir: str): # pylint: disable=unused-argument
- """test config handling"""
+def test_main(tmpworkdir: str): # pylint: disable=unused-argument
+ """test main"""
- Path("rwm.conf").touch()
- rwm_main([])
+ assert rwm_main([]) == 0
+ Path("rwm.conf").write_text(
+ dedent("""
+ S3_ENDPOINT_URL: "dummy"
+ S3_ACCESS_KEY: "dummy"
+ S3_SECRET_KEY: "dummy"
-def buckets_plain_list(full_response):
- """boto3 helper"""
+ RCC_CRYPT_BUCKET: "dummy-nasbackup-test1"
+ RCC_CRYPT_PASSWORD: "dummy"
- return [x["Name"] for x in full_response["Buckets"]]
+ RES_BUCKET: "dummy"
+ RES_PASSWORD: "dummy"
+ """),
+ encoding="utf-8"
+ )
+ assert rwm_main([]) == 0
-def objects_plain_list(full_response):
- """boto3 helper"""
+ assert rwm_main(["aws", "--", "--version"]) == 0
+ assert rwm_main(["aws", "notexist"]) != 0
- return [x["Key"] for x in full_response["Contents"]]
+ assert rwm_main(["rclone", "version"]) == 0
+ assert rwm_main(["rclone_crypt", "version"]) == 0
+
+ assert rwm_main(["restic", "version"]) == 0
-def test_aws(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
+def test_aws_cmd(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
"""test aws command"""
- rwm_conf = {
+ rwm = RWM({
"S3_ENDPOINT_URL": motoserver,
"S3_ACCESS_KEY": "dummy",
"S3_SECRET_KEY": "dummy",
- }
+ })
s3 = boto3.client('s3', endpoint_url=motoserver, aws_access_key_id="dummy", aws_secret_access_key="dummy")
test_bucket = "testbucket"
assert test_bucket not in buckets_plain_list(s3.list_buckets())
- rwm_main(["aws", "s3", "mb", f"s3://{test_bucket}"], rwm_conf)
+ rwm.aws_cmd(["s3", "mb", f"s3://{test_bucket}"])
assert test_bucket in buckets_plain_list(s3.list_buckets())
- rwm_main(["aws", "s3", "rb", f"s3://{test_bucket}"], rwm_conf)
+ rwm.aws_cmd(["s3", "rb", f"s3://{test_bucket}"])
assert test_bucket not in buckets_plain_list(s3.list_buckets())
-def test_rclone(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
+def test_rclone_cmd(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
"""test rclone command"""
- rwm_conf = {
+ rwm = RWM({
"S3_ENDPOINT_URL": motoserver,
"S3_ACCESS_KEY": "dummy",
"S3_SECRET_KEY": "dummy",
- }
+ })
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(["rc", "mkdir", f"rwmbe:/{test_bucket}/"], rwm_conf)
- rwm_main(["rc", "copy", test_file, f"rwmbe:/{test_bucket}/"], rwm_conf)
+ rwm.rclone_cmd(["mkdir", f"rwmbe:/{test_bucket}/"])
+ rwm.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))
-def test_rclone_argscheck():
- """test rclone args checking"""
-
- assert rwm_main(["rc", "dummy"]) == 1
-
-
-def test_rclone_crypt(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
+def test_rclone_crypt_cmd(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
"""test rclone with crypt overlay"""
- rwm_conf = {
+ rwm = RWM({
"S3_ENDPOINT_URL": motoserver,
"S3_ACCESS_KEY": "dummy",
"S3_SECRET_KEY": "dummy",
"RCC_CRYPT_BUCKET": "cryptdata_test",
"RCC_CRYPT_PASSWORD": rclone_obscure_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(["rcc", "copy", test_file, f"rwmbe:/{test_bucket}/"], rwm_conf)
- assert len(objects_plain_list(s3.list_objects_v2(Bucket=rwm_conf["RCC_CRYPT_BUCKET"]))) == 1
+ rwm.rclone_crypt_cmd(["copy", test_file, f"rwmbe:/{test_bucket}/"])
+ assert len(objects_plain_list(s3.list_objects_v2(Bucket=rwm.config["RCC_CRYPT_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
+ rwm.rclone_crypt_cmd(["delete", f"rwmbe:/{test_bucket}/{test_file}"])
+ assert s3.list_objects_v2(Bucket=rwm.config["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
+ rwm.rclone_crypt_cmd(["sync", ".", f"rwmbe:/{test_bucket}/"])
+ assert s3.list_objects_v2(Bucket=rwm.config["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
-
-
-def test_rclone_crypt_argscheck():
- """test rclone crypt args checking"""
-
- assert rwm_main(["rcc", "dummy"]) == 1
+ rwm.rclone_crypt_cmd(["sync", ".", f"rwmbe:/{test_bucket}/"])
+ assert s3.list_objects_v2(Bucket=rwm.config["RCC_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