mirror of
https://github.com/home-assistant/core.git
synced 2026-06-06 07:26:58 +01:00
Add button platform to Proxmox (#163791)
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Erwin Douna
GitHub
@@ -37,7 +37,10 @@ from .const import (
|
||||
)
|
||||
from .coordinator import ProxmoxConfigEntry, ProxmoxCoordinator
|
||||
|
||||
PLATFORMS = [Platform.BINARY_SENSOR]
|
||||
PLATFORMS = [
|
||||
Platform.BINARY_SENSOR,
|
||||
Platform.BUTTON,
|
||||
]
|
||||
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
|
||||
@@ -0,0 +1,339 @@
|
||||
"""Button platform for Proxmox VE."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import abstractmethod
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from proxmoxer import AuthenticationError
|
||||
from proxmoxer.core import ResourceException
|
||||
import requests
|
||||
from requests.exceptions import ConnectTimeout, SSLError
|
||||
|
||||
from homeassistant.components.button import (
|
||||
ButtonDeviceClass,
|
||||
ButtonEntity,
|
||||
ButtonEntityDescription,
|
||||
)
|
||||
from homeassistant.const import EntityCategory
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import ProxmoxConfigEntry, ProxmoxCoordinator, ProxmoxNodeData
|
||||
from .entity import ProxmoxContainerEntity, ProxmoxNodeEntity, ProxmoxVMEntity
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class ProxmoxNodeButtonNodeEntityDescription(ButtonEntityDescription):
|
||||
"""Class to hold Proxmox node button description."""
|
||||
|
||||
press_action: Callable[[ProxmoxCoordinator, str], None]
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class ProxmoxVMButtonEntityDescription(ButtonEntityDescription):
|
||||
"""Class to hold Proxmox VM button description."""
|
||||
|
||||
press_action: Callable[[ProxmoxCoordinator, str, int], None]
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class ProxmoxContainerButtonEntityDescription(ButtonEntityDescription):
|
||||
"""Class to hold Proxmox container button description."""
|
||||
|
||||
press_action: Callable[[ProxmoxCoordinator, str, int], None]
|
||||
|
||||
|
||||
NODE_BUTTONS: tuple[ProxmoxNodeButtonNodeEntityDescription, ...] = (
|
||||
ProxmoxNodeButtonNodeEntityDescription(
|
||||
key="reboot",
|
||||
press_action=lambda coordinator, node: coordinator.proxmox.nodes(
|
||||
node
|
||||
).status.post(command="reboot"),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
device_class=ButtonDeviceClass.RESTART,
|
||||
),
|
||||
ProxmoxNodeButtonNodeEntityDescription(
|
||||
key="shutdown",
|
||||
translation_key="shutdown",
|
||||
press_action=lambda coordinator, node: coordinator.proxmox.nodes(
|
||||
node
|
||||
).status.post(command="shutdown"),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
ProxmoxNodeButtonNodeEntityDescription(
|
||||
key="start_all",
|
||||
translation_key="start_all",
|
||||
press_action=lambda coordinator, node: coordinator.proxmox.nodes(
|
||||
node
|
||||
).startall.post(),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
ProxmoxNodeButtonNodeEntityDescription(
|
||||
key="stop_all",
|
||||
translation_key="stop_all",
|
||||
press_action=lambda coordinator, node: coordinator.proxmox.nodes(
|
||||
node
|
||||
).stopall.post(),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
)
|
||||
|
||||
VM_BUTTONS: tuple[ProxmoxVMButtonEntityDescription, ...] = (
|
||||
ProxmoxVMButtonEntityDescription(
|
||||
key="start",
|
||||
translation_key="start",
|
||||
press_action=lambda coordinator, node, vmid: (
|
||||
coordinator.proxmox.nodes(node).qemu(vmid).status.start.post()
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
ProxmoxVMButtonEntityDescription(
|
||||
key="stop",
|
||||
translation_key="stop",
|
||||
press_action=lambda coordinator, node, vmid: (
|
||||
coordinator.proxmox.nodes(node).qemu(vmid).status.stop.post()
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
ProxmoxVMButtonEntityDescription(
|
||||
key="restart",
|
||||
press_action=lambda coordinator, node, vmid: (
|
||||
coordinator.proxmox.nodes(node).qemu(vmid).status.restart.post()
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
device_class=ButtonDeviceClass.RESTART,
|
||||
),
|
||||
ProxmoxVMButtonEntityDescription(
|
||||
key="hibernate",
|
||||
translation_key="hibernate",
|
||||
press_action=lambda coordinator, node, vmid: (
|
||||
coordinator.proxmox.nodes(node).qemu(vmid).status.hibernate.post()
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
ProxmoxVMButtonEntityDescription(
|
||||
key="reset",
|
||||
translation_key="reset",
|
||||
press_action=lambda coordinator, node, vmid: (
|
||||
coordinator.proxmox.nodes(node).qemu(vmid).status.reset.post()
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
)
|
||||
|
||||
CONTAINER_BUTTONS: tuple[ProxmoxContainerButtonEntityDescription, ...] = (
|
||||
ProxmoxContainerButtonEntityDescription(
|
||||
key="start",
|
||||
translation_key="start",
|
||||
press_action=lambda coordinator, node, vmid: (
|
||||
coordinator.proxmox.nodes(node).lxc(vmid).status.start.post()
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
ProxmoxContainerButtonEntityDescription(
|
||||
key="stop",
|
||||
translation_key="stop",
|
||||
press_action=lambda coordinator, node, vmid: (
|
||||
coordinator.proxmox.nodes(node).lxc(vmid).status.stop.post()
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
ProxmoxContainerButtonEntityDescription(
|
||||
key="restart",
|
||||
press_action=lambda coordinator, node, vmid: (
|
||||
coordinator.proxmox.nodes(node).lxc(vmid).status.restart.post()
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
device_class=ButtonDeviceClass.RESTART,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: ProxmoxConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up ProxmoxVE buttons."""
|
||||
coordinator = entry.runtime_data
|
||||
|
||||
def _async_add_new_nodes(nodes: list[ProxmoxNodeData]) -> None:
|
||||
"""Add new node buttons."""
|
||||
async_add_entities(
|
||||
ProxmoxNodeButtonEntity(coordinator, entity_description, node)
|
||||
for node in nodes
|
||||
for entity_description in NODE_BUTTONS
|
||||
)
|
||||
|
||||
def _async_add_new_vms(
|
||||
vms: list[tuple[ProxmoxNodeData, dict[str, Any]]],
|
||||
) -> None:
|
||||
"""Add new VM buttons."""
|
||||
async_add_entities(
|
||||
ProxmoxVMButtonEntity(coordinator, entity_description, vm, node_data)
|
||||
for (node_data, vm) in vms
|
||||
for entity_description in VM_BUTTONS
|
||||
)
|
||||
|
||||
def _async_add_new_containers(
|
||||
containers: list[tuple[ProxmoxNodeData, dict[str, Any]]],
|
||||
) -> None:
|
||||
"""Add new container buttons."""
|
||||
async_add_entities(
|
||||
ProxmoxContainerButtonEntity(
|
||||
coordinator, entity_description, container, node_data
|
||||
)
|
||||
for (node_data, container) in containers
|
||||
for entity_description in CONTAINER_BUTTONS
|
||||
)
|
||||
|
||||
coordinator.new_nodes_callbacks.append(_async_add_new_nodes)
|
||||
coordinator.new_vms_callbacks.append(_async_add_new_vms)
|
||||
coordinator.new_containers_callbacks.append(_async_add_new_containers)
|
||||
|
||||
_async_add_new_nodes(
|
||||
[
|
||||
node_data
|
||||
for node_data in coordinator.data.values()
|
||||
if node_data.node["node"] in coordinator.known_nodes
|
||||
]
|
||||
)
|
||||
_async_add_new_vms(
|
||||
[
|
||||
(node_data, vm_data)
|
||||
for node_data in coordinator.data.values()
|
||||
for vmid, vm_data in node_data.vms.items()
|
||||
if (node_data.node["node"], vmid) in coordinator.known_vms
|
||||
]
|
||||
)
|
||||
_async_add_new_containers(
|
||||
[
|
||||
(node_data, container_data)
|
||||
for node_data in coordinator.data.values()
|
||||
for vmid, container_data in node_data.containers.items()
|
||||
if (node_data.node["node"], vmid) in coordinator.known_containers
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
class ProxmoxBaseButton(ButtonEntity):
|
||||
"""Common base for Proxmox buttons. Basically to ensure the async_press logic isn't duplicated."""
|
||||
|
||||
entity_description: ButtonEntityDescription
|
||||
coordinator: ProxmoxCoordinator
|
||||
|
||||
@abstractmethod
|
||||
async def _async_press_call(self) -> None:
|
||||
"""Abstract method used per Proxmox button class."""
|
||||
|
||||
async def async_press(self) -> None:
|
||||
"""Trigger the Proxmox button press service."""
|
||||
try:
|
||||
await self._async_press_call()
|
||||
except AuthenticationError as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="cannot_connect_no_details",
|
||||
) from err
|
||||
except SSLError as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="invalid_auth_no_details",
|
||||
) from err
|
||||
except ConnectTimeout as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="timeout_connect_no_details",
|
||||
) from err
|
||||
except (ResourceException, requests.exceptions.ConnectionError) as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="api_error_no_details",
|
||||
) from err
|
||||
|
||||
|
||||
class ProxmoxNodeButtonEntity(ProxmoxNodeEntity, ProxmoxBaseButton):
|
||||
"""Represents a Proxmox Node button entity."""
|
||||
|
||||
entity_description: ProxmoxNodeButtonNodeEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: ProxmoxCoordinator,
|
||||
entity_description: ProxmoxNodeButtonNodeEntityDescription,
|
||||
node_data: ProxmoxNodeData,
|
||||
) -> None:
|
||||
"""Initialize the Proxmox Node button entity."""
|
||||
self.entity_description = entity_description
|
||||
super().__init__(coordinator, node_data)
|
||||
|
||||
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{node_data.node['id']}_{entity_description.key}"
|
||||
|
||||
async def _async_press_call(self) -> None:
|
||||
"""Execute the node button action via executor."""
|
||||
await self.hass.async_add_executor_job(
|
||||
self.entity_description.press_action,
|
||||
self.coordinator,
|
||||
self._node_data.node["node"],
|
||||
)
|
||||
|
||||
|
||||
class ProxmoxVMButtonEntity(ProxmoxVMEntity, ProxmoxBaseButton):
|
||||
"""Represents a Proxmox VM button entity."""
|
||||
|
||||
entity_description: ProxmoxVMButtonEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: ProxmoxCoordinator,
|
||||
entity_description: ProxmoxVMButtonEntityDescription,
|
||||
vm_data: dict[str, Any],
|
||||
node_data: ProxmoxNodeData,
|
||||
) -> None:
|
||||
"""Initialize the Proxmox VM button entity."""
|
||||
self.entity_description = entity_description
|
||||
super().__init__(coordinator, vm_data, node_data)
|
||||
|
||||
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{self.device_id}_{entity_description.key}"
|
||||
|
||||
async def _async_press_call(self) -> None:
|
||||
"""Execute the VM button action via executor."""
|
||||
await self.hass.async_add_executor_job(
|
||||
self.entity_description.press_action,
|
||||
self.coordinator,
|
||||
self._node_name,
|
||||
self.vm_data["vmid"],
|
||||
)
|
||||
|
||||
|
||||
class ProxmoxContainerButtonEntity(ProxmoxContainerEntity, ProxmoxBaseButton):
|
||||
"""Represents a Proxmox Container button entity."""
|
||||
|
||||
entity_description: ProxmoxContainerButtonEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: ProxmoxCoordinator,
|
||||
entity_description: ProxmoxContainerButtonEntityDescription,
|
||||
container_data: dict[str, Any],
|
||||
node_data: ProxmoxNodeData,
|
||||
) -> None:
|
||||
"""Initialize the Proxmox Container button entity."""
|
||||
self.entity_description = entity_description
|
||||
super().__init__(coordinator, container_data, node_data)
|
||||
|
||||
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{self.device_id}_{entity_description.key}"
|
||||
|
||||
async def _async_press_call(self) -> None:
|
||||
"""Execute the container button action via executor."""
|
||||
await self.hass.async_add_executor_job(
|
||||
self.entity_description.press_action,
|
||||
self.coordinator,
|
||||
self._node_name,
|
||||
self.container_data["vmid"],
|
||||
)
|
||||
@@ -0,0 +1,18 @@
|
||||
{
|
||||
"entity": {
|
||||
"button": {
|
||||
"hibernate": {
|
||||
"default": "mdi:power-sleep"
|
||||
},
|
||||
"reset": {
|
||||
"default": "mdi:restart"
|
||||
},
|
||||
"start": {
|
||||
"default": "mdi:play"
|
||||
},
|
||||
"stop": {
|
||||
"default": "mdi:stop"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -54,15 +54,47 @@
|
||||
"status": {
|
||||
"name": "Status"
|
||||
}
|
||||
},
|
||||
"button": {
|
||||
"hibernate": {
|
||||
"name": "Hibernate"
|
||||
},
|
||||
"reset": {
|
||||
"name": "Reset"
|
||||
},
|
||||
"shutdown": {
|
||||
"name": "Shutdown"
|
||||
},
|
||||
"start": {
|
||||
"name": "Start"
|
||||
},
|
||||
"start_all": {
|
||||
"name": "Start all"
|
||||
},
|
||||
"stop": {
|
||||
"name": "Stop"
|
||||
},
|
||||
"stop_all": {
|
||||
"name": "Stop all"
|
||||
}
|
||||
}
|
||||
},
|
||||
"exceptions": {
|
||||
"api_error_no_details": {
|
||||
"message": "An error occurred while communicating with the Proxmox VE instance."
|
||||
},
|
||||
"cannot_connect": {
|
||||
"message": "An error occurred while trying to connect to the Proxmox VE instance: {error}"
|
||||
},
|
||||
"cannot_connect_no_details": {
|
||||
"message": "Could not connect to the Proxmox VE instance."
|
||||
},
|
||||
"invalid_auth": {
|
||||
"message": "An error occurred while trying to authenticate: {error}"
|
||||
},
|
||||
"invalid_auth_no_details": {
|
||||
"message": "Authentication failed for the Proxmox VE instance."
|
||||
},
|
||||
"no_nodes_found": {
|
||||
"message": "No active nodes were found on the Proxmox VE server."
|
||||
},
|
||||
@@ -71,6 +103,9 @@
|
||||
},
|
||||
"timeout_connect": {
|
||||
"message": "A timeout occurred while trying to connect to the Proxmox VE instance: {error}"
|
||||
},
|
||||
"timeout_connect_no_details": {
|
||||
"message": "A timeout occurred while trying to connect to the Proxmox VE instance."
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
|
||||
@@ -89,31 +89,41 @@ def mock_proxmox_client():
|
||||
qemu_by_vmid = {vm["vmid"]: vm for vm in qemu_list}
|
||||
lxc_by_vmid = {vm["vmid"]: vm for vm in lxc_list}
|
||||
|
||||
# Note to reviewer: I will expand on these fixtures in a next PR
|
||||
# Necessary evil to handle the binary_sensor tests properly
|
||||
# Cache resource mocks by vmid so callers (e.g. button tests) can
|
||||
# inspect specific call counts after pressing a button.
|
||||
qemu_mocks: dict[int, MagicMock] = {}
|
||||
lxc_mocks: dict[int, MagicMock] = {}
|
||||
|
||||
def _qemu_resource(vmid: int) -> MagicMock:
|
||||
"""Return a mock resource the QEMU."""
|
||||
resource = MagicMock()
|
||||
vm = qemu_by_vmid[vmid]
|
||||
resource.status.current.get.return_value = {
|
||||
"name": vm["name"],
|
||||
"status": vm["status"],
|
||||
}
|
||||
return resource
|
||||
"""Return a cached mock resource for a QEMU VM."""
|
||||
if vmid not in qemu_mocks:
|
||||
resource = MagicMock()
|
||||
vm = qemu_by_vmid[vmid]
|
||||
resource.status.current.get.return_value = {
|
||||
"name": vm["name"],
|
||||
"status": vm["status"],
|
||||
}
|
||||
qemu_mocks[vmid] = resource
|
||||
return qemu_mocks[vmid]
|
||||
|
||||
def _lxc_resource(vmid: int) -> MagicMock:
|
||||
"""Return a mock resource the LXC."""
|
||||
resource = MagicMock()
|
||||
ct = lxc_by_vmid[vmid]
|
||||
resource.status.current.get.return_value = {
|
||||
"name": ct["name"],
|
||||
"status": ct["status"],
|
||||
}
|
||||
return resource
|
||||
"""Return a cached mock resource for an LXC container."""
|
||||
if vmid not in lxc_mocks:
|
||||
resource = MagicMock()
|
||||
ct = lxc_by_vmid[vmid]
|
||||
resource.status.current.get.return_value = {
|
||||
"name": ct["name"],
|
||||
"status": ct["status"],
|
||||
}
|
||||
lxc_mocks[vmid] = resource
|
||||
return lxc_mocks[vmid]
|
||||
|
||||
node_mock.qemu.side_effect = _qemu_resource
|
||||
node_mock.lxc.side_effect = _lxc_resource
|
||||
|
||||
mock_instance._qemu_mocks = qemu_mocks
|
||||
mock_instance._lxc_mocks = lxc_mocks
|
||||
|
||||
nodes_mock = MagicMock()
|
||||
nodes_mock.get.return_value = load_json_array_fixture(
|
||||
"nodes/nodes.json", DOMAIN
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,315 @@
|
||||
"""Tests for the ProxmoxVE button platform."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from proxmoxer import AuthenticationError
|
||||
from proxmoxer.core import ResourceException
|
||||
import pytest
|
||||
from requests.exceptions import ConnectTimeout, SSLError
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.button import SERVICE_PRESS
|
||||
from homeassistant.const import ATTR_ENTITY_ID, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from . import setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry, snapshot_platform
|
||||
|
||||
BUTTON_DOMAIN = "button"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def enable_all_entities(entity_registry_enabled_by_default: None) -> None:
|
||||
"""Enable all entities for button tests."""
|
||||
|
||||
|
||||
async def test_all_button_entities(
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
mock_proxmox_client: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Snapshot test for all ProxmoxVE button entities."""
|
||||
with patch(
|
||||
"homeassistant.components.proxmoxve.PLATFORMS",
|
||||
[Platform.BUTTON],
|
||||
):
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
await snapshot_platform(
|
||||
hass, entity_registry, snapshot, mock_config_entry.entry_id
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("entity_id", "command"),
|
||||
[
|
||||
("button.pve1_restart", "reboot"),
|
||||
("button.pve1_shutdown", "shutdown"),
|
||||
],
|
||||
)
|
||||
async def test_node_buttons(
|
||||
hass: HomeAssistant,
|
||||
mock_proxmox_client: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_id: str,
|
||||
command: str,
|
||||
) -> None:
|
||||
"""Test pressing a ProxmoxVE node action button triggers the correct API call."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
method_mock = mock_proxmox_client._node_mock.status.post
|
||||
pre_calls = len(method_mock.mock_calls)
|
||||
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
SERVICE_PRESS,
|
||||
{ATTR_ENTITY_ID: entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
assert len(method_mock.mock_calls) == pre_calls + 1
|
||||
method_mock.assert_called_with(command=command)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("entity_id", "attr"),
|
||||
[
|
||||
("button.pve1_start_all", "startall"),
|
||||
("button.pve1_stop_all", "stopall"),
|
||||
],
|
||||
)
|
||||
async def test_node_startall_stopall_buttons(
|
||||
hass: HomeAssistant,
|
||||
mock_proxmox_client: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_id: str,
|
||||
attr: str,
|
||||
) -> None:
|
||||
"""Test pressing a ProxmoxVE node start all / stop all button triggers the correct API call."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
method_mock = getattr(mock_proxmox_client._node_mock, attr).post
|
||||
pre_calls = len(method_mock.mock_calls)
|
||||
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
SERVICE_PRESS,
|
||||
{ATTR_ENTITY_ID: entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
assert len(method_mock.mock_calls) == pre_calls + 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("entity_id", "vmid", "action"),
|
||||
[
|
||||
("button.vm_web_start", 100, "start"),
|
||||
("button.vm_web_stop", 100, "stop"),
|
||||
("button.vm_web_restart", 100, "restart"),
|
||||
("button.vm_web_hibernate", 100, "hibernate"),
|
||||
("button.vm_web_reset", 100, "reset"),
|
||||
],
|
||||
)
|
||||
async def test_vm_buttons(
|
||||
hass: HomeAssistant,
|
||||
mock_proxmox_client: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_id: str,
|
||||
vmid: int,
|
||||
action: str,
|
||||
) -> None:
|
||||
"""Test pressing a ProxmoxVE VM action button triggers the correct API call."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
mock_proxmox_client._node_mock.qemu(vmid)
|
||||
method_mock = getattr(mock_proxmox_client._qemu_mocks[vmid].status, action).post
|
||||
pre_calls = len(method_mock.mock_calls)
|
||||
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
SERVICE_PRESS,
|
||||
{ATTR_ENTITY_ID: entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
assert len(method_mock.mock_calls) == pre_calls + 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("entity_id", "vmid", "action"),
|
||||
[
|
||||
("button.ct_nginx_start", 200, "start"),
|
||||
("button.ct_nginx_stop", 200, "stop"),
|
||||
("button.ct_nginx_restart", 200, "restart"),
|
||||
],
|
||||
)
|
||||
async def test_container_buttons(
|
||||
hass: HomeAssistant,
|
||||
mock_proxmox_client: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_id: str,
|
||||
vmid: int,
|
||||
action: str,
|
||||
) -> None:
|
||||
"""Test pressing a ProxmoxVE container action button triggers the correct API call."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
mock_proxmox_client._node_mock.lxc(vmid)
|
||||
method_mock = getattr(mock_proxmox_client._lxc_mocks[vmid].status, action).post
|
||||
pre_calls = len(method_mock.mock_calls)
|
||||
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
SERVICE_PRESS,
|
||||
{ATTR_ENTITY_ID: entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
assert len(method_mock.mock_calls) == pre_calls + 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("entity_id", "exception"),
|
||||
[
|
||||
("button.pve1_restart", AuthenticationError("auth failed")),
|
||||
("button.pve1_restart", SSLError("ssl error")),
|
||||
("button.pve1_restart", ConnectTimeout("timeout")),
|
||||
("button.pve1_shutdown", ResourceException(500, "error", {})),
|
||||
],
|
||||
)
|
||||
async def test_node_buttons_exceptions(
|
||||
hass: HomeAssistant,
|
||||
mock_proxmox_client: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_id: str,
|
||||
exception: Exception,
|
||||
) -> None:
|
||||
"""Test that ProxmoxVE node button errors are raised as HomeAssistantError."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
mock_proxmox_client._node_mock.status.post.side_effect = exception
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
SERVICE_PRESS,
|
||||
{ATTR_ENTITY_ID: entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("entity_id", "vmid", "action", "exception"),
|
||||
[
|
||||
(
|
||||
"button.vm_web_start",
|
||||
100,
|
||||
"start",
|
||||
AuthenticationError("auth failed"),
|
||||
),
|
||||
(
|
||||
"button.vm_web_start",
|
||||
100,
|
||||
"start",
|
||||
SSLError("ssl error"),
|
||||
),
|
||||
(
|
||||
"button.vm_web_hibernate",
|
||||
100,
|
||||
"hibernate",
|
||||
ConnectTimeout("timeout"),
|
||||
),
|
||||
(
|
||||
"button.vm_web_reset",
|
||||
100,
|
||||
"reset",
|
||||
ResourceException(500, "error", {}),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_vm_buttons_exceptions(
|
||||
hass: HomeAssistant,
|
||||
mock_proxmox_client: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_id: str,
|
||||
vmid: int,
|
||||
action: str,
|
||||
exception: Exception,
|
||||
) -> None:
|
||||
"""Test that ProxmoxVE VM button errors are raised as HomeAssistantError."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
mock_proxmox_client._node_mock.qemu(vmid)
|
||||
getattr(
|
||||
mock_proxmox_client._qemu_mocks[vmid].status, action
|
||||
).post.side_effect = exception
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
SERVICE_PRESS,
|
||||
{ATTR_ENTITY_ID: entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("entity_id", "vmid", "action", "exception"),
|
||||
[
|
||||
(
|
||||
"button.ct_nginx_start",
|
||||
200,
|
||||
"start",
|
||||
AuthenticationError("auth failed"),
|
||||
),
|
||||
(
|
||||
"button.ct_nginx_start",
|
||||
200,
|
||||
"start",
|
||||
SSLError("ssl error"),
|
||||
),
|
||||
(
|
||||
"button.ct_nginx_restart",
|
||||
200,
|
||||
"restart",
|
||||
ConnectTimeout("timeout"),
|
||||
),
|
||||
(
|
||||
"button.ct_nginx_stop",
|
||||
200,
|
||||
"stop",
|
||||
ResourceException(500, "error", {}),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_container_buttons_exceptions(
|
||||
hass: HomeAssistant,
|
||||
mock_proxmox_client: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_id: str,
|
||||
vmid: int,
|
||||
action: str,
|
||||
exception: Exception,
|
||||
) -> None:
|
||||
"""Test that ProxmoxVE container button errors are raised as HomeAssistantError."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
mock_proxmox_client._node_mock.lxc(vmid)
|
||||
getattr(
|
||||
mock_proxmox_client._lxc_mocks[vmid].status, action
|
||||
).post.side_effect = exception
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
SERVICE_PRESS,
|
||||
{ATTR_ENTITY_ID: entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
Reference in New Issue
Block a user