Fuxino
0829c33053
Accept None for the missing inputs, output, exclude and keep values and replace them with their defaults in __init__, and annotate the SSH client, which mypy typed as None and so skipped checking the remote code.
268 lines
9.5 KiB
Python
268 lines
9.5 KiB
Python
"""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
|
|
|
|
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_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
|