Codex
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app import commands
|
|
|
|
|
|
def test_start_proxmox_backup_parses_upid(monkeypatch):
|
|
expected = "UPID:pve:0001:0002:0003:vzdump:110:root@pam:"
|
|
|
|
def fake_run(command, timeout=None):
|
|
assert command[:3] == ["pvesh", "create", "/nodes/pve/vzdump"]
|
|
assert "--output-format" in command
|
|
return commands.CommandResult(stdout=json.dumps(expected), stderr="")
|
|
|
|
monkeypatch.setattr(commands, "run", fake_run)
|
|
|
|
assert commands.start_proxmox_backup(
|
|
node="pve",
|
|
vmid=110,
|
|
storage="backup-store",
|
|
mode="snapshot",
|
|
compression="zstd",
|
|
) == expected
|
|
|
|
|
|
def test_start_proxmox_backup_parses_plain_upid(monkeypatch):
|
|
expected = "UPID:pve:0001:0002:0003:vzdump:110:root@pam:"
|
|
|
|
monkeypatch.setattr(commands, "run", lambda command, timeout=None: commands.CommandResult(stdout=expected, stderr=""))
|
|
|
|
assert commands.start_proxmox_backup(
|
|
node="pve",
|
|
vmid=110,
|
|
storage="backup-store",
|
|
mode="snapshot",
|
|
compression="zstd",
|
|
) == expected
|
|
|
|
|
|
def test_start_proxmox_backup_parses_upid_after_info_lines(monkeypatch):
|
|
expected = "UPID:pve:0001:0002:0003:vzdump:110:root@pam:"
|
|
output = f"INFO: starting new backup job\nINFO: Backup job finished successfully\n\"{expected}\""
|
|
|
|
monkeypatch.setattr(commands, "run", lambda command, timeout=None: commands.CommandResult(stdout=output, stderr=""))
|
|
|
|
assert commands.start_proxmox_backup(
|
|
node="pve",
|
|
vmid=110,
|
|
storage="backup-store",
|
|
mode="snapshot",
|
|
compression="zstd",
|
|
) == expected
|
|
|
|
|
|
def test_list_guests_normalizes_vm_and_lxc(monkeypatch):
|
|
payload = [
|
|
{"vmid": "110", "name": "test-guest", "type": "qemu", "node": "pve", "status": "running"},
|
|
{"vmid": "111", "name": "ct", "type": "lxc", "node": "pve", "status": "stopped"},
|
|
]
|
|
|
|
monkeypatch.setattr(commands, "pvesh_json", lambda args: payload)
|
|
|
|
guests = commands.list_guests()
|
|
|
|
assert guests == [
|
|
{"vmid": 110, "name": "test-guest", "type": "vm", "node": "pve", "status": "running"},
|
|
{"vmid": 111, "name": "ct", "type": "lxc", "node": "pve", "status": "stopped"},
|
|
]
|
|
|
|
|
|
def test_discover_archive_finds_newest_matching_file(tmp_path):
|
|
old_file = tmp_path / "vzdump-qemu-110-2026_01_01-00_00_00.vma.zst"
|
|
new_file = tmp_path / "vzdump-qemu-110-2026_01_01-00_01_00.vma.zst"
|
|
log_file = tmp_path / "vzdump-qemu-110-2026_01_01-00_01_00.log"
|
|
wrong_guest = tmp_path / "vzdump-qemu-999-2026_01_01-00_01_00.vma.zst"
|
|
for path in (old_file, new_file, log_file, wrong_guest):
|
|
path.write_text("data")
|
|
old_file.touch()
|
|
new_file.touch()
|
|
log_file.touch()
|
|
|
|
found = commands.discover_archive(str(tmp_path), 110, "vm", "2020-01-01T00:00:00+00:00")
|
|
|
|
assert found == new_file
|
|
|
|
|
|
def test_discover_archive_raises_when_missing(tmp_path):
|
|
with pytest.raises(FileNotFoundError):
|
|
commands.discover_archive(str(tmp_path), 110, "vm", "2020-01-01T00:00:00+00:00")
|
|
|
|
|
|
def test_rclone_commands_use_exact_objects(monkeypatch):
|
|
calls = []
|
|
|
|
def fake_run(command, timeout=None):
|
|
calls.append(command)
|
|
return commands.CommandResult(stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(commands, "run", fake_run)
|
|
|
|
commands.rclone_copyto("/usr/bin/rclone", "/tmp/archive.zst", "crypt:backup/archive.zst")
|
|
commands.rclone_delete_file("/usr/bin/rclone", "crypt:backup/archive.zst")
|
|
commands.rclone_rmdir("/usr/bin/rclone", "crypt:backup")
|
|
|
|
assert calls == [
|
|
["/usr/bin/rclone", "copyto", "/tmp/archive.zst", "crypt:backup/archive.zst"],
|
|
["/usr/bin/rclone", "deletefile", "crypt:backup/archive.zst"],
|
|
["/usr/bin/rclone", "rmdir", "crypt:backup"],
|
|
]
|
|
assert all("sync" not in call for call in calls for call in call)
|