Add ruff linting

Configure ruff in pyproject.toml and apply its fixes: modern annotations
(X | None, builtin generics, collections.abc), sorted imports, explicit
stacklevel in test warnings.
This commit is contained in:
Fuxino
2026-10-16 13:47:24 +02:00
parent 7dbb1bcaee
commit 17c104c6c8
6 changed files with 65 additions and 57 deletions
+6
View File
@@ -57,6 +57,12 @@ warn_unused_ignores = true
module = ["paramiko", "systemd", "dbus"]
ignore_missing_imports = true
[tool.ruff]
line-length = 120
[tool.ruff.lint]
select = ["E", "F", "W", "I", "B", "UP"]
[tool.pytest.ini_options]
testpaths = ["tests"]
# Import the repo copy, not an installed one
+33 -33
View File
@@ -14,27 +14,27 @@ Classes:
# Avoid evaluating annotations at definition time, as they reference optional modules (paramiko)
from __future__ import annotations
# Import libraries
import sys
import os
import pwd
import io
import re
from typing import Callable, Iterator, List, Optional, ParamSpec, TypedDict, TypeVar
import warnings
from functools import wraps
from shutil import get_terminal_size, rmtree, which
import shlex
import argparse
import configparser
import io
import logging
from logging import StreamHandler
from timeit import default_timer
from subprocess import Popen, PIPE, STDOUT
import os
import pwd
import re
import shlex
import sys
import warnings
from collections.abc import Callable, Iterator
from datetime import datetime
from tempfile import mkstemp
from functools import wraps
from getpass import GetPassWarning, getpass
from glob import glob
from logging import StreamHandler
from shutil import get_terminal_size, rmtree, which
from subprocess import PIPE, STDOUT, Popen
from tempfile import mkstemp
from timeit import default_timer
from typing import ParamSpec, TypedDict, TypeVar
try:
import paramiko
@@ -204,9 +204,9 @@ class Backup:
Perform the backup
"""
def __init__(self, inputs: Optional[List[str]], output: Optional[str], exclude: Optional[List[str]],
keep: Optional[int], options: str,
ssh_host: Optional[str] = None, ssh_user: Optional[str] = None, ssh_keyfile: Optional[str] = None,
def __init__(self, inputs: list[str] | None, output: str | None, exclude: list[str] | None,
keep: int | None, options: str,
ssh_host: str | None = None, ssh_user: str | None = None, ssh_keyfile: str | None = None,
remote_sudo: bool = False, remove_before: bool = False, verbose: bool = False,
progress: bool = False) -> None:
# Missing values are reported by check_params()
@@ -228,7 +228,7 @@ class Backup:
self._inputs_path = ''
self._exclude_path = ''
self._remote = False
self._ssh: Optional[paramiko.client.SSHClient] = None
self._ssh: paramiko.client.SSHClient | None = None
self._password_auth = False
self._removed_count = 0
@@ -281,7 +281,7 @@ class Backup:
return 0
def _resolve_remote_path(self, path: str) -> Optional[str]:
def _resolve_remote_path(self, path: str) -> str | None:
"""Return the absolute path of a directory on the server, or None if it doesn't exist"""
assert self._ssh is not None
@@ -450,7 +450,7 @@ class Backup:
except IndexError:
logger.info('No previous backups available')
def _ssh_connect(self, homedir: str = '') -> Optional[paramiko.client.SSHClient]:
def _ssh_connect(self, homedir: str = '') -> paramiko.client.SSHClient | None:
try:
ssh = paramiko.SSHClient()
except NameError:
@@ -469,7 +469,7 @@ class Backup:
if connected is None:
ssh.close()
def _ssh_login(self, ssh: paramiko.client.SSHClient, homedir: str) -> Optional[paramiko.client.SSHClient]:
def _ssh_login(self, ssh: paramiko.client.SSHClient, homedir: str) -> paramiko.client.SSHClient | None:
try:
ssh.load_host_keys(filename=f'{homedir}/.ssh/known_hosts')
except FileNotFoundError:
@@ -553,7 +553,7 @@ class Backup:
return ssh
def _load_ssh_key(self) -> Optional[paramiko.PKey]:
def _load_ssh_key(self) -> paramiko.PKey | None:
try:
return paramiko.PKey.from_path(self.ssh_keyfile)
except TypeError:
@@ -808,7 +808,7 @@ def _read_records(stream: io.BufferedIOBase) -> Iterator[str]:
yield buffer.decode('utf-8', errors='replace')
def _detect_user() -> Optional[str]:
def _detect_user() -> str | None:
if os.geteuid() == 0:
return os.getenv('SUDO_USER') or os.getenv('DOAS_USER')
@@ -873,8 +873,8 @@ def _expand_home(path: str, user: str) -> str:
return os.path.expanduser(path)
def _expand_inputs(inputs: List[str], user: Optional[str] = None) -> List[str]:
expanded_inputs: List[str] = []
def _expand_inputs(inputs: list[str], user: str | None = None) -> list[str]:
expanded_inputs: list[str] = []
for i in inputs:
if i == '':
@@ -900,13 +900,13 @@ def _expand_inputs(inputs: List[str], user: Optional[str] = None) -> List[str]:
class ConfigArgs(TypedDict):
"""Values read from the configuration file"""
inputs: Optional[List[str]]
output: Optional[str]
exclude: Optional[List[str]]
inputs: list[str] | None
output: str | None
exclude: list[str] | None
keep: int
ssh_host: Optional[str]
ssh_user: Optional[str]
ssh_keyfile: Optional[str]
ssh_host: str | None
ssh_user: str | None
ssh_keyfile: str | None
remote_sudo: bool
numeric_ids: bool
@@ -919,7 +919,7 @@ def _get_bool(config: configparser.ConfigParser, key: str) -> bool:
return config.getboolean('server', key)
def _read_config(config_file: str, user: Optional[str] = None) -> ConfigArgs:
def _read_config(config_file: str, user: str | None = None) -> ConfigArgs:
config_args: ConfigArgs = {'inputs': None,
'output': None,
'exclude': None,
+1 -1
View File
@@ -1,7 +1,7 @@
"""Shared fixtures for the simple_backup tests"""
import sys
from typing import Callable
from collections.abc import Callable
import pytest
+12 -11
View File
@@ -2,8 +2,9 @@
import os
import pwd
from collections.abc import Callable
from pathlib import Path
from typing import Any, Callable, Optional
from typing import Any
import pytest
@@ -25,8 +26,8 @@ class 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,
ssh_host: str | None = None, ssh_user: str | None = None,
ssh_keyfile: str | None = None, remote_sudo: bool = False,
remove_before: bool = False, verbose: bool = False, progress: bool = False) -> None:
self.inputs = inputs
self.output = output
@@ -40,7 +41,7 @@ class FakeBackup:
self.remove_before = remove_before
self.verbose = verbose
self.progress = progress
self.homedir: Optional[str] = None
self.homedir: str | None = None
self.ran = False
self.closed = False
FakeBackup.instances.append(self)
@@ -86,13 +87,13 @@ def as_root(monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.fixture
def config_calls(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, Optional[str]]]:
def config_calls(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str | None]]:
"""Record the config file path and user instead of reading the file"""
calls: list[tuple[str, Optional[str]]] = []
calls: list[tuple[str, str | None]] = []
read_config = sb._read_config
def record(config_file: str, user: Optional[str] = None) -> sb.ConfigArgs:
def record(config_file: str, user: str | None = None) -> sb.ConfigArgs:
calls.append((config_file, user))
return read_config('/nonexistent/simple_backup.conf', user)
@@ -249,20 +250,20 @@ ssh_keyfile=/keys/id
class TestUser:
@pytest.mark.usefixtures('as_user')
def test_current_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]]) -> None:
def test_current_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]]) -> None:
assert run_cli() == 0
assert config_calls == [(f'{HOME}/.config/simple_backup/simple_backup.conf', USER)]
assert last_backup().homedir == HOME
@pytest.mark.usefixtures('as_root')
def test_explicit_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]]) -> None:
def test_explicit_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]]) -> None:
assert run_cli('-u', USER, '-c', '~/other.conf') == 0
assert config_calls == [(f'{HOME}/other.conf', USER)]
assert last_backup().homedir == HOME
@pytest.mark.usefixtures('as_root')
@pytest.mark.parametrize('var', ['SUDO_USER', 'DOAS_USER'])
def test_sudo_or_doas_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]],
def test_sudo_or_doas_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]],
monkeypatch: pytest.MonkeyPatch, var: str) -> None:
monkeypatch.setenv(var, USER)
@@ -271,7 +272,7 @@ class TestUser:
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]]],
def test_root_without_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]],
caplog: pytest.LogCaptureFixture) -> None:
assert run_cli() == 0
# '~' is left to expanduser, i.e. $HOME
+3 -2
View File
@@ -2,9 +2,10 @@
import io
import os
from collections.abc import Callable, Iterator
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Callable, Iterator, Optional
from typing import Any
import pytest
@@ -52,7 +53,7 @@ def out(tmp_path: Path) -> Path:
return out
def make_backup(out: Path, inputs: list[str], keep: int = -1, exclude: Optional[list[str]] = None,
def make_backup(out: Path, inputs: list[str], keep: int = -1, exclude: list[str] | None = 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)
+10 -10
View File
@@ -10,7 +10,7 @@ import warnings
from getpass import GetPassWarning
from pathlib import Path
from shutil import which
from typing import Any, Optional
from typing import Any
import pytest
@@ -30,7 +30,7 @@ class FakeSSH:
self.home = home
self.bin_dir = bin_dir
self.commands: list[str] = []
self.fail_on: Optional[str] = None
self.fail_on: str | None = None
self.closed = False
def exec_command(self, command: str) -> tuple[None, io.BytesIO, io.BytesIO]:
@@ -186,7 +186,7 @@ class TestRemoveOldBackups:
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:
def test_nothing_to_remove(self, remote_home: Path, names: list[str] | None) -> None:
if names is not None:
make_backups(remote_home, *names)
@@ -348,11 +348,11 @@ class TestRsyncArgs:
class FakeClient:
"""paramiko.SSHClient stand-in; connect() takes its outcomes from a list"""
def __init__(self, outcomes: list[Optional[BaseException]]) -> None:
def __init__(self, outcomes: list[BaseException | None]) -> None:
self.outcomes = outcomes
self.connects: list[dict[str, Any]] = []
self.policies: list[str] = []
self.known_hosts: Optional[str] = None
self.known_hosts: str | None = None
self.closed = False
def load_host_keys(self, filename: str) -> None:
@@ -370,7 +370,7 @@ class FakeClient:
if isinstance(outcome, UserWarning):
# Like WarningPolicy for an unknown host key
warnings.warn(outcome)
warnings.warn(outcome, stacklevel=2)
elif outcome is not None:
raise outcome
@@ -462,7 +462,7 @@ class TestLogin:
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)
warnings.warn('Can not control echo on the terminal.', GetPassWarning, stacklevel=2)
return ''
@@ -474,7 +474,7 @@ class TestLogin:
@pytest.mark.parametrize('outcome, connected', [(None, True), ('error', False)])
def test_keyfile(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
outcome: Optional[str], connected: bool) -> None:
outcome: str | None, connected: bool) -> None:
pkey = object()
client.outcomes = [ssh_error(), ssh_error(), ssh_error() if outcome else None]
monkeypatch.setattr(sb.Backup, '_load_ssh_key', lambda self: pkey)
@@ -548,7 +548,7 @@ class TestLoadKey:
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)
warnings.warn('Can not control echo on the terminal.', GetPassWarning, stacklevel=2)
return ''
@@ -558,7 +558,7 @@ class TestLoadKey:
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:
def test_bad_file(self, tmp_path: Path, caplog: pytest.LogCaptureFixture, content: str | None) -> None:
keyfile = tmp_path / 'id'
if content is not None: