Compare commits

...

10 Commits

11 changed files with 379 additions and 202 deletions

8
NEWS
View File

@ -1,3 +1,11 @@
1.6.3.dev0
* #541: Add "borgmatic list --find" flag for searching for files across multiple archives, useful
for hunting down that file you accidentally deleted so you can extract it. See the documentation
for more information:
https://torsion.org/borgmatic/docs/how-to/inspect-your-backups/#searching-for-a-file
* Deprecate "borgmatic list --successful" flag, as listing only non-checkpoint (successful)
archives is now the default in newer versions of Borg.
1.6.2
* #523: Reduce the default consistency check frequency and support configuring the frequency
independently for each check. Also add "borgmatic check --force" flag to ignore configured

View File

@ -1,4 +1,6 @@
import copy
import logging
import re
from borgmatic.borg.flags import make_flags, make_flags_from_arguments
from borgmatic.execute import execute_command
@ -6,17 +8,11 @@ from borgmatic.execute import execute_command
logger = logging.getLogger(__name__)
# A hack to convince Borg to exclude archives ending in ".checkpoint". This assumes that a
# non-checkpoint archive name ends in a digit (e.g. from a timestamp).
BORG_EXCLUDE_CHECKPOINTS_GLOB = '*[0123456789]'
def resolve_archive_name(repository, archive, storage_config, local_path='borg', remote_path=None):
'''
Given a local or remote repository path, an archive name, a storage config dict, a local Borg
path, and a remote Borg path, simply return the archive name. But if the archive name is
"latest", then instead introspect the repository for the latest successful (non-checkpoint)
archive, and return its name.
"latest", then instead introspect the repository for the latest archive and return its name.
Raise ValueError if "latest" is given but there are no archives in the repository.
'''
@ -31,7 +27,6 @@ def resolve_archive_name(repository, archive, storage_config, local_path='borg',
+ (('--debug', '--show-rc') if logger.isEnabledFor(logging.DEBUG) else ())
+ make_flags('remote-path', remote_path)
+ make_flags('lock-wait', lock_wait)
+ make_flags('glob-archives', BORG_EXCLUDE_CHECKPOINTS_GLOB)
+ make_flags('last', 1)
+ ('--short', repository)
)
@ -47,17 +42,20 @@ def resolve_archive_name(repository, archive, storage_config, local_path='borg',
return latest_archive
def list_archives(repository, storage_config, list_arguments, local_path='borg', remote_path=None):
MAKE_FLAGS_EXCLUDES = ('repository', 'archive', 'successful', 'paths', 'find_paths')
def make_list_command(
repository, storage_config, list_arguments, local_path='borg', remote_path=None
):
'''
Given a local or remote repository path, a storage config dict, and the arguments to the list
action, display the output of listing Borg archives in the repository or return JSON output. Or,
if an archive name is given, listing the files in that archive.
Given a local or remote repository path, a storage config dict, the arguments to the list
action, and local and remote Borg paths, return a command as a tuple to list archives or paths
within an archive.
'''
lock_wait = storage_config.get('lock_wait', None)
if list_arguments.successful:
list_arguments.glob_archives = BORG_EXCLUDE_CHECKPOINTS_GLOB
full_command = (
return (
(local_path, 'list')
+ (
('--info',)
@ -71,19 +69,92 @@ def list_archives(repository, storage_config, list_arguments, local_path='borg',
)
+ make_flags('remote-path', remote_path)
+ make_flags('lock-wait', lock_wait)
+ make_flags_from_arguments(
list_arguments, excludes=('repository', 'archive', 'paths', 'successful')
)
+ make_flags_from_arguments(list_arguments, excludes=MAKE_FLAGS_EXCLUDES,)
+ (
'::'.join((repository, list_arguments.archive))
('::'.join((repository, list_arguments.archive)),)
if list_arguments.archive
else repository,
else (repository,)
)
+ (tuple(list_arguments.paths) if list_arguments.paths else ())
)
return execute_command(
full_command,
output_log_level=None if list_arguments.json else logging.WARNING,
borg_local_path=local_path,
def make_find_paths(find_paths):
'''
Given a sequence of path fragments or patterns as passed to `--find`, transform all path
fragments into glob patterns. Pass through existing patterns untouched.
For example, given find_paths of:
['foo.txt', 'pp:root/somedir']
... transform that into:
['sh:**/*foo.txt*/**', 'pp:root/somedir']
'''
if not find_paths:
return ()
return tuple(
find_path
if re.compile(r'([-!+RrPp] )|(\w\w:)').match(find_path)
else f'sh:**/*{find_path}*/**'
for find_path in find_paths
)
def list_archives(repository, storage_config, list_arguments, local_path='borg', remote_path=None):
'''
Given a local or remote repository path, a storage config dict, the arguments to the list
action, and local and remote Borg paths, display the output of listing Borg archives in the
repository or return JSON output. Or, if an archive name is given, list the files in that
archive. Or, if list_arguments.find_paths are given, list the files by searching across multiple
archives.
'''
# If there are any paths to find (and there's not a single archive already selected), start by
# getting a list of archives to search.
if list_arguments.find_paths and not list_arguments.archive:
repository_arguments = copy.copy(list_arguments)
repository_arguments.archive = None
repository_arguments.json = False
repository_arguments.format = None
# Ask Borg to list archives. Capture its output for use below.
archive_lines = tuple(
execute_command(
make_list_command(
repository, storage_config, repository_arguments, local_path, remote_path
),
output_log_level=None,
borg_local_path=local_path,
)
.strip('\n')
.split('\n')
)
else:
archive_lines = (list_arguments.archive,)
# For each archive listed by Borg, run list on the contents of that archive.
for archive_line in archive_lines:
try:
archive = archive_line.split()[0]
except (AttributeError, IndexError):
archive = None
if archive:
logger.warning(archive_line)
archive_arguments = copy.copy(list_arguments)
archive_arguments.archive = archive
main_command = make_list_command(
repository, storage_config, archive_arguments, local_path, remote_path
) + make_find_paths(list_arguments.find_paths)
output = execute_command(
main_command,
output_log_level=None if list_arguments.json else logging.WARNING,
borg_local_path=local_path,
)
if list_arguments.json:
return output

View File

@ -554,7 +554,14 @@ def make_parsers():
metavar='PATH',
nargs='+',
dest='paths',
help='Paths to list from archive, defaults to the entire archive',
help='Paths or patterns to list from a single selected archive (via "--archive"), defaults to listing the entire archive',
)
list_group.add_argument(
'--find',
metavar='PATH',
nargs='+',
dest='find_paths',
help='Partial paths or patterns to search for and list across multiple archives',
)
list_group.add_argument(
'--short', default=False, action='store_true', help='Output only archive or path names'
@ -571,9 +578,9 @@ def make_parsers():
)
list_group.add_argument(
'--successful',
default=False,
default=True,
action='store_true',
help='Only list archive names of successful (non-checkpoint) backups',
help='Deprecated in favor of listing successful (non-checkpoint) backups by default in newer versions of Borg',
)
list_group.add_argument(
'--sort-by', metavar='KEYS', help='Comma-separated list of sorting keys'
@ -681,9 +688,6 @@ def parse_arguments(*unparsed_arguments):
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 arguments['list'].glob_archives and arguments['list'].successful:
raise ValueError('The --glob-archives and --successful options cannot be used together')
if (
'list' in arguments
and 'info' in arguments

View File

@ -116,7 +116,7 @@ Omit the `--archive` flag to mount all archives (lazy-loaded):
borgmatic mount --mount-point /mnt
```
Or use the "latest" value for the archive to mount the latest successful archive:
Or use the "latest" value for the archive to mount the latest archive:
```bash
borgmatic mount --archive latest --mount-point /mnt

View File

@ -51,6 +51,31 @@ borgmatic info
`--info`. Or upgrade borgmatic!)
### Searching for a file
Let's say you've accidentally deleted a file and want to find the backup
archive(s) containing it. `borgmatic list` provides a `--find` flag for
exactly this purpose (as of borgmatic 1.6.3). For instance, if you're looking
for a `foo.txt`:
```bash
borgmatic list --find foo.txt
```
This will list your archives and indicate those with files matching
`*foo.txt*` anywhere in the archive. The `--find` parameter can alternatively
be a [Borg
pattern](https://borgbackup.readthedocs.io/en/stable/usage/help.html#borg-patterns).
To limit the archives searched, use the standard `list` parameters for
filtering archives such as `--last`, `--archive`, `--glob-archives`, etc. For
example, to search only the last five archives:
```bash
borgmatic list --find foo.txt --last 5
```
## Logging
By default, borgmatic logs to a local syslog-compatible daemon if one is

View File

@ -286,35 +286,12 @@ output only shows up at the console, and not in syslog.
* [Borgmacator GNOME AppIndicator](https://github.com/N-Coder/borgmacator/)
### Successful backups
`borgmatic list` includes support for a `--successful` flag that only lists
successful (non-checkpoint) backups. This flag works via a basic heuristic: It
assumes that non-checkpoint archive names end with a digit (e.g. from a
timestamp), while checkpoint archive names do not. This means that if you're
using custom archive names that do not end in a digit, the `--successful` flag
will not work as expected.
Combined with a built-in Borg flag like `--last`, you can list the last
successful backup for use in your monitoring scripts. Here's an example
combined with `--json`:
```bash
borgmatic list --successful --last 1 --json
```
Note that this particular combination will only work if you've got a single
backup "series" in your repository. If you're instead backing up, say, from
multiple different hosts into a single repository, then you'll need to get
fancier with your archive listing. See `borg list --help` for more flags.
### Latest backups
All borgmatic actions that accept an "--archive" flag allow you to specify an
archive name of "latest". This lets you get the latest successful archive
without having to first run "borgmatic list" manually, which can be handy in
automated scripts. Here's an example:
archive name of "latest". This lets you get the latest archive without having
to first run "borgmatic list" manually, which can be handy in automated
scripts. Here's an example:
```bash
borgmatic info --archive latest

View File

@ -92,6 +92,7 @@ installing borgmatic:
* [Alpine Linux](https://pkgs.alpinelinux.org/packages?name=borgmatic)
* [OpenBSD](http://ports.su/sysutils/borgmatic)
* [openSUSE](https://software.opensuse.org/package/borgmatic)
* [macOS (via Homebrew)](https://formulae.brew.sh/formula/borgmatic)
* [Ansible role](https://github.com/borgbase/ansible-role-borgbackup)
* [virtualenv](https://virtualenv.pypa.io/en/stable/)
@ -311,21 +312,30 @@ Access](https://projects.torsion.org/borgmatic-collective/borgmatic/issues/293).
borgmatic includes a shell completion script (currently only for Bash) to
support tab-completing borgmatic command-line actions and flags. Depending on
how you installed borgmatic, this may be enabled by default. But if it's not,
you can install the shell completion script globally:
start by installing the `bash-completion` Linux package or the
[`bash-completion@2`](https://formulae.brew.sh/formula/bash-completion@2)
macOS Homebrew formula. Then, install the shell completion script globally:
```bash
sudo su -c "borgmatic --bash-completion > $(pkg-config --variable=completionsdir bash-completion)/borgmatic"
```
If you don't have `pkg-config` installed, you can try the following path
instead:
```bash
sudo su -c "borgmatic --bash-completion > /usr/share/bash-completion/completions/borgmatic"
```
Alternatively, if you'd like to install the script for just the current user:
Or, if you'd like to install the script for just the current user:
```bash
mkdir --parents ~/.local/share/bash-completion/completions
borgmatic --bash-completion > ~/.local/share/bash-completion/completions/borgmatic
```
In either case, you may also need to install the `bash-completion` Linux
package and restart your shell (`exit` and open a new shell).
Finally, restart your shell (`exit` and open a new shell) so the completions
take effect.
### Colored output

View File

@ -1,6 +1,6 @@
from setuptools import find_packages, setup
VERSION = '1.6.2'
VERSION = '1.6.3.dev0'
setup(

View File

@ -296,15 +296,6 @@ def test_parse_arguments_disallows_init_and_dry_run():
)
def test_parse_arguments_disallows_glob_archives_with_successful():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])
with pytest.raises(ValueError):
module.parse_arguments(
'--config', 'myconfig', 'list', '--glob-archives', '*glob*', '--successful'
)
def test_parse_arguments_disallows_repository_unless_action_consumes_it():
flexmock(module.collect).should_receive('get_default_config_paths').and_return(['default'])

View File

@ -1,3 +1,4 @@
import argparse
import logging
import pytest
@ -8,8 +9,6 @@ from borgmatic.borg import list as module
from ..test_verbosity import insert_logging_mock
BORG_LIST_LATEST_ARGUMENTS = (
'--glob-archives',
module.BORG_EXCLUDE_CHECKPOINTS_GLOB,
'--last',
'1',
'--short',
@ -108,156 +107,125 @@ def test_resolve_archive_name_with_lock_wait_calls_borg_with_lock_wait_parameter
)
def test_list_archives_calls_borg_with_parameters():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', 'repo'), output_log_level=logging.WARNING, borg_local_path='borg'
)
module.list_archives(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False, successful=False),
)
def test_list_archives_with_log_info_calls_borg_with_info_parameter():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--info', 'repo'), output_log_level=logging.WARNING, borg_local_path='borg'
)
def test_make_list_command_includes_log_info():
insert_logging_mock(logging.INFO)
module.list_archives(
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False, successful=False),
list_arguments=flexmock(archive=None, paths=None, json=False),
)
assert command == ('borg', 'list', '--info', 'repo')
def test_list_archives_with_log_info_and_json_suppresses_most_borg_output():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--json', 'repo'), output_log_level=None, borg_local_path='borg'
)
def test_make_list_command_includes_json_but_not_info():
insert_logging_mock(logging.INFO)
module.list_archives(
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=True, successful=False),
list_arguments=flexmock(archive=None, paths=None, json=True),
)
assert command == ('borg', 'list', '--json', 'repo')
def test_list_archives_with_log_debug_calls_borg_with_debug_parameter():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--debug', '--show-rc', 'repo'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
def test_make_list_command_includes_log_debug():
insert_logging_mock(logging.DEBUG)
module.list_archives(
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False, successful=False),
list_arguments=flexmock(archive=None, paths=None, json=False),
)
assert command == ('borg', 'list', '--debug', '--show-rc', 'repo')
def test_list_archives_with_log_debug_and_json_suppresses_most_borg_output():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--json', 'repo'), output_log_level=None, borg_local_path='borg'
)
def test_make_list_command_includes_json_but_not_debug():
insert_logging_mock(logging.DEBUG)
module.list_archives(
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=True, successful=False),
list_arguments=flexmock(archive=None, paths=None, json=True),
)
def test_list_archives_with_lock_wait_calls_borg_with_lock_wait_parameters():
storage_config = {'lock_wait': 5}
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--lock-wait', '5', 'repo'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
module.list_archives(
repository='repo',
storage_config=storage_config,
list_arguments=flexmock(archive=None, paths=None, json=False, successful=False),
)
assert command == ('borg', 'list', '--json', 'repo')
def test_list_archives_with_archive_calls_borg_with_archive_parameter():
storage_config = {}
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', 'repo::archive'), output_log_level=logging.WARNING, borg_local_path='borg'
)
module.list_archives(
repository='repo',
storage_config=storage_config,
list_arguments=flexmock(archive='archive', paths=None, json=False, successful=False),
)
def test_list_archives_with_path_calls_borg_with_path_parameter():
storage_config = {}
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', 'repo::archive', 'var/lib'),
output_log_level=logging.WARNING,
borg_local_path='borg',
)
module.list_archives(
repository='repo',
storage_config=storage_config,
list_arguments=flexmock(archive='archive', paths=['var/lib'], json=False, successful=False),
)
def test_list_archives_with_local_path_calls_borg_via_local_path():
flexmock(module).should_receive('execute_command').with_args(
('borg1', 'list', 'repo'), output_log_level=logging.WARNING, borg_local_path='borg1'
)
module.list_archives(
def test_make_list_command_includes_json():
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False, successful=False),
local_path='borg1',
list_arguments=flexmock(archive=None, paths=None, json=True),
)
assert command == ('borg', 'list', '--json', 'repo')
def test_list_archives_with_remote_path_calls_borg_with_remote_path_parameters():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--remote-path', 'borg1', 'repo'),
output_log_level=logging.WARNING,
borg_local_path='borg',
def test_make_list_command_includes_lock_wait():
command = module.make_list_command(
repository='repo',
storage_config={'lock_wait': 5},
list_arguments=flexmock(archive=None, paths=None, json=False),
)
module.list_archives(
assert command == ('borg', 'list', '--lock-wait', '5', 'repo')
def test_make_list_command_includes_archive():
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False, successful=False),
remote_path='borg1',
list_arguments=flexmock(archive='archive', paths=None, json=False),
)
assert command == ('borg', 'list', 'repo::archive')
def test_list_archives_with_short_calls_borg_with_short_parameter():
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--short', 'repo'),
output_log_level=logging.WARNING,
borg_local_path='borg',
).and_return('[]')
module.list_archives(
def test_make_list_command_includes_archive_and_path():
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False, successful=False, short=True),
list_arguments=flexmock(archive='archive', paths=['var/lib'], json=False),
)
assert command == ('borg', 'list', 'repo::archive', 'var/lib')
def test_make_list_command_includes_local_path():
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False),
local_path='borg2',
)
assert command == ('borg2', 'list', 'repo')
def test_make_list_command_includes_remote_path():
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False),
remote_path='borg2',
)
assert command == ('borg', 'list', '--remote-path', 'borg2', 'repo')
def test_make_list_command_includes_short():
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False, short=True),
)
assert command == ('borg', 'list', '--short', 'repo')
@pytest.mark.parametrize(
'argument_name',
@ -273,45 +241,156 @@ def test_list_archives_with_short_calls_borg_with_short_parameter():
'patterns_from',
),
)
def test_list_archives_passes_through_arguments_to_borg(argument_name):
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--' + argument_name.replace('_', '-'), 'value', 'repo'),
output_log_level=logging.WARNING,
borg_local_path='borg',
).and_return('[]')
module.list_archives(
def test_make_list_command_includes_additional_flags(argument_name):
command = module.make_list_command(
repository='repo',
storage_config={},
list_arguments=flexmock(
archive=None, paths=None, json=False, successful=False, **{argument_name: 'value'}
archive=None,
paths=None,
json=False,
find_paths=None,
format=None,
**{argument_name: 'value'}
),
)
assert command == ('borg', 'list', '--' + argument_name.replace('_', '-'), 'value', 'repo')
def test_list_archives_with_successful_calls_borg_to_exclude_checkpoints():
def test_make_find_paths_considers_none_as_empty_paths():
assert module.make_find_paths(None) == ()
def test_make_find_paths_passes_through_patterns():
find_paths = (
'fm:*',
'sh:**/*.txt',
're:^.*$',
'pp:root/somedir',
'pf:root/foo.txt',
'R /',
'r /',
'p /',
'P /',
'+ /',
'- /',
'! /',
)
assert module.make_find_paths(find_paths) == find_paths
def test_make_find_paths_adds_globs_to_path_fragments():
assert module.make_find_paths(('foo.txt',)) == ('sh:**/*foo.txt*/**',)
def test_list_archives_calls_borg_with_parameters():
list_arguments = argparse.Namespace(archive=None, paths=None, json=False, find_paths=None)
flexmock(module).should_receive('make_list_command').with_args(
repository='repo',
storage_config={},
list_arguments=list_arguments,
local_path='borg',
remote_path=None,
).and_return(('borg', 'list', 'repo'))
flexmock(module).should_receive('make_find_paths').and_return(())
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--glob-archives', module.BORG_EXCLUDE_CHECKPOINTS_GLOB, 'repo'),
output_log_level=logging.WARNING,
borg_local_path='borg',
).and_return('[]')
('borg', 'list', 'repo'), output_log_level=logging.WARNING, borg_local_path='borg'
).once()
module.list_archives(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=False, successful=True),
repository='repo', storage_config={}, list_arguments=list_arguments,
)
def test_list_archives_with_json_calls_borg_with_json_parameter():
def test_list_archives_with_json_suppresses_most_borg_output():
list_arguments = argparse.Namespace(archive=None, paths=None, json=True, find_paths=None)
flexmock(module).should_receive('make_list_command').with_args(
repository='repo',
storage_config={},
list_arguments=list_arguments,
local_path='borg',
remote_path=None,
).and_return(('borg', 'list', 'repo'))
flexmock(module).should_receive('make_find_paths').and_return(())
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', '--json', 'repo'), output_log_level=None, borg_local_path='borg'
).and_return('[]')
('borg', 'list', 'repo'), output_log_level=None, borg_local_path='borg'
).once()
json_output = module.list_archives(
repository='repo',
storage_config={},
list_arguments=flexmock(archive=None, paths=None, json=True, successful=False),
module.list_archives(
repository='repo', storage_config={}, list_arguments=list_arguments,
)
assert json_output == '[]'
def test_list_archives_calls_borg_with_local_path():
list_arguments = argparse.Namespace(archive=None, paths=None, json=False, find_paths=None)
flexmock(module).should_receive('make_list_command').with_args(
repository='repo',
storage_config={},
list_arguments=list_arguments,
local_path='borg2',
remote_path=None,
).and_return(('borg2', 'list', 'repo'))
flexmock(module).should_receive('make_find_paths').and_return(())
flexmock(module).should_receive('execute_command').with_args(
('borg2', 'list', 'repo'), output_log_level=logging.WARNING, borg_local_path='borg2'
).once()
module.list_archives(
repository='repo', storage_config={}, list_arguments=list_arguments, local_path='borg2',
)
def test_list_archives_calls_borg_multiple_times_with_find_paths():
glob_paths = ('**/*foo.txt*/**',)
list_arguments = argparse.Namespace(
archive=None, paths=None, json=False, find_paths=['foo.txt'], format=None
)
flexmock(module).should_receive('make_list_command').and_return(
('borg', 'list', 'repo')
).and_return(('borg', 'list', 'repo::archive1')).and_return(('borg', 'list', 'repo::archive2'))
flexmock(module).should_receive('make_find_paths').and_return(glob_paths)
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', 'repo'), output_log_level=None, borg_local_path='borg'
).and_return(
'archive1 Sun, 2022-05-29 15:27:04 [abc]\narchive2 Mon, 2022-05-30 19:47:15 [xyz]'
).once()
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', 'repo::archive1') + glob_paths,
output_log_level=logging.WARNING,
borg_local_path='borg',
).once()
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', 'repo::archive2') + glob_paths,
output_log_level=logging.WARNING,
borg_local_path='borg',
).once()
module.list_archives(
repository='repo', storage_config={}, list_arguments=list_arguments,
)
def test_list_archives_calls_borg_with_archive():
list_arguments = argparse.Namespace(archive='archive', paths=None, json=False, find_paths=None)
flexmock(module).should_receive('make_list_command').with_args(
repository='repo',
storage_config={},
list_arguments=list_arguments,
local_path='borg',
remote_path=None,
).and_return(('borg', 'list', 'repo::archive'))
flexmock(module).should_receive('make_find_paths').and_return(())
flexmock(module).should_receive('execute_command').with_args(
('borg', 'list', 'repo::archive'), output_log_level=logging.WARNING, borg_local_path='borg'
).once()
module.list_archives(
repository='repo', storage_config={}, list_arguments=list_arguments,
)

View File

@ -69,10 +69,18 @@ def test_format_buffered_logs_for_payload_without_handler_produces_empty_payload
assert payload == ''
def mock_logger():
logger = flexmock()
logger.should_receive('addHandler')
logger.should_receive('removeHandler')
flexmock(module.logging).should_receive('getLogger').and_return(logger)
def test_initialize_monitor_creates_log_handler_with_ping_body_limit():
ping_body_limit = 100
monitoring_log_level = 1
mock_logger()
flexmock(module).should_receive('Forgetful_buffering_handler').with_args(
ping_body_limit - len(module.PAYLOAD_TRUNCATION_INDICATOR), monitoring_log_level
).once()
@ -85,6 +93,7 @@ def test_initialize_monitor_creates_log_handler_with_ping_body_limit():
def test_initialize_monitor_creates_log_handler_with_default_ping_body_limit():
monitoring_log_level = 1
mock_logger()
flexmock(module).should_receive('Forgetful_buffering_handler').with_args(
module.DEFAULT_PING_BODY_LIMIT_BYTES - len(module.PAYLOAD_TRUNCATION_INDICATOR),
monitoring_log_level,
@ -97,6 +106,7 @@ def test_initialize_monitor_creates_log_handler_with_zero_ping_body_limit():
ping_body_limit = 0
monitoring_log_level = 1
mock_logger()
flexmock(module).should_receive('Forgetful_buffering_handler').with_args(
ping_body_limit, monitoring_log_level
).once()
@ -107,6 +117,7 @@ def test_initialize_monitor_creates_log_handler_with_zero_ping_body_limit():
def test_initialize_monitor_creates_log_handler_when_send_logs_true():
mock_logger()
flexmock(module).should_receive('Forgetful_buffering_handler').once()
module.initialize_monitor(
@ -115,6 +126,7 @@ def test_initialize_monitor_creates_log_handler_when_send_logs_true():
def test_initialize_monitor_bails_when_send_logs_false():
mock_logger()
flexmock(module).should_receive('Forgetful_buffering_handler').never()
module.initialize_monitor(