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