Fuxino
9e6aadf7d3
A missing /usr/bin/rsync ended in a traceback, possibly after --remove-before-backup had removed old backups, and left SSHPASS in the environment.
388 lines
15 KiB
Python
388 lines
15 KiB
Python
"""Tests for local backups, with real rsync runs and with a fake Popen"""
|
|
|
|
import errno
|
|
import io
|
|
import os
|
|
from collections.abc import Callable, Iterator
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from shutil import which
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from simple_backup import simple_backup as sb
|
|
|
|
OPTIONS = '-r -v --ignore-missing-args --mkpath -a -h -H -X -s'
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fake_now(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Give each backup a new timestamp, as real runs within the same second would collide"""
|
|
|
|
def times() -> Iterator[datetime]:
|
|
t = datetime(2026, 10, 14, 10, 0, 0)
|
|
|
|
while True:
|
|
yield t
|
|
t += timedelta(hours=1)
|
|
|
|
it = times()
|
|
|
|
class FakeDatetime:
|
|
@staticmethod
|
|
def now() -> datetime:
|
|
return next(it)
|
|
|
|
monkeypatch.setattr(sb, 'datetime', FakeDatetime)
|
|
|
|
|
|
@pytest.fixture
|
|
def src(tmp_path: Path) -> Path:
|
|
src = tmp_path / 'src dir'
|
|
src.mkdir()
|
|
(src / 'same.txt').write_text('unchanged')
|
|
(src / 'changed.txt').write_text('first')
|
|
|
|
return src
|
|
|
|
|
|
@pytest.fixture
|
|
def out(tmp_path: Path) -> Path:
|
|
out = tmp_path / 'out dir'
|
|
out.mkdir()
|
|
|
|
return out
|
|
|
|
|
|
def make_backup(out: Path, inputs: list[str], keep: int = -1, exclude: list[str] | None = None,
|
|
remove_before: bool = False, progress: bool = False) -> sb.Backup:
|
|
backup = sb.Backup(inputs, str(out), exclude or [], keep, OPTIONS, remove_before=remove_before,
|
|
progress=progress)
|
|
|
|
assert backup.check_params() == 0
|
|
|
|
return backup
|
|
|
|
|
|
def run_backup(out: Path, inputs: list[str], **kwargs: Any) -> int:
|
|
return make_backup(out, inputs, **kwargs).run()
|
|
|
|
|
|
def backups(out: Path) -> list[str]:
|
|
return sorted(d.name for d in (out / 'simple_backup').iterdir() if d.is_dir() and not d.is_symlink())
|
|
|
|
|
|
def copied(out: Path, backup: str, path: Path) -> Path:
|
|
"""Path of a file inside a backup (rsync keeps the full source path)"""
|
|
|
|
return out / 'simple_backup' / backup / str(path).lstrip('/')
|
|
|
|
|
|
class TestCheckParams:
|
|
def test_no_inputs(self, out: Path) -> None:
|
|
assert sb.Backup([], str(out), [], -1, OPTIONS).check_params() == 1
|
|
|
|
def test_no_output(self) -> None:
|
|
assert sb.Backup(['/x'], None, [], -1, OPTIONS).check_params() == 2
|
|
|
|
def test_missing_output(self, tmp_path: Path) -> None:
|
|
assert sb.Backup(['/x'], str(tmp_path / 'missing'), [], -1, OPTIONS).check_params() == 2
|
|
|
|
def test_relative_output(self, out: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.chdir(out.parent)
|
|
backup = sb.Backup(['/x'], out.name, [], None, OPTIONS)
|
|
|
|
assert backup.check_params() == 0
|
|
assert backup.output == str(out)
|
|
assert backup.keep == -1
|
|
|
|
|
|
class TestRsync:
|
|
def test_first_backup(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture) -> None:
|
|
assert run_backup(out, [str(src)]) == 0
|
|
|
|
assert backups(out) == ['2026-10-14_10-00-00']
|
|
assert copied(out, '2026-10-14_10-00-00', src / 'same.txt').read_text() == 'unchanged'
|
|
assert 'Backup completed' in caplog.text
|
|
assert 'rsync: sent' in caplog.text
|
|
|
|
def test_hard_links(self, src: Path, out: Path) -> None:
|
|
assert run_backup(out, [str(src)]) == 0
|
|
(src / 'changed.txt').write_text('second')
|
|
assert run_backup(out, [str(src)]) == 0
|
|
|
|
first, second = backups(out)
|
|
|
|
assert copied(out, first, src / 'same.txt').stat().st_ino == \
|
|
copied(out, second, src / 'same.txt').stat().st_ino
|
|
assert copied(out, first, src / 'changed.txt').stat().st_ino != \
|
|
copied(out, second, src / 'changed.txt').stat().st_ino
|
|
assert copied(out, second, src / 'changed.txt').read_text() == 'second'
|
|
|
|
def test_exclude_and_missing_input(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture) -> None:
|
|
(src / 'skip.bak').touch()
|
|
|
|
assert run_backup(out, [str(src), str(src / 'missing')], exclude=['*.bak']) == 0
|
|
|
|
(backup,) = backups(out)
|
|
assert copied(out, backup, src / 'same.txt').exists()
|
|
assert not copied(out, backup, src / 'skip.bak').exists()
|
|
assert 'Input' in caplog.text and 'not found' in caplog.text
|
|
|
|
def test_all_inputs_missing(self, src: Path, out: Path) -> None:
|
|
assert run_backup(out, [str(src / 'missing')]) == 1
|
|
assert not (out / 'simple_backup').exists()
|
|
|
|
@pytest.mark.parametrize('keep, remaining', [
|
|
(-1, ['2026-10-14_10-00-00', '2026-10-14_11-00-00', '2026-10-14_12-00-00']),
|
|
(1, ['2026-10-14_11-00-00', '2026-10-14_12-00-00']),
|
|
(0, ['2026-10-14_12-00-00']),
|
|
])
|
|
def test_keep(self, src: Path, out: Path, keep: int, remaining: list[str]) -> None:
|
|
for _ in range(3):
|
|
assert run_backup(out, [str(src)], keep=keep) == 0
|
|
|
|
assert backups(out) == remaining
|
|
|
|
def test_removal_logged(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture) -> None:
|
|
for _ in range(3):
|
|
assert run_backup(out, [str(src)]) == 0
|
|
|
|
assert run_backup(out, [str(src)], keep=0) == 0
|
|
assert 'Removing backup 2026-10-14_10-00-00 (1/3)...' in caplog.text
|
|
assert 'Removing backup 2026-10-14_12-00-00 (3/3)...' in caplog.text
|
|
assert 'Removed 3 backups' in caplog.text
|
|
|
|
def test_remove_before(self, src: Path, out: Path) -> None:
|
|
assert run_backup(out, [str(src)]) == 0
|
|
assert run_backup(out, [str(src)], keep=0, remove_before=True) == 0
|
|
|
|
(backup,) = backups(out)
|
|
assert backup == '2026-10-14_11-00-00'
|
|
# Nothing to link against, as the old backup was removed first
|
|
assert copied(out, backup, src / 'same.txt').stat().st_nlink == 1
|
|
|
|
def test_removal_os_error(self, src: Path, out: Path, monkeypatch: pytest.MonkeyPatch,
|
|
caplog: pytest.LogCaptureFixture) -> None:
|
|
assert run_backup(out, [str(src)]) == 0
|
|
|
|
def busy(path: str) -> None:
|
|
raise OSError(errno.EBUSY, os.strerror(errno.EBUSY), path)
|
|
|
|
monkeypatch.setattr(sb, 'rmtree', busy)
|
|
|
|
assert run_backup(out, [str(src)], keep=0) == 0
|
|
assert backups(out) == ['2026-10-14_10-00-00', '2026-10-14_11-00-00']
|
|
assert 'Error while removing backup 2026-10-14_10-00-00. Device or resource busy' in caplog.text
|
|
|
|
def test_stray_files_and_symlinks(self, src: Path, out: Path) -> None:
|
|
assert run_backup(out, [str(src)]) == 0
|
|
|
|
base = out / 'simple_backup'
|
|
(base / 'notes.txt').write_text('keep me')
|
|
# Left by old versions; must not count as a backup, nor be used for --link-dest
|
|
(base / 'last_backup').symlink_to(base / backups(out)[0])
|
|
|
|
assert run_backup(out, [str(src)], keep=0) == 0
|
|
|
|
assert backups(out) == ['2026-10-14_11-00-00']
|
|
assert (base / 'notes.txt').read_text() == 'keep me'
|
|
assert (base / 'last_backup').is_symlink()
|
|
|
|
def test_old_folder_name_format(self, src: Path, out: Path) -> None:
|
|
assert run_backup(out, [str(src)]) == 0
|
|
|
|
base = out / 'simple_backup'
|
|
(base / '2026-10-14_10-00-00').rename(base / '2026-10-03 20:15:25')
|
|
|
|
assert run_backup(out, [str(src)]) == 0
|
|
assert backups(out) == ['2026-10-03 20:15:25', '2026-10-14_11-00-00']
|
|
assert copied(out, '2026-10-03 20:15:25', src / 'same.txt').stat().st_ino == \
|
|
copied(out, '2026-10-14_11-00-00', src / 'same.txt').stat().st_ino
|
|
|
|
assert run_backup(out, [str(src)], keep=0) == 0
|
|
assert backups(out) == ['2026-10-14_12-00-00']
|
|
|
|
def test_progress(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture,
|
|
capsys: pytest.CaptureFixture[str]) -> None:
|
|
assert run_backup(out, [str(src)], progress=True) == 0
|
|
|
|
assert '100%' in capsys.readouterr().err
|
|
# Only the summary is logged, not the progress updates
|
|
assert 'rsync: sent' in caplog.text
|
|
assert 'rsync: total size' in caplog.text
|
|
assert '%' not in caplog.text
|
|
|
|
def test_cli(self, src: Path, out: Path, tmp_path: Path, run_cli: Callable[..., int]) -> None:
|
|
config_file = str(tmp_path / 'missing.conf')
|
|
|
|
for _ in range(2):
|
|
assert run_cli('-c', config_file, '-i', str(src), '-o', str(out), '-k', '0') == 0
|
|
|
|
assert backups(out) == ['2026-10-14_11-00-00']
|
|
|
|
|
|
class ChunkStream(io.BufferedIOBase):
|
|
"""Returns the given chunks one per read1() call, like a pipe"""
|
|
|
|
def __init__(self, chunks: list[bytes]) -> None:
|
|
self.chunks = chunks
|
|
|
|
def read1(self, size: int = -1) -> bytes:
|
|
return self.chunks.pop(0) if self.chunks else b''
|
|
|
|
|
|
class TestReadRecords:
|
|
def test_split_on_newline_and_cr(self) -> None:
|
|
stream = ChunkStream([b'file\n\r 1.00M 50%\r 2.00M 100%\n\nsent 1 bytes\ntotal'])
|
|
|
|
assert list(sb._read_records(stream)) == ['file', '', ' 1.00M 50%', ' 2.00M 100%', '', 'sent 1 bytes',
|
|
'total']
|
|
|
|
def test_records_across_chunks(self) -> None:
|
|
stream = ChunkStream([b'fi', b'le\r', b'\nna', 'mé\n'.encode(), b'\xff\n'])
|
|
|
|
assert list(sb._read_records(stream)) == ['file', '', 'namé', '�']
|
|
|
|
|
|
class TestProgressBar:
|
|
@pytest.fixture(autouse=True)
|
|
def terminal(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(sb, 'get_terminal_size', lambda: os.terminal_size((81, 24)))
|
|
# 30 s elapsed at each update
|
|
monkeypatch.setattr(sb, 'default_timer', iter([0.0, 30.0]).__next__)
|
|
|
|
def test_scanning(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
bar = sb._ProgressBar()
|
|
|
|
assert bar.update(' 2.00M 25% 1.47MB/s 0:00:02 (xfr#1, ir-chk=1009/2632)')
|
|
assert capsys.readouterr().err == '\r1,623 files checked 2.00M copied (scanning)\033[K'
|
|
|
|
def test_update(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
bar = sb._ProgressBar()
|
|
|
|
# With --link-dest rsync's percentage leaves out linked files, so the files are counted instead
|
|
assert bar.update(' 4,000,000 1% 2.97MB/s 0:00:01 (xfr#2, to-chk=1/4)')
|
|
assert capsys.readouterr().err == \
|
|
'\r[######################--------] 75% 3/4 files 4,000,000 copied ETA 0:00:10\033[K'
|
|
|
|
def test_keeps_counts_within_file(self, monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str]) -> None:
|
|
monkeypatch.setattr(sb, 'default_timer', iter([0.0, 30.0, 30.0]).__next__)
|
|
bar = sb._ProgressBar()
|
|
|
|
assert bar.update(' 2.00M 1% 1.47MB/s 0:00:02 (xfr#1, to-chk=2/4)')
|
|
assert bar.update(' 3.00M 2% 1.47MB/s 0:00:03')
|
|
assert capsys.readouterr().err.endswith('] 50% 2/4 files 3.00M copied ETA 0:00:30\033[K')
|
|
|
|
def test_not_progress(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
bar = sb._ProgressBar()
|
|
|
|
for record in ['home/user/file 50% 1:00:00', 'sent 1 bytes received 2 bytes', 'total size is 3']:
|
|
assert not bar.update(record)
|
|
|
|
bar.clear()
|
|
assert capsys.readouterr().err == ''
|
|
|
|
def test_narrow_terminal(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
|
monkeypatch.setattr(sb, 'get_terminal_size', lambda: os.terminal_size((20, 24)))
|
|
|
|
assert sb._ProgressBar().update(' 6.00M 100% 1.48MB/s 0:00:03 (xfr#3, to-chk=0/4)')
|
|
# No room for the bar
|
|
assert capsys.readouterr().err == '\r100% 4/4 files 6.\033[K'
|
|
|
|
|
|
class FakePopen:
|
|
"""Stands in for rsync: creates the destination and returns a chosen return code"""
|
|
|
|
returncode = 0
|
|
calls: list[list[str]] = []
|
|
|
|
def __init__(self, args: list[str], **kwargs: Any) -> None:
|
|
FakePopen.calls.append(args)
|
|
os.makedirs(args[args.index('/') + 1])
|
|
self.stdin = io.BytesIO()
|
|
self.stdout = io.BytesIO(b'sending incremental file list\nsent 1 bytes received 2 bytes\ntotal size is 3\n')
|
|
|
|
def __enter__(self) -> 'FakePopen':
|
|
return self
|
|
|
|
def __exit__(self, *args: Any) -> None:
|
|
pass
|
|
|
|
def wait(self) -> int:
|
|
return FakePopen.returncode
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_popen(monkeypatch: pytest.MonkeyPatch) -> type[FakePopen]:
|
|
FakePopen.returncode = 0
|
|
FakePopen.calls = []
|
|
monkeypatch.setattr(sb, 'Popen', FakePopen)
|
|
|
|
return FakePopen
|
|
|
|
|
|
@pytest.fixture
|
|
def old_backups(out: Path) -> list[str]:
|
|
names = ['2026-10-01_10-00-00', '2026-10-02_10-00-00']
|
|
|
|
for name in names:
|
|
(out / 'simple_backup' / name).mkdir(parents=True)
|
|
|
|
return names
|
|
|
|
|
|
class TestReturnCodes:
|
|
def test_args(self, src: Path, out: Path, fake_popen: type[FakePopen], old_backups: list[str]) -> None:
|
|
assert run_backup(out, [str(src)]) == 0
|
|
|
|
(args,) = fake_popen.calls
|
|
assert args[0] == which('rsync')
|
|
assert f'--link-dest={out}/simple_backup/{old_backups[-1]}' in args
|
|
assert args[args.index('/') + 1] == f'{out}/simple_backup/2026-10-14_10-00-00'
|
|
|
|
def test_rsync_missing(self, src: Path, out: Path, tmp_path: Path, fake_popen: type[FakePopen],
|
|
old_backups: list[str], monkeypatch: pytest.MonkeyPatch,
|
|
caplog: pytest.LogCaptureFixture) -> None:
|
|
monkeypatch.setenv('PATH', str(tmp_path / 'empty'))
|
|
|
|
assert run_backup(out, [str(src)], keep=0, remove_before=True) == 4
|
|
assert fake_popen.calls == []
|
|
# Nothing was removed
|
|
assert backups(out) == old_backups
|
|
assert 'rsync not found' in caplog.text
|
|
|
|
@pytest.mark.parametrize('returncode, result, removed', [
|
|
(0, 0, True),
|
|
(24, 0, True),
|
|
(23, 4, False),
|
|
(2, 4, False),
|
|
])
|
|
def test_old_backups_removed_only_on_success(self, src: Path, out: Path, fake_popen: type[FakePopen],
|
|
old_backups: list[str], caplog: pytest.LogCaptureFixture,
|
|
returncode: int, result: int, removed: bool) -> None:
|
|
fake_popen.returncode = returncode
|
|
|
|
assert run_backup(out, [str(src)], keep=0) == result
|
|
|
|
if removed:
|
|
assert backups(out) == ['2026-10-14_10-00-00']
|
|
else:
|
|
assert backups(out) == [*old_backups, '2026-10-14_10-00-00']
|
|
assert 'Old backups will not be removed' in caplog.text
|
|
assert f'return code {returncode}' in caplog.text
|
|
|
|
@pytest.mark.parametrize('keep, message', [
|
|
(0, 'No complete backup is available'),
|
|
(1, 'Old backups were removed before the backup'),
|
|
])
|
|
def test_failure_after_remove_before(self, src: Path, out: Path, fake_popen: type[FakePopen],
|
|
old_backups: list[str], caplog: pytest.LogCaptureFixture,
|
|
keep: int, message: str) -> None:
|
|
fake_popen.returncode = 23
|
|
|
|
assert run_backup(out, [str(src)], keep=keep, remove_before=True) == 4
|
|
assert message in caplog.text
|