301 lines
8.4 KiB
Python
301 lines
8.4 KiB
Python
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
|