diff --git a/rwm.conf.example b/rwm.conf.example index 06b9654719bf3dde1882825a04a3ef4e1e93e6f6..6413b7d2bb70829e3b07d6fb01c8011772846e65 100644 --- a/rwm.conf.example +++ b/rwm.conf.example @@ -1,6 +1,11 @@ -S3_ENDPOINT_URL: "" -S3_ACCESS_KEY: "" -S3_SECRET_KEY: "" +RWM_S3_ENDPOINT_URL: "" +RWM_S3_ACCESS_KEY: "" +RWM_S3_SECRET_KEY: "" -RCC_CRYPT_BUCKET: "rwmcrypt" -RCC_CRYPT_PASSWORD: "" +# RWM_RCLONE + +RWM_RCLONE_CRYPT_BUCKET: "rwmcrypt" +RWM_RCLONE_CRYPT_PASSWORD: "" + +RWM_RESTIC_BUCKET: "rwmcrypt" +RWM_RESTIC_PASSWORD: "" \ No newline at end of file diff --git a/rwm.py b/rwm.py index 9ff8ffd21d2df996b5d6e9cc77ad1e1f34f428ac..3cc6f74148bdc7492b5517d5032d6b54805d1783 100755 --- a/rwm.py +++ b/rwm.py @@ -92,15 +92,15 @@ class RWM: env = { "PATH": os.environ["PATH"], "AWS_METADATA_SERVICE_NUM_ATTEMPTS": "0", - "AWS_ACCESS_KEY_ID": self.config["S3_ACCESS_KEY"], - "AWS_SECRET_ACCESS_KEY": self.config["S3_SECRET_KEY"] + "AWS_ACCESS_KEY_ID": self.config["RWM_S3_ACCESS_KEY"], + "AWS_SECRET_ACCESS_KEY": self.config["RWM_S3_SECRET_KEY"] } if is_sublist(["s3", "mb"], args): # region must be set and empty for awscil >=2.x and ?du? ceph s3 env.update({"AWS_DEFAULT_REGION": ""}) # aws cli does not have endpoint-url as env config option - return run_command(["aws", "--endpoint-url", self.config["S3_ENDPOINT_URL"]] + args, env=env) + return run_command(["aws", "--endpoint-url", self.config["RWM_S3_ENDPOINT_URL"]] + args, env=env) def rclone_cmd(self, args): """ @@ -114,9 +114,9 @@ class RWM: env = { "RCLONE_CONFIG": "", "RCLONE_CONFIG_RWMBE_TYPE": "s3", - "RCLONE_CONFIG_RWMBE_ENDPOINT": self.config["S3_ENDPOINT_URL"], - "RCLONE_CONFIG_RWMBE_ACCESS_KEY_ID": self.config["S3_ACCESS_KEY"], - "RCLONE_CONFIG_RWMBE_SECRET_ACCESS_KEY": self.config["S3_SECRET_KEY"], + "RCLONE_CONFIG_RWMBE_ENDPOINT": self.config["RWM_S3_ENDPOINT_URL"], + "RCLONE_CONFIG_RWMBE_ACCESS_KEY_ID": self.config["RWM_S3_ACCESS_KEY"], + "RCLONE_CONFIG_RWMBE_SECRET_ACCESS_KEY": self.config["RWM_S3_SECRET_KEY"], "RCLONE_CONFIG_RWMBE_PROVIDER": "Ceph", "RCLONE_CONFIG_RWMBE_ENV_AUTH": "false", "RCLONE_CONFIG_RWMBE_REGION": "", @@ -137,14 +137,14 @@ class RWM: env = { "RCLONE_CONFIG": "", "RCLONE_CONFIG_RWMBE_TYPE": "crypt", - "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_RWMBE_REMOTE": f"rwmbes3:/{self.config['RWM_RCLONE_CRYPT_BUCKET']}", + "RCLONE_CONFIG_RWMBE_PASSWORD": rclone_obscure_password(self.config["RWM_RCLONE_CRYPT_PASSWORD"]), + "RCLONE_CONFIG_RWMBE_PASSWORD2": rclone_obscure_password(self.config["RWM_RCLONE_CRYPT_PASSWORD"]), "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_ENDPOINT": self.config["RWM_S3_ENDPOINT_URL"], + "RCLONE_CONFIG_RWMBES3_ACCESS_KEY_ID": self.config["RWM_S3_ACCESS_KEY"], + "RCLONE_CONFIG_RWMBES3_SECRET_ACCESS_KEY": self.config["RWM_S3_SECRET_KEY"], "RCLONE_CONFIG_RWMBES3_PROVIDER": "Ceph", "RCLONE_CONFIG_RWMBES3_ENV_AUTH": "false", "RCLONE_CONFIG_RWMBES3_REGION": "", @@ -157,10 +157,10 @@ class RWM: env = { "HOME": os.environ["HOME"], "PATH": os.environ["PATH"], - "AWS_ACCESS_KEY_ID": self.config["S3_ACCESS_KEY"], - "AWS_SECRET_ACCESS_KEY": self.config["S3_SECRET_KEY"], - "RESTIC_PASSWORD": self.config["RES_PASSWORD"], - "RESTIC_REPOSITORY": f"s3:{self.config['S3_ENDPOINT_URL']}/{self.config['RES_BUCKET']}", + "AWS_ACCESS_KEY_ID": self.config["RWM_S3_ACCESS_KEY"], + "AWS_SECRET_ACCESS_KEY": self.config["RWM_S3_SECRET_KEY"], + "RESTIC_PASSWORD": self.config["RWM_RESTIC_PASSWORD"], + "RESTIC_REPOSITORY": f"s3:{self.config['RWM_S3_ENDPOINT_URL']}/{self.config['RWM_RESTIC_BUCKET']}", } return run_command(["restic"] + args, env=env) diff --git a/tests/test_default.py b/tests/test_default.py index 3fa46a95e74147eee13fbfc249ce266de60083c6..504abc96f3fca44f6c9ed4ac6d547e2a2792e507 100644 --- a/tests/test_default.py +++ b/tests/test_default.py @@ -2,9 +2,11 @@ from pathlib import Path from textwrap import dedent +from unittest.mock import Mock, patch import boto3 -from rwm import is_sublist, main as rwm_main, rclone_obscure_password, RWM +import rwm +from rwm import is_sublist, main as rwm_main, rclone_obscure_password, RWM, wrap_output def buckets_plain_list(full_response): @@ -27,44 +29,34 @@ def test_sublist(): assert not is_sublist([1, 3], [5, 4, 1, 2, 3, 6, 7]) -def test_main(tmpworkdir: str): # pylint: disable=unused-argument - """test main""" +def test_wrap_output(): + """test wrap_output""" - assert rwm_main([]) == 0 + assert wrap_output(11, "dummy", "dummy") == 11 - Path("rwm.conf").write_text( - dedent(""" - S3_ENDPOINT_URL: "dummy" - S3_ACCESS_KEY: "dummy" - S3_SECRET_KEY: "dummy" - RCC_CRYPT_BUCKET: "dummy-nasbackup-test1" - RCC_CRYPT_PASSWORD: "dummy" - - RES_BUCKET: "dummy" - RES_PASSWORD: "dummy" - """), - encoding="utf-8" - ) +def test_main(tmpworkdir: str): # pylint: disable=unused-argument + """test main""" + # optional and default config hanling + assert rwm_main([]) == 0 + Path("rwm.conf").touch() assert rwm_main([]) == 0 - assert rwm_main(["aws", "--", "--version"]) == 0 - assert rwm_main(["aws", "notexist"]) != 0 - - assert rwm_main(["rclone", "version"]) == 0 - assert rwm_main(["rclone_crypt", "version"]) == 0 - - assert rwm_main(["restic", "version"]) == 0 + # command branches + mock = Mock(return_value=(0, "", "")) + for item in ["aws", "rclone", "rclone_crypt", "restic"]: + with patch.object(rwm.RWM, f"{item}_cmd", mock): + assert rwm_main([item]) == 0 def test_aws_cmd(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument """test aws command""" rwm = RWM({ - "S3_ENDPOINT_URL": motoserver, - "S3_ACCESS_KEY": "dummy", - "S3_SECRET_KEY": "dummy", + "RWM_S3_ENDPOINT_URL": motoserver, + "RWM_S3_ACCESS_KEY": "dummy", + "RWM_S3_SECRET_KEY": "dummy", }) s3 = boto3.client('s3', endpoint_url=motoserver, aws_access_key_id="dummy", aws_secret_access_key="dummy") test_bucket = "testbucket" @@ -82,9 +74,9 @@ def test_rclone_cmd(tmpworkdir: str, motoserver: str): # pylint: disable=unused """test rclone command""" rwm = RWM({ - "S3_ENDPOINT_URL": motoserver, - "S3_ACCESS_KEY": "dummy", - "S3_SECRET_KEY": "dummy", + "RWM_S3_ENDPOINT_URL": motoserver, + "RWM_S3_ACCESS_KEY": "dummy", + "RWM_S3_SECRET_KEY": "dummy", }) s3 = boto3.client('s3', endpoint_url=motoserver, aws_access_key_id="dummy", aws_secret_access_key="dummy") @@ -102,11 +94,11 @@ def test_rclone_crypt_cmd(tmpworkdir: str, motoserver: str): # pylint: disable= """test rclone with crypt overlay""" 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"), + "RWM_S3_ENDPOINT_URL": motoserver, + "RWM_S3_ACCESS_KEY": "dummy", + "RWM_S3_SECRET_KEY": "dummy", + "RWM_RCLONE_CRYPT_BUCKET": "cryptdata_test", + "RWM_RCLONE_CRYPT_PASSWORD": rclone_obscure_password("dummydummydummydummydummydummydummydummy"), }) s3 = boto3.client('s3', endpoint_url=motoserver, aws_access_key_id="dummy", aws_secret_access_key="dummy") @@ -115,19 +107,19 @@ def test_rclone_crypt_cmd(tmpworkdir: str, motoserver: str): # pylint: disable= 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["RCC_CRYPT_BUCKET"]))) == 1 + assert len(objects_plain_list(s3.list_objects_v2(Bucket=rwm.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["RCC_CRYPT_BUCKET"])["KeyCount"] == 0 + assert s3.list_objects_v2(Bucket=rwm.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["RCC_CRYPT_BUCKET"])["KeyCount"] == 2 + assert s3.list_objects_v2(Bucket=rwm.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["RCC_CRYPT_BUCKET"])["KeyCount"] == 1 + 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