feat: shutdown_host and begin_stop_vm through the API

shutdown_host() posts command=shutdown to the node itself; its guests
stop as on any shutdown. begin_stop_vm(name) asks a guest for its clean
shutdown and returns once Proxmox took it, without waiting for the
task: the caller reads get_vm_status until it stopped. Both raise
RuntimeError when Proxmox refuses; an unknown guest is a ValueError.

The contracts HostShutdownMixin and VmStopRequestMixin of
napalm-device-types (NAPALM/napalm-device-types#55); for netOrk's graph
workflows (NetOrk/netork#1636).

CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `shutdown_host()`: the node shuts itself down through the API
+  (`POST /nodes/{node}/status command=shutdown`), and `begin_stop_vm(name)`: a
+  guest's clean shutdown, returning once Proxmox took it instead of waiting for
+  the task — the contracts `HostShutdownMixin` and `VmStopRequestMixin` of
+  napalm-device-types (NetOrk/netork#1636).
+
 ## [0.8.0] - 2026-10-23
 
 The first release on PyPI. It collects everything since 0.1.0.

napalm_proxmox/vm_contract_mixin.py

@@ -2,7 +2,9 @@
 
 ``power_vm`` stays for callers that already use it; these are what a
 hypervisor-neutral caller talks to. They raise instead of returning a
-``{"success": ...}`` dict, and block until Proxmox reports the task finished.
+``{"success": ...}`` dict, and block until Proxmox reports the task finished —
+except ``begin_stop_vm`` and ``shutdown_host``, which return once Proxmox took
+the request (``VmStopRequestMixin``, ``HostShutdownMixin``).
 """
 
 from __future__ import annotations
@@ -162,6 +164,14 @@ class ProxmoxVMContractMixin:
     def stop_vm(self, name: str, force: bool = False) -> None:
         self._run_power(*self._resolve_vm(name), "stop" if force else "shutdown")
 
+    def begin_stop_vm(self, name: str) -> None:
+        """Ask for the guest's clean shutdown; return once Proxmox accepted it."""
+        vmid, vm_type = self._resolve_vm(name)
+        try:
+            self._guest_api(vmid, vm_type).status.shutdown.post()
+        except Exception as exc:
+            raise RuntimeError(f"shutdown of {vm_type} {vmid} refused: {exc}") from exc
+
     def reboot_vm(self, name: str, force: bool = False) -> None:
         vmid, vm_type = self._resolve_vm(name)
         if not force:
@@ -191,3 +201,10 @@ class ProxmoxVMContractMixin:
             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
+
+    def shutdown_host(self) -> None:
+        """Shut this Proxmox node down through the API; its guests stop as on any shutdown."""
+        try:
+            self._node_api().status.post(command="shutdown")
+        except Exception as exc:
+            raise RuntimeError(f"Shutdown of node {self._node_name!r} refused: {exc}") from exc

tests/test_shutdown_requests.py

@@ -0,0 +1,74 @@
+"""shutdown_host and begin_stop_vm: shutting down without waiting.
+
+Through the API, never over SSH: ``shutdown_host`` asks the node to shut
+itself down, ``begin_stop_vm`` asks a guest for its clean (ACPI) shutdown and
+returns once Proxmox accepted it — the caller reads ``get_vm_status`` until it
+stopped (NetOrk/netork#1636). Neither waits for a task.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+def test_shutdown_host_posts_shutdown_to_the_node(driver):
+    driver.shutdown_host()
+
+    driver._node_api().status.post.assert_called_once_with(command="shutdown")
+
+
+def test_a_refused_shutdown_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.shutdown_host()
+
+
+def _guests(driver, qemu, lxc=()):
+    node = driver._node_api()
+    node.qemu.get.return_value = list(qemu)
+    node.lxc.get.return_value = list(lxc)
+    return node
+
+
+def test_begin_stop_vm_asks_for_a_clean_shutdown_and_does_not_wait(driver, monkeypatch):
+    node = _guests(driver, [{"vmid": 650, "name": "build-01"}])
+    waited = []
+    monkeypatch.setattr(driver, "_wait_for_task", lambda *a, **k: waited.append(a), raising=False)
+
+    driver.begin_stop_vm("650")
+
+    node.qemu.assert_called_with(650)
+    node.qemu(650).status.shutdown.post.assert_called_once_with()
+    assert waited == []
+
+
+def test_begin_stop_vm_finds_a_guest_by_name(driver):
+    node = _guests(driver, [{"vmid": 650, "name": "build-01"}])
+
+    driver.begin_stop_vm("build-01")
+
+    node.qemu(650).status.shutdown.post.assert_called_once_with()
+
+
+def test_begin_stop_vm_of_a_container(driver):
+    node = _guests(driver, [], [{"vmid": 700, "name": "ct"}])
+
+    driver.begin_stop_vm("700")
+
+    node.lxc(700).status.shutdown.post.assert_called_once_with()
+
+
+def test_an_unknown_guest_is_a_value_error(driver):
+    _guests(driver, [])
+
+    with pytest.raises(ValueError):
+        driver.begin_stop_vm("999")
+
+
+def test_a_refused_request_is_a_runtime_error(driver):
+    node = _guests(driver, [{"vmid": 650, "name": "build-01"}])
+    node.qemu(650).status.shutdown.post.side_effect = Exception("VM is locked")
+
+    with pytest.raises(RuntimeError, match="VM is locked"):
+        driver.begin_stop_vm("650")