psw/borgmatic
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Run arbitrary Borg commands with new "borgmatic borg" action (#425).

Browse source
This commit is contained in:
Dan Helfman 2021-06-17 20:41:44 -07:00
parent b37dd1a79e
commit cf8882f2bc
13 changed files with 438 additions and 132 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -2,7 +2,7 @@
*.pyc
*.swp
.cache
.coverage
.coverage*
.pytest_cache
.tox
__pycache__

2
NEWS
View file

@ -1,6 +1,8 @@
1.5.15.dev0
* #419: Document use case of running backups conditionally based on laptop power level:
https://torsion.org/borgmatic/docs/how-to/backup-to-a-removable-drive-or-an-intermittent-server/
* #425: Run arbitrary Borg commands with new "borgmatic borg" action. See the documentation for
more information: https://torsion.org/borgmatic/docs/how-to/run-arbitrary-borg-commands/
1.5.14
* #390: Add link to Hetzner storage offering from the documentation.

45
borgmatic/borg/borg.py Normal file
View file

@ -0,0 +1,45 @@
import logging
from borgmatic.borg.flags import make_flags
from borgmatic.execute import execute_command
logger = logging.getLogger(__name__)
REPOSITORYLESS_BORG_COMMANDS = {'serve', None}
def run_arbitrary_borg(
repository, storage_config, options, archive=None, local_path='borg', remote_path=None
):
'''
Given a local or remote repository path, a storage config dict, a sequence of arbitrary
command-line Borg options, and an optional archive name, run an arbitrary Borg command on the
given repository/archive.
'''
lock_wait = storage_config.get('lock_wait', None)
try:
options = options[1:] if options[0] == '--' else options
borg_command = options[0]
command_options = tuple(options[1:])
except IndexError:
borg_command = None
command_options = ()
repository_archive = '::'.join((repository, archive)) if repository and archive else repository
full_command = (
(local_path,)
+ ((borg_command,) if borg_command else ())
+ ((repository_archive,) if borg_command and repository_archive else ())
+ command_options
+ (('--info',) if logger.getEffectiveLevel() == logging.INFO else ())
+ (('--debug', '--show-rc') if logger.isEnabledFor(logging.DEBUG) else ())
+ make_flags('remote-path', remote_path)
+ make_flags('lock-wait', lock_wait)
)
return execute_command(
full_command, output_log_level=logging.WARNING, borg_local_path=local_path,
)

3
borgmatic/borg/create.py
View file

@ -220,7 +220,8 @@ def create_archive(
extra_borg_options = storage_config.get('extra_borg_options', {}).get('create', '')
full_command = (
(local_path, 'create')
tuple(local_path.split(' '))
+ ('create',)
+ _make_pattern_flags(location_config, pattern_file.name if pattern_file else None)
+ _make_exclude_flags(location_config, exclude_file.name if exclude_file else None)
+ (('--checkpoint-interval', str(checkpoint_interval)) if checkpoint_interval else ())

93
borgmatic/commands/arguments.py
View file

@ -15,17 +15,18 @@ SUBPARSER_ALIASES = {
'restore': ['--restore', '-r'],
'list': ['--list', '-l'],
'info': ['--info', '-i'],
'borg': [],
}
def parse_subparser_arguments(unparsed_arguments, subparsers):
'''
Given a sequence of arguments, and a subparsers object as returned by
argparse.ArgumentParser().add_subparsers(), give each requested action's subparser a shot at
parsing all arguments. This allows common arguments like "--repository" to be shared across
multiple subparsers.
Given a sequence of arguments and a dict from subparser name to argparse.ArgumentParser
instance, give each requested action's subparser a shot at parsing all arguments. This allows
common arguments like "--repository" to be shared across multiple subparsers.
Return the result as a dict mapping from subparser name to a parsed namespace of arguments.
Return the result as a tuple of (a dict mapping from subparser name to a parsed namespace of
arguments, a list of remaining arguments not claimed by any subparser).
'''
arguments = collections.OrderedDict()
remaining_arguments = list(unparsed_arguments)
@ -35,7 +36,12 @@ def parse_subparser_arguments(unparsed_arguments, subparsers):
for alias in aliases
}
for subparser_name, subparser in subparsers.choices.items():
# If the "borg" action is used, skip all other subparsers. This avoids confusion like
# "borg list" triggering borgmatic's own list action.
if 'borg' in unparsed_arguments:
subparsers = {'borg': subparsers['borg']}
for subparser_name, subparser in subparsers.items():
if subparser_name not in remaining_arguments:
continue
@ -47,11 +53,11 @@ def parse_subparser_arguments(unparsed_arguments, subparsers):
parsed, unused_remaining = subparser.parse_known_args(unparsed_arguments)
for value in vars(parsed).values():
if isinstance(value, str):
if value in subparsers.choices:
if value in subparsers:
remaining_arguments.remove(value)
elif isinstance(value, list):
for item in value:
if item in subparsers.choices:
if item in subparsers:
remaining_arguments.remove(item)
arguments[canonical_name] = parsed
@ -59,47 +65,33 @@ def parse_subparser_arguments(unparsed_arguments, subparsers):
# 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]
subparser = subparsers[subparser_name]
parsed, unused_remaining = subparser.parse_known_args(unparsed_arguments)
arguments[subparser_name] = parsed
return arguments
def parse_global_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(), parse and return any global
arguments as a parsed argparse.Namespace instance.
'''
# Ask each subparser, one by one, to greedily consume arguments. Any arguments that remain
# are global arguments.
remaining_arguments = list(unparsed_arguments)
present_subparser_names = set()
for subparser_name, subparser in subparsers.choices.items():
if subparser_name not in remaining_arguments:
# Now ask each subparser, one by one, to greedily consume arguments.
for subparser_name, subparser in subparsers.items():
if subparser_name not in arguments.keys():
continue
present_subparser_names.add(subparser_name)
subparser = subparsers[subparser_name]
unused_parsed, remaining_arguments = subparser.parse_known_args(remaining_arguments)
# If no actions are explicitly requested, assume defaults: prune, create, and check.
if (
not present_subparser_names
and '--help' not in unparsed_arguments
and '-h' not in unparsed_arguments
):
for subparser_name in ('prune', 'create', 'check'):
subparser = subparsers.choices[subparser_name]
unused_parsed, remaining_arguments = subparser.parse_known_args(remaining_arguments)
# Special case: If "borg" is present in the arguments, consume all arguments after (+1) the
# "borg" action.
if 'borg' in arguments:
borg_options_index = remaining_arguments.index('borg') + 1
arguments['borg'].options = remaining_arguments[borg_options_index:]
remaining_arguments = remaining_arguments[:borg_options_index]
# Remove the subparser names themselves.
for subparser_name in present_subparser_names:
for subparser_name, subparser in subparsers.items():
if subparser_name in remaining_arguments:
remaining_arguments.remove(subparser_name)
return top_level_parser.parse_args(remaining_arguments)
return (arguments, remaining_arguments)
class Extend_action(Action):
@ -510,8 +502,7 @@ def parse_arguments(*unparsed_arguments):
)
list_group = list_parser.add_argument_group('list arguments')
list_group.add_argument(
'--repository',
help='Path of repository to list, defaults to the configured repository if there is only one',
'--repository', help='Path of repository to list, defaults to the configured repositories',
)
list_group.add_argument('--archive', help='Name of archive to list (or "latest")')
list_group.add_argument(
@ -601,8 +592,32 @@ def parse_arguments(*unparsed_arguments):
)
info_group.add_argument('-h', '--help', action='help', help='Show this help message and exit')
arguments = parse_subparser_arguments(unparsed_arguments, subparsers)
arguments['global'] = parse_global_arguments(unparsed_arguments, top_level_parser, subparsers)
borg_parser = subparsers.add_parser(
'borg',
aliases=SUBPARSER_ALIASES['borg'],
help='Run an arbitrary Borg command',
description='Run an arbitrary Borg command based on borgmatic\'s configuration',
add_help=False,
)
borg_group = borg_parser.add_argument_group('borg arguments')
borg_group.add_argument(
'--repository',
help='Path of repository to pass to Borg, defaults to the configured repositories',
)
borg_group.add_argument('--archive', help='Name of archive to pass to Borg (or "latest")')
borg_group.add_argument(
'--',
metavar='OPTION',
dest='options',
nargs='+',
help='Options to pass to Borg, command first ("create", "list", etc). "--" is optional. To specify the repository or the archive, you must use --repository or --archive instead of providing them here.',
)
borg_group.add_argument('-h', '--help', action='help', help='Show this help message and exit')
arguments, remaining_arguments = parse_subparser_arguments(
unparsed_arguments, subparsers.choices
)
arguments['global'] = top_level_parser.parse_args(remaining_arguments)
if arguments['global'].excludes_filename:
raise ValueError(

17
borgmatic/commands/borgmatic.py
View file

@ -9,6 +9,7 @@ from subprocess import CalledProcessError
import colorama
import pkg_resources
from borgmatic.borg import borg as borg_borg
from borgmatic.borg import check as borg_check
from borgmatic.borg import create as borg_create
from borgmatic.borg import environment as borg_environment
@ -543,6 +544,22 @@ def run_actions(
)
if json_output:
yield json.loads(json_output)
if 'borg' in arguments:
if arguments['borg'].repository is None or validate.repositories_match(
repository, arguments['borg'].repository
):
logger.warning('{}: Running arbitrary Borg command'.format(repository))
archive_name = borg_list.resolve_archive_name(
repository, arguments['borg'].archive, storage, local_path, remote_path
)
borg_borg.run_arbitrary_borg(
repository,
storage,
options=arguments['borg'].options,
archive=archive_name,
local_path=local_path,
remote_path=remote_path,
)
def load_configurations(config_filenames, overrides=None):

2
docs/Dockerfile
View file

@ -3,7 +3,7 @@ FROM python:3.8-alpine3.12 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 \
&& for action in init prune create check extract mount umount restore list info; do \
&& for action in init prune create check extract export-tar mount umount restore list info borg; do \
echo -e "\n--------------------------------------------------------------------------------\n" >> /command-line.txt \
&& borgmatic "$action" --help >> /command-line.txt; done

2
docs/how-to/develop-on-borgmatic.md
View file

@ -3,7 +3,7 @@ title: How to develop on borgmatic
eleventyNavigation:
key: Develop on borgmatic
parent: How-to guides
order: 11
order: 12
---
## Source code

94
docs/how-to/run-arbitrary-borg-commands.md Normal file
View file

@ -0,0 +1,94 @@
---
title: How to run arbitrary Borg commands
eleventyNavigation:
key: Run arbitrary Borg commands
parent: How-to guides
order: 10
---
## Running Borg with borgmatic
Borg has several commands (and options) that borgmatic does not currently
support. Sometimes though, as a borgmatic user, you may find yourself wanting
to take advantage of these off-the-beaten-path Borg features. You could of
course drop down to running Borg directly. But then you'd give up all the
niceties of your borgmatic configuration. You could file a [borgmatic
ticket](https://torsion.org/borgmatic/#issues) or even a [pull
request](https://torsion.org/borgmatic/#contributing) to add the feature. But
what if you need it *now*?
That's where borgmatic's support for running "arbitrary" Borg commands comes
in. Running Borg commands with borgmatic takes advantage of the following, all
based on your borgmatic configuration files or command-line arguments:
* configured repositories (automatically runs your Borg command once for each
one)
* local and remote Borg binary paths
* SSH settings and Borg environment variables
* lock wait settings
* verbosity
### borg action
The way you run Borg with borgmatic is via the `borg` action. Here's a simple
example:
```bash
borgmatic borg break-lock
```
(No `borg` action in borgmatic? Time to upgrade!)
This runs Borg's `break-lock` command once on each configured borgmatic
repository. Notice how the repository isn't present in the specified Borg
options, as that part is provided by borgmatic.
You can also specify Borg options for relevant commands:
```bash
borgmatic borg list --progress
```
This runs Borg's `list` command once on each configured borgmatic
repository. However, the native `borgmatic list` action should be preferred
for most use.
What if you only want to run Borg on a single configured borgmatic repository
when you've got several configured? Not a problem.
```bash
borgmatic borg --repository repo.borg break-lock
```
And what about a single archive?
```bash
borgmatic borg --archive your-archive-name list
```
### Limitations
borgmatic's `borg` action is not without limitations:
* The Borg command you want to run (`create`, `list`, etc.) *must* come first
after the `borg` action. If you have any other Borg options to specify,
provide them after. For instance, `borgmatic borg list --progress` will work,
but `borgmatic borg --progress list` will not.
* borgmatic supplies the repository/archive name to Borg for you (based on
your borgmatic configuration or the `borgmatic borg --repository`/`--archive`
arguments), so do not specify the repository/archive otherwise.
* The `borg` action will not currently work for any Borg commands like `borg
serve` that do not accept a repository/archive name.
* Do not specify any global borgmatic arguments to the right of the `borg`
action. (They will be passed to Borg instead of borgmatic.) If you have
global borgmatic arguments, specify them *before* the `borg` action.
* Unlike other borgmatic actions, you cannot combine the `borg` action with
other borgmatic actions. This is to prevent ambiguity in commands like
`borgmatic borg list`, in which `list` is both a valid Borg command and a
borgmatic action. In this case, only the Borg command is run.
* Unlike normal borgmatic actions that support JSON, the `borg` action will
not disable certain borgmatic logs to avoid interfering with JSON output.
In general, this `borgmatic borg` feature should be considered an escape
valve—a feature of second resort. In the long run, it's preferable to wrap
Borg commands with borgmatic actions that can support them fully.

2
docs/how-to/upgrade.md
View file

@ -3,7 +3,7 @@ title: How to upgrade borgmatic
eleventyNavigation:
key: Upgrade borgmatic
parent: How-to guides
order: 10
order: 11
---
## Upgrading

18
tests/integration/commands/test_arguments.py
View file

@ -163,6 +163,24 @@ def test_parse_arguments_with_help_and_action_shows_action_help(capsys):
assert 'create arguments:' in captured.out
def test_parse_arguments_with_action_before_global_options_parses_options():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
arguments = module.parse_arguments('prune', '--verbosity', '2')
assert 'prune' in arguments
assert arguments['global'].verbosity == 2
def test_parse_arguments_with_global_options_before_action_parses_options():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
arguments = module.parse_arguments('--verbosity', '2', 'prune')
assert 'prune' in arguments
assert arguments['global'].verbosity == 2
def test_parse_arguments_with_prune_action_leaves_other_actions_disabled():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])

123
tests/unit/borg/test_borg.py Normal file
View file

@ -0,0 +1,123 @@
import logging
from flexmock import flexmock
from borgmatic.borg import borg as module
from ..test_verbosity import insert_logging_mock
def test_run_arbitrary_borg_calls_borg_with_parameters():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'break-lock', 'repo'), output_log_level=logging.WARNING, borg_local_path='borg'
)
module.run_arbitrary_borg(
repository='repo', storage_config={}, options=['break-lock'],
)
def test_run_arbitrary_borg_with_log_info_calls_borg_with_info_parameter():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'break-lock', 'repo', '--info'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
insert_logging_mock(logging.INFO)
module.run_arbitrary_borg(
repository='repo', storage_config={}, options=['break-lock'],
)
def test_run_arbitrary_borg_with_log_debug_calls_borg_with_debug_parameter():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'break-lock', 'repo', '--debug', '--show-rc'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
insert_logging_mock(logging.DEBUG)
module.run_arbitrary_borg(
repository='repo', storage_config={}, options=['break-lock'],
)
def test_run_arbitrary_borg_with_lock_wait_calls_borg_with_lock_wait_parameters():
storage_config = {'lock_wait': 5}
flexmock(module).should_receive('execute_command').with_args(
('borg', 'break-lock', 'repo', '--lock-wait', '5'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
module.run_arbitrary_borg(
repository='repo', storage_config=storage_config, options=['break-lock'],
)
def test_run_arbitrary_borg_with_archive_calls_borg_with_archive_parameter():
storage_config = {}
flexmock(module).should_receive('execute_command').with_args(
('borg', 'break-lock', 'repo::archive'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
module.run_arbitrary_borg(
repository='repo', storage_config=storage_config, options=['break-lock'], archive='archive',
)
def test_run_arbitrary_borg_with_local_path_calls_borg_via_local_path():
flexmock(module).should_receive('execute_command').with_args(
('borg1', 'break-lock', 'repo'), output_log_level=logging.WARNING, borg_local_path='borg1'
)
module.run_arbitrary_borg(
repository='repo', storage_config={}, options=['break-lock'], local_path='borg1',
)
def test_run_arbitrary_borg_with_remote_path_calls_borg_with_remote_path_parameters():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'break-lock', 'repo', '--remote-path', 'borg1'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
module.run_arbitrary_borg(
repository='repo', storage_config={}, options=['break-lock'], remote_path='borg1',
)
def test_run_arbitrary_borg_passes_borg_specific_parameters_to_borg():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', 'repo', '--progress'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
module.run_arbitrary_borg(
repository='repo', storage_config={}, options=['list', '--progress'],
)
def test_run_arbitrary_borg_omits_dash_dash_in_parameters_passed_to_borg():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'break-lock', 'repo'), output_log_level=logging.WARNING, borg_local_path='borg',
)
module.run_arbitrary_borg(
repository='repo', storage_config={}, options=['--', 'break-lock'],
)
def test_run_arbitrary_borg_without_borg_specific_parameters_does_not_raise():
flexmock(module).should_receive('execute_command').with_args(
('borg',), output_log_level=logging.WARNING, borg_local_path='borg',
)
module.run_arbitrary_borg(
repository='repo', storage_config={}, options=[],
)

167
tests/unit/commands/test_arguments.py
View file

@ -5,146 +5,137 @@ from borgmatic.commands import arguments as module
def test_parse_subparser_arguments_consumes_subparser_arguments_before_subparser_name():
action_namespace = flexmock(foo=True)
subparsers = flexmock(
choices={
'action': flexmock(parse_known_args=lambda arguments: (action_namespace, [])),
'other': flexmock(),
}
subparsers = {
'action': flexmock(parse_known_args=lambda arguments: (action_namespace, ['action'])),
'other': flexmock(),
}
arguments, remaining_arguments = module.parse_subparser_arguments(
('--foo', 'true', 'action'), subparsers
)
arguments = module.parse_subparser_arguments(('--foo', 'true', 'action'), subparsers)
assert arguments == {'action': action_namespace}
assert remaining_arguments == []
def test_parse_subparser_arguments_consumes_subparser_arguments_after_subparser_name():
action_namespace = flexmock(foo=True)
subparsers = flexmock(
choices={
'action': flexmock(parse_known_args=lambda arguments: (action_namespace, [])),
'other': flexmock(),
}
subparsers = {
'action': flexmock(parse_known_args=lambda arguments: (action_namespace, ['action'])),
'other': flexmock(),
}
arguments, remaining_arguments = module.parse_subparser_arguments(
('action', '--foo', 'true'), subparsers
)
arguments = module.parse_subparser_arguments(('action', '--foo', 'true'), subparsers)
assert arguments == {'action': action_namespace}
assert remaining_arguments == []
def test_parse_subparser_arguments_consumes_subparser_arguments_with_alias():
action_namespace = flexmock(foo=True)
action_subparser = flexmock(parse_known_args=lambda arguments: (action_namespace, []))
subparsers = flexmock(
choices={
'action': action_subparser,
'-a': action_subparser,
'other': flexmock(),
'-o': flexmock(),
}
)
action_subparser = flexmock(parse_known_args=lambda arguments: (action_namespace, ['action']))
subparsers = {
'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'), subparsers)
arguments, remaining_arguments = module.parse_subparser_arguments(
('-a', '--foo', 'true'), subparsers
)
assert arguments == {'action': action_namespace}
assert remaining_arguments == []
def test_parse_subparser_arguments_consumes_multiple_subparser_arguments():
action_namespace = flexmock(foo=True)
other_namespace = flexmock(bar=3)
subparsers = flexmock(
choices={
'action': flexmock(
parse_known_args=lambda arguments: (action_namespace, ['--bar', '3'])
),
'other': flexmock(parse_known_args=lambda arguments: (other_namespace, [])),
}
)
subparsers = {
'action': flexmock(
parse_known_args=lambda arguments: (action_namespace, ['action', '--bar', '3'])
),
'other': flexmock(parse_known_args=lambda arguments: (other_namespace, [])),
}
arguments = module.parse_subparser_arguments(
arguments, remaining_arguments = module.parse_subparser_arguments(
('action', '--foo', 'true', 'other', '--bar', '3'), subparsers
)
assert arguments == {'action': action_namespace, 'other': other_namespace}
assert remaining_arguments == []
def test_parse_subparser_arguments_applies_default_subparsers():
prune_namespace = flexmock()
create_namespace = flexmock(progress=True)
check_namespace = flexmock()
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(),
}
)
subparsers = {
'prune': flexmock(
parse_known_args=lambda arguments: (prune_namespace, ['prune', '--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'), subparsers)
arguments, remaining_arguments = module.parse_subparser_arguments(('--progress'), subparsers)
assert arguments == {
'prune': prune_namespace,
'create': create_namespace,
'check': check_namespace,
}
assert remaining_arguments == []
def test_parse_global_arguments_with_help_does_not_apply_default_subparsers():
global_namespace = flexmock(verbosity='lots')
def test_parse_subparser_arguments_passes_through_unknown_arguments_before_subparser_name():
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(),
}
subparsers = {
'action': flexmock(
parse_known_args=lambda arguments: (action_namespace, ['action', '--verbosity', 'lots'])
),
'other': flexmock(),
}
arguments, remaining_arguments = module.parse_subparser_arguments(
('--verbosity', 'lots', 'action'), subparsers
)
arguments = module.parse_global_arguments(
('--verbosity', 'lots', '--help'), top_level_parser, subparsers
)
assert arguments == global_namespace
assert arguments == {'action': action_namespace}
assert remaining_arguments == ['--verbosity', 'lots']
def test_parse_global_arguments_consumes_global_arguments_before_subparser_name():
global_namespace = flexmock(verbosity='lots')
def test_parse_subparser_arguments_passes_through_unknown_arguments_after_subparser_name():
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(),
}
subparsers = {
'action': flexmock(
parse_known_args=lambda arguments: (action_namespace, ['action', '--verbosity', 'lots'])
),
'other': flexmock(),
}
arguments, remaining_arguments = module.parse_subparser_arguments(
('action', '--verbosity', 'lots'), subparsers
)
arguments = module.parse_global_arguments(
('--verbosity', 'lots', 'action'), top_level_parser, subparsers
)
assert arguments == global_namespace
assert arguments == {'action': action_namespace}
assert remaining_arguments == ['--verbosity', 'lots']
def test_parse_global_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(),
}
)
def test_parse_subparser_arguments_parses_borg_options_and_skips_other_subparsers():
action_namespace = flexmock(options=[])
subparsers = {
'borg': flexmock(parse_known_args=lambda arguments: (action_namespace, ['borg', 'list'])),
'list': flexmock(),
}
arguments = module.parse_global_arguments(
('action', '--verbosity', 'lots'), top_level_parser, subparsers
)
arguments, remaining_arguments = module.parse_subparser_arguments(('borg', 'list'), subparsers)
assert arguments == global_namespace
assert arguments == {'borg': action_namespace}
assert arguments['borg'].options == ['list']
assert remaining_arguments == []