From 28ecff230645c8eb02eb5ef65ed3daaebdd9636c Mon Sep 17 00:00:00 2001 From: Fuxino Date: Wed, 14 Oct 2026 08:15:07 +0200 Subject: [PATCH] Add tests for local backups Run real rsync into a temp dir to test hard links, keep, remove-before-backup, stray files in the backup dir and the old folder name format, and use a fake Popen to test how rsync return codes are handled. --- tests/test_local.py | 267 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 267 insertions(+) create mode 100644 tests/test_local.py diff --git a/tests/test_local.py b/tests/test_local.py new file mode 100644 index 0000000..a668993 --- /dev/null +++ b/tests/test_local.py @@ -0,0 +1,267 @@ +"""Tests for local backups, with real rsync runs and with a fake Popen""" + +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) -> sb.Backup: + backup = sb.Backup(inputs, str(out), exclude or [], keep, OPTIONS, remove_before=remove_before) + + 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 # type: ignore[arg-type] + + 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) # type: ignore[arg-type] + + 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_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_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 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]) + + def __enter__(self) -> 'FakePopen': + return self + + def __exit__(self, *args: Any) -> None: + pass + + def communicate(self) -> tuple[bytes, None]: + return b'sending incremental file list\nsent 1 bytes received 2 bytes\ntotal size is 3\n', None + + +@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