mirror of
https://github.com/home-assistant/core.git
synced 2026-09-26 09:23:17 -04:00
migrate volkszaehler to config flow (#169500)
Co-authored-by: Copilot <copilot@github.com> Co-authored-by: Joostlek <joostlek@outlook.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
StefanSchoof
co-authored by
Copilot
Joostlek
Claude Opus 4.8
Copilot
Joostlek
Claude Opus 4.8
parent
60e4cbebfd
commit
7d47c4a355
@@ -1 +1,80 @@
|
||||
"""The volkszaehler component."""
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from volkszaehler import Volkszaehler
|
||||
from volkszaehler.exceptions import VolkszaehlerApiConnectionError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_UUID, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
from .const import SUBENTRY_TYPE_CHANNEL
|
||||
|
||||
_PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=1)
|
||||
|
||||
|
||||
type VolkszaehlerConfigEntry = ConfigEntry[dict[str, VolkszaehlerData]]
|
||||
|
||||
|
||||
class VolkszaehlerData:
|
||||
"""The class for handling the data retrieval from the Volkszaehler API."""
|
||||
|
||||
def __init__(self, api: Volkszaehler) -> None:
|
||||
"""Initialize the data object."""
|
||||
self.api = api
|
||||
self.available = True
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
||||
async def async_update(self) -> None:
|
||||
"""Get the latest data from the Volkszaehler REST API."""
|
||||
|
||||
try:
|
||||
await self.api.get_data()
|
||||
self.available = True
|
||||
except VolkszaehlerApiConnectionError:
|
||||
_LOGGER.error("Unable to fetch data from the Volkszaehler API")
|
||||
self.available = False
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: VolkszaehlerConfigEntry
|
||||
) -> bool:
|
||||
"""Set up Volkszaehler from a config entry."""
|
||||
runtime_data: dict[str, VolkszaehlerData] = {}
|
||||
|
||||
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_CHANNEL):
|
||||
vz_data = VolkszaehlerData(
|
||||
Volkszaehler(
|
||||
async_get_clientsession(hass),
|
||||
subentry.data[CONF_UUID],
|
||||
host=entry.data[CONF_HOST],
|
||||
port=entry.data[CONF_PORT],
|
||||
)
|
||||
)
|
||||
await vz_data.async_update()
|
||||
if not vz_data.available or vz_data.api.data is None:
|
||||
raise ConfigEntryNotReady(
|
||||
"Unable to fetch initial data from the Volkszaehler API"
|
||||
)
|
||||
|
||||
runtime_data[subentry.subentry_id] = vz_data
|
||||
|
||||
entry.runtime_data = runtime_data
|
||||
await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(
|
||||
hass: HomeAssistant, entry: VolkszaehlerConfigEntry
|
||||
) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
"""Config flow for Volkszaehler integration."""
|
||||
|
||||
import logging
|
||||
from types import MappingProxyType
|
||||
from typing import Any, override
|
||||
|
||||
import probatio
|
||||
from volkszaehler import Volkszaehler
|
||||
from volkszaehler.exceptions import (
|
||||
VolkszaehlerApiConnectionError,
|
||||
VolkszaehlerNoDataAvailable,
|
||||
)
|
||||
|
||||
from homeassistant.config_entries import (
|
||||
ConfigEntry,
|
||||
ConfigFlow,
|
||||
ConfigFlowResult,
|
||||
ConfigSubentry,
|
||||
ConfigSubentryFlow,
|
||||
SubentryFlowResult,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, CONF_UUID
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
|
||||
from .const import DEFAULT_PORT, DOMAIN, SUBENTRY_TYPE_CHANNEL
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
STEP_USER_DATA_SCHEMA = probatio.Schema(
|
||||
{
|
||||
probatio.Required(CONF_HOST): cv.string,
|
||||
probatio.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
|
||||
probatio.Required(CONF_UUID): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
STEP_SUBENTRY_DATA_SCHEMA = probatio.Schema({probatio.Required(CONF_UUID): cv.string})
|
||||
|
||||
|
||||
async def _validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None:
|
||||
"""Validate the user input allows us to connect."""
|
||||
api = Volkszaehler(
|
||||
session=async_get_clientsession(hass),
|
||||
uuid=data[CONF_UUID],
|
||||
host=data[CONF_HOST],
|
||||
port=data[CONF_PORT],
|
||||
)
|
||||
await api.get_data()
|
||||
|
||||
|
||||
async def _async_validate_input_errors(
|
||||
hass: HomeAssistant, data: dict[str, Any]
|
||||
) -> str | None:
|
||||
"""Validate input and return a config flow error key if validation fails."""
|
||||
try:
|
||||
await _validate_input(hass, data)
|
||||
except VolkszaehlerApiConnectionError:
|
||||
return "cannot_connect"
|
||||
except VolkszaehlerNoDataAvailable:
|
||||
return "no_data"
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
return "unknown"
|
||||
return None
|
||||
|
||||
|
||||
def _is_uuid_configured(hass: HomeAssistant, uuid: str) -> bool:
|
||||
"""Return if a Volkszaehler channel UUID is already configured."""
|
||||
for entry in hass.config_entries.async_entries(DOMAIN):
|
||||
if any(
|
||||
subentry.unique_id == uuid
|
||||
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_CHANNEL)
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class VolkszaehlerConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Volkszaehler."""
|
||||
|
||||
@classmethod
|
||||
@callback
|
||||
@override
|
||||
def async_get_supported_subentry_types(
|
||||
cls, config_entry: ConfigEntry
|
||||
) -> dict[str, type[ConfigSubentryFlow]]:
|
||||
"""Return subentries supported by this handler."""
|
||||
return {SUBENTRY_TYPE_CHANNEL: VolkszaehlerSubentryFlow}
|
||||
|
||||
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
|
||||
"""Set the config entry up from yaml."""
|
||||
if error := await _async_validate_input_errors(self.hass, import_data):
|
||||
return self.async_abort(reason=error)
|
||||
|
||||
uuid = import_data[CONF_UUID]
|
||||
if _is_uuid_configured(self.hass, uuid):
|
||||
return self.async_abort(reason="already_configured")
|
||||
|
||||
channel_subentry = ConfigSubentry(
|
||||
subentry_type=SUBENTRY_TYPE_CHANNEL,
|
||||
unique_id=uuid,
|
||||
title=import_data.get(CONF_NAME, uuid),
|
||||
data=MappingProxyType({CONF_UUID: uuid}),
|
||||
)
|
||||
|
||||
for entry in self.hass.config_entries.async_entries(DOMAIN):
|
||||
if (
|
||||
entry.data.get(CONF_HOST) == import_data[CONF_HOST]
|
||||
and entry.data.get(CONF_PORT, DEFAULT_PORT) == import_data[CONF_PORT]
|
||||
):
|
||||
self.hass.config_entries.async_add_subentry(entry, channel_subentry)
|
||||
return self.async_abort(reason="subentry_added")
|
||||
|
||||
return self.async_create_entry(
|
||||
title=import_data[CONF_HOST],
|
||||
data={
|
||||
CONF_HOST: import_data[CONF_HOST],
|
||||
CONF_PORT: import_data[CONF_PORT],
|
||||
},
|
||||
subentries=[channel_subentry.as_dict()],
|
||||
)
|
||||
|
||||
@override
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the initial step."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
self._async_abort_entries_match(
|
||||
{
|
||||
CONF_HOST: user_input[CONF_HOST],
|
||||
CONF_PORT: user_input[CONF_PORT],
|
||||
}
|
||||
)
|
||||
if error := await _async_validate_input_errors(self.hass, user_input):
|
||||
errors["base"] = error
|
||||
else:
|
||||
if _is_uuid_configured(self.hass, user_input[CONF_UUID]):
|
||||
return self.async_abort(reason="already_configured")
|
||||
return self.async_create_entry(
|
||||
title=user_input[CONF_HOST],
|
||||
data={
|
||||
CONF_HOST: user_input[CONF_HOST],
|
||||
CONF_PORT: user_input[CONF_PORT],
|
||||
},
|
||||
subentries=[
|
||||
{
|
||||
"subentry_type": SUBENTRY_TYPE_CHANNEL,
|
||||
"title": user_input[CONF_UUID],
|
||||
"data": {CONF_UUID: user_input[CONF_UUID]},
|
||||
"unique_id": user_input[CONF_UUID],
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=self.add_suggested_values_to_schema(
|
||||
STEP_USER_DATA_SCHEMA, user_input
|
||||
)
|
||||
if user_input
|
||||
else STEP_USER_DATA_SCHEMA,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
|
||||
class VolkszaehlerSubentryFlow(ConfigSubentryFlow):
|
||||
"""Handle Volkszaehler channel subentry flow."""
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> SubentryFlowResult:
|
||||
"""Add a Volkszaehler channel subentry."""
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
if user_input is not None:
|
||||
if _is_uuid_configured(self.hass, user_input[CONF_UUID]):
|
||||
return self.async_abort(reason="already_configured")
|
||||
|
||||
entry = self._get_entry()
|
||||
if error := await _async_validate_input_errors(
|
||||
self.hass,
|
||||
{
|
||||
CONF_HOST: entry.data[CONF_HOST],
|
||||
CONF_PORT: entry.data[CONF_PORT],
|
||||
CONF_UUID: user_input[CONF_UUID],
|
||||
},
|
||||
):
|
||||
errors["base"] = error
|
||||
else:
|
||||
return self.async_create_entry(
|
||||
title=user_input[CONF_UUID],
|
||||
data={CONF_UUID: user_input[CONF_UUID]},
|
||||
unique_id=user_input[CONF_UUID],
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=self.add_suggested_values_to_schema(
|
||||
STEP_SUBENTRY_DATA_SCHEMA, user_input
|
||||
)
|
||||
if user_input
|
||||
else STEP_SUBENTRY_DATA_SCHEMA,
|
||||
errors=errors,
|
||||
)
|
||||
@@ -0,0 +1,7 @@
|
||||
"""Constants for the Volkszaehler integration."""
|
||||
|
||||
DOMAIN = "volkszaehler"
|
||||
SUBENTRY_TYPE_CHANNEL = "channel"
|
||||
DEFAULT_HOST = "localhost"
|
||||
DEFAULT_NAME = "Volkszaehler"
|
||||
DEFAULT_PORT = 80
|
||||
@@ -2,7 +2,9 @@
|
||||
"domain": "volkszaehler",
|
||||
"name": "Volkszaehler",
|
||||
"codeowners": [],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/volkszaehler",
|
||||
"integration_type": "device",
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["volkszaehler"],
|
||||
"quality_scale": "legacy",
|
||||
|
||||
@@ -1,12 +1,8 @@
|
||||
"""Support for consuming values for the Volkszaehler API."""
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import override
|
||||
|
||||
import probatio
|
||||
from volkszaehler import Volkszaehler
|
||||
from volkszaehler.exceptions import VolkszaehlerApiConnectionError
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
|
||||
@@ -14,6 +10,7 @@ from homeassistant.components.sensor import (
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_IMPORT
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_MONITORED_CONDITIONS,
|
||||
@@ -23,21 +20,23 @@ from homeassistant.const import (
|
||||
UnitOfEnergy,
|
||||
UnitOfPower,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import PlatformNotReady
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.helpers import config_validation as cv, issue_registry as ir
|
||||
from homeassistant.helpers.entity_platform import (
|
||||
AddConfigEntryEntitiesCallback,
|
||||
AddEntitiesCallback,
|
||||
)
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_HOST = "localhost"
|
||||
DEFAULT_NAME = "Volkszaehler"
|
||||
DEFAULT_PORT = 80
|
||||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=1)
|
||||
from . import VolkszaehlerConfigEntry, VolkszaehlerData
|
||||
from .const import (
|
||||
DEFAULT_HOST,
|
||||
DEFAULT_NAME,
|
||||
DEFAULT_PORT,
|
||||
DOMAIN,
|
||||
SUBENTRY_TYPE_CHANNEL,
|
||||
)
|
||||
|
||||
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
|
||||
SensorEntityDescription(
|
||||
@@ -91,42 +90,96 @@ async def async_setup_platform(
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the Volkszaehler sensors."""
|
||||
"""Import Volkszaehler sensor YAML config into config flow."""
|
||||
validated = PLATFORM_SCHEMA(config)
|
||||
data = {
|
||||
CONF_HOST: validated[CONF_HOST],
|
||||
CONF_NAME: validated[CONF_NAME],
|
||||
CONF_PORT: validated[CONF_PORT],
|
||||
CONF_UUID: validated[CONF_UUID],
|
||||
}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=data,
|
||||
)
|
||||
if result["type"] is FlowResultType.ABORT and result["reason"] not in (
|
||||
"already_configured",
|
||||
"subentry_added",
|
||||
):
|
||||
ir.async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
f"deprecated_yaml_import_issue_{result['reason']}",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
severity=ir.IssueSeverity.WARNING,
|
||||
translation_key=f"deprecated_yaml_import_issue_{result['reason']}",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": DEFAULT_NAME,
|
||||
},
|
||||
breaks_in_ha_version="2027.3.0",
|
||||
)
|
||||
return
|
||||
|
||||
host: str = config[CONF_HOST]
|
||||
name: str = config[CONF_NAME]
|
||||
port: int = config[CONF_PORT]
|
||||
uuid: str = config[CONF_UUID]
|
||||
conditions: list[str] = config[CONF_MONITORED_CONDITIONS]
|
||||
ir.async_create_issue(
|
||||
hass,
|
||||
HOMEASSISTANT_DOMAIN,
|
||||
f"deprecated_yaml_{DOMAIN}",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
severity=ir.IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": DEFAULT_NAME,
|
||||
},
|
||||
breaks_in_ha_version="2027.3.0",
|
||||
)
|
||||
|
||||
session = async_get_clientsession(hass)
|
||||
vz_api = VolkszaehlerData(Volkszaehler(session, uuid, host=host, port=port))
|
||||
|
||||
await vz_api.async_update()
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: VolkszaehlerConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up Volkszaehler sensors from a config entry."""
|
||||
conditions = SENSOR_KEYS
|
||||
|
||||
if vz_api.api.data is None:
|
||||
raise PlatformNotReady
|
||||
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_CHANNEL):
|
||||
vz_api = entry.runtime_data[subentry.subentry_id]
|
||||
|
||||
entities = [
|
||||
VolkszaehlerSensor(vz_api, name, description)
|
||||
for description in SENSOR_TYPES
|
||||
if description.key in conditions
|
||||
]
|
||||
entities = [
|
||||
VolkszaehlerSensor(
|
||||
vz_api,
|
||||
subentry.title,
|
||||
subentry.data[CONF_UUID],
|
||||
description,
|
||||
)
|
||||
for description in SENSOR_TYPES
|
||||
if description.key in conditions
|
||||
]
|
||||
|
||||
async_add_entities(entities, True)
|
||||
async_add_entities(entities, False, config_subentry_id=subentry.subentry_id)
|
||||
|
||||
|
||||
class VolkszaehlerSensor(SensorEntity):
|
||||
"""Implementation of a Volkszaehler sensor."""
|
||||
|
||||
def __init__(
|
||||
self, vz_api: VolkszaehlerData, name: str, description: SensorEntityDescription
|
||||
self,
|
||||
vz_api: VolkszaehlerData,
|
||||
name: str,
|
||||
uuid: str,
|
||||
description: SensorEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the Volkszaehler sensor."""
|
||||
self.entity_description = description
|
||||
self.vz_api = vz_api
|
||||
|
||||
self._attr_name = f"{name} {description.name}"
|
||||
self._attr_unique_id = f"{uuid}_{description.key}"
|
||||
|
||||
@property
|
||||
@override
|
||||
@@ -142,23 +195,3 @@ class VolkszaehlerSensor(SensorEntity):
|
||||
self._attr_native_value = round(
|
||||
getattr(self.vz_api.api, self.entity_description.key), 2
|
||||
)
|
||||
|
||||
|
||||
class VolkszaehlerData:
|
||||
"""The class for handling the data retrieval from the Volkszaehler API."""
|
||||
|
||||
def __init__(self, api: Volkszaehler) -> None:
|
||||
"""Initialize the data object."""
|
||||
self.api = api
|
||||
self.available = True
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
||||
async def async_update(self) -> None:
|
||||
"""Get the latest data from the Volkszaehler REST API."""
|
||||
|
||||
try:
|
||||
await self.api.get_data()
|
||||
self.available = True
|
||||
except VolkszaehlerApiConnectionError:
|
||||
_LOGGER.error("Unable to fetch data from the Volkszaehler API")
|
||||
self.available = False
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"no_data": "No data available for this channel",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"port": "[%key:common::config_flow::data::port%]",
|
||||
"uuid": "UUID"
|
||||
},
|
||||
"data_description": {
|
||||
"host": "IP address or hostname of the Volkszaehler server.",
|
||||
"port": "Port number of the Volkszaehler server.",
|
||||
"uuid": "UUID of the channel to monitor. In the Volkszaehler Web UI, click the (i) icon next to the channel to find it."
|
||||
},
|
||||
"description": "Enter your Volkszaehler server and first channel details.",
|
||||
"title": "Set up Volkszaehler"
|
||||
}
|
||||
}
|
||||
},
|
||||
"config_subentries": {
|
||||
"channel": {
|
||||
"entry_type": "Channel",
|
||||
"initiate_flow": {
|
||||
"user": "Add Channel"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"uuid": "UUID"
|
||||
},
|
||||
"data_description": {
|
||||
"uuid": "UUID of the channel to monitor. In the Volkszaehler Web UI, click the (i) icon next to the channel to find it."
|
||||
},
|
||||
"description": "Add another Volkszaehler channel.",
|
||||
"title": "Add Volkszaehler channel"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"deprecated_yaml_import_issue_cannot_connect": {
|
||||
"description": "Importing Volkszaehler sensor configuration from YAML failed because the server could not be reached. Remove the YAML configuration and set up the integration through the Home Assistant UI.",
|
||||
"title": "YAML import failed"
|
||||
},
|
||||
"deprecated_yaml_import_issue_no_data": {
|
||||
"description": "Importing Volkszaehler sensor configuration from YAML failed because no data was available for the channel. Remove the YAML configuration and set up the integration through the Home Assistant UI.",
|
||||
"title": "YAML import failed"
|
||||
},
|
||||
"deprecated_yaml_import_issue_unknown": {
|
||||
"description": "Importing Volkszaehler sensor configuration from YAML is deprecated or failed. Remove the YAML configuration and set up the integration through the Home Assistant UI.",
|
||||
"title": "YAML import is deprecated"
|
||||
}
|
||||
}
|
||||
}
|
||||
Generated
+1
@@ -876,6 +876,7 @@ FLOWS = {
|
||||
"vlc_telnet",
|
||||
"vodafone_station",
|
||||
"voip",
|
||||
"volkszaehler",
|
||||
"volumio",
|
||||
"volvo",
|
||||
"wake_on_lan",
|
||||
|
||||
@@ -8088,8 +8088,8 @@
|
||||
},
|
||||
"volkszaehler": {
|
||||
"name": "Volkszaehler",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"integration_type": "device",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"volumio": {
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for the Volkszaehler component."""
|
||||
@@ -0,0 +1,63 @@
|
||||
"""Test fixtures for volkszaehler."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.volkszaehler.const import DOMAIN, SUBENTRY_TYPE_CHANNEL
|
||||
from homeassistant.config_entries import ConfigSubentryData
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_UUID
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Override async_setup_entry."""
|
||||
with patch(
|
||||
"homeassistant.components.volkszaehler.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture(name="mock_api")
|
||||
def mock_client_api() -> Generator[Mock]:
|
||||
"""Set up fake Volkszaehler API responses."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.volkszaehler.Volkszaehler",
|
||||
autospec=True,
|
||||
) as mock_api,
|
||||
patch(
|
||||
"homeassistant.components.volkszaehler.config_flow.Volkszaehler",
|
||||
new=mock_api,
|
||||
),
|
||||
):
|
||||
api = mock_api.return_value
|
||||
api.get_data = AsyncMock(return_value=None)
|
||||
|
||||
yield api
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
"""Fixture for a config entry with one existing channel subentry."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="localhost",
|
||||
data={
|
||||
CONF_HOST: "localhost",
|
||||
CONF_PORT: 80,
|
||||
},
|
||||
subentries_data=[
|
||||
ConfigSubentryData(
|
||||
subentry_type=SUBENTRY_TYPE_CHANNEL,
|
||||
title="existing-uuid",
|
||||
data={CONF_UUID: "existing-uuid"},
|
||||
unique_id="existing-uuid",
|
||||
subentry_id="existing-subentry-id",
|
||||
)
|
||||
],
|
||||
)
|
||||
@@ -0,0 +1,353 @@
|
||||
"""Test config flow for Volkszaehler integration."""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from volkszaehler.exceptions import (
|
||||
VolkszaehlerApiConnectionError,
|
||||
VolkszaehlerNoDataAvailable,
|
||||
)
|
||||
|
||||
from homeassistant.components.volkszaehler.const import DOMAIN, SUBENTRY_TYPE_CHANNEL
|
||||
from homeassistant.config_entries import SOURCE_USER
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_MONITORED_CONDITIONS,
|
||||
CONF_NAME,
|
||||
CONF_PLATFORM,
|
||||
CONF_PORT,
|
||||
CONF_UUID,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_create_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: AsyncMock,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test that the config flow creates an entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_UUID: "test-uuid",
|
||||
CONF_HOST: "localhost",
|
||||
CONF_PORT: 80,
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "localhost"
|
||||
assert result["data"] == {
|
||||
CONF_HOST: "localhost",
|
||||
CONF_PORT: 80,
|
||||
}
|
||||
|
||||
entry = result["result"]
|
||||
assert len(entry.subentries) == 1
|
||||
subentry = next(iter(entry.subentries.values()))
|
||||
assert subentry.subentry_type == SUBENTRY_TYPE_CHANNEL
|
||||
assert subentry.title == "test-uuid"
|
||||
assert subentry.unique_id == "test-uuid"
|
||||
assert subentry.data == {CONF_UUID: "test-uuid"}
|
||||
|
||||
assert mock_api.get_data.call_count == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("side_effect", "expected_error"),
|
||||
[
|
||||
(VolkszaehlerApiConnectionError, "cannot_connect"),
|
||||
(VolkszaehlerNoDataAvailable, "no_data"),
|
||||
(Exception, "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_user_errors(
|
||||
hass: HomeAssistant,
|
||||
mock_api: AsyncMock,
|
||||
side_effect: type[Exception],
|
||||
expected_error: str,
|
||||
) -> None:
|
||||
"""Test error handling in the config flow user step."""
|
||||
user_input = {
|
||||
CONF_UUID: "test-uuid",
|
||||
CONF_HOST: "localhost",
|
||||
CONF_PORT: 80,
|
||||
}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
|
||||
mock_api.get_data.side_effect = side_effect
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"]["base"] == expected_error
|
||||
|
||||
mock_api.get_data.side_effect = None
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "localhost"
|
||||
assert result["data"] == {
|
||||
CONF_HOST: "localhost",
|
||||
CONF_PORT: 80,
|
||||
}
|
||||
|
||||
|
||||
async def test_create_subentry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test that the subentry flow creates an additional channel."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.subentries.async_init(
|
||||
(mock_config_entry.entry_id, SUBENTRY_TYPE_CHANNEL),
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"], {CONF_UUID: "new-uuid"}
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "new-uuid"
|
||||
assert result["data"] == {CONF_UUID: "new-uuid"}
|
||||
assert len(mock_config_entry.subentries) == 2
|
||||
|
||||
assert mock_api.get_data.call_count == 1
|
||||
|
||||
|
||||
async def test_import(hass: HomeAssistant, mock_api: AsyncMock) -> None:
|
||||
"""Test that we can import a config entry."""
|
||||
import_data = {
|
||||
CONF_UUID: "import-uuid",
|
||||
CONF_HOST: "importhost",
|
||||
CONF_NAME: "2.8.0",
|
||||
CONF_PLATFORM: "volkszaehler",
|
||||
CONF_MONITORED_CONDITIONS: ["consumption"],
|
||||
}
|
||||
await async_setup_component(hass, "sensor", {"sensor": import_data})
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 1
|
||||
entry = entries[0]
|
||||
assert entry.data == {
|
||||
CONF_HOST: "importhost",
|
||||
CONF_PORT: 80,
|
||||
}
|
||||
assert entry.title == "importhost"
|
||||
assert len(entry.subentries) == 1
|
||||
subentry = next(iter(entry.subentries.values()))
|
||||
assert subentry.subentry_type is SUBENTRY_TYPE_CHANNEL
|
||||
assert subentry.title == "2.8.0"
|
||||
assert subentry.data == {CONF_UUID: "import-uuid"}
|
||||
|
||||
assert mock_api.get_data.call_count == 2
|
||||
|
||||
|
||||
async def test_import_once(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test that we import a config entry only once."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
import_data = {
|
||||
CONF_UUID: "existing-uuid",
|
||||
CONF_HOST: "localhost",
|
||||
CONF_PORT: 80,
|
||||
CONF_PLATFORM: "volkszaehler",
|
||||
}
|
||||
await async_setup_component(hass, "sensor", {"sensor": import_data})
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 1
|
||||
assert entries[0].entry_id == mock_config_entry.entry_id
|
||||
assert len(entries[0].subentries) == 1
|
||||
|
||||
assert mock_api.get_data.call_count == 1
|
||||
|
||||
|
||||
async def test_import_add_second_subentry_same_host(
|
||||
hass: HomeAssistant, mock_api: AsyncMock
|
||||
) -> None:
|
||||
"""Test that import adds a second channel to the existing entry on same host."""
|
||||
await async_setup_component(
|
||||
hass,
|
||||
"sensor",
|
||||
{
|
||||
"sensor": [
|
||||
{
|
||||
CONF_UUID: "import-uuid-1",
|
||||
CONF_HOST: "importhost",
|
||||
CONF_PORT: 8080,
|
||||
CONF_PLATFORM: "volkszaehler",
|
||||
},
|
||||
{
|
||||
CONF_UUID: "import-uuid-2",
|
||||
CONF_HOST: "importhost",
|
||||
CONF_PORT: 8080,
|
||||
CONF_PLATFORM: "volkszaehler",
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 1
|
||||
assert len(entries[0].subentries) == 2
|
||||
|
||||
assert mock_api.get_data.call_count == 3
|
||||
|
||||
|
||||
async def test_import_validation_error(
|
||||
hass: HomeAssistant, mock_api: AsyncMock
|
||||
) -> None:
|
||||
"""Test that import aborts when input validation fails."""
|
||||
mock_api.get_data.side_effect = VolkszaehlerApiConnectionError
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": "import"},
|
||||
data={
|
||||
CONF_UUID: "import-uuid",
|
||||
CONF_HOST: "importhost",
|
||||
CONF_PORT: 80,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "cannot_connect"
|
||||
|
||||
|
||||
async def test_user_duplicate_uuid_from_entry_unique_id(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test user flow duplicate UUID detection from an entry unique_id."""
|
||||
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
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_UUID: "existing-uuid",
|
||||
CONF_HOST: "new-host",
|
||||
CONF_PORT: 80,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
assert mock_api.get_data.call_count == 1
|
||||
|
||||
|
||||
async def test_import_same_host_different_port_creates_new_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test import with same host and different port creates a new entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
import_data = {
|
||||
CONF_UUID: "import-uuid-new-port",
|
||||
CONF_HOST: "localhost",
|
||||
CONF_PORT: 8080,
|
||||
CONF_PLATFORM: "volkszaehler",
|
||||
}
|
||||
await async_setup_component(hass, "sensor", {"sensor": import_data})
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 2
|
||||
assert any(
|
||||
entry.data == {CONF_HOST: "localhost", CONF_PORT: 8080} for entry in entries
|
||||
)
|
||||
|
||||
assert mock_api.get_data.call_count == 3
|
||||
|
||||
|
||||
async def test_subentry_duplicate_uuid(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test that subentry flow aborts for duplicate UUID."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.subentries.async_init(
|
||||
(mock_config_entry.entry_id, SUBENTRY_TYPE_CHANNEL),
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"],
|
||||
{CONF_UUID: "existing-uuid"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_subentry_validation_error(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test that subentry flow returns form error on validation failure."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.subentries.async_init(
|
||||
(mock_config_entry.entry_id, SUBENTRY_TYPE_CHANNEL),
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
|
||||
mock_api.get_data.side_effect = VolkszaehlerApiConnectionError
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"],
|
||||
{CONF_UUID: "new-uuid"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"]["base"] == "cannot_connect"
|
||||
|
||||
mock_api.get_data.side_effect = None
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"],
|
||||
{CONF_UUID: "new-uuid"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "new-uuid"
|
||||
assert result["data"] == {CONF_UUID: "new-uuid"}
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Tests for the Volkszaehler integration setup."""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from volkszaehler.exceptions import VolkszaehlerApiConnectionError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_setup_and_unload_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test successful setup and unload of a config entry."""
|
||||
mock_api.data = {"rows": []}
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
assert "existing-subentry-id" in mock_config_entry.runtime_data
|
||||
|
||||
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_setup_entry_connection_error(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test setup retry when the API cannot be reached."""
|
||||
mock_api.get_data.side_effect = VolkszaehlerApiConnectionError()
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_setup_entry_no_data(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_api: AsyncMock,
|
||||
) -> None:
|
||||
"""Test setup retry when the API returns no data."""
|
||||
mock_api.data = None
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
Reference in New Issue
Block a user