Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00
Code Activity

Add Config Flow for ProxmoxVE (#142432)

Browse Source 
Co-authored-by: Robert Resch <robert@resch.dev>
Co-authored-by: Abílio Costa <abmantis@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Erwin Douna
2026-01-17 12:59:58 +01:00
committed by GitHub
parent 171013c0d0
commit c993cd9bee
20 changed files with 1390 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
CODEOWNERS generated
View File

@@ -1273,7 +1273,8 @@ build.json @home-assistant/supervisor
/tests/components/prosegur/ @dgomes
/homeassistant/components/proximity/ @mib1185
/tests/components/proximity/ @mib1185
/homeassistant/components/proxmoxve/ @jhollowe @Corbeno
/homeassistant/components/proxmoxve/ @jhollowe @Corbeno @erwindouna
/tests/components/proxmoxve/ @jhollowe @Corbeno @erwindouna
/homeassistant/components/ps4/ @ktnrg45
/tests/components/ps4/ @ktnrg45
/homeassistant/components/pterodactyl/ @elmurato

269
homeassistant/components/proxmoxve/__init__.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from proxmoxer import AuthenticationError, ProxmoxAPI
@@ -10,6 +11,7 @@ import requests.exceptions
from requests.exceptions import ConnectTimeout, SSLError
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
@@ -18,26 +20,29 @@ from homeassistant.const import (
CONF_VERIFY_SSL,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, issue_registry as ir
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .common import ProxmoxClient, call_api_container_vm, parse_api_container_vm
from .common import (
ProxmoxClient,
ResourceException,
call_api_container_vm,
parse_api_container_vm,
)
from .const import (
_LOGGER,
CONF_CONTAINERS,
CONF_NODE,
CONF_NODES,
CONF_REALM,
CONF_VMS,
COORDINATORS,
DEFAULT_PORT,
DEFAULT_REALM,
DEFAULT_VERIFY_SSL,
DOMAIN,
PROXMOX_CLIENTS,
TYPE_CONTAINER,
TYPE_VM,
UPDATE_INTERVAL,
@@ -45,6 +50,10 @@ from .const import (
PLATFORMS = [Platform.BINARY_SENSOR]
type ProxmoxConfigEntry = ConfigEntry[
dict[str, dict[str, dict[int, DataUpdateCoordinator[dict[str, Any] | None]]]]
]
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
@@ -84,109 +93,154 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the platform."""
hass.data.setdefault(DOMAIN, {})
"""Import the Proxmox configuration from YAML."""
if DOMAIN not in config:
return True
def build_client() -> ProxmoxAPI:
"""Build the Proxmox client connection."""
hass.data[PROXMOX_CLIENTS] = {}
for entry in config[DOMAIN]:
host = entry[CONF_HOST]
port = entry[CONF_PORT]
user = entry[CONF_USERNAME]
realm = entry[CONF_REALM]
password = entry[CONF_PASSWORD]
verify_ssl = entry[CONF_VERIFY_SSL]
hass.data[PROXMOX_CLIENTS][host] = None
try:
# Construct an API client with the given data for the given host
proxmox_client = ProxmoxClient(
host, port, user, realm, password, verify_ssl
)
proxmox_client.build_client()
except AuthenticationError:
_LOGGER.warning(
"Invalid credentials for proxmox instance %s:%d", host, port
)
continue
except SSLError:
_LOGGER.error(
(
"Unable to verify proxmox server SSL. "
'Try using "verify_ssl: false" for proxmox instance %s:%d'
),
host,
port,
)
continue
except ConnectTimeout:
_LOGGER.warning("Connection to host %s timed out during setup", host)
continue
except requests.exceptions.ConnectionError:
_LOGGER.warning("Host %s is not reachable", host)
continue
hass.data[PROXMOX_CLIENTS][host] = proxmox_client
await hass.async_add_executor_job(build_client)
coordinators: dict[
str, dict[str, dict[int, DataUpdateCoordinator[dict[str, Any] | None]]]
] = {}
hass.data[DOMAIN][COORDINATORS] = coordinators
# Create a coordinator for each vm/container
for host_config in config[DOMAIN]:
host_name = host_config["host"]
coordinators[host_name] = {}
proxmox_client = hass.data[PROXMOX_CLIENTS][host_name]
# Skip invalid hosts
if proxmox_client is None:
continue
proxmox = proxmox_client.get_api_client()
for node_config in host_config["nodes"]:
node_name = node_config["node"]
node_coordinators = coordinators[host_name][node_name] = {}
for vm_id in node_config["vms"]:
coordinator = create_coordinator_container_vm(
hass, proxmox, host_name, node_name, vm_id, TYPE_VM
)
# Fetch initial data
await coordinator.async_refresh()
node_coordinators[vm_id] = coordinator
for container_id in node_config["containers"]:
coordinator = create_coordinator_container_vm(
hass, proxmox, host_name, node_name, container_id, TYPE_CONTAINER
)
# Fetch initial data
await coordinator.async_refresh()
node_coordinators[container_id] = coordinator
for component in PLATFORMS:
await hass.async_create_task(
async_load_platform(hass, component, DOMAIN, {"config": config}, config)
)
hass.async_create_task(_async_setup(hass, config))
return True
def create_coordinator_container_vm(
async def _async_setup(hass: HomeAssistant, config: ConfigType) -> None:
for entry_config in config[DOMAIN]:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=entry_config,
)
if (
result.get("type") is FlowResultType.ABORT
and result.get("reason") != "already_configured"
):
ir.async_create_issue(
hass,
DOMAIN,
f"deprecated_yaml_import_issue_{result.get('reason')}",
breaks_in_ha_version="2026.8.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.WARNING,
translation_key=f"deprecated_yaml_import_issue_{result.get('reason')}",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Proxmox VE",
},
)
return
ir.async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
"deprecated_yaml",
breaks_in_ha_version="2026.8.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Proxmox VE",
},
)
async def async_setup_entry(hass: HomeAssistant, entry: ProxmoxConfigEntry) -> bool:
"""Set up a ProxmoxVE instance from a config entry."""
def build_client() -> ProxmoxClient:
"""Build and return the Proxmox client connection."""
host = entry.data[CONF_HOST]
port = entry.data[CONF_PORT]
user = entry.data[CONF_USERNAME]
realm = entry.data[CONF_REALM]
password = entry.data[CONF_PASSWORD]
verify_ssl = entry.data[CONF_VERIFY_SSL]
try:
client = ProxmoxClient(host, port, user, realm, password, verify_ssl)
client.build_client()
except AuthenticationError as ex:
raise ConfigEntryAuthFailed("Invalid credentials") from ex
except SSLError as ex:
raise ConfigEntryAuthFailed(
f"Unable to verify proxmox server SSL. Try using 'verify_ssl: false' for proxmox instance {host}:{port}"
) from ex
except ConnectTimeout as ex:
raise ConfigEntryNotReady("Connection timed out") from ex
except requests.exceptions.ConnectionError as ex:
raise ConfigEntryNotReady(f"Host {host} is not reachable: {ex}") from ex
else:
return client
proxmox_client = await hass.async_add_executor_job(build_client)
coordinators: dict[
str, dict[str, dict[int, DataUpdateCoordinator[dict[str, Any] | None]]]
] = {}
entry.runtime_data = coordinators
host_name = entry.data[CONF_HOST]
coordinators[host_name] = {}
proxmox: ProxmoxAPI = proxmox_client.get_api_client()
for node_config in entry.data[CONF_NODES]:
node_name = node_config[CONF_NODE]
node_coordinators = coordinators[host_name][node_name] = {}
try:
vms, containers = await hass.async_add_executor_job(
_get_vms_containers, proxmox, node_config
)
except (ResourceException, requests.exceptions.ConnectionError) as err:
LOGGER.error("Unable to get vms/containers for node %s: %s", node_name, err)
continue
for vm in vms:
coordinator = _create_coordinator_container_vm(
hass, entry, proxmox, host_name, node_name, vm["vmid"], TYPE_VM
)
await coordinator.async_config_entry_first_refresh()
node_coordinators[vm["vmid"]] = coordinator
for container in containers:
coordinator = _create_coordinator_container_vm(
hass,
entry,
proxmox,
host_name,
node_name,
container["vmid"],
TYPE_CONTAINER,
)
await coordinator.async_config_entry_first_refresh()
node_coordinators[container["vmid"]] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
def _get_vms_containers(
proxmox: ProxmoxAPI,
node_config: dict[str, Any],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""Get vms and containers for a node."""
vms = proxmox.nodes(node_config[CONF_NODE]).qemu.get()
containers = proxmox.nodes(node_config[CONF_NODE]).lxc.get()
assert vms is not None and containers is not None
return vms, containers
def _create_coordinator_container_vm(
hass: HomeAssistant,
entry: ProxmoxConfigEntry,
proxmox: ProxmoxAPI,
host_name: str,
node_name: str,
@@ -205,7 +259,7 @@ def create_coordinator_container_vm(
vm_status = await hass.async_add_executor_job(poll_api)
if vm_status is None:
_LOGGER.warning(
LOGGER.warning(
"Vm/Container %s unable to be found in node %s", vm_id, node_name
)
return None
@@ -214,9 +268,14 @@ def create_coordinator_container_vm(
return DataUpdateCoordinator(
hass,
_LOGGER,
config_entry=None,
LOGGER,
config_entry=entry,
name=f"proxmox_coordinator_{host_name}_{node_name}_{vm_id}",
update_method=async_update_data,
update_interval=timedelta(seconds=UPDATE_INTERVAL),
)
async def async_unload_entry(hass: HomeAssistant, entry: ProxmoxConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

53
homeassistant/components/proxmoxve/binary_sensor.py
View File

@@ -2,55 +2,48 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import COORDINATORS, DOMAIN, PROXMOX_CLIENTS
from . import ProxmoxConfigEntry
from .const import CONF_CONTAINERS, CONF_NODE, CONF_NODES, CONF_VMS
from .entity import ProxmoxEntity
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
entry: ProxmoxConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up binary sensors."""
if discovery_info is None:
return
sensors = []
for host_config in discovery_info["config"][DOMAIN]:
host_name = host_config["host"]
host_name_coordinators = hass.data[DOMAIN][COORDINATORS][host_name]
host_name = entry.data[CONF_HOST]
host_name_coordinators = entry.runtime_data[host_name]
if hass.data[PROXMOX_CLIENTS][host_name] is None:
continue
for node_config in entry.data[CONF_NODES]:
node_name = node_config[CONF_NODE]
for node_config in host_config["nodes"]:
node_name = node_config["node"]
for dev_id in node_config[CONF_VMS] + node_config[CONF_CONTAINERS]:
coordinator = host_name_coordinators[node_name][dev_id]
for dev_id in node_config["vms"] + node_config["containers"]:
coordinator = host_name_coordinators[node_name][dev_id]
if TYPE_CHECKING:
assert coordinator.data is not None
name = coordinator.data["name"]
sensor = create_binary_sensor(
coordinator, host_name, node_name, dev_id, name
)
sensors.append(sensor)
# unfound case
if (coordinator_data := coordinator.data) is None:
continue
name = coordinator_data["name"]
sensor = create_binary_sensor(
coordinator, host_name, node_name, dev_id, name
)
sensors.append(sensor)
add_entities(sensors)
async_add_entities(sensors)
def create_binary_sensor(

175
homeassistant/components/proxmoxve/config_flow.py Normal file
View File

@@ -0,0 +1,175 @@
"""Config flow for Proxmox VE integration."""
from __future__ import annotations
import logging
from typing import Any
from proxmoxer import AuthenticationError, ProxmoxAPI
import requests
from requests.exceptions import ConnectTimeout, SSLError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from .common import ResourceException
from .const import (
CONF_CONTAINERS,
CONF_NODE,
CONF_NODES,
CONF_REALM,
CONF_VMS,
DEFAULT_PORT,
DEFAULT_REALM,
DEFAULT_VERIFY_SSL,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_REALM, default=DEFAULT_REALM): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_VERIFY_SSL, default=DEFAULT_VERIFY_SSL): cv.boolean,
}
)
def _sanitize_userid(data: dict[str, Any]) -> str:
"""Sanitize the user ID."""
return (
data[CONF_USERNAME]
if "@" in data[CONF_USERNAME]
else f"{data[CONF_USERNAME]}@{data[CONF_REALM]}"
)
def _get_nodes_data(data: dict[str, Any]) -> list[dict[str, Any]]:
"""Validate the user input and fetch data (sync, for executor)."""
try:
client = ProxmoxAPI(
data[CONF_HOST],
port=data[CONF_PORT],
user=_sanitize_userid(data),
password=data[CONF_PASSWORD],
verify_ssl=data.get(CONF_VERIFY_SSL, DEFAULT_VERIFY_SSL),
)
nodes = client.nodes.get()
except AuthenticationError as err:
raise ProxmoxAuthenticationError from err
except SSLError as err:
raise ProxmoxSSLError from err
except ConnectTimeout as err:
raise ProxmoxConnectTimeout from err
except (ResourceException, requests.exceptions.ConnectionError) as err:
raise ProxmoxNoNodesFound from err
_LOGGER.debug("Proxmox nodes: %s", nodes)
nodes_data: list[dict[str, Any]] = []
for node in nodes:
try:
vms = client.nodes(node["node"]).qemu.get()
containers = client.nodes(node["node"]).lxc.get()
except (ResourceException, requests.exceptions.ConnectionError) as err:
raise ProxmoxNoNodesFound from err
nodes_data.append(
{
CONF_NODE: node["node"],
CONF_VMS: [vm["vmid"] for vm in vms],
CONF_CONTAINERS: [container["vmid"] for container in containers],
}
)
_LOGGER.debug("Nodes with data: %s", nodes_data)
return nodes_data
class ProxmoxveConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Proxmox VE."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
proxmox_nodes: list[dict[str, Any]] = []
if user_input is not None:
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
try:
proxmox_nodes = await self.hass.async_add_executor_job(
_get_nodes_data, user_input
)
except ProxmoxConnectTimeout:
errors["base"] = "connect_timeout"
except ProxmoxAuthenticationError:
errors["base"] = "invalid_auth"
except ProxmoxSSLError:
errors["base"] = "ssl_error"
except ProxmoxNoNodesFound:
errors["base"] = "no_nodes_found"
if not errors:
return self.async_create_entry(
title=user_input[CONF_HOST],
data={**user_input, CONF_NODES: proxmox_nodes},
)
return self.async_show_form(
step_id="user",
data_schema=CONFIG_SCHEMA,
errors=errors,
)
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
"""Handle a flow initiated by configuration file."""
self._async_abort_entries_match({CONF_HOST: import_data[CONF_HOST]})
try:
proxmox_nodes = await self.hass.async_add_executor_job(
_get_nodes_data, import_data
)
except ProxmoxConnectTimeout:
return self.async_abort(reason="connect_timeout")
except ProxmoxAuthenticationError:
return self.async_abort(reason="invalid_auth")
except ProxmoxSSLError:
return self.async_abort(reason="ssl_error")
except ProxmoxNoNodesFound:
return self.async_abort(reason="no_nodes_found")
return self.async_create_entry(
title=import_data[CONF_HOST],
data={**import_data, CONF_NODES: proxmox_nodes},
)
class ProxmoxNoNodesFound(HomeAssistantError):
"""Error to indicate no nodes found."""
class ProxmoxConnectTimeout(HomeAssistantError):
"""Error to indicate a connection timeout."""
class ProxmoxSSLError(HomeAssistantError):
"""Error to indicate an SSL error."""
class ProxmoxAuthenticationError(HomeAssistantError):
"""Error to indicate an authentication error."""

6
homeassistant/components/proxmoxve/const.py
View File

@@ -1,16 +1,12 @@
"""Constants for ProxmoxVE."""
import logging
DOMAIN = "proxmoxve"
PROXMOX_CLIENTS = "proxmox_clients"
CONF_REALM = "realm"
CONF_NODE = "node"
CONF_NODES = "nodes"
CONF_VMS = "vms"
CONF_CONTAINERS = "containers"
COORDINATORS = "coordinators"
DEFAULT_PORT = 8006
DEFAULT_REALM = "pam"
@@ -18,5 +14,3 @@ DEFAULT_VERIFY_SSL = True
TYPE_VM = 0
TYPE_CONTAINER = 1
UPDATE_INTERVAL = 60
_LOGGER = logging.getLogger(__package__)

3
homeassistant/components/proxmoxve/manifest.json
View File

@@ -1,7 +1,8 @@
{
"domain": "proxmoxve",
"name": "Proxmox VE",
"codeowners": ["@jhollowe", "@Corbeno"],
"codeowners": ["@jhollowe", "@Corbeno", "@erwindouna"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/proxmoxve",
"iot_class": "local_polling",
"loggers": ["proxmoxer"],

46
homeassistant/components/proxmoxve/strings.json Normal file
View File

@@ -0,0 +1,46 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "Cannot connect to Proxmox VE server",
"connect_timeout": "[%key:common::config_flow::error::timeout_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"no_nodes_found": "No active nodes found",
"ssl_error": "SSL check failed. Check the SSL settings"
},
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]",
"port": "[%key:common::config_flow::data::port%]",
"realm": "Realm",
"username": "[%key:common::config_flow::data::username%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
},
"description": "Enter your Proxmox VE server details to set up the integration.",
"title": "Connect to Proxmox VE"
}
}
},
"issues": {
"deprecated_yaml_import_issue_connect_timeout": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, a connection timeout occurred. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"deprecated_yaml_import_issue_invalid_auth": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, invalid authentication details were found. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "[%key:component::proxmoxve::issues::deprecated_yaml_import_issue_connect_timeout::title%]"
},
"deprecated_yaml_import_issue_no_nodes_found": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, no active nodes were found on the Proxmox VE server. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "[%key:component::proxmoxve::issues::deprecated_yaml_import_issue_connect_timeout::title%]"
},
"deprecated_yaml_import_issue_ssl_error": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an SSL error occurred. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "[%key:component::proxmoxve::issues::deprecated_yaml_import_issue_connect_timeout::title%]"
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@@ -537,6 +537,7 @@ FLOWS = {
"prosegur",
"prowl",
"proximity",
"proxmoxve",
"prusalink",
"ps4",
"pterodactyl",

2
homeassistant/generated/integrations.json
View File

@@ -5246,7 +5246,7 @@
"proxmoxve": {
"name": "Proxmox VE",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "local_polling"
},
"proxy": {

3
requirements_test_all.txt generated
View File

@@ -1522,6 +1522,9 @@ prometheus-client==0.21.0
# homeassistant.components.prowl
prowlpy==1.1.1
# homeassistant.components.proxmoxve
proxmoxer==2.0.1
# homeassistant.components.hardware
# homeassistant.components.recorder
# homeassistant.components.systemmonitor

26
tests/components/proxmoxve/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
"""Tests for Proxmox VE integration."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
async def setup_integration(
hass: HomeAssistant,
config_entry: MockConfigEntry,
) -> None:
"""Set up the Proxmox VE integration for testing and enable all entities."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
entity_registry = er.async_get(hass)
for entry in er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
):
if entry.disabled_by is not None:
entity_registry.async_update_entity(entry.entity_id, disabled_by=None)
await hass.async_block_till_done()

142
tests/components/proxmoxve/conftest.py Normal file
View File

@@ -0,0 +1,142 @@
"""Common fixtures for the ProxmoxVE tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.proxmoxve.const import (
CONF_CONTAINERS,
CONF_NODE,
CONF_NODES,
CONF_REALM,
CONF_VMS,
DOMAIN,
)
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from tests.common import (
MockConfigEntry,
load_json_array_fixture,
load_json_object_fixture,
)
MOCK_TEST_CONFIG = {
CONF_HOST: "127.0.0.1",
CONF_PORT: 8006,
CONF_REALM: "pam",
CONF_USERNAME: "test_user@pam",
CONF_PASSWORD: "test_password",
CONF_VERIFY_SSL: True,
CONF_NODES: [
{
CONF_NODE: "pve1",
CONF_VMS: [100, 101],
CONF_CONTAINERS: [200, 201],
},
{
CONF_NODE: "pve2",
CONF_VMS: [100, 101],
CONF_CONTAINERS: [200, 201],
},
],
}
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.proxmoxve.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_proxmox_client():
"""Mock Proxmox client with dynamic exception injection support."""
with (
patch(
"homeassistant.components.proxmoxve.ProxmoxAPI", autospec=True
) as mock_api,
patch(
"homeassistant.components.proxmoxve.common.ProxmoxAPI", autospec=True
) as mock_api_common,
patch(
"homeassistant.components.proxmoxve.config_flow.ProxmoxAPI"
) as mock_api_cf,
):
mock_instance = MagicMock()
mock_api.return_value = mock_instance
mock_api_common.return_value = mock_instance
mock_api_cf.return_value = mock_instance
mock_instance.access.ticket.post.return_value = load_json_object_fixture(
"access_ticket.json", DOMAIN
)
# Make a separate mock for the qemu and lxc endpoints
node_mock = MagicMock()
qemu_list = load_json_array_fixture("nodes/qemu.json", DOMAIN)
lxc_list = load_json_array_fixture("nodes/lxc.json", DOMAIN)
node_mock.qemu.get.return_value = qemu_list
node_mock.lxc.get.return_value = lxc_list
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
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
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
node_mock.qemu.side_effect = _qemu_resource
node_mock.lxc.side_effect = _lxc_resource
nodes_mock = MagicMock()
nodes_mock.get.return_value = load_json_array_fixture(
"nodes/nodes.json", DOMAIN
)
nodes_mock.__getitem__.side_effect = lambda key: node_mock
nodes_mock.return_value = node_mock
mock_instance.nodes = nodes_mock
mock_instance._node_mock = node_mock
mock_instance._nodes_mock = nodes_mock
yield mock_instance
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock a config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="ProxmoxVE test",
data=MOCK_TEST_CONFIG,
)

5
tests/components/proxmoxve/fixtures/access_ticket.json Normal file
View File

@@ -0,0 +1,5 @@
{
"username": "test_user@pam",
"ticket": "test_ticket",
"CSRFPreventionToken": "test_token"
}

18
tests/components/proxmoxve/fixtures/nodes/lxc.json Normal file
View File

@@ -0,0 +1,18 @@
[
{
"vmid": 200,
"name": "ct-nginx",
"status": "running",
"maxmem": 1073741824,
"mem": 536870912,
"cpu": 0.05,
"maxdisk": 21474836480,
"disk": 1125899906,
"uptime": 43200
},
{
"vmid": 201,
"name": "ct-backup",
"status": "stopped"
}
]

32
tests/components/proxmoxve/fixtures/nodes/nodes.json Normal file
View File

@@ -0,0 +1,32 @@
[
{
"id": "node/pve1",
"node": "pve1",
"status": "online",
"level": "",
"type": "node",
"maxmem": 34359738368,
"mem": 12884901888,
"maxcpu": 8,
"cpu": 0.12,
"uptime": 86400,
"maxdisk": 500000000000,
"disk": 100000000000,
"ssl_fingerprint": "5C:D2:AB:...:D9"
},
{
"id": "node/pve2",
"node": "pve2",
"status": "online",
"level": "",
"type": "node",
"maxmem": 34359738368,
"mem": 16106127360,
"maxcpu": 8,
"cpu": 0.25,
"uptime": 72000,
"maxdisk": 500000000000,
"disk": 120000000000,
"ssl_fingerprint": "7A:E1:DF:...:AC"
}
]

18
tests/components/proxmoxve/fixtures/nodes/qemu.json Normal file
View File

@@ -0,0 +1,18 @@
[
{
"vmid": 100,
"name": "vm-web",
"status": "running",
"maxmem": 2147483648,
"mem": 1073741824,
"cpu": 0.15,
"maxdisk": 34359738368,
"disk": 1234567890,
"uptime": 86400
},
{
"vmid": 101,
"name": "vm-db",
"status": "stopped"
}
]

409
tests/components/proxmoxve/snapshots/test_binary_sensor.ambr Normal file
View File

@@ -0,0 +1,409 @@
# serializer version: 1
# name: test_all_entities[binary_sensor.pve1_ct_backup-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.pve1_ct_backup',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'pve1_ct-backup',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': '',
'original_name': 'pve1_ct-backup',
'platform': 'proxmoxve',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'proxmox_pve1_201_running',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.pve1_ct_backup-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'pve1_ct-backup',
'icon': '',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pve1_ct_backup',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[binary_sensor.pve1_ct_nginx-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.pve1_ct_nginx',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'pve1_ct-nginx',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': '',
'original_name': 'pve1_ct-nginx',
'platform': 'proxmoxve',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'proxmox_pve1_200_running',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.pve1_ct_nginx-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'pve1_ct-nginx',
'icon': '',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pve1_ct_nginx',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[binary_sensor.pve1_vm_db-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.pve1_vm_db',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'pve1_vm-db',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': '',
'original_name': 'pve1_vm-db',
'platform': 'proxmoxve',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'proxmox_pve1_101_running',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.pve1_vm_db-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'pve1_vm-db',
'icon': '',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pve1_vm_db',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[binary_sensor.pve1_vm_web-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.pve1_vm_web',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'pve1_vm-web',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': '',
'original_name': 'pve1_vm-web',
'platform': 'proxmoxve',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'proxmox_pve1_100_running',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.pve1_vm_web-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'pve1_vm-web',
'icon': '',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pve1_vm_web',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[binary_sensor.pve2_ct_backup-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.pve2_ct_backup',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'pve2_ct-backup',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': '',
'original_name': 'pve2_ct-backup',
'platform': 'proxmoxve',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'proxmox_pve2_201_running',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.pve2_ct_backup-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'pve2_ct-backup',
'icon': '',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pve2_ct_backup',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[binary_sensor.pve2_ct_nginx-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.pve2_ct_nginx',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'pve2_ct-nginx',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': '',
'original_name': 'pve2_ct-nginx',
'platform': 'proxmoxve',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'proxmox_pve2_200_running',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.pve2_ct_nginx-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'pve2_ct-nginx',
'icon': '',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pve2_ct_nginx',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[binary_sensor.pve2_vm_db-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.pve2_vm_db',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'pve2_vm-db',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': '',
'original_name': 'pve2_vm-db',
'platform': 'proxmoxve',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'proxmox_pve2_101_running',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.pve2_vm_db-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'pve2_vm-db',
'icon': '',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pve2_vm_db',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[binary_sensor.pve2_vm_web-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.pve2_vm_web',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'pve2_vm-web',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': '',
'original_name': 'pve2_vm-web',
'platform': 'proxmoxve',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'proxmox_pve2_100_running',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.pve2_vm_web-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'pve2_vm-web',
'icon': '',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pve2_vm_web',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

33
tests/components/proxmoxve/test_binary_sensor.py Normal file
View File

@@ -0,0 +1,33 @@
"""Test the Proxmox VE binary sensor platform."""
from unittest.mock import MagicMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
import homeassistant.helpers.entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.usefixture("entity_registry_enabled_by_default")
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_proxmox_client: MagicMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all entities."""
with patch(
"homeassistant.components.proxmoxve.PLATFORMS",
[Platform.BINARY_SENSOR],
):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(
hass, entity_registry, snapshot, mock_config_entry.entry_id
)

230
tests/components/proxmoxve/test_config_flow.py Normal file
View File

@@ -0,0 +1,230 @@
"""Test the config flow for Proxmox VE."""
from __future__ import annotations
from unittest.mock import MagicMock
from proxmoxer import AuthenticationError
import pytest
from requests.exceptions import ConnectTimeout, SSLError
from homeassistant.components.proxmoxve import CONF_HOST, CONF_REALM
from homeassistant.components.proxmoxve.common import ResourceException
from homeassistant.components.proxmoxve.const import CONF_NODES, DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER, ConfigEntryState
from homeassistant.const import CONF_PASSWORD, CONF_PORT, CONF_USERNAME, CONF_VERIFY_SSL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import MOCK_TEST_CONFIG
from tests.common import MockConfigEntry
MOCK_USER_STEP = {
CONF_HOST: "127.0.0.1",
CONF_USERNAME: "test_user@pam",
CONF_PASSWORD: "test_password",
CONF_VERIFY_SSL: True,
CONF_PORT: 8006,
CONF_REALM: "pam",
}
MOCK_USER_SETUP = {CONF_NODES: ["pve1"]}
MOCK_USER_FINAL = {
**MOCK_USER_STEP,
**MOCK_USER_SETUP,
}
async def test_form(
hass: HomeAssistant,
mock_proxmox_client: MagicMock,
) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=MOCK_USER_STEP
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "127.0.0.1"
assert result["data"] == MOCK_TEST_CONFIG
@pytest.mark.parametrize(
("exception", "reason"),
[
(
AuthenticationError("Invalid credentials"),
"invalid_auth",
),
(
SSLError("SSL handshake failed"),
"ssl_error",
),
(
ConnectTimeout("Connection timed out"),
"connect_timeout",
),
],
)
async def test_form_exceptions(
hass: HomeAssistant,
mock_proxmox_client: MagicMock,
exception: Exception,
reason: str,
) -> None:
"""Test we handle all exceptions."""
mock_proxmox_client.nodes.get.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=MOCK_USER_STEP,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": reason}
mock_proxmox_client.nodes.get.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=MOCK_USER_STEP
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_form_no_nodes_exception(
hass: HomeAssistant,
mock_proxmox_client: MagicMock,
) -> None:
"""Test we handle no nodes found exception."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
mock_proxmox_client.nodes.get.side_effect = ResourceException(
"404", "status_message", "content"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=MOCK_USER_STEP
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "no_nodes_found"}
mock_proxmox_client.nodes.get.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=MOCK_USER_STEP
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_duplicate_entry(
hass: HomeAssistant,
mock_proxmox_client: MagicMock,
mock_setup_entry: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test we handle duplicate entries."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=MOCK_USER_STEP
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_import_flow(
hass: HomeAssistant,
mock_setup_entry: MagicMock,
mock_proxmox_client: MagicMock,
) -> None:
"""Test importing from YAML creates a config entry and sets it up."""
MOCK_IMPORT_CONFIG = {
DOMAIN: {
**MOCK_USER_STEP,
**MOCK_USER_SETUP,
}
}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=MOCK_IMPORT_CONFIG[DOMAIN]
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "127.0.0.1"
assert result["data"][CONF_HOST] == "127.0.0.1"
assert len(mock_setup_entry.mock_calls) == 1
assert result["result"].state is ConfigEntryState.LOADED
@pytest.mark.parametrize(
("exception", "reason"),
[
(
AuthenticationError("Invalid credentials"),
"invalid_auth",
),
(
SSLError("SSL handshake failed"),
"ssl_error",
),
(
ConnectTimeout("Connection timed out"),
"connect_timeout",
),
(
ResourceException("404", "status_message", "content"),
"no_nodes_found",
),
],
)
async def test_import_flow_exceptions(
hass: HomeAssistant,
mock_setup_entry: MagicMock,
mock_proxmox_client: MagicMock,
exception: Exception,
reason: str,
) -> None:
"""Test importing from YAML creates a config entry and sets it up."""
MOCK_IMPORT_CONFIG = {
DOMAIN: {
**MOCK_USER_STEP,
**MOCK_USER_SETUP,
}
}
mock_proxmox_client.nodes.get.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=MOCK_IMPORT_CONFIG[DOMAIN]
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == reason
assert len(mock_setup_entry.mock_calls) == 0
assert len(hass.config_entries.async_entries(DOMAIN)) == 0

60
tests/components/proxmoxve/test_init.py Normal file
View File

@@ -0,0 +1,60 @@
"""Tests for the Proxmox VE integration initialization."""
from unittest.mock import MagicMock
from homeassistant.components.proxmoxve.const import (
CONF_CONTAINERS,
CONF_NODE,
CONF_NODES,
CONF_REALM,
CONF_VMS,
DOMAIN,
)
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
import homeassistant.helpers.issue_registry as ir
from homeassistant.setup import async_setup_component
async def test_config_import(
hass: HomeAssistant,
mock_proxmox_client: MagicMock,
mock_setup_entry: MagicMock,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test sensor initialization."""
await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: [
{
CONF_HOST: "127.0.0.1",
CONF_PORT: 8006,
CONF_REALM: "pam",
CONF_USERNAME: "test_user@pam",
CONF_PASSWORD: "test_password",
CONF_VERIFY_SSL: True,
CONF_NODES: [
{
CONF_NODE: "pve1",
CONF_VMS: [100, 101],
CONF_CONTAINERS: [200, 201],
},
],
}
]
},
)
await hass.async_block_till_done()
assert len(issue_registry.issues) == 1
assert (HOMEASSISTANT_DOMAIN, "deprecated_yaml") in issue_registry.issues
assert len(hass.config_entries.async_entries(DOMAIN)) == 1