179 lines
5.5 KiB
Python
179 lines
5.5 KiB
Python
import argparse
|
|
from unittest import mock
|
|
import subprocess
|
|
import pytest
|
|
|
|
from locutus.restore import run_restore
|
|
|
|
|
|
def make_args(config, profile, archive, targetdir):
|
|
args = argparse.Namespace()
|
|
args.config = config
|
|
args.profile = profile
|
|
args.archive = archive
|
|
args.targetdir = targetdir
|
|
return args
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run')
|
|
def test_restore_by_index_success(
|
|
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
|
):
|
|
mock_config.return_value.get_repo.return_value = '/repo'
|
|
mock_config.return_value.env = {}
|
|
archives = ['archA', 'archB', 'archC']
|
|
mock_run.side_effect = [
|
|
mock.Mock(stdout='\n'.join(archives) + '\n', returncode=0), # borg list
|
|
mock.Mock(returncode=0), # borg extract
|
|
]
|
|
|
|
args = make_args('dummy.toml', 'dummy.rc', '1', str(tmp_path))
|
|
rc = run_restore(args)
|
|
assert rc == 0
|
|
out = capsys.readouterr().out
|
|
assert 'restored to' in out
|
|
assert 'archB' in out
|
|
extract_call = mock_run.call_args_list[1][0][0]
|
|
assert extract_call == ['borg', 'extract', '/repo::archB']
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run')
|
|
def test_restore_by_name_success(
|
|
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
|
):
|
|
mock_config.return_value.get_repo.return_value = '/repo'
|
|
mock_config.return_value.env = {}
|
|
mock_run.return_value.returncode = 0
|
|
|
|
args = make_args('dummy.toml', 'dummy.rc', 'named-archive', str(tmp_path))
|
|
rc = run_restore(args)
|
|
assert rc == 0
|
|
out = capsys.readouterr().out
|
|
assert 'restored to' in out
|
|
assert 'named-archive' in out
|
|
call_args = mock_run.call_args[0][0]
|
|
assert call_args == ['borg', 'extract', '/repo::named-archive']
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run')
|
|
def test_restore_by_index_out_of_range(
|
|
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
|
):
|
|
mock_config.return_value.get_repo.return_value = '/repo'
|
|
mock_config.return_value.env = {}
|
|
archives = ['a', 'b']
|
|
mock_run.return_value.stdout = '\n'.join(archives) + '\n'
|
|
mock_run.return_value.returncode = 0
|
|
|
|
args = make_args('dummy.toml', 'dummy.rc', '5', str(tmp_path))
|
|
rc = run_restore(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'Archive index out of range' in err
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run')
|
|
def test_restore_by_index_no_archives(
|
|
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
|
):
|
|
mock_config.return_value.get_repo.return_value = '/repo'
|
|
mock_config.return_value.env = {}
|
|
mock_run.return_value.stdout = ''
|
|
mock_run.return_value.returncode = 0
|
|
|
|
args = make_args('dummy.toml', 'dummy.rc', '0', str(tmp_path))
|
|
rc = run_restore(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'No archives found' in err
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
@mock.patch('os.path.isdir', return_value=False)
|
|
def test_restore_invalid_targetdir(mock_isdir, mock_config, capsys):
|
|
mock_config.return_value.get_repo.return_value = '/repo'
|
|
mock_config.return_value.env = {}
|
|
|
|
args = make_args('dummy.toml', 'dummy.rc', 'archive', '/notadir')
|
|
rc = run_restore(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'does not exist or is not a directory' in err
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
def test_restore_no_repo(mock_config, tmp_path, capsys):
|
|
mock_config.return_value.get_repo.return_value = None
|
|
mock_config.return_value.env = {}
|
|
args = make_args('dummy.toml', 'dummy.rc', '0', str(tmp_path))
|
|
rc = run_restore(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'No BORG_REPO configured' in err
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch(
|
|
'subprocess.run',
|
|
side_effect=subprocess.CalledProcessError(1, ['borg', 'extract']),
|
|
)
|
|
def test_restore_extract_calledprocesserror(
|
|
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
|
):
|
|
mock_config.return_value.get_repo.return_value = '/repo'
|
|
mock_config.return_value.env = {}
|
|
|
|
args = make_args('dummy.toml', 'dummy.rc', 'named-archive', str(tmp_path))
|
|
rc = run_restore(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'borg extract failed' in err
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run', side_effect=Exception('fail!'))
|
|
def test_restore_extract_generic_exception(
|
|
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
|
):
|
|
mock_config.return_value.get_repo.return_value = '/repo'
|
|
mock_config.return_value.env = {}
|
|
|
|
args = make_args('dummy.toml', 'dummy.rc', 'named-archive', str(tmp_path))
|
|
rc = run_restore(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'borg extract failed (unexpected): fail!' in err
|
|
|
|
|
|
def test_restore_usage_message(capsys):
|
|
args = argparse.Namespace()
|
|
rc = run_restore(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'Usage:' in err
|
|
|
|
|
|
@mock.patch('locutus.restore.LocutusConfig')
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run', side_effect=Exception('index explode'))
|
|
def test_restore_index_lookup_generic_exception(
|
|
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
|
):
|
|
mock_config.return_value.get_repo.return_value = '/repo'
|
|
mock_config.return_value.env = {}
|
|
args = make_args('dummy.toml', 'dummy.rc', '0', str(tmp_path))
|
|
rc = run_restore(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'borg restore failed during archive lookup: index explode' in err
|