pve-cloud-backups/backend/tests/test_backup_service.py

import asyncio
from pathlib import Path

import pytest

from app.db import db, utc_now
from app.jobs import create_job
from app.settings_store import set_settings


def _settings(tmp_path):
    return {
        "setup_complete": False,
        "proxmox_node": "pve",
        "proxmox_storage": "backup-store",
        "local_backup_dir": str(tmp_path),
        "rclone_path": "/usr/bin/rclone",
        "rclone_remote": "onedrive",
        "rclone_remote_path": "Backups/Proxmox",
        "discord_webhook_url": "",
        "allowed_cors_origins": [],
        "default_compression": "zstd",
        "default_backup_mode": "snapshot",
        "default_retention_type": "days",
        "default_retention_value": 30,
        "max_concurrent_backups": 1,
    }


def _job():
    return create_job(
        {
            "guest_vmid": 110,
            "guest_name": "test-guest",
            "guest_type": "vm",
            "node": "pve",
            "enabled": True,
            "cron_schedule": "0 2 * * *",
            "backup_mode": "snapshot",
            "compression": "zstd",
            "retention_type": "latest",
            "retention_value": 1,
        }
    )


def test_backup_state_transitions_complete_and_delete_local_safely(
    isolated_db, no_notifications, tmp_path, monkeypatch
):
    from app import backup_service, commands

    set_settings(_settings(tmp_path))
    job = _job()
    backup = backup_service.queue_backup_for_job(job)
    archive = tmp_path / "vzdump-qemu-110-2026_01_01-00_00_00.vma.zst"
    archive.write_bytes(b"backup")

    monkeypatch.setattr(commands, "start_proxmox_backup", lambda **kwargs: "UPID:pve:test")
    monkeypatch.setattr(commands, "get_task_status", lambda node, upid: {"status": "stopped", "exitstatus": "OK"})
    monkeypatch.setattr(commands, "discover_archive", lambda *args: archive)
    monkeypatch.setattr(commands, "rclone_copyto", lambda *args: None)
    monkeypatch.setattr(commands, "rclone_object_exists", lambda *args: True)
    monkeypatch.setattr(backup_service, "apply_retention_for_job", lambda job_id: asyncio.sleep(0))

    asyncio.run(backup_service.process_backup(backup["id"]))
    final = backup_service.get_backup(backup["id"])

    assert final["state"] == "completed"
    assert final["upid"] == "UPID:pve:test"
    assert final["remote_path"].startswith("onedrive:Backups/Proxmox/")
    assert not archive.exists()


def test_only_one_worker_can_claim_queued_backup(isolated_db, no_notifications, tmp_path):
    from app import backup_service

    set_settings(_settings(tmp_path))
    job = _job()
    backup = backup_service.queue_backup_for_job(job)

    first = backup_service._claim_queued_backup(backup["id"])
    second = backup_service._claim_queued_backup(backup["id"])

    assert first is not None
    assert first["state"] == "pve_running"
    assert second is None


def test_pve_running_without_upid_fails_without_polling_none(isolated_db, no_notifications, monkeypatch):
    from app import backup_service, commands

    poll_calls = []
    monkeypatch.setattr(commands, "get_task_status", lambda *args: poll_calls.append(args))

    with db() as conn:
        conn.execute(
            """
            INSERT INTO backups(id, guest_vmid, guest_name, guest_type, node, state, started_at, updated_at)
            VALUES ('missing-upid', 110, 'test-guest', 'lxc', 'pve', 'pve_running', ?, ?)
            """,
            (utc_now(), utc_now()),
        )

    asyncio.run(backup_service.process_backup("missing-upid"))
    final = backup_service.get_backup("missing-upid")

    assert final["state"] == "failed"
    assert "no Proxmox UPID" in final["error_message"]
    assert poll_calls == []


def test_local_archive_is_not_deleted_when_remote_verification_fails(
    isolated_db, no_notifications, tmp_path, monkeypatch
):
    from app import backup_service, commands

    set_settings(_settings(tmp_path))
    job = _job()
    backup = backup_service.queue_backup_for_job(job)
    archive = tmp_path / "vzdump-qemu-110-2026_01_01-00_00_00.vma.zst"
    archive.write_bytes(b"backup")

    monkeypatch.setattr(commands, "start_proxmox_backup", lambda **kwargs: "UPID:pve:test")
    monkeypatch.setattr(commands, "get_task_status", lambda node, upid: {"status": "stopped", "exitstatus": "OK"})
    monkeypatch.setattr(commands, "discover_archive", lambda *args: archive)
    monkeypatch.setattr(commands, "rclone_copyto", lambda *args: None)
    monkeypatch.setattr(commands, "rclone_object_exists", lambda *args: False)

    asyncio.run(backup_service.process_backup(backup["id"]))
    final = backup_service.get_backup(backup["id"])

    assert final["state"] == "failed"
    assert archive.exists()


