Initial commit
This commit is contained in:
Alexander Wainwright
300
test/test_backup.py
Normal file
300
test/test_backup.py
Normal file
@@ -0,0 +1,300 @@
|
||||
import argparse
|
||||
import pytest
|
||||
import subprocess
|
||||
|
||||
from unittest import mock
|
||||
|
||||
from locutus.backup import (
|
||||
run_backup,
|
||||
run_borg_create,
|
||||
run_borg_prune,
|
||||
run_borg_compact,
|
||||
)
|
||||
|
||||
|
||||
def make_args(config, profile, dry_run=False):
|
||||
args = argparse.Namespace()
|
||||
args.config = config
|
||||
args.profile = profile
|
||||
args.dry_run = dry_run
|
||||
return args
|
||||
|
||||
|
||||
class DummyCfg:
|
||||
def __init__(self, includes=[], excludes=[], prune=True, repo='', env=''):
|
||||
self.includes = includes
|
||||
self.excludes = excludes
|
||||
self.prune = prune
|
||||
self.env = env
|
||||
self._repo = repo
|
||||
|
||||
def get_repo(self):
|
||||
return self._repo
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def dummy_cfg(tmp_path):
|
||||
return DummyCfg(
|
||||
includes=['/home/alex/docs', '/home/alex/pics'],
|
||||
excludes=['*.cache', '/tmp'],
|
||||
repo='user@host:/repo',
|
||||
env={'BORG_PASSPHRASE': 'hunter2'},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def dummy_compact_cfg():
|
||||
return DummyCfg(
|
||||
repo='user@host:/repo',
|
||||
env={'BORG_PASSPHRASE': 'hunter2'},
|
||||
)
|
||||
|
||||
|
||||
@mock.patch('locutus.backup.LocutusConfig')
|
||||
@mock.patch('locutus.backup.run_borg_create')
|
||||
@mock.patch('locutus.backup.run_borg_prune')
|
||||
@mock.patch('locutus.backup.run_borg_compact')
|
||||
def test_backup_success(mock_compact, mock_prune, mock_create, mock_config):
|
||||
mock_create.return_value = True
|
||||
mock_prune.return_value = True
|
||||
mock_compact.return_value = True
|
||||
# Simulate cfg.compact = True
|
||||
instance = mock_config.return_value
|
||||
instance.compact = True
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
assert run_backup(args) == 0
|
||||
assert mock_create.called
|
||||
assert mock_prune.called
|
||||
assert mock_compact.called
|
||||
|
||||
|
||||
@mock.patch('locutus.backup.LocutusConfig')
|
||||
@mock.patch('locutus.backup.run_borg_create')
|
||||
@mock.patch('locutus.backup.run_borg_prune')
|
||||
@mock.patch('locutus.backup.run_borg_compact')
|
||||
def test_backup_success_no_compact(
|
||||
mock_compact, mock_prune, mock_create, mock_config
|
||||
):
|
||||
mock_create.return_value = True
|
||||
mock_prune.return_value = True
|
||||
# Compact should not be called if cfg.compact is False
|
||||
instance = mock_config.return_value
|
||||
instance.compact = False
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
assert run_backup(args) == 0
|
||||
assert mock_create.called
|
||||
assert mock_prune.called
|
||||
assert not mock_compact.called
|
||||
|
||||
|
||||
@mock.patch('locutus.backup.LocutusConfig')
|
||||
@mock.patch('locutus.backup.run_borg_create')
|
||||
@mock.patch('locutus.backup.run_borg_prune')
|
||||
@mock.patch('locutus.backup.run_borg_compact')
|
||||
def test_backup_fails_on_create(
|
||||
mock_compact, mock_prune, mock_create, mock_config
|
||||
):
|
||||
mock_create.return_value = False
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
assert run_backup(args) == 1
|
||||
|
||||
|
||||
@mock.patch('locutus.backup.LocutusConfig')
|
||||
@mock.patch('locutus.backup.run_borg_create')
|
||||
@mock.patch('locutus.backup.run_borg_prune')
|
||||
@mock.patch('locutus.backup.run_borg_compact')
|
||||
def test_backup_fails_on_prune(
|
||||
mock_compact, mock_prune, mock_create, mock_config
|
||||
):
|
||||
mock_create.return_value = True
|
||||
mock_prune.return_value = False
|
||||
instance = mock_config.return_value
|
||||
instance.compact = True
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
assert run_backup(args) == 1
|
||||
|
||||
|
||||
@mock.patch('locutus.backup.LocutusConfig')
|
||||
@mock.patch('locutus.backup.run_borg_create')
|
||||
@mock.patch('locutus.backup.run_borg_prune')
|
||||
@mock.patch('locutus.backup.run_borg_compact')
|
||||
def test_backup_fails_on_compact(
|
||||
mock_compact, mock_prune, mock_create, mock_config
|
||||
):
|
||||
mock_create.return_value = True
|
||||
mock_prune.return_value = True
|
||||
mock_compact.return_value = False
|
||||
instance = mock_config.return_value
|
||||
instance.compact = True
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
assert run_backup(args) == 1
|
||||
|
||||
|
||||
@mock.patch(
|
||||
'locutus.backup.LocutusConfig', side_effect=Exception('config error')
|
||||
)
|
||||
def test_backup_config_exception(mock_config):
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
assert run_backup(args) == 1
|
||||
|
||||
|
||||
@mock.patch('subprocess.run')
|
||||
def test_borg_create_success(mock_run, dummy_cfg):
|
||||
mock_run.return_value.returncode = 0
|
||||
assert run_borg_create(dummy_cfg, dry_run=False) is True
|
||||
# Check command construction
|
||||
args = mock_run.call_args[0][0]
|
||||
assert args[0:2] == ['borg', 'create']
|
||||
assert '--dry-run' not in args
|
||||
assert '--exclude' in args
|
||||
assert any('/home/alex/docs' in a or '/home/alex/pics' in a for a in args)
|
||||
# Check env contains BORG_PASSPHRASE
|
||||
assert mock_run.call_args[1]['env']['BORG_PASSPHRASE'] == 'hunter2'
|
||||
|
||||
|
||||
@mock.patch('subprocess.run')
|
||||
def test_borg_create_success_dry_run(mock_run, dummy_cfg):
|
||||
mock_run.return_value.returncode = 0
|
||||
assert run_borg_create(dummy_cfg, dry_run=True) is True
|
||||
args = mock_run.call_args[0][0]
|
||||
assert '--dry-run' in args
|
||||
|
||||
|
||||
@mock.patch('subprocess.run')
|
||||
def test_borg_create_no_repo(mock_run, dummy_cfg):
|
||||
dummy_cfg._repo = None
|
||||
assert run_borg_create(dummy_cfg, dry_run=False) is False
|
||||
assert not mock_run.called
|
||||
|
||||
|
||||
@mock.patch(
|
||||
'subprocess.run',
|
||||
side_effect=subprocess.CalledProcessError(1, ['borg', 'create']),
|
||||
)
|
||||
def test_borg_create_calledprocesserror(mock_run, dummy_cfg, capsys):
|
||||
assert run_borg_create(dummy_cfg, dry_run=False) is False
|
||||
out = capsys.readouterr().err
|
||||
assert 'borg create failed' in out
|
||||
|
||||
|
||||
@mock.patch(
|
||||
'subprocess.run', side_effect=mock.Mock(side_effect=OSError('fail'))
|
||||
)
|
||||
def test_borg_create_raises_other_exception(mock_run, dummy_cfg, capsys):
|
||||
# Simulate unexpected subprocess failure (should still be caught)
|
||||
with mock.patch('subprocess.CalledProcessError', OSError):
|
||||
assert run_borg_create(dummy_cfg, dry_run=False) is False
|
||||
out = capsys.readouterr().err
|
||||
assert 'borg create failed' in out
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def dummy_prune_cfg():
|
||||
return DummyCfg(
|
||||
repo='user@host:/repo',
|
||||
env={'BORG_PASSPHRASE': 'hunter2'},
|
||||
prune={'keep_last': 3, 'keep_daily': 7},
|
||||
)
|
||||
|
||||
|
||||
@mock.patch('subprocess.run')
|
||||
def test_prune_success(mock_run, dummy_prune_cfg):
|
||||
mock_run.return_value.returncode = 0
|
||||
assert run_borg_prune(dummy_prune_cfg, dry_run=False) is True
|
||||
args = mock_run.call_args[0][0]
|
||||
assert 'borg' in args
|
||||
assert 'prune' in args
|
||||
assert '--keep-last' in args
|
||||
assert '3' in args
|
||||
assert '--keep-daily' in args
|
||||
assert '7' in args
|
||||
assert '--dry-run' not in args
|
||||
|
||||
|
||||
@mock.patch('subprocess.run')
|
||||
def test_prune_success_dry_run(mock_run, dummy_prune_cfg):
|
||||
mock_run.return_value.returncode = 0
|
||||
assert run_borg_prune(dummy_prune_cfg, dry_run=True) is True
|
||||
args = mock_run.call_args[0][0]
|
||||
assert '--dry-run' in args
|
||||
|
||||
|
||||
def test_prune_no_repo():
|
||||
cfg = DummyCfg(repo=None, env={}, prune={})
|
||||
assert run_borg_prune(cfg, dry_run=False) is False
|
||||
|
||||
|
||||
@mock.patch('subprocess.run', side_effect=Exception('fail'))
|
||||
def test_prune_subprocess_exception(mock_run, dummy_prune_cfg, capsys):
|
||||
assert run_borg_prune(dummy_prune_cfg, dry_run=False) is False
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg prune failed' in err
|
||||
|
||||
|
||||
@mock.patch(
|
||||
'subprocess.run', side_effect=mock.Mock(side_effect=OSError('fail'))
|
||||
)
|
||||
def test_prune_raises_other_exception(mock_run, dummy_prune_cfg, capsys):
|
||||
with mock.patch('subprocess.CalledProcessError', OSError):
|
||||
assert run_borg_prune(dummy_prune_cfg, dry_run=False) is False
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg prune failed' in err
|
||||
|
||||
|
||||
@mock.patch(
|
||||
'subprocess.run',
|
||||
side_effect=mock.Mock(
|
||||
side_effect=pytest.importorskip('subprocess').CalledProcessError(
|
||||
1, ['borg', 'prune']
|
||||
)
|
||||
),
|
||||
)
|
||||
def test_prune_calledprocesserror(mock_run, dummy_prune_cfg, capsys):
|
||||
assert run_borg_prune(dummy_prune_cfg, dry_run=False) is False
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg prune failed' in err
|
||||
|
||||
|
||||
@mock.patch('subprocess.run')
|
||||
def test_compact_success(mock_run, dummy_compact_cfg):
|
||||
mock_run.return_value.returncode = 0
|
||||
assert run_borg_compact(dummy_compact_cfg, dry_run=False) is True
|
||||
args = mock_run.call_args[0][0]
|
||||
assert args[:2] == ['borg', 'compact']
|
||||
assert '--verbose' in args
|
||||
|
||||
|
||||
@mock.patch('subprocess.run')
|
||||
def test_compact_no_repo(mock_run):
|
||||
cfg = DummyCfg(repo=None, env={})
|
||||
assert run_borg_compact(cfg, dry_run=False) is False
|
||||
assert not mock_run.called
|
||||
|
||||
|
||||
def test_compact_dry_run(dummy_compact_cfg, capsys):
|
||||
assert run_borg_compact(dummy_compact_cfg, dry_run=True) is True
|
||||
out = capsys.readouterr().out
|
||||
assert '(dry-run)' in out
|
||||
|
||||
|
||||
@mock.patch('subprocess.run', side_effect=Exception('fail'))
|
||||
def test_compact_subprocess_exception(mock_run, dummy_compact_cfg, capsys):
|
||||
assert run_borg_compact(dummy_compact_cfg, dry_run=False) is False
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg compact failed' in err
|
||||
|
||||
|
||||
@mock.patch(
|
||||
'subprocess.run',
|
||||
side_effect=mock.Mock(
|
||||
side_effect=pytest.importorskip('subprocess').CalledProcessError(
|
||||
1, ['borg', 'compact']
|
||||
)
|
||||
),
|
||||
)
|
||||
def test_compact_calledprocesserror(mock_run, dummy_compact_cfg, capsys):
|
||||
assert run_borg_compact(dummy_compact_cfg, dry_run=False) is False
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg compact failed' in err
|
||||
101
test/test_config.py
Normal file
101
test/test_config.py
Normal file
@@ -0,0 +1,101 @@
|
||||
import pytest
|
||||
|
||||
from locutus import LocutusConfig
|
||||
|
||||
|
||||
def make_toml(path: str, text: str) -> None:
|
||||
with open(path, 'wb') as f:
|
||||
f.write(text.encode())
|
||||
|
||||
|
||||
def make_rc(path: str, text: str) -> None:
|
||||
with open(path, 'w') as f:
|
||||
f.write(text)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_config_files(tmp_path):
|
||||
# Make temp locutus.toml
|
||||
toml_content = """
|
||||
[includes]
|
||||
paths = ["/test/inc1", "/test/inc2"]
|
||||
|
||||
[excludes]
|
||||
paths = ["*.cache"]
|
||||
|
||||
[prune]
|
||||
keep_last = 2
|
||||
keep_daily = 3
|
||||
|
||||
[compact]
|
||||
enabled = true
|
||||
"""
|
||||
toml_path = tmp_path / 'locutus.toml'
|
||||
make_toml(str(toml_path), toml_content)
|
||||
|
||||
# Make temp locutus.rc
|
||||
rc_content = (
|
||||
'export BORG_REPO="/tmp/repo"\nexport BORG_PASSPHRASE="hunter2"\n'
|
||||
)
|
||||
rc_path = tmp_path / 'locutus.rc'
|
||||
make_rc(str(rc_path), rc_content)
|
||||
|
||||
return str(toml_path), str(rc_path)
|
||||
|
||||
|
||||
def test_config_loads_correctly(temp_config_files):
|
||||
toml_path, rc_path = temp_config_files
|
||||
cfg = LocutusConfig(toml_path, rc_path)
|
||||
|
||||
assert cfg.toml_path == toml_path
|
||||
assert cfg.rc_path == rc_path
|
||||
|
||||
assert cfg.includes == ['/test/inc1', '/test/inc2']
|
||||
assert cfg.excludes == ['*.cache']
|
||||
assert cfg.prune['keep_last'] == 2
|
||||
assert cfg.prune['keep_daily'] == 3
|
||||
assert cfg.compact is True
|
||||
|
||||
assert cfg.get_repo() == '/tmp/repo'
|
||||
assert cfg.get_passphrase() == 'hunter2'
|
||||
|
||||
|
||||
def test_missing_rc(tmp_path):
|
||||
# Valid toml, missing rc
|
||||
toml_content = """
|
||||
[includes]
|
||||
paths = ["/test/inc1"]
|
||||
"""
|
||||
toml_path = tmp_path / 'locutus.toml'
|
||||
make_toml(str(toml_path), toml_content)
|
||||
rc_path = tmp_path / 'missing.rc'
|
||||
|
||||
cfg = LocutusConfig(str(toml_path), str(rc_path))
|
||||
|
||||
assert cfg.includes == ['/test/inc1']
|
||||
assert cfg.get_repo() is None
|
||||
assert cfg.get_passphrase() is None
|
||||
|
||||
|
||||
def test_missing_toml(tmp_path):
|
||||
# Missing toml
|
||||
toml_path = tmp_path / 'missing.toml'
|
||||
rc_path = tmp_path / 'locutus.rc'
|
||||
make_rc(str(rc_path), 'export BORG_REPO="/tmp/repo"\n')
|
||||
with pytest.raises(FileNotFoundError):
|
||||
LocutusConfig(str(toml_path), str(rc_path))
|
||||
|
||||
|
||||
def test_partial_rc(tmp_path):
|
||||
# Valid toml, partial rc (no passphrase)
|
||||
toml_content = """
|
||||
[includes]
|
||||
paths = ["/home/test"]
|
||||
"""
|
||||
toml_path = tmp_path / 'locutus.toml'
|
||||
make_toml(str(toml_path), toml_content)
|
||||
rc_path = tmp_path / 'locutus.rc'
|
||||
make_rc(str(rc_path), 'export BORG_REPO="/somewhere/repo"\n')
|
||||
cfg = LocutusConfig(str(toml_path), str(rc_path))
|
||||
assert cfg.get_repo() == '/somewhere/repo'
|
||||
assert cfg.get_passphrase() is None
|
||||
100
test/test_info.py
Normal file
100
test/test_info.py
Normal file
@@ -0,0 +1,100 @@
|
||||
import argparse
|
||||
from unittest import mock
|
||||
import subprocess
|
||||
import pytest
|
||||
|
||||
from locutus.info import run_info
|
||||
|
||||
|
||||
def make_args(config, profile):
|
||||
args = argparse.Namespace()
|
||||
args.config = config
|
||||
args.profile = profile
|
||||
return args
|
||||
|
||||
|
||||
@mock.patch('locutus.info.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_info_basic_output(mock_run, mock_config, capsys):
|
||||
mock_config.return_value.toml_path = '/tmp/test.toml'
|
||||
mock_config.return_value.rc_path = '/tmp/test.rc'
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.get_passphrase.return_value = 'hunter2'
|
||||
mock_config.return_value.includes = ['/etc', '/home']
|
||||
mock_config.return_value.excludes = ['*.cache']
|
||||
mock_config.return_value.prune = {'keep_last': 3, 'keep_daily': 1}
|
||||
|
||||
args = make_args('/tmp/test.toml', '/tmp/test.rc')
|
||||
mock_run.return_value.returncode = 0
|
||||
|
||||
run_info(args)
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert 'Config: /tmp/test.toml' in out
|
||||
assert 'Profile: /tmp/test.rc' in out
|
||||
assert 'Repo: /repo' in out
|
||||
assert 'Passphrase: set' in out
|
||||
assert 'Includes:' in out and '/etc' in out and '/home' in out
|
||||
assert 'Excludes:' in out and '*.cache' in out
|
||||
assert 'Prune: keep_last=3, keep_daily=1' in out
|
||||
assert '[borg info output below]' in out
|
||||
mock_run.assert_called_with(
|
||||
['borg', 'info', '/repo'], env=mock.ANY, check=True
|
||||
)
|
||||
|
||||
|
||||
@mock.patch('locutus.info.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_info_no_repo(mock_run, mock_config, capsys):
|
||||
mock_config.return_value.toml_path = '/tmp/test.toml'
|
||||
mock_config.return_value.rc_path = '/tmp/test.rc'
|
||||
mock_config.return_value.get_repo.return_value = None
|
||||
mock_config.return_value.get_passphrase.return_value = None
|
||||
mock_config.return_value.includes = []
|
||||
mock_config.return_value.excludes = []
|
||||
mock_config.return_value.prune = {}
|
||||
|
||||
args = make_args('/tmp/test.toml', '/tmp/test.rc')
|
||||
run_info(args)
|
||||
out = capsys.readouterr().out
|
||||
assert 'Repo: (not set)' in out
|
||||
assert 'Passphrase: (not set)' in out
|
||||
assert '[borg info output below]' in out
|
||||
mock_run.assert_not_called()
|
||||
|
||||
|
||||
@mock.patch('locutus.info.LocutusConfig')
|
||||
@mock.patch(
|
||||
'subprocess.run',
|
||||
side_effect=subprocess.CalledProcessError(1, ['borg', 'info']),
|
||||
)
|
||||
def test_info_borg_info_fails(mock_run, mock_config, capsys):
|
||||
mock_config.return_value.toml_path = '/tmp/test.toml'
|
||||
mock_config.return_value.rc_path = '/tmp/test.rc'
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.get_passphrase.return_value = 'hunter2'
|
||||
mock_config.return_value.includes = []
|
||||
mock_config.return_value.excludes = []
|
||||
mock_config.return_value.prune = {}
|
||||
|
||||
args = make_args('/tmp/test.toml', '/tmp/test.rc')
|
||||
run_info(args)
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg info failed:' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.info.LocutusConfig')
|
||||
@mock.patch('subprocess.run', side_effect=Exception('fail'))
|
||||
def test_info_borg_info_generic_exception(mock_run, mock_config, capsys):
|
||||
mock_config.return_value.toml_path = '/tmp/test.toml'
|
||||
mock_config.return_value.rc_path = '/tmp/test.rc'
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.get_passphrase.return_value = 'hunter2'
|
||||
mock_config.return_value.includes = []
|
||||
mock_config.return_value.excludes = []
|
||||
mock_config.return_value.prune = {}
|
||||
|
||||
args = make_args('/tmp/test.toml', '/tmp/test.rc')
|
||||
run_info(args)
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg info failed: fail' in err
|
||||
169
test/test_init.py
Normal file
169
test/test_init.py
Normal file
@@ -0,0 +1,169 @@
|
||||
import argparse
|
||||
import subprocess
|
||||
from unittest import mock
|
||||
|
||||
from locutus.init import run_init
|
||||
|
||||
|
||||
def make_args(profile_path):
|
||||
args = argparse.Namespace()
|
||||
args.profile = profile_path
|
||||
return args
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
@mock.patch('getpass.getpass')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_run_init_success_no_passphrase(
|
||||
mock_subproc, mock_getpass, mock_input, tmp_path
|
||||
):
|
||||
# Simulate user enters repo path and empty passphrase
|
||||
mock_input.return_value = '/test/repo'
|
||||
mock_getpass.return_value = ''
|
||||
mock_subproc.return_value = mock.Mock(returncode=0)
|
||||
|
||||
rc_path = tmp_path / 'test.rc'
|
||||
args = make_args(str(rc_path))
|
||||
result = run_init(args)
|
||||
|
||||
assert result == 0
|
||||
with open(rc_path) as f:
|
||||
content = f.read()
|
||||
assert 'export BORG_REPO="/test/repo"' in content
|
||||
assert 'BORG_PASSPHRASE' not in content
|
||||
|
||||
mock_subproc.assert_called_once()
|
||||
call_args = mock_subproc.call_args[0][0]
|
||||
assert call_args == ['borg', 'init', '--encryption=none', '/test/repo']
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
@mock.patch('getpass.getpass')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_run_init_success_with_passphrase(
|
||||
mock_subproc, mock_getpass, mock_input, tmp_path
|
||||
):
|
||||
mock_input.return_value = '/secure/repo'
|
||||
# First call is passphrase, second is confirm, both match
|
||||
mock_getpass.side_effect = ['hunter2', 'hunter2']
|
||||
mock_subproc.return_value = mock.Mock(returncode=0)
|
||||
|
||||
rc_path = tmp_path / 'secure.rc'
|
||||
args = make_args(str(rc_path))
|
||||
result = run_init(args)
|
||||
|
||||
assert result == 0
|
||||
with open(rc_path) as f:
|
||||
content = f.read()
|
||||
assert 'export BORG_REPO="/secure/repo"' in content
|
||||
assert 'export BORG_PASSPHRASE="hunter2"' in content
|
||||
|
||||
mock_subproc.assert_called_once()
|
||||
call_args = mock_subproc.call_args[0][0]
|
||||
assert call_args == ['borg', 'init', '--encryption=repokey', '/secure/repo']
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
def test_run_init_empty_repo_aborts(mock_input, tmp_path):
|
||||
mock_input.return_value = ''
|
||||
args = make_args(str(tmp_path / 'fail.rc'))
|
||||
result = run_init(args)
|
||||
assert result == 1
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
@mock.patch('getpass.getpass')
|
||||
def test_run_init_passphrase_mismatch(mock_getpass, mock_input, tmp_path):
|
||||
mock_input.return_value = '/fail/repo'
|
||||
mock_getpass.side_effect = [
|
||||
'pass1',
|
||||
'pass2',
|
||||
'',
|
||||
] # Third call: user gives up
|
||||
args = make_args(str(tmp_path / 'fail2.rc'))
|
||||
with mock.patch('subprocess.run'):
|
||||
result = run_init(args)
|
||||
assert result == 0 # Exits with no passphrase (encryption=none)
|
||||
with open(args.profile) as f:
|
||||
content = f.read()
|
||||
assert 'BORG_PASSPHRASE' not in content
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
@mock.patch('getpass.getpass')
|
||||
@mock.patch('subprocess.run', side_effect=Exception('init failed'))
|
||||
def test_run_init_subprocess_failure(
|
||||
mock_subproc, mock_getpass, mock_input, tmp_path
|
||||
):
|
||||
mock_input.return_value = '/failinit/repo'
|
||||
mock_getpass.return_value = ''
|
||||
args = make_args(str(tmp_path / 'failinit.rc'))
|
||||
result = run_init(args)
|
||||
assert result == 1
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
@mock.patch('getpass.getpass')
|
||||
@mock.patch('os.makedirs', side_effect=Exception('cannot create dir'))
|
||||
def test_run_init_makedirs_exception(
|
||||
mock_makedirs, mock_getpass, mock_input, tmp_path
|
||||
):
|
||||
mock_input.return_value = '/faildir/repo'
|
||||
mock_getpass.return_value = ''
|
||||
args = make_args(str(tmp_path / 'faildir.rc'))
|
||||
with mock.patch('subprocess.run'):
|
||||
result = run_init(args)
|
||||
assert result == 1
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
@mock.patch('getpass.getpass')
|
||||
@mock.patch('os.makedirs')
|
||||
@mock.patch('builtins.open', side_effect=Exception('cannot open file'))
|
||||
def test_run_init_open_exception(
|
||||
mock_open, mock_makedirs, mock_getpass, mock_input, tmp_path
|
||||
):
|
||||
mock_input.return_value = '/failfile/repo'
|
||||
mock_getpass.return_value = ''
|
||||
args = make_args(str(tmp_path / 'failfile.rc'))
|
||||
with mock.patch('subprocess.run'):
|
||||
result = run_init(args)
|
||||
assert result == 1
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
@mock.patch('getpass.getpass')
|
||||
@mock.patch('os.makedirs')
|
||||
@mock.patch('builtins.open')
|
||||
@mock.patch('subprocess.run', side_effect=Exception('unexpected'))
|
||||
def test_run_init_unexpected_exception(
|
||||
mock_subproc, mock_open, mock_makedirs, mock_getpass, mock_input, tmp_path
|
||||
):
|
||||
mock_input.return_value = '/unexpected/repo'
|
||||
mock_getpass.return_value = ''
|
||||
args = make_args(str(tmp_path / 'unexpected.rc'))
|
||||
result = run_init(args)
|
||||
assert result == 1
|
||||
|
||||
|
||||
@mock.patch('builtins.input', side_effect=KeyboardInterrupt)
|
||||
def test_run_init_keyboard_interrupt(mock_input, tmp_path):
|
||||
args = make_args(str(tmp_path / 'kbint.rc'))
|
||||
result = run_init(args)
|
||||
assert result == 1
|
||||
|
||||
|
||||
@mock.patch('builtins.input')
|
||||
@mock.patch('getpass.getpass')
|
||||
@mock.patch(
|
||||
'subprocess.run',
|
||||
side_effect=subprocess.CalledProcessError(1, ['borg', 'init']),
|
||||
)
|
||||
def test_run_init_calledprocesserror(
|
||||
mock_subproc, mock_getpass, mock_input, tmp_path
|
||||
):
|
||||
mock_input.return_value = '/failsubproc/repo'
|
||||
mock_getpass.return_value = ''
|
||||
args = make_args(str(tmp_path / 'failsubproc.rc'))
|
||||
result = run_init(args)
|
||||
assert result == 1
|
||||
179
test/test_list.py
Normal file
179
test/test_list.py
Normal file
@@ -0,0 +1,179 @@
|
||||
import argparse
|
||||
from unittest import mock
|
||||
import pytest
|
||||
|
||||
from locutus.list import run_list
|
||||
|
||||
|
||||
def make_args(config, profile, target=None):
|
||||
args = argparse.Namespace()
|
||||
args.config = config
|
||||
args.profile = profile
|
||||
if target is not None:
|
||||
args.target = target
|
||||
return args
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_list_prints_chrono_and_numbers_reverse(
|
||||
mock_subproc, mock_config, capsys
|
||||
):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
|
||||
# Oldest to newest
|
||||
archives = ['archA', 'archB', 'archC']
|
||||
mock_subproc.return_value.stdout = '\n'.join(archives) + '\n'
|
||||
mock_subproc.return_value.returncode = 0
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
rc = run_list(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out.strip().splitlines()
|
||||
# Should print:
|
||||
# 2: archA
|
||||
# 1: archB
|
||||
# 0: archC
|
||||
assert out[0].strip() == '2: archA'
|
||||
assert out[1].strip() == '1: archB'
|
||||
assert out[2].strip() == '0: archC'
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_list_by_index_most_recent(mock_subproc, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
|
||||
# Oldest to newest
|
||||
archives = ['oldest', 'middle', 'newest']
|
||||
# First subprocess.run: get archives (reversed in code)
|
||||
mock_subproc.side_effect = [
|
||||
mock.Mock(stdout='\n'.join(archives) + '\n', returncode=0),
|
||||
mock.Mock(stdout='file1\nfile2\n', returncode=0),
|
||||
]
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc', target='0')
|
||||
rc = run_list(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
# Should show file list for "newest"
|
||||
assert 'file1' in out and 'file2' in out
|
||||
call_args = mock_subproc.call_args_list[1][0][0]
|
||||
# Should call borg list ...::newest
|
||||
assert call_args == ['borg', 'list', '/repo::newest']
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_list_by_index_out_of_range(mock_subproc, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
|
||||
archives = ['archA', 'archB']
|
||||
mock_subproc.return_value.stdout = '\n'.join(archives) + '\n'
|
||||
mock_subproc.return_value.returncode = 0
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc', target='3') # out of range
|
||||
rc = run_list(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'Archive index out of range' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_list_by_name(mock_subproc, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
# Only need one subprocess.run (direct by name)
|
||||
mock_subproc.return_value.stdout = 'filex\nfiley\n'
|
||||
mock_subproc.return_value.returncode = 0
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc', target='archive-explicit')
|
||||
rc = run_list(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert 'filex' in out and 'filey' in out
|
||||
call_args = mock_subproc.call_args[0][0]
|
||||
assert call_args == ['borg', 'list', '/repo::archive-explicit']
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
def test_list_no_repo(mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = None
|
||||
mock_config.return_value.env = {}
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
rc = run_list(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'No BORG_REPO configured' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch('subprocess.run', side_effect=Exception('fail'))
|
||||
def test_list_unexpected_exception(mock_subproc, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
rc = run_list(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg list failed' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch(
|
||||
'subprocess.run',
|
||||
side_effect=pytest.importorskip('subprocess').CalledProcessError(
|
||||
1, ['borg', 'list']
|
||||
),
|
||||
)
|
||||
def test_list_calledprocesserror(mock_subproc, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
rc = run_list(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg list failed' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_list_no_archives(mock_subproc, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_subproc.return_value.stdout = ''
|
||||
mock_subproc.return_value.returncode
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_list_no_archives_main(mock_subproc, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_subproc.return_value.stdout = '' # No archives
|
||||
mock_subproc.return_value.returncode = 0
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc')
|
||||
rc = run_list(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert 'No archives found' in out
|
||||
|
||||
|
||||
@mock.patch('locutus.list.LocutusConfig')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_list_by_index_no_archives(mock_subproc, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_subproc.return_value.stdout = '' # No archives
|
||||
mock_subproc.return_value.returncode = 0
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc', target='0')
|
||||
rc = run_list(args)
|
||||
assert rc == 1
|
||||
out = capsys.readouterr().out
|
||||
assert 'No archives found' in out
|
||||
208
test/test_mount.py
Normal file
208
test/test_mount.py
Normal file
@@ -0,0 +1,208 @@
|
||||
import argparse
|
||||
from unittest import mock
|
||||
import pytest
|
||||
import subprocess
|
||||
|
||||
from locutus.mount import run_mount
|
||||
|
||||
|
||||
def make_args(config, profile, mountpoint, archive=None):
|
||||
args = argparse.Namespace()
|
||||
args.config = config
|
||||
args.profile = profile
|
||||
args.mountpoint = mountpoint
|
||||
if archive is not None:
|
||||
args.archive = archive
|
||||
return args
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_all_archives_success(mock_run, mock_isdir, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_isdir.return_value = True
|
||||
mock_run.return_value.returncode = 0
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc', '/mnt/test')
|
||||
rc = run_mount(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert 'Archives mounted on /mnt/test' in out
|
||||
cmd_args = mock_run.call_args[0][0]
|
||||
assert cmd_args == ['borg', 'mount', '/repo', '/mnt/test']
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_by_name_success(mock_run, mock_isdir, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_isdir.return_value = True
|
||||
mock_run.return_value.returncode = 0
|
||||
|
||||
args = make_args(
|
||||
'dummy.toml', 'dummy.rc', '/mnt/test', archive='archive-xyz'
|
||||
)
|
||||
rc = run_mount(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert "Archive 'archive-xyz' mounted on /mnt/test" in out
|
||||
cmd_args = mock_run.call_args[0][0]
|
||||
assert cmd_args == ['borg', 'mount', '/repo::archive-xyz', '/mnt/test']
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_by_index_success(mock_run, mock_isdir, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_isdir.return_value = True
|
||||
mock_run.return_value.returncode = 0
|
||||
archives = ['archA', 'archB', 'archC'] # Oldest to newest
|
||||
mock_run.side_effect = [
|
||||
# First run: borg list
|
||||
mock.Mock(stdout='\n'.join(archives) + '\n', returncode=0),
|
||||
# Second run: borg mount
|
||||
mock.Mock(returncode=0),
|
||||
]
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc', '/mnt/test', archive='1')
|
||||
rc = run_mount(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
# Index 1 in reversed list is archB (so index 0=newest=archC, 1=archB, 2=archA)
|
||||
assert "Archive 'archB' mounted on /mnt/test" in out
|
||||
cmd_args = mock_run.call_args_list[1][0][0]
|
||||
assert cmd_args == ['borg', 'mount', '/repo::archB', '/mnt/test']
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
def test_mount_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', str(tmp_path), archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'No BORG_REPO configured' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=False)
|
||||
def test_mount_invalid_mountpoint(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', '/not/a/dir', archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'does not exist or is not a directory' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run', side_effect=Exception('fail'))
|
||||
def test_mount_unexpected_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', str(tmp_path), archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg mount failed' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_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.side_effect = [
|
||||
mock.Mock(stdout='\n'.join(archives) + '\n', returncode=0)
|
||||
]
|
||||
args = make_args(
|
||||
'dummy.toml', 'dummy.rc', str(tmp_path), archive='5'
|
||||
) # Out of range
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'Archive index out of range' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_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 = '' # No archives
|
||||
mock_run.return_value.returncode = 0
|
||||
args = make_args('dummy.toml', 'dummy.rc', str(tmp_path), archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'No archives found' in err
|
||||
|
||||
|
||||
def test_mount_usage_message(capsys):
|
||||
args = argparse.Namespace()
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'Usage:' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_calledprocesserror_on_mount(
|
||||
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', 'c']
|
||||
mock_run.side_effect = [
|
||||
mock.Mock(
|
||||
stdout='\n'.join(archives) + '\n', returncode=0
|
||||
), # listing works
|
||||
subprocess.CalledProcessError(1, ['borg', 'mount']), # mount fails
|
||||
]
|
||||
args = make_args('dummy.toml', 'dummy.rc', str(tmp_path), archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg mount failed:' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_exception_on_mount(
|
||||
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', 'c']
|
||||
mock_run.side_effect = [
|
||||
mock.Mock(
|
||||
stdout='\n'.join(archives) + '\n', returncode=0
|
||||
), # listing works
|
||||
Exception(), # mount fails
|
||||
]
|
||||
args = make_args('dummy.toml', 'dummy.rc', str(tmp_path), archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg mount failed (unexpected)' in err
|
||||
178
test/test_restore.py
Normal file
178
test/test_restore.py
Normal file
@@ -0,0 +1,178 @@
|
||||
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
|
||||
97
test/test_umount.py
Normal file
97
test/test_umount.py
Normal file
@@ -0,0 +1,97 @@
|
||||
import argparse
|
||||
import pytest
|
||||
import subprocess
|
||||
|
||||
from unittest import mock
|
||||
|
||||
from locutus.umount import run_umount
|
||||
|
||||
|
||||
def make_args(mountpoint):
|
||||
args = argparse.Namespace()
|
||||
args.mountpoint = mountpoint
|
||||
return args
|
||||
|
||||
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_umount_borg_success(mock_run, mock_isdir, capsys):
|
||||
mock_run.return_value.returncode = 0
|
||||
args = make_args('/mnt/test')
|
||||
rc = run_umount(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert 'Unmounted /mnt/test' in out
|
||||
cmd_args = mock_run.call_args[0][0]
|
||||
assert cmd_args == ['borg', 'umount', '/mnt/test']
|
||||
|
||||
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_umount_fusermount_success(mock_run, mock_isdir, capsys):
|
||||
# Simulate borg umount fails, fusermount -u succeeds
|
||||
def side_effect(cmd, **kwargs):
|
||||
if cmd[0] == 'borg':
|
||||
raise subprocess.CalledProcessError(1, cmd)
|
||||
return mock.Mock(returncode=0)
|
||||
|
||||
mock_run.side_effect = side_effect
|
||||
|
||||
args = make_args('/mnt/backup')
|
||||
rc = run_umount(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert 'Unmounted /mnt/backup' in out
|
||||
# Second call is fusermount -u
|
||||
assert mock_run.call_args_list[1][0][0] == [
|
||||
'fusermount',
|
||||
'-u',
|
||||
'/mnt/backup',
|
||||
]
|
||||
|
||||
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run', side_effect=Exception('fail'))
|
||||
def test_umount_all_fail(mock_run, mock_isdir, capsys):
|
||||
args = make_args('/mnt/doesnotunmount')
|
||||
rc = run_umount(args)
|
||||
assert rc == 2
|
||||
err = capsys.readouterr().err
|
||||
assert 'Failed to unmount' in err
|
||||
|
||||
|
||||
@mock.patch('os.path.isdir', return_value=False)
|
||||
def test_umount_bad_mountpoint(mock_isdir, capsys):
|
||||
args = make_args('/not/a/dir')
|
||||
rc = run_umount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'does not exist or is not a directory' in err
|
||||
|
||||
|
||||
def test_umount_usage_message(capsys):
|
||||
args = argparse.Namespace()
|
||||
rc = run_umount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'Usage:' in err
|
||||
|
||||
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_umount_fusermount_inner_exception(mock_run, mock_isdir, capsys):
|
||||
# borg umount fails (CalledProcessError), fusermount -u raises generic Exception
|
||||
def side_effect(cmd, **kwargs):
|
||||
if cmd[0] == 'borg':
|
||||
raise subprocess.CalledProcessError(1, cmd)
|
||||
if cmd[0] == 'fusermount':
|
||||
raise Exception('fusermount explosion!')
|
||||
return mock.Mock(returncode=0)
|
||||
|
||||
mock_run.side_effect = side_effect
|
||||
|
||||
args = make_args('/mnt/boom')
|
||||
rc = run_umount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'Failed to unmount /mnt/boom: fusermount explosion!' in err
|
||||
Reference in New Issue
Block a user