Compare commits

...
18 Commits
Author SHA1 Message Date
Dan Helfman 1676a98c51 Fix for Borg create error output not showing up at borgmatic verbosity level zero (#198). 2019-06-24 09:55:41 -07:00
Dan Helfman 358ed53da0 Only show build status badge for master branch. 2019-06-23 16:53:33 -07:00
Dan Helfman 90925c9428 Provide tips about old-style flags for those on older versions. 2019-06-23 16:42:23 -07:00
Dan Helfman cd192a6909 Bump version for release. 2019-06-23 16:30:16 -07:00
Dan Helfman 7185146481 A bunch of tests for parse_subparser_arguments(). 2019-06-23 16:06:39 -07:00
Dan Helfman c15e6c5fe5 More actions help. 2019-06-23 09:46:22 -07:00
Dan Helfman 79c2b9df06 Don't make major version bump after all. 2019-06-23 09:23:51 -07:00
Dan Helfman acd6772148 Update documentation to refer to dashless action sub-commands. 2019-06-22 22:09:50 -07:00
Dan Helfman cd91dbd4f7 Include sub-command help in documentation. 2019-06-22 22:04:56 -07:00
Dan Helfman 8fc4efff88 Add subcommand note to NEWS. 2019-06-22 21:35:40 -07:00
Dan Helfman 4bf3e906a1 Break out main borgmatic arguments-parsing code into a separate file. 2019-06-22 21:32:27 -07:00
Dan Helfman 0ca43ef67a Get tests passing. 2019-06-22 21:23:48 -07:00
Dan Helfman 603e055a39 Fix borgmatic command unit tests for new parsed arguments. 2019-06-22 16:29:25 -07:00
Dan Helfman 75c04611dc Refactor to support subparsed-based parsed arguments. 2019-06-22 16:10:07 -07:00
Dan Helfman 881dc9b01e Make each subparser get a crack all all arguments. 2019-06-21 23:12:37 -07:00
Dan Helfman 8c72e909a7 Initial stab at subparsers for argument parsing. Not yet fully working. 2019-06-21 22:27:16 -07:00
Dan Helfman 74ac148747 Disable console color via "color" option in borgmatic configuration output section (#191). 2019-06-19 20:48:54 -07:00
Dan Helfman be7887c071 Demote log level of unhelpful info log (#194). 2019-06-19 12:01:03 -07:00
25 changed files with 1182 additions and 687 deletions
+2
View File
@@ -0,0 +1,2 @@
.git
.tox
+10
View File
@@ -1,3 +1,13 @@
1.3.10
* #198: Fix for Borg create error output not showing up at borgmatic verbosity level zero.
1.3.9
* #195: Switch to command-line actions as more traditional sub-commands, e.g. "borgmatic create",
"borgmatic prune", etc. However, the classic dashed options like "--create" still work!
1.3.8
* #191: Disable console color via "color" option in borgmatic configuration output section.
1.3.7
* #196: Fix for unclear error message for invalid YAML merge include.
* #197: Don't color syslog output.
+1 -1
View File
@@ -2,7 +2,7 @@
title: borgmatic
permalink: index.html
---
<a href="https://build.torsion.org/witten/borgmatic" alt="build status">![Build Status](https://build.torsion.org/api/badges/witten/borgmatic/status.svg)</a>
<a href="https://build.torsion.org/witten/borgmatic" alt="build status">![Build Status](https://build.torsion.org/api/badges/witten/borgmatic/status.svg?ref=refs/heads/master)</a>
## Overview
+296
View File
@@ -0,0 +1,296 @@
import collections
from argparse import ArgumentParser
from borgmatic.config import collect
SUBPARSER_ALIASES = {
'init': ['--init', '-I'],
'prune': ['--prune', '-p'],
'create': ['--create', '-C'],
'check': ['--check', '-k'],
'extract': ['--extract', '-x'],
'list': ['--list', '-l'],
'info': ['--info', '-i'],
}
def parse_subparser_arguments(unparsed_arguments, top_level_parser, subparsers):
'''
Given a sequence of arguments, a top-level parser (containing subparsers), and a subparsers
object as returned by argparse.ArgumentParser().add_subparsers(), ask each subparser to parse
its own arguments and the top-level parser to parse any remaining arguments.
Return the result as a dict mapping from subparser name (or "global") to a parsed namespace of
arguments.
'''
arguments = collections.OrderedDict()
remaining_arguments = list(unparsed_arguments)
alias_to_subparser_name = {
alias: subparser_name
for subparser_name, aliases in SUBPARSER_ALIASES.items()
for alias in aliases
}
# Give each requested action's subparser a shot at parsing all arguments.
for subparser_name, subparser in subparsers.choices.items():
if subparser_name not in unparsed_arguments:
continue
remaining_arguments.remove(subparser_name)
canonical_name = alias_to_subparser_name.get(subparser_name, subparser_name)
parsed, remaining = subparser.parse_known_args(unparsed_arguments)
arguments[canonical_name] = parsed
# If no actions are explicitly requested, assume defaults: prune, create, and check.
if not arguments and '--help' not in unparsed_arguments and '-h' not in unparsed_arguments:
for subparser_name in ('prune', 'create', 'check'):
subparser = subparsers.choices[subparser_name]
parsed, remaining = subparser.parse_known_args(unparsed_arguments)
arguments[subparser_name] = parsed
# Then ask each subparser, one by one, to greedily consume arguments. Any arguments that remain
# are global arguments.
for subparser_name in arguments.keys():
subparser = subparsers.choices[subparser_name]
parsed, remaining_arguments = subparser.parse_known_args(remaining_arguments)
arguments['global'] = top_level_parser.parse_args(remaining_arguments)
return arguments
def parse_arguments(*unparsed_arguments):
'''
Given command-line arguments with which this script was invoked, parse the arguments and return
them as a dict mapping from subparser name (or "global") to an argparse.Namespace instance.
'''
config_paths = collect.get_default_config_paths()
global_parser = ArgumentParser(add_help=False)
global_group = global_parser.add_argument_group('global arguments')
global_group.add_argument(
'-c',
'--config',
nargs='*',
dest='config_paths',
default=config_paths,
help='Configuration filenames or directories, defaults to: {}'.format(
' '.join(config_paths)
),
)
global_group.add_argument(
'--excludes',
dest='excludes_filename',
help='Deprecated in favor of exclude_patterns within configuration',
)
global_group.add_argument(
'-n',
'--dry-run',
dest='dry_run',
action='store_true',
help='Go through the motions, but do not actually write to any repositories',
)
global_group.add_argument(
'-nc', '--no-color', dest='no_color', action='store_true', help='Disable colored output'
)
global_group.add_argument(
'-v',
'--verbosity',
type=int,
choices=range(0, 3),
default=0,
help='Display verbose progress to the console (from none to lots: 0, 1, or 2)',
)
global_group.add_argument(
'--syslog-verbosity',
type=int,
choices=range(0, 3),
default=0,
help='Display verbose progress to syslog (from none to lots: 0, 1, or 2)',
)
global_group.add_argument(
'--version',
dest='version',
default=False,
action='store_true',
help='Display installed version number of borgmatic and exit',
)
top_level_parser = ArgumentParser(
description='''
A simple wrapper script for the Borg backup software that creates and prunes backups.
If none of the action options are given, then borgmatic defaults to: prune, create, and
check archives.
''',
parents=[global_parser],
)
subparsers = top_level_parser.add_subparsers(
title='actions',
metavar='',
help='Specify zero or more actions. Defaults to prune, create, and check. Use --help with action for details:',
)
init_parser = subparsers.add_parser(
'init',
aliases=SUBPARSER_ALIASES['init'],
help='Initialize an empty Borg repository',
description='Initialize an empty Borg repository',
add_help=False,
)
init_group = init_parser.add_argument_group('init arguments')
init_group.add_argument(
'-e',
'--encryption',
dest='encryption_mode',
help='Borg repository encryption mode',
required=True,
)
init_group.add_argument(
'--append-only',
dest='append_only',
action='store_true',
help='Create an append-only repository',
)
init_group.add_argument(
'--storage-quota',
dest='storage_quota',
help='Create a repository with a fixed storage quota',
)
init_group.add_argument('-h', '--help', action='help', help='Show this help message and exit')
prune_parser = subparsers.add_parser(
'prune',
aliases=SUBPARSER_ALIASES['prune'],
help='Prune archives according to the retention policy',
description='Prune archives according to the retention policy',
add_help=False,
)
prune_group = prune_parser.add_argument_group('prune arguments')
prune_group.add_argument(
'--stats',
dest='stats',
default=False,
action='store_true',
help='Display statistics of archive',
)
prune_group.add_argument('-h', '--help', action='help', help='Show this help message and exit')
create_parser = subparsers.add_parser(
'create',
aliases=SUBPARSER_ALIASES['create'],
help='Create archives (actually perform backups)',
description='Create archives (actually perform backups)',
add_help=False,
)
create_group = create_parser.add_argument_group('create arguments')
create_group.add_argument(
'--progress',
dest='progress',
default=False,
action='store_true',
help='Display progress for each file as it is processed',
)
create_group.add_argument(
'--stats',
dest='stats',
default=False,
action='store_true',
help='Display statistics of archive',
)
create_group.add_argument(
'--json', dest='json', default=False, action='store_true', help='Output results as JSON'
)
create_group.add_argument('-h', '--help', action='help', help='Show this help message and exit')
check_parser = subparsers.add_parser(
'check',
aliases=SUBPARSER_ALIASES['check'],
help='Check archives for consistency',
description='Check archives for consistency',
add_help=False,
)
check_group = check_parser.add_argument_group('check arguments')
check_group.add_argument('-h', '--help', action='help', help='Show this help message and exit')
extract_parser = subparsers.add_parser(
'extract',
aliases=SUBPARSER_ALIASES['extract'],
help='Extract a named archive to the current directory',
description='Extract a named archive to the current directory',
add_help=False,
)
extract_group = extract_parser.add_argument_group('extract arguments')
extract_group.add_argument(
'--repository',
help='Path of repository to use, defaults to the configured repository if there is only one',
)
extract_group.add_argument('--archive', help='Name of archive to operate on', required=True)
extract_group.add_argument(
'--restore-path',
nargs='+',
dest='restore_paths',
help='Paths to restore from archive, defaults to the entire archive',
)
extract_group.add_argument(
'--progress',
dest='progress',
default=False,
action='store_true',
help='Display progress for each file as it is processed',
)
extract_group.add_argument(
'-h', '--help', action='help', help='Show this help message and exit'
)
list_parser = subparsers.add_parser(
'list',
aliases=SUBPARSER_ALIASES['list'],
help='List archives',
description='List archives',
add_help=False,
)
list_group = list_parser.add_argument_group('list arguments')
list_group.add_argument(
'--repository',
help='Path of repository to use, defaults to the configured repository if there is only one',
)
list_group.add_argument('--archive', help='Name of archive to operate on')
list_group.add_argument(
'--json', dest='json', default=False, action='store_true', help='Output results as JSON'
)
list_group.add_argument('-h', '--help', action='help', help='Show this help message and exit')
info_parser = subparsers.add_parser(
'info',
aliases=SUBPARSER_ALIASES['info'],
help='Display summary information on archives',
description='Display summary information on archives',
add_help=False,
)
info_group = info_parser.add_argument_group('info arguments')
info_group.add_argument(
'--json', dest='json', default=False, action='store_true', help='Output results as JSON'
)
info_group.add_argument('-h', '--help', action='help', help='Show this help message and exit')
arguments = parse_subparser_arguments(unparsed_arguments, top_level_parser, subparsers)
if arguments['global'].excludes_filename:
raise ValueError(
'The --excludes option has been replaced with exclude_patterns in configuration'
)
if 'init' in arguments and arguments['global'].dry_run:
raise ValueError('The init action cannot be used with the --dry-run option')
if (
'list' in arguments
and 'info' in arguments
and arguments['list'].json
and arguments['info'].json
):
raise ValueError('With the --json option, list and info actions cannot be used together')
return arguments
+136 -305
View File
@@ -3,7 +3,6 @@ import json
import logging
import os
import sys
from argparse import ArgumentParser
from subprocess import CalledProcessError
import colorama
@@ -18,6 +17,7 @@ from borgmatic.borg import info as borg_info
from borgmatic.borg import init as borg_init
from borgmatic.borg import list as borg_list
from borgmatic.borg import prune as borg_prune
from borgmatic.commands.arguments import parse_arguments
from borgmatic.config import checks, collect, convert, validate
from borgmatic.logger import configure_logging, should_do_markup
from borgmatic.signals import configure_signals
@@ -28,243 +28,10 @@ logger = logging.getLogger(__name__)
LEGACY_CONFIG_PATH = '/etc/borgmatic/config'
def parse_arguments(*arguments):
def run_configuration(config_filename, config, arguments): # pragma: no cover
'''
Given command-line arguments with which this script was invoked, parse the arguments and return
them as an argparse.ArgumentParser instance.
'''
config_paths = collect.get_default_config_paths()
parser = ArgumentParser(
description='''
A simple wrapper script for the Borg backup software that creates and prunes backups.
If none of the action options are given, then borgmatic defaults to: prune, create, and
check archives.
''',
add_help=False,
)
actions_group = parser.add_argument_group('actions')
actions_group.add_argument(
'-I', '--init', dest='init', action='store_true', help='Initialize an empty Borg repository'
)
actions_group.add_argument(
'-p',
'--prune',
dest='prune',
action='store_true',
help='Prune archives according to the retention policy',
)
actions_group.add_argument(
'-C',
'--create',
dest='create',
action='store_true',
help='Create archives (actually perform backups)',
)
actions_group.add_argument(
'-k', '--check', dest='check', action='store_true', help='Check archives for consistency'
)
actions_group.add_argument(
'-x',
'--extract',
dest='extract',
action='store_true',
help='Extract a named archive to the current directory',
)
actions_group.add_argument(
'-l', '--list', dest='list', action='store_true', help='List archives'
)
actions_group.add_argument(
'-i',
'--info',
dest='info',
action='store_true',
help='Display summary information on archives',
)
init_group = parser.add_argument_group('options for --init')
init_group.add_argument(
'-e', '--encryption', dest='encryption_mode', help='Borg repository encryption mode'
)
init_group.add_argument(
'--append-only',
dest='append_only',
action='store_true',
help='Create an append-only repository',
)
init_group.add_argument(
'--storage-quota',
dest='storage_quota',
help='Create a repository with a fixed storage quota',
)
prune_group = parser.add_argument_group('options for --prune')
stats_argument = prune_group.add_argument(
'--stats',
dest='stats',
default=False,
action='store_true',
help='Display statistics of archive',
)
create_group = parser.add_argument_group('options for --create')
progress_argument = create_group.add_argument(
'--progress',
dest='progress',
default=False,
action='store_true',
help='Display progress for each file as it is processed',
)
create_group._group_actions.append(stats_argument)
json_argument = create_group.add_argument(
'--json', dest='json', default=False, action='store_true', help='Output results as JSON'
)
extract_group = parser.add_argument_group('options for --extract')
repository_argument = extract_group.add_argument(
'--repository',
help='Path of repository to use, defaults to the configured repository if there is only one',
)
archive_argument = extract_group.add_argument('--archive', help='Name of archive to operate on')
extract_group.add_argument(
'--restore-path',
nargs='+',
dest='restore_paths',
help='Paths to restore from archive, defaults to the entire archive',
)
extract_group._group_actions.append(progress_argument)
list_group = parser.add_argument_group('options for --list')
list_group._group_actions.append(repository_argument)
list_group._group_actions.append(archive_argument)
list_group._group_actions.append(json_argument)
info_group = parser.add_argument_group('options for --info')
info_group._group_actions.append(json_argument)
common_group = parser.add_argument_group('common options')
common_group.add_argument(
'-c',
'--config',
nargs='+',
dest='config_paths',
default=config_paths,
help='Configuration filenames or directories, defaults to: {}'.format(
' '.join(config_paths)
),
)
common_group.add_argument(
'--excludes',
dest='excludes_filename',
help='Deprecated in favor of exclude_patterns within configuration',
)
common_group.add_argument(
'-n',
'--dry-run',
dest='dry_run',
action='store_true',
help='Go through the motions, but do not actually write to any repositories',
)
common_group.add_argument(
'-nc', '--no-color', dest='no_color', action='store_true', help='Disable colored output'
)
common_group.add_argument(
'-v',
'--verbosity',
type=int,
choices=range(0, 3),
default=0,
help='Display verbose progress to the console (from none to lots: 0, 1, or 2)',
)
common_group.add_argument(
'--syslog-verbosity',
type=int,
choices=range(0, 3),
default=0,
help='Display verbose progress to syslog (from none to lots: 0, 1, or 2)',
)
common_group.add_argument(
'--version',
dest='version',
default=False,
action='store_true',
help='Display installed version number of borgmatic and exit',
)
common_group.add_argument('--help', action='help', help='Show this help information and exit')
args = parser.parse_args(arguments)
if args.excludes_filename:
raise ValueError(
'The --excludes option has been replaced with exclude_patterns in configuration'
)
if (args.encryption_mode or args.append_only or args.storage_quota) and not args.init:
raise ValueError(
'The --encryption, --append-only, and --storage-quota options can only be used with the --init option'
)
if args.init and args.dry_run:
raise ValueError('The --init option cannot be used with the --dry-run option')
if args.init and not args.encryption_mode:
raise ValueError('The --encryption option is required with the --init option')
if not args.extract:
if not args.list:
if args.repository:
raise ValueError(
'The --repository option can only be used with the --extract and --list options'
)
if args.archive:
raise ValueError(
'The --archive option can only be used with the --extract and --list options'
)
if args.restore_paths:
raise ValueError('The --restore-path option can only be used with the --extract option')
if args.extract and not args.archive:
raise ValueError('The --archive option is required with the --extract option')
if args.progress and not (args.create or args.extract):
raise ValueError(
'The --progress option can only be used with the --create and --extract options'
)
if args.json and not (args.create or args.list or args.info):
raise ValueError(
'The --json option can only be used with the --create, --list, or --info options'
)
if args.json and args.list and args.info:
raise ValueError(
'With the --json option, options --list and --info cannot be used together'
)
# If any of the action flags are explicitly requested, leave them as-is. Otherwise, assume
# defaults: Mutate the given arguments to enable the default actions.
if (
not args.init
and not args.prune
and not args.create
and not args.check
and not args.extract
and not args.list
and not args.info
):
args.prune = True
args.create = True
args.check = True
if args.stats and not (args.create or args.prune):
raise ValueError('The --stats option can only be used when creating or pruning archives')
return args
def run_configuration(config_filename, config, args): # pragma: no cover
'''
Given a config filename and the corresponding parsed config dict, execute its defined pruning,
Given a config filename, the corresponding parsed config dict, and command-line arguments as a
dict from subparser name to a namespace of parsed arguments, execute its defined pruning,
backups, consistency checks, and/or other actions.
Yield JSON output strings from executing any actions that produce JSON.
@@ -273,24 +40,25 @@ def run_configuration(config_filename, config, args): # pragma: no cover
config.get(section_name, {})
for section_name in ('location', 'storage', 'retention', 'consistency', 'hooks')
)
global_arguments = arguments['global']
try:
local_path = location.get('local_path', 'borg')
remote_path = location.get('remote_path')
borg_environment.initialize(storage)
if args.create:
if 'create' in arguments:
hook.execute_hook(
hooks.get('before_backup'),
hooks.get('umask'),
config_filename,
'pre-backup',
args.dry_run,
global_arguments.dry_run,
)
for repository_path in location['repositories']:
yield from run_actions(
args=args,
arguments=arguments,
location=location,
storage=storage,
retention=retention,
@@ -300,23 +68,35 @@ def run_configuration(config_filename, config, args): # pragma: no cover
repository_path=repository_path,
)
if args.create:
if 'create' in arguments:
hook.execute_hook(
hooks.get('after_backup'),
hooks.get('umask'),
config_filename,
'post-backup',
args.dry_run,
global_arguments.dry_run,
)
except (OSError, CalledProcessError):
hook.execute_hook(
hooks.get('on_error'), hooks.get('umask'), config_filename, 'on-error', args.dry_run
hooks.get('on_error'),
hooks.get('umask'),
config_filename,
'on-error',
global_arguments.dry_run,
)
raise
def run_actions(
*, args, location, storage, retention, consistency, local_path, remote_path, repository_path
*,
arguments,
location,
storage,
retention,
consistency,
local_path,
remote_path,
repository_path
): # pragma: no cover
'''
Given parsed command-line arguments as an argparse.ArgumentParser instance, several different
@@ -326,119 +106,147 @@ def run_actions(
Yield JSON output strings from executing any actions that produce JSON.
'''
repository = os.path.expanduser(repository_path)
dry_run_label = ' (dry run; not making any changes)' if args.dry_run else ''
if args.init:
global_arguments = arguments['global']
dry_run_label = ' (dry run; not making any changes)' if global_arguments.dry_run else ''
if 'init' in arguments:
logger.info('{}: Initializing repository'.format(repository))
borg_init.initialize_repository(
repository,
args.encryption_mode,
args.append_only,
args.storage_quota,
arguments['init'].encryption_mode,
arguments['init'].append_only,
arguments['init'].storage_quota,
local_path=local_path,
remote_path=remote_path,
)
if args.prune:
if 'prune' in arguments:
logger.info('{}: Pruning archives{}'.format(repository, dry_run_label))
borg_prune.prune_archives(
args.dry_run,
global_arguments.dry_run,
repository,
storage,
retention,
local_path=local_path,
remote_path=remote_path,
stats=args.stats,
stats=arguments['prune'].stats,
)
if args.create:
if 'create' in arguments:
logger.info('{}: Creating archive{}'.format(repository, dry_run_label))
json_output = borg_create.create_archive(
args.dry_run,
global_arguments.dry_run,
repository,
location,
storage,
local_path=local_path,
remote_path=remote_path,
progress=args.progress,
stats=args.stats,
json=args.json,
progress=arguments['create'].progress,
stats=arguments['create'].stats,
json=arguments['create'].json,
)
if json_output:
yield json.loads(json_output)
if args.check and checks.repository_enabled_for_checks(repository, consistency):
if 'check' in arguments and checks.repository_enabled_for_checks(repository, consistency):
logger.info('{}: Running consistency checks'.format(repository))
borg_check.check_archives(
repository, storage, consistency, local_path=local_path, remote_path=remote_path
)
if args.extract:
if args.repository is None or repository == args.repository:
logger.info('{}: Extracting archive {}'.format(repository, args.archive))
if 'extract' in arguments:
if arguments['extract'].repository is None or repository == arguments['extract'].repository:
logger.info(
'{}: Extracting archive {}'.format(repository, arguments['extract'].archive)
)
borg_extract.extract_archive(
args.dry_run,
global_arguments.dry_run,
repository,
args.archive,
args.restore_paths,
arguments['extract'].archive,
arguments['extract'].restore_paths,
location,
storage,
local_path=local_path,
remote_path=remote_path,
progress=args.progress,
progress=arguments['extract'].progress,
)
if args.list:
if args.repository is None or repository == args.repository:
if 'list' in arguments:
if arguments['list'].repository is None or repository == arguments['list'].repository:
logger.info('{}: Listing archives'.format(repository))
json_output = borg_list.list_archives(
repository,
storage,
args.archive,
arguments['list'].archive,
local_path=local_path,
remote_path=remote_path,
json=args.json,
json=arguments['list'].json,
)
if json_output:
yield json.loads(json_output)
if args.info:
if 'info' in arguments:
logger.info('{}: Displaying summary info for archives'.format(repository))
json_output = borg_info.display_archives_info(
repository, storage, local_path=local_path, remote_path=remote_path, json=args.json
repository,
storage,
local_path=local_path,
remote_path=remote_path,
json=arguments['info'].json,
)
if json_output:
yield json.loads(json_output)
def collect_configuration_run_summary_logs(config_filenames, args):
def load_configurations(config_filenames):
'''
Given a sequence of configuration filenames and parsed command-line arguments as an
argparse.ArgumentParser instance, run each configuration file and yield a series of
logging.LogRecord instances containing summary information about each run.
As a side effect of running through these configuration files, output their JSON results, if
any, to stdout.
Given a sequence of configuration filenames, load and validate each configuration file. Return
the results as a tuple of: dict of configuration filename to corresponding parsed configuration,
and sequence of logging.LogRecord instances containing any parse errors.
'''
# Dict mapping from config filename to corresponding parsed config dict.
configs = collections.OrderedDict()
logs = []
# Parse and load each configuration file.
for config_filename in config_filenames:
try:
logger.info('{}: Parsing configuration file'.format(config_filename))
configs[config_filename] = validate.parse_configuration(
config_filename, validate.schema_filename()
)
except (ValueError, OSError, validate.Validation_error) as error:
yield logging.makeLogRecord(
dict(
levelno=logging.CRITICAL,
levelname='CRITICAL',
msg='{}: Error parsing configuration file'.format(config_filename),
)
)
yield logging.makeLogRecord(
dict(levelno=logging.CRITICAL, levelname='CRITICAL', msg=error)
logs.extend(
[
logging.makeLogRecord(
dict(
levelno=logging.CRITICAL,
levelname='CRITICAL',
msg='{}: Error parsing configuration file'.format(config_filename),
)
),
logging.makeLogRecord(
dict(levelno=logging.CRITICAL, levelname='CRITICAL', msg=error)
),
]
)
return (configs, logs)
def collect_configuration_run_summary_logs(configs, arguments):
'''
Given a dict of configuration filename to corresponding parsed configuration, and parsed
command-line arguments as a dict from subparser name to a parsed namespace of arguments, run
each configuration file and yield a series of logging.LogRecord instances containing summary
information about each run.
As a side effect of running through these configuration files, output their JSON results, if
any, to stdout.
'''
# Run cross-file validation checks.
if args.extract or (args.list and args.archive):
if 'extract' in arguments:
repository = arguments['extract'].repository
elif 'list' in arguments and arguments['list'].archive:
repository = arguments['list'].repository
else:
repository = None
if repository:
try:
validate.guard_configuration_contains_repository(args.repository, configs)
validate.guard_configuration_contains_repository(repository, configs)
except ValueError as error:
yield logging.makeLogRecord(
dict(levelno=logging.CRITICAL, levelname='CRITICAL', msg=error)
@@ -449,7 +257,7 @@ def collect_configuration_run_summary_logs(config_filenames, args):
json_results = []
for config_filename, config in configs.items():
try:
json_results.extend(list(run_configuration(config_filename, config, args)))
json_results.extend(list(run_configuration(config_filename, config, arguments)))
yield logging.makeLogRecord(
dict(
levelno=logging.INFO,
@@ -457,7 +265,21 @@ def collect_configuration_run_summary_logs(config_filenames, args):
msg='{}: Successfully ran configuration file'.format(config_filename),
)
)
except (ValueError, OSError, CalledProcessError) as error:
except CalledProcessError as error:
yield logging.makeLogRecord(
dict(
levelno=logging.CRITICAL,
levelname='CRITICAL',
msg='{}: Error running configuration file'.format(config_filename),
)
)
yield logging.makeLogRecord(
dict(levelno=logging.CRITICAL, levelname='CRITICAL', msg=error.output)
)
yield logging.makeLogRecord(
dict(levelno=logging.CRITICAL, levelname='CRITICAL', msg=error)
)
except (ValueError, OSError) as error:
yield logging.makeLogRecord(
dict(
levelno=logging.CRITICAL,
@@ -472,12 +294,14 @@ def collect_configuration_run_summary_logs(config_filenames, args):
if json_results:
sys.stdout.write(json.dumps(json_results))
if not config_filenames:
if not configs:
yield logging.makeLogRecord(
dict(
levelno=logging.CRITICAL,
levelname='CRITICAL',
msg='{}: No configuration files found'.format(' '.join(args.config_paths)),
msg='{}: No configuration files found'.format(
' '.join(arguments['global'].config_paths)
),
)
)
@@ -495,7 +319,7 @@ def main(): # pragma: no cover
configure_signals()
try:
args = parse_arguments(*sys.argv[1:])
arguments = parse_arguments(*sys.argv[1:])
except ValueError as error:
configure_logging(logging.CRITICAL)
logger.critical(error)
@@ -507,25 +331,32 @@ def main(): # pragma: no cover
logger.critical('Error parsing arguments: {}'.format(' '.join(sys.argv)))
exit_with_help_link()
colorama.init(autoreset=True, strip=not should_do_markup(args.no_color))
configure_logging(
verbosity_to_log_level(args.verbosity), verbosity_to_log_level(args.syslog_verbosity)
)
if args.version:
global_arguments = arguments['global']
if global_arguments.version:
print(pkg_resources.require('borgmatic')[0].version)
sys.exit(0)
config_filenames = tuple(collect.collect_config_filenames(args.config_paths))
config_filenames = tuple(collect.collect_config_filenames(global_arguments.config_paths))
configs, parse_logs = load_configurations(config_filenames)
colorama.init(autoreset=True, strip=not should_do_markup(global_arguments.no_color, configs))
configure_logging(
verbosity_to_log_level(global_arguments.verbosity),
verbosity_to_log_level(global_arguments.syslog_verbosity),
)
logger.debug('Ensuring legacy configuration is upgraded')
convert.guard_configuration_upgraded(LEGACY_CONFIG_PATH, config_filenames)
summary_logs = tuple(collect_configuration_run_summary_logs(config_filenames, args))
summary_logs = list(collect_configuration_run_summary_logs(configs, arguments))
logger.info('')
logger.info('summary:')
[logger.handle(log) for log in summary_logs if log.levelno >= logger.getEffectiveLevel()]
[
logger.handle(log)
for log in parse_logs + summary_logs
if log.levelno >= logger.getEffectiveLevel()
]
if any(log.levelno == logging.CRITICAL for log in summary_logs):
exit_with_help_link()
+10
View File
@@ -301,6 +301,16 @@ map:
this prefix. Borg placeholders can be used. See the output of
"borg help placeholders" for details. Defaults to "{hostname}-".
example: sourcehostname
output:
desc: |
Options for customizing borgmatic's own output and logging.
map:
color:
type: bool
desc: |
Apply color to console output. Can be overridden with --no-color command-line
flag. Defaults to true.
example: false
hooks:
desc: |
Shell commands or scripts to execute before and after a backup or if an error has occurred.
-9
View File
@@ -7,8 +7,6 @@ import ruamel.yaml
from borgmatic.config import load
logger = logging.getLogger(__name__)
def schema_filename():
'''
@@ -65,13 +63,6 @@ def apply_logical_validation(config_filename, parsed_configuration):
),
)
consistency_prefix = parsed_configuration.get('consistency', {}).get('prefix')
if archive_name_format and not consistency_prefix:
logger.warning(
'Since version 1.1.16, if you provide `archive_name_format`, you should also'
' specify `consistency.prefix`.'
)
def parse_configuration(config_filename, schema_filename):
'''
+19 -5
View File
@@ -4,7 +4,11 @@ import subprocess
logger = logging.getLogger(__name__)
ERROR_OUTPUT_MAX_LINE_COUNT = 25
def execute_and_log_output(full_command, output_log_level, shell):
last_lines = []
process = subprocess.Popen(
full_command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=shell
)
@@ -14,10 +18,13 @@ def execute_and_log_output(full_command, output_log_level, shell):
if not line:
continue
if line.startswith('borg: error:'):
logger.error(line)
else:
logger.log(output_log_level, line)
# Keep the last few lines of output in case the command errors, and we need the output for
# the exception below.
last_lines.append(line)
if len(last_lines) > ERROR_OUTPUT_MAX_LINE_COUNT:
last_lines.pop(0)
logger.log(output_log_level, line)
remaining_output = process.stdout.read().rstrip().decode()
if remaining_output: # pragma: no cover
@@ -25,7 +32,14 @@ def execute_and_log_output(full_command, output_log_level, shell):
exit_code = process.poll()
if exit_code != 0:
raise subprocess.CalledProcessError(exit_code, full_command)
# If an error occurs, include its output in the raised exception so that we don't
# inadvertently hide error output.
if len(last_lines) == ERROR_OUTPUT_MAX_LINE_COUNT:
last_lines.insert(0, '...')
raise subprocess.CalledProcessError(
exit_code, ' '.join(full_command), '\n'.join(last_lines)
)
def execute_command(full_command, output_log_level=logging.INFO, shell=False):
+6 -2
View File
@@ -21,13 +21,17 @@ def to_bool(arg):
return False
def should_do_markup(no_color):
def should_do_markup(no_color, configs):
'''
Determine if we should enable colorama marking up.
Given the value of the command-line no-color argument, and a dict of configuration filename to
corresponding parsed configuration, determine if we should enable colorama marking up.
'''
if no_color:
return False
if any(config.get('output', {}).get('color') is False for config in configs.values()):
return False
py_colors = os.environ.get('PY_COLORS', None)
if py_colors is not None:
+4 -1
View File
@@ -2,7 +2,10 @@ FROM python:3.7.3-alpine3.9 as borgmatic
COPY . /app
RUN pip install --no-cache /app && generate-borgmatic-config && chmod +r /etc/borgmatic/config.yaml
RUN borgmatic --help > /command-line.txt
RUN borgmatic --help > /command-line.txt \
&& for action in init prune create check extract list info; do \
echo -e "\n--------------------------------------------------------------------------------\n" >> /command-line.txt \
&& borgmatic "$action" --help >> /command-line.txt; done
FROM node:11.15.0-alpine as html
+8 -5
View File
@@ -17,15 +17,18 @@ run borgmatic's pruning, creating, or checking actions separately. For
instance, the the following optional flags are available:
```bash
borgmatic --prune
borgmatic --create
borgmatic --check
borgmatic prune
borgmatic create
borgmatic check
```
(No borgmatic `prune`, `create`, or `check` actions? Try the old-style
`--prune`, `--create`, or `--check`. Or upgrade borgmatic!)
You can run with only one of these flags provided, or you can mix and match
any number of them in a single borgmatic run. This supports approaches like
making backups with `--create` on a frequent schedule, while only running
expensive consistency checks with `--check` on a much less frequent basis from
making backups with `create` on a frequent schedule, while only running
expensive consistency checks with `check` on a much less frequent basis from
a separate cron job.
### Consistency check configuration
+7 -4
View File
@@ -29,10 +29,13 @@ functionality:
```bash
borgmatic --list
borgmatic --info
borgmatic list
borgmatic info
```
(No borgmatic `list` or `info` actions? Try the old-style `--list` or
`--info`. Or upgrade borgmatic!)
## Logging
By default, borgmatic logs to a local syslog-compatible daemon if one is
@@ -71,8 +74,8 @@ already has this rate limit disabled.
## Scripting borgmatic
To consume the output of borgmatic in other software, you can include an
optional `--json` flag with `--create`, `--list`, or `--info` to get the
output formatted as JSON.
optional `--json` flag with `create`, `list`, or `info` to get the output
formatted as JSON.
Note that when you specify the `--json` flag, Borg's other non-JSON output is
suppressed so as not to interfere with the captured JSON. Also note that JSON
+12 -6
View File
@@ -5,12 +5,15 @@ title: How to restore a backup
When the worst happens—or you want to test your backups—the first step is
to figure out which archive to restore. A good way to do that is to use the
`--list` action:
`list` action:
```bash
borgmatic --list
borgmatic list
```
(No borgmatic `list` action? Try the old-style `--list`, or upgrade
borgmatic!)
That should yield output looking something like:
```text
@@ -22,9 +25,12 @@ Assuming that you want to restore the archive with the most up-to-date files
and therefore the latest timestamp, run a command like:
```bash
borgmatic --extract --archive host-2019-01-02T04:06:07.080910
borgmatic extract --archive host-2019-01-02T04:06:07.080910
```
(No borgmatic `extract` action? Try the old-style `--extract`, or upgrade
borgmatic!)
The `--archive` value is the name of the archive to restore. This extracts the
entire contents of the archive to the current directory, so make sure you're
in the right place before running the command.
@@ -33,13 +39,13 @@ in the right place before running the command.
## Repository selection
If you have a single repository in your borgmatic configuration file(s), no
problem: the `--extract` action figures out which repository to use.
problem: the `extract` action figures out which repository to use.
But if you have multiple repositories configured, then you'll need to specify
the repository path containing the archive to extract. Here's an example:
```bash
borgmatic --extract --repository repo.borg --archive host-2019-...
borgmatic extract --repository repo.borg --archive host-2019-...
```
## Restore particular files
@@ -49,7 +55,7 @@ everything from an archive. To do that, tack on one or more `--restore-path`
values. For instance:
```bash
borgmatic --extract --archive host-2019-... --restore-path /path/1 /path/2
borgmatic extract --archive host-2019-... --restore-path /path/1 /path/2
```
Like a whole-archive restore, this also restores into the current directory.
+10 -6
View File
@@ -121,9 +121,12 @@ this step if you already have a Borg repository.) To create a repository, run
a command like the following:
```bash
borgmatic --init --encryption repokey
borgmatic init --encryption repokey
```
(No borgmatic `init` action? Try the old-style `--init` flag, or upgrade
borgmatic!)
This uses the borgmatic configuration file you created above to determine
which local or remote repository to create, and encrypts it with the
encryption passphrase specified there if one is provided. Read about [Borg
@@ -199,12 +202,13 @@ sudo systemctl start borgmatic.timer
Feel free to modify the timer file based on how frequently you'd like
borgmatic to run.
## Colored Output
## Colored output
Borgmatic uses [colorama](https://pypi.org/project/colorama/) to produce
colored terminal output by default. It is disabled when a non-interactive
terminal is detected (like a cron job). Otherwise, it can be disabled by
passing `--no-color` or by setting the environment variable `PY_COLORS=False`.
Borgmatic produces colored terminal output by default. It is disabled when a
non-interactive terminal is detected (like a cron job). Otherwise, you can
disable it by passing the `--no-color` flag, setting the environment variable
`PY_COLORS=False`, or setting the `color` option to `false` in the `output`
section of configuration.
## Troubleshooting
+2 -1
View File
@@ -3,7 +3,8 @@ title: borgmatic command-line reference
---
## borgmatic options
Here are all of the available borgmatic command-line options:
Here are all of the available borgmatic command-line options. This includes the separate options for
each action sub-command:
```
{% include borgmatic/command-line.txt %}
+4
View File
@@ -3,4 +3,8 @@
set -e
docker build --tag borgmatic-docs --file docs/Dockerfile .
echo
echo "You can view dev docs at http://localhost:8080"
echo "Note that links within these docs will go to the online docs, so you will need to fiddle with URLs manually to stay in the dev docs."
echo
docker run --interactive --tty --publish 8080:80 --rm borgmatic-docs
+1 -1
View File
@@ -18,7 +18,7 @@ mv temp.yaml.uncommented temp.yaml
for sub_command in prune create check list info; do
echo "********** borg $sub_command **********"
for line in $(borgmatic --config temp.yaml --$sub_command -v 2 2>&1 | grep "borg\w* $sub_command") ; do
for line in $(borgmatic --config temp.yaml $sub_command -v 2 2>&1 | grep "borg\w* $sub_command") ; do
echo "$line" | grep '^-' >> borgmatic_borg_flags
done
sort borgmatic_borg_flags > borgmatic_borg_flags.sorted
+1 -1
View File
@@ -1,6 +1,6 @@
from setuptools import find_packages, setup
VERSION = '1.3.7'
VERSION = '1.3.10'
setup(
@@ -0,0 +1,348 @@
import pytest
from flexmock import flexmock
from borgmatic.commands import arguments as module
def test_parse_arguments_with_no_arguments_uses_defaults():
config_paths = ['default']
flexmock(module.collect).should_receive('get_default_config_paths').and_return(config_paths)
arguments = module.parse_arguments()
global_arguments = arguments['global']
assert global_arguments.config_paths == config_paths
assert global_arguments.excludes_filename is None
assert global_arguments.verbosity == 0
assert global_arguments.syslog_verbosity == 0
def test_parse_arguments_with_multiple_config_paths_parses_as_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
arguments = module.parse_arguments('--config', 'myconfig', 'otherconfig')
global_arguments = arguments['global']
assert global_arguments.config_paths == ['myconfig', 'otherconfig']
assert global_arguments.verbosity == 0
assert global_arguments.syslog_verbosity == 0
def test_parse_arguments_with_verbosity_overrides_default():
config_paths = ['default']
flexmock(module.collect).should_receive('get_default_config_paths').and_return(config_paths)
arguments = module.parse_arguments('--verbosity', '1')
global_arguments = arguments['global']
assert global_arguments.config_paths == config_paths
assert global_arguments.excludes_filename is None
assert global_arguments.verbosity == 1
assert global_arguments.syslog_verbosity == 0
def test_parse_arguments_with_syslog_verbosity_overrides_default():
config_paths = ['default']
flexmock(module.collect).should_receive('get_default_config_paths').and_return(config_paths)
arguments = module.parse_arguments('--syslog-verbosity', '2')
global_arguments = arguments['global']
assert global_arguments.config_paths == config_paths
assert global_arguments.excludes_filename is None
assert global_arguments.verbosity == 0
assert global_arguments.syslog_verbosity == 2
def test_parse_arguments_with_list_json_overrides_default():
arguments = module.parse_arguments('list', '--json')
assert 'list' in arguments
assert arguments['list'].json is True
def test_parse_arguments_with_dashed_list_json_overrides_default():
arguments = module.parse_arguments('--list', '--json')
assert 'list' in arguments
assert arguments['list'].json is True
def test_parse_arguments_with_no_actions_defaults_to_all_actions_enabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
arguments = module.parse_arguments()
assert 'prune' in arguments
assert 'create' in arguments
assert 'check' in arguments
def test_parse_arguments_with_help_and_no_actions_shows_global_help(capsys):
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit) as exit:
module.parse_arguments('--help')
assert exit.value.code == 0
captured = capsys.readouterr()
assert 'global arguments:' in captured.out
assert 'actions:' in captured.out
def test_parse_arguments_with_help_and_action_shows_action_help(capsys):
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit) as exit:
module.parse_arguments('create', '--help')
assert exit.value.code == 0
captured = capsys.readouterr()
assert 'global arguments:' not in captured.out
assert 'actions:' not in captured.out
assert 'create arguments:' in captured.out
def test_parse_arguments_with_prune_action_leaves_other_actions_disabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
arguments = module.parse_arguments('prune')
assert 'prune' in arguments
assert 'create' not in arguments
assert 'check' not in arguments
def test_parse_arguments_with_dashed_prune_action_leaves_other_actions_disabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
arguments = module.parse_arguments('--prune')
assert 'prune' in arguments
assert 'create' not in arguments
assert 'check' not in arguments
def test_parse_arguments_with_multiple_actions_leaves_other_action_disabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
arguments = module.parse_arguments('create', 'check')
assert 'prune' not in arguments
assert 'create' in arguments
assert 'check' in arguments
def test_parse_arguments_with_multiple_dashed_actions_leaves_other_action_disabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
arguments = module.parse_arguments('--create', '--check')
assert 'prune' not in arguments
assert 'create' in arguments
assert 'check' in arguments
def test_parse_arguments_with_invalid_arguments_exits():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--posix-me-harder')
def test_parse_arguments_disallows_deprecated_excludes_option():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--excludes', 'myexcludes')
def test_parse_arguments_disallows_encryption_mode_without_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--config', 'myconfig', '--encryption', 'repokey')
def test_parse_arguments_allows_encryption_mode_with_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', 'init', '--encryption', 'repokey')
def test_parse_arguments_allows_encryption_mode_with_dashed_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', '--init', '--encryption', 'repokey')
def test_parse_arguments_requires_encryption_mode_with_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--config', 'myconfig', 'init')
def test_parse_arguments_disallows_append_only_without_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--config', 'myconfig', '--append-only')
def test_parse_arguments_disallows_storage_quota_without_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--config', 'myconfig', '--storage-quota', '5G')
def test_parse_arguments_allows_init_and_prune():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', 'init', '--encryption', 'repokey', 'prune')
def test_parse_arguments_allows_init_and_create():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', 'init', '--encryption', 'repokey', 'create')
def test_parse_arguments_disallows_init_and_dry_run():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments(
'--config', 'myconfig', 'init', '--encryption', 'repokey', '--dry-run'
)
def test_parse_arguments_disallows_repository_without_extract_or_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--config', 'myconfig', '--repository', 'test.borg')
def test_parse_arguments_allows_repository_with_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments(
'--config', 'myconfig', 'extract', '--repository', 'test.borg', '--archive', 'test'
)
def test_parse_arguments_allows_repository_with_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', 'list', '--repository', 'test.borg')
def test_parse_arguments_disallows_archive_without_extract_or_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--config', 'myconfig', '--archive', 'test')
def test_parse_arguments_disallows_restore_paths_without_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--config', 'myconfig', '--restore-path', 'test')
def test_parse_arguments_allows_archive_with_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', 'extract', '--archive', 'test')
def test_parse_arguments_allows_archive_with_dashed_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', '--extract', '--archive', 'test')
def test_parse_arguments_allows_archive_with_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', 'list', '--archive', 'test')
def test_parse_arguments_requires_archive_with_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--config', 'myconfig', 'extract')
def test_parse_arguments_allows_progress_before_create():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--progress', 'create', 'list')
def test_parse_arguments_allows_progress_after_create():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('create', '--progress', 'list')
def test_parse_arguments_allows_progress_and_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--progress', 'extract', '--archive', 'test', 'list')
def test_parse_arguments_disallows_progress_without_create():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--progress', 'list')
def test_parse_arguments_with_stats_and_create_flags_does_not_raise():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--stats', 'create', 'list')
def test_parse_arguments_with_stats_and_prune_flags_does_not_raise():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--stats', 'prune', 'list')
def test_parse_arguments_with_stats_flag_but_no_create_or_prune_flag_raises_value_error():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--stats', 'list')
def test_parse_arguments_with_just_stats_flag_does_not_raise():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--stats')
def test_parse_arguments_allows_json_with_list_or_info():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('list', '--json')
module.parse_arguments('info', '--json')
def test_parse_arguments_allows_json_with_dashed_info():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--info', '--json')
def test_parse_arguments_disallows_json_with_both_list_and_info():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('list', 'info', '--json')
@@ -1,281 +1,10 @@
import subprocess
import pytest
from flexmock import flexmock
from borgmatic.commands import borgmatic as module
def test_parse_arguments_with_no_arguments_uses_defaults():
config_paths = ['default']
flexmock(module.collect).should_receive('get_default_config_paths').and_return(config_paths)
parser = module.parse_arguments()
assert parser.config_paths == config_paths
assert parser.excludes_filename is None
assert parser.verbosity == 0
assert parser.syslog_verbosity == 0
assert parser.json is False
def test_parse_arguments_with_multiple_config_paths_parses_as_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
parser = module.parse_arguments('--config', 'myconfig', 'otherconfig')
assert parser.config_paths == ['myconfig', 'otherconfig']
assert parser.verbosity == 0
assert parser.syslog_verbosity == 0
def test_parse_arguments_with_verbosity_overrides_default():
config_paths = ['default']
flexmock(module.collect).should_receive('get_default_config_paths').and_return(config_paths)
parser = module.parse_arguments('--verbosity', '1')
assert parser.config_paths == config_paths
assert parser.excludes_filename is None
assert parser.verbosity == 1
assert parser.syslog_verbosity == 0
def test_parse_arguments_with_syslog_verbosity_overrides_default():
config_paths = ['default']
flexmock(module.collect).should_receive('get_default_config_paths').and_return(config_paths)
parser = module.parse_arguments('--syslog-verbosity', '2')
assert parser.config_paths == config_paths
assert parser.excludes_filename is None
assert parser.verbosity == 0
assert parser.syslog_verbosity == 2
def test_parse_arguments_with_json_overrides_default():
parser = module.parse_arguments('--list', '--json')
assert parser.json is True
def test_parse_arguments_with_no_actions_defaults_to_all_actions_enabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
parser = module.parse_arguments()
assert parser.prune is True
assert parser.create is True
assert parser.check is True
def test_parse_arguments_with_prune_action_leaves_other_actions_disabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
parser = module.parse_arguments('--prune')
assert parser.prune is True
assert parser.create is False
assert parser.check is False
def test_parse_arguments_with_multiple_actions_leaves_other_action_disabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
parser = module.parse_arguments('--create', '--check')
assert parser.prune is False
assert parser.create is True
assert parser.check is True
def test_parse_arguments_with_invalid_arguments_exits():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(SystemExit):
module.parse_arguments('--posix-me-harder')
def test_parse_arguments_disallows_deprecated_excludes_option():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--excludes', 'myexcludes')
def test_parse_arguments_disallows_encryption_mode_without_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--encryption', 'repokey')
def test_parse_arguments_allows_encryption_mode_with_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', '--init', '--encryption', 'repokey')
def test_parse_arguments_requires_encryption_mode_with_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--init')
def test_parse_arguments_disallows_append_only_without_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--append-only')
def test_parse_arguments_disallows_storage_quota_without_init():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--storage-quota', '5G')
def test_parse_arguments_allows_init_and_prune():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', '--init', '--encryption', 'repokey', '--prune')
def test_parse_arguments_allows_init_and_create():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', '--init', '--encryption', 'repokey', '--create')
def test_parse_arguments_disallows_init_and_dry_run():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments(
'--config', 'myconfig', '--init', '--encryption', 'repokey', '--dry-run'
)
def test_parse_arguments_disallows_repository_without_extract_or_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--repository', 'test.borg')
def test_parse_arguments_allows_repository_with_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments(
'--config', 'myconfig', '--extract', '--repository', 'test.borg', '--archive', 'test'
)
def test_parse_arguments_allows_repository_with_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', '--list', '--repository', 'test.borg')
def test_parse_arguments_disallows_archive_without_extract_or_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--archive', 'test')
def test_parse_arguments_disallows_restore_paths_without_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--restore-path', 'test')
def test_parse_arguments_allows_archive_with_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', '--extract', '--archive', 'test')
def test_parse_arguments_allows_archive_with_list():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--config', 'myconfig', '--list', '--archive', 'test')
def test_parse_arguments_requires_archive_with_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--config', 'myconfig', '--extract')
def test_parse_arguments_allows_progress_and_create():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--progress', '--create', '--list')
def test_parse_arguments_allows_progress_and_extract():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--progress', '--extract', '--archive', 'test', '--list')
def test_parse_arguments_disallows_progress_without_create():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--progress', '--list')
def test_parse_arguments_with_stats_and_create_flags_does_not_raise():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--stats', '--create', '--list')
def test_parse_arguments_with_stats_and_prune_flags_does_not_raise():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--stats', '--prune', '--list')
def test_parse_arguments_with_stats_flag_but_no_create_or_prune_flag_raises_value_error():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--stats', '--list')
def test_parse_arguments_with_just_stats_flag_does_not_raise():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--stats')
def test_parse_arguments_allows_json_with_list_or_info():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
module.parse_arguments('--list', '--json')
module.parse_arguments('--info', '--json')
def test_parse_arguments_disallows_json_without_list_or_info():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--json')
def test_parse_arguments_disallows_json_with_both_list_and_info():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments('--list', '--info', '--json')
def test_borgmatic_version_matches_news_version():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
+15 -5
View File
@@ -15,12 +15,22 @@ def test_execute_and_log_output_logs_each_line_separately():
module.execute_and_log_output(['echo', 'there'], output_log_level=logging.INFO, shell=False)
def test_execute_and_log_output_logs_borg_error_as_error():
flexmock(module.logger).should_receive('error').with_args('borg: error: oopsie').once()
def test_execute_and_log_output_includes_borg_error_output_in_exception():
with pytest.raises(subprocess.CalledProcessError) as error:
module.execute_and_log_output(['grep'], output_log_level=logging.INFO, shell=False)
module.execute_and_log_output(
['echo', 'borg: error: oopsie'], output_log_level=logging.INFO, shell=False
)
assert error.value.returncode == 2
assert error.value.output
def test_execute_and_log_output_truncates_long_borg_error_output():
flexmock(module).ERROR_OUTPUT_MAX_LINE_COUNT = 0
with pytest.raises(subprocess.CalledProcessError) as error:
module.execute_and_log_output(['grep'], output_log_level=logging.INFO, shell=False)
assert error.value.returncode == 2
assert error.value.output.startswith('...')
def test_execute_and_log_output_with_no_output_logs_nothing():
+171
View File
@@ -0,0 +1,171 @@
from flexmock import flexmock
from borgmatic.commands import arguments as module
def test_parse_subparser_arguments_consumes_subparser_arguments_before_subparser_name():
global_namespace = flexmock()
action_namespace = flexmock(foo=True)
top_level_parser = flexmock(parse_args=lambda arguments: global_namespace)
subparsers = flexmock(
choices={
'action': flexmock(parse_known_args=lambda arguments: (action_namespace, [])),
'other': flexmock(),
}
)
arguments = module.parse_subparser_arguments(
('--foo', 'true', 'action'), top_level_parser, subparsers
)
assert arguments == {'action': action_namespace, 'global': global_namespace}
def test_parse_subparser_arguments_consumes_subparser_arguments_after_subparser_name():
global_namespace = flexmock()
action_namespace = flexmock(foo=True)
top_level_parser = flexmock(parse_args=lambda arguments: global_namespace)
subparsers = flexmock(
choices={
'action': flexmock(parse_known_args=lambda arguments: (action_namespace, [])),
'other': flexmock(),
}
)
arguments = module.parse_subparser_arguments(
('action', '--foo', 'true'), top_level_parser, subparsers
)
assert arguments == {'action': action_namespace, 'global': global_namespace}
def test_parse_subparser_arguments_consumes_global_arguments_before_subparser_name():
global_namespace = flexmock(verbosity='lots')
action_namespace = flexmock()
top_level_parser = flexmock(parse_args=lambda arguments: global_namespace)
subparsers = flexmock(
choices={
'action': flexmock(
parse_known_args=lambda arguments: (action_namespace, ['--verbosity', 'lots'])
),
'other': flexmock(),
}
)
arguments = module.parse_subparser_arguments(
('--verbosity', 'lots', 'action'), top_level_parser, subparsers
)
assert arguments == {'action': action_namespace, 'global': global_namespace}
def test_parse_subparser_arguments_consumes_global_arguments_after_subparser_name():
global_namespace = flexmock(verbosity='lots')
action_namespace = flexmock()
top_level_parser = flexmock(parse_args=lambda arguments: global_namespace)
subparsers = flexmock(
choices={
'action': flexmock(
parse_known_args=lambda arguments: (action_namespace, ['--verbosity', 'lots'])
),
'other': flexmock(),
}
)
arguments = module.parse_subparser_arguments(
('action', '--verbosity', 'lots'), top_level_parser, subparsers
)
assert arguments == {'action': action_namespace, 'global': global_namespace}
def test_parse_subparser_arguments_consumes_subparser_arguments_with_alias():
global_namespace = flexmock()
action_namespace = flexmock(foo=True)
top_level_parser = flexmock(parse_args=lambda arguments: global_namespace)
action_subparser = flexmock(parse_known_args=lambda arguments: (action_namespace, []))
subparsers = flexmock(
choices={
'action': action_subparser,
'-a': action_subparser,
'other': flexmock(),
'-o': flexmock(),
}
)
flexmock(module).SUBPARSER_ALIASES = {'action': ['-a'], 'other': ['-o']}
arguments = module.parse_subparser_arguments(
('-a', '--foo', 'true'), top_level_parser, subparsers
)
assert arguments == {'action': action_namespace, 'global': global_namespace}
def test_parse_subparser_arguments_consumes_multiple_subparser_arguments():
global_namespace = flexmock()
action_namespace = flexmock(foo=True)
other_namespace = flexmock(bar=3)
top_level_parser = flexmock(parse_args=lambda arguments: global_namespace)
subparsers = flexmock(
choices={
'action': flexmock(
parse_known_args=lambda arguments: (action_namespace, ['--bar', '3'])
),
'other': flexmock(parse_known_args=lambda arguments: (other_namespace, [])),
}
)
arguments = module.parse_subparser_arguments(
('action', '--foo', 'true', 'other', '--bar', '3'), top_level_parser, subparsers
)
assert arguments == {
'action': action_namespace,
'other': other_namespace,
'global': global_namespace,
}
def test_parse_subparser_arguments_applies_default_subparsers():
global_namespace = flexmock()
prune_namespace = flexmock()
create_namespace = flexmock(progress=True)
check_namespace = flexmock()
top_level_parser = flexmock(parse_args=lambda arguments: global_namespace)
subparsers = flexmock(
choices={
'prune': flexmock(parse_known_args=lambda arguments: (prune_namespace, ['--progress'])),
'create': flexmock(parse_known_args=lambda arguments: (create_namespace, [])),
'check': flexmock(parse_known_args=lambda arguments: (check_namespace, [])),
'other': flexmock(),
}
)
arguments = module.parse_subparser_arguments(('--progress'), top_level_parser, subparsers)
assert arguments == {
'prune': prune_namespace,
'create': create_namespace,
'check': check_namespace,
'global': global_namespace,
}
def test_parse_subparser_arguments_with_help_does_not_apply_default_subparsers():
global_namespace = flexmock(verbosity='lots')
action_namespace = flexmock()
top_level_parser = flexmock(parse_args=lambda arguments: global_namespace)
subparsers = flexmock(
choices={
'action': flexmock(
parse_known_args=lambda arguments: (action_namespace, ['--verbosity', 'lots'])
),
'other': flexmock(),
}
)
arguments = module.parse_subparser_arguments(
('--verbosity', 'lots', '--help'), top_level_parser, subparsers
)
assert arguments == {'global': global_namespace}
+76 -40
View File
@@ -1,101 +1,137 @@
import subprocess
from flexmock import flexmock
from borgmatic.commands import borgmatic as module
def test_collect_configuration_run_summary_logs_info_for_success():
flexmock(module.validate).should_receive('parse_configuration').and_return({'test.yaml': {}})
flexmock(module).should_receive('run_configuration').and_return([])
args = flexmock(extract=False, list=False)
def test_load_configurations_collects_parsed_configurations():
configuration = flexmock()
other_configuration = flexmock()
flexmock(module.validate).should_receive('parse_configuration').and_return(
configuration
).and_return(other_configuration)
logs = tuple(module.collect_configuration_run_summary_logs(('test.yaml',), args=args))
configs, logs = tuple(module.load_configurations(('test.yaml', 'other.yaml')))
assert configs == {'test.yaml': configuration, 'other.yaml': other_configuration}
assert logs == []
def test_load_configurations_logs_critical_for_parse_error():
flexmock(module.validate).should_receive('parse_configuration').and_raise(ValueError)
configs, logs = tuple(module.load_configurations(('test.yaml',)))
assert configs == {}
assert any(log for log in logs if log.levelno == module.logging.CRITICAL)
def test_collect_configuration_run_summary_logs_info_for_success():
flexmock(module).should_receive('run_configuration').and_return([])
arguments = {}
logs = tuple(
module.collect_configuration_run_summary_logs({'test.yaml': {}}, arguments=arguments)
)
assert all(log for log in logs if log.levelno == module.logging.INFO)
def test_collect_configuration_run_summary_logs_info_for_success_with_extract():
flexmock(module.validate).should_receive('parse_configuration').and_return({'test.yaml': {}})
flexmock(module.validate).should_receive('guard_configuration_contains_repository')
flexmock(module).should_receive('run_configuration').and_return([])
args = flexmock(extract=True, list=False, repository='repo')
arguments = {'extract': flexmock(repository='repo')}
logs = tuple(module.collect_configuration_run_summary_logs(('test.yaml',), args=args))
logs = tuple(
module.collect_configuration_run_summary_logs({'test.yaml': {}}, arguments=arguments)
)
assert all(log for log in logs if log.levelno == module.logging.INFO)
def test_collect_configuration_run_summary_logs_critical_for_extract_with_repository_error():
flexmock(module.validate).should_receive('parse_configuration').and_return({'test.yaml': {}})
flexmock(module.validate).should_receive('guard_configuration_contains_repository').and_raise(
ValueError
)
args = flexmock(extract=True, list=False, repository='repo')
arguments = {'extract': flexmock(repository='repo')}
logs = tuple(module.collect_configuration_run_summary_logs(('test.yaml',), args=args))
logs = tuple(
module.collect_configuration_run_summary_logs({'test.yaml': {}}, arguments=arguments)
)
assert any(log for log in logs if log.levelno == module.logging.CRITICAL)
def test_collect_configuration_run_summary_logs_critical_for_list_with_archive_and_repository_error():
flexmock(module.validate).should_receive('parse_configuration').and_return({'test.yaml': {}})
flexmock(module.validate).should_receive('guard_configuration_contains_repository').and_raise(
ValueError
)
args = flexmock(extract=False, list=True, repository='repo', archive='test')
arguments = {'list': flexmock(repository='repo', archive='test')}
logs = tuple(module.collect_configuration_run_summary_logs(('test.yaml',), args=args))
logs = tuple(
module.collect_configuration_run_summary_logs({'test.yaml': {}}, arguments=arguments)
)
assert any(log for log in logs if log.levelno == module.logging.CRITICAL)
def test_collect_configuration_run_summary_logs_info_for_success_with_list():
flexmock(module.validate).should_receive('parse_configuration').and_return({'test.yaml': {}})
flexmock(module).should_receive('run_configuration').and_return([])
args = flexmock(extract=False, list=True, repository='repo', archive=None)
arguments = {'list': flexmock(repository='repo', archive=None)}
logs = tuple(module.collect_configuration_run_summary_logs(('test.yaml',), args=args))
logs = tuple(
module.collect_configuration_run_summary_logs({'test.yaml': {}}, arguments=arguments)
)
assert all(log for log in logs if log.levelno == module.logging.INFO)
def test_collect_configuration_run_summary_logs_critical_for_parse_error():
flexmock(module.validate).should_receive('parse_configuration').and_raise(ValueError)
args = flexmock(extract=False, list=False)
logs = tuple(module.collect_configuration_run_summary_logs(('test.yaml',), args=args))
assert any(log for log in logs if log.levelno == module.logging.CRITICAL)
def test_collect_configuration_run_summary_logs_critical_for_run_error():
flexmock(module.validate).should_receive('parse_configuration').and_return({'test.yaml': {}})
def test_collect_configuration_run_summary_logs_critical_for_run_value_error():
flexmock(module.validate).should_receive('guard_configuration_contains_repository')
flexmock(module).should_receive('run_configuration').and_raise(ValueError)
args = flexmock(extract=False, list=False)
arguments = {}
logs = tuple(module.collect_configuration_run_summary_logs(('test.yaml',), args=args))
logs = tuple(
module.collect_configuration_run_summary_logs({'test.yaml': {}}, arguments=arguments)
)
assert any(log for log in logs if log.levelno == module.logging.CRITICAL)
def test_collect_configuration_run_summary_logs_critical_for_missing_configs():
flexmock(module.validate).should_receive('parse_configuration').and_return({'test.yaml': {}})
flexmock(module).should_receive('run_configuration').and_return([])
args = flexmock(config_paths=(), extract=False, list=False)
def test_collect_configuration_run_summary_logs_critical_including_output_for_run_process_error():
flexmock(module.validate).should_receive('guard_configuration_contains_repository')
flexmock(module).should_receive('run_configuration').and_raise(
subprocess.CalledProcessError(1, 'command', 'error output')
)
arguments = {}
logs = tuple(module.collect_configuration_run_summary_logs(config_filenames=(), args=args))
logs = tuple(
module.collect_configuration_run_summary_logs({'test.yaml': {}}, arguments=arguments)
)
assert any(log for log in logs if log.levelno == module.logging.CRITICAL)
assert any(log for log in logs if 'error output' in str(log))
def test_collect_configuration_run_summary_logs_outputs_merged_json_results():
flexmock(module.validate).should_receive('parse_configuration').and_return(
{'test.yaml': {}, 'test2.yaml': {}}
)
flexmock(module).should_receive('run_configuration').and_return(['foo', 'bar']).and_return(
['baz']
)
flexmock(module.sys.stdout).should_receive('write').with_args('["foo", "bar", "baz"]').once()
args = flexmock(extract=False, list=False)
arguments = {}
tuple(module.collect_configuration_run_summary_logs(('test.yaml', 'test2.yaml'), args=args))
tuple(
module.collect_configuration_run_summary_logs(
{'test.yaml': {}, 'test2.yaml': {}}, arguments=arguments
)
)
def test_collect_configuration_run_summary_logs_critical_for_missing_configs():
flexmock(module).should_receive('run_configuration').and_return([])
arguments = {'global': flexmock(config_paths=[])}
logs = tuple(module.collect_configuration_run_summary_logs({}, arguments=arguments))
assert any(log for log in logs if log.levelno == module.logging.CRITICAL)
+1 -19
View File
@@ -1,5 +1,4 @@
import pytest
from flexmock import flexmock
from borgmatic.config import validate as module
@@ -37,20 +36,6 @@ def test_apply_logical_validation_raises_if_archive_name_format_present_without_
)
def test_apply_logical_validation_warns_if_archive_name_format_present_without_consistency_prefix():
logger = flexmock(module.logger)
logger.should_receive('warning').once()
module.apply_logical_validation(
'config.yaml',
{
'storage': {'archive_name_format': '{hostname}-{now}'},
'retention': {'prefix': '{hostname}-'},
'consistency': {},
},
)
def test_apply_locical_validation_raises_if_unknown_repository_in_check_repositories():
with pytest.raises(module.Validation_error):
module.apply_logical_validation(
@@ -74,10 +59,7 @@ def test_apply_locical_validation_does_not_raise_if_known_repository_in_check_re
)
def test_apply_logical_validation_does_not_raise_or_warn_if_archive_name_format_and_prefix_present():
logger = flexmock(module.logger)
logger.should_receive('warning').never()
def test_apply_logical_validation_does_not_raise_if_archive_name_format_and_prefix_present():
module.apply_logical_validation(
'config.yaml',
{
+42 -5
View File
@@ -21,34 +21,71 @@ def test_to_bool_passes_none_through():
def test_should_do_markup_respects_no_color_value():
assert module.should_do_markup(no_color=True) is False
assert module.should_do_markup(no_color=True, configs={}) is False
def test_should_do_markup_respects_config_value():
assert (
module.should_do_markup(no_color=False, configs={'foo.yaml': {'output': {'color': False}}})
is False
)
def test_should_do_markup_prefers_any_false_config_value():
assert (
module.should_do_markup(
no_color=False,
configs={
'foo.yaml': {'output': {'color': True}},
'bar.yaml': {'output': {'color': False}},
},
)
is False
)
def test_should_do_markup_respects_PY_COLORS_environment_variable():
flexmock(module.os.environ).should_receive('get').and_return('True')
flexmock(module).should_receive('to_bool').and_return(True)
assert module.should_do_markup(no_color=False) is True
assert module.should_do_markup(no_color=False, configs={}) is True
def test_should_do_markup_prefers_no_color_value_to_config_value():
assert (
module.should_do_markup(no_color=True, configs={'foo.yaml': {'output': {'color': True}}})
is False
)
def test_should_do_markup_prefers_config_value_to_PY_COLORS():
flexmock(module.os.environ).should_receive('get').and_return('True')
flexmock(module).should_receive('to_bool').and_return(True)
assert (
module.should_do_markup(no_color=False, configs={'foo.yaml': {'output': {'color': False}}})
is False
)
def test_should_do_markup_prefers_no_color_value_to_PY_COLORS():
flexmock(module.os.environ).should_receive('get').and_return('True')
flexmock(module).should_receive('to_bool').and_return(True)
assert module.should_do_markup(no_color=True) is False
assert module.should_do_markup(no_color=True, configs={}) is False
def test_should_do_markup_respects_stdout_tty_value():
flexmock(module.os.environ).should_receive('get').and_return(None)
assert module.should_do_markup(no_color=False) is False
assert module.should_do_markup(no_color=False, configs={}) is False
def test_should_do_markup_prefers_PY_COLORS_to_stdout_tty_value():
flexmock(module.os.environ).should_receive('get').and_return('True')
flexmock(module).should_receive('to_bool').and_return(True)
assert module.should_do_markup(no_color=False) is True
assert module.should_do_markup(no_color=False, configs={}) is True
def test_console_color_formatter_format_includes_log_message():