def test_retention_latest_selects_expired_remote_backups(
    isolated_db, no_notifications, tmp_path, monkeypatch
):
    from app import backup_service

    set_settings(_settings(tmp_path))
    job = _job()
    deleted = []

    async def fake_delete_remote_backup(backup, reason):
        deleted.append((backup["id"], reason))

    monkeypatch.setattr(backup_service, "delete_remote_backup", fake_delete_remote_backup)
    with db() as conn:
        for index in range(3):
            backup_id = f"backup-{index}"
            conn.execute(
                """
                INSERT INTO backups(
                    id, job_id, guest_vmid, guest_name, guest_type, node, state,
                    remote_path, started_at, completed_at, updated_at
                )
                VALUES (?, ?, 110, 'test-guest', 'vm', 'pve', 'completed', ?, ?, ?, ?)
                """,
                (
                    backup_id,
                    job["id"],
                    f"crypt:backup/{index}",
                    f"2026-01-01T00:0{index}:00+00:00",
                    f"2026-01-01T00:0{index}:30+00:00",
                    utc_now(),
                ),
            )

    asyncio.run(backup_service.apply_retention_for_job(job["id"]))

    assert deleted == [
        ("backup-1", "retention latest=1"),
        ("backup-0", "retention latest=1"),
    ]


def test_startup_recovery_resumes_known_states(isolated_db, no_notifications, monkeypatch):
    from app import backup_service

    resumed = []
    deleted = []

    async def fake_process_backup(backup_id):
        resumed.append(backup_id)

    async def fake_delete_remote_backup(backup, reason):
        deleted.append((backup["id"], reason))

    monkeypatch.setattr(backup_service, "process_backup", fake_process_backup)
    monkeypatch.setattr(backup_service, "delete_remote_backup", fake_delete_remote_backup)

    with db() as conn:
        for backup_id, state in (("b1", "local_ready"), ("b2", "deleting"), ("b3", "completed")):
            conn.execute(
                """
                INSERT INTO backups(id, guest_vmid, guest_name, guest_type, node, state, started_at, updated_at)
                VALUES (?, 110, 'test-guest', 'vm', 'pve', ?, ?, ?)
                """,
                (backup_id, state, utc_now(), utc_now()),
            )

    asyncio.run(backup_service.recover_unfinished())

    assert resumed == ["b1"]
    assert deleted == [("b2", "recovery delete resume")]


def test_retention_sweep_applies_all_jobs_with_completed_remote_backups(
    isolated_db, no_notifications, tmp_path, monkeypatch
):
    from app import backup_service

    set_settings(_settings(tmp_path))
    job = _job()
    swept = []

    async def fake_apply_retention_for_job(job_id):
        swept.append(job_id)

    monkeypatch.setattr(backup_service, "apply_retention_for_job", fake_apply_retention_for_job)
    with db() as conn:
        conn.execute(
            """
            INSERT INTO backups(
                id, job_id, guest_vmid, guest_name, guest_type, node, state,
                remote_path, started_at, completed_at, updated_at
            )
            VALUES ('sweep-test', ?, 110, 'test-guest', 'vm', 'pve', 'completed',
                    'onedrive:pve-cloud-backup/sweep-test/archive.tar.zst', ?, ?, ?)
            """,
            (job["id"], utc_now(), utc_now(), utc_now()),
        )

    asyncio.run(backup_service.retention_sweep())

    assert swept == [job["id"]]


def test_manual_backup_delete_removes_remote_and_known_local_file(
    isolated_db, no_notifications, tmp_path, monkeypatch
):
    from app import backup_service, commands

    set_settings(_settings(tmp_path))
    local_file = tmp_path / "orphaned-local.tar.zst"
    local_file.write_text("backup")
    deleted_files = []
    removed_dirs = []

    monkeypatch.setattr(commands, "rclone_object_exists", lambda *args: True)
    monkeypatch.setattr(commands, "rclone_delete_file", lambda *args: deleted_files.append(args))
    monkeypatch.setattr(commands, "rclone_rmdir", lambda *args: removed_dirs.append(args))

    with db() as conn:
        conn.execute(
            """
            INSERT INTO backups(
                id, guest_vmid, guest_name, guest_type, node, state,
                local_path, remote_path, started_at, completed_at, updated_at
            )
            VALUES ('manual-delete', 110, 'test-guest', 'vm', 'pve', 'completed',
                    ?, 'onedrive:pve-cloud-backup/manual-delete/archive.tar.zst', ?, ?, ?)
            """,
            (str(local_file), utc_now(), utc_now(), utc_now()),
        )

    deleted = asyncio.run(backup_service.delete_backup_manually("manual-delete"))

    assert deleted["state"] == "deleted"
    assert not local_file.exists()
    assert deleted_files == [
        (
            "/usr/bin/rclone",
            "onedrive:pve-cloud-backup/manual-delete/archive.tar.zst",
        )
    ]
    assert removed_dirs == [
        (
            "/usr/bin/rclone",
            "onedrive:pve-cloud-backup/manual-delete",
        )
    ]


