bak-and-rest/tests/test_backup.py

from __future__ import annotations

from pathlib import Path
import subprocess

from bakrest.backup import (
    _build_robocopy_command,
    _build_rsync_command,
    _robocopy_destination_dir,
    _run_command,
    _run_robocopy,
    _split_existing_files,
)
from bakrest.config import parse_config
from bakrest.registry import ChangeEvent, ChangeRegistry


def test_split_existing_files_ignores_deleted(tmp_path: Path) -> None:
    existing = tmp_path / "image.jpg"
    existing.write_text("ok", encoding="utf-8")
    events = [
        ChangeEvent(path=str(existing), event_type="modified"),
        ChangeEvent(path=str(tmp_path / "missing.jpg"), event_type="deleted"),
    ]

    present, skipped = _split_existing_files(events)

    assert [event.path for event in present] == [str(existing)]
    assert [event.path for event in skipped] == [str(tmp_path / "missing.jpg")]


def test_build_rsync_command_uses_files_from(tmp_path: Path) -> None:
    source = tmp_path / "source"
    source.mkdir()
    image = source / "image.jpg"
    image.write_text("ok", encoding="utf-8")
    config = parse_config(
        {
            "watch": {
                "include_dirs": [str(source)],
                "include_extensions": [".jpg"],
                "exclude_dirs": [],
                "exclude_patterns": [],
            },
            "backup": {
                "rsync_path": "rsync",
                "remote_destination": "user@host:/backup/",
                "server_check": {"type": "tcp", "port": 22, "interval_seconds": 300, "timeout_seconds": 1},
            },
        },
        tmp_path / "config.toml",
    )

    command = _build_rsync_command(config, [ChangeEvent(path=str(image), event_type="modified")])

    assert command[0] == "rsync"
    assert "-av" in command
    assert "--relative" in command
    assert any(part.startswith("--files-from=") for part in command)
    assert command[-1] == "user@host:/backup/"


def test_build_robocopy_command_preserves_path_under_include_root(tmp_path: Path) -> None:
    source = tmp_path / "Lavori"
    nested = source / "Cliente" / "Grafica"
    nested.mkdir(parents=True)
    image = nested / "image.psd"
    image.write_text("ok", encoding="utf-8")
    config = parse_config(
        {
            "watch": {
                "include_dirs": [str(source)],
                "include_extensions": [".psd"],
                "exclude_dirs": [],
                "exclude_patterns": [],
            },
            "backup": {
                "engine": "robocopy",
                "robocopy_path": "robocopy",
                "remote_destinations": ["\\\\server\\Backup1", "\\\\server\\Backup2"],
                "server_check": {"type": "tcp", "port": 445, "interval_seconds": 300, "timeout_seconds": 1},
            },
        },
        tmp_path / "config.toml",
    )

    destination = _robocopy_destination_dir(config, image, Path("\\\\server\\Backup1"))
    command = _build_robocopy_command(
        config,
        ChangeEvent(path=str(image), event_type="modified"),
        "\\\\server\\Backup1",
    )

    assert destination == Path("\\\\server\\Backup1") / "Lavori" / "Cliente" / "Grafica"
    assert command[0] == "robocopy"
    assert command[1] == str(nested)
    assert command[2] == str(destination)
    assert command[3] == "image.psd"
    assert "/MIR" not in command
    assert "/PURGE" not in command


def test_robocopy_marks_each_file_copied_after_all_destinations(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = tmp_path / "Lavori"
    source.mkdir()
    first = source / "a.docx"
    second = source / "b.docx"
    first.write_text("a", encoding="utf-8")
    second.write_text("b", encoding="utf-8")
    config = parse_config(
        {
            "watch": {
                "include_dirs": [str(source)],
                "include_extensions": [".docx"],
                "exclude_dirs": [],
                "exclude_patterns": [],
            },
            "backup": {
                "engine": "robocopy",
                "robocopy_path": "robocopy",
                "remote_destinations": ["\\\\server\\Backup1", "\\\\server\\Backup2"],
                "server_check": {"type": "tcp", "port": 445, "interval_seconds": 1800, "timeout_seconds": 1},
            },
            "storage": {
                "change_registry": str(tmp_path / "changes.sqlite"),
                "logs_dir": str(tmp_path / "logs"),
                "status_file": str(tmp_path / "status.json"),
            },
            "stability": {
                "min_age_seconds": 0,
                "unchanged_check_interval_seconds": 0,
                "unchanged_checks_required": 0,
            },
        },
        tmp_path / "config.toml",
    )
    registry = ChangeRegistry(config.storage.change_registry)
    events = [
        ChangeEvent(path=str(first), event_type="modified"),
        ChangeEvent(path=str(second), event_type="modified"),
    ]
    for event in events:
        registry.record(event)

    pending_counts_after_commands: list[int] = []

    def fake_run(command, **kwargs):
        pending_counts_after_commands.append(registry.pending_count())
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    monkeypatch.setattr("bakrest.backup.subprocess.run", fake_run)

    result = _run_robocopy(config, events, _NullLogger(), registry)

    assert result.success
    assert result.copied_count == 2
    assert registry.pending_count() == 0
    assert pending_counts_after_commands == [2, 2, 1, 1]


class _NullLogger:
    def info(self, *args, **kwargs) -> None:
        pass

    def warning(self, *args, **kwargs) -> None:
        pass

    def error(self, *args, **kwargs) -> None:
        pass

    def exception(self, *args, **kwargs) -> None:
        pass


def test_run_command_uses_lossy_utf8_decoding(monkeypatch) -> None:
    calls = {}

    def fake_run(*args, **kwargs):
        calls.update(kwargs)
        return subprocess.CompletedProcess(args[0], 0, stdout="", stderr="")

    monkeypatch.setattr("bakrest.backup.subprocess.run", fake_run)

    _run_command(["robocopy"])

    assert calls["encoding"] == "utf-8"
    assert calls["errors"] == "replace"
    assert calls["text"] is True