From 08bfb5c1c0018f0059869b291394587f5259d6d6 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Thu, 24 Sep 2026 10:00:12 +0200 Subject: [PATCH] feat: VM snapshots and reboot_host through the API 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. --- CHANGELOG.md | 4 ++ napalm_proxmox/driver.py | 2 + napalm_proxmox/vm_contract_mixin.py | 9 +++ napalm_proxmox/vm_snapshot_mixin.py | 74 ++++++++++++++++++++++ tests/test_reboot_host.py | 16 +++++ tests/test_vm_snapshots.py | 95 +++++++++++++++++++++++++++++ 6 files changed, 200 insertions(+) create mode 100644 napalm_proxmox/vm_snapshot_mixin.py create mode 100644 tests/test_reboot_host.py create mode 100644 tests/test_vm_snapshots.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a4cdc01..fbb7a3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `suspend_vm` and `get_vm_config`. They accept a VM's name or vmid, raise `ValueError`/`RuntimeError` instead of returning a result dict, and wait for the Proxmox task to finish. `power_vm` is unchanged. +- Snapshot methods `get_vm_snapshots`, `create_vm_snapshot`, + `delete_vm_snapshot`, `rollback_vm_snapshot` for VMs and containers + (containers never save RAM state). +- `reboot_host()` restarts the node through the API instead of SSH. ### Changed - `get_vms()` reports `vmid` as a string (`"100"`), following diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index 035b359..515dca4 100644 --- a/napalm_proxmox/driver.py +++ b/napalm_proxmox/driver.py @@ -50,6 +50,7 @@ from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin from napalm_proxmox.config_mixin import ProxmoxConfigMixin from napalm_proxmox.vm_mixin import ProxmoxVMMixin from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin +from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin from napalm_proxmox.system_mixin import ProxmoxSystemMixin @@ -71,6 +72,7 @@ class ProxmoxDriver( ProxmoxConfigMixin, ProxmoxVMMixin, ProxmoxVMContractMixin, + ProxmoxVMSnapshotMixin, ProxmoxVMProvisionMixin, ProxmoxRoutingMixin, ProxmoxSystemMixin, diff --git a/napalm_proxmox/vm_contract_mixin.py b/napalm_proxmox/vm_contract_mixin.py index d71a117..4f4b44d 100644 --- a/napalm_proxmox/vm_contract_mixin.py +++ b/napalm_proxmox/vm_contract_mixin.py @@ -182,3 +182,12 @@ class ProxmoxVMContractMixin: vmid, vm_type = self._resolve_vm(name) cfg = self._guest_api(vmid, vm_type).config.get() or {} return parse_vm_config(str(vmid), vm_type, cfg) + + # -- the node itself -------------------------------------------------------- + + def reboot_host(self) -> None: + """Restart this Proxmox node through the API (no SSH involved).""" + try: + self._node_api().status.post(command="reboot") + except Exception as exc: + raise RuntimeError(f"Reboot of node {self._node_name!r} refused: {exc}") from exc diff --git a/napalm_proxmox/vm_snapshot_mixin.py b/napalm_proxmox/vm_snapshot_mixin.py new file mode 100644 index 0000000..6c6d052 --- /dev/null +++ b/napalm_proxmox/vm_snapshot_mixin.py @@ -0,0 +1,74 @@ +"""HypervisorDriver snapshot methods for Proxmox VE guests (QEMU and LXC).""" + +from __future__ import annotations + +from typing import Any + +from napalm_device_types.models import SnapshotDict + +#: Proxmox lists the live state as a pseudo-snapshot of this name. +_CURRENT = "current" +#: A RAM snapshot of a large VM takes a while to write out. +_SNAPSHOT_TIMEOUT = 600 + + +class ProxmoxVMSnapshotMixin: + """Relies on ``_resolve_vm``/``_guest_api`` from ProxmoxVMContractMixin.""" + + _resolve_vm: Any + _guest_api: Any + _wait_for_task: Any + + def _snapshots(self, name: str) -> tuple[Any, str, str, list[dict[str, Any]]]: + vmid, vm_type = self._resolve_vm(name) + api = self._guest_api(vmid, vm_type) + raw = [s for s in api.snapshot.get() or [] if s.get("name") != _CURRENT] + return api, str(vmid), vm_type, raw + + def _run_task(self, call: Any, *args: Any, **kwargs: Any) -> None: + try: + upid = call(*args, **kwargs) + except Exception as exc: + raise RuntimeError(str(exc)) from exc + if upid: + self._wait_for_task(upid, timeout=_SNAPSHOT_TIMEOUT) + + @staticmethod + def _require(raw: list[dict[str, Any]], snapshot: str, vm: str) -> None: + if not any(s.get("name") == snapshot for s in raw): + raise ValueError(f"VM {vm!r} has no snapshot named {snapshot!r}") + + def get_vm_snapshots(self, name: str) -> list[SnapshotDict]: + _, _, _, raw = self._snapshots(name) + return [ + { + "name": s["name"], + "vm": name, + "created": float(s.get("snaptime", 0)), + "description": s.get("description", ""), + "has_memory": bool(s.get("vmstate")), + "parent": s.get("parent", ""), + } + for s in raw + ] + + def create_vm_snapshot( + self, name: str, snapshot: str, description: str = "", include_memory: bool = False + ) -> None: + api, _, vm_type, raw = self._snapshots(name) + if any(s.get("name") == snapshot for s in raw): + raise ValueError(f"VM {name!r} already has a snapshot named {snapshot!r}") + kwargs: dict[str, Any] = {"snapname": snapshot, "description": description} + if vm_type == "vm": # containers have no RAM state to save + kwargs["vmstate"] = 1 if include_memory else 0 + self._run_task(api.snapshot.post, **kwargs) + + def delete_vm_snapshot(self, name: str, snapshot: str) -> None: + api, _, _, raw = self._snapshots(name) + self._require(raw, snapshot, name) + self._run_task(api.snapshot(snapshot).delete) + + def rollback_vm_snapshot(self, name: str, snapshot: str) -> None: + api, _, _, raw = self._snapshots(name) + self._require(raw, snapshot, name) + self._run_task(api.snapshot(snapshot).rollback.post) diff --git a/tests/test_reboot_host.py b/tests/test_reboot_host.py new file mode 100644 index 0000000..2b3d08e --- /dev/null +++ b/tests/test_reboot_host.py @@ -0,0 +1,16 @@ +"""reboot_host: restart the Proxmox node itself through the API, not over SSH.""" + +from __future__ import annotations + +import pytest + + +def test_posts_reboot_to_the_node(driver): + driver.reboot_host() + driver._node_api().status.post.assert_called_once_with(command="reboot") + + +def test_api_refusal_is_a_runtime_error(driver): + driver._node_api().status.post.side_effect = Exception("Permission check failed") + with pytest.raises(RuntimeError, match="Permission check failed"): + driver.reboot_host() diff --git a/tests/test_vm_snapshots.py b/tests/test_vm_snapshots.py new file mode 100644 index 0000000..c878d7f --- /dev/null +++ b/tests/test_vm_snapshots.py @@ -0,0 +1,95 @@ +"""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")