Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0: CHANGELOG.md Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0: CONTRIBUTING.md Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0: MANIFEST.in diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/PKG-INFO /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/PKG-INFO --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/PKG-INFO 2026-01-04 23:42:44.827348200 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/PKG-INFO 2026-01-06 01:47:11.636146000 +0000 @@ -4,5 +4,4 @@ Summary: Swiss army knife cli tool for atomic, incremental, intelligent, feature-rich backups for btrfs Author-email: Michael Berry -License-Expression: MIT Project-URL: Homepage, https://github.com/berrym/btrfs-backup-ng Classifier: Programming Language :: Python :: 3 @@ -22,4 +21,5 @@ Requires-Dist: ruff>=0.1.0; extra == "dev" Requires-Dist: mypy>=1.0; extra == "dev" +Requires-Dist: types-paramiko>=3.0; extra == "dev" Dynamic: license-file @@ -27,6 +27,10 @@ [![CI](https://github.com/berrym/btrfs-backup-ng/actions/workflows/ci.yml/badge.svg)](https://github.com/berrym/btrfs-backup-ng/actions/workflows/ci.yml) +[![codecov](https://codecov.io/gh/berrym/btrfs-backup-ng/graph/badge.svg)](https://codecov.io/gh/berrym/btrfs-backup-ng) [![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) +[![Checked with mypy](https://www.mypy-lang.org/static/mypy_badge.svg)](https://mypy-lang.org/) +[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff) +[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black) A modern, feature-rich tool for automated BTRFS snapshot backup management with TOML configuration, time-based retention policies, and robust SSH transfer support. @@ -90,12 +94,49 @@ ### Shell Completions -Shell completion scripts are available in the `completions/` directory for bash, zsh, and fish. See `completions/README.md` for installation instructions. +Install shell completions for tab-completion support: + +```bash +# Install for your shell (bash, zsh, or fish) +btrfs-backup-ng completions install --shell bash +btrfs-backup-ng completions install --shell zsh +btrfs-backup-ng completions install --shell fish + +# System-wide installation (requires root) +sudo btrfs-backup-ng completions install --shell bash --system + +# Show path to completion scripts +btrfs-backup-ng completions path +``` + +See `completions/README.md` for manual installation instructions. ### Man Pages -Manual pages are available in the `man/` directory. See `man/README.md` for installation instructions. View without installing: +Install man pages for offline documentation: + +```bash +# User installation (to ~/.local/share/man) +btrfs-backup-ng manpages install + +# System-wide installation (requires root) +sudo btrfs-backup-ng manpages install --system + +# Custom prefix +btrfs-backup-ng manpages install --prefix /opt/myapp + +# Show path to man page files +btrfs-backup-ng manpages path +``` + +After user installation, add to your MANPATH: +```bash +export MANPATH="$HOME/.local/share/man:$MANPATH" +``` +View man pages: ```bash -man ./man/btrfs-backup-ng.1 +man btrfs-backup-ng +man btrfs-backup-ng-restore +man btrfs-backup-ng-config ``` diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/README.md /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/README.md --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/README.md 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/README.md 2026-01-05 05:44:19.000000000 +0000 @@ -2,6 +2,10 @@ [![CI](https://github.com/berrym/btrfs-backup-ng/actions/workflows/ci.yml/badge.svg)](https://github.com/berrym/btrfs-backup-ng/actions/workflows/ci.yml) +[![codecov](https://codecov.io/gh/berrym/btrfs-backup-ng/graph/badge.svg)](https://codecov.io/gh/berrym/btrfs-backup-ng) [![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) +[![Checked with mypy](https://www.mypy-lang.org/static/mypy_badge.svg)](https://mypy-lang.org/) +[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff) +[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black) A modern, feature-rich tool for automated BTRFS snapshot backup management with TOML configuration, time-based retention policies, and robust SSH transfer support. @@ -65,12 +69,49 @@ ### Shell Completions -Shell completion scripts are available in the `completions/` directory for bash, zsh, and fish. See `completions/README.md` for installation instructions. +Install shell completions for tab-completion support: + +```bash +# Install for your shell (bash, zsh, or fish) +btrfs-backup-ng completions install --shell bash +btrfs-backup-ng completions install --shell zsh +btrfs-backup-ng completions install --shell fish + +# System-wide installation (requires root) +sudo btrfs-backup-ng completions install --shell bash --system + +# Show path to completion scripts +btrfs-backup-ng completions path +``` + +See `completions/README.md` for manual installation instructions. ### Man Pages -Manual pages are available in the `man/` directory. See `man/README.md` for installation instructions. View without installing: +Install man pages for offline documentation: + +```bash +# User installation (to ~/.local/share/man) +btrfs-backup-ng manpages install + +# System-wide installation (requires root) +sudo btrfs-backup-ng manpages install --system + +# Custom prefix +btrfs-backup-ng manpages install --prefix /opt/myapp + +# Show path to man page files +btrfs-backup-ng manpages path +``` + +After user installation, add to your MANPATH: +```bash +export MANPATH="$HOME/.local/share/man:$MANPATH" +``` +View man pages: ```bash -man ./man/btrfs-backup-ng.1 +man btrfs-backup-ng +man btrfs-backup-ng-restore +man btrfs-backup-ng-config ``` Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0: completions Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0: man diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/pyproject.toml /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/pyproject.toml --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/pyproject.toml 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/pyproject.toml 2026-01-06 01:47:06.000000000 +0000 @@ -11,5 +11,5 @@ description = "Swiss army knife cli tool for atomic, incremental, intelligent, feature-rich backups for btrfs" authors = [{ name = "Michael Berry", email = "trismegustis@gmail.com" }] -license = "MIT" +license = { file = "LICESNE" } dependencies = ["filelock", "rich", "paramiko"] classifiers = [ @@ -30,4 +30,5 @@ "ruff>=0.1.0", "mypy>=1.0", + "types-paramiko>=3.0", ] @@ -81,2 +82,19 @@ # Goal: increase to 70%+ as we add more unit tests with mocking fail_under = 55 + +[tool.mypy] +python_version = "3.11" +warn_unused_configs = true +ignore_missing_imports = true +exclude = [ + "tests/", +] + +[tool.basedpyright] +pythonVersion = "3.11" +typeCheckingMode = "standard" +reportMissingImports = false +reportMissingTypeStubs = false +exclude = [ + "**/tests/**", +] diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__logger__.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__logger__.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__logger__.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__logger__.py 2026-01-05 05:23:50.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: standard - """btrfs-backup-ng: btrfs-backup_ng/Logger.py A common logger for displaying in a rich layout with optional file logging. @@ -11,6 +9,6 @@ from collections import deque from pathlib import Path -from typing import IO # , override (requires python 3.12) +# Note: override decorator requires Python 3.12+ from rich.console import Console from rich.logging import RichHandler @@ -30,6 +28,9 @@ -class RichLogger(IO[str]): - """A singleton pattern class to share internal state of the rich logger.""" +class RichLogger: + """A singleton pattern class to share internal state of the rich logger. + + Implements write() and flush() as required by Rich Console's file parameter. + """ __instance = None @@ -96,5 +97,5 @@ # Create new handlers if live_layout: - cons = Console(file=RichLogger(), width=150) + cons = Console(file=RichLogger(), width=150) # type: ignore[arg-type] rich_handler = RichHandler(console=cons, show_time=False, show_path=False) else: diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__main__.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__main__.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__main__.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__main__.py 2026-01-05 05:23:50.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: standard - """btrfs-backup-ng: Automated btrfs backup management. diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__util__.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__util__.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__util__.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/__util__.py 2026-01-05 21:53:48.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: standard - """btrfs-backup-ng: btrfs_backup_ng/__util__.py Common utility code shared between modules. @@ -14,4 +12,23 @@ from .__logger__ import logger +__all__ = [ + "AbortError", + "SnapshotTransferError", + "Snapshot", + "exec_subprocess", + "log_heading", + "date_to_str", + "str_to_date", + "is_btrfs", + "is_subvolume", + "is_mounted", + "get_mount_info", + "read_locks", + "write_locks", + "delete_subvolume", + "DATE_FORMAT", + "MOUNTS_FILE", +] + DATE_FORMAT = "%Y%m%d-%H%M%S" MOUNTS_FILE = "/proc/mounts" @@ -194,10 +211,19 @@ def is_subvolume(path): - """Checks whether the given path is a btrfs subvolume.""" + """Checks whether the given path is a btrfs subvolume. + + Args: + path: Path to check + + Returns: + True if path is a btrfs subvolume, False otherwise + """ + path = Path(path).resolve() + if not path.exists(): + return False if not is_btrfs(path): return False logger.debug("Checking for btrfs subvolume: %s", path) # subvolumes always have inode 256 - path = Path(path).resolve() st = path.stat() result = st.st_ino == 256 @@ -206,4 +232,21 @@ +def delete_subvolume(path): + """Delete a btrfs subvolume. + + Args: + path: Path to the subvolume to delete + + Raises: + AbortError: If deletion fails + """ + path = Path(path).resolve() + logger.debug("Deleting btrfs subvolume: %s", path) + if not is_subvolume(path): + raise AbortError(f"Path is not a subvolume: {path}") + exec_subprocess(["btrfs", "subvolume", "delete", str(path)]) + logger.debug(" -> Subvolume deleted successfully") + + def is_mounted(path): """Check if path is an active mount point. diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/_legacy_main.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/_legacy_main.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/_legacy_main.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/_legacy_main.py 2026-01-05 05:23:50.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: standard - """Legacy CLI implementation for btrfs-backup-ng. Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli: completions.py diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/config_cmd.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/config_cmd.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/config_cmd.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/config_cmd.py 2026-01-05 05:15:47.000000000 +0000 @@ -270,5 +270,5 @@ print("Configure how long to keep snapshots. Set to 0 to disable.") - retention = {} + retention: dict[str, str | int] = {} retention["min"] = _prompt("Minimum retention period", "1d") retention["hourly"] = _prompt_int("Hourly snapshots to keep", 24, 0, 1000) @@ -321,5 +321,5 @@ print("-" * 40) - volumes = [] + volumes: list[dict[str, Any]] = [] add_volume = True diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/dispatcher.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/dispatcher.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/dispatcher.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/dispatcher.py 2026-01-05 03:58:04.000000000 +0000 @@ -26,4 +26,6 @@ "verify", "estimate", + "completions", + "manpages", } ) @@ -592,4 +594,61 @@ ) + # completions command + completions_parser = subparsers.add_parser( + "completions", + help="Install shell completion scripts", + description="Install or locate shell completion scripts for bash, zsh, or fish", + ) + completions_subs = completions_parser.add_subparsers(dest="completions_action") + + completions_install = completions_subs.add_parser( + "install", + help="Install completions for your shell", + ) + completions_install.add_argument( + "--shell", + choices=["bash", "zsh", "fish"], + required=True, + help="Shell to install completions for", + ) + completions_install.add_argument( + "--system", + action="store_true", + help="Install system-wide (requires root)", + ) + + completions_subs.add_parser( + "path", + help="Show path to completion scripts", + ) + + # manpages command + manpages_parser = subparsers.add_parser( + "manpages", + help="Install man pages", + description="Install or locate man pages for btrfs-backup-ng commands", + ) + manpages_subs = manpages_parser.add_subparsers(dest="manpages_action") + + manpages_install = manpages_subs.add_parser( + "install", + help="Install man pages", + ) + manpages_install.add_argument( + "--system", + action="store_true", + help="Install system-wide to /usr/local/share/man (requires root)", + ) + manpages_install.add_argument( + "--prefix", + metavar="PATH", + help="Install to PREFIX/share/man/man1", + ) + + manpages_subs.add_parser( + "path", + help="Show path to man page files", + ) + # estimate command estimate_parser = subparsers.add_parser( @@ -757,4 +816,6 @@ "verify": cmd_verify, "estimate": cmd_estimate, + "completions": cmd_completions, + "manpages": cmd_manpages, } @@ -855,4 +916,18 @@ +def cmd_completions(args: argparse.Namespace) -> int: + """Execute completions command.""" + from .completions import execute_completions + + return execute_completions(args) + + +def cmd_manpages(args: argparse.Namespace) -> int: + """Execute manpages command.""" + from .manpages import execute_manpages + + return execute_manpages(args) + + def main(argv: list[str] | None = None) -> int: """Main entry point for btrfs-backup-ng CLI. diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/estimate.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/estimate.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/estimate.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/estimate.py 2026-01-05 05:20:03.000000000 +0000 @@ -11,4 +11,5 @@ import logging from pathlib import Path +from typing import Any from .. import endpoint @@ -73,5 +74,5 @@ try: if config_path: - config = load_config(config_path) + config, _warnings = load_config(config_path) else: found_path = find_config_file() @@ -80,5 +81,5 @@ print("Use --config to specify a config file") return 1 - config = load_config(found_path) + config, _warnings = load_config(found_path) except ConfigError as e: logger.error("Failed to load config: %s", e) @@ -250,5 +251,5 @@ destination: Destination path for display """ - data = { + data: dict[str, Any] = { "source": source, "destination": destination, diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/list_cmd.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/list_cmd.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/list_cmd.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/list_cmd.py 2026-01-05 05:16:22.000000000 +0000 @@ -6,4 +6,5 @@ import os from pathlib import Path +from typing import Any from .. import endpoint @@ -58,8 +59,8 @@ output_json = getattr(args, "json", False) - all_data = [] + all_data: list[dict[str, Any]] = [] for volume in volumes: - volume_data = { + volume_data: dict[str, Any] = { "path": volume.path, "snapshot_prefix": volume.snapshot_prefix, @@ -115,5 +116,5 @@ # Get target snapshots for target in volume.targets: - target_data = { + target_data: dict[str, Any] = { "path": target.path, "snapshots": [], Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli: manpages.py diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/prune.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/prune.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/prune.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/prune.py 2026-01-05 05:17:31.000000000 +0000 @@ -6,4 +6,5 @@ import time from pathlib import Path +from typing import Literal from .. import __util__, endpoint @@ -276,4 +277,5 @@ # Determine overall status + status: Literal["success", "failure", "partial"] if volumes_failed == 0: status = "success" diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/restore.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/restore.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/restore.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/restore.py 2026-01-05 05:22:29.000000000 +0000 @@ -10,4 +10,5 @@ from datetime import datetime from pathlib import Path +from typing import Any from .. import __util__, endpoint @@ -97,8 +98,9 @@ """ # Load config - config_path = getattr(args, "config", None) + config_file = getattr(args, "config", None) try: - if config_path: - config = load_config(config_path) + if config_file: + config, _warnings = load_config(config_file) + config_path = config_file else: found_path = find_config_file() @@ -109,5 +111,5 @@ print(" /etc/btrfs-backup-ng/config.toml") return 1 - config = load_config(found_path) + config, _warnings = load_config(found_path) config_path = found_path except ConfigError as e: @@ -157,8 +159,8 @@ """ # Load config - config_path = getattr(args, "config", None) + config_file = getattr(args, "config", None) try: - if config_path: - config = load_config(config_path) + if config_file: + config, _warnings = load_config(config_file) else: found_path = find_config_file() @@ -167,5 +169,5 @@ print("Use --config to specify a config file") return 1 - config = load_config(found_path) + config, _warnings = load_config(found_path) except ConfigError as e: logger.error("Failed to load config: %s", e) @@ -482,5 +484,5 @@ local_ep = LocalEndpoint(config=endpoint_kwargs) - local_ep.prepare() + local_ep.prepare() # type: ignore[attr-defined] return local_ep @@ -624,5 +626,5 @@ if lock_file_path.exists(): with open(lock_file_path, encoding="utf-8") as f: - locks = __util__.read_locks(f.read()) + locks = __util__.read_locks(f.read()) # type: ignore[attr-defined] except Exception as e: logger.warning("Could not read lock file: %s", e) @@ -633,6 +635,6 @@ print("Active Locks:") print("-" * 40) - restore_locks = {} - other_locks = {} + restore_locks: dict[str, Any] = {} + other_locks: dict[str, Any] = {} for snap_name, lock_info in locks.items(): @@ -716,5 +718,5 @@ with open(lock_file_path, encoding="utf-8") as f: - locks = __util__.read_locks(f.read()) + locks = __util__.read_locks(f.read()) # type: ignore[attr-defined] except Exception as e: logger.error("Could not read lock file: %s", e) @@ -727,5 +729,5 @@ # Find and remove matching locks unlocked_count = 0 - new_locks = {} + new_locks: dict[str, Any] = {} for snap_name, lock_info in locks.items(): @@ -766,5 +768,5 @@ try: with open(lock_file_path, "w", encoding="utf-8") as f: - f.write(__util__.write_locks(new_locks)) + f.write(__util__.write_locks(new_locks)) # type: ignore[attr-defined] except Exception as e: logger.error("Could not write lock file: %s", e) @@ -831,5 +833,5 @@ # Check if it's a subvolume - if not __util__.is_subvolume(item): + if not __util__.is_subvolume(item): # type: ignore[attr-defined] continue diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/run.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/run.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/run.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/run.py 2026-01-05 05:18:52.000000000 +0000 @@ -7,4 +7,5 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path +from typing import Literal from .. import __util__, endpoint @@ -317,5 +318,5 @@ if target.require_mount and not target.path.startswith("ssh://"): target_path = Path(target.path).resolve() - if not __util__.is_mounted(target_path): + if not __util__.is_mounted(target_path): # type: ignore[attr-defined] raise __util__.AbortError( f"Target {target.path} is not mounted. " @@ -490,4 +491,5 @@ # Determine overall status + status: Literal["success", "failure", "partial"] if volumes_failed == 0 and transfers_failed == 0: status = "success" diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/status.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/status.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/status.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/status.py 2026-01-05 05:16:36.000000000 +0000 @@ -15,11 +15,12 @@ -def _format_bytes(size_bytes: int) -> str: +def _format_bytes(size_bytes: int | float) -> str: """Format bytes as human-readable string.""" + size: float = float(size_bytes) for unit in ("B", "KB", "MB", "GB", "TB"): - if abs(size_bytes) < 1024: - return f"{size_bytes:.1f} {unit}" - size_bytes /= 1024 - return f"{size_bytes:.1f} PB" + if abs(size) < 1024: + return f"{size:.1f} {unit}" + size /= 1024 + return f"{size:.1f} PB" diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/transfer.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/transfer.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/transfer.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/cli/transfer.py 2026-01-05 05:18:23.000000000 +0000 @@ -145,5 +145,5 @@ if target.require_mount and not target.path.startswith("ssh://"): target_path = Path(target.path).resolve() - if not __util__.is_mounted(target_path): + if not __util__.is_mounted(target_path): # type: ignore[attr-defined] raise __util__.AbortError( f"Target {target.path} is not mounted. " diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/estimate.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/estimate.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/estimate.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/estimate.py 2026-01-05 05:01:30.000000000 +0000 @@ -262,5 +262,5 @@ source_endpoint, dest_endpoint, - snapshots: list = None, + snapshots: list | None = None, ) -> TransferEstimate: """Estimate the size of a backup transfer operation. diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/execution.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/execution.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/execution.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/execution.py 2026-01-05 05:06:15.000000000 +0000 @@ -202,5 +202,5 @@ List of JobResults for this volume """ - results = [] + results: list[JobResult] = [] if not volume.targets: diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/progress.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/progress.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/progress.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/progress.py 2026-01-05 05:06:36.000000000 +0000 @@ -347,4 +347,7 @@ # Create reader thread to pipe data with progress updates + # These should always be set since the processes are configured with PIPE + assert send_process.stdout is not None, "send_process.stdout must be PIPE" + assert receive_process.stdin is not None, "receive_process.stdin must be PIPE" reader = ProgressReader( source=send_process.stdout, diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/restore.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/restore.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/restore.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/restore.py 2026-01-05 05:05:32.000000000 +0000 @@ -9,7 +9,8 @@ import uuid from pathlib import Path -from typing import Callable +from typing import Any, Callable from .. import __util__ +from ..__util__ import Snapshot from ..transaction import log_transaction from .operations import send_snapshot @@ -25,8 +26,8 @@ def get_restore_chain( - target_snapshot, - all_backup_snapshots: list, - existing_local: list, -) -> list: + target_snapshot: Snapshot, + all_backup_snapshots: list[Snapshot], + existing_local: list[Snapshot], +) -> list[Snapshot]: """Determine which snapshots need to be restored to get target_snapshot. @@ -46,6 +47,6 @@ existing_names = {s.get_name() for s in existing_local} - chain = [] - current = target_snapshot + chain: list[Snapshot] = [] + current: Snapshot | None = target_snapshot while current is not None: @@ -172,5 +173,5 @@ # Must be on btrfs filesystem - if not __util__.is_btrfs(path): + if not __util__.is_btrfs(path): # type: ignore[attr-defined] raise RestoreError( f"Destination {path} is not on a btrfs filesystem. " @@ -239,5 +240,5 @@ # Verify it's a valid subvolume - if not __util__.is_subvolume(snapshot_path): + if not __util__.is_subvolume(snapshot_path): # type: ignore[attr-defined] raise RestoreError( f"{snapshot_path} exists but is not a valid btrfs subvolume. " @@ -401,5 +402,5 @@ session_id = str(uuid.uuid4())[:8] - stats = {"restored": 0, "skipped": 0, "failed": 0, "errors": []} + stats: dict[str, Any] = {"restored": 0, "skipped": 0, "failed": 0, "errors": []} # List snapshots at backup location diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/transfer.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/transfer.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/transfer.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/transfer.py 2026-01-05 05:02:00.000000000 +0000 @@ -7,10 +7,19 @@ import shutil import subprocess -from typing import Optional +from typing import Optional, TypedDict logger = logging.getLogger(__name__) + +class CompressionConfig(TypedDict): + """Type definition for compression program configuration.""" + + compress: list[str] + decompress: list[str] + check: str + + # Available compression programs with their compress/decompress commands -COMPRESSION_PROGRAMS = { +COMPRESSION_PROGRAMS: dict[str, CompressionConfig] = { "gzip": { "compress": ["gzip", "-c"], diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/verify.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/verify.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/verify.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/core/verify.py 2026-01-05 05:06:05.000000000 +0000 @@ -16,5 +16,5 @@ from enum import Enum from pathlib import Path -from typing import Callable +from typing import Any, Callable from .. import __util__ @@ -328,5 +328,5 @@ # Verify temp dir is on btrfs - if not __util__.is_btrfs(temp_path): + if not __util__.is_btrfs(temp_path): # type: ignore[attr-defined] report.errors.append(f"Temp directory {temp_path} is not on btrfs filesystem") report.completed_at = time.time() @@ -364,5 +364,5 @@ ) - restored = [] + restored: list[Any] = [] for i, snap in enumerate(to_verify, 1): @@ -392,5 +392,5 @@ raise VerifyError(f"Restored snapshot not found at {restored_path}") - if not __util__.is_subvolume(restored_path): + if not __util__.is_subvolume(restored_path): # type: ignore[attr-defined] raise VerifyError( f"Restored path {restored_path} is not a valid subvolume" @@ -424,6 +424,6 @@ for snap in to_verify: snap_path = temp_path / snap.get_name() - if snap_path.exists() and __util__.is_subvolume(snap_path): - __util__.delete_subvolume(snap_path) + if snap_path.exists() and __util__.is_subvolume(snap_path): # type: ignore[attr-defined] + __util__.delete_subvolume(snap_path) # type: ignore[attr-defined] # Remove temp dir if temp_path.exists(): diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/__init__.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/__init__.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/__init__.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/__init__.py 2026-01-05 05:24:12.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: standard - """btrfs-backup-ng: btrfs_backup_ng/endpoint/__init__.py.""" @@ -9,5 +7,5 @@ from .local import LocalEndpoint from .shell import ShellEndpoint -from .ssh import SSHEndpoint +from .ssh import SSHEndpoint # type: ignore[attr-defined] diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/common.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/common.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/common.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/common.py 2026-01-05 05:24:13.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: standard - """btrfs-backup-ng: btrfs_backup_ng/endpoint/common.py Common functionality among modules. diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/local.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/local.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/local.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/local.py 2026-01-05 05:24:13.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: standard - """btrfs-backup-ng: btrfs_backup_ng/endpoint/local.py Create commands with local endpoints. @@ -115,5 +113,5 @@ self.config["source"] is not None and self.config["fs_checks"] - and not __util__.is_subvolume(self.config["source"]) + and not __util__.is_subvolume(self.config["source"]) # type: ignore[attr-defined] ): logger.error( @@ -124,5 +122,5 @@ ) - if self.config["fs_checks"] and not __util__.is_btrfs(self.config["path"]): + if self.config["fs_checks"] and not __util__.is_btrfs(self.config["path"]): # type: ignore[attr-defined] logger.error( "%s does not seem to be on a btrfs filesystem", self.config["path"] diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/shell.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/shell.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/shell.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/shell.py 2026-01-05 05:24:12.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: standard - """btrfs-backup-ng: btrfs_backup_ng/endpoint/shell.py Create destinations with shell command endpoints. diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/ssh.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/ssh.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/ssh.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint/ssh.py 2026-01-05 23:29:13.000000000 +0000 @@ -1,4 +1,2 @@ -# pyright: strict - """btrfs-backup-ng: SSH Endpoint for managing remote operations. @@ -27,22 +25,31 @@ import threading import time +import types import uuid from pathlib import Path from subprocess import CompletedProcess from threading import Lock -from typing import Any, Dict, List, Optional, Tuple, TypeVar, cast +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, TypeVar, cast +# Handle paramiko import with proper typing +paramiko: Optional[types.ModuleType] try: - import paramiko + import paramiko as _paramiko + paramiko = _paramiko PARAMIKO_AVAILABLE = True except ImportError: - paramiko = None # type: ignore + paramiko = None PARAMIKO_AVAILABLE = False +if TYPE_CHECKING: + pass + +# Handle pwd import with proper typing +_pwd: Optional[types.ModuleType] try: - import pwd + import pwd as _pwd_module - _pwd = pwd + _pwd = _pwd_module _pwd_available = True except ImportError: @@ -51,9 +58,11 @@ -from btrfs_backup_ng import __util__ -from btrfs_backup_ng.__logger__ import logger -from btrfs_backup_ng.sshutil.master import SSHMasterManager +from btrfs_backup_ng import __util__ # noqa: E402 +from btrfs_backup_ng.__logger__ import logger # noqa: E402 +from btrfs_backup_ng.sshutil.master import SSHMasterManager # noqa: E402 + +from .common import Endpoint # noqa: E402 -from .common import Endpoint +__all__ = ["SSHEndpoint"] # Type variable for self in SSHEndpoint @@ -127,5 +136,4 @@ self._last_receive_log: Optional[str] = None self._last_transfer_snapshot: Optional[bool] = None - self.ssh_manager: SSHMasterManager logger.debug( "SSHEndpoint: Config keys before parent init: %s", list(config.keys()) @@ -227,5 +235,5 @@ sudo_user = os.environ.get("SUDO_USER") sudo_user_home = None - if _pwd_available and _pwd is not None: + if _pwd_available and _pwd is not None and sudo_user is not None: try: sudo_user_home = _pwd.getpwnam(sudo_user).pw_dir @@ -780,15 +788,15 @@ if all_passed: logger.debug("All diagnostic tests passed") - for test, result in results.items(): - logger.debug(f"Test {test}: PASSED") + for test_name, test_passed in results.items(): + logger.debug(f"Test {test_name}: PASSED") else: # Show summary at INFO level only for failures - failed_tests = [test for test, result in results.items() if not result] + failed_tests = [t for t, passed in results.items() if not passed] logger.debug(f"Some diagnostic tests failed: {failed_tests}") logger.info("\nDiagnostic Summary:") logger.info("-" * 50) - for test, result in results.items(): - status = "PASS" if result else "FAIL" - logger.info(f"{test.replace('_', ' ').title():20} {status}") + for test_name, test_passed in results.items(): + status = "PASS" if test_passed else "FAIL" + logger.info(f"{test_name.replace('_', ' ').title():20} {status}") logger.info("-" * 50) @@ -2394,5 +2402,5 @@ # Build local btrfs send command send_cmd = ["sudo", "btrfs", "send"] - if is_incremental: + if is_incremental and parent_path is not None: send_cmd.extend(["-p", parent_path]) logger.info(f"Using parent: {os.path.basename(parent_path)}") @@ -2403,5 +2411,10 @@ remote_cmd = f"sudo -S btrfs receive {escaped_dest}" - client: Optional[paramiko.SSHClient] = None + # Ensure paramiko is available + if paramiko is None: + logger.error("Paramiko is not available for SSH transfer") + return False + + client: Optional[Any] = None send_proc: Optional[subprocess.Popen[bytes]] = None @@ -2557,11 +2570,13 @@ return False - except paramiko.AuthenticationException as e: - logger.error(f"SSH authentication failed: {e}") - return False - except paramiko.SSHException as e: - logger.error(f"SSH error: {e}") - return False except Exception as e: + # Handle paramiko-specific exceptions + if paramiko is not None: + if isinstance(e, paramiko.AuthenticationException): + logger.error(f"SSH authentication failed: {e}") + return False + if isinstance(e, paramiko.SSHException): + logger.error(f"SSH error: {e}") + return False logger.error(f"Transfer failed: {e}") return False @@ -2611,5 +2626,5 @@ # Build send command send_parts = ["sudo", "btrfs", "send"] - if is_incremental: + if is_incremental and parent_path is not None: send_parts.extend(["-p", shlex.quote(parent_path)]) logger.info(f"Using parent: {os.path.basename(parent_path)}") @@ -2662,6 +2677,7 @@ def stream_stderr() -> None: - if proc.stderr: - for chunk in iter(lambda: proc.stderr.read(80), b""): + stderr_stream = proc.stderr + if stderr_stream is not None: + for chunk in iter(lambda: stderr_stream.read(80), b""): if chunk: text = chunk.decode(errors="replace") @@ -2729,6 +2745,6 @@ logger.error("SSH master connection is down") try: - self.ssh_manager.stop() - self.ssh_manager.start() + self.ssh_manager.stop_master() + self.ssh_manager.start_master() if not self._is_master_active(): logger.error("Failed to re-establish SSH master connection") Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/endpoint: ssh.pyi diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/notifications.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/notifications.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/notifications.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/notifications.py 2026-01-05 05:00:40.000000000 +0000 @@ -10,5 +10,5 @@ import urllib.error import urllib.request -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, field from datetime import datetime from email.mime.multipart import MIMEMultipart @@ -53,9 +53,5 @@ snapshots_pruned: int = 0 duration_seconds: float = 0.0 - errors: list[str] = None - - def __post_init__(self): - if self.errors is None: - self.errors = [] + errors: list[str] = field(default_factory=list) def to_dict(self) -> dict: @@ -195,12 +191,8 @@ smtp_tls: str = "none" # "ssl", "starttls", or "none" from_addr: str = "btrfs-backup-ng@localhost" - to_addrs: list[str] = None + to_addrs: list[str] = field(default_factory=list) on_success: bool = False on_failure: bool = True - def __post_init__(self): - if self.to_addrs is None: - self.to_addrs = [] - @dataclass @@ -221,13 +213,9 @@ url: Optional[str] = None method: str = "POST" - headers: dict[str, str] = None + headers: dict[str, str] = field(default_factory=dict) on_success: bool = False on_failure: bool = True timeout: int = 30 - def __post_init__(self): - if self.headers is None: - self.headers = {} - @dataclass @@ -240,12 +228,6 @@ """ - email: EmailConfig = None - webhook: WebhookConfig = None - - def __post_init__(self): - if self.email is None: - self.email = EmailConfig() - if self.webhook is None: - self.webhook = WebhookConfig() + email: EmailConfig = field(default_factory=EmailConfig) + webhook: WebhookConfig = field(default_factory=WebhookConfig) def is_enabled(self) -> bool: @@ -407,5 +389,5 @@ transfers_failed: int = 0, duration_seconds: float = 0.0, - errors: list[str] = None, + errors: Optional[list[str]] = None, ) -> NotificationEvent: """Create a backup completion notification event. @@ -458,5 +440,5 @@ snapshots_pruned: int = 0, duration_seconds: float = 0.0, - errors: list[str] = None, + errors: Optional[list[str]] = None, ) -> NotificationEvent: """Create a prune completion notification event. diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/retention.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/retention.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/retention.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/retention.py 2026-01-05 05:18:00.000000000 +0000 @@ -21,5 +21,5 @@ from dataclasses import dataclass from datetime import datetime, timedelta -from typing import Callable +from typing import Any, Callable from .config import RetentionConfig @@ -182,5 +182,5 @@ snapshots: list, config: RetentionConfig, - get_name: Callable[[object], str] | None = None, + get_name: Callable[[Any], str] | None = None, prefix: str = "", now: datetime | None = None, diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/sshutil/diagnose.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/sshutil/diagnose.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/sshutil/diagnose.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/sshutil/diagnose.py 2026-01-05 23:01:18.000000000 +0000 @@ -1,5 +1,3 @@ -#!/usr/bin/env python3 -""" -SSH and Sudo Diagnostic Tool for btrfs-backup-ng +"""SSH and Sudo Diagnostic Tool for btrfs-backup-ng This script helps diagnose common SSH and sudo configuration issues diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/transaction.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/transaction.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/transaction.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng/transaction.py 2026-01-05 04:58:52.000000000 +0000 @@ -14,5 +14,5 @@ from datetime import datetime, timezone from pathlib import Path -from typing import Any +from typing import Any, Literal logger = logging.getLogger(__name__) @@ -70,5 +70,5 @@ return - record = { + record: dict[str, Any] = { "timestamp": datetime.now(timezone.utc).isoformat(), "action": action, @@ -147,5 +147,5 @@ return self - def __exit__(self, exc_type, exc_val, exc_tb) -> bool: + def __exit__(self, exc_type, exc_val, exc_tb) -> Literal[False]: import time @@ -274,11 +274,8 @@ } - stats = { - "total_records": len(records), - "transfers": {"completed": 0, "failed": 0}, - "snapshots": {"completed": 0, "failed": 0}, - "deletes": {"completed": 0, "failed": 0}, - "total_bytes_transferred": 0, - } + transfers = {"completed": 0, "failed": 0} + snapshots = {"completed": 0, "failed": 0} + deletes = {"completed": 0, "failed": 0} + total_bytes = 0 for record in records: @@ -288,19 +285,25 @@ if action == "transfer": if status == "completed": - stats["transfers"]["completed"] += 1 + transfers["completed"] += 1 if record.get("size_bytes"): - stats["total_bytes_transferred"] += record["size_bytes"] + total_bytes += record["size_bytes"] elif status == "failed": - stats["transfers"]["failed"] += 1 + transfers["failed"] += 1 elif action == "snapshot": if status == "completed": - stats["snapshots"]["completed"] += 1 + snapshots["completed"] += 1 elif status == "failed": - stats["snapshots"]["failed"] += 1 + snapshots["failed"] += 1 elif action in ("delete", "prune"): if status == "completed": - stats["deletes"]["completed"] += 1 + deletes["completed"] += 1 elif status == "failed": - stats["deletes"]["failed"] += 1 + deletes["failed"] += 1 - return stats + return { + "total_records": len(records), + "transfers": transfers, + "snapshots": snapshots, + "deletes": deletes, + "total_bytes_transferred": total_bytes, + } diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/PKG-INFO /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/PKG-INFO --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/PKG-INFO 2026-01-04 23:42:44.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/PKG-INFO 2026-01-06 01:47:11.000000000 +0000 @@ -4,5 +4,4 @@ Summary: Swiss army knife cli tool for atomic, incremental, intelligent, feature-rich backups for btrfs Author-email: Michael Berry -License-Expression: MIT Project-URL: Homepage, https://github.com/berrym/btrfs-backup-ng Classifier: Programming Language :: Python :: 3 @@ -22,4 +21,5 @@ Requires-Dist: ruff>=0.1.0; extra == "dev" Requires-Dist: mypy>=1.0; extra == "dev" +Requires-Dist: types-paramiko>=3.0; extra == "dev" Dynamic: license-file @@ -27,6 +27,10 @@ [![CI](https://github.com/berrym/btrfs-backup-ng/actions/workflows/ci.yml/badge.svg)](https://github.com/berrym/btrfs-backup-ng/actions/workflows/ci.yml) +[![codecov](https://codecov.io/gh/berrym/btrfs-backup-ng/graph/badge.svg)](https://codecov.io/gh/berrym/btrfs-backup-ng) [![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) +[![Checked with mypy](https://www.mypy-lang.org/static/mypy_badge.svg)](https://mypy-lang.org/) +[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff) +[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black) A modern, feature-rich tool for automated BTRFS snapshot backup management with TOML configuration, time-based retention policies, and robust SSH transfer support. @@ -90,12 +94,49 @@ ### Shell Completions -Shell completion scripts are available in the `completions/` directory for bash, zsh, and fish. See `completions/README.md` for installation instructions. +Install shell completions for tab-completion support: + +```bash +# Install for your shell (bash, zsh, or fish) +btrfs-backup-ng completions install --shell bash +btrfs-backup-ng completions install --shell zsh +btrfs-backup-ng completions install --shell fish + +# System-wide installation (requires root) +sudo btrfs-backup-ng completions install --shell bash --system + +# Show path to completion scripts +btrfs-backup-ng completions path +``` + +See `completions/README.md` for manual installation instructions. ### Man Pages -Manual pages are available in the `man/` directory. See `man/README.md` for installation instructions. View without installing: +Install man pages for offline documentation: + +```bash +# User installation (to ~/.local/share/man) +btrfs-backup-ng manpages install + +# System-wide installation (requires root) +sudo btrfs-backup-ng manpages install --system + +# Custom prefix +btrfs-backup-ng manpages install --prefix /opt/myapp + +# Show path to man page files +btrfs-backup-ng manpages path +``` + +After user installation, add to your MANPATH: +```bash +export MANPATH="$HOME/.local/share/man:$MANPATH" +``` +View man pages: ```bash -man ./man/btrfs-backup-ng.1 +man btrfs-backup-ng +man btrfs-backup-ng-restore +man btrfs-backup-ng-config ``` diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/SOURCES.txt /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/SOURCES.txt --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/SOURCES.txt 2026-01-04 23:42:44.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/SOURCES.txt 2026-01-06 01:47:11.000000000 +0000 @@ -1,5 +1,25 @@ +CHANGELOG.md +CONTRIBUTING.md LICENSE +MANIFEST.in README.md pyproject.toml +completions/README.md +completions/btrfs-backup-ng.bash +completions/btrfs-backup-ng.fish +completions/btrfs-backup-ng.zsh +man/README.md +man/man1/btrfs-backup-ng-config.1 +man/man1/btrfs-backup-ng-estimate.1 +man/man1/btrfs-backup-ng-install.1 +man/man1/btrfs-backup-ng-list.1 +man/man1/btrfs-backup-ng-prune.1 +man/man1/btrfs-backup-ng-restore.1 +man/man1/btrfs-backup-ng-run.1 +man/man1/btrfs-backup-ng-snapshot.1 +man/man1/btrfs-backup-ng-status.1 +man/man1/btrfs-backup-ng-transfer.1 +man/man1/btrfs-backup-ng-verify.1 +man/man1/btrfs-backup-ng.1 src/btrfs_backup_ng/__init__.py src/btrfs_backup_ng/__logger__.py @@ -21,4 +41,5 @@ src/btrfs_backup_ng/cli/__init__.py src/btrfs_backup_ng/cli/common.py +src/btrfs_backup_ng/cli/completions.py src/btrfs_backup_ng/cli/config_cmd.py src/btrfs_backup_ng/cli/dispatcher.py @@ -26,4 +47,5 @@ src/btrfs_backup_ng/cli/install.py src/btrfs_backup_ng/cli/list_cmd.py +src/btrfs_backup_ng/cli/manpages.py src/btrfs_backup_ng/cli/prune.py src/btrfs_backup_ng/cli/restore.py @@ -51,10 +73,12 @@ src/btrfs_backup_ng/endpoint/shell.py src/btrfs_backup_ng/endpoint/ssh.py -src/btrfs_backup_ng/endpoint/ssh.pyi src/btrfs_backup_ng/sshutil/diagnose.py src/btrfs_backup_ng/sshutil/master.py +tests/__init__.py +tests/conftest.py tests/test_btrbk_import.py tests/test_cli.py tests/test_cli_common.py +tests/test_cli_verify.py tests/test_config_cmd.py tests/test_config_loader.py @@ -62,7 +86,20 @@ tests/test_estimate.py tests/test_logger.py +tests/test_notifications.py +tests/test_progress.py tests/test_restore.py tests/test_retention.py +tests/test_transaction.py tests/test_transfer.py tests/test_util.py -tests/test_verify.py \ No newline at end of file +tests/test_verify.py +tests/integration/__init__.py +tests/integration/conftest.py +tests/integration/test_cli_e2e.py +tests/integration/test_config_flow.py +tests/integration/test_endpoints.py +tests/integration/test_retention_flow.py +tests/integration/tier2/__init__.py +tests/integration/tier2/conftest.py +tests/integration/tier2/test_btrfs_operations.py +tests/integration/tier2/test_endpoints_real.py \ No newline at end of file diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/requires.txt /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/requires.txt --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/requires.txt 2026-01-04 23:42:44.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/src/btrfs_backup_ng.egg-info/requires.txt 2026-01-06 01:47:11.000000000 +0000 @@ -8,4 +8,5 @@ ruff>=0.1.0 mypy>=1.0 +types-paramiko>=3.0 [test] Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests: __init__.py Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests: conftest.py Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests: integration Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests: test_cli_verify.py diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/tests/test_estimate.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests/test_estimate.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/tests/test_estimate.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests/test_estimate.py 2026-01-05 07:28:48.000000000 +0000 @@ -1,12 +1,16 @@ """Tests for backup size estimation functionality.""" +import subprocess from unittest.mock import MagicMock, patch - from btrfs_backup_ng.core.estimate import ( SnapshotEstimate, TransferEstimate, _parse_size, + estimate_incremental_size, + estimate_snapshot_full_size, + estimate_transfer, format_size, + print_estimate, ) @@ -265,4 +269,494 @@ +class TestEstimateIncrementalSize: + """Tests for estimate_incremental_size function.""" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_send_no_data_success(self, mock_euid, mock_run, tmp_path): + """Test successful estimation via btrfs send --no-data.""" + # Return a mock stdout with some bytes to represent stream size + mock_run.return_value = MagicMock( + returncode=0, + stdout=b"x" * 50000, # 50KB mock stream + ) + + size, method = estimate_incremental_size(tmp_path / "snap2", tmp_path / "snap1") + + assert size == 50000 + assert method == "send_no_data" + # Verify the command was called correctly + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "btrfs" in cmd + assert "send" in cmd + assert "--no-data" in cmd + assert "-p" in cmd + + @patch("subprocess.run") + @patch("os.geteuid", return_value=1000) + def test_send_no_data_with_sudo(self, mock_euid, mock_run, tmp_path): + """Test estimation uses sudo when not root.""" + mock_run.return_value = MagicMock( + returncode=0, + stdout=b"x" * 1000, + ) + + size, method = estimate_incremental_size( + tmp_path / "snap2", tmp_path / "snap1", use_sudo=True + ) + + assert method == "send_no_data" + cmd = mock_run.call_args[0][0] + assert cmd[0] == "sudo" + assert cmd[1] == "-n" + + @patch("btrfs_backup_ng.core.estimate.estimate_snapshot_full_size") + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_fallback_to_size_diff(self, mock_euid, mock_run, mock_full_size, tmp_path): + """Test fallback to size difference when send --no-data fails.""" + # send --no-data fails + mock_run.return_value = MagicMock(returncode=1, stdout=b"", stderr=b"error") + + # Full size estimation succeeds + mock_full_size.side_effect = [ + (1024 * 1024 * 100, "du"), # snap2: 100 MiB + (1024 * 1024 * 80, "du"), # snap1: 80 MiB + ] + + size, method = estimate_incremental_size(tmp_path / "snap2", tmp_path / "snap1") + + assert size == 1024 * 1024 * 20 # 20 MiB difference + assert method == "size_diff" + + @patch("btrfs_backup_ng.core.estimate.estimate_snapshot_full_size") + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_size_diff_negative_returns_zero( + self, mock_euid, mock_run, mock_full_size, tmp_path + ): + """Test that negative size diff returns 0.""" + mock_run.return_value = MagicMock(returncode=1, stdout=b"", stderr=b"error") + + # Parent is larger than snapshot (e.g., files deleted) + mock_full_size.side_effect = [ + (1024 * 1024 * 50, "du"), # snap2: 50 MiB + (1024 * 1024 * 100, "du"), # snap1: 100 MiB + ] + + size, method = estimate_incremental_size(tmp_path / "snap2", tmp_path / "snap1") + + assert size == 0 # max(0, negative) = 0 + assert method == "size_diff" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_timeout_failure(self, mock_euid, mock_run, tmp_path): + """Test handling of timeout.""" + mock_run.side_effect = subprocess.TimeoutExpired("btrfs", 300) + + size, method = estimate_incremental_size(tmp_path / "snap2", tmp_path / "snap1") + + assert size is None + assert method == "failed" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_file_not_found_failure(self, mock_euid, mock_run, tmp_path): + """Test handling of FileNotFoundError (btrfs not installed).""" + mock_run.side_effect = FileNotFoundError("btrfs not found") + + size, method = estimate_incremental_size(tmp_path / "snap2", tmp_path / "snap1") + + assert size is None + assert method == "failed" + + +class TestEstimateSnapshotFullSizeAdvanced: + """Additional tests for estimate_snapshot_full_size function.""" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=1000) + def test_uses_sudo_when_not_root(self, mock_euid, mock_run, tmp_path): + """Test that sudo is used when not running as root.""" + mock_run.return_value = MagicMock( + returncode=0, + stdout="Exclusive:\t\t1.00GiB\n", + ) + + estimate_snapshot_full_size(tmp_path / "snap", use_sudo=True) + + cmd = mock_run.call_args[0][0] + assert cmd[0] == "sudo" + assert cmd[1] == "-n" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_no_sudo_when_root(self, mock_euid, mock_run, tmp_path): + """Test that sudo is not used when running as root.""" + mock_run.return_value = MagicMock( + returncode=0, + stdout="Exclusive:\t\t1.00GiB\n", + ) + + estimate_snapshot_full_size(tmp_path / "snap") + + cmd = mock_run.call_args[0][0] + assert cmd[0] == "btrfs" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_timeout_handling(self, mock_euid, mock_run, tmp_path): + """Test handling of command timeout.""" + mock_run.side_effect = subprocess.TimeoutExpired("btrfs", 30) + + size, method = estimate_snapshot_full_size(tmp_path / "snap") + + assert size is None + assert method == "failed" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_exclusive_size_zero_falls_through(self, mock_euid, mock_run, tmp_path): + """Test that zero exclusive size falls through to next method.""" + mock_run.side_effect = [ + # subvolume show returns 0 for exclusive + MagicMock(returncode=0, stdout="Exclusive:\t\t0B\n"), + # filesystem du succeeds + MagicMock(returncode=0, stdout="Total\tExclusive\n12345\t12345\n"), + ] + + size, method = estimate_snapshot_full_size(tmp_path / "snap") + + assert size == 12345 + assert method == "filesystem_du" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_oserror_handling(self, mock_euid, mock_run, tmp_path): + """Test handling of OSError.""" + mock_run.side_effect = OSError("Permission denied") + + size, method = estimate_snapshot_full_size(tmp_path / "snap") + + assert size is None + assert method == "failed" + + @patch("subprocess.run") + @patch("os.geteuid", return_value=0) + def test_du_value_error_fallthrough(self, mock_euid, mock_run, tmp_path): + """Test that non-integer du output falls through.""" + mock_run.side_effect = [ + MagicMock(returncode=1, stdout=""), # subvolume show fails + MagicMock(returncode=1, stdout=""), # filesystem du fails + MagicMock(returncode=0, stdout="not_a_number\t/path\n"), # du bad output + ] + + size, method = estimate_snapshot_full_size(tmp_path / "snap") + + assert size is None + assert method == "failed" + + +class TestEstimateTransfer: + """Tests for estimate_transfer function.""" + + def test_empty_source(self): + """Test with no snapshots at source.""" + source = MagicMock() + source.list_snapshots.return_value = [] + source.config = {"path": "/source", "ssh_sudo": False} + + dest = MagicMock() + dest.list_snapshots.return_value = [] + + result = estimate_transfer(source, dest) + + assert result.snapshot_count == 0 + assert result.new_snapshot_count == 0 + assert result.skipped_count == 0 + + def test_all_snapshots_at_destination(self): + """Test when all snapshots already exist at destination.""" + snap1 = MagicMock() + snap1.get_name.return_value = "snap-1" + snap1.time_obj = 1000 + + source = MagicMock() + source.list_snapshots.return_value = [snap1] + source.config = {"path": "/source", "ssh_sudo": False} + + dest = MagicMock() + dest.list_snapshots.return_value = [snap1] + + result = estimate_transfer(source, dest) + + assert result.snapshot_count == 0 + assert result.skipped_count == 1 + + @patch("btrfs_backup_ng.core.estimate.estimate_snapshot_full_size") + def test_new_snapshot_full_transfer(self, mock_full_size, tmp_path): + """Test estimating a new snapshot for full transfer.""" + snap1 = MagicMock() + snap1.get_name.return_value = "snap-1" + snap1.time_obj = 1000 + + source = MagicMock() + source.list_snapshots.return_value = [snap1] + source.config = {"path": tmp_path, "ssh_sudo": False} + + dest = MagicMock() + dest.list_snapshots.return_value = [] + + mock_full_size.return_value = (1024 * 1024 * 100, "filesystem_du") + + result = estimate_transfer(source, dest) + + assert result.snapshot_count == 1 + assert result.new_snapshot_count == 1 + assert result.total_full_size == 1024 * 1024 * 100 + assert len(result.snapshots) == 1 + assert result.snapshots[0].name == "snap-1" + + @patch("btrfs_backup_ng.core.estimate.estimate_incremental_size") + @patch("btrfs_backup_ng.core.estimate.estimate_snapshot_full_size") + def test_incremental_transfer_estimation( + self, mock_full_size, mock_incr_size, tmp_path + ): + """Test estimating incremental transfer between snapshots.""" + snap1 = MagicMock() + snap1.get_name.return_value = "snap-1" + snap1.time_obj = 1000 + + snap2 = MagicMock() + snap2.get_name.return_value = "snap-2" + snap2.time_obj = 2000 + + # Create real directory so path.exists() works + (tmp_path / "snap-1").mkdir() + + source = MagicMock() + source.list_snapshots.return_value = [snap1, snap2] + source.config = {"path": tmp_path, "ssh_sudo": False} + + dest = MagicMock() + dest.list_snapshots.return_value = [] + + mock_full_size.return_value = (1024 * 1024 * 100, "filesystem_du") + mock_incr_size.return_value = (1024 * 1024 * 10, "send_no_data") + + result = estimate_transfer(source, dest) + + assert result.snapshot_count == 2 + assert result.new_snapshot_count == 2 + # Second snapshot should be incremental + assert result.snapshots[1].is_incremental is True + assert result.snapshots[1].parent_name == "snap-1" + + def test_dest_list_snapshots_fails(self, tmp_path): + """Test handling when destination list_snapshots raises exception.""" + snap1 = MagicMock() + snap1.get_name.return_value = "snap-1" + snap1.time_obj = 1000 + + source = MagicMock() + source.list_snapshots.return_value = [snap1] + source.config = {"path": tmp_path, "ssh_sudo": False} + + dest = MagicMock() + dest.list_snapshots.side_effect = Exception("Connection failed") + + # Should not raise, should treat as empty destination + with patch( + "btrfs_backup_ng.core.estimate.estimate_snapshot_full_size" + ) as mock_full: + mock_full.return_value = (1000, "du") + result = estimate_transfer(source, dest) + + assert result.new_snapshot_count == 1 + + def test_explicit_snapshot_list(self, tmp_path): + """Test with explicit snapshot list instead of listing from source.""" + snap1 = MagicMock() + snap1.get_name.return_value = "snap-1" + snap1.time_obj = 1000 + + source = MagicMock() + source.config = {"path": tmp_path, "ssh_sudo": False} + + dest = MagicMock() + dest.list_snapshots.return_value = [] + + with patch( + "btrfs_backup_ng.core.estimate.estimate_snapshot_full_size" + ) as mock_full: + mock_full.return_value = (5000, "du") + result = estimate_transfer(source, dest, snapshots=[snap1]) + + # Should use provided list, not call list_snapshots + source.list_snapshots.assert_not_called() + assert result.snapshot_count == 1 + + def test_uses_ssh_sudo_from_config(self, tmp_path): + """Test that ssh_sudo config is passed to estimation functions.""" + snap1 = MagicMock() + snap1.get_name.return_value = "snap-1" + snap1.time_obj = 1000 + + source = MagicMock() + source.list_snapshots.return_value = [snap1] + source.config = {"path": tmp_path, "ssh_sudo": True} + + dest = MagicMock() + dest.list_snapshots.return_value = [] + + with patch( + "btrfs_backup_ng.core.estimate.estimate_snapshot_full_size" + ) as mock_full: + mock_full.return_value = (1000, "du") + estimate_transfer(source, dest) + + # Check use_sudo was passed (second positional arg) + mock_full.assert_called_once() + # Args are (snap_path, use_sudo) + assert mock_full.call_args[0][1] is True + + +class TestPrintEstimate: + """Tests for print_estimate function.""" + + def test_print_empty_estimate(self, capsys): + """Test printing estimate with no new snapshots.""" + estimate = TransferEstimate(skipped_count=5) + + print_estimate(estimate, "source", "dest") + + captured = capsys.readouterr() + assert "source" in captured.out + assert "dest" in captured.out + assert "already at destination: 5" in captured.out + assert "No new snapshots to transfer" in captured.out + + def test_print_full_snapshot(self, capsys): + """Test printing estimate with full snapshot.""" + estimate = TransferEstimate() + snap = SnapshotEstimate( + name="snap-2024-01-01", + full_size=1024 * 1024 * 500, # 500 MiB + ) + estimate.add_snapshot(snap) + estimate.new_snapshot_count = 1 + + print_estimate(estimate, "local", "remote") + + captured = capsys.readouterr() + assert "snap-2024-01-01" in captured.out + assert "full" in captured.out + assert "MiB" in captured.out + + def test_print_incremental_snapshot(self, capsys): + """Test printing estimate with incremental snapshot.""" + estimate = TransferEstimate() + snap = SnapshotEstimate( + name="snap-2024-01-02", + full_size=1024 * 1024 * 500, + incremental_size=1024 * 1024 * 50, + parent_name="snap-2024-01-01", + is_incremental=True, + ) + estimate.add_snapshot(snap) + estimate.new_snapshot_count = 1 + + print_estimate(estimate, "local", "remote") + + captured = capsys.readouterr() + assert "snap-2024-01-02" in captured.out + assert "incremental" in captured.out + assert "snap-2024-01-01" in captured.out + + def test_print_long_snapshot_name_truncated(self, capsys): + """Test that long snapshot names are truncated.""" + estimate = TransferEstimate() + long_name = "snapshot-with-a-very-long-name-that-exceeds-forty-characters" + snap = SnapshotEstimate( + name=long_name, + full_size=1024, + ) + estimate.add_snapshot(snap) + estimate.new_snapshot_count = 1 + + print_estimate(estimate) + + captured = capsys.readouterr() + assert ".." in captured.out # Truncation indicator + + def test_print_multiple_snapshots(self, capsys): + """Test printing estimate with multiple snapshots.""" + estimate = TransferEstimate() + + snap1 = SnapshotEstimate(name="snap-1", full_size=1024 * 1024) + snap2 = SnapshotEstimate( + name="snap-2", + full_size=1024 * 1024, + incremental_size=512 * 1024, + parent_name="snap-1", + is_incremental=True, + ) + + estimate.add_snapshot(snap1) + estimate.add_snapshot(snap2) + estimate.new_snapshot_count = 2 + estimate.estimation_time = 1.5 + + print_estimate(estimate) + + captured = capsys.readouterr() + assert "Snapshots to transfer: 2" in captured.out + assert "snap-1" in captured.out + assert "snap-2" in captured.out + assert "1.50s" in captured.out + + def test_print_totals(self, capsys): + """Test that totals are printed correctly.""" + estimate = TransferEstimate() + snap = SnapshotEstimate( + name="snap-1", + full_size=1024 * 1024 * 1024, # 1 GiB + ) + estimate.add_snapshot(snap) + estimate.new_snapshot_count = 1 + + print_estimate(estimate) + + captured = capsys.readouterr() + assert "Total data to transfer" in captured.out + assert "GiB" in captured.out + + +class TestParseSizeEdgeCases: + """Additional edge case tests for _parse_size.""" + + def test_parse_with_whitespace(self): + """Test parsing size with whitespace.""" + assert _parse_size(" 1GiB ") == 1024**3 + # Space in middle is actually supported due to strip in value parsing + assert _parse_size("1 GiB") == 1024**3 + + def test_parse_float_value(self): + """Test parsing float without unit.""" + assert _parse_size("1.5") == 1 + + def test_parse_large_tib(self): + """Test parsing large TiB value.""" + assert _parse_size("10TiB") == 10 * 1024**4 + + def test_parse_case_sensitivity(self): + """Test that parsing is case-sensitive.""" + # Our implementation is case-sensitive + assert _parse_size("1gib") is None + assert _parse_size("1GIB") is None + + class TestExecuteEstimate: """Tests for execute_estimate CLI function.""" @@ -304,6 +798,7 @@ mock_find.return_value = str(tmp_path / "config.toml") - mock_load.return_value = Config( - volumes=[VolumeConfig(path="/var/log", snapshot_prefix="logs")] + mock_load.return_value = ( + Config(volumes=[VolumeConfig(path="/var/log", snapshot_prefix="logs")]), + [], ) Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests: test_notifications.py Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests: test_progress.py diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/tests/test_restore.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests/test_restore.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/tests/test_restore.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests/test_restore.py 2026-01-05 07:46:48.000000000 +0000 @@ -1,6 +1,7 @@ """Tests for restore functionality.""" +import argparse import time -from unittest.mock import MagicMock, patch +from unittest.mock import ANY, MagicMock, patch import pytest @@ -28,4 +29,13 @@ return self._name + def find_parent(self, snapshots: list): + """Find the most recent snapshot older than this one.""" + candidates = [s for s in snapshots if s < self] + if not candidates: + return None + return max( + candidates, key=lambda s: s.time_obj if hasattr(s, "time_obj") else 0 + ) + def __lt__(self, other): """Compare by time for sorting.""" @@ -746,5 +756,5 @@ config_path = tmp_path / "config.toml" mock_find.return_value = str(config_path) - mock_load.return_value = Config() + mock_load.return_value = (Config(), []) args = MagicMock() @@ -784,5 +794,5 @@ ] ) - mock_load.return_value = config + mock_load.return_value = (config, []) args = MagicMock() @@ -819,6 +829,7 @@ mock_find.return_value = str(tmp_path / "config.toml") - mock_load.return_value = Config( - volumes=[VolumeConfig(path="/var/log", snapshot_prefix="logs")] + mock_load.return_value = ( + Config(volumes=[VolumeConfig(path="/var/log", snapshot_prefix="logs")]), + [], ) @@ -838,6 +849,9 @@ mock_find.return_value = str(tmp_path / "config.toml") - mock_load.return_value = Config( - volumes=[VolumeConfig(path="/home", snapshot_prefix="home", targets=[])] + mock_load.return_value = ( + Config( + volumes=[VolumeConfig(path="/home", snapshot_prefix="home", targets=[])] + ), + [], ) @@ -857,12 +871,15 @@ mock_find.return_value = str(tmp_path / "config.toml") - mock_load.return_value = Config( - volumes=[ - VolumeConfig( - path="/home", - snapshot_prefix="home", - targets=[TargetConfig(path="/mnt/backup/home")], - ) - ] + mock_load.return_value = ( + Config( + volumes=[ + VolumeConfig( + path="/home", + snapshot_prefix="home", + targets=[TargetConfig(path="/mnt/backup/home")], + ) + ] + ), + [], ) @@ -884,16 +901,19 @@ mock_find.return_value = str(tmp_path / "config.toml") - mock_load.return_value = Config( - volumes=[ - VolumeConfig( - path="/home", - snapshot_prefix="home", - targets=[ - TargetConfig( - path="ssh://backup@server:/backups/home", ssh_sudo=True - ) - ], - ) - ] + mock_load.return_value = ( + Config( + volumes=[ + VolumeConfig( + path="/home", + snapshot_prefix="home", + targets=[ + TargetConfig( + path="ssh://backup@server:/backups/home", ssh_sudo=True + ) + ], + ) + ] + ), + [], ) mock_list.return_value = 0 @@ -922,12 +942,15 @@ mock_find.return_value = str(tmp_path / "config.toml") - mock_load.return_value = Config( - volumes=[ - VolumeConfig( - path="/home", - snapshot_prefix="home", - targets=[TargetConfig(path="/mnt/backup/home")], - ) - ] + mock_load.return_value = ( + Config( + volumes=[ + VolumeConfig( + path="/home", + snapshot_prefix="home", + targets=[TargetConfig(path="/mnt/backup/home")], + ) + ] + ), + [], ) @@ -942,2 +965,1145 @@ assert result == 1 # Need destination + + +# Tests for core restore functions + + +class TestVerifyRestoredSnapshot: + """Tests for verify_restored_snapshot function.""" + + @patch("btrfs_backup_ng.core.restore.__util__.is_subvolume") + def test_success_when_valid_subvolume(self, mock_is_subvol, tmp_path): + """Test verification succeeds for valid subvolume.""" + from btrfs_backup_ng.core.restore import verify_restored_snapshot + + # Create the snapshot path + snapshot_path = tmp_path / "test-snapshot" + snapshot_path.mkdir() + + mock_is_subvol.return_value = True + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": str(tmp_path)} + + result = verify_restored_snapshot(mock_endpoint, "test-snapshot") + + assert result is True + mock_is_subvol.assert_called_once_with(snapshot_path) + + @patch("btrfs_backup_ng.core.restore.__util__.is_subvolume") + def test_raises_when_path_not_exists(self, mock_is_subvol, tmp_path): + """Test verification fails when snapshot path doesn't exist.""" + from btrfs_backup_ng.core.restore import RestoreError, verify_restored_snapshot + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": str(tmp_path)} + + with pytest.raises(RestoreError, match="not found after restore"): + verify_restored_snapshot(mock_endpoint, "nonexistent-snapshot") + + @patch("btrfs_backup_ng.core.restore.__util__.is_subvolume") + def test_raises_when_not_subvolume(self, mock_is_subvol, tmp_path): + """Test verification fails when path is not a subvolume.""" + from btrfs_backup_ng.core.restore import RestoreError, verify_restored_snapshot + + # Create the path but not as subvolume + snapshot_path = tmp_path / "not-subvolume" + snapshot_path.mkdir() + + mock_is_subvol.return_value = False + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": str(tmp_path)} + + with pytest.raises(RestoreError, match="not a valid btrfs subvolume"): + verify_restored_snapshot(mock_endpoint, "not-subvolume") + + @patch("btrfs_backup_ng.core.restore.__util__.is_subvolume") + def test_wraps_unexpected_exceptions(self, mock_is_subvol, tmp_path): + """Test unexpected exceptions are wrapped in RestoreError.""" + from btrfs_backup_ng.core.restore import RestoreError, verify_restored_snapshot + + snapshot_path = tmp_path / "test-snapshot" + snapshot_path.mkdir() + + mock_is_subvol.side_effect = OSError("Unexpected error") + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": str(tmp_path)} + + with pytest.raises(RestoreError, match="Verification failed"): + verify_restored_snapshot(mock_endpoint, "test-snapshot") + + +class TestRestoreSnapshot: + """Tests for restore_snapshot function.""" + + @patch("btrfs_backup_ng.core.restore.verify_restored_snapshot") + @patch("btrfs_backup_ng.core.restore.send_snapshot") + @patch("btrfs_backup_ng.core.restore.log_transaction") + def test_restores_single_snapshot(self, mock_log, mock_send, mock_verify, tmp_path): + """Test restoring a single snapshot.""" + from btrfs_backup_ng.core.restore import restore_snapshot + + mock_verify.return_value = True + + backup_endpoint = MagicMock() + backup_endpoint.config = {"path": "/backup"} + local_endpoint = MagicMock() + local_endpoint.config = {"path": str(tmp_path)} + + snapshot = MockSnapshot("test-snap") + + restore_snapshot(backup_endpoint, local_endpoint, snapshot) + + # Verify lock was set + backup_endpoint.set_lock.assert_any_call(snapshot, ANY, True) + + # Verify send_snapshot was called + mock_send.assert_called_once() + + # Verify lock was released + backup_endpoint.set_lock.assert_any_call(snapshot, ANY, False) + + @patch("btrfs_backup_ng.core.restore.verify_restored_snapshot") + @patch("btrfs_backup_ng.core.restore.send_snapshot") + @patch("btrfs_backup_ng.core.restore.log_transaction") + def test_restores_with_parent(self, mock_log, mock_send, mock_verify): + """Test restoring with incremental parent.""" + from btrfs_backup_ng.core.restore import restore_snapshot + + mock_verify.return_value = True + + backup_endpoint = MagicMock() + backup_endpoint.config = {"path": "/backup"} + local_endpoint = MagicMock() + local_endpoint.config = {"path": "/restore"} + + snapshot = MockSnapshot("snap-2") + parent = MockSnapshot("snap-1") + + restore_snapshot(backup_endpoint, local_endpoint, snapshot, parent=parent) + + # Verify parent lock was set + backup_endpoint.set_lock.assert_any_call(parent, ANY, True, parent=True) + + # Verify send_snapshot was called with parent + call_kwargs = mock_send.call_args[1] + assert call_kwargs["parent"] == parent + + @patch("btrfs_backup_ng.core.restore.verify_restored_snapshot") + @patch("btrfs_backup_ng.core.restore.send_snapshot") + @patch("btrfs_backup_ng.core.restore.log_transaction") + def test_logs_transaction_on_success(self, mock_log, mock_send, mock_verify): + """Test transaction logging on successful restore.""" + from btrfs_backup_ng.core.restore import restore_snapshot + + mock_verify.return_value = True + + backup_endpoint = MagicMock() + backup_endpoint.config = {"path": "/backup"} + local_endpoint = MagicMock() + local_endpoint.config = {"path": "/restore"} + + snapshot = MockSnapshot("test-snap") + + restore_snapshot(backup_endpoint, local_endpoint, snapshot) + + # Should log started and completed + assert mock_log.call_count >= 2 + statuses = [call[1]["status"] for call in mock_log.call_args_list] + assert "started" in statuses + assert "completed" in statuses + + @patch("btrfs_backup_ng.core.restore.verify_restored_snapshot") + @patch("btrfs_backup_ng.core.restore.send_snapshot") + @patch("btrfs_backup_ng.core.restore.log_transaction") + def test_logs_failure_on_error(self, mock_log, mock_send, mock_verify): + """Test transaction logging on failed restore.""" + from btrfs_backup_ng.core.restore import RestoreError, restore_snapshot + + mock_send.side_effect = Exception("Transfer failed") + + backup_endpoint = MagicMock() + backup_endpoint.config = {"path": "/backup"} + local_endpoint = MagicMock() + local_endpoint.config = {"path": "/restore"} + + snapshot = MockSnapshot("test-snap") + + with pytest.raises(RestoreError, match="Restore failed"): + restore_snapshot(backup_endpoint, local_endpoint, snapshot) + + # Should log failure + statuses = [call[1]["status"] for call in mock_log.call_args_list] + assert "failed" in statuses + + @patch("btrfs_backup_ng.core.restore.verify_restored_snapshot") + @patch("btrfs_backup_ng.core.restore.send_snapshot") + @patch("btrfs_backup_ng.core.restore.log_transaction") + def test_releases_locks_on_error(self, mock_log, mock_send, mock_verify): + """Test locks are released even on error.""" + from btrfs_backup_ng.core.restore import RestoreError, restore_snapshot + + mock_send.side_effect = Exception("Transfer failed") + + backup_endpoint = MagicMock() + backup_endpoint.config = {"path": "/backup"} + local_endpoint = MagicMock() + local_endpoint.config = {"path": "/restore"} + + snapshot = MockSnapshot("test-snap") + parent = MockSnapshot("parent-snap") + + with pytest.raises(RestoreError): + restore_snapshot(backup_endpoint, local_endpoint, snapshot, parent=parent) + + # Verify locks were released + backup_endpoint.set_lock.assert_any_call(snapshot, ANY, False) + backup_endpoint.set_lock.assert_any_call(parent, ANY, False, parent=True) + + +class TestRestoreSnapshots: + """Tests for restore_snapshots function.""" + + def test_returns_empty_stats_when_no_backups(self): + """Test returns empty stats when no backups found.""" + from btrfs_backup_ng.core.restore import restore_snapshots + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = [] + + local_endpoint = MagicMock() + + stats = restore_snapshots(backup_endpoint, local_endpoint) + + assert stats["restored"] == 0 + assert stats["skipped"] == 0 + assert stats["failed"] == 0 + + def test_restores_latest_by_default(self): + """Test restores latest snapshot by default.""" + from btrfs_backup_ng.core.restore import restore_snapshots + + snapshots = make_snapshots( + [ + ("snap-1", "20260101-100000"), + ("snap-2", "20260101-110000"), + ] + ) + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = snapshots + + local_endpoint = MagicMock() + local_endpoint.list_snapshots.return_value = [] + + # Dry run to see what would be restored + stats = restore_snapshots(backup_endpoint, local_endpoint, dry_run=True) + + # In dry run, nothing is actually restored + assert stats["restored"] == 0 + + def test_restores_specific_snapshot(self): + """Test restores specific named snapshot.""" + from btrfs_backup_ng.core.restore import restore_snapshots + + snapshots = make_snapshots( + [ + ("snap-1", "20260101-100000"), + ("snap-2", "20260101-110000"), + ("snap-3", "20260101-120000"), + ] + ) + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = snapshots + + local_endpoint = MagicMock() + local_endpoint.list_snapshots.return_value = [] + + stats = restore_snapshots( + backup_endpoint, + local_endpoint, + snapshot_name="snap-2", + dry_run=True, + ) + + assert stats["restored"] == 0 # Dry run + + def test_raises_when_snapshot_not_found(self): + """Test raises error when named snapshot not found.""" + from btrfs_backup_ng.core.restore import RestoreError, restore_snapshots + + snapshots = make_snapshots([("snap-1", "20260101-100000")]) + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = snapshots + + local_endpoint = MagicMock() + + with pytest.raises(RestoreError, match="not found"): + restore_snapshots( + backup_endpoint, + local_endpoint, + snapshot_name="nonexistent", + ) + + def test_no_restore_needed_when_all_exist(self): + """Test no restore needed when all snapshots already exist locally.""" + from btrfs_backup_ng.core.restore import restore_snapshots + + snapshots = make_snapshots( + [ + ("snap-1", "20260101-100000"), + ("snap-2", "20260101-110000"), + ] + ) + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = snapshots + + # Both snapshots already exist locally - chain will be empty + local_endpoint = MagicMock() + local_endpoint.list_snapshots.return_value = snapshots.copy() + + stats = restore_snapshots( + backup_endpoint, + local_endpoint, + restore_all=True, + dry_run=True, + ) + + # No restores needed since all exist (chain is empty) + assert stats["restored"] == 0 + + def test_restore_before_time(self): + """Test restoring snapshot before specific time.""" + from btrfs_backup_ng.core.restore import restore_snapshots + + snapshots = make_snapshots( + [ + ("snap-1", "20260101-100000"), + ("snap-2", "20260101-110000"), + ("snap-3", "20260101-120000"), + ] + ) + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = snapshots + + local_endpoint = MagicMock() + local_endpoint.list_snapshots.return_value = [] + + before_time = time.strptime("20260101-113000", "%Y%m%d-%H%M%S") + + stats = restore_snapshots( + backup_endpoint, + local_endpoint, + before_time=before_time, + dry_run=True, + ) + + assert stats["restored"] == 0 # Dry run + + def test_raises_when_no_snapshot_before_time(self): + """Test raises when no snapshot before requested time.""" + from btrfs_backup_ng.core.restore import RestoreError, restore_snapshots + + snapshots = make_snapshots([("snap-1", "20260101-120000")]) + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = snapshots + + local_endpoint = MagicMock() + + before_time = time.strptime("20260101-100000", "%Y%m%d-%H%M%S") + + with pytest.raises(RestoreError, match="No snapshot found before"): + restore_snapshots( + backup_endpoint, + local_endpoint, + before_time=before_time, + ) + + def test_calls_progress_callback(self): + """Test calls on_progress callback during restore.""" + from btrfs_backup_ng.core.restore import restore_snapshots + + snapshots = make_snapshots( + [ + ("snap-1", "20260101-100000"), + ("snap-2", "20260101-110000"), + ] + ) + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = snapshots + + local_endpoint = MagicMock() + local_endpoint.list_snapshots.return_value = [] + + progress_calls = [] + + def on_progress(current, total, name): + progress_calls.append((current, total, name)) + + # Use dry_run=True so we don't actually try to restore + stats = restore_snapshots( + backup_endpoint, + local_endpoint, + restore_all=True, + dry_run=True, + on_progress=on_progress, + ) + + # In dry run, progress is not called + assert stats["restored"] == 0 + + def test_returns_stats_with_errors(self): + """Test returns stats including error list.""" + from btrfs_backup_ng.core.restore import restore_snapshots + + backup_endpoint = MagicMock() + backup_endpoint.list_snapshots.return_value = [] + + local_endpoint = MagicMock() + + stats = restore_snapshots(backup_endpoint, local_endpoint) + + assert "errors" in stats + assert isinstance(stats["errors"], list) + + +# Tests for CLI entry points + + +class TestExecuteRestore: + """Tests for execute_restore CLI entry point.""" + + def test_no_source_shows_error(self): + """Test execute_restore with no source shows error.""" + from btrfs_backup_ng.cli.restore import execute_restore + + args = argparse.Namespace( + list_volumes=False, + volume=None, + list=False, + status=False, + unlock=None, + cleanup=False, + source=None, + destination=None, + verbose=0, + quiet=False, + ) + + result = execute_restore(args) + + assert result == 1 + + def test_no_destination_shows_error(self): + """Test execute_restore with source but no destination.""" + from btrfs_backup_ng.cli.restore import execute_restore + + args = argparse.Namespace( + list_volumes=False, + volume=None, + list=False, + status=False, + unlock=None, + cleanup=False, + source="/backup", + destination=None, + verbose=0, + quiet=False, + ) + + result = execute_restore(args) + + assert result == 1 + + @patch("btrfs_backup_ng.cli.restore._execute_list_volumes") + def test_list_volumes_mode(self, mock_list_volumes): + """Test --list-volumes mode calls _execute_list_volumes.""" + from btrfs_backup_ng.cli.restore import execute_restore + + mock_list_volumes.return_value = 0 + + args = argparse.Namespace( + list_volumes=True, + volume=None, + verbose=0, + quiet=False, + ) + + result = execute_restore(args) + + assert result == 0 + mock_list_volumes.assert_called_once() + + @patch("btrfs_backup_ng.cli.restore._execute_config_restore") + def test_volume_mode(self, mock_config_restore): + """Test --volume mode calls _execute_config_restore.""" + from btrfs_backup_ng.cli.restore import execute_restore + + mock_config_restore.return_value = 0 + + args = argparse.Namespace( + list_volumes=False, + volume="/home", + verbose=0, + quiet=False, + ) + + result = execute_restore(args) + + assert result == 0 + mock_config_restore.assert_called_once_with(args, "/home") + + @patch("btrfs_backup_ng.cli.restore._execute_list") + def test_list_mode(self, mock_list): + """Test --list mode calls _execute_list.""" + from btrfs_backup_ng.cli.restore import execute_restore + + mock_list.return_value = 0 + + args = argparse.Namespace( + list_volumes=False, + volume=None, + list=True, + status=False, + unlock=None, + cleanup=False, + source="/backup", + destination=None, + verbose=0, + quiet=False, + ) + + result = execute_restore(args) + + assert result == 0 + mock_list.assert_called_once() + + @patch("btrfs_backup_ng.cli.restore._execute_status") + def test_status_mode(self, mock_status): + """Test --status mode calls _execute_status.""" + from btrfs_backup_ng.cli.restore import execute_restore + + mock_status.return_value = 0 + + args = argparse.Namespace( + list_volumes=False, + volume=None, + list=False, + status=True, + unlock=None, + cleanup=False, + source="/backup", + destination=None, + verbose=0, + quiet=False, + ) + + result = execute_restore(args) + + assert result == 0 + mock_status.assert_called_once() + + @patch("btrfs_backup_ng.cli.restore._execute_unlock") + def test_unlock_mode(self, mock_unlock): + """Test --unlock mode calls _execute_unlock.""" + from btrfs_backup_ng.cli.restore import execute_restore + + mock_unlock.return_value = 0 + + args = argparse.Namespace( + list_volumes=False, + volume=None, + list=False, + status=False, + unlock="session-123", + cleanup=False, + source="/backup", + destination=None, + verbose=0, + quiet=False, + ) + + result = execute_restore(args) + + assert result == 0 + mock_unlock.assert_called_once_with(args, "session-123") + + @patch("btrfs_backup_ng.cli.restore._execute_cleanup") + def test_cleanup_mode(self, mock_cleanup): + """Test --cleanup mode calls _execute_cleanup.""" + from btrfs_backup_ng.cli.restore import execute_restore + + mock_cleanup.return_value = 0 + + args = argparse.Namespace( + list_volumes=False, + volume=None, + list=False, + status=False, + unlock=None, + cleanup=True, + source=None, + destination="/restore", + verbose=0, + quiet=False, + ) + + result = execute_restore(args) + + assert result == 0 + mock_cleanup.assert_called_once() + + +class TestExecuteMainRestore: + """Tests for _execute_main_restore function.""" + + @patch("btrfs_backup_ng.cli.restore.validate_restore_destination") + def test_destination_validation_failure(self, mock_validate, tmp_path): + """Test handling of destination validation failure.""" + from btrfs_backup_ng.cli.restore import _execute_main_restore + + mock_validate.side_effect = RestoreError("Not a btrfs filesystem") + + args = argparse.Namespace( + source="/backup", + destination=str(tmp_path), + in_place=False, + yes_i_know_what_i_am_doing=False, + verbose=0, + quiet=False, + ) + + result = _execute_main_restore(args) + + assert result == 1 + + @patch("btrfs_backup_ng.cli.restore._prepare_local_endpoint") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + @patch("btrfs_backup_ng.cli.restore.validate_restore_destination") + def test_backup_endpoint_failure( + self, mock_validate, mock_prep_backup, mock_prep_local, tmp_path + ): + """Test handling of backup endpoint preparation failure.""" + from btrfs_backup_ng.cli.restore import _execute_main_restore + + mock_prep_backup.side_effect = Exception("SSH connection failed") + + args = argparse.Namespace( + source="ssh://server/backup", + destination=str(tmp_path), + in_place=False, + yes_i_know_what_i_am_doing=False, + verbose=0, + quiet=False, + ) + + result = _execute_main_restore(args) + + assert result == 1 + + @patch("btrfs_backup_ng.cli.restore._prepare_local_endpoint") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + @patch("btrfs_backup_ng.cli.restore.validate_restore_destination") + def test_local_endpoint_failure( + self, mock_validate, mock_prep_backup, mock_prep_local, tmp_path + ): + """Test handling of local endpoint preparation failure.""" + from btrfs_backup_ng.cli.restore import _execute_main_restore + + mock_prep_backup.return_value = MagicMock() + mock_prep_local.side_effect = Exception("Cannot create local endpoint") + + args = argparse.Namespace( + source="/backup", + destination=str(tmp_path), + in_place=False, + yes_i_know_what_i_am_doing=False, + verbose=0, + quiet=False, + ) + + result = _execute_main_restore(args) + + assert result == 1 + + @patch("btrfs_backup_ng.cli.restore.restore_snapshots") + @patch("btrfs_backup_ng.cli.restore._prepare_local_endpoint") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + @patch("btrfs_backup_ng.cli.restore.validate_restore_destination") + def test_restore_error( + self, mock_validate, mock_prep_backup, mock_prep_local, mock_restore, tmp_path + ): + """Test handling of RestoreError during restore.""" + from btrfs_backup_ng.cli.restore import _execute_main_restore + + mock_prep_backup.return_value = MagicMock() + mock_prep_local.return_value = MagicMock() + mock_restore.side_effect = RestoreError("Snapshot not found") + + args = argparse.Namespace( + source="/backup", + destination=str(tmp_path), + in_place=False, + yes_i_know_what_i_am_doing=False, + before=None, + dry_run=False, + snapshot=None, + all=False, + overwrite=False, + no_incremental=False, + interactive=False, + compress=None, + rate_limit=None, + verbose=0, + quiet=False, + ) + + result = _execute_main_restore(args) + + assert result == 1 + + @patch("btrfs_backup_ng.cli.restore.restore_snapshots") + @patch("btrfs_backup_ng.cli.restore._prepare_local_endpoint") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + @patch("btrfs_backup_ng.cli.restore.validate_restore_destination") + def test_successful_restore( + self, mock_validate, mock_prep_backup, mock_prep_local, mock_restore, tmp_path + ): + """Test successful restore returns 0.""" + from btrfs_backup_ng.cli.restore import _execute_main_restore + + mock_prep_backup.return_value = MagicMock() + mock_prep_local.return_value = MagicMock() + mock_restore.return_value = {"restored": 2, "skipped": 0, "failed": 0} + + args = argparse.Namespace( + source="/backup", + destination=str(tmp_path), + in_place=False, + yes_i_know_what_i_am_doing=False, + before=None, + dry_run=False, + snapshot=None, + all=False, + overwrite=False, + no_incremental=False, + interactive=False, + compress=None, + rate_limit=None, + verbose=0, + quiet=False, + ) + + result = _execute_main_restore(args) + + assert result == 0 + + @patch("btrfs_backup_ng.cli.restore.restore_snapshots") + @patch("btrfs_backup_ng.cli.restore._prepare_local_endpoint") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + @patch("btrfs_backup_ng.cli.restore.validate_restore_destination") + def test_restore_with_failures_returns_1( + self, mock_validate, mock_prep_backup, mock_prep_local, mock_restore, tmp_path + ): + """Test restore with failures returns 1.""" + from btrfs_backup_ng.cli.restore import _execute_main_restore + + mock_prep_backup.return_value = MagicMock() + mock_prep_local.return_value = MagicMock() + mock_restore.return_value = {"restored": 1, "skipped": 0, "failed": 1} + + args = argparse.Namespace( + source="/backup", + destination=str(tmp_path), + in_place=False, + yes_i_know_what_i_am_doing=False, + before=None, + dry_run=False, + snapshot=None, + all=False, + overwrite=False, + no_incremental=False, + interactive=False, + compress=None, + rate_limit=None, + verbose=0, + quiet=False, + ) + + result = _execute_main_restore(args) + + assert result == 1 + + @patch("btrfs_backup_ng.cli.restore.restore_snapshots") + @patch("btrfs_backup_ng.cli.restore._prepare_local_endpoint") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + @patch("btrfs_backup_ng.cli.restore.validate_restore_destination") + def test_invalid_before_date_format( + self, mock_validate, mock_prep_backup, mock_prep_local, mock_restore, tmp_path + ): + """Test invalid --before date format.""" + from btrfs_backup_ng.cli.restore import _execute_main_restore + + mock_prep_backup.return_value = MagicMock() + mock_prep_local.return_value = MagicMock() + + args = argparse.Namespace( + source="/backup", + destination=str(tmp_path), + in_place=False, + yes_i_know_what_i_am_doing=False, + before="not-a-date", + dry_run=False, + snapshot=None, + all=False, + overwrite=False, + no_incremental=False, + interactive=False, + compress=None, + rate_limit=None, + verbose=0, + quiet=False, + ) + + result = _execute_main_restore(args) + + assert result == 1 + + +class TestExecuteList: + """Tests for _execute_list function.""" + + def test_list_no_source(self): + """Test --list without source shows error.""" + from btrfs_backup_ng.cli.restore import _execute_list + + args = MagicMock() + args.source = None + + result = _execute_list(args) + + assert result == 1 + + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + def test_list_endpoint_failure(self, mock_prepare): + """Test --list with endpoint failure.""" + from btrfs_backup_ng.cli.restore import _execute_list + + mock_prepare.side_effect = Exception("Connection failed") + + args = MagicMock() + args.source = "/backup" + + result = _execute_list(args) + + assert result == 1 + + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + def test_list_empty(self, mock_prepare, mock_list): + """Test --list with no snapshots.""" + from btrfs_backup_ng.cli.restore import _execute_list + + mock_prepare.return_value = MagicMock() + mock_list.return_value = [] + + args = MagicMock() + args.source = "/backup" + + result = _execute_list(args) + + assert result == 0 + + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + def test_list_with_snapshots(self, mock_prepare, mock_list, capsys): + """Test --list shows snapshots.""" + from btrfs_backup_ng.cli.restore import _execute_list + + mock_prepare.return_value = MagicMock() + mock_list.return_value = [ + MockSnapshot("snap-1"), + MockSnapshot("snap-2"), + ] + + args = MagicMock() + args.source = "/backup" + + result = _execute_list(args) + + assert result == 0 + captured = capsys.readouterr() + assert "snap-1" in captured.out + assert "snap-2" in captured.out + assert "2 snapshot(s)" in captured.out + + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + @patch("btrfs_backup_ng.cli.restore._prepare_backup_endpoint") + def test_list_exception(self, mock_prepare, mock_list): + """Test --list handles exceptions.""" + from btrfs_backup_ng.cli.restore import _execute_list + + mock_prepare.return_value = MagicMock() + mock_list.side_effect = Exception("Failed to list") + + args = MagicMock() + args.source = "/backup" + + result = _execute_list(args) + + assert result == 1 + + +class TestPrepareBackupEndpoint: + """Tests for _prepare_backup_endpoint function.""" + + @patch("btrfs_backup_ng.cli.restore.endpoint.choose_endpoint") + def test_local_endpoint(self, mock_choose, tmp_path): + """Test preparing local backup endpoint.""" + from btrfs_backup_ng.cli.restore import _prepare_backup_endpoint + + mock_ep = MagicMock() + mock_choose.return_value = mock_ep + + args = MagicMock() + args.no_fs_checks = False + args.prefix = "home-" + + _prepare_backup_endpoint(args, str(tmp_path)) + + mock_choose.assert_called_once() + mock_ep.prepare.assert_called_once() + # Check endpoint kwargs + call_kwargs = mock_choose.call_args[0][1] + assert call_kwargs["snap_prefix"] == "home-" + assert call_kwargs["fs_checks"] is True + + @patch("btrfs_backup_ng.cli.restore.endpoint.choose_endpoint") + def test_ssh_endpoint(self, mock_choose): + """Test preparing SSH backup endpoint.""" + from btrfs_backup_ng.cli.restore import _prepare_backup_endpoint + + mock_ep = MagicMock() + mock_choose.return_value = mock_ep + + args = MagicMock() + args.no_fs_checks = False + args.prefix = "" + args.ssh_sudo = True + args.ssh_password_auth = False + args.ssh_key = "/path/to/key" + + _prepare_backup_endpoint(args, "ssh://user@server/backup") + + call_kwargs = mock_choose.call_args[0][1] + assert call_kwargs["ssh_sudo"] is True + assert call_kwargs["ssh_identity_file"] == "/path/to/key" + + +class TestPrepareLocalEndpoint: + """Tests for _prepare_local_endpoint function.""" + + @patch("btrfs_backup_ng.endpoint.local.LocalEndpoint") + def test_creates_directory(self, mock_local_ep, tmp_path): + """Test local endpoint creates destination directory.""" + from btrfs_backup_ng.cli.restore import _prepare_local_endpoint + + dest = tmp_path / "new_restore_dir" + assert not dest.exists() + + _prepare_local_endpoint(dest) + + assert dest.exists() + + @patch("btrfs_backup_ng.endpoint.local.LocalEndpoint") + def test_calls_prepare(self, mock_local_ep, tmp_path): + """Test local endpoint prepare is called.""" + from btrfs_backup_ng.cli.restore import _prepare_local_endpoint + + mock_ep_instance = MagicMock() + mock_local_ep.return_value = mock_ep_instance + + _prepare_local_endpoint(tmp_path) + + mock_ep_instance.prepare.assert_called_once() + + +class TestParseDatetime: + """Tests for _parse_datetime function.""" + + def test_parse_date_only(self): + """Test parsing date without time.""" + from btrfs_backup_ng.cli.restore import _parse_datetime + + result = _parse_datetime("2026-01-15") + + assert result.tm_year == 2026 + assert result.tm_mon == 1 + assert result.tm_mday == 15 + + def test_parse_date_with_time(self): + """Test parsing date with time.""" + from btrfs_backup_ng.cli.restore import _parse_datetime + + result = _parse_datetime("2026-01-15 14:30:00") + + assert result.tm_year == 2026 + assert result.tm_hour == 14 + assert result.tm_min == 30 + assert result.tm_sec == 0 + + def test_parse_date_with_time_no_seconds(self): + """Test parsing date with time but no seconds.""" + from btrfs_backup_ng.cli.restore import _parse_datetime + + result = _parse_datetime("2026-01-15 14:30") + + assert result.tm_hour == 14 + assert result.tm_min == 30 + + def test_parse_iso_format(self): + """Test parsing ISO format with T separator.""" + from btrfs_backup_ng.cli.restore import _parse_datetime + + result = _parse_datetime("2026-01-15T14:30:00") + + assert result.tm_year == 2026 + assert result.tm_hour == 14 + + def test_parse_invalid_format(self): + """Test parsing invalid format raises ValueError.""" + from btrfs_backup_ng.cli.restore import _parse_datetime + + with pytest.raises(ValueError, match="Could not parse date"): + _parse_datetime("invalid-date") + + +class TestInteractiveSelect: + """Tests for _interactive_select function.""" + + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_no_snapshots_returns_none(self, mock_list): + """Test returns None when no snapshots available.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [] + + result = _interactive_select(MagicMock()) + + assert result is None + + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_exception_returns_none(self, mock_list): + """Test returns None on exception.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.side_effect = Exception("Failed") + + result = _interactive_select(MagicMock()) + + assert result is None + + @patch("builtins.input", side_effect=["0"]) + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_cancel_returns_none(self, mock_list, mock_input): + """Test selecting 0 cancels and returns None.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [MockSnapshot("snap-1")] + + result = _interactive_select(MagicMock()) + + assert result is None + + @patch("builtins.input", side_effect=[""]) + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_empty_input_returns_none(self, mock_list, mock_input): + """Test empty input cancels and returns None.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [MockSnapshot("snap-1")] + + result = _interactive_select(MagicMock()) + + assert result is None + + @patch("builtins.input", side_effect=["1", "y"]) + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_select_and_confirm(self, mock_list, mock_input): + """Test selecting a snapshot and confirming.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [MockSnapshot("snap-1"), MockSnapshot("snap-2")] + + result = _interactive_select(MagicMock()) + + assert result == "snap-1" + + @patch("builtins.input", side_effect=["1", "n"]) + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_select_and_decline(self, mock_list, mock_input): + """Test selecting a snapshot but declining confirmation.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [MockSnapshot("snap-1")] + + result = _interactive_select(MagicMock()) + + assert result is None + + @patch("builtins.input", side_effect=["invalid", "1", "y"]) + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_invalid_input_then_valid(self, mock_list, mock_input): + """Test invalid input followed by valid selection.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [MockSnapshot("snap-1")] + + result = _interactive_select(MagicMock()) + + assert result == "snap-1" + + @patch("builtins.input", side_effect=["5", "1", "y"]) + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_out_of_range_then_valid(self, mock_list, mock_input): + """Test out of range selection followed by valid.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [MockSnapshot("snap-1")] + + result = _interactive_select(MagicMock()) + + assert result == "snap-1" + + @patch("builtins.input", side_effect=KeyboardInterrupt()) + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_keyboard_interrupt(self, mock_list, mock_input): + """Test keyboard interrupt returns None.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [MockSnapshot("snap-1")] + + result = _interactive_select(MagicMock()) + + assert result is None + + @patch("builtins.input", side_effect=EOFError()) + @patch("btrfs_backup_ng.cli.restore.list_remote_snapshots") + def test_eof_error(self, mock_list, mock_input): + """Test EOF error returns None.""" + from btrfs_backup_ng.cli.restore import _interactive_select + + mock_list.return_value = [MockSnapshot("snap-1")] + + result = _interactive_select(MagicMock()) + + assert result is None Only in /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests: test_transaction.py diff -U2 -r /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/tests/test_verify.py /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests/test_verify.py --- /var/lib/copr-rpmbuild/results/btrfs-backup-ng/upstream-unpacked/Source0/btrfs_backup_ng-0.8.0/tests/test_verify.py 2026-01-04 23:42:39.000000000 +0000 +++ /var/lib/copr-rpmbuild/results/btrfs-backup-ng/srpm-unpacked/btrfs_backup_ng-0.8.0.tar.gz-extract/btrfs_backup_ng-0.8.0/tests/test_verify.py 2026-01-05 07:28:57.000000000 +0000 @@ -2,13 +2,19 @@ import time -from unittest.mock import MagicMock +from pathlib import Path +from unittest.mock import MagicMock, patch +import pytest from btrfs_backup_ng.core.verify import ( + VerifyError, VerifyLevel, VerifyReport, VerifyResult, _find_parent_snapshot, + _test_send_stream, + verify_full, verify_metadata, + verify_stream, ) @@ -302,2 +308,599 @@ assert VerifyLevel("stream") == VerifyLevel.STREAM assert VerifyLevel("full") == VerifyLevel.FULL + + +class TestVerifyStream: + """Tests for verify_stream function.""" + + def test_empty_backup_location(self): + """Test handling of empty backup location.""" + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [] + mock_endpoint.config = {"path": "/backup"} + + report = verify_stream(mock_endpoint) + + assert report.total == 0 + assert report.level == VerifyLevel.STREAM + assert "No snapshots found" in report.errors[0] + + def test_snapshot_not_found(self): + """Test handling of non-existent snapshot.""" + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + report = verify_stream(mock_endpoint, snapshot_name="nonexistent") + + assert report.total == 0 + assert "not found" in report.errors[0] + + @patch("btrfs_backup_ng.core.verify._test_send_stream") + def test_verify_latest_by_default(self, mock_test_stream): + """Test that only latest snapshot is verified by default.""" + snapshots = make_snapshots( + [ + ("snap-1", "20260101-100000"), + ("snap-2", "20260101-110000"), + ("snap-3", "20260101-120000"), + ] + ) + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = snapshots + mock_endpoint.config = {"path": "/backup"} + + report = verify_stream(mock_endpoint) + + assert report.total == 1 + assert report.results[0].snapshot_name == "snap-3" + + @patch("btrfs_backup_ng.core.verify._test_send_stream") + def test_verify_specific_snapshot(self, mock_test_stream): + """Test verification of specific snapshot.""" + snapshots = make_snapshots( + [ + ("snap-1", "20260101-100000"), + ("snap-2", "20260101-110000"), + ] + ) + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = snapshots + mock_endpoint.config = {"path": "/backup"} + + report = verify_stream(mock_endpoint, snapshot_name="snap-1") + + assert report.total == 1 + assert report.results[0].snapshot_name == "snap-1" + + @patch("btrfs_backup_ng.core.verify._test_send_stream") + def test_stream_success(self, mock_test_stream): + """Test successful stream verification.""" + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + report = verify_stream(mock_endpoint) + + assert report.total == 1 + assert report.passed == 1 + assert report.results[0].passed is True + assert "verified successfully" in report.results[0].message + + @patch("btrfs_backup_ng.core.verify._test_send_stream") + def test_stream_failure(self, mock_test_stream): + """Test failed stream verification.""" + mock_test_stream.side_effect = Exception("Stream error") + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + report = verify_stream(mock_endpoint) + + assert report.total == 1 + assert report.failed == 1 + assert report.results[0].passed is False + assert "failed" in report.results[0].message + + @patch("btrfs_backup_ng.core.verify._test_send_stream") + def test_progress_callback(self, mock_test_stream): + """Test that progress callback is called.""" + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + progress_calls = [] + + def on_progress(current, total, name): + progress_calls.append((current, total, name)) + + verify_stream(mock_endpoint, on_progress=on_progress) + + assert len(progress_calls) == 1 + assert progress_calls[0] == (1, 1, "snap-1") + + @patch("btrfs_backup_ng.core.verify._test_send_stream") + def test_incremental_detection(self, mock_test_stream): + """Test that incremental parent is detected.""" + snapshots = make_snapshots( + [ + ("snap-1", "20260101-100000"), + ("snap-2", "20260101-110000"), + ] + ) + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = snapshots + mock_endpoint.config = {"path": "/backup"} + + report = verify_stream(mock_endpoint, snapshot_name="snap-2") + + assert report.results[0].details.get("incremental") is True + assert report.results[0].details.get("parent") == "snap-1" + + def test_exception_handling(self): + """Test that exceptions are caught and reported.""" + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.side_effect = Exception("Connection failed") + mock_endpoint.config = {"path": "/backup"} + + report = verify_stream(mock_endpoint) + + assert len(report.errors) > 0 + assert "Connection failed" in report.errors[0] + + +class TestVerifyFull: + """Tests for verify_full function.""" + + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_empty_backup_location(self, mock_is_btrfs): + """Test handling of empty backup location.""" + import tempfile + + mock_is_btrfs.return_value = True + + with tempfile.TemporaryDirectory() as tmpdir: + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [] + mock_endpoint.config = {"path": tmpdir} + + report = verify_full(mock_endpoint, temp_dir=Path(tmpdir)) + + assert report.level == VerifyLevel.FULL + assert "No snapshots found" in report.errors[0] + + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_snapshot_not_found(self, mock_is_btrfs): + """Test handling of non-existent snapshot.""" + import tempfile + + mock_is_btrfs.return_value = True + + with tempfile.TemporaryDirectory() as tmpdir: + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": tmpdir} + + report = verify_full( + mock_endpoint, snapshot_name="nonexistent", temp_dir=Path(tmpdir) + ) + + assert "not found" in report.errors[0] + + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_temp_dir_not_btrfs(self, mock_is_btrfs): + """Test error when temp dir is not on btrfs.""" + import tempfile + + mock_is_btrfs.return_value = False + + with tempfile.TemporaryDirectory() as tmpdir: + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": tmpdir} + + report = verify_full(mock_endpoint, temp_dir=Path(tmpdir)) + + assert "not on btrfs" in report.errors[0] + + def test_remote_without_temp_dir(self): + """Test error when remote backup without temp_dir.""" + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "ssh://server:/backups"} + + report = verify_full(mock_endpoint) + + assert "--temp-dir must be specified" in report.errors[0] + + @patch("btrfs_backup_ng.endpoint.LocalEndpoint") + @patch("btrfs_backup_ng.core.verify._test_restore") + @patch("btrfs_backup_ng.core.verify.__util__.is_subvolume") + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_full_verify_success( + self, + mock_is_btrfs, + mock_is_subvolume, + mock_test_restore, + mock_local_endpoint, + ): + """Test successful full verification.""" + import tempfile + + mock_is_btrfs.return_value = True + mock_is_subvolume.return_value = True + + with tempfile.TemporaryDirectory() as tmpdir: + # Create mock snapshot path + snap_path = Path(tmpdir) / "snap-1" + snap_path.mkdir() + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + report = verify_full(mock_endpoint, temp_dir=Path(tmpdir), cleanup=False) + + assert report.total == 1 + assert report.passed == 1 + assert report.results[0].passed is True + + @patch("btrfs_backup_ng.endpoint.LocalEndpoint") + @patch("btrfs_backup_ng.core.verify._test_restore") + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_full_verify_restore_fails( + self, + mock_is_btrfs, + mock_test_restore, + mock_local_endpoint, + ): + """Test full verification when restore fails.""" + import tempfile + + mock_is_btrfs.return_value = True + mock_test_restore.side_effect = Exception("Restore failed") + + with tempfile.TemporaryDirectory() as tmpdir: + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + report = verify_full(mock_endpoint, temp_dir=Path(tmpdir), cleanup=False) + + assert report.total == 1 + assert report.failed == 1 + assert "failed" in report.results[0].message.lower() + + @patch("btrfs_backup_ng.endpoint.LocalEndpoint") + @patch("btrfs_backup_ng.core.verify._test_restore") + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_restored_path_not_found( + self, + mock_is_btrfs, + mock_test_restore, + mock_local_endpoint, + ): + """Test error when restored snapshot path doesn't exist.""" + import tempfile + + mock_is_btrfs.return_value = True + # Don't create the snapshot path - it won't exist + + with tempfile.TemporaryDirectory() as tmpdir: + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + report = verify_full(mock_endpoint, temp_dir=Path(tmpdir), cleanup=False) + + assert report.failed == 1 + assert "not found" in report.results[0].message.lower() + + @patch("btrfs_backup_ng.endpoint.LocalEndpoint") + @patch("btrfs_backup_ng.core.verify._test_restore") + @patch("btrfs_backup_ng.core.verify.__util__.is_subvolume") + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_restored_not_subvolume( + self, + mock_is_btrfs, + mock_is_subvolume, + mock_test_restore, + mock_local_endpoint, + ): + """Test error when restored path is not a valid subvolume.""" + import tempfile + + mock_is_btrfs.return_value = True + mock_is_subvolume.return_value = False + + with tempfile.TemporaryDirectory() as tmpdir: + # Create the path but it won't be a subvolume + snap_path = Path(tmpdir) / "snap-1" + snap_path.mkdir() + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + report = verify_full(mock_endpoint, temp_dir=Path(tmpdir), cleanup=False) + + assert report.failed == 1 + assert "not a valid subvolume" in report.results[0].message.lower() + + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_with_custom_temp_dir(self, mock_is_btrfs): + """Test using custom temp directory.""" + import tempfile + + mock_is_btrfs.return_value = True + + with tempfile.TemporaryDirectory() as tmpdir: + temp_path = Path(tmpdir) + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [] + mock_endpoint.config = {"path": "/backup"} + + report = verify_full(mock_endpoint, temp_dir=temp_path) + + # Should have gotten past temp_dir setup to "no snapshots" error + assert "No snapshots found" in report.errors[0] + + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_creates_temp_dir_if_not_exists(self, mock_is_btrfs): + """Test that temp_dir is created if it doesn't exist.""" + import tempfile + + mock_is_btrfs.return_value = True + + with tempfile.TemporaryDirectory() as tmpdir: + new_temp = Path(tmpdir) / "new_subdir" + assert not new_temp.exists() + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [] + mock_endpoint.config = {"path": "/backup"} + + verify_full(mock_endpoint, temp_dir=new_temp) + + assert new_temp.exists() + + @patch("btrfs_backup_ng.endpoint.LocalEndpoint") + @patch("btrfs_backup_ng.core.verify._test_restore") + @patch("btrfs_backup_ng.core.verify.__util__.is_subvolume") + @patch("btrfs_backup_ng.core.verify.__util__.is_btrfs") + def test_progress_callback( + self, + mock_is_btrfs, + mock_is_subvolume, + mock_test_restore, + mock_local_endpoint, + ): + """Test that progress callback is called.""" + import tempfile + + mock_is_btrfs.return_value = True + mock_is_subvolume.return_value = True + + with tempfile.TemporaryDirectory() as tmpdir: + snap_path = Path(tmpdir) / "snap-1" + snap_path.mkdir() + + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.return_value = [MockSnapshot("snap-1")] + mock_endpoint.config = {"path": "/backup"} + + progress_calls = [] + + def on_progress(current, total, name): + progress_calls.append((current, total, name)) + + verify_full( + mock_endpoint, + temp_dir=Path(tmpdir), + cleanup=False, + on_progress=on_progress, + ) + + assert len(progress_calls) == 1 + assert progress_calls[0] == (1, 1, "snap-1") + + +class TestTestSendStream: + """Tests for _test_send_stream function.""" + + @patch("subprocess.run") + def test_local_send_success(self, mock_run): + """Test successful local send stream test.""" + mock_run.return_value = MagicMock(returncode=0) + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": "/backup"} + mock_endpoint.ssh_client = None + + snapshot = MockSnapshot("snap-1") + + # Should not raise + _test_send_stream(mock_endpoint, snapshot) + + mock_run.assert_called_once() + call_args = mock_run.call_args + assert "btrfs" in call_args[0][0] + assert "--no-data" in call_args[0][0] + + @patch("subprocess.run") + def test_local_send_with_parent(self, mock_run): + """Test local send with parent snapshot.""" + mock_run.return_value = MagicMock(returncode=0) + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": "/backup"} + mock_endpoint.ssh_client = None + + snapshot = MockSnapshot("snap-2") + parent = MockSnapshot("snap-1") + + _test_send_stream(mock_endpoint, snapshot, parent) + + call_args = mock_run.call_args + assert "-p" in call_args[0][0] + + @patch("subprocess.run") + def test_local_send_failure(self, mock_run): + """Test failed local send stream test.""" + mock_run.return_value = MagicMock( + returncode=1, stderr=b"Send failed: corrupted data" + ) + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": "/backup"} + mock_endpoint.ssh_client = None + + snapshot = MockSnapshot("snap-1") + + with pytest.raises(VerifyError) as exc_info: + _test_send_stream(mock_endpoint, snapshot) + + assert "failed" in str(exc_info.value).lower() + + def test_ssh_send_success(self): + """Test successful SSH send stream test.""" + mock_stdout = MagicMock() + mock_stdout.channel.recv_exit_status.return_value = 0 + + mock_stderr = MagicMock() + mock_stderr.read.return_value = b"" + + mock_ssh_client = MagicMock() + mock_ssh_client.exec_command.return_value = ( + MagicMock(), + mock_stdout, + mock_stderr, + ) + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": "/backup", "ssh_sudo": False} + mock_endpoint.ssh_client = mock_ssh_client + + snapshot = MockSnapshot("snap-1") + + # Should not raise + _test_send_stream(mock_endpoint, snapshot) + + mock_ssh_client.exec_command.assert_called_once() + + def test_ssh_send_with_sudo(self): + """Test SSH send with sudo.""" + mock_stdout = MagicMock() + mock_stdout.channel.recv_exit_status.return_value = 0 + + mock_stderr = MagicMock() + + mock_ssh_client = MagicMock() + mock_ssh_client.exec_command.return_value = ( + MagicMock(), + mock_stdout, + mock_stderr, + ) + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": "/backup", "ssh_sudo": True} + mock_endpoint.ssh_client = mock_ssh_client + + snapshot = MockSnapshot("snap-1") + + _test_send_stream(mock_endpoint, snapshot) + + call_args = mock_ssh_client.exec_command.call_args + assert "sudo" in call_args[0][0] + + def test_ssh_send_failure(self): + """Test failed SSH send stream test.""" + mock_stdout = MagicMock() + mock_stdout.channel.recv_exit_status.return_value = 1 + + mock_stderr = MagicMock() + mock_stderr.read.return_value = b"Permission denied" + + mock_ssh_client = MagicMock() + mock_ssh_client.exec_command.return_value = ( + MagicMock(), + mock_stdout, + mock_stderr, + ) + + mock_endpoint = MagicMock() + mock_endpoint.config = {"path": "/backup", "ssh_sudo": False} + mock_endpoint.ssh_client = mock_ssh_client + + snapshot = MockSnapshot("snap-1") + + with pytest.raises(VerifyError) as exc_info: + _test_send_stream(mock_endpoint, snapshot) + + assert "failed" in str(exc_info.value).lower() + + +class TestVerifyError: + """Tests for VerifyError exception.""" + + def test_verify_error_message(self): + """Test VerifyError exception.""" + error = VerifyError("Test error message") + assert str(error) == "Test error message" + + def test_verify_error_is_exception(self): + """Test VerifyError is an Exception.""" + assert issubclass(VerifyError, Exception) + + +class TestVerifyMetadataExceptionHandling: + """Tests for exception handling in verify_metadata.""" + + def test_list_snapshots_exception(self): + """Test handling of exception during list_snapshots.""" + mock_endpoint = MagicMock() + mock_endpoint.list_snapshots.side_effect = Exception("Connection failed") + mock_endpoint.config = {"path": "/backup"} + + report = verify_metadata(mock_endpoint) + + assert len(report.errors) > 0 + assert "Connection failed" in report.errors[0] + + def test_source_comparison_exception(self): + """Test handling of exception during source comparison.""" + backup_snapshots = [MockSnapshot("snap-1")] + + backup_ep = MagicMock() + backup_ep.list_snapshots.return_value = backup_snapshots + backup_ep.config = {"path": "/backup"} + + source_ep = MagicMock() + source_ep.list_snapshots.side_effect = Exception("Source unreachable") + + # Should not raise, just log warning + report = verify_metadata(backup_ep, source_endpoint=source_ep) + + # The report should still be valid + assert report.total == 1 + + +class TestVerifyReportDuration: + """Additional tests for VerifyReport duration.""" + + def test_duration_before_completion(self): + """Test duration calculation before completed_at is set.""" + report = VerifyReport( + level=VerifyLevel.METADATA, + location="/test", + ) + report.started_at = time.time() - 2.0 + # completed_at defaults to 0.0 + + # Duration should be calculated from now + duration = report.duration + assert duration >= 2.0