Skip to content
Snippets Groups Projects
Commit bf3b99c7 authored by Radoslav Bodó's avatar Radoslav Bodó
Browse files

general: refactor config, splitup tests to be more unit

parent 9a5f18fb
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
rwm.conf.example
+ 10
− 5
View file @ bf3b99c7
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
rwm.py
+ 16
− 16
View file @ bf3b99c7
......@@ -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)
......
tests/test_default.py
+ 31
− 39
View file @ bf3b99c7
......@@ -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
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment