peter/borgmatic
1
0
Fork 0
mirror of https://github.com/borgmatic-collective/borgmatic.git synced 2026-08-07 00:23:20 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add key import action to import a copy of repository key from backup (#345).

Browse Source 
Reviewed-on: https://projects.torsion.org/borgmatic-collective/borgmatic/pulls/1036
Reviewed-by: Dan Helfman <witten@torsion.org>
This commit is contained in:
Dan Helfman
2025-03-21 21:22:50 +00:00
parent 587d31de7c 8aaa5ba8a6
commit 9ac2a2e286
8 changed files with 466 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/config/test_schema.py
+7 -1
View File
@@ -14,7 +14,13 @@ def test_schema_line_length_stays_under_limit():
assert len(line.rstrip('\n')) <= MAXIMUM_LINE_LENGTH
ACTIONS_MODULE_NAMES_TO_OMIT = {'arguments', 'change_passphrase', 'export_key', 'json'}
ACTIONS_MODULE_NAMES_TO_OMIT = {
'arguments',
'change_passphrase',
'export_key',
'import_key',
'json',
}
ACTIONS_MODULE_NAMES_TO_ADD = {'key', 'umount'}
tests/unit/actions/test_import_key.py
+20
View File
@@ -0,0 +1,20 @@
from flexmock import flexmock
from borgmatic.actions import import_key as module
def test_run_import_key_does_not_raise():
flexmock(module.logger).answer = lambda message: None
flexmock(module.borgmatic.config.validate).should_receive('repositories_match').and_return(True)
flexmock(module.borgmatic.borg.import_key).should_receive('import_key')
import_arguments = flexmock(repository=flexmock())
module.run_import_key(
repository={'path': 'repo'},
config={},
local_borg_version=None,
import_arguments=import_arguments,
global_arguments=flexmock(),
local_path=None,
remote_path=None,
)
tests/unit/borg/test_import_key.py
+279
View File
@@ -0,0 +1,279 @@
import logging
import pytest
from flexmock import flexmock
from borgmatic.borg import import_key as module
from ..test_verbosity import insert_logging_mock
def insert_execute_command_mock(
command, input_file=module.DO_NOT_CAPTURE, working_directory=None, borg_exit_codes=None
):
flexmock(module.environment).should_receive('make_environment')
flexmock(module.borgmatic.config.paths).should_receive('get_working_directory').and_return(
working_directory,
)
flexmock(module).should_receive('execute_command').with_args(
command,
input_file=input_file,
output_log_level=module.logging.INFO,
environment=None,
working_directory=working_directory,
borg_local_path=command[0],
borg_exit_codes=borg_exit_codes,
).once()
def test_import_key_calls_borg_with_required_flags():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', 'repo'))
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_calls_borg_with_local_path():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg1', 'key', 'import', 'repo'))
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
local_path='borg1',
)
def test_import_key_calls_borg_using_exit_codes():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
borg_exit_codes = flexmock()
insert_execute_command_mock(('borg', 'key', 'import', 'repo'), borg_exit_codes=borg_exit_codes)
module.import_key(
repository_path='repo',
config={'borg_exit_codes': borg_exit_codes},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_calls_borg_with_remote_path_flags():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', '--remote-path', 'borg1', 'repo'))
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
remote_path='borg1',
)
def test_import_key_calls_borg_with_umask_flags():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', '--umask', '0770', 'repo'))
module.import_key(
repository_path='repo',
config={'umask': '0770'},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_calls_borg_with_log_json_flags():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', '--log-json', 'repo'))
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=True),
)
def test_import_key_calls_borg_with_lock_wait_flags():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', '--lock-wait', '5', 'repo'))
module.import_key(
repository_path='repo',
config={'lock_wait': '5'},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_with_log_info_calls_borg_with_info_parameter():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', '--info', 'repo'))
insert_logging_mock(logging.INFO)
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_with_log_debug_calls_borg_with_debug_flags():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', '--debug', '--show-rc', 'repo'))
insert_logging_mock(logging.DEBUG)
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_calls_borg_with_paper_flags():
flexmock(module.flags).should_receive('make_flags').and_return(('--paper',))
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', '--paper', 'repo'))
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=True, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_calls_borg_with_path_argument():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').with_args('source').and_return(True)
insert_execute_command_mock(('borg', 'key', 'import', 'repo', 'source'), input_file=None)
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path='source'),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_with_non_existent_path_raises():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').and_return(False)
flexmock(module).should_receive('execute_command').never()
with pytest.raises(ValueError):
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path='source'),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_with_stdin_path_calls_borg_without_path_argument():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', 'repo'))
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path='-'),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_with_dry_run_skips_borg_call():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
flexmock(module).should_receive('execute_command').never()
module.import_key(
repository_path='repo',
config={},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=True, log_json=False),
)
def test_import_key_calls_borg_with_working_directory():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').never()
insert_execute_command_mock(('borg', 'key', 'import', 'repo'), working_directory='/working/dir')
module.import_key(
repository_path='repo',
config={'working_directory': '/working/dir'},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path=None),
global_arguments=flexmock(dry_run=False, log_json=False),
)
def test_import_key_calls_borg_with_path_argument_and_working_directory():
flexmock(module.flags).should_receive('make_flags').and_return(())
flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
flexmock(module.os.path).should_receive('exists').with_args('/working/dir/source').and_return(
True
).once()
insert_execute_command_mock(
('borg', 'key', 'import', 'repo', 'source'),
input_file=None,
working_directory='/working/dir',
)
module.import_key(
repository_path='repo',
config={'working_directory': '/working/dir'},
local_borg_version='1.2.3',
import_arguments=flexmock(paper=False, path='source'),
global_arguments=flexmock(dry_run=False, log_json=False),
)
tests/unit/commands/test_borgmatic.py
+20
View File
@@ -1390,6 +1390,26 @@ def test_run_actions_runs_export_key():
)
def test_run_actions_runs_import_key():
flexmock(module).should_receive('add_custom_log_levels')
flexmock(module).should_receive('get_skip_actions').and_return([])
flexmock(module.command).should_receive('Before_after_hooks').and_return(flexmock())
flexmock(borgmatic.actions.import_key).should_receive('run_import_key').once()
tuple(
module.run_actions(
arguments={'global': flexmock(dry_run=False, log_file='foo'), 'import': flexmock()},
config_filename=flexmock(),
config={'repositories': []},
config_paths=[],
local_path=flexmock(),
remote_path=flexmock(),
local_borg_version=flexmock(),
repository={'path': 'repo'},
)
)
def test_run_actions_runs_change_passphrase():
flexmock(module).should_receive('add_custom_log_levels')
flexmock(module).should_receive('get_skip_actions').and_return([])