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

general: logging and backup tests

parent 7d610afc
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
rwm.conf.example
+ 8
− 1
View file @ 276698ec
# all commands
RWM_S3_ENDPOINT_URL: ""
RWM_S3_ACCESS_KEY: ""
RWM_S3_SECRET_KEY: ""
# rclone_crypt
RWM_RCLONE_CRYPT_BUCKET: "rwmcrypt"
RWM_RCLONE_CRYPT_PASSWORD: ""
# restic, backup
RWM_RESTIC_BUCKET: "rwmcrypt"
RWM_RESTIC_PASSWORD: ""
\ No newline at end of file
RWM_RESTIC_PASSWORD: ""
# backup
RWM_BACKUPS: []
RWM_RETENTION: []
rwm.py
+ 83
− 11
View file @ 276698ec
......@@ -4,6 +4,7 @@
import base64
import logging
import os
import shlex
import subprocess
import sys
from argparse import ArgumentParser
......@@ -16,7 +17,7 @@ from cryptography.hazmat.backends import default_backend
__version__ = "0.2"
logger = logging.getLogger("rwm")
logger.setLevel(logging.DEBUG)
logger.setLevel(logging.INFO)
def is_sublist(needle, haystack):
......@@ -44,7 +45,7 @@ def run_command(*args, **kwargs):
"text": True,
"encoding": "utf-8",
})
logger.debug("run_command, %s", (args, kwargs))
logger.debug("run_command: %s", shlex.join(args[0]))
return subprocess.run(*args, **kwargs, check=False)
......@@ -147,14 +148,69 @@ class RWM:
}
return run_command(["restic"] + args, env=env)
def backup(self, name):
"""do restic backup from config"""
def main(argv=None):
"""main"""
if self.restic_cmd(["cat", "config"]).returncode != 0:
if (proc := self.restic_cmd(["init"])).returncode != 0:
logger.error("failed to autoinitialize restic repository")
return proc
conf = self.config["RWM_BACKUPS"][name]
# restic backup
excludes = conf.get("excludes", [])
excludes = [x for pair in zip(["--exclude"] * len(excludes), excludes) for x in pair]
extras = conf.get("extras", [])
cmd_args = ["backup"] + extras + excludes + conf["filesdirs"]
logger.info("running backup")
backup_proc = self.restic_cmd(cmd_args)
wrap_output(backup_proc)
if backup_proc.returncode != 0:
logger.error("backup failed, %s", backup_proc)
return backup_proc
# restic forget prune
keeps = []
for key, val in self.config.get("RWM_RETENTION", {}).items():
keeps += [f"--{key}", val]
if not keeps:
logger.error("no retention policy found")
cmd_args = ["forget", "--prune"] + keeps
logger.info("running forget prune")
forget_proc = self.restic_cmd(cmd_args)
wrap_output(forget_proc)
if forget_proc.returncode != 0:
logger.error("forget prune failed, %s", forget_proc)
return forget_proc
return backup_proc
def configure_logging(debug):
"""configure logger"""
log_handler = logging.StreamHandler(sys.stdout)
log_handler.setFormatter(
logging.Formatter(
fmt="%(asctime)s %(name)s[%(process)d]: %(levelname)s %(message)s"
)
)
logger.addHandler(log_handler)
if debug: # pragma: no cover ; would reconfigure pylint environment
logger.setLevel(logging.DEBUG)
def parse_arguments(argv):
"""parse arguments"""
parser = ArgumentParser(description="restics3 worm manager")
parser.add_argument("--debug", action="store_true")
parser.add_argument("--config", default="rwm.conf")
subparsers = parser.add_subparsers(title="commands", dest="command", required=False)
subparsers = parser.add_subparsers(title="commands", dest="command", required=False)
subparsers.add_parser("version", help="show version")
aws_cmd_parser = subparsers.add_parser("aws", help="aws command")
aws_cmd_parser.add_argument("cmd_args", nargs="*")
......@@ -164,27 +220,43 @@ def main(argv=None):
rcc_cmd_parser.add_argument("cmd_args", nargs="*")
res_cmd_parser = subparsers.add_parser("restic", help="restic command")
res_cmd_parser.add_argument("cmd_args", nargs="*")
backup_cmd_parser = subparsers.add_parser("backup", help="backup command")
backup_cmd_parser.add_argument("name", help="backup config name")
return parser.parse_args(argv)
args = parser.parse_args(argv)
def main(argv=None):
"""main"""
args = parse_arguments(argv)
configure_logging(args.debug)
config = {}
if args.config:
config.update(get_config(args.config))
logger.debug("config, %s", config)
# assert config ?
rwmi = RWM(config)
if args.command == "version":
print(__version__)
return 0
ret = -1
if args.command == "aws":
return wrap_output(rwmi.aws_cmd(args.cmd_args))
ret = wrap_output(rwmi.aws_cmd(args.cmd_args))
if args.command == "rclone":
return wrap_output(rwmi.rclone_cmd(args.cmd_args))
ret = wrap_output(rwmi.rclone_cmd(args.cmd_args))
if args.command == "rclone_crypt":
return wrap_output(rwmi.rclone_crypt_cmd(args.cmd_args))
ret = wrap_output(rwmi.rclone_crypt_cmd(args.cmd_args))
if args.command == "restic":
return wrap_output(rwmi.restic_cmd(args.cmd_args))
ret = wrap_output(rwmi.restic_cmd(args.cmd_args))
if args.command == "backup":
ret = rwmi.backup(args.name).returncode
return 0
logger.info("rwm finished with %s (ret %d)", "success" if ret == 0 else "errors", ret)
return ret
if __name__ == "__main__": # pragma: nocover
......
tests/test_default.py
+ 132
− 2
View file @ 276698ec
......@@ -40,7 +40,7 @@ def test_main(tmpworkdir: str): # pylint: disable=unused-argument
"""test main"""
# optional and default config hanling
assert rwm_main([]) == 0
assert rwm_main(["version"]) == 0
Path("rwm.conf").touch()
assert rwm_main(["version"]) == 0
......@@ -48,7 +48,10 @@ def test_main(tmpworkdir: str): # pylint: disable=unused-argument
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
assert rwm_main([item, "dummy"]) == 0
with patch.object(rwm.RWM, "backup", mock):
assert rwm_main(["backup", "dummy"]) == 0
def test_aws_cmd(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
......@@ -137,3 +140,130 @@ def test_restic_cmd(tmpworkdir: str, motoserver: str): # pylint: disable=unused
assert trwm.restic_cmd(["init"]).returncode == 0
proc = trwm.restic_cmd(["cat", "config"])
assert "id" in json.loads(proc.stdout)
def _list_snapshots(trwm):
"""test helper"""
return json.loads(trwm.restic_cmd(["snapshots", "--json"]).stdout)
def _list_files(trwm, snapshot_id):
"""test helper"""
snapshot_ls = [
json.loads(x)
for x in
trwm.restic_cmd(["ls", snapshot_id, "--json"]).stdout.splitlines()
]
return [
x["path"] for x in snapshot_ls
if (x["struct_type"] == "node" and x["type"] == "file")
]
def test_backup(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
"""test backup 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",
"RWM_BACKUPS": {
"testcfg": {
"filesdirs": ["testdatadir/"],
"excludes": ["testfile_to_be_ignored"],
"extras": ["--tag", "dummytag"],
}
},
"RWM_RETENTION": {
"keep-daily": "1"
}
})
Path("testdatadir").mkdir()
Path("testdatadir/testdata1.txt").write_text("dummydata", encoding="utf-8")
Path("testdatadir/testfile_to_be_ignored").write_text("dummydata", encoding="utf-8")
assert trwm.backup("testcfg").returncode == 0
snapshots = _list_snapshots(trwm)
assert len(snapshots) == 1
snapshot_files = _list_files(trwm, snapshots[0]["id"])
assert "/testdatadir/testdata1.txt" in snapshot_files
def test_backup_autoinit(): # pylint: disable=unused-argument
"""mocked error handling test"""
mock = Mock(return_value=CompletedProcess(args='dummy', returncode=1))
with patch.object(rwm.RWM, "restic_cmd", mock):
RWM({}).backup("dummy")
def test_backup_excludes(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
"""test backup 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",
"RWM_BACKUPS": {
"testcfg": {
"filesdirs": ["testdatadir"],
"excludes": ["proc", "*.ignored"],
"extras": ["--tag", "dummytag"],
}
}
})
Path("testdatadir").mkdir()
Path("testdatadir/etc").mkdir()
Path("testdatadir/etc/config").write_text("dummydata", encoding="utf-8")
Path("testdatadir/etc/config2").write_text("dummydata", encoding="utf-8")
Path("testdatadir/etc/config3.ignored").write_text("dummydata", encoding="utf-8")
Path("testdatadir/proc").mkdir()
Path("testdatadir/proc/to_be_also_excluded").write_text("dummydata", encoding="utf-8")
assert trwm.backup("testcfg").returncode == 0
snapshots = _list_snapshots(trwm)
assert len(snapshots) == 1
snapshot_files = _list_files(trwm, snapshots[0]["id"])
assert "/testdatadir/etc/config" in snapshot_files
assert "/testdatadir/etc/config2" in snapshot_files
def test_backup_error_handling(tmpworkdir: str, motoserver: str): # pylint: disable=unused-argument
"""test backup command err cases"""
mock = Mock(side_effect=[
CompletedProcess(args='dummy', returncode=0), # autoinit
CompletedProcess(args='dummy', returncode=11) # backup
])
with patch.object(rwm.RWM, "restic_cmd", mock):
trwm = RWM({
"RWM_BACKUPS": {
"dummycfg": {"filesdirs": ["dummydir"]}
}
})
proc = trwm.backup("dummycfg")
assert proc.returncode == 11
mock = Mock(side_effect=[
CompletedProcess(args='dummy', returncode=0), # autoinit
CompletedProcess(args='dummy', returncode=0), # backup
CompletedProcess(args='dummy', returncode=12) # forget
])
with patch.object(rwm.RWM, "restic_cmd", mock):
trwm = RWM({
"RWM_BACKUPS": {
"dummycfg": {"filesdirs": ["dummydir"]}
}
})
proc = trwm.backup("dummycfg")
assert proc.returncode == 12
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment