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
|
@@ -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
|
sudo pacman -Syu rsync
|
||||||
```
|
```
|
||||||
|
|
||||||
It's also required to have python-dotenv
|
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.
|
||||||
|
|
||||||
Optional dependencies are systemd-python to enable using systemd journal for logging, and dbus-python for desktop notifications.
|
|
||||||
|
|
||||||
## Install
|
## 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
|
```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:
|
Then build the package (this requires the Python `build` module, e.g. the python-build package on Arch Linux):
|
||||||
|
|
||||||
```bash
|
|
||||||
pip install --upgrade build wheel
|
|
||||||
```
|
|
||||||
|
|
||||||
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
|
||||||
|
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
|
## 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 --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.
|
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
|
### 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.
|
||||||
If needed, it's possible to specify the ssh key location with the --keyfile argument or in the configuration file.
|
If needed, it's possible to specify the ssh key location with the --keyfile argument or in the configuration file.
|
||||||
|
|||||||
+37
-7
@@ -1,4 +1,4 @@
|
|||||||
.TH SIMPLE_BACKUP 1 2023-06-15 SIMPLE_BACKUP 3.2.6
|
.TH SIMPLE_BACKUP 1 2026-10-14 SIMPLE_BACKUP 4.4.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
|
||||||
@@ -52,7 +52,15 @@ or to use single or double quotes around them.
|
|||||||
.B \-o, \-\-output DIR
|
.B \-o, \-\-output DIR
|
||||||
Specify the directory where the files will be copied. The program will automatically
|
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
|
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
|
.TP
|
||||||
.B \-e, \-\-exclude FILE|DIR|PATTERN [FILE|...]]
|
.B \-e, \-\-exclude FILE|DIR|PATTERN [FILE|...]]
|
||||||
Specify files, directories or patterns to exclude from the backup. Matching files and directories
|
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
|
.B \-k, \-\-keep N
|
||||||
Specify how many old backups (so excluding the current one) will be kept. The default behavior
|
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).
|
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
|
.TP
|
||||||
.B \-u, \-\-user USERNAME
|
.B \-u, \-\-user USERNAME
|
||||||
Explicitly specify the user running the backup (in case it is needed for home directory expansion).
|
Explicitly specify the user running the backup (in case it is needed for home directory expansion).
|
||||||
@@ -86,10 +98,19 @@ if saving bandwith is needed.
|
|||||||
Remove old backups (if necessary) before creating the new backup. Useful to free some space
|
Remove old backups (if necessary) before creating the new backup. Useful to free some space
|
||||||
before performing the backup.
|
before performing the backup.
|
||||||
Default behavior is to remove old backups after successfully completing 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
|
.TP
|
||||||
.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
|
||||||
@@ -100,7 +121,7 @@ Using \-\-rsync\-options it is possible to manually select which options to use.
|
|||||||
.P
|
.P
|
||||||
\-a, \-l, \-p, \-t, \-g, \-o, \-c, \-h, \-D, \-H, \-X, \-s
|
\-a, \-l, \-p, \-t, \-g, \-o, \-c, \-h, \-D, \-H, \-X, \-s
|
||||||
.P
|
.P
|
||||||
Options \-r and \-v are used in any case. Not that options must be specified without dash (\-), for example:
|
Options \-r, \-v, \-\-ignore\-missing\-args and \-\-mkpath are used in any case. Note that options must be specified without dash (\-), for example:
|
||||||
.P
|
.P
|
||||||
.EX
|
.EX
|
||||||
simple_backup \-\-rsync\-options a l p
|
simple_backup \-\-rsync\-options a l p
|
||||||
@@ -131,6 +152,14 @@ Copy it to the default location ($HOME/.config/simple_backup) and edit it as nee
|
|||||||
It is possible to choose a directory on a remote server as destination for the backup. The files
|
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).
|
||||||
|
.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
|
||||||
@@ -165,7 +194,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.
|
||||||
@@ -177,16 +207,16 @@ 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).
|
||||||
.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 SEE ALSO
|
.SH SEE ALSO
|
||||||
.BR rsync (1)
|
.BR rsync (1)
|
||||||
.SH AUTHORS
|
.SH AUTHORS
|
||||||
.MT https://github.com/Fuxino
|
.MT https://git.shouldnt.work/fuxino
|
||||||
Daniele Fucini
|
Daniele Fucini
|
||||||
.ME
|
.ME
|
||||||
|
|||||||
+61
-1
@@ -1,3 +1,63 @@
|
|||||||
[build-system]
|
[build-system]
|
||||||
requires = ['setuptools']
|
requires = ['setuptools>=77']
|
||||||
build-backend = 'setuptools.build_meta'
|
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."""
|
"""Init."""
|
||||||
|
|
||||||
__version__ = '4.1.2'
|
__version__ = '4.4.0'
|
||||||
|
|||||||
@@ -2,9 +2,9 @@
|
|||||||
|
|
||||||
[backup]
|
[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).
|
# 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
|
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
|
||||||
|
|||||||
+632
-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