Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b65955f9f6
|
||
|
|
6d54019a18
|
||
|
|
7547db1ea8
|
||
|
|
37e87f31b5
|
||
|
|
4ac8b64085
|
||
|
|
34e3e41382
|
||
|
|
0ffb5c47ac
|
||
|
|
59a3f94ec8
|
||
|
|
408beefc66
|
||
|
|
b54f7e1817
|
||
|
|
9e6aadf7d3
|
||
|
|
ac59969d82
|
||
|
|
db4059ff36
|
||
|
|
dc50ef1b07
|
||
|
|
fa3ba2a6bd
|
||
|
|
e8b088f13e
|
||
|
|
535f7a33fe
|
||
|
|
85bce54b05
|
||
|
|
125b77a839
|
||
|
|
17c104c6c8
|
||
|
|
7dbb1bcaee
|
||
|
|
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
|
@@ -1,3 +1,4 @@
|
|||||||
|
CLAUDE.md
|
||||||
.idea/
|
.idea/
|
||||||
__pycache__/
|
__pycache__/
|
||||||
test/
|
test/
|
||||||
|
|||||||
@@ -4,9 +4,9 @@
|
|||||||
|
|
||||||
pkgname=simple_backup
|
pkgname=simple_backup
|
||||||
pkgdesc='Simple backup script that uses rsync to copy files'
|
pkgdesc='Simple backup script that uses rsync to copy files'
|
||||||
pkgver=4.1.7
|
pkgver=4.5.0
|
||||||
pkgrel=1
|
pkgrel=1
|
||||||
url="https://git.shouldnt.work/fuxino/${pkgname}"
|
url="https://git.shouldnt.work/fuxino/$pkgname"
|
||||||
arch=('any')
|
arch=('any')
|
||||||
license=('GPL-3.0-or-later')
|
license=('GPL-3.0-or-later')
|
||||||
makedepends=('git'
|
makedepends=('git'
|
||||||
@@ -18,21 +18,31 @@ depends=('python>=3.10'
|
|||||||
optdepends=('python-systemd: use systemd log'
|
optdepends=('python-systemd: use systemd log'
|
||||||
'python-dbus: for desktop notifications'
|
'python-dbus: for desktop notifications'
|
||||||
'python-paramiko: for remote backup through ssh')
|
'python-paramiko: for remote backup through ssh')
|
||||||
|
checkdepends=('python-pytest'
|
||||||
|
'python-paramiko'
|
||||||
|
'openssh')
|
||||||
conflicts=('simple_backup-git')
|
conflicts=('simple_backup-git')
|
||||||
source=(git+${url}?signed#tag=${pkgver})
|
source=("git+$url?signed#tag=$pkgver")
|
||||||
validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
|
validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
|
||||||
sha256sums=('011cd5546745e413ac7603b64531def2688c2bf6e4ea19876a67547a1db208ce')
|
sha256sums=('c0f7c38d99208e1327c0a6af1d58d5d13797d1cef54f99e1b2fcb4f4a9480609')
|
||||||
|
|
||||||
build()
|
build()
|
||||||
{
|
{
|
||||||
cd ${srcdir}/${pkgname}
|
cd "$srcdir"/"$pkgname"
|
||||||
python -m build --wheel --no-isolation
|
python -m build --wheel --no-isolation
|
||||||
}
|
}
|
||||||
|
|
||||||
|
check()
|
||||||
|
{
|
||||||
|
cd "$srcdir"/"$pkgname"
|
||||||
|
python -m pytest
|
||||||
|
}
|
||||||
|
|
||||||
package()
|
package()
|
||||||
{
|
{
|
||||||
cd ${srcdir}/${pkgname}
|
cd "$srcdir"/"$pkgname"
|
||||||
python -m installer --destdir=${pkgdir} dist/*.whl
|
python -m installer --destdir="$pkgdir" dist/*.whl
|
||||||
install -Dm644 ${pkgname}/${pkgname}.conf ${pkgdir}/usr/share/doc/${pkgname}/${pkgname}.conf
|
install -Dm644 "$pkgname"/"$pkgname".conf "$pkgdir"/usr/share/doc/"$pkgname"/"$pkgname".conf
|
||||||
install -Dm644 man/${pkgname}.1 ${pkgdir}/usr/share/man/man1/${pkgname}.1
|
install -Dm644 README.md "$pkgdir"/usr/share/doc/"$pkgname"/README.md
|
||||||
|
install -Dm644 man/"$pkgname".1 "$pkgdir"/usr/share/man/man1/"$pkgname".1
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -20,44 +20,74 @@ The script uses rsync to actually run the backup, so you will have to install it
|
|||||||
sudo pacman -Syu rsync
|
sudo pacman -Syu rsync
|
||||||
```
|
```
|
||||||
|
|
||||||
Optional dependencies are systemd-python to enable using systemd journal for logging, dbus-python for desktop notifications, and paramiko for remote backup.
|
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
|
## Installation
|
||||||
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
|
||||||
|
|
||||||
|
Install one with your AUR helper, e.g. `aur sync simple_backup` with aurutils.
|
||||||
|
|
||||||
|
### Other distributions
|
||||||
|
First clone the repository:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
git clone https://git.shouldnt.work/fuxino/simple_backup.git
|
git clone https://git.shouldnt.work/fuxino/simple_backup.git
|
||||||
```
|
```
|
||||||
|
|
||||||
Then install the tools required to build the package:
|
Then build the package (this requires the Python `build` module, e.g. the python-build package on Arch Linux):
|
||||||
|
|
||||||
```bash
|
|
||||||
pip install --upgrade build
|
|
||||||
```
|
|
||||||
|
|
||||||
Finally, run:
|
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
cd simple_backup
|
cd simple_backup
|
||||||
python -m build --wheel
|
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):
|
Many distributions don't allow installing packages with pip into the system Python ("externally-managed-environment" error).
|
||||||
- **simple_backup** for the release version
|
In that case, install the wheel in a virtual environment, for example with pipx:
|
||||||
- **simple_backup-git** for the git version
|
|
||||||
|
```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
|
## Old backups
|
||||||
The number of old backups to keep can be set with the --keep (or -k) argument (or in the configuration file).
|
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.
|
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.
|
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.
|
||||||
|
|
||||||
|
## Last backup
|
||||||
|
After each successful backup, simple_backup saves its time and directory to `~/.local/state/simple_backup/last_backup` (or `$XDG_STATE_HOME/simple_backup/last_backup`).
|
||||||
|
When run with sudo or doas, the file goes to the home directory of the user who ran the command.
|
||||||
|
Print it with:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
simple_backup --last-backup
|
||||||
|
```
|
||||||
|
|
||||||
|
The file has two lines: the completion time in ISO 8601 format (e.g. `2026-10-25T14:03:12+02:00`) and the backup directory, so scripts (e.g. a status bar indicator) can easily read it to remind you when a backup is overdue.
|
||||||
|
|
||||||
## Remote backup
|
## Remote backup
|
||||||
> **Warning**
|
> **Warning**
|
||||||
> This feature is experimental
|
> This feature is experimental
|
||||||
|
|
||||||
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).
|
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.
|
For this to work, rsync must be installed on the server too.
|
||||||
|
The host can be an alias from `~/.ssh/config`: its HostName, Port and IdentityFile are used (ProxyJump and ProxyCommand are not supported).
|
||||||
|
|
||||||
|
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
|
### 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.
|
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.
|
||||||
|
|||||||
+36
-4
@@ -1,4 +1,4 @@
|
|||||||
.TH SIMPLE_BACKUP 1 2026-10-08 SIMPLE_BACKUP 4.2.0
|
.TH SIMPLE_BACKUP 1 2026-10-25 SIMPLE_BACKUP 4.6.0
|
||||||
.SH NAME
|
.SH NAME
|
||||||
simple_backup \- Backup files and folders using rsync
|
simple_backup \- Backup files and folders using rsync
|
||||||
.SH SYNOPSIS
|
.SH SYNOPSIS
|
||||||
@@ -105,6 +105,12 @@ according to \-\-keep will be left, and the new backup may be incomplete. In par
|
|||||||
.B \-\-no\-syslog
|
.B \-\-no\-syslog
|
||||||
Don't use systemd journal for logging.
|
Don't use systemd journal for logging.
|
||||||
.TP
|
.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...]
|
.B \-\-rsync\-options OPTIONS [OPTION...]
|
||||||
By default, the following rsync options are used:
|
By default, the following rsync options are used:
|
||||||
.RS
|
.RS
|
||||||
@@ -139,6 +145,11 @@ needs to be allowed to run without password in the same way.
|
|||||||
.TP
|
.TP
|
||||||
.B \-\-numeric\-ids
|
.B \-\-numeric\-ids
|
||||||
Use rsync \-\-numeric\-ids option. This causes rsync to use numeric uid/gid instead of trying to map uid/gid names from the local machine to the server.
|
Use rsync \-\-numeric\-ids option. This causes rsync to use numeric uid/gid instead of trying to map uid/gid names from the local machine to the server.
|
||||||
|
.TP
|
||||||
|
.B \-\-last\-backup
|
||||||
|
Print the date and directory of the last successful backup, then exit. The exit status is 0, or 1
|
||||||
|
if no backup has been recorded yet or the state file can\(aqt be read (see
|
||||||
|
.BR FILES ).
|
||||||
.SH CONFIGURATION
|
.SH CONFIGURATION
|
||||||
An example configuration file is provided at \(aq/usr/share/doc/simple_backup/simple_backup.conf\(aq.
|
An example configuration file is provided at \(aq/usr/share/doc/simple_backup/simple_backup.conf\(aq.
|
||||||
Copy it to the default location ($HOME/.config/simple_backup) and edit it as needed.
|
Copy it to the default location ($HOME/.config/simple_backup) and edit it as needed.
|
||||||
@@ -146,6 +157,16 @@ 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
|
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
|
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).
|
configuration file, or on the command line (\(aq\-\-ssh\-host\(aq and \(aq\-\-ssh\-user\(aq options).
|
||||||
|
The host can be an alias from \fI~/.ssh/config\fR: its HostName, Port and IdentityFile are used
|
||||||
|
(ProxyJump and ProxyCommand are not supported).
|
||||||
|
.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
|
.SS AUTHENTICATION
|
||||||
For authentication, it is possible to use SSH key or password.
|
For authentication, it is possible to use SSH key or password.
|
||||||
.P
|
.P
|
||||||
@@ -180,7 +201,8 @@ for details. For this reason, use SSH key authentication if possible.
|
|||||||
.SH EXIT STATUS
|
.SH EXIT STATUS
|
||||||
.TP
|
.TP
|
||||||
.B 0
|
.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
|
.TP
|
||||||
.B 1
|
.B 1
|
||||||
No valid inputs selected for backup.
|
No valid inputs selected for backup.
|
||||||
@@ -192,13 +214,23 @@ Backup failed because output directory for storing the backup does not exist.
|
|||||||
Permission denied to access the output directory.
|
Permission denied to access the output directory.
|
||||||
.TP
|
.TP
|
||||||
.B 4
|
.B 4
|
||||||
rsync error (rsync returned a non-zero value).
|
rsync error (rsync returned a value other than 0 or 24, or rsync is not installed).
|
||||||
.TP
|
.TP
|
||||||
.B 5
|
.B 5
|
||||||
SSH connection failed.
|
SSH connection failed.
|
||||||
.TP
|
.TP
|
||||||
.B 6
|
.B 6
|
||||||
Bad configuration file.
|
Bad configuration file or options (for example, only one of SSH host and SSH user specified).
|
||||||
|
.SH FILES
|
||||||
|
.TP
|
||||||
|
.I $XDG_STATE_HOME/simple_backup/last_backup
|
||||||
|
Written after each successful backup (local or remote), replacing the previous one. The first line
|
||||||
|
is the completion time in ISO 8601 format with the UTC offset (for example
|
||||||
|
\(aq2026\-10\-25T14:03:12+02:00\(aq), the second line the backup directory (prefixed with
|
||||||
|
\(aquser@host:\(aq for remote backups). The default location is
|
||||||
|
\fI~/.local/state/simple_backup/last_backup\fR. When running as root with sudo, doas or \-\-user,
|
||||||
|
the file is written to that user\(aqs \fI~/.local/state\fR and owned by them. Useful for scripts
|
||||||
|
that remind you when a backup is overdue.
|
||||||
.SH SEE ALSO
|
.SH SEE ALSO
|
||||||
.BR rsync (1)
|
.BR rsync (1)
|
||||||
.SH AUTHORS
|
.SH AUTHORS
|
||||||
|
|||||||
+22
-1
@@ -33,7 +33,7 @@ classifiers = [
|
|||||||
[project.optional-dependencies]
|
[project.optional-dependencies]
|
||||||
JOURNAL = ["systemd-python"]
|
JOURNAL = ["systemd-python"]
|
||||||
NOTIFICATIONS = ["dbus-python"]
|
NOTIFICATIONS = ["dbus-python"]
|
||||||
REMOTE = ["paramiko"]
|
REMOTE = ["paramiko>=3.2"]
|
||||||
|
|
||||||
[project.urls]
|
[project.urls]
|
||||||
Homepage = "https://git.shouldnt.work/fuxino/simple_backup"
|
Homepage = "https://git.shouldnt.work/fuxino/simple_backup"
|
||||||
@@ -46,3 +46,24 @@ packages = ["simple_backup"]
|
|||||||
|
|
||||||
[tool.setuptools.dynamic]
|
[tool.setuptools.dynamic]
|
||||||
version = { attr = "simple_backup.__version__" }
|
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.ruff]
|
||||||
|
line-length = 120
|
||||||
|
|
||||||
|
[tool.ruff.lint]
|
||||||
|
select = ["E", "F", "W", "I", "B", "UP"]
|
||||||
|
|
||||||
|
[tool.pytest.ini_options]
|
||||||
|
testpaths = ["tests"]
|
||||||
|
# Import the repo copy, not an installed one
|
||||||
|
pythonpath = ["."]
|
||||||
|
|||||||
@@ -1,3 +1,3 @@
|
|||||||
"""Init."""
|
"""Init."""
|
||||||
|
|
||||||
__version__ = '4.2.0'
|
__version__ = '4.6.0'
|
||||||
|
|||||||
@@ -4,7 +4,7 @@
|
|||||||
# 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).
|
# 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/user
|
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
|
backup_dir=/media/Backup
|
||||||
|
|
||||||
# Files, directories and patterns to exclude from the backup. Multiple items can be separated using a comma.
|
# 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_host=
|
||||||
# ssh_user=
|
# ssh_user=
|
||||||
# ssh_keyfile=
|
# ssh_keyfile=
|
||||||
# remote_sudo=
|
# remote_sudo=false
|
||||||
# numeric_ids=
|
# numeric_ids=false
|
||||||
|
|||||||
+625
-318
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,54 @@
|
|||||||
|
"""Shared fixtures for the simple_backup tests"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
from collections.abc import Callable, Iterator
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from simple_backup import simple_backup as sb
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def no_side_effects(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||||
|
"""Keep tests out of the systemd journal, the desktop notifications and the user's state directory"""
|
||||||
|
|
||||||
|
if hasattr(sb, 'j_handler'):
|
||||||
|
sb.logger.removeHandler(sb.j_handler)
|
||||||
|
|
||||||
|
monkeypatch.setattr(sb, '_notify', lambda text: None)
|
||||||
|
monkeypatch.setenv('XDG_STATE_HOME', str(tmp_path / 'state'))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
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 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,388 @@
|
|||||||
|
"""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) -> 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: 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: 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
|
||||||
|
|
||||||
|
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
|
||||||
@@ -0,0 +1,243 @@
|
|||||||
|
"""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_glob_characters_in_name(tmp_path: Path) -> None:
|
||||||
|
(tmp_path / 'Photos [2020]').mkdir()
|
||||||
|
(tmp_path / 'Photos 2').mkdir()
|
||||||
|
|
||||||
|
assert sb._expand_inputs([f'{tmp_path}/Photos [2020]'], USER) == [f'{tmp_path}/Photos [2020]']
|
||||||
|
|
||||||
|
|
||||||
|
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)
|
||||||
|
|
||||||
|
|
||||||
|
def test_inputs_and_backup_dir_optional(tmp_path: Path) -> None:
|
||||||
|
config_args = sb._read_config(write_config(tmp_path, '[backup]\nkeep=2\n'), USER)
|
||||||
|
|
||||||
|
assert config_args['inputs'] is None
|
||||||
|
assert config_args['output'] is None
|
||||||
|
assert config_args['keep'] == 2
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize('text', [
|
||||||
|
'[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,396 @@
|
|||||||
|
"""Tests for local backups, with real rsync runs and with a fake Popen"""
|
||||||
|
|
||||||
|
import errno
|
||||||
|
import io
|
||||||
|
import os
|
||||||
|
from collections.abc import Callable
|
||||||
|
from pathlib import Path
|
||||||
|
from shutil import which
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from simple_backup import simple_backup as sb
|
||||||
|
|
||||||
|
OPTIONS = '-r -v --ignore-missing-args --mkpath -a -h -H -X -s'
|
||||||
|
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.usefixtures('fake_now')
|
||||||
|
|
||||||
|
|
||||||
|
@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: list[str] | None = None,
|
||||||
|
remove_before: bool = False, progress: bool = False, verbose: bool = False) -> sb.Backup:
|
||||||
|
backup = sb.Backup(inputs, str(out), exclude or [], keep, OPTIONS, remove_before=remove_before,
|
||||||
|
progress=progress, verbose=verbose)
|
||||||
|
|
||||||
|
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_removal_os_error(self, src: Path, out: Path, monkeypatch: pytest.MonkeyPatch,
|
||||||
|
caplog: pytest.LogCaptureFixture) -> None:
|
||||||
|
assert run_backup(out, [str(src)]) == 0
|
||||||
|
|
||||||
|
def busy(path: str) -> None:
|
||||||
|
raise OSError(errno.EBUSY, os.strerror(errno.EBUSY), path)
|
||||||
|
|
||||||
|
monkeypatch.setattr(sb, 'rmtree', busy)
|
||||||
|
|
||||||
|
assert run_backup(out, [str(src)], keep=0) == 0
|
||||||
|
assert backups(out) == ['2026-10-14_10-00-00', '2026-10-14_11-00-00']
|
||||||
|
assert 'Error while removing backup 2026-10-14_10-00-00. Device or resource busy' in caplog.text
|
||||||
|
|
||||||
|
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
|
||||||
|
|
||||||
|
# The fake clock also advances when the state file is written
|
||||||
|
assert backups(out) == ['2026-10-14_12-00-00']
|
||||||
|
state = (tmp_path / 'state' / 'simple_backup' / 'last_backup').read_text().splitlines()
|
||||||
|
assert state[0].startswith('2026-10-14T13:00:00')
|
||||||
|
assert state[1] == f'{out}/simple_backup/2026-10-14_12-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]] = []
|
||||||
|
output = b'sending incremental file list\nsent 1 bytes received 2 bytes\ntotal size is 3\n'
|
||||||
|
|
||||||
|
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(FakePopen.output)
|
||||||
|
|
||||||
|
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] == which('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('lines, logged', [
|
||||||
|
([f'file{n}' for n in range(5)] + ['sent 1 bytes', 'total size is 3'], ['sent 1 bytes', 'total size is 3']),
|
||||||
|
([], []),
|
||||||
|
(['total size is 3'], ['total size is 3']),
|
||||||
|
])
|
||||||
|
def test_summary_logged(self, src: Path, out: Path, fake_popen: type[FakePopen], monkeypatch: pytest.MonkeyPatch,
|
||||||
|
caplog: pytest.LogCaptureFixture, lines: list[str], logged: list[str]) -> None:
|
||||||
|
monkeypatch.setattr(fake_popen, 'output', ''.join(f'{line}\n' for line in lines).encode())
|
||||||
|
|
||||||
|
assert run_backup(out, [str(src)]) == 0
|
||||||
|
assert [m[7:] for m in caplog.messages if m.startswith('rsync: ')] == logged
|
||||||
|
|
||||||
|
@pytest.mark.parametrize('returncode, level', [(0, 'INFO'), (23, 'WARNING'), (1, 'ERROR')])
|
||||||
|
def test_verbose_one_line_per_message(self, src: Path, out: Path, fake_popen: type[FakePopen],
|
||||||
|
caplog: pytest.LogCaptureFixture, returncode: int, level: str) -> None:
|
||||||
|
fake_popen.returncode = returncode
|
||||||
|
|
||||||
|
run_backup(out, [str(src)], verbose=True)
|
||||||
|
|
||||||
|
rsync = [(r.levelname, r.getMessage()) for r in caplog.records if r.getMessage().startswith('rsync: ')]
|
||||||
|
lines = ['sending incremental file list', 'sent 1 bytes received 2 bytes', 'total size is 3']
|
||||||
|
assert rsync == [(level, f'rsync: {line}') for line in lines]
|
||||||
|
|
||||||
|
def test_rsync_missing(self, src: Path, out: Path, tmp_path: Path, fake_popen: type[FakePopen],
|
||||||
|
old_backups: list[str], monkeypatch: pytest.MonkeyPatch,
|
||||||
|
caplog: pytest.LogCaptureFixture) -> None:
|
||||||
|
monkeypatch.setenv('PATH', str(tmp_path / 'empty'))
|
||||||
|
|
||||||
|
assert run_backup(out, [str(src)], keep=0, remove_before=True) == 4
|
||||||
|
assert fake_popen.calls == []
|
||||||
|
# Nothing was removed
|
||||||
|
assert backups(out) == old_backups
|
||||||
|
assert 'rsync not found' in caplog.text
|
||||||
|
|
||||||
|
@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,710 @@
|
|||||||
|
"""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
|
||||||
|
|
||||||
|
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: str | None = 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: list[str] | None) -> 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]] = []
|
||||||
|
envs: list[dict[str, str] | None] = []
|
||||||
|
|
||||||
|
def __init__(self, args: list[str], **kwargs: Any) -> None:
|
||||||
|
RecordPopen.calls.append(args)
|
||||||
|
RecordPopen.envs.append(kwargs.get('env'))
|
||||||
|
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 = []
|
||||||
|
RecordPopen.envs = []
|
||||||
|
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.delenv('SSHPASS', raising=False)
|
||||||
|
backup = connected_backup()
|
||||||
|
backup.inputs = [str(tmp_path / 'input')]
|
||||||
|
backup._password = 'secret'
|
||||||
|
|
||||||
|
assert backup.run() == 0
|
||||||
|
|
||||||
|
(args,) = RecordPopen.calls
|
||||||
|
assert shlex.split(self.ssh_command(args)) == ['sshpass', '-e', 'ssh', '-l', 'user', '-o',
|
||||||
|
'StrictHostKeyChecking=accept-new']
|
||||||
|
# Only rsync gets the password, never our own environment
|
||||||
|
(env,) = RecordPopen.envs
|
||||||
|
assert env is not None and env['SSHPASS'] == 'secret'
|
||||||
|
assert 'SSHPASS' not in os.environ
|
||||||
|
|
||||||
|
def test_no_password_env(self, tmp_path: Path) -> None:
|
||||||
|
self.run(tmp_path)
|
||||||
|
|
||||||
|
assert RecordPopen.envs == [None]
|
||||||
|
|
||||||
|
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[BaseException | None]) -> None:
|
||||||
|
self.outcomes = outcomes
|
||||||
|
self.connects: list[dict[str, Any]] = []
|
||||||
|
self.policies: list[str] = []
|
||||||
|
self.known_hosts: str | None = 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({'hostname': hostname, **kwargs})
|
||||||
|
outcome = self.outcomes.pop(0)
|
||||||
|
|
||||||
|
if isinstance(outcome, UserWarning):
|
||||||
|
# Like WarningPolicy for an unknown host key
|
||||||
|
warnings.warn(outcome, stacklevel=2)
|
||||||
|
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 client.connects[0]['hostname'] == 'server'
|
||||||
|
assert 'port' not in client.connects[0]
|
||||||
|
assert not client.closed
|
||||||
|
|
||||||
|
def test_ssh_config(self, client: FakeClient, tmp_path: Path) -> None:
|
||||||
|
(tmp_path / '.ssh').mkdir()
|
||||||
|
key = tmp_path / '.ssh' / 'id_backup'
|
||||||
|
key.touch()
|
||||||
|
(tmp_path / '.ssh' / 'config').write_text(
|
||||||
|
f'Host server\n HostName backup.example.org\n Port 2222\n'
|
||||||
|
f' IdentityFile {key}\n IdentityFile {tmp_path}/.ssh/missing\n')
|
||||||
|
client.outcomes = [ssh_error(), None]
|
||||||
|
|
||||||
|
assert remote_backup()._ssh_connect(str(tmp_path)) is client
|
||||||
|
|
||||||
|
for params in client.connects:
|
||||||
|
assert (params['hostname'], params['port']) == ('backup.example.org', 2222)
|
||||||
|
assert params['key_filename'] == [str(key)]
|
||||||
|
|
||||||
|
def test_bad_ssh_config(self, client: FakeClient, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||||
|
(tmp_path / '.ssh').mkdir()
|
||||||
|
(tmp_path / '.ssh' / 'config').write_text('Host server\n Port abc\n')
|
||||||
|
client.outcomes = [None]
|
||||||
|
|
||||||
|
assert remote_backup()._ssh_connect(str(tmp_path)) is client
|
||||||
|
assert client.connects[0]['hostname'] == 'server'
|
||||||
|
assert 'port' not in client.connects[0]
|
||||||
|
assert 'Cannot read' in caplog.text
|
||||||
|
|
||||||
|
@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.delenv('SSHPASS', raising=False)
|
||||||
|
backup = remote_backup()
|
||||||
|
|
||||||
|
assert backup._ssh_connect() is client
|
||||||
|
assert client.connects[-1]['password'] == 'secret'
|
||||||
|
assert backup._password == 'secret'
|
||||||
|
assert 'SSHPASS' not in os.environ
|
||||||
|
|
||||||
|
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, stacklevel=2)
|
||||||
|
|
||||||
|
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: str | None, 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, stacklevel=2)
|
||||||
|
|
||||||
|
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: str | None) -> 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
|
||||||
|
|
||||||
|
|
||||||
|
FAKE_SSH = """#!/bin/sh
|
||||||
|
# Record the arguments, then run the remote command locally in the fake remote home
|
||||||
|
for a in "$@"; do printf '%s\\n' "$a"; done >> "{log}"
|
||||||
|
echo --- >> "{log}"
|
||||||
|
|
||||||
|
while [ $# -gt 0 ]; do
|
||||||
|
case $1 in
|
||||||
|
-i|-l|-o|-p) shift 2 ;;
|
||||||
|
-*) shift ;;
|
||||||
|
*) break ;;
|
||||||
|
esac
|
||||||
|
done
|
||||||
|
|
||||||
|
# The host
|
||||||
|
shift
|
||||||
|
cd "{home}" && HOME="{home}" exec sh -c "$*"
|
||||||
|
"""
|
||||||
|
|
||||||
|
FAKE_SSHPASS = """#!/bin/sh
|
||||||
|
printf '%s' "${{SSHPASS-unset}}" > "{log}"
|
||||||
|
# -e
|
||||||
|
shift
|
||||||
|
exec "$@"
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.usefixtures('connect', 'fake_now')
|
||||||
|
class TestRsyncOverSsh:
|
||||||
|
"""Whole remote backups: real rsync, with fake ssh and sshpass commands that run the remote side locally"""
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def fake_ssh(self, ssh: FakeSSH, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
self.ssh_log = tmp_path / 'ssh.log'
|
||||||
|
self.sshpass_log = tmp_path / 'sshpass.log'
|
||||||
|
|
||||||
|
for name, script, log in [('ssh', FAKE_SSH, self.ssh_log), ('sshpass', FAKE_SSHPASS, self.sshpass_log)]:
|
||||||
|
path = ssh.bin_dir / name
|
||||||
|
path.write_text(script.format(log=log, home=ssh.home))
|
||||||
|
path.chmod(0o755)
|
||||||
|
|
||||||
|
monkeypatch.setenv('PATH', f'{ssh.bin_dir}:{os.environ["PATH"]}')
|
||||||
|
monkeypatch.delenv('SSHPASS', raising=False)
|
||||||
|
|
||||||
|
self.src = tmp_path / 'src dir'
|
||||||
|
self.src.mkdir()
|
||||||
|
(self.src / 'same.txt').write_text('unchanged')
|
||||||
|
(self.src / 'changed.txt').write_text('first')
|
||||||
|
|
||||||
|
def run(self, keep: int = -1, password: str | None = None, **kwargs: Any) -> int:
|
||||||
|
backup = connected_backup(keep=keep, **kwargs)
|
||||||
|
backup.inputs = [str(self.src)]
|
||||||
|
backup._password = password
|
||||||
|
|
||||||
|
return backup.run()
|
||||||
|
|
||||||
|
def ssh_calls(self) -> list[list[str]]:
|
||||||
|
return [c.strip('\n').split('\n') for c in self.ssh_log.read_text().split('---\n') if c != '']
|
||||||
|
|
||||||
|
def copied(self, remote_home: Path, backup: str, name: str) -> Path:
|
||||||
|
return remote_home / 'backups' / 'simple_backup' / backup / str(self.src).lstrip('/') / name
|
||||||
|
|
||||||
|
def test_backups(self, remote_home: Path) -> None:
|
||||||
|
assert self.run() == 0
|
||||||
|
(self.src / 'changed.txt').write_text('second')
|
||||||
|
assert self.run(keep=1) == 0
|
||||||
|
assert self.run(keep=1) == 0
|
||||||
|
|
||||||
|
old, new = '2026-10-14_11-00-00', '2026-10-14_12-00-00'
|
||||||
|
assert remote_dirs(remote_home / 'backups' / 'simple_backup') == [old, new]
|
||||||
|
assert self.copied(remote_home, new, 'changed.txt').read_text() == 'second'
|
||||||
|
# Unchanged files are hard-linked to the previous backup (--link-dest on the server)
|
||||||
|
assert self.copied(remote_home, old, 'same.txt').stat().st_ino == \
|
||||||
|
self.copied(remote_home, new, 'same.txt').stat().st_ino
|
||||||
|
|
||||||
|
for call in self.ssh_calls():
|
||||||
|
assert call[:5] == ['-o', 'StrictHostKeyChecking=accept-new', '-l', 'user', 'server']
|
||||||
|
# rsync passes the remote command as separate arguments
|
||||||
|
assert call[5:7] == ['rsync', '--server']
|
||||||
|
|
||||||
|
def test_keyfile(self, remote_home: Path) -> None:
|
||||||
|
assert self.run(ssh_keyfile='/keys/my key') == 0
|
||||||
|
|
||||||
|
(call,) = self.ssh_calls()
|
||||||
|
assert call[:4] == ['-i', '/keys/my key', '-o', 'StrictHostKeyChecking=accept-new']
|
||||||
|
|
||||||
|
def test_password(self, remote_home: Path) -> None:
|
||||||
|
assert self.run(password='secret') == 0
|
||||||
|
|
||||||
|
assert self.sshpass_log.read_text() == 'secret'
|
||||||
|
assert 'SSHPASS' not in os.environ
|
||||||
|
assert len(remote_dirs(remote_home / 'backups' / 'simple_backup')) == 1
|
||||||
|
|
||||||
|
def test_remote_sudo(self, remote_home: Path) -> None:
|
||||||
|
assert self.run() == 0
|
||||||
|
assert self.run(keep=0, remote_sudo=True) == 0
|
||||||
|
|
||||||
|
assert remote_dirs(remote_home / 'backups' / 'simple_backup') == ['2026-10-14_11-00-00']
|
||||||
|
assert self.ssh_calls()[-1][5:7] == ['sudo rsync', '--server']
|
||||||
|
|
||||||
|
def test_progress(self, remote_home: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
assert self.run(progress=True) == 0
|
||||||
|
|
||||||
|
assert '100%' in capsys.readouterr().err
|
||||||
@@ -0,0 +1,137 @@
|
|||||||
|
"""Tests for the last backup state file and --last-backup"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import pwd
|
||||||
|
from collections.abc import Callable
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from simple_backup import simple_backup as sb
|
||||||
|
|
||||||
|
DESTINATION = '/backups/simple_backup/2026-10-25_10-00-00'
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def as_user(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
"""Run as a non-root user (the tests may run as root in a chroot)"""
|
||||||
|
|
||||||
|
monkeypatch.setattr(os, 'geteuid', lambda: 1000)
|
||||||
|
monkeypatch.setenv('USER', 'user')
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def state(tmp_path: Path) -> Path:
|
||||||
|
return tmp_path / 'state' / 'simple_backup' / 'last_backup'
|
||||||
|
|
||||||
|
|
||||||
|
def write_state(state: Path, text: str) -> None:
|
||||||
|
state.parent.mkdir(parents=True)
|
||||||
|
state.write_text(text)
|
||||||
|
|
||||||
|
|
||||||
|
def test_write(state: Path) -> None:
|
||||||
|
sb._write_state(None, DESTINATION)
|
||||||
|
|
||||||
|
time, destination = state.read_text().splitlines()
|
||||||
|
assert abs(datetime.now().astimezone() - datetime.fromisoformat(time)) < timedelta(minutes=1)
|
||||||
|
assert destination == DESTINATION
|
||||||
|
# No temporary file left behind
|
||||||
|
assert os.listdir(state.parent) == ['last_backup']
|
||||||
|
|
||||||
|
|
||||||
|
def test_overwrite(state: Path) -> None:
|
||||||
|
write_state(state, '2026-10-01T10:00:00+02:00\n/old\n')
|
||||||
|
|
||||||
|
sb._write_state(None, DESTINATION)
|
||||||
|
|
||||||
|
assert state.read_text().splitlines()[1] == DESTINATION
|
||||||
|
|
||||||
|
|
||||||
|
def test_relative_xdg_state_home_ignored(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||||
|
monkeypatch.setenv('XDG_STATE_HOME', 'relative')
|
||||||
|
monkeypatch.setenv('HOME', str(tmp_path / 'home'))
|
||||||
|
|
||||||
|
sb._write_state(None, DESTINATION)
|
||||||
|
|
||||||
|
assert (tmp_path / 'home' / '.local' / 'state' / 'simple_backup' / 'last_backup').exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_write_error(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||||
|
# A file where the state directory should be
|
||||||
|
(tmp_path / 'not a dir').write_text('')
|
||||||
|
monkeypatch.setenv('XDG_STATE_HOME', str(tmp_path / 'not a dir'))
|
||||||
|
|
||||||
|
sb._write_state(None, DESTINATION)
|
||||||
|
|
||||||
|
assert 'Cannot save the time of the last backup' in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_root_writes_as_user(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||||
|
calls: list[tuple[str, int]] = []
|
||||||
|
egid = os.getegid()
|
||||||
|
owner = pwd.struct_passwd(('user', 'x', 1000, 1001, '', str(tmp_path / 'home'), '/bin/sh'))
|
||||||
|
|
||||||
|
monkeypatch.setattr(os, 'geteuid', lambda: 0)
|
||||||
|
monkeypatch.setattr(pwd, 'getpwnam', lambda name: owner)
|
||||||
|
monkeypatch.setattr(os, 'seteuid', lambda uid: calls.append(('euid', uid)))
|
||||||
|
monkeypatch.setattr(os, 'setegid', lambda gid: calls.append(('egid', gid)))
|
||||||
|
|
||||||
|
sb._write_state('user', DESTINATION)
|
||||||
|
|
||||||
|
# Root's XDG_STATE_HOME is not used
|
||||||
|
assert (tmp_path / 'home' / '.local' / 'state' / 'simple_backup' / 'last_backup').exists()
|
||||||
|
assert calls == [('egid', 1001), ('euid', 1000), ('euid', 0), ('egid', egid)]
|
||||||
|
|
||||||
|
|
||||||
|
class TestStateOwner:
|
||||||
|
def test_non_root(self) -> None:
|
||||||
|
assert sb._state_owner('user') is None
|
||||||
|
|
||||||
|
@pytest.mark.parametrize('name, expected', [('user', 'user'), ('root', None), ('unknown', None), (None, None)])
|
||||||
|
def test_root(self, monkeypatch: pytest.MonkeyPatch, name: str | None, expected: str | None) -> None:
|
||||||
|
users = {'user': pwd.struct_passwd(('user', 'x', 1000, 1000, '', '/home/user', '/bin/sh')),
|
||||||
|
'root': pwd.struct_passwd(('root', 'x', 0, 0, '', '/root', '/bin/sh'))}
|
||||||
|
|
||||||
|
def getpwnam(name: str) -> pwd.struct_passwd:
|
||||||
|
if name not in users:
|
||||||
|
raise KeyError(name)
|
||||||
|
|
||||||
|
return users[name]
|
||||||
|
|
||||||
|
monkeypatch.setattr(os, 'geteuid', lambda: 0)
|
||||||
|
monkeypatch.setattr(pwd, 'getpwnam', getpwnam)
|
||||||
|
|
||||||
|
owner = sb._state_owner(name)
|
||||||
|
assert (owner.pw_name if owner is not None else None) == expected
|
||||||
|
|
||||||
|
|
||||||
|
class TestShow:
|
||||||
|
@pytest.mark.parametrize('days, age', [(0, 'today'), (1, 'yesterday'), (3, '3 days ago')])
|
||||||
|
def test_age(self, run_cli: Callable[..., int], state: Path, capsys: pytest.CaptureFixture[str],
|
||||||
|
days: int, age: str) -> None:
|
||||||
|
completed = (datetime.now() - timedelta(days=days)).replace(hour=12, minute=30).astimezone()
|
||||||
|
write_state(state, f'{completed.isoformat(timespec="seconds")}\n{DESTINATION}\n')
|
||||||
|
|
||||||
|
assert run_cli('--last-backup') == 0
|
||||||
|
assert capsys.readouterr().out == \
|
||||||
|
f'Last backup: {completed:%Y-%m-%d} 12:30 ({age})\nDirectory: {DESTINATION}\n'
|
||||||
|
|
||||||
|
def test_after_write(self, run_cli: Callable[..., int], capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
sb._write_state(None, DESTINATION)
|
||||||
|
|
||||||
|
assert run_cli('--last-backup') == 0
|
||||||
|
assert '(today)' in capsys.readouterr().out
|
||||||
|
|
||||||
|
def test_no_backup(self, run_cli: Callable[..., int], capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
assert run_cli('--last-backup') == 1
|
||||||
|
assert capsys.readouterr().out == 'No backup recorded yet\n'
|
||||||
|
|
||||||
|
@pytest.mark.parametrize('text', ['', '2026-10-25T10:00:00+02:00\n', 'yesterday\n/backups\n'])
|
||||||
|
def test_bad_file(self, run_cli: Callable[..., int], state: Path, capsys: pytest.CaptureFixture[str],
|
||||||
|
text: str) -> None:
|
||||||
|
write_state(state, text)
|
||||||
|
|
||||||
|
assert run_cli('--last-backup') == 1
|
||||||
|
assert 'Cannot read' in capsys.readouterr().err
|
||||||
Reference in New Issue
Block a user