mirror of
https://github.com/home-assistant/core.git
synced 2025-04-24 01:08:12 +00:00
Add Iskra integration (#121488)
* Add iskra integration * iskra non resettable counters naming fix * added iskra config_flow test * fixed iskra integration according to code review * changed iskra config flow test * iskra integration, fixed codeowners * Removed counters code & minor fixes * added comment * Update homeassistant/components/iskra/__init__.py Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com> * Updated Iskra integration according to review * Update homeassistant/components/iskra/strings.json Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com> * Updated iskra integration according to review * minor iskra integration change * iskra integration changes according to review * iskra integration changes according to review * Changed iskra integration according to review * added iskra config_flow range validation * Fixed tests for iskra integration * Update homeassistant/components/iskra/coordinator.py * Update homeassistant/components/iskra/config_flow.py Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com> * Fixed iskra integration according to review * Changed voluptuous schema for iskra integration and added data_descriptions * Iskra integration tests lint error fix --------- Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Iskra kranj
GitHub
parent
da0d1b71ce
commit
b557e9e826
@ -728,6 +728,8 @@ build.json @home-assistant/supervisor
|
||||
/tests/components/iron_os/ @tr4nt0r
|
||||
/homeassistant/components/isal/ @bdraco
|
||||
/tests/components/isal/ @bdraco
|
||||
/homeassistant/components/iskra/ @iskramis
|
||||
/tests/components/iskra/ @iskramis
|
||||
/homeassistant/components/islamic_prayer_times/ @engrbm87 @cpfair
|
||||
/tests/components/islamic_prayer_times/ @engrbm87 @cpfair
|
||||
/homeassistant/components/israel_rail/ @shaiu
|
||||
|
||||
100
homeassistant/components/iskra/__init__.py
Normal file
100
homeassistant/components/iskra/__init__.py
Normal file
@ -0,0 +1,100 @@
|
||||
"""The iskra integration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pyiskra.adapters import Modbus, RestAPI
|
||||
from pyiskra.devices import Device
|
||||
from pyiskra.exceptions import DeviceConnectionError, DeviceNotSupported, NotAuthorised
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_ADDRESS,
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_PROTOCOL,
|
||||
CONF_USERNAME,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
|
||||
from .const import DOMAIN, MANUFACTURER
|
||||
from .coordinator import IskraDataUpdateCoordinator
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
|
||||
type IskraConfigEntry = ConfigEntry[list[IskraDataUpdateCoordinator]]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: IskraConfigEntry) -> bool:
|
||||
"""Set up iskra device from a config entry."""
|
||||
conf = entry.data
|
||||
adapter = None
|
||||
|
||||
if conf[CONF_PROTOCOL] == "modbus_tcp":
|
||||
adapter = Modbus(
|
||||
ip_address=conf[CONF_HOST],
|
||||
protocol="tcp",
|
||||
port=conf[CONF_PORT],
|
||||
modbus_address=conf[CONF_ADDRESS],
|
||||
)
|
||||
elif conf[CONF_PROTOCOL] == "rest_api":
|
||||
authentication = None
|
||||
if (username := conf.get(CONF_USERNAME)) is not None and (
|
||||
password := conf.get(CONF_PASSWORD)
|
||||
) is not None:
|
||||
authentication = {
|
||||
"username": username,
|
||||
"password": password,
|
||||
}
|
||||
adapter = RestAPI(ip_address=conf[CONF_HOST], authentication=authentication)
|
||||
|
||||
# Try connecting to the device and create pyiskra device object
|
||||
try:
|
||||
base_device = await Device.create_device(adapter)
|
||||
except DeviceConnectionError as e:
|
||||
raise ConfigEntryNotReady("Cannot connect to the device") from e
|
||||
except NotAuthorised as e:
|
||||
raise ConfigEntryNotReady("Not authorised to connect to the device") from e
|
||||
except DeviceNotSupported as e:
|
||||
raise ConfigEntryNotReady("Device not supported") from e
|
||||
|
||||
# Initialize the device
|
||||
await base_device.init()
|
||||
|
||||
# if the device is a gateway, add all child devices, otherwise add the device itself.
|
||||
if base_device.is_gateway:
|
||||
# Add the gateway device to the device registry
|
||||
device_registry = dr.async_get(hass)
|
||||
device_registry.async_get_or_create(
|
||||
config_entry_id=entry.entry_id,
|
||||
identifiers={(DOMAIN, base_device.serial)},
|
||||
manufacturer=MANUFACTURER,
|
||||
name=base_device.model,
|
||||
model=base_device.model,
|
||||
sw_version=base_device.fw_version,
|
||||
)
|
||||
|
||||
coordinators = [
|
||||
IskraDataUpdateCoordinator(hass, child_device)
|
||||
for child_device in base_device.get_child_devices()
|
||||
]
|
||||
else:
|
||||
coordinators = [IskraDataUpdateCoordinator(hass, base_device)]
|
||||
|
||||
for coordinator in coordinators:
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
entry.runtime_data = coordinators
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: IskraConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
253
homeassistant/components/iskra/config_flow.py
Normal file
253
homeassistant/components/iskra/config_flow.py
Normal file
@ -0,0 +1,253 @@
|
||||
"""Config flow for iskra integration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pyiskra.adapters import Modbus, RestAPI
|
||||
from pyiskra.exceptions import (
|
||||
DeviceConnectionError,
|
||||
DeviceTimeoutError,
|
||||
InvalidResponseCode,
|
||||
NotAuthorised,
|
||||
)
|
||||
from pyiskra.helper import BasicInfo
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
|
||||
from homeassistant.const import (
|
||||
CONF_ADDRESS,
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_PROTOCOL,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.selector import (
|
||||
NumberSelector,
|
||||
NumberSelectorConfig,
|
||||
NumberSelectorMode,
|
||||
SelectSelector,
|
||||
SelectSelectorConfig,
|
||||
SelectSelectorMode,
|
||||
)
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
STEP_USER_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST): str,
|
||||
vol.Required(CONF_PROTOCOL, default="rest_api"): SelectSelector(
|
||||
SelectSelectorConfig(
|
||||
options=["rest_api", "modbus_tcp"],
|
||||
mode=SelectSelectorMode.LIST,
|
||||
translation_key="protocol",
|
||||
),
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
STEP_AUTHENTICATION_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_USERNAME): str,
|
||||
vol.Required(CONF_PASSWORD): str,
|
||||
}
|
||||
)
|
||||
|
||||
# CONF_ADDRESS validation is done later in code, as if ranges are set in voluptuous it turns into a slider
|
||||
STEP_MODBUS_TCP_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_PORT, default=10001): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=0, max=65535)
|
||||
),
|
||||
vol.Required(CONF_ADDRESS, default=33): NumberSelector(
|
||||
NumberSelectorConfig(min=1, max=255, mode=NumberSelectorMode.BOX)
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def test_rest_api_connection(host: str, user_input: dict[str, Any]) -> BasicInfo:
|
||||
"""Check if the RestAPI requires authentication."""
|
||||
|
||||
rest_api = RestAPI(ip_address=host, authentication=user_input)
|
||||
try:
|
||||
basic_info = await rest_api.get_basic_info()
|
||||
except NotAuthorised as e:
|
||||
raise NotAuthorised from e
|
||||
except (DeviceConnectionError, DeviceTimeoutError, InvalidResponseCode) as e:
|
||||
raise CannotConnect from e
|
||||
except Exception as e:
|
||||
_LOGGER.error("Unexpected exception: %s", e)
|
||||
raise UnknownException from e
|
||||
|
||||
return basic_info
|
||||
|
||||
|
||||
async def test_modbus_connection(host: str, user_input: dict[str, Any]) -> BasicInfo:
|
||||
"""Test the Modbus connection."""
|
||||
modbus_api = Modbus(
|
||||
ip_address=host,
|
||||
protocol="tcp",
|
||||
port=user_input[CONF_PORT],
|
||||
modbus_address=user_input[CONF_ADDRESS],
|
||||
)
|
||||
|
||||
try:
|
||||
basic_info = await modbus_api.get_basic_info()
|
||||
except NotAuthorised as e:
|
||||
raise NotAuthorised from e
|
||||
except (DeviceConnectionError, DeviceTimeoutError, InvalidResponseCode) as e:
|
||||
raise CannotConnect from e
|
||||
except Exception as e:
|
||||
_LOGGER.error("Unexpected exception: %s", e)
|
||||
raise UnknownException from e
|
||||
|
||||
return basic_info
|
||||
|
||||
|
||||
class IskraConfigFlowFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for iskra."""
|
||||
|
||||
VERSION = 1
|
||||
host: str
|
||||
protocol: str
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle a flow initiated by the user."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
self.host = user_input[CONF_HOST]
|
||||
self.protocol = user_input[CONF_PROTOCOL]
|
||||
if self.protocol == "rest_api":
|
||||
# Check if authentication is required.
|
||||
try:
|
||||
device_info = await test_rest_api_connection(self.host, user_input)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except NotAuthorised:
|
||||
# Proceed to authentication step.
|
||||
return await self.async_step_authentication()
|
||||
except UnknownException:
|
||||
errors["base"] = "unknown"
|
||||
# If the connection was not successful, show an error.
|
||||
|
||||
# If the connection was successful, create the device.
|
||||
if not errors:
|
||||
return await self._create_entry(
|
||||
host=self.host,
|
||||
protocol=self.protocol,
|
||||
device_info=device_info,
|
||||
user_input=user_input,
|
||||
)
|
||||
|
||||
if self.protocol == "modbus_tcp":
|
||||
# Proceed to modbus step.
|
||||
return await self.async_step_modbus_tcp()
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=STEP_USER_DATA_SCHEMA,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_authentication(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the authentication step."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
try:
|
||||
device_info = await test_rest_api_connection(self.host, user_input)
|
||||
# If the connection failed, abort.
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
# If the authentication failed, show an error and authentication form again.
|
||||
except NotAuthorised:
|
||||
errors["base"] = "invalid_auth"
|
||||
except UnknownException:
|
||||
errors["base"] = "unknown"
|
||||
|
||||
# if the connection was successful, create the device.
|
||||
if not errors:
|
||||
return await self._create_entry(
|
||||
self.host,
|
||||
self.protocol,
|
||||
device_info=device_info,
|
||||
user_input=user_input,
|
||||
)
|
||||
|
||||
# If there's no user_input or there was an error, show the authentication form again.
|
||||
return self.async_show_form(
|
||||
step_id="authentication",
|
||||
data_schema=STEP_AUTHENTICATION_DATA_SCHEMA,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_modbus_tcp(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the Modbus TCP step."""
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
# If there's user_input, check the connection.
|
||||
if user_input is not None:
|
||||
# convert to integer
|
||||
user_input[CONF_ADDRESS] = int(user_input[CONF_ADDRESS])
|
||||
|
||||
try:
|
||||
device_info = await test_modbus_connection(self.host, user_input)
|
||||
|
||||
# If the connection failed, show an error.
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except UnknownException:
|
||||
errors["base"] = "unknown"
|
||||
|
||||
# If the connection was successful, create the device.
|
||||
if not errors:
|
||||
return await self._create_entry(
|
||||
host=self.host,
|
||||
protocol=self.protocol,
|
||||
device_info=device_info,
|
||||
user_input=user_input,
|
||||
)
|
||||
|
||||
# If there's no user_input or there was an error, show the modbus form again.
|
||||
return self.async_show_form(
|
||||
step_id="modbus_tcp",
|
||||
data_schema=STEP_MODBUS_TCP_DATA_SCHEMA,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def _create_entry(
|
||||
self,
|
||||
host: str,
|
||||
protocol: str,
|
||||
device_info: BasicInfo,
|
||||
user_input: dict[str, Any],
|
||||
) -> ConfigFlowResult:
|
||||
"""Create the config entry."""
|
||||
|
||||
await self.async_set_unique_id(device_info.serial)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
return self.async_create_entry(
|
||||
title=device_info.model,
|
||||
data={CONF_HOST: host, CONF_PROTOCOL: protocol, **user_input},
|
||||
)
|
||||
|
||||
|
||||
class CannotConnect(HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
|
||||
|
||||
class UnknownException(HomeAssistantError):
|
||||
"""Error to indicate an unknown exception occurred."""
|
||||
25
homeassistant/components/iskra/const.py
Normal file
25
homeassistant/components/iskra/const.py
Normal file
@ -0,0 +1,25 @@
|
||||
"""Constants for the iskra integration."""
|
||||
|
||||
DOMAIN = "iskra"
|
||||
MANUFACTURER = "Iskra d.o.o"
|
||||
|
||||
# POWER
|
||||
ATTR_TOTAL_APPARENT_POWER = "total_apparent_power"
|
||||
ATTR_TOTAL_REACTIVE_POWER = "total_reactive_power"
|
||||
ATTR_TOTAL_ACTIVE_POWER = "total_active_power"
|
||||
ATTR_PHASE1_POWER = "phase1_power"
|
||||
ATTR_PHASE2_POWER = "phase2_power"
|
||||
ATTR_PHASE3_POWER = "phase3_power"
|
||||
|
||||
# Voltage
|
||||
ATTR_PHASE1_VOLTAGE = "phase1_voltage"
|
||||
ATTR_PHASE2_VOLTAGE = "phase2_voltage"
|
||||
ATTR_PHASE3_VOLTAGE = "phase3_voltage"
|
||||
|
||||
# Current
|
||||
ATTR_PHASE1_CURRENT = "phase1_current"
|
||||
ATTR_PHASE2_CURRENT = "phase2_current"
|
||||
ATTR_PHASE3_CURRENT = "phase3_current"
|
||||
|
||||
# Frequency
|
||||
ATTR_FREQUENCY = "frequency"
|
||||
57
homeassistant/components/iskra/coordinator.py
Normal file
57
homeassistant/components/iskra/coordinator.py
Normal file
@ -0,0 +1,57 @@
|
||||
"""Coordinator for Iskra integration."""
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from pyiskra.devices import Device
|
||||
from pyiskra.exceptions import (
|
||||
DeviceConnectionError,
|
||||
DeviceTimeoutError,
|
||||
InvalidResponseCode,
|
||||
NotAuthorised,
|
||||
)
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IskraDataUpdateCoordinator(DataUpdateCoordinator[None]):
|
||||
"""Class to manage fetching Iskra data."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, device: Device) -> None:
|
||||
"""Initialize."""
|
||||
self.device = device
|
||||
|
||||
update_interval = timedelta(seconds=60)
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=DOMAIN,
|
||||
update_interval=update_interval,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> None:
|
||||
"""Fetch data from Iskra device."""
|
||||
try:
|
||||
await self.device.update_status()
|
||||
except DeviceTimeoutError as e:
|
||||
raise UpdateFailed(
|
||||
f"Timeout error occurred while updating data for device {self.device.serial}"
|
||||
) from e
|
||||
except DeviceConnectionError as e:
|
||||
raise UpdateFailed(
|
||||
f"Connection error occurred while updating data for device {self.device.serial}"
|
||||
) from e
|
||||
except NotAuthorised as e:
|
||||
raise UpdateFailed(
|
||||
f"Not authorised to fetch data from device {self.device.serial}"
|
||||
) from e
|
||||
except InvalidResponseCode as e:
|
||||
raise UpdateFailed(
|
||||
f"Invalid response code from device {self.device.serial}"
|
||||
) from e
|
||||
38
homeassistant/components/iskra/entity.py
Normal file
38
homeassistant/components/iskra/entity.py
Normal file
@ -0,0 +1,38 @@
|
||||
"""Base entity for Iskra devices."""
|
||||
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN, MANUFACTURER
|
||||
from .coordinator import IskraDataUpdateCoordinator
|
||||
|
||||
|
||||
class IskraEntity(CoordinatorEntity[IskraDataUpdateCoordinator]):
|
||||
"""Representation a base Iskra device."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(self, coordinator: IskraDataUpdateCoordinator) -> None:
|
||||
"""Initialize the Iskra device."""
|
||||
super().__init__(coordinator)
|
||||
self.device = coordinator.device
|
||||
gateway = self.device.parent_device
|
||||
|
||||
if gateway is not None:
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self.device.serial)},
|
||||
manufacturer=MANUFACTURER,
|
||||
model=self.device.model,
|
||||
name=self.device.model,
|
||||
sw_version=self.device.fw_version,
|
||||
serial_number=self.device.serial,
|
||||
via_device=(DOMAIN, gateway.serial),
|
||||
)
|
||||
else:
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self.device.serial)},
|
||||
manufacturer=MANUFACTURER,
|
||||
model=self.device.model,
|
||||
sw_version=self.device.fw_version,
|
||||
serial_number=self.device.serial,
|
||||
)
|
||||
11
homeassistant/components/iskra/manifest.json
Normal file
11
homeassistant/components/iskra/manifest.json
Normal file
@ -0,0 +1,11 @@
|
||||
{
|
||||
"domain": "iskra",
|
||||
"name": "iskra",
|
||||
"codeowners": ["@iskramis"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/iskra",
|
||||
"integration_type": "hub",
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["pyiskra"],
|
||||
"requirements": ["pyiskra==0.1.8"]
|
||||
}
|
||||
229
homeassistant/components/iskra/sensor.py
Normal file
229
homeassistant/components/iskra/sensor.py
Normal file
@ -0,0 +1,229 @@
|
||||
"""Support for Iskra."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
|
||||
from pyiskra.devices import Device
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
UnitOfApparentPower,
|
||||
UnitOfElectricCurrent,
|
||||
UnitOfElectricPotential,
|
||||
UnitOfFrequency,
|
||||
UnitOfPower,
|
||||
UnitOfReactivePower,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import IskraConfigEntry
|
||||
from .const import (
|
||||
ATTR_FREQUENCY,
|
||||
ATTR_PHASE1_CURRENT,
|
||||
ATTR_PHASE1_POWER,
|
||||
ATTR_PHASE1_VOLTAGE,
|
||||
ATTR_PHASE2_CURRENT,
|
||||
ATTR_PHASE2_POWER,
|
||||
ATTR_PHASE2_VOLTAGE,
|
||||
ATTR_PHASE3_CURRENT,
|
||||
ATTR_PHASE3_POWER,
|
||||
ATTR_PHASE3_VOLTAGE,
|
||||
ATTR_TOTAL_ACTIVE_POWER,
|
||||
ATTR_TOTAL_APPARENT_POWER,
|
||||
ATTR_TOTAL_REACTIVE_POWER,
|
||||
)
|
||||
from .coordinator import IskraDataUpdateCoordinator
|
||||
from .entity import IskraEntity
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class IskraSensorEntityDescription(SensorEntityDescription):
|
||||
"""Describes Iskra sensor entity."""
|
||||
|
||||
value_func: Callable[[Device], float | None]
|
||||
|
||||
|
||||
SENSOR_TYPES: tuple[IskraSensorEntityDescription, ...] = (
|
||||
# Power
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_TOTAL_ACTIVE_POWER,
|
||||
translation_key="total_active_power",
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfPower.WATT,
|
||||
value_func=lambda device: device.measurements.total.active_power.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_TOTAL_REACTIVE_POWER,
|
||||
translation_key="total_reactive_power",
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfReactivePower.VOLT_AMPERE_REACTIVE,
|
||||
value_func=lambda device: device.measurements.total.reactive_power.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_TOTAL_APPARENT_POWER,
|
||||
translation_key="total_apparent_power",
|
||||
device_class=SensorDeviceClass.APPARENT_POWER,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfApparentPower.VOLT_AMPERE,
|
||||
value_func=lambda device: device.measurements.total.apparent_power.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE1_POWER,
|
||||
translation_key="phase1_power",
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfPower.WATT,
|
||||
value_func=lambda device: device.measurements.phases[0].active_power.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE2_POWER,
|
||||
translation_key="phase2_power",
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfPower.WATT,
|
||||
value_func=lambda device: device.measurements.phases[1].active_power.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE3_POWER,
|
||||
translation_key="phase3_power",
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfPower.WATT,
|
||||
value_func=lambda device: device.measurements.phases[2].active_power.value,
|
||||
),
|
||||
# Voltage
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE1_VOLTAGE,
|
||||
translation_key="phase1_voltage",
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
value_func=lambda device: device.measurements.phases[0].voltage.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE2_VOLTAGE,
|
||||
translation_key="phase2_voltage",
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
value_func=lambda device: device.measurements.phases[1].voltage.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE3_VOLTAGE,
|
||||
translation_key="phase3_voltage",
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
value_func=lambda device: device.measurements.phases[2].voltage.value,
|
||||
),
|
||||
# Current
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE1_CURRENT,
|
||||
translation_key="phase1_current",
|
||||
device_class=SensorDeviceClass.CURRENT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
|
||||
value_func=lambda device: device.measurements.phases[0].current.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE2_CURRENT,
|
||||
translation_key="phase2_current",
|
||||
device_class=SensorDeviceClass.CURRENT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
|
||||
value_func=lambda device: device.measurements.phases[1].current.value,
|
||||
),
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_PHASE3_CURRENT,
|
||||
translation_key="phase3_current",
|
||||
device_class=SensorDeviceClass.CURRENT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
|
||||
value_func=lambda device: device.measurements.phases[2].current.value,
|
||||
),
|
||||
# Frequency
|
||||
IskraSensorEntityDescription(
|
||||
key=ATTR_FREQUENCY,
|
||||
translation_key="frequency",
|
||||
device_class=SensorDeviceClass.FREQUENCY,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
native_unit_of_measurement=UnitOfFrequency.HERTZ,
|
||||
value_func=lambda device: device.measurements.frequency.value,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: IskraConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up Iskra sensors based on config_entry."""
|
||||
|
||||
# Device that uses the config entry.
|
||||
coordinators = entry.runtime_data
|
||||
|
||||
entities: list[IskraSensor] = []
|
||||
|
||||
# Add sensors for each device.
|
||||
for coordinator in coordinators:
|
||||
device = coordinator.device
|
||||
sensors = []
|
||||
|
||||
# Add measurement sensors.
|
||||
if device.supports_measurements:
|
||||
sensors.append(ATTR_FREQUENCY)
|
||||
sensors.append(ATTR_TOTAL_APPARENT_POWER)
|
||||
sensors.append(ATTR_TOTAL_ACTIVE_POWER)
|
||||
sensors.append(ATTR_TOTAL_REACTIVE_POWER)
|
||||
if device.phases >= 1:
|
||||
sensors.append(ATTR_PHASE1_VOLTAGE)
|
||||
sensors.append(ATTR_PHASE1_POWER)
|
||||
sensors.append(ATTR_PHASE1_CURRENT)
|
||||
if device.phases >= 2:
|
||||
sensors.append(ATTR_PHASE2_VOLTAGE)
|
||||
sensors.append(ATTR_PHASE2_POWER)
|
||||
sensors.append(ATTR_PHASE2_CURRENT)
|
||||
if device.phases >= 3:
|
||||
sensors.append(ATTR_PHASE3_VOLTAGE)
|
||||
sensors.append(ATTR_PHASE3_POWER)
|
||||
sensors.append(ATTR_PHASE3_CURRENT)
|
||||
|
||||
entities.extend(
|
||||
IskraSensor(coordinator, description)
|
||||
for description in SENSOR_TYPES
|
||||
if description.key in sensors
|
||||
)
|
||||
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class IskraSensor(IskraEntity, SensorEntity):
|
||||
"""Representation of a Sensor."""
|
||||
|
||||
entity_description: IskraSensorEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: IskraDataUpdateCoordinator,
|
||||
description: IskraSensorEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.entity_description = description
|
||||
self._attr_unique_id = f"{coordinator.device.serial}_{description.key}"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
"""Return the state of the sensor."""
|
||||
return self.entity_description.value_func(self.device)
|
||||
92
homeassistant/components/iskra/strings.json
Normal file
92
homeassistant/components/iskra/strings.json
Normal file
@ -0,0 +1,92 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Configure Iskra Device",
|
||||
"description": "Enter the IP address of your Iskra Device and select protocol.",
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]"
|
||||
},
|
||||
"data_description": {
|
||||
"host": "Hostname or IP address of your Iskra device."
|
||||
}
|
||||
},
|
||||
"authentication": {
|
||||
"title": "Configure Rest API Credentials",
|
||||
"description": "Enter username and password",
|
||||
"data": {
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]"
|
||||
}
|
||||
},
|
||||
"modbus_tcp": {
|
||||
"title": "Configure Modbus TCP",
|
||||
"description": "Enter Modbus TCP port and device's Modbus address.",
|
||||
"data": {
|
||||
"port": "[%key:common::config_flow::data::port%]",
|
||||
"address": "Modbus address"
|
||||
},
|
||||
"data_description": {
|
||||
"port": "Port number can be found in the device's settings menu.",
|
||||
"address": "Modbus address can be found in the device's settings menu."
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
}
|
||||
},
|
||||
"selector": {
|
||||
"protocol": {
|
||||
"options": {
|
||||
"rest_api": "Rest API",
|
||||
"modbus_tcp": "Modbus TCP"
|
||||
}
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"sensor": {
|
||||
"total_active_power": {
|
||||
"name": "Total active power"
|
||||
},
|
||||
"total_apparent_power": {
|
||||
"name": "Total apparent power"
|
||||
},
|
||||
"total_reactive_power": {
|
||||
"name": "Total reactive power"
|
||||
},
|
||||
"phase1_power": {
|
||||
"name": "Phase 1 power"
|
||||
},
|
||||
"phase2_power": {
|
||||
"name": "Phase 2 power"
|
||||
},
|
||||
"phase3_power": {
|
||||
"name": "Phase 3 power"
|
||||
},
|
||||
"phase1_voltage": {
|
||||
"name": "Phase 1 voltage"
|
||||
},
|
||||
"phase2_voltage": {
|
||||
"name": "Phase 2 voltage"
|
||||
},
|
||||
"phase3_voltage": {
|
||||
"name": "Phase 3 voltage"
|
||||
},
|
||||
"phase1_current": {
|
||||
"name": "Phase 1 current"
|
||||
},
|
||||
"phase2_current": {
|
||||
"name": "Phase 2 current"
|
||||
},
|
||||
"phase3_current": {
|
||||
"name": "Phase 3 current"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -285,6 +285,7 @@ FLOWS = {
|
||||
"ipp",
|
||||
"iqvia",
|
||||
"iron_os",
|
||||
"iskra",
|
||||
"islamic_prayer_times",
|
||||
"israel_rail",
|
||||
"iss",
|
||||
|
||||
@ -2908,6 +2908,12 @@
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"iskra": {
|
||||
"name": "iskra",
|
||||
"integration_type": "hub",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"islamic_prayer_times": {
|
||||
"integration_type": "hub",
|
||||
"config_flow": true,
|
||||
|
||||
@ -1956,6 +1956,9 @@ pyiqvia==2022.04.0
|
||||
# homeassistant.components.irish_rail_transport
|
||||
pyirishrail==0.0.2
|
||||
|
||||
# homeassistant.components.iskra
|
||||
pyiskra==0.1.8
|
||||
|
||||
# homeassistant.components.iss
|
||||
pyiss==1.0.1
|
||||
|
||||
|
||||
@ -1567,6 +1567,9 @@ pyipp==0.16.0
|
||||
# homeassistant.components.iqvia
|
||||
pyiqvia==2022.04.0
|
||||
|
||||
# homeassistant.components.iskra
|
||||
pyiskra==0.1.8
|
||||
|
||||
# homeassistant.components.iss
|
||||
pyiss==1.0.1
|
||||
|
||||
|
||||
1
tests/components/iskra/__init__.py
Normal file
1
tests/components/iskra/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Tests for the Iskra component."""
|
||||
46
tests/components/iskra/conftest.py
Normal file
46
tests/components/iskra/conftest.py
Normal file
@ -0,0 +1,46 @@
|
||||
"""Fixtures for mocking pyiskra's different protocols.
|
||||
|
||||
Fixtures:
|
||||
- `mock_pyiskra_rest`: Mock pyiskra Rest API protocol.
|
||||
- `mock_pyiskra_modbus`: Mock pyiskra Modbus protocol.
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from .const import PQ_MODEL, SERIAL, SG_MODEL
|
||||
|
||||
|
||||
class MockBasicInfo:
|
||||
"""Mock BasicInfo class."""
|
||||
|
||||
def __init__(self, model) -> None:
|
||||
"""Initialize the mock class."""
|
||||
self.serial = SERIAL
|
||||
self.model = model
|
||||
self.description = "Iskra mock device"
|
||||
self.location = "imagination"
|
||||
self.sw_ver = "1.0.0"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_pyiskra_rest():
|
||||
"""Mock Iskra API authenticate with Rest API protocol."""
|
||||
|
||||
with patch(
|
||||
"pyiskra.adapters.RestAPI.RestAPI.get_basic_info",
|
||||
return_value=MockBasicInfo(model=SG_MODEL),
|
||||
) as basic_info_mock:
|
||||
yield basic_info_mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_pyiskra_modbus():
|
||||
"""Mock Iskra API authenticate with Rest API protocol."""
|
||||
|
||||
with patch(
|
||||
"pyiskra.adapters.Modbus.Modbus.get_basic_info",
|
||||
return_value=MockBasicInfo(model=PQ_MODEL),
|
||||
) as basic_info_mock:
|
||||
yield basic_info_mock
|
||||
10
tests/components/iskra/const.py
Normal file
10
tests/components/iskra/const.py
Normal file
@ -0,0 +1,10 @@
|
||||
"""Constants used in the Iskra component tests."""
|
||||
|
||||
SG_MODEL = "SG-W1"
|
||||
PQ_MODEL = "MC784"
|
||||
SERIAL = "XXXXXXX"
|
||||
HOST = "192.1.0.1"
|
||||
MODBUS_PORT = 10001
|
||||
MODBUS_ADDRESS = 33
|
||||
USERNAME = "test_username"
|
||||
PASSWORD = "test_password"
|
||||
300
tests/components/iskra/test_config_flow.py
Normal file
300
tests/components/iskra/test_config_flow.py
Normal file
@ -0,0 +1,300 @@
|
||||
"""Tests for the Iskra config flow."""
|
||||
|
||||
from pyiskra.exceptions import (
|
||||
DeviceConnectionError,
|
||||
DeviceTimeoutError,
|
||||
InvalidResponseCode,
|
||||
NotAuthorised,
|
||||
)
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.iskra import DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_USER
|
||||
from homeassistant.const import (
|
||||
CONF_ADDRESS,
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_PROTOCOL,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from .const import (
|
||||
HOST,
|
||||
MODBUS_ADDRESS,
|
||||
MODBUS_PORT,
|
||||
PASSWORD,
|
||||
PQ_MODEL,
|
||||
SERIAL,
|
||||
SG_MODEL,
|
||||
USERNAME,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
# Test step_user with Rest API protocol
|
||||
async def test_user_rest_no_auth(hass: HomeAssistant, mock_pyiskra_rest) -> None:
|
||||
"""Test the user flow with Rest API protocol."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
# Test if user form is provided
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
# Test no authentication required
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_HOST: HOST, CONF_PROTOCOL: "rest_api"},
|
||||
)
|
||||
|
||||
# Test successful Rest API configuration
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == SG_MODEL
|
||||
assert result["data"] == {CONF_HOST: HOST, CONF_PROTOCOL: "rest_api"}
|
||||
|
||||
|
||||
async def test_user_rest_auth(hass: HomeAssistant, mock_pyiskra_rest) -> None:
|
||||
"""Test the user flow with Rest API protocol and authentication required."""
|
||||
mock_pyiskra_rest.side_effect = NotAuthorised
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
# Test if user form is provided
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
# Test if prompted to enter username and password if not authorised
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_HOST: HOST, CONF_PROTOCOL: "rest_api"},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "authentication"
|
||||
|
||||
# Test failed authentication
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": "invalid_auth"}
|
||||
assert result["step_id"] == "authentication"
|
||||
|
||||
# Test successful authentication
|
||||
mock_pyiskra_rest.side_effect = None
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
|
||||
# Test successful Rest API configuration
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == SG_MODEL
|
||||
assert result["data"] == {
|
||||
CONF_HOST: HOST,
|
||||
CONF_PROTOCOL: "rest_api",
|
||||
CONF_USERNAME: USERNAME,
|
||||
CONF_PASSWORD: PASSWORD,
|
||||
}
|
||||
|
||||
|
||||
async def test_user_modbus(hass: HomeAssistant, mock_pyiskra_modbus) -> None:
|
||||
"""Test the user flow with Modbus TCP protocol."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
# Test if user form is provided
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_HOST: HOST, CONF_PROTOCOL: "modbus_tcp"},
|
||||
)
|
||||
|
||||
# Test if propmpted to enter port and address
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "modbus_tcp"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_PORT: MODBUS_PORT,
|
||||
CONF_ADDRESS: MODBUS_ADDRESS,
|
||||
},
|
||||
)
|
||||
|
||||
# Test successful Modbus TCP configuration
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == PQ_MODEL
|
||||
assert result["data"] == {
|
||||
CONF_HOST: HOST,
|
||||
CONF_PROTOCOL: "modbus_tcp",
|
||||
CONF_PORT: MODBUS_PORT,
|
||||
CONF_ADDRESS: MODBUS_ADDRESS,
|
||||
}
|
||||
|
||||
|
||||
async def test_modbus_abort_if_already_setup(
|
||||
hass: HomeAssistant, mock_pyiskra_modbus
|
||||
) -> None:
|
||||
"""Test we abort if Iskra is already setup."""
|
||||
|
||||
MockConfigEntry(domain=DOMAIN, unique_id=SERIAL).add_to_hass(hass)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_HOST: HOST, CONF_PROTOCOL: "modbus_tcp"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "modbus_tcp"
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_PORT: MODBUS_PORT,
|
||||
CONF_ADDRESS: MODBUS_ADDRESS,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_rest_api_abort_if_already_setup(
|
||||
hass: HomeAssistant, mock_pyiskra_rest
|
||||
) -> None:
|
||||
"""Test we abort if Iskra is already setup."""
|
||||
|
||||
MockConfigEntry(domain=DOMAIN, unique_id=SERIAL).add_to_hass(hass)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_HOST: HOST, CONF_PROTOCOL: "rest_api"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("s_effect", "reason"),
|
||||
[
|
||||
(DeviceConnectionError, "cannot_connect"),
|
||||
(DeviceTimeoutError, "cannot_connect"),
|
||||
(InvalidResponseCode, "cannot_connect"),
|
||||
(Exception, "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_modbus_device_error(
|
||||
hass: HomeAssistant,
|
||||
mock_pyiskra_modbus,
|
||||
s_effect,
|
||||
reason,
|
||||
) -> None:
|
||||
"""Test device error with Modbus TCP protocol."""
|
||||
mock_pyiskra_modbus.side_effect = s_effect
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_HOST: HOST, CONF_PROTOCOL: "modbus_tcp"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "modbus_tcp"
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_PORT: MODBUS_PORT,
|
||||
CONF_ADDRESS: MODBUS_ADDRESS,
|
||||
},
|
||||
)
|
||||
|
||||
# Test if error returned
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "modbus_tcp"
|
||||
assert result["errors"] == {"base": reason}
|
||||
|
||||
# Remove side effect
|
||||
mock_pyiskra_modbus.side_effect = None
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_PORT: MODBUS_PORT,
|
||||
CONF_ADDRESS: MODBUS_ADDRESS,
|
||||
},
|
||||
)
|
||||
|
||||
# Test successful Modbus TCP configuration
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == PQ_MODEL
|
||||
assert result["data"] == {
|
||||
CONF_HOST: HOST,
|
||||
CONF_PROTOCOL: "modbus_tcp",
|
||||
CONF_PORT: MODBUS_PORT,
|
||||
CONF_ADDRESS: MODBUS_ADDRESS,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("s_effect", "reason"),
|
||||
[
|
||||
(DeviceConnectionError, "cannot_connect"),
|
||||
(DeviceTimeoutError, "cannot_connect"),
|
||||
(InvalidResponseCode, "cannot_connect"),
|
||||
(Exception, "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_rest_device_error(
|
||||
hass: HomeAssistant,
|
||||
mock_pyiskra_rest,
|
||||
s_effect,
|
||||
reason,
|
||||
) -> None:
|
||||
"""Test device error with Modbus TCP protocol."""
|
||||
mock_pyiskra_rest.side_effect = s_effect
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_HOST: HOST, CONF_PROTOCOL: "rest_api"},
|
||||
)
|
||||
|
||||
# Test if error returned
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": reason}
|
||||
|
||||
# Remove side effect
|
||||
mock_pyiskra_rest.side_effect = None
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_HOST: HOST, CONF_PROTOCOL: "rest_api"},
|
||||
)
|
||||
|
||||
# Test successful Rest API configuration
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == SG_MODEL
|
||||
assert result["data"] == {CONF_HOST: HOST, CONF_PROTOCOL: "rest_api"}
|
||||
Loading…
x
Reference in New Issue
Block a user