Christian Manivong
08bfb5c1c0
get_vm_snapshots, create_vm_snapshot, delete_vm_snapshot and
rollback_vm_snapshot for VMs and containers, so netOrk's snapshot view
works on Proxmox as it does on VMware. Proxmox lists the live state as a
pseudo-snapshot named "current"; it is never reported or addressable.
Containers have no RAM state, so include_memory is ignored for them.
reboot_host() restarts the node with POST /nodes/{node}/status
command=reboot instead of /sbin/reboot over SSH.
96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
"""HypervisorDriver snapshot methods on Proxmox (QEMU and LXC)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
SNAPSHOTS = [
|
|
{"name": "base", "description": "clean install", "snaptime": 1700000000, "vmstate": 0},
|
|
{"name": "upgrade", "description": "", "snaptime": 1700000100, "parent": "base", "vmstate": 1},
|
|
{"name": "current", "description": "You are here!", "parent": "upgrade", "running": 1},
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def api(driver):
|
|
node = driver._node_api()
|
|
node.qemu.get.return_value = [{"vmid": 100, "name": "web01"}]
|
|
node.lxc.get.return_value = [{"vmid": 200, "name": "dns01"}]
|
|
node.qemu.return_value.snapshot.get.return_value = SNAPSHOTS
|
|
driver._wait_for_task = MagicMock()
|
|
return node
|
|
|
|
|
|
class TestList:
|
|
def test_flattens_and_skips_the_current_marker(self, driver, api):
|
|
assert driver.get_vm_snapshots("web01") == [
|
|
{
|
|
"name": "base",
|
|
"vm": "web01",
|
|
"created": 1700000000.0,
|
|
"description": "clean install",
|
|
"has_memory": False,
|
|
"parent": "",
|
|
},
|
|
{
|
|
"name": "upgrade",
|
|
"vm": "web01",
|
|
"created": 1700000100.0,
|
|
"description": "",
|
|
"has_memory": True,
|
|
"parent": "base",
|
|
},
|
|
]
|
|
|
|
def test_unknown_vm(self, driver, api):
|
|
with pytest.raises(ValueError):
|
|
driver.get_vm_snapshots("nope")
|
|
|
|
|
|
class TestCreate:
|
|
def test_vm_with_memory(self, driver, api):
|
|
api.qemu.return_value.snapshot.post.return_value = "UPID:snap"
|
|
driver.create_vm_snapshot("100", "pre", description="d", include_memory=True)
|
|
api.qemu.return_value.snapshot.post.assert_called_once_with(
|
|
snapname="pre", description="d", vmstate=1
|
|
)
|
|
driver._wait_for_task.assert_called_once_with("UPID:snap", timeout=600)
|
|
|
|
def test_container_never_saves_memory(self, driver, api):
|
|
api.lxc.return_value.snapshot.get.return_value = []
|
|
driver.create_vm_snapshot("200", "pre", include_memory=True)
|
|
api.lxc.return_value.snapshot.post.assert_called_once_with(snapname="pre", description="")
|
|
|
|
def test_duplicate_name(self, driver, api):
|
|
with pytest.raises(ValueError, match="already"):
|
|
driver.create_vm_snapshot("web01", "base")
|
|
|
|
def test_api_refusal(self, driver, api):
|
|
api.qemu.return_value.snapshot.post.side_effect = Exception(
|
|
"snapshot feature is not available"
|
|
)
|
|
with pytest.raises(RuntimeError, match="not available"):
|
|
driver.create_vm_snapshot("web01", "new")
|
|
|
|
|
|
class TestDeleteAndRollback:
|
|
def test_delete(self, driver, api):
|
|
driver.delete_vm_snapshot("web01", "base")
|
|
api.qemu.return_value.snapshot.assert_called_with("base")
|
|
api.qemu.return_value.snapshot.return_value.delete.assert_called_once_with()
|
|
|
|
def test_rollback(self, driver, api):
|
|
driver.rollback_vm_snapshot("web01", "upgrade")
|
|
api.qemu.return_value.snapshot.return_value.rollback.post.assert_called_once_with()
|
|
|
|
@pytest.mark.parametrize("method", ["delete_vm_snapshot", "rollback_vm_snapshot"])
|
|
def test_unknown_snapshot(self, driver, api, method):
|
|
with pytest.raises(ValueError, match="no snapshot"):
|
|
getattr(driver, method)("web01", "nope")
|
|
|
|
def test_current_is_not_a_snapshot(self, driver, api):
|
|
with pytest.raises(ValueError):
|
|
driver.rollback_vm_snapshot("web01", "current")
|