Fuxino
8a479a8455
With --link-dest unchanged files are linked, not copied, so rsync's byte percentage stayed near zero and the ETA was hours. Use the checked/total file counts instead, and show no percentage or ETA while still scanning.
361 lines
14 KiB
Python
361 lines
14 KiB
Python
"""Tests for local backups, with real rsync runs and with a fake Popen"""
|
|
|
|
import io
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterator, Optional
|
|
|
|
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: Optional[list[str]] = 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_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] == '/usr/bin/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'
|
|
|
|
@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
|