def test_manual_backup_delete_removes_empty_remote_folder_when_file_is_already_missing(
    isolated_db, no_notifications, tmp_path, monkeypatch
):
    from app import backup_service, commands

    set_settings(_settings(tmp_path))
    removed_dirs = []

    monkeypatch.setattr(commands, "rclone_object_exists", lambda *args: False)
    monkeypatch.setattr(commands, "rclone_delete_file", lambda *args: pytest.fail("deletefile should not run"))
    monkeypatch.setattr(commands, "rclone_rmdir", lambda *args: removed_dirs.append(args))

    with db() as conn:
        conn.execute(
            """
            INSERT INTO backups(
                id, guest_vmid, guest_name, guest_type, node, state,
                remote_path, started_at, completed_at, updated_at
            )
            VALUES ('missing-remote-file', 110, 'test-guest', 'vm', 'pve', 'completed',
                    'onedrive:pve-cloud-backup/missing-remote-file/archive.tar.zst', ?, ?, ?)
            """,
            (utc_now(), utc_now(), utc_now()),
        )

    deleted = asyncio.run(backup_service.delete_backup_manually("missing-remote-file"))

    assert deleted["state"] == "deleted"
    assert removed_dirs == [
        (
            "/usr/bin/rclone",
            "onedrive:pve-cloud-backup/missing-remote-file",
        )
    ]


def test_manual_backup_delete_refuses_active_backup(isolated_db, no_notifications):
    from app import backup_service

    with db() as conn:
        conn.execute(
            """
            INSERT INTO backups(id, guest_vmid, guest_name, guest_type, node, state, started_at, updated_at)
            VALUES ('active-delete', 110, 'test-guest', 'vm', 'pve', 'uploading', ?, ?)
            """,
            (utc_now(), utc_now()),
        )

    with pytest.raises(RuntimeError):
        asyncio.run(backup_service.delete_backup_manually("active-delete"))


def test_deleting_already_deleted_backup_purges_metadata_row(isolated_db, no_notifications):
    from app import backup_service

    with db() as conn:
        conn.execute(
            """
            INSERT INTO backups(
                id, guest_vmid, guest_name, guest_type, node, state,
                remote_path, started_at, deleted_at, updated_at
            )
            VALUES ('purge-deleted', 110, 'test-guest', 'vm', 'pve', 'deleted',
                    'onedrive:pve-cloud-backup/purge-deleted/archive.tar.zst', ?, ?, ?)
            """,
            (utc_now(), utc_now(), utc_now()),
        )

    result = asyncio.run(backup_service.delete_backup_manually("purge-deleted"))

    assert result == {"id": "purge-deleted", "purged": True}
    with db() as conn:
        assert conn.execute("SELECT COUNT(*) AS count FROM backups WHERE id = 'purge-deleted'").fetchone()["count"] == 0


def test_delete_job_deletes_backups_before_removing_job(
    isolated_db, no_notifications, tmp_path, monkeypatch
):
    from app import backup_service

    set_settings(_settings(tmp_path))
    job = _job()
    deleted = []

    async def fake_delete_backup_manually(backup_id, reason="manual deletion"):
        deleted.append((backup_id, reason))
        with db() as conn:
            conn.execute("UPDATE backups SET state = 'deleted', deleted_at = ?, updated_at = ? WHERE id = ?", (utc_now(), utc_now(), backup_id))
            row = conn.execute("SELECT * FROM backups WHERE id = ?", (backup_id,)).fetchone()
        return dict(row)

    monkeypatch.setattr(backup_service, "delete_backup_manually", fake_delete_backup_manually)
    with db() as conn:
        conn.execute(
            """
            INSERT INTO backups(id, job_id, guest_vmid, guest_name, guest_type, node, state, remote_path, started_at, completed_at, updated_at)
            VALUES ('job-delete-backup', ?, 110, 'test-guest', 'vm', 'pve', 'completed',
                    'onedrive:pve-cloud-backup/job-delete-backup/archive.tar.zst', ?, ?, ?)
            """,
            (job["id"], utc_now(), utc_now(), utc_now()),
        )

    result = asyncio.run(backup_service.delete_job_with_backups(job["id"]))

    assert result == {"job_id": job["id"], "deleted_backups": ["job-delete-backup"]}
    assert deleted == [("job-delete-backup", f"job {job['id']} deleted")]
    with db() as conn:
        assert conn.execute("SELECT COUNT(*) AS count FROM backup_jobs WHERE id = ?", (job["id"],)).fetchone()["count"] == 0