From c8a8887c5d07a65d4e929a65e7e47754333531f0 Mon Sep 17 00:00:00 2001 From: Florian Charlaix Date: Wed, 7 Feb 2024 10:45:08 +0100 Subject: [PATCH] Improve args parsing and checks --- nextcloud_backup.py | 87 ++++++++++++++++++++++++++++++++++----------- 1 file changed, 66 insertions(+), 21 deletions(-) diff --git a/nextcloud_backup.py b/nextcloud_backup.py index 5829328..2ca37a1 100644 --- a/nextcloud_backup.py +++ b/nextcloud_backup.py @@ -5,11 +5,12 @@ from logging import getLogger, basicConfig, DEBUG, INFO, WARN, ERROR, CRITICAL from subprocess import run, PIPE, CalledProcessError from json import loads from tempfile import NamedTemporaryFile, _TemporaryFileWrapper -from os import environ +from os import environ, access, X_OK from sys import stderr from argparse import ArgumentParser, BooleanOptionalAction, ArgumentTypeError from pathlib import Path -from typing import Optional, Any +from typing import Optional, Any, Dict +from shutil import which log_levels = { @@ -21,6 +22,37 @@ log_levels = { } +def executable(value: str) -> Path: + """ + Check if the argument is a valid executable + + :param value: Raw path to executable file + :return: Path to executable file + """ + path = which(value) + if not path: + raise ArgumentTypeError(f"{value} not found") + path = Path(path) + if not access(path, X_OK): + raise ArgumentTypeError(f"{value} is not executable") + return path + + +def directory(value: str) -> Path: + """ + Check if the argument is an existing directory + + :param value: Raw path + :return: Path + """ + path = Path(value) + if not path.exists(): + raise ArgumentTypeError(f"{value} doesn't exist") + if not path.is_dir(): + raise ArgumentTypeError(f"{value} is not a directory") + return path + + class NextcloudBackup: """ Nextcloud backup class @@ -335,6 +367,19 @@ class NextcloudBackup: TODO/WIP """ + def tests(self): + """ + Launch all tests + """ + if not self.test_nextcloud(): + raise ValueError("Nextcloud check faild") + + if not self.test_db(): + raise ValueError("Database check faild") + + if not self.test_s3(): + raise ValueError("S3 check faild") + def main(): """ @@ -347,11 +392,13 @@ def main(): help="Verbosity (between 1-4 occurrences with more leading to more " "verbose logging). CRITICAL=0, ERROR=1, WARN=2, INFO=3, " "DEBUG=4") + parser.add_argument("--log-file", "-f", type=Path) + parser.add_argument("--php", "-p", default="php", type=executable) + parser.add_argument("--nextcloud", "-n", default="/opt/nextcloud", type=directory) + parser.add_argument("--backup", "-b", default="/opt/backupnextcloud", type=directory) + parser_backup = subparsers.add_parser("backup") - parser_backup.add_argument("--php", "-p", default="php", type=Path) - parser_backup.add_argument("--nextcloud", "-n", default="/opt/nextcloud", type=Path) - parser_backup.add_argument("--backup", "-b", default="/opt/backupnextcloud", type=Path) parser_backup.add_argument("--maintenance", "-m", action=BooleanOptionalAction, default=True) parser_backup.add_argument("--rclone-remote-control", "-r", @@ -359,10 +406,17 @@ def main(): parser_check = subparsers.add_parser("check") - parser_check.add_argument("--backup", "-b", default="/opt/backupnextcloud", type=Path) args = parser.parse_args() - basicConfig(level=log_levels[args.verbosity]) + logging_confg : Dict[str, Any] = { + "level": log_levels[args.verbosity] + } + + if "file_log" in args and args.file_log: + logging_confg["filename"] = args.log_file + logging_confg["filemode"] = "a" + + basicConfig(**logging_confg) if not args.nextcloud.exists() or not args.nextcloud.is_dir(): raise ArgumentTypeError("Invalid Nextcloud path") @@ -370,23 +424,14 @@ def main(): if not args.backup.exists() or not args.backup.is_dir(): raise ArgumentTypeError("Invalid backup path") - nextcloud = { - "php": args.php, - "path": args.nextcloud - } + nb = NextcloudBackup({"php": args.php, "path": args.nextcloud}, args.backup) - nb = NextcloudBackup(nextcloud, args.backup) + try: + nb.tests() + except ValueError as e: + nb.logger.error(str(e)) if args.action == "backup": - if not nb.test_nextcloud(): - raise ArgumentTypeError("Nextcloud check faild") - - if not nb.test_db(): - raise ArgumentTypeError("Database check faild") - - if not nb.test_s3(): - raise ArgumentTypeError("S3 check faild") - nb.backup(args.maintenance, args.rclone_remote_control) elif args.action == "check": nb.check()