Fuxino
ec27b8d77f
With incremental recursion rsync only knows the total near the end, so the bar showed no ETA for most of the backup. Save the file count as a third line of the state file and use it as an estimate until rsync reports the real total.
403 lines
15 KiB
Python
403 lines
15 KiB
Python
"""Tests for combining command line and config values, and for detecting the user"""
|
|
|
|
import os
|
|
import pwd
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from simple_backup import simple_backup as sb
|
|
|
|
USER = pwd.getpwuid(os.getuid()).pw_name
|
|
HOME = pwd.getpwuid(os.getuid()).pw_dir
|
|
UID = os.getuid()
|
|
|
|
DEFAULT_OPTIONS = '-r -v --ignore-missing-args --mkpath -a -h -H -X -s'
|
|
|
|
RunCli = Callable[..., int]
|
|
|
|
|
|
class FakeBackup:
|
|
"""Records the values simple_backup() passes to Backup"""
|
|
|
|
instances: list['FakeBackup'] = []
|
|
check_code = 0
|
|
run_code = 0
|
|
destination = '/backups/simple_backup/2026-10-25_10-00-00'
|
|
|
|
def __init__(self, inputs: Any, output: Any, exclude: Any, keep: Any, options: str,
|
|
ssh_host: str | None = None, ssh_user: str | None = None,
|
|
ssh_keyfile: str | None = None, remote_sudo: bool = False,
|
|
remove_before: bool = False, verbose: bool = False, progress: bool = False,
|
|
file_estimate: int = 0) -> None:
|
|
self.inputs = inputs
|
|
self.output = output
|
|
self.exclude = exclude
|
|
self.keep = keep
|
|
self.options = options
|
|
self.ssh_host = ssh_host
|
|
self.ssh_user = ssh_user
|
|
self.ssh_keyfile = ssh_keyfile
|
|
self.remote_sudo = remote_sudo
|
|
self.remove_before = remove_before
|
|
self.verbose = verbose
|
|
self.progress = progress
|
|
self.file_estimate = file_estimate
|
|
self.file_count: int | None = None
|
|
self.homedir: str | None = None
|
|
self.ran = False
|
|
self.closed = False
|
|
FakeBackup.instances.append(self)
|
|
|
|
def check_params(self, homedir: str = '') -> int:
|
|
self.homedir = homedir
|
|
|
|
return FakeBackup.check_code
|
|
|
|
def run(self) -> int:
|
|
self.ran = True
|
|
|
|
return FakeBackup.run_code
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fake_backup(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
FakeBackup.instances = []
|
|
FakeBackup.check_code = 0
|
|
FakeBackup.run_code = 0
|
|
monkeypatch.setattr(sb, 'Backup', FakeBackup)
|
|
|
|
|
|
@pytest.fixture
|
|
def as_user(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Run as the current (non-root) user"""
|
|
|
|
monkeypatch.setattr(os, 'geteuid', lambda: UID if UID != 0 else 1000)
|
|
monkeypatch.setenv('USER', USER)
|
|
monkeypatch.setenv('HOME', HOME)
|
|
|
|
|
|
@pytest.fixture
|
|
def as_root(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Run as root, with no sudo/doas variables set"""
|
|
|
|
monkeypatch.setattr(os, 'geteuid', lambda: 0)
|
|
# Writing the state file as root would call seteuid() (covered in test_state.py)
|
|
monkeypatch.setattr(sb, '_write_state', lambda user, destination, file_count=None: None)
|
|
|
|
for var in ['SUDO_USER', 'SUDO_UID', 'DOAS_USER']:
|
|
monkeypatch.delenv(var, raising=False)
|
|
|
|
|
|
@pytest.fixture
|
|
def config_calls(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str | None]]:
|
|
"""Record the config file path and user instead of reading the file"""
|
|
|
|
calls: list[tuple[str, str | None]] = []
|
|
read_config = sb._read_config
|
|
|
|
def record(config_file: str, user: str | None = None) -> sb.ConfigArgs:
|
|
calls.append((config_file, user))
|
|
|
|
return read_config('/nonexistent/simple_backup.conf', user)
|
|
|
|
monkeypatch.setattr(sb, '_read_config', record)
|
|
|
|
return calls
|
|
|
|
|
|
@pytest.fixture
|
|
def config(tmp_path: Path) -> Callable[[str], str]:
|
|
"""Write a config file with existing inputs, followed by the given text"""
|
|
|
|
(tmp_path / 'data').mkdir()
|
|
|
|
def write(text: str = '') -> str:
|
|
config_file = tmp_path / 'simple_backup.conf'
|
|
config_file.write_text(f'[backup]\ninputs={tmp_path}/data\n{text}\n')
|
|
|
|
return str(config_file)
|
|
|
|
return write
|
|
|
|
|
|
def last_backup() -> FakeBackup:
|
|
assert len(FakeBackup.instances) == 1
|
|
|
|
return FakeBackup.instances[0]
|
|
|
|
|
|
@pytest.mark.usefixtures('as_user')
|
|
class TestValues:
|
|
def test_config_values(self, run_cli: RunCli, config: Callable[[str], str], tmp_path: Path) -> None:
|
|
config_file = config("""backup_dir=/backups
|
|
exclude=*.bak
|
|
keep=3
|
|
[server]
|
|
ssh_host=server
|
|
ssh_user=user
|
|
ssh_keyfile=/keys/id
|
|
remote_sudo=true
|
|
numeric_ids=true
|
|
""")
|
|
|
|
assert run_cli('-c', config_file) == 0
|
|
|
|
b = last_backup()
|
|
assert b.inputs == [f'{tmp_path}/data']
|
|
assert b.output == '/backups'
|
|
assert b.exclude == ['*.bak']
|
|
assert b.keep == 3
|
|
assert (b.ssh_host, b.ssh_user, b.ssh_keyfile) == ('server', 'user', '/keys/id')
|
|
assert b.remote_sudo is True
|
|
assert b.options == f'{DEFAULT_OPTIONS} --numeric-ids'
|
|
assert b.ran and b.closed
|
|
|
|
def test_cli_overrides_config(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
|
|
config_file = config("""backup_dir=/backups
|
|
exclude=*.bak
|
|
keep=3
|
|
[server]
|
|
ssh_host=server
|
|
ssh_user=user
|
|
ssh_keyfile=/keys/id
|
|
""")
|
|
|
|
assert run_cli('-c', config_file, '-i', '/cli/in', '-o', '/cli/out', '-e', '*.tmp', '-k', '0',
|
|
'--ssh-host', 'cli-host', '--ssh-user', 'cli-user', '--keyfile', '/cli/key') == 0
|
|
|
|
b = last_backup()
|
|
assert b.inputs == ['/cli/in']
|
|
assert b.output == '/cli/out'
|
|
assert b.exclude == ['*.tmp']
|
|
assert b.keep == 0
|
|
assert (b.ssh_host, b.ssh_user, b.ssh_keyfile) == ('cli-host', 'cli-user', '/cli/key')
|
|
|
|
def test_inputs_and_output_only_from_cli(self, run_cli: RunCli, tmp_path: Path) -> None:
|
|
config_file = tmp_path / 'simple_backup.conf'
|
|
config_file.write_text('[backup]\nkeep=3\n')
|
|
|
|
assert run_cli('-c', str(config_file), '-i', '/cli/in', '-o', '/cli/out') == 0
|
|
|
|
b = last_backup()
|
|
assert (b.inputs, b.output, b.keep) == (['/cli/in'], '/cli/out', 3)
|
|
|
|
def test_ssh_host_and_user_from_cli_and_config(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
|
|
config_file = config('backup_dir=/backups\n[server]\nssh_user=user')
|
|
|
|
assert run_cli('-c', config_file, '--ssh-host', 'server') == 0
|
|
assert (last_backup().ssh_host, last_backup().ssh_user) == ('server', 'user')
|
|
|
|
@pytest.mark.parametrize('server, argv', [
|
|
('ssh_host=server', []),
|
|
('ssh_user=user', []),
|
|
('', ['--ssh-host', 'server']),
|
|
('', ['--ssh-user', 'user']),
|
|
])
|
|
def test_only_one_of_ssh_host_and_user(self, run_cli: RunCli, config: Callable[[str], str],
|
|
caplog: pytest.LogCaptureFixture, server: str,
|
|
argv: list[str]) -> None:
|
|
config_file = config(f'backup_dir=/backups\n[server]\n{server}')
|
|
|
|
assert run_cli('-c', config_file, *argv) == 6
|
|
assert FakeBackup.instances == []
|
|
assert 'Both SSH host and SSH user are needed' in caplog.text
|
|
|
|
@pytest.mark.parametrize('config_keep, argv', [
|
|
('keep=-2', []),
|
|
('', ['-k', '-2']),
|
|
])
|
|
def test_keep_below_minus_one(self, run_cli: RunCli, config: Callable[[str], str],
|
|
caplog: pytest.LogCaptureFixture, config_keep: str, argv: list[str]) -> None:
|
|
config_file = config(f'backup_dir=/backups\n{config_keep}')
|
|
|
|
assert run_cli('-c', config_file, *argv) == 6
|
|
assert FakeBackup.instances == []
|
|
assert 'Invalid number of backups to keep: -2' in caplog.text
|
|
|
|
def test_bad_config(self, run_cli: RunCli, config: Callable[[str], str],
|
|
caplog: pytest.LogCaptureFixture) -> None:
|
|
config_file = config('backup_dir=/backups\nkeep=abc')
|
|
|
|
assert run_cli('-c', config_file) == 6
|
|
assert FakeBackup.instances == []
|
|
assert 'Bad configuration file' in caplog.text
|
|
|
|
def test_local_output_expanded(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
|
|
assert run_cli('-c', config('backup_dir=~/backups')) == 0
|
|
assert last_backup().output == f'{HOME}/backups'
|
|
|
|
def test_remote_output_not_expanded(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
|
|
config_file = config('backup_dir=~/backups\n[server]\nssh_host=server\nssh_user=user')
|
|
|
|
assert run_cli('-c', config_file) == 0
|
|
# Expanded on the server
|
|
assert last_backup().output == '~/backups'
|
|
|
|
@pytest.mark.parametrize('argv, options', [
|
|
([], DEFAULT_OPTIONS),
|
|
(['--rsync-options', 'a', 'l'], '-r -v --ignore-missing-args --mkpath -a -l'),
|
|
(['-s', '-z', '--numeric-ids'], f'{DEFAULT_OPTIONS} -c -z --numeric-ids'),
|
|
])
|
|
def test_rsync_options(self, run_cli: RunCli, config: Callable[[str], str], argv: list[str],
|
|
options: str) -> None:
|
|
assert run_cli('-c', config('backup_dir=/backups'), *argv) == 0
|
|
assert last_backup().options == options
|
|
|
|
def test_flags(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
|
|
assert run_cli('-c', config('backup_dir=/backups'), '--remote-sudo', '--remove-before-backup', '-v') == 0
|
|
|
|
b = last_backup()
|
|
assert b.remote_sudo and b.remove_before and b.verbose
|
|
|
|
@pytest.mark.parametrize('tty, argv, progress', [
|
|
(True, [], True),
|
|
(True, ['--no-progress'], False),
|
|
(False, [], False),
|
|
])
|
|
def test_progress(self, run_cli: RunCli, config: Callable[[str], str], monkeypatch: pytest.MonkeyPatch,
|
|
tty: bool, argv: list[str], progress: bool) -> None:
|
|
monkeypatch.setattr(sb.sys.stderr, 'isatty', lambda: tty)
|
|
|
|
assert run_cli('-c', config('backup_dir=/backups'), *argv) == 0
|
|
assert last_backup().progress is progress
|
|
|
|
@pytest.mark.parametrize('tty, estimate', [(True, 1234), (False, 0)])
|
|
def test_file_estimate(self, run_cli: RunCli, config: Callable[[str], str], monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path, tty: bool, estimate: int) -> None:
|
|
monkeypatch.setattr(sb.sys.stderr, 'isatty', lambda: tty)
|
|
state = tmp_path / 'state' / 'simple_backup' / 'last_backup'
|
|
state.parent.mkdir(parents=True)
|
|
state.write_text('2026-10-25T10:00:00+02:00\n/backups/old\n1234\n')
|
|
|
|
assert run_cli('-c', config('backup_dir=/backups')) == 0
|
|
assert last_backup().file_estimate == estimate
|
|
|
|
def test_check_params_failure(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
|
|
FakeBackup.check_code = 2
|
|
|
|
assert run_cli('-c', config('backup_dir=/backups')) == 2
|
|
|
|
b = last_backup()
|
|
assert not b.ran
|
|
assert b.closed
|
|
|
|
@pytest.mark.parametrize('check_code, run_code, saved', [(0, 0, True), (2, 0, False), (0, 4, False)])
|
|
def test_state_saved_only_on_success(self, run_cli: RunCli, config: Callable[[str], str], tmp_path: Path,
|
|
check_code: int, run_code: int, saved: bool) -> None:
|
|
FakeBackup.check_code = check_code
|
|
FakeBackup.run_code = run_code
|
|
|
|
assert run_cli('-c', config('backup_dir=/backups')) == (check_code or run_code)
|
|
|
|
state = tmp_path / 'state' / 'simple_backup' / 'last_backup'
|
|
assert state.exists() is saved
|
|
|
|
if saved:
|
|
assert state.read_text().splitlines()[1] == FakeBackup.destination
|
|
|
|
|
|
class TestUser:
|
|
@pytest.mark.usefixtures('as_user')
|
|
def test_current_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]]) -> None:
|
|
assert run_cli() == 0
|
|
assert config_calls == [(f'{HOME}/.config/simple_backup/simple_backup.conf', USER)]
|
|
assert last_backup().homedir == HOME
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_explicit_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]]) -> None:
|
|
assert run_cli('-u', USER, '-c', '~/other.conf') == 0
|
|
assert config_calls == [(f'{HOME}/other.conf', USER)]
|
|
assert last_backup().homedir == HOME
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
@pytest.mark.parametrize('var', ['SUDO_USER', 'DOAS_USER'])
|
|
def test_sudo_or_doas_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]],
|
|
monkeypatch: pytest.MonkeyPatch, var: str) -> None:
|
|
monkeypatch.setenv(var, USER)
|
|
|
|
assert run_cli() == 0
|
|
assert config_calls == [(f'{HOME}/.config/simple_backup/simple_backup.conf', USER)]
|
|
assert last_backup().homedir == HOME
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_root_without_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]],
|
|
caplog: pytest.LogCaptureFixture) -> None:
|
|
assert run_cli() == 0
|
|
# '~' is left to expanduser, i.e. $HOME
|
|
assert config_calls[0][1] is None
|
|
assert last_backup().homedir == ''
|
|
assert 'Failed to detect user' in caplog.text
|
|
|
|
|
|
class TestDetectUser:
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_root_prefers_sudo_user(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv('SUDO_USER', 'alice')
|
|
monkeypatch.setenv('DOAS_USER', 'bob')
|
|
|
|
assert sb._detect_user() == 'alice'
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_root_doas_user(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv('DOAS_USER', 'bob')
|
|
|
|
assert sb._detect_user() == 'bob'
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_root_no_user(self) -> None:
|
|
assert sb._detect_user() is None
|
|
|
|
@pytest.mark.usefixtures('as_user')
|
|
def test_non_root(self) -> None:
|
|
assert sb._detect_user() == USER
|
|
|
|
|
|
def test_run_as_script(tmp_path: Path) -> None:
|
|
"""Running the module directly passes simple_backup()'s return code to the shell"""
|
|
|
|
repo = Path(sb.__file__).resolve().parents[1]
|
|
env = {**os.environ, 'PYTHONPATH': str(repo), 'HOME': str(tmp_path)}
|
|
|
|
# No config file and no inputs: check_params() returns 1
|
|
result = subprocess.run([sys.executable, '-m', 'simple_backup.simple_backup', '--no-syslog',
|
|
'-c', str(tmp_path / 'missing.conf')], env=env, cwd=tmp_path,
|
|
capture_output=True, check=False)
|
|
|
|
assert result.returncode == 1
|
|
|
|
|
|
class TestTargetUid:
|
|
@pytest.mark.usefixtures('as_user')
|
|
def test_non_root(self) -> None:
|
|
assert sb._target_uid() == os.geteuid()
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_sudo_uid(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv('SUDO_UID', '1234')
|
|
|
|
assert sb._target_uid() == 1234
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_doas_user(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv('DOAS_USER', USER)
|
|
|
|
assert sb._target_uid() == UID
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_unknown_doas_user(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv('DOAS_USER', 'no-such-user-simple-backup')
|
|
|
|
assert sb._target_uid() is None
|
|
|
|
@pytest.mark.usefixtures('as_root')
|
|
def test_no_user(self) -> None:
|
|
assert sb._target_uid() is None
|