From 17c104c6c8d63ac1cf08809746fe63fd7d17e079 Mon Sep 17 00:00:00 2001 From: Fuxino Date: Fri, 16 Oct 2026 13:47:24 +0200 Subject: [PATCH] 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. --- pyproject.toml | 6 ++++ simple_backup/simple_backup.py | 66 +++++++++++++++++----------------- tests/conftest.py | 2 +- tests/test_cli.py | 23 ++++++------ tests/test_local.py | 5 +-- tests/test_remote.py | 20 +++++------ 6 files changed, 65 insertions(+), 57 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 24f0daa..64bf754 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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 diff --git a/simple_backup/simple_backup.py b/simple_backup/simple_backup.py index 1aaa193..6760d2a 100755 --- a/simple_backup/simple_backup.py +++ b/simple_backup/simple_backup.py @@ -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, diff --git a/tests/conftest.py b/tests/conftest.py index 947f455..434bd08 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,7 @@ """Shared fixtures for the simple_backup tests""" import sys -from typing import Callable +from collections.abc import Callable import pytest diff --git a/tests/test_cli.py b/tests/test_cli.py index 30fe202..0e99901 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -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 diff --git a/tests/test_local.py b/tests/test_local.py index 77de71b..09b8ff9 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -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) diff --git a/tests/test_remote.py b/tests/test_remote.py index 18cc314..73e0f27 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -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: