Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a7fd325fa7
|
||
|
|
8a479a8455
|
||
|
|
3e486eb3ab
|
||
|
|
331e72b373
|
||
|
|
13cae98776
|
||
|
|
7cd90f8d11
|
||
|
|
d0fb5b65fb
|
||
|
|
a3ace84c8b
|
||
|
|
10eccd360c
|
||
|
|
0829c33053
|
||
|
|
cfe3b28436
|
||
|
|
1fb0513467
|
||
|
|
d803790c1a
|
||
|
|
8d08032b5d
|
||
|
|
28ecff2306
|
||
|
|
dbb0aadc84
|
||
|
|
41a45dad41
|
||
|
|
4b28cb7edf
|
||
|
|
63774e92d7
|
||
|
|
166c9dbc47
|
||
|
|
28d1b0d0d0
|
||
|
|
cd80c941ad
|
||
|
|
29958896a2
|
||
|
|
d815b0832a
|
||
|
|
e6ac9e6cab
|
||
|
|
9298f4232c
|
||
|
|
1dca989310
|
||
|
|
9a68a8660d
|
||
|
|
73de4f8d25
|
||
|
|
baf5b07d23
|
||
|
|
8b30fefed3
|
||
|
|
2a5562d1a9
|
||
|
|
7091118663
|
||
|
|
3113e182c7
|
||
|
|
8fc3c42a5e
|
||
|
|
0671f86579
|
||
|
|
2f207efcf9
|
||
|
|
3cbca8e4d8
|
||
|
|
54880651be
|
||
|
|
793a30fad8
|
||
|
|
871dd0f49a
|
||
|
|
30e4416f8f
|
||
|
|
d0f888cbb4
|
||
|
|
7ef29cf551
|
||
|
|
4f5499eba9
|
||
|
|
fc165fccbf
|
||
|
|
d79ea56c0f
|
||
|
|
5189910131
|
||
|
|
7d4c43f8ff
|
||
|
|
ac7e6ec21b
|
||
|
|
578d992a19
|
||
|
|
e2d8897a45
|
||
|
|
dadc4b0265
|
||
|
|
7e7d11c520
|
||
|
|
22e0556b50
|
||
|
|
2b91178e94
|
||
|
|
2c2649113d
|
||
|
|
0f4a100ac7
|
||
|
|
60d83151dc
|
||
|
|
47aec456be
|
||
|
|
c715188973
|
||
|
|
57613649c2
|
||
|
|
67dbb49a67
|
||
|
|
e62a668f3e
|
||
|
|
bd8934d57f
|
||
|
|
96fe3c7813
|
||
|
|
9fdb959540
|
||
|
|
eb889beee7
|
||
|
|
2a37eb4172
|
||
|
|
45a07205a1
|
@@ -0,0 +1,48 @@
|
||||
# PKGBUILD
|
||||
|
||||
# Maintainer: Daniele Fucini <dfucini [at] gmail [dot] com>
|
||||
|
||||
pkgname=simple_backup
|
||||
pkgdesc='Simple backup script that uses rsync to copy files'
|
||||
pkgver=4.4.0
|
||||
pkgrel=2
|
||||
url="https://git.shouldnt.work/fuxino/$pkgname"
|
||||
arch=('any')
|
||||
license=('GPL-3.0-or-later')
|
||||
makedepends=('git'
|
||||
'python-setuptools'
|
||||
'python-build'
|
||||
'python-installer')
|
||||
depends=('python>=3.10'
|
||||
'rsync')
|
||||
optdepends=('python-systemd: use systemd log'
|
||||
'python-dbus: for desktop notifications'
|
||||
'python-paramiko: for remote backup through ssh')
|
||||
checkdepends=('python-pytest'
|
||||
'python-paramiko'
|
||||
'openssh')
|
||||
conflicts=('simple_backup-git')
|
||||
source=("git+$url?signed#tag=$pkgver")
|
||||
validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
|
||||
sha256sums=('cc5944558ed4bc30a0d1a00b49dfced371909dd6c4f39b0c1ccf8f39b445e907')
|
||||
|
||||
build()
|
||||
{
|
||||
cd "$srcdir"/"$pkgname"
|
||||
python -m build --wheel --no-isolation
|
||||
}
|
||||
|
||||
check()
|
||||
{
|
||||
cd "$srcdir"/"$pkgname"
|
||||
python -m pytest
|
||||
}
|
||||
|
||||
package()
|
||||
{
|
||||
cd "$srcdir"/"$pkgname"
|
||||
python -m installer --destdir="$pkgdir" dist/*.whl
|
||||
install -Dm644 "$pkgname"/"$pkgname".conf "$pkgdir"/usr/share/doc/"$pkgname"/"$pkgname".conf
|
||||
install -Dm644 README.md "$pkgdir"/usr/share/doc/"$pkgname"/README.md
|
||||
install -Dm644 man/"$pkgname".1 "$pkgdir"/usr/share/man/man1/"$pkgname".1
|
||||
}
|
||||
@@ -20,42 +20,60 @@ The script uses rsync to actually run the backup, so you will have to install it
|
||||
sudo pacman -Syu rsync
|
||||
```
|
||||
|
||||
It's also required to have python-dotenv
|
||||
|
||||
Optional dependencies are systemd-python to enable using systemd journal for logging, and dbus-python for desktop notifications.
|
||||
Optional dependencies are systemd-python to enable using systemd journal for logging, dbus-python for desktop notifications, and paramiko (version 3.2 or later) for remote backup.
|
||||
|
||||
## Install
|
||||
To install the program, first clone the repository:
|
||||
For Arch Linux and Arch-based distros, two packages are available in the AUR (aur.archlinux.org):
|
||||
- **simple_backup** for the release version
|
||||
- **simple_backup-git** for the git version
|
||||
|
||||
To install it from source, first clone the repository:
|
||||
|
||||
```bash
|
||||
git clone https://github.com/Fuxino/simple_backup.git
|
||||
git clone https://git.shouldnt.work/fuxino/simple_backup.git
|
||||
```
|
||||
|
||||
Then install the tools required to build the package:
|
||||
|
||||
```bash
|
||||
pip install --upgrade build wheel
|
||||
```
|
||||
|
||||
Finally, run:
|
||||
Then build the package (this requires the Python `build` module, e.g. the python-build package on Arch Linux):
|
||||
|
||||
```bash
|
||||
cd simple_backup
|
||||
python -m build --wheel
|
||||
python -m pip install dist/*.whl
|
||||
```
|
||||
|
||||
For Arch Linux and Arch-based distros, two packages are available in the AUR (aur.archlinux.org):
|
||||
- **simple_backup** for the release version
|
||||
- **simple_backup-git** for the git version
|
||||
Many distributions don't allow installing packages with pip into the system Python ("externally-managed-environment" error).
|
||||
In that case, install the wheel in a virtual environment, for example with pipx:
|
||||
|
||||
```bash
|
||||
pipx install dist/*.whl
|
||||
```
|
||||
|
||||
Optional dependencies can be installed as extras, for example `pipx install 'dist/simple_backup-<version>-py3-none-any.whl[REMOTE]'` (available extras: JOURNAL, NOTIFICATIONS, REMOTE).
|
||||
|
||||
## Running as root
|
||||
To back up files that your user can't read, run simple_backup with sudo or doas.
|
||||
In that case the user who ran the command is detected automatically, so that '~' (in the inputs and in the default configuration file path) refers to that user's home directory, not root's.
|
||||
When running as root in other ways (e.g. from a systemd service), specify the user with the --user (or -u) argument.
|
||||
|
||||
## Old backups
|
||||
The number of old backups to keep can be set with the --keep (or -k) argument (or in the configuration file).
|
||||
Old backups are only removed if rsync completes successfully, so that they are not deleted when the new backup may be incomplete.
|
||||
The only rsync error that is ignored is return code 24 (source files vanished during the transfer), which is common when backing up a live system.
|
||||
|
||||
## Remote backup
|
||||
> **Warning**
|
||||
> This feature is experimental
|
||||
|
||||
It's possible to use a remote server as destination for the backup. Just use the --username (or -u) and --host arguments (or set them in the configuration file).
|
||||
It's possible to use a remote server as destination for the backup. Just use the --ssh-user and --ssh-host arguments (or set them in the configuration file).
|
||||
Both are required: if only one of them is set, simple_backup exits with an error.
|
||||
For this to work, rsync must be installed on the server too.
|
||||
|
||||
The output directory is a path on the server. A leading '~' and relative paths refer to the user's home directory on the server.
|
||||
On the command line, quote '~' so that your local shell doesn't expand it:
|
||||
|
||||
```bash
|
||||
simple_backup --ssh-host server --ssh-user user -o '~/backups'
|
||||
```
|
||||
|
||||
### Server authentication
|
||||
The best way to handle the authentication is to have an ssh agent running on your system, otherwise if a passphrase is necessary to unlock the ssh key, it will be necessary to enter it more than once.
|
||||
If needed, it's possible to specify the ssh key location with the --keyfile argument or in the configuration file.
|
||||
|
||||
+37
-7
@@ -1,4 +1,4 @@
|
||||
.TH SIMPLE_BACKUP 1 2023-06-15 SIMPLE_BACKUP 3.2.6
|
||||
.TH SIMPLE_BACKUP 1 2026-10-15 SIMPLE_BACKUP 4.4.1
|
||||
.SH NAME
|
||||
simple_backup \- Backup files and folders using rsync
|
||||
.SH SYNOPSIS
|
||||
@@ -52,7 +52,15 @@ or to use single or double quotes around them.
|
||||
.B \-o, \-\-output DIR
|
||||
Specify the directory where the files will be copied. The program will automatically
|
||||
create a subdirectory called \(aqsimple_backup\(aq (if it does not already exist) and
|
||||
inside this directory the actual backup directory (using the current date and time).
|
||||
inside this directory the actual backup directory (using the current date and time, for example
|
||||
\(aq2026\-10\-08_17\-34\-47\(aq).
|
||||
.RS
|
||||
.P
|
||||
Backups are best stored on a Linux filesystem (e.g. ext4 or btrfs). FAT and exFAT filesystems
|
||||
are supported, but they don\(aqt support hard links, so each backup will be a full copy instead
|
||||
of only storing files that changed since the previous one. File ownership and permissions are
|
||||
not preserved either.
|
||||
.RE
|
||||
.TP
|
||||
.B \-e, \-\-exclude FILE|DIR|PATTERN [FILE|...]]
|
||||
Specify files, directories or patterns to exclude from the backup. Matching files and directories
|
||||
@@ -61,6 +69,10 @@ will not be copied. Multiple elements can be specified, in the same way as for t
|
||||
.B \-k, \-\-keep N
|
||||
Specify how many old backups (so excluding the current one) will be kept. The default behavior
|
||||
is to keep them all (same as N=\-1).
|
||||
Old backups are only removed if rsync completes successfully, to avoid deleting them when the
|
||||
new backup may be incomplete. rsync return code 24 (source files vanished during the transfer)
|
||||
is treated as success, while any other error (including return code 23, partial transfer due to error)
|
||||
will prevent old backups from being removed. This does not apply when using \-\-remove\-before\-backup.
|
||||
.TP
|
||||
.B \-u, \-\-user USERNAME
|
||||
Explicitly specify the user running the backup (in case it is needed for home directory expansion).
|
||||
@@ -86,10 +98,19 @@ if saving bandwith is needed.
|
||||
Remove old backups (if necessary) before creating the new backup. Useful to free some space
|
||||
before performing the backup.
|
||||
Default behavior is to remove old backups after successfully completing the backup.
|
||||
Note that if the backup fails after old backups have been removed, only the backups kept
|
||||
according to \-\-keep will be left, and the new backup may be incomplete. In particular, with
|
||||
\-\-keep 0 there will be no complete backup available until the next successful one.
|
||||
.TP
|
||||
.B \-\-no\-syslog
|
||||
Don't use systemd journal for logging.
|
||||
.TP
|
||||
.B \-\-no\-progress
|
||||
Don't show a progress bar while rsync copies the files. The progress bar is only shown when
|
||||
the output is a terminal, so it never appears in the systemd journal or in redirected output.
|
||||
While rsync is still looking for files, the bar shows \(aq(scanning)\(aq, and the percentage
|
||||
and estimated time are only approximate.
|
||||
.TP
|
||||
.B \-\-rsync\-options OPTIONS [OPTION...]
|
||||
By default, the following rsync options are used:
|
||||
.RS
|
||||
@@ -100,7 +121,7 @@ Using \-\-rsync\-options it is possible to manually select which options to use.
|
||||
.P
|
||||
\-a, \-l, \-p, \-t, \-g, \-o, \-c, \-h, \-D, \-H, \-X, \-s
|
||||
.P
|
||||
Options \-r and \-v are used in any case. Not that options must be specified without dash (\-), for example:
|
||||
Options \-r, \-v, \-\-ignore\-missing\-args and \-\-mkpath are used in any case. Note that options must be specified without dash (\-), for example:
|
||||
.P
|
||||
.EX
|
||||
simple_backup \-\-rsync\-options a l p
|
||||
@@ -131,6 +152,14 @@ Copy it to the default location ($HOME/.config/simple_backup) and edit it as nee
|
||||
It is possible to choose a directory on a remote server as destination for the backup. The files
|
||||
are copied by rsync through SSH. Server hostname and username must be specified, either in the
|
||||
configuration file, or on the command line (\(aq\-\-ssh\-host\(aq and \(aq\-\-ssh\-user\(aq options).
|
||||
.P
|
||||
The output directory is a path on the server. A leading \(aq~\(aq and relative paths refer to the
|
||||
home directory of the user on the server. On the command line, quote \(aq~\(aq so that the local
|
||||
shell doesn\(aqt expand it, for example:
|
||||
.P
|
||||
.EX
|
||||
simple_backup \-\-ssh\-host server \-\-ssh\-user user \-o \(aq~/backups\(aq
|
||||
.EE
|
||||
.SS AUTHENTICATION
|
||||
For authentication, it is possible to use SSH key or password.
|
||||
.P
|
||||
@@ -165,7 +194,8 @@ for details. For this reason, use SSH key authentication if possible.
|
||||
.SH EXIT STATUS
|
||||
.TP
|
||||
.B 0
|
||||
The backup was completed without errors.
|
||||
The backup was completed without errors. This includes rsync return code 24 (some source files
|
||||
vanished during the transfer), which only logs a warning.
|
||||
.TP
|
||||
.B 1
|
||||
No valid inputs selected for backup.
|
||||
@@ -177,16 +207,16 @@ Backup failed because output directory for storing the backup does not exist.
|
||||
Permission denied to access the output directory.
|
||||
.TP
|
||||
.B 4
|
||||
rsync error (rsync returned a non-zero value).
|
||||
rsync error (rsync returned a value other than 0 or 24).
|
||||
.TP
|
||||
.B 5
|
||||
SSH connection failed.
|
||||
.TP
|
||||
.B 6
|
||||
Bad configuration file.
|
||||
Bad configuration file or options (for example, only one of SSH host and SSH user specified).
|
||||
.SH SEE ALSO
|
||||
.BR rsync (1)
|
||||
.SH AUTHORS
|
||||
.MT https://github.com/Fuxino
|
||||
.MT https://git.shouldnt.work/fuxino
|
||||
Daniele Fucini
|
||||
.ME
|
||||
|
||||
+61
-1
@@ -1,3 +1,63 @@
|
||||
[build-system]
|
||||
requires = ['setuptools']
|
||||
requires = ['setuptools>=77']
|
||||
build-backend = 'setuptools.build_meta'
|
||||
|
||||
[project]
|
||||
name = "simple_backup"
|
||||
dynamic = ["version"]
|
||||
description = "Simple backup script that uses rsync to copy files"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
license = "GPL-3.0-or-later"
|
||||
license-files = ["LICENSE"]
|
||||
authors = [
|
||||
{ name = "Daniele Fucini", email = "dfucini@gmail.com" }
|
||||
]
|
||||
maintainers = [
|
||||
{ name = "Daniele Fucini", email = "dfucini@gmail.com" }
|
||||
]
|
||||
classifiers = [
|
||||
"Development Status :: 4 - Beta",
|
||||
"Environment :: Console",
|
||||
"Natural Language :: English",
|
||||
"Operating System :: POSIX :: Linux",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
"Programming Language :: Python :: 3.14",
|
||||
"Topic :: System :: Archiving :: Backup",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
JOURNAL = ["systemd-python"]
|
||||
NOTIFICATIONS = ["dbus-python"]
|
||||
REMOTE = ["paramiko>=3.2"]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://git.shouldnt.work/fuxino/simple_backup"
|
||||
|
||||
[project.scripts]
|
||||
simple_backup = "simple_backup.simple_backup:simple_backup"
|
||||
|
||||
[tool.setuptools]
|
||||
packages = ["simple_backup"]
|
||||
|
||||
[tool.setuptools.dynamic]
|
||||
version = { attr = "simple_backup.__version__" }
|
||||
|
||||
[tool.mypy]
|
||||
files = ["simple_backup", "tests"]
|
||||
disallow_untyped_defs = true
|
||||
warn_unused_ignores = true
|
||||
|
||||
# No usable stubs (types-paramiko lags behind paramiko 5)
|
||||
[[tool.mypy.overrides]]
|
||||
module = ["paramiko", "systemd", "dbus"]
|
||||
ignore_missing_imports = true
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
# Import the repo copy, not an installed one
|
||||
pythonpath = ["."]
|
||||
|
||||
@@ -1,38 +0,0 @@
|
||||
[metadata]
|
||||
name = simple_backup
|
||||
version = attr: simple_backup.__version__
|
||||
description = Simple backup script using rsync
|
||||
long_description = file: README.md
|
||||
author = Daniele Fucini
|
||||
author_email = dfucini@gmail.com
|
||||
license = GPL3
|
||||
url = https://github.com/Fuxino/simple_backup
|
||||
classifiers =
|
||||
Development Status :: 4 - Beta
|
||||
Environment :: Console
|
||||
License :: OSI Approved :: GNU General Public License v3 (GPLv3)
|
||||
Natural Language :: English
|
||||
Operating System :: POSIX :: Linux
|
||||
Programming Language :: Python :: 3.10
|
||||
Programming Language :: Python :: 3.11
|
||||
Programming Language :: Python :: 3.12
|
||||
Topic :: System :: Archiving :: Backup
|
||||
|
||||
[options]
|
||||
packages = simple_backup
|
||||
python_requires = >=3.10
|
||||
install_requires =
|
||||
python-dotenv
|
||||
|
||||
[options.extras_require]
|
||||
JOURNAL =
|
||||
systemd-python
|
||||
NOTIFICATIONS =
|
||||
dbus-python
|
||||
REMOTE =
|
||||
paramiko
|
||||
|
||||
[options.entry_points]
|
||||
console_scripts =
|
||||
simple_backup = simple_backup.simple_backup:simple_backup
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""Init."""
|
||||
|
||||
__version__ = '4.1.2'
|
||||
__version__ = '4.4.1'
|
||||
|
||||
@@ -2,9 +2,9 @@
|
||||
|
||||
[backup]
|
||||
# Files and directories to backup. Multiple items can be separated using a comma (','). It is possible to use wildcards (i.e. '*' to match multiple characters and '~' for the user's home directory).
|
||||
inputs=/home/my_home,/etc
|
||||
inputs=/home/user
|
||||
|
||||
# Output directory.
|
||||
# Output directory. For remote backups, this is a path on the server ('~' and relative paths refer to the home directory on the server).
|
||||
backup_dir=/media/Backup
|
||||
|
||||
# Files, directories and patterns to exclude from the backup. Multiple items can be separated using a comma.
|
||||
@@ -18,5 +18,5 @@ keep=-1
|
||||
# ssh_host=
|
||||
# ssh_user=
|
||||
# ssh_keyfile=
|
||||
# remote_sudo=
|
||||
# numeric_ids=
|
||||
# remote_sudo=false
|
||||
# numeric_ids=false
|
||||
|
||||
+639
-346
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,30 @@
|
||||
"""Shared fixtures for the simple_backup tests"""
|
||||
|
||||
import sys
|
||||
from typing import Callable
|
||||
|
||||
import pytest
|
||||
|
||||
from simple_backup import simple_backup as sb
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def no_side_effects(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Keep tests out of the systemd journal and the desktop notifications"""
|
||||
|
||||
if hasattr(sb, 'j_handler'):
|
||||
sb.logger.removeHandler(sb.j_handler)
|
||||
|
||||
monkeypatch.setattr(sb, '_notify', lambda text: None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def run_cli(monkeypatch: pytest.MonkeyPatch) -> Callable[..., int]:
|
||||
"""Run simple_backup() with the given command line arguments"""
|
||||
|
||||
def run(*argv: str) -> int:
|
||||
monkeypatch.setattr(sys, 'argv', ['simple_backup', *argv])
|
||||
|
||||
return sb.simple_backup()
|
||||
|
||||
return run
|
||||
@@ -0,0 +1,331 @@
|
||||
"""Tests for combining command line and config values, and for detecting the user"""
|
||||
|
||||
import os
|
||||
import pwd
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
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
|
||||
|
||||
def __init__(self, inputs: Any, output: Any, exclude: Any, keep: Any, options: str,
|
||||
ssh_host: Optional[str] = None, ssh_user: Optional[str] = None,
|
||||
ssh_keyfile: Optional[str] = None, remote_sudo: bool = False,
|
||||
remove_before: bool = False, verbose: bool = False, progress: bool = False) -> 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.homedir: Optional[str] = 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 0
|
||||
|
||||
def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fake_backup(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
FakeBackup.instances = []
|
||||
FakeBackup.check_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)
|
||||
|
||||
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, Optional[str]]]:
|
||||
"""Record the config file path and user instead of reading the file"""
|
||||
|
||||
calls: list[tuple[str, Optional[str]]] = []
|
||||
read_config = sb._read_config
|
||||
|
||||
def record(config_file: str, user: Optional[str] = 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_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
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
|
||||
class TestUser:
|
||||
@pytest.mark.usefixtures('as_user')
|
||||
def test_current_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]]) -> 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, Optional[str]]]) -> 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, Optional[str]]],
|
||||
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, Optional[str]]],
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
@@ -0,0 +1,230 @@
|
||||
"""Tests for reading the configuration file and expanding paths"""
|
||||
|
||||
import configparser
|
||||
import os
|
||||
import pwd
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from simple_backup import simple_backup as sb
|
||||
|
||||
USER = pwd.getpwuid(os.getuid()).pw_name
|
||||
HOME = pwd.getpwuid(os.getuid()).pw_dir
|
||||
|
||||
|
||||
def write_config(path: Path, text: str) -> str:
|
||||
config_file = path / 'simple_backup.conf'
|
||||
config_file.write_text(text)
|
||||
|
||||
return str(config_file)
|
||||
|
||||
|
||||
def test_repo_copy_imported() -> None:
|
||||
repo = Path(__file__).resolve().parents[1]
|
||||
|
||||
assert Path(sb.__file__).resolve().is_relative_to(repo)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('path, expected', [
|
||||
('~', HOME),
|
||||
('~/docs', f'{HOME}/docs'),
|
||||
('/abs/~/path', '/abs/~/path'),
|
||||
('relative', 'relative'),
|
||||
])
|
||||
def test_expand_home(path: str, expected: str) -> None:
|
||||
assert sb._expand_home(path, USER) == expected
|
||||
|
||||
|
||||
def test_expand_inputs_glob(tmp_path: Path) -> None:
|
||||
for name in ['a.txt', 'b.txt', 'c.log']:
|
||||
(tmp_path / name).touch()
|
||||
|
||||
result = sb._expand_inputs([f'{tmp_path}/*.txt', ''], USER)
|
||||
|
||||
assert sorted(result) == [f'{tmp_path}/a.txt', f'{tmp_path}/b.txt']
|
||||
|
||||
|
||||
def test_expand_inputs_no_match(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
assert sb._expand_inputs([f'{tmp_path}/missing'], USER) == []
|
||||
assert 'No file or directory matching input' in caplog.text
|
||||
|
||||
|
||||
def test_expand_inputs_tilde_without_user(caplog: pytest.LogCaptureFixture) -> None:
|
||||
sb._expand_inputs(['~/nothing-here'], None)
|
||||
|
||||
assert 'Cannot expand' in caplog.text
|
||||
|
||||
|
||||
def test_missing_config(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
config_args = sb._read_config(str(tmp_path / 'missing.conf'), USER)
|
||||
|
||||
assert config_args['inputs'] is None
|
||||
assert config_args['output'] is None
|
||||
assert config_args['keep'] == -1
|
||||
assert 'does not exist' in caplog.text
|
||||
|
||||
|
||||
def test_missing_config_without_user(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
sb._read_config(str(tmp_path / 'missing.conf'), None)
|
||||
|
||||
assert 'User not specified' in caplog.text
|
||||
|
||||
|
||||
def test_read_config(tmp_path: Path) -> None:
|
||||
(tmp_path / 'one').mkdir()
|
||||
(tmp_path / 'two').mkdir()
|
||||
|
||||
config_file = write_config(tmp_path, f"""
|
||||
[backup]
|
||||
inputs={tmp_path}/one, {tmp_path}/two, {tmp_path}/one
|
||||
backup_dir=~/backups
|
||||
exclude=*.bak, , *.tmp
|
||||
keep=3
|
||||
""")
|
||||
|
||||
config_args = sb._read_config(config_file, USER)
|
||||
|
||||
assert config_args['inputs'] is not None
|
||||
assert sorted(config_args['inputs']) == [f'{tmp_path}/one', f'{tmp_path}/two']
|
||||
# Expanded later, as it may be a path on the server
|
||||
assert config_args['output'] == '~/backups'
|
||||
assert config_args['exclude'] == ['*.bak', '*.tmp']
|
||||
assert config_args['keep'] == 3
|
||||
assert config_args['ssh_host'] is None
|
||||
assert config_args['ssh_user'] is None
|
||||
assert config_args['ssh_keyfile'] is None
|
||||
assert config_args['remote_sudo'] is False
|
||||
assert config_args['numeric_ids'] is False
|
||||
|
||||
|
||||
def test_optional_keys_missing(tmp_path: Path) -> None:
|
||||
config_file = write_config(tmp_path, f"""
|
||||
[backup]
|
||||
inputs={tmp_path}
|
||||
backup_dir=/backups
|
||||
""")
|
||||
|
||||
config_args = sb._read_config(config_file, USER)
|
||||
|
||||
assert config_args['exclude'] == []
|
||||
assert config_args['keep'] == -1
|
||||
|
||||
|
||||
def test_old_default_section(tmp_path: Path) -> None:
|
||||
config_file = write_config(tmp_path, f"""
|
||||
[default]
|
||||
inputs={tmp_path}
|
||||
backup_dir=/backups
|
||||
""")
|
||||
|
||||
config_args = sb._read_config(config_file, USER)
|
||||
|
||||
assert config_args['inputs'] == [str(tmp_path)]
|
||||
assert config_args['output'] == '/backups'
|
||||
|
||||
|
||||
def test_percent_in_values(tmp_path: Path) -> None:
|
||||
config_file = write_config(tmp_path, f"""
|
||||
[backup]
|
||||
inputs={tmp_path}
|
||||
backup_dir=/backups/100%
|
||||
exclude=%temp%
|
||||
""")
|
||||
|
||||
config_args = sb._read_config(config_file, USER)
|
||||
|
||||
assert config_args['output'] == '/backups/100%'
|
||||
assert config_args['exclude'] == ['%temp%']
|
||||
|
||||
|
||||
def test_server_section(tmp_path: Path) -> None:
|
||||
config_file = write_config(tmp_path, f"""
|
||||
[backup]
|
||||
inputs={tmp_path}
|
||||
backup_dir=backups
|
||||
|
||||
[server]
|
||||
ssh_host=server
|
||||
ssh_user=user
|
||||
ssh_keyfile=/keys/id_ed25519
|
||||
remote_sudo=yes
|
||||
numeric_ids=true
|
||||
""")
|
||||
|
||||
config_args = sb._read_config(config_file, USER)
|
||||
|
||||
assert config_args['ssh_host'] == 'server'
|
||||
assert config_args['ssh_user'] == 'user'
|
||||
assert config_args['ssh_keyfile'] == '/keys/id_ed25519'
|
||||
assert config_args['remote_sudo'] is True
|
||||
assert config_args['numeric_ids'] is True
|
||||
|
||||
|
||||
def test_empty_server_values(tmp_path: Path) -> None:
|
||||
config_file = write_config(tmp_path, f"""
|
||||
[backup]
|
||||
inputs={tmp_path}
|
||||
backup_dir=/backups
|
||||
|
||||
[server]
|
||||
ssh_host=
|
||||
ssh_user=
|
||||
ssh_keyfile=
|
||||
remote_sudo=
|
||||
numeric_ids=
|
||||
""")
|
||||
|
||||
config_args = sb._read_config(config_file, USER)
|
||||
|
||||
assert config_args['ssh_host'] is None
|
||||
assert config_args['ssh_user'] is None
|
||||
assert config_args['ssh_keyfile'] is None
|
||||
assert config_args['remote_sudo'] is False
|
||||
assert config_args['numeric_ids'] is False
|
||||
|
||||
|
||||
def test_only_ssh_host(tmp_path: Path) -> None:
|
||||
config_file = write_config(tmp_path, f"""
|
||||
[backup]
|
||||
inputs={tmp_path}
|
||||
backup_dir=/backups
|
||||
|
||||
[server]
|
||||
ssh_host=server
|
||||
""")
|
||||
|
||||
config_args = sb._read_config(config_file, USER)
|
||||
|
||||
# Kept, so that the missing user is reported instead of silently backing up locally
|
||||
assert config_args['ssh_host'] == 'server'
|
||||
assert config_args['ssh_user'] is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize('line', ['keep=abc', 'remote_sudo=maybe'])
|
||||
def test_bad_value(tmp_path: Path, line: str) -> None:
|
||||
section = '[server]' if line.startswith('remote_sudo') else ''
|
||||
|
||||
config_file = write_config(tmp_path, f"""
|
||||
[backup]
|
||||
inputs={tmp_path}
|
||||
backup_dir=/backups
|
||||
{section}
|
||||
{line}
|
||||
""")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
sb._read_config(config_file, USER)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('text', [
|
||||
'[backup]\nbackup_dir=/backups\n',
|
||||
'[backup]\ninputs=/tmp\n',
|
||||
'[other]\ninputs=/tmp\n',
|
||||
'inputs=/tmp\n',
|
||||
])
|
||||
def test_bad_config(tmp_path: Path, text: str) -> None:
|
||||
config_file = write_config(tmp_path, text)
|
||||
|
||||
with pytest.raises(configparser.Error):
|
||||
sb._read_config(config_file, USER)
|
||||
@@ -0,0 +1,360 @@
|
||||
"""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
|
||||
@@ -0,0 +1,568 @@
|
||||
"""Tests for remote backups, with a fake SSH client that runs commands locally"""
|
||||
|
||||
import importlib.util
|
||||
import io
|
||||
import os
|
||||
import shlex
|
||||
import socket
|
||||
import subprocess
|
||||
import warnings
|
||||
from getpass import GetPassWarning
|
||||
from pathlib import Path
|
||||
from shutil import which
|
||||
from typing import Any, Optional
|
||||
|
||||
import pytest
|
||||
|
||||
from simple_backup import simple_backup as sb
|
||||
|
||||
OPTIONS = '-r -v --ignore-missing-args --mkpath -a -h -H -X -s'
|
||||
|
||||
needs_paramiko = pytest.mark.skipif(importlib.util.find_spec('paramiko') is None,
|
||||
reason='paramiko not installed')
|
||||
needs_ssh_keygen = pytest.mark.skipif(which('ssh-keygen') is None, reason='ssh-keygen not installed')
|
||||
|
||||
|
||||
class FakeSSH:
|
||||
"""SSHClient stand-in: runs commands with sh in a fake remote home"""
|
||||
|
||||
def __init__(self, home: Path, bin_dir: Path) -> None:
|
||||
self.home = home
|
||||
self.bin_dir = bin_dir
|
||||
self.commands: list[str] = []
|
||||
self.fail_on: Optional[str] = None
|
||||
self.closed = False
|
||||
|
||||
def exec_command(self, command: str) -> tuple[None, io.BytesIO, io.BytesIO]:
|
||||
self.commands.append(command)
|
||||
|
||||
if self.fail_on is not None and self.fail_on in command:
|
||||
return None, io.BytesIO(b''), io.BytesIO(b'Permission denied\n')
|
||||
|
||||
env = {'HOME': str(self.home), 'PATH': f'{self.bin_dir}:{os.environ["PATH"]}'}
|
||||
p = subprocess.run(['sh', '-c', command], cwd=self.home, env=env, capture_output=True, check=False)
|
||||
|
||||
return None, io.BytesIO(p.stdout), io.BytesIO(p.stderr)
|
||||
|
||||
def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def remote_home(tmp_path: Path) -> Path:
|
||||
home = tmp_path / 'remote home'
|
||||
(home / 'backups').mkdir(parents=True)
|
||||
|
||||
return home
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ssh(remote_home: Path, tmp_path: Path) -> FakeSSH:
|
||||
# A 'sudo' that just runs the command, for --remote-sudo
|
||||
bin_dir = tmp_path / 'bin'
|
||||
bin_dir.mkdir()
|
||||
sudo = bin_dir / 'sudo'
|
||||
sudo.write_text('#!/bin/sh\nexec "$@"\n')
|
||||
sudo.chmod(0o755)
|
||||
|
||||
return FakeSSH(remote_home, bin_dir)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def connect(monkeypatch: pytest.MonkeyPatch, ssh: FakeSSH) -> FakeSSH:
|
||||
monkeypatch.setattr(sb.Backup, '_ssh_connect', lambda self, homedir='': ssh)
|
||||
|
||||
return ssh
|
||||
|
||||
|
||||
def remote_backup(output: str = '~/backups', keep: int = -1, **kwargs: Any) -> sb.Backup:
|
||||
return sb.Backup(['/x'], output, [], keep, OPTIONS, ssh_host='server', ssh_user='user', **kwargs)
|
||||
|
||||
|
||||
def connected_backup(output: str = '~/backups', keep: int = -1, **kwargs: Any) -> sb.Backup:
|
||||
backup = remote_backup(output, keep, **kwargs)
|
||||
|
||||
assert backup.check_params() == 0
|
||||
|
||||
return backup
|
||||
|
||||
|
||||
def make_backups(remote_home: Path, *names: str) -> Path:
|
||||
base = remote_home / 'backups' / 'simple_backup'
|
||||
base.mkdir()
|
||||
|
||||
for name in names:
|
||||
(base / name).mkdir()
|
||||
|
||||
return base
|
||||
|
||||
|
||||
def remote_dirs(base: Path) -> list[str]:
|
||||
return sorted(d.name for d in base.iterdir() if d.is_dir())
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('connect')
|
||||
class TestRemotePath:
|
||||
@pytest.mark.parametrize('output', ['~/backups', 'backups', '~/backups/', '{home}/backups'])
|
||||
def test_resolved(self, remote_home: Path, output: str) -> None:
|
||||
backup = connected_backup(output.format(home=remote_home))
|
||||
|
||||
assert backup.output == f'{remote_home}/backups'
|
||||
|
||||
def test_home(self, remote_home: Path) -> None:
|
||||
assert connected_backup('~').output == str(remote_home)
|
||||
|
||||
def test_shell_characters_not_expanded(self, remote_home: Path) -> None:
|
||||
name = 'a $HOME `touch pwned` b'
|
||||
(remote_home / name).mkdir()
|
||||
|
||||
assert connected_backup(f'~/{name}').output == f'{remote_home}/{name}'
|
||||
assert not (remote_home / 'pwned').exists()
|
||||
|
||||
def test_missing(self, caplog: pytest.LogCaptureFixture) -> None:
|
||||
assert remote_backup('~/missing').check_params() == 2
|
||||
assert 'Output path for backup does not exist' in caplog.text
|
||||
|
||||
def test_needs_host_and_user(self, tmp_path: Path) -> None:
|
||||
# Only one of them: a local backup
|
||||
backup = sb.Backup(['/x'], str(tmp_path), [], -1, OPTIONS, ssh_host='server')
|
||||
|
||||
assert backup.check_params() == 0
|
||||
assert not backup._remote
|
||||
|
||||
|
||||
class TestConnectionErrors:
|
||||
def test_connection_failed(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sb.Backup, '_ssh_connect', lambda self, homedir='': None)
|
||||
|
||||
assert remote_backup().check_params() == 5
|
||||
|
||||
@pytest.mark.parametrize('error', [socket.gaierror(-2, 'Name or service not known'),
|
||||
ConnectionRefusedError(111, 'Connection refused'),
|
||||
TimeoutError('timed out')])
|
||||
def test_os_error(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture,
|
||||
error: OSError) -> None:
|
||||
def fail(self: sb.Backup, homedir: str = '') -> None:
|
||||
raise error
|
||||
|
||||
monkeypatch.setattr(sb.Backup, '_ssh_connect', fail)
|
||||
|
||||
assert remote_backup().check_params() == 5
|
||||
assert 'Can\'t connect to the server server' in caplog.text
|
||||
|
||||
def test_paramiko_missing(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
|
||||
monkeypatch.delattr(sb, 'paramiko', raising=False)
|
||||
|
||||
assert remote_backup().check_params() == 5
|
||||
assert 'Install paramiko' in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('connect')
|
||||
class TestRemoveOldBackups:
|
||||
def test_keep(self, remote_home: Path) -> None:
|
||||
base = make_backups(remote_home, '2026-10-01 10:00:00', '2026-10-02_10-00-00', '2026-10-03_10-00-00')
|
||||
(base / 'notes.txt').touch()
|
||||
|
||||
# The newest one stands for the backup just made
|
||||
connected_backup(keep=1).remove_old_backups()
|
||||
|
||||
assert remote_dirs(base) == ['2026-10-02_10-00-00', '2026-10-03_10-00-00']
|
||||
assert (base / 'notes.txt').exists()
|
||||
|
||||
def test_remove_before(self, remote_home: Path) -> None:
|
||||
base = make_backups(remote_home, '2026-10-01_10-00-00', '2026-10-02_10-00-00')
|
||||
|
||||
connected_backup(keep=0, remove_before=True).remove_old_backups()
|
||||
|
||||
assert remote_dirs(base) == []
|
||||
assert base.is_dir()
|
||||
|
||||
def test_remote_sudo(self, remote_home: Path, ssh: FakeSSH) -> None:
|
||||
base = make_backups(remote_home, '2026-10-01_10-00-00', '2026-10-02_10-00-00')
|
||||
|
||||
connected_backup(keep=0, remote_sudo=True).remove_old_backups()
|
||||
|
||||
assert remote_dirs(base) == ['2026-10-02_10-00-00']
|
||||
assert [c for c in ssh.commands if 'rm -r' in c][0].startswith('sudo rm -r ')
|
||||
|
||||
@pytest.mark.parametrize('names', [[], None])
|
||||
def test_nothing_to_remove(self, remote_home: Path, names: Optional[list[str]]) -> None:
|
||||
if names is not None:
|
||||
make_backups(remote_home, *names)
|
||||
|
||||
connected_backup(keep=0, remove_before=True).remove_old_backups()
|
||||
|
||||
# The backup folder itself must never be removed
|
||||
assert (remote_home / 'backups').is_dir()
|
||||
assert (remote_home / 'backups' / 'simple_backup').is_dir() == (names is not None)
|
||||
|
||||
def test_list_error(self, remote_home: Path, ssh: FakeSSH, caplog: pytest.LogCaptureFixture) -> None:
|
||||
base = make_backups(remote_home, '2026-10-01_10-00-00', '2026-10-02_10-00-00')
|
||||
backup = connected_backup(keep=0)
|
||||
ssh.fail_on = 'find'
|
||||
|
||||
backup.remove_old_backups()
|
||||
|
||||
assert remote_dirs(base) == ['2026-10-01_10-00-00', '2026-10-02_10-00-00']
|
||||
assert 'Cannot list old backups on the server' in caplog.text
|
||||
|
||||
def test_remove_error(self, remote_home: Path, ssh: FakeSSH, caplog: pytest.LogCaptureFixture) -> None:
|
||||
make_backups(remote_home, '2026-10-01_10-00-00', '2026-10-02_10-00-00')
|
||||
backup = connected_backup(keep=0)
|
||||
ssh.fail_on = 'rm -r'
|
||||
|
||||
backup.remove_old_backups()
|
||||
|
||||
assert 'Error while removing backup 2026-10-01_10-00-00' in caplog.text
|
||||
assert backup._removed_count == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('connect')
|
||||
class TestFindLastBackup:
|
||||
def test_newest(self, remote_home: Path) -> None:
|
||||
base = make_backups(remote_home, '2026-10-01 10:00:00', '2026-10-02_10-00-00')
|
||||
(base / 'zzz.txt').touch()
|
||||
backup = connected_backup()
|
||||
|
||||
backup.find_last_backup()
|
||||
|
||||
assert backup._last_backup == f'{base}/2026-10-02_10-00-00'
|
||||
|
||||
@pytest.mark.parametrize('create', [True, False])
|
||||
def test_none(self, remote_home: Path, caplog: pytest.LogCaptureFixture, create: bool) -> None:
|
||||
if create:
|
||||
make_backups(remote_home)
|
||||
|
||||
backup = connected_backup()
|
||||
backup.find_last_backup()
|
||||
|
||||
assert backup._last_backup == ''
|
||||
assert 'No previous backups available' in caplog.text
|
||||
|
||||
def test_no_connection(self) -> None:
|
||||
backup = connected_backup()
|
||||
backup._ssh = None
|
||||
|
||||
with pytest.raises(SystemExit) as e:
|
||||
backup.find_last_backup()
|
||||
|
||||
assert e.value.code == 5
|
||||
|
||||
|
||||
class RecordPopen:
|
||||
"""Records the rsync arguments instead of running rsync"""
|
||||
|
||||
returncode = 0
|
||||
calls: list[list[str]] = []
|
||||
|
||||
def __init__(self, args: list[str], **kwargs: Any) -> None:
|
||||
RecordPopen.calls.append(args)
|
||||
self.stdin = io.BytesIO()
|
||||
self.stdout = io.BytesIO(b'sent 1 bytes received 2 bytes\ntotal size is 3\n')
|
||||
|
||||
def __enter__(self) -> 'RecordPopen':
|
||||
return self
|
||||
|
||||
def __exit__(self, *args: Any) -> None:
|
||||
pass
|
||||
|
||||
def wait(self) -> int:
|
||||
return RecordPopen.returncode
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('connect')
|
||||
class TestRsyncArgs:
|
||||
@pytest.fixture(autouse=True)
|
||||
def record_popen(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
RecordPopen.returncode = 0
|
||||
RecordPopen.calls = []
|
||||
monkeypatch.setattr(sb, 'Popen', RecordPopen)
|
||||
# The input must exist locally
|
||||
(tmp_path / 'input').mkdir()
|
||||
|
||||
def run(self, tmp_path: Path, **kwargs: Any) -> tuple[int, list[str]]:
|
||||
backup = connected_backup(**kwargs)
|
||||
backup.inputs = [str(tmp_path / 'input')]
|
||||
returncode = backup.run()
|
||||
(args,) = RecordPopen.calls
|
||||
|
||||
return returncode, args
|
||||
|
||||
@staticmethod
|
||||
def ssh_command(args: list[str]) -> str:
|
||||
return args[args.index('-e') + 1]
|
||||
|
||||
def test_default(self, tmp_path: Path, remote_home: Path) -> None:
|
||||
returncode, args = self.run(tmp_path)
|
||||
|
||||
assert returncode == 0
|
||||
assert args[args.index('/') + 1].startswith(f'user@server:{remote_home}/backups/simple_backup/')
|
||||
assert self.ssh_command(args) == 'ssh -o StrictHostKeyChecking=accept-new'
|
||||
assert not any(a.startswith('--rsync-path') for a in args)
|
||||
assert not any(a.startswith('--link-dest') for a in args)
|
||||
|
||||
def test_link_dest(self, tmp_path: Path, remote_home: Path) -> None:
|
||||
base = make_backups(remote_home, '2026-10-01_10-00-00')
|
||||
|
||||
_, args = self.run(tmp_path)
|
||||
|
||||
assert f'--link-dest={base}/2026-10-01_10-00-00' in args
|
||||
|
||||
def test_keyfile_with_spaces(self, tmp_path: Path) -> None:
|
||||
_, args = self.run(tmp_path, ssh_keyfile='/keys/my key')
|
||||
|
||||
# rsync splits the -e command like a shell
|
||||
assert shlex.split(self.ssh_command(args)) == ['ssh', '-i', '/keys/my key', '-o',
|
||||
'StrictHostKeyChecking=accept-new']
|
||||
|
||||
def test_remote_sudo(self, tmp_path: Path) -> None:
|
||||
_, args = self.run(tmp_path, remote_sudo=True)
|
||||
|
||||
assert '--rsync-path=sudo rsync' in args
|
||||
|
||||
def test_password_auth(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sb, 'which', lambda cmd: f'/usr/bin/{cmd}')
|
||||
monkeypatch.setenv('SSHPASS', 'secret')
|
||||
backup = connected_backup()
|
||||
backup.inputs = [str(tmp_path / 'input')]
|
||||
backup._password_auth = True
|
||||
|
||||
assert backup.run() == 0
|
||||
|
||||
(args,) = RecordPopen.calls
|
||||
assert shlex.split(self.ssh_command(args)) == ['sshpass', '-e', 'ssh', '-l', 'user', '-o',
|
||||
'StrictHostKeyChecking=accept-new']
|
||||
# The password must not stay in the environment
|
||||
assert 'SSHPASS' not in os.environ
|
||||
|
||||
def test_failure(self, tmp_path: Path, remote_home: Path) -> None:
|
||||
base = make_backups(remote_home, '2026-10-01_10-00-00')
|
||||
RecordPopen.returncode = 23
|
||||
|
||||
returncode, _ = self.run(tmp_path, keep=0)
|
||||
|
||||
assert returncode == 4
|
||||
assert remote_dirs(base) == ['2026-10-01_10-00-00']
|
||||
|
||||
|
||||
class FakeClient:
|
||||
"""paramiko.SSHClient stand-in; connect() takes its outcomes from a list"""
|
||||
|
||||
def __init__(self, outcomes: list[Optional[BaseException]]) -> None:
|
||||
self.outcomes = outcomes
|
||||
self.connects: list[dict[str, Any]] = []
|
||||
self.policies: list[str] = []
|
||||
self.known_hosts: Optional[str] = None
|
||||
self.closed = False
|
||||
|
||||
def load_host_keys(self, filename: str) -> None:
|
||||
self.known_hosts = filename
|
||||
|
||||
if not os.path.isfile(filename):
|
||||
raise FileNotFoundError(filename)
|
||||
|
||||
def set_missing_host_key_policy(self, policy: Any) -> None:
|
||||
self.policies.append(type(policy).__name__)
|
||||
|
||||
def connect(self, hostname: str, **kwargs: Any) -> None:
|
||||
self.connects.append(kwargs)
|
||||
outcome = self.outcomes.pop(0)
|
||||
|
||||
if isinstance(outcome, UserWarning):
|
||||
# Like WarningPolicy for an unknown host key
|
||||
warnings.warn(outcome)
|
||||
elif outcome is not None:
|
||||
raise outcome
|
||||
|
||||
def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(monkeypatch: pytest.MonkeyPatch) -> FakeClient:
|
||||
fake = FakeClient([])
|
||||
monkeypatch.setattr(sb.paramiko, 'SSHClient', lambda: fake)
|
||||
|
||||
def no_terminal(prompt: str = '') -> str:
|
||||
raise AssertionError(f'Unexpected prompt: {prompt}')
|
||||
|
||||
monkeypatch.setattr(sb, 'getpass', no_terminal)
|
||||
monkeypatch.setattr('builtins.input', no_terminal)
|
||||
|
||||
return fake
|
||||
|
||||
|
||||
def ssh_error() -> Exception:
|
||||
return sb.paramiko.SSHException('Authentication failed')
|
||||
|
||||
|
||||
@needs_paramiko
|
||||
class TestLogin:
|
||||
def test_agent(self, client: FakeClient, tmp_path: Path) -> None:
|
||||
client.outcomes = [None]
|
||||
|
||||
assert remote_backup()._ssh_connect(str(tmp_path)) is client
|
||||
assert client.known_hosts == f'{tmp_path}/.ssh/known_hosts'
|
||||
assert client.connects[0]['timeout'] == sb.SSH_TIMEOUT
|
||||
assert not client.closed
|
||||
|
||||
@pytest.mark.parametrize('answer, connected', [('y', True), (' Yes', True), ('n', False), ('', False)])
|
||||
def test_unknown_host_key(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
|
||||
answer: str, connected: bool) -> None:
|
||||
client.outcomes = [UserWarning('Unknown host key'), None]
|
||||
monkeypatch.setattr('builtins.input', lambda prompt: answer)
|
||||
|
||||
result = remote_backup()._ssh_connect()
|
||||
|
||||
assert (result is client) == connected
|
||||
assert client.closed != connected
|
||||
|
||||
if connected:
|
||||
assert client.policies == ['WarningPolicy', 'AutoAddPolicy']
|
||||
|
||||
def test_unknown_host_key_no_terminal(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
def eof(prompt: str) -> str:
|
||||
raise EOFError
|
||||
|
||||
client.outcomes = [UserWarning('Unknown host key')]
|
||||
monkeypatch.setattr('builtins.input', eof)
|
||||
|
||||
assert remote_backup()._ssh_connect() is None
|
||||
assert client.closed
|
||||
|
||||
def test_bad_host_key(self, client: FakeClient, caplog: pytest.LogCaptureFixture) -> None:
|
||||
class Key:
|
||||
def get_base64(self) -> str:
|
||||
return 'AAAA'
|
||||
|
||||
client.outcomes = [sb.paramiko.BadHostKeyException('server', Key(), Key())]
|
||||
|
||||
assert remote_backup()._ssh_connect() is None
|
||||
assert client.closed
|
||||
assert 'Can\'t connect to the server' in caplog.text
|
||||
|
||||
def test_password(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
client.outcomes = [ssh_error(), ssh_error(), None]
|
||||
monkeypatch.setattr(sb, 'getpass', lambda prompt: 'secret')
|
||||
monkeypatch.setenv('SSHPASS', '')
|
||||
backup = remote_backup()
|
||||
|
||||
assert backup._ssh_connect() is client
|
||||
assert client.connects[-1]['password'] == 'secret'
|
||||
assert backup._password_auth
|
||||
assert os.environ['SSHPASS'] == 'secret'
|
||||
|
||||
def test_wrong_password(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
client.outcomes = [ssh_error(), ssh_error(), ssh_error()]
|
||||
monkeypatch.setattr(sb, 'getpass', lambda prompt: 'wrong')
|
||||
|
||||
assert remote_backup()._ssh_connect() is None
|
||||
assert client.closed
|
||||
|
||||
def test_password_no_terminal(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
|
||||
caplog: pytest.LogCaptureFixture) -> None:
|
||||
def no_tty(prompt: str) -> str:
|
||||
warnings.warn('Can not control echo on the terminal.', GetPassWarning)
|
||||
|
||||
return ''
|
||||
|
||||
client.outcomes = [ssh_error(), ssh_error()]
|
||||
monkeypatch.setattr(sb, 'getpass', no_tty)
|
||||
|
||||
assert remote_backup()._ssh_connect() is None
|
||||
assert 'Unable to get password' in caplog.text
|
||||
|
||||
@pytest.mark.parametrize('outcome, connected', [(None, True), ('error', False)])
|
||||
def test_keyfile(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
|
||||
outcome: Optional[str], connected: bool) -> None:
|
||||
pkey = object()
|
||||
client.outcomes = [ssh_error(), ssh_error(), ssh_error() if outcome else None]
|
||||
monkeypatch.setattr(sb.Backup, '_load_ssh_key', lambda self: pkey)
|
||||
|
||||
assert (remote_backup(ssh_keyfile='/keys/id')._ssh_connect() is client) == connected
|
||||
assert client.connects[-1]['pkey'] is pkey
|
||||
|
||||
def test_keyfile_not_loaded(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
client.outcomes = [ssh_error(), ssh_error()]
|
||||
monkeypatch.setattr(sb.Backup, '_load_ssh_key', lambda self: None)
|
||||
|
||||
assert remote_backup(ssh_keyfile='/keys/id')._ssh_connect() is None
|
||||
assert len(client.connects) == 2
|
||||
assert client.closed
|
||||
|
||||
|
||||
@pytest.fixture(scope='module')
|
||||
def keys(tmp_path_factory: pytest.TempPathFactory) -> Path:
|
||||
"""Real keys made by ssh-keygen; encrypted ones use the passphrase 'secret'"""
|
||||
|
||||
key_dir = tmp_path_factory.mktemp('keys')
|
||||
key_types = {'ed25519': ['-t', 'ed25519'], 'ecdsa': ['-t', 'ecdsa'],
|
||||
'rsa': ['-t', 'rsa', '-b', '2048'], 'rsa_pem': ['-t', 'rsa', '-b', '2048', '-m', 'PEM']}
|
||||
|
||||
for name, args in key_types.items():
|
||||
for suffix, passphrase in [('', ''), ('_enc', 'secret')]:
|
||||
subprocess.run(['ssh-keygen', '-q', *args, '-N', passphrase, '-f', str(key_dir / f'{name}{suffix}')],
|
||||
check=True)
|
||||
|
||||
return key_dir
|
||||
|
||||
|
||||
@needs_paramiko
|
||||
@needs_ssh_keygen
|
||||
class TestLoadKey:
|
||||
@pytest.fixture(autouse=True)
|
||||
def prompts(self, monkeypatch: pytest.MonkeyPatch) -> list[str]:
|
||||
asked: list[str] = []
|
||||
|
||||
def answer(prompt: str) -> str:
|
||||
asked.append(prompt)
|
||||
|
||||
return self.passphrase
|
||||
|
||||
self.passphrase = 'secret'
|
||||
monkeypatch.setattr(sb, 'getpass', answer)
|
||||
|
||||
return asked
|
||||
|
||||
@pytest.mark.parametrize('name', ['ed25519', 'ecdsa', 'rsa', 'rsa_pem'])
|
||||
def test_unencrypted(self, keys: Path, prompts: list[str], name: str) -> None:
|
||||
key = remote_backup(ssh_keyfile=str(keys / name))._load_ssh_key()
|
||||
|
||||
assert isinstance(key, sb.paramiko.PKey)
|
||||
assert prompts == []
|
||||
|
||||
@pytest.mark.parametrize('name', ['ed25519_enc', 'ecdsa_enc', 'rsa_enc', 'rsa_pem_enc'])
|
||||
def test_encrypted(self, keys: Path, prompts: list[str], name: str) -> None:
|
||||
key = remote_backup(ssh_keyfile=str(keys / name))._load_ssh_key()
|
||||
|
||||
assert isinstance(key, sb.paramiko.PKey)
|
||||
assert len(prompts) == 1
|
||||
|
||||
@pytest.mark.parametrize('name', ['ed25519_enc', 'rsa_pem_enc'])
|
||||
def test_wrong_passphrase(self, keys: Path, caplog: pytest.LogCaptureFixture, name: str) -> None:
|
||||
self.passphrase = 'wrong'
|
||||
|
||||
assert remote_backup(ssh_keyfile=str(keys / name))._load_ssh_key() is None
|
||||
assert 'wrong passphrase?' in caplog.text
|
||||
|
||||
def test_no_terminal(self, keys: Path, monkeypatch: pytest.MonkeyPatch,
|
||||
caplog: pytest.LogCaptureFixture) -> None:
|
||||
def no_tty(prompt: str) -> str:
|
||||
warnings.warn('Can not control echo on the terminal.', GetPassWarning)
|
||||
|
||||
return ''
|
||||
|
||||
monkeypatch.setattr(sb, 'getpass', no_tty)
|
||||
|
||||
assert remote_backup(ssh_keyfile=str(keys / 'ed25519_enc'))._load_ssh_key() is None
|
||||
assert 'Unable to get passphrase' in caplog.text
|
||||
|
||||
@pytest.mark.parametrize('content', [None, 'not a key\n'])
|
||||
def test_bad_file(self, tmp_path: Path, caplog: pytest.LogCaptureFixture, content: Optional[str]) -> None:
|
||||
keyfile = tmp_path / 'id'
|
||||
|
||||
if content is not None:
|
||||
keyfile.write_text(content)
|
||||
|
||||
assert remote_backup(ssh_keyfile=str(keyfile))._load_ssh_key() is None
|
||||
assert 'Unable to load SSH key' in caplog.text
|
||||
Reference in New Issue
Block a user