home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add custom panel for LCN configuration (#108664)

Browse Source 
* Add LCN panel using lcn-frontend module

* Move panel from sidebar to integration configuration

* Change OptionFlow to reconfigure step

* Change OptionFlow to reconfigure step

* Remove deprecation warning

* Fix docstring

* Add tests for lcn websockets

* Remove deepcopy

* Bump lcn-frontend to 0.1.3

* Add tests for lcn websockets

* Remove websocket command lcn/hosts

* Websocket scan tests cover modules not stored in config_entry

* Add comment to mock of hass.http

* Add a decorater to ensure the config_entry exists and return it

* Use entry_id instead of host_id

* Bump lcn-frontend to 0.1.5

* Use auto_id for websocket client send_json

* Create issues on yaml import errors

* Remove low level key deprecation warnings

* Method renaming

* Change issue id in issue creation

* Update tests for issue creation
This commit is contained in:
Andre Lengwenus
2024-08-21 11:33:47 +02:00
committed by GitHub
parent f33328308c
commit c276cfc371
24 changed files with 1234 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
homeassistant/components/lcn/__init__.py
View File

@ -27,6 +27,7 @@ from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import ConfigType
from .const import (
ADD_ENTITIES_CALLBACKS,
CONF_DIM_MODE,
CONF_DOMAIN_DATA,
CONF_SK_NUM_TRIES,
@ -47,6 +48,7 @@ from .helpers import (
)
from .schemas import CONFIG_SCHEMA # noqa: F401
from .services import SERVICES
from .websocket import register_panel_and_ws_api
_LOGGER = logging.getLogger(__name__)
@ -115,6 +117,7 @@ async def async_setup_entry(
_LOGGER.debug('LCN connected to "%s"', config_entry.title)
hass.data[DOMAIN][config_entry.entry_id] = {
CONNECTION: lcn_connection,
ADD_ENTITIES_CALLBACKS: {},
}
# Update config_entry with LCN device serials
await async_update_config_entry(hass, config_entry)
@ -140,6 +143,8 @@ async def async_setup_entry(
DOMAIN, service_name, service(hass).async_call_service, service.schema
)
await register_panel_and_ws_api(hass)
return True

11
homeassistant/components/lcn/binary_sensor.py
View File

@ -15,7 +15,13 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType
from . import LcnEntity
from .const import BINSENSOR_PORTS, CONF_DOMAIN_DATA, SETPOINTS
from .const import (
ADD_ENTITIES_CALLBACKS,
BINSENSOR_PORTS,
CONF_DOMAIN_DATA,
DOMAIN,
SETPOINTS,
)
from .helpers import DeviceConnectionType, InputType, get_device_connection
@ -43,6 +49,9 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
{DOMAIN_BINARY_SENSOR: (async_add_entities, create_lcn_binary_sensor_entity)}
)
async_add_entities(
create_lcn_binary_sensor_entity(hass, entity_config, config_entry)

5
homeassistant/components/lcn/climate.py
View File

@ -28,11 +28,13 @@ from homeassistant.helpers.typing import ConfigType
from . import LcnEntity
from .const import (
ADD_ENTITIES_CALLBACKS,
CONF_DOMAIN_DATA,
CONF_LOCKABLE,
CONF_MAX_TEMP,
CONF_MIN_TEMP,
CONF_SETPOINT,
DOMAIN,
)
from .helpers import DeviceConnectionType, InputType, get_device_connection
@ -56,6 +58,9 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
{DOMAIN_CLIMATE: (async_add_entities, create_lcn_climate_entity)}
)
async_add_entities(
create_lcn_climate_entity(hass, entity_config, config_entry)

176
homeassistant/components/lcn/config_flow.py
View File

@ -3,32 +3,50 @@
from __future__ import annotations
import logging
from typing import Any
import pypck
import voluptuous as vol
from homeassistant.config_entries import (
SOURCE_IMPORT,
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
)
from homeassistant import config_entries
from homeassistant.const import (
CONF_BASE,
CONF_DEVICES,
CONF_ENTITIES,
CONF_HOST,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType
from .const import CONF_DIM_MODE, CONF_SK_NUM_TRIES, DOMAIN
from .const import CONF_DIM_MODE, CONF_SK_NUM_TRIES, DIM_MODES, DOMAIN
from .helpers import purge_device_registry, purge_entity_registry
_LOGGER = logging.getLogger(__name__)
CONFIG_DATA = {
vol.Required(CONF_IP_ADDRESS, default=""): str,
vol.Required(CONF_PORT, default=4114): cv.positive_int,
vol.Required(CONF_USERNAME, default=""): str,
vol.Required(CONF_PASSWORD, default=""): str,
vol.Required(CONF_SK_NUM_TRIES, default=0): cv.positive_int,
vol.Required(CONF_DIM_MODE, default="STEPS200"): vol.In(DIM_MODES),
}
def get_config_entry(hass: HomeAssistant, data: ConfigType) -> ConfigEntry | None:
USER_DATA = {vol.Required(CONF_HOST, default="pchk"): str, **CONFIG_DATA}
CONFIG_SCHEMA = vol.Schema(CONFIG_DATA)
USER_SCHEMA = vol.Schema(USER_DATA)
def get_config_entry(
hass: HomeAssistant, data: ConfigType
) -> config_entries.ConfigEntry | None:
"""Check config entries for already configured entries based on the ip address/port."""
return next(
(
@ -41,8 +59,10 @@ def get_config_entry(hass: HomeAssistant, data: ConfigType) -> ConfigEntry | Non
)
async def validate_connection(host_name: str, data: ConfigType) -> ConfigType:
async def validate_connection(data: ConfigType) -> str | None:
"""Validate if a connection to LCN can be established."""
error = None
host_name = data[CONF_HOST]
host = data[CONF_IP_ADDRESS]
port = data[CONF_PORT]
username = data[CONF_USERNAME]
@ -61,43 +81,71 @@ async def validate_connection(host_name: str, data: ConfigType) -> ConfigType:
host, port, username, password, settings=settings
)
await connection.async_connect(timeout=5)
try:
await connection.async_connect(timeout=5)
_LOGGER.debug("LCN connection validated")
except pypck.connection.PchkAuthenticationError:
_LOGGER.warning('Authentication on PCHK "%s" failed', host_name)
error = "authentication_error"
except pypck.connection.PchkLicenseError:
_LOGGER.warning(
'Maximum number of connections on PCHK "%s" was '
"reached. An additional license key is required",
host_name,
)
error = "license_error"
except (TimeoutError, ConnectionRefusedError):
_LOGGER.warning('Connection to PCHK "%s" failed', host_name)
error = "connection_refused"
_LOGGER.debug("LCN connection validated")
await connection.async_close()
return data
return error
class LcnFlowHandler(ConfigFlow, domain=DOMAIN):
class LcnFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a LCN config flow."""
VERSION = 1
async def async_step_import(self, data: ConfigType) -> ConfigFlowResult:
async def async_step_import(
self, data: ConfigType
) -> config_entries.ConfigFlowResult:
"""Import existing configuration from LCN."""
host_name = data[CONF_HOST]
# validate the imported connection parameters
try:
await validate_connection(host_name, data)
except pypck.connection.PchkAuthenticationError:
_LOGGER.warning('Authentication on PCHK "%s" failed', host_name)
return self.async_abort(reason="authentication_error")
except pypck.connection.PchkLicenseError:
_LOGGER.warning(
(
'Maximum number of connections on PCHK "%s" was '
"reached. An additional license key is required"
),
host_name,
if error := await validate_connection(data):
async_create_issue(
self.hass,
DOMAIN,
error,
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.ERROR,
translation_key=error,
translation_placeholders={
"url": "/config/integrations/dashboard/add?domain=lcn"
},
)
return self.async_abort(reason="license_error")
except TimeoutError:
_LOGGER.warning('Connection to PCHK "%s" failed', host_name)
return self.async_abort(reason="connection_timeout")
return self.async_abort(reason=error)
async_create_issue(
self.hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2024.12.0",
is_fixable=False,
is_persistent=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "LCN",
},
)
# check if we already have a host with the same address configured
if entry := get_config_entry(self.hass, data):
entry.source = SOURCE_IMPORT
entry.source = config_entries.SOURCE_IMPORT
# Cleanup entity and device registry, if we imported from configuration.yaml to
# remove orphans when entities were removed from configuration
purge_entity_registry(self.hass, entry.entry_id, data)
@ -106,4 +154,64 @@ class LcnFlowHandler(ConfigFlow, domain=DOMAIN):
self.hass.config_entries.async_update_entry(entry, data=data)
return self.async_abort(reason="existing_configuration_updated")
return self.async_create_entry(title=f"{host_name}", data=data)
return self.async_create_entry(title=f"{data[CONF_HOST]}", data=data)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> config_entries.ConfigFlowResult:
"""Handle a flow initiated by the user."""
if user_input is None:
return self.async_show_form(step_id="user", data_schema=USER_SCHEMA)
errors = None
if get_config_entry(self.hass, user_input):
errors = {CONF_BASE: "already_configured"}
elif (error := await validate_connection(user_input)) is not None:
errors = {CONF_BASE: error}
if errors is not None:
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
USER_SCHEMA, user_input
),
errors=errors,
)
data: dict = {
**user_input,
CONF_DEVICES: [],
CONF_ENTITIES: [],
}
return self.async_create_entry(title=data[CONF_HOST], data=data)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> config_entries.ConfigFlowResult:
"""Reconfigure LCN configuration."""
errors = None
entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
assert entry
if user_input is not None:
user_input[CONF_HOST] = entry.data[CONF_HOST]
await self.hass.config_entries.async_unload(entry.entry_id)
if (error := await validate_connection(user_input)) is not None:
errors = {CONF_BASE: error}
if errors is None:
data = entry.data.copy()
data.update(user_input)
self.hass.config_entries.async_update_entry(entry, data=data)
await self.hass.config_entries.async_setup(entry.entry_id)
return self.async_abort(reason="reconfigure_successful")
await self.hass.config_entries.async_setup(entry.entry_id)
return self.async_show_form(
step_id="reconfigure",
data_schema=self.add_suggested_values_to_schema(CONFIG_SCHEMA, entry.data),
errors=errors or {},
)

1
homeassistant/components/lcn/const.py
View File

@ -18,6 +18,7 @@ DOMAIN = "lcn"
DATA_LCN = "lcn"
DEFAULT_NAME = "pchk"
ADD_ENTITIES_CALLBACKS = "add_entities_callbacks"
CONNECTION = "connection"
CONF_HARDWARE_SERIAL = "hardware_serial"
CONF_SOFTWARE_SERIAL = "software_serial"

11
homeassistant/components/lcn/cover.py
View File

@ -14,7 +14,13 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType
from . import LcnEntity
from .const import CONF_DOMAIN_DATA, CONF_MOTOR, CONF_REVERSE_TIME
from .const import (
ADD_ENTITIES_CALLBACKS,
CONF_DOMAIN_DATA,
CONF_MOTOR,
CONF_REVERSE_TIME,
DOMAIN,
)
from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0
@ -40,6 +46,9 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN cover entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
{DOMAIN_COVER: (async_add_entities, create_lcn_cover_entity)}
)
async_add_entities(
create_lcn_cover_entity(hass, entity_config, config_entry)

10
homeassistant/components/lcn/helpers.py
View File

@ -423,6 +423,16 @@ async def async_update_config_entry(
hass.config_entries.async_update_entry(config_entry, data=new_data)
def get_device_config(
address: AddressType, config_entry: ConfigEntry
) -> ConfigType | None:
"""Return the device configuration for given address and ConfigEntry."""
for device_config in config_entry.data[CONF_DEVICES]:
if tuple(device_config[CONF_ADDRESS]) == address:
return cast(ConfigType, device_config)
return None
def has_unique_host_names(hosts: list[ConfigType]) -> list[ConfigType]:
"""Validate that all connection names are unique.

5
homeassistant/components/lcn/light.py
View File

@ -22,10 +22,12 @@ from homeassistant.helpers.typing import ConfigType
from . import LcnEntity
from .const import (
ADD_ENTITIES_CALLBACKS,
CONF_DIMMABLE,
CONF_DOMAIN_DATA,
CONF_OUTPUT,
CONF_TRANSITION,
DOMAIN,
OUTPUT_PORTS,
)
from .helpers import DeviceConnectionType, InputType, get_device_connection
@ -53,6 +55,9 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN light entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
{DOMAIN_LIGHT: (async_add_entities, create_lcn_light_entity)}
)
async_add_entities(
create_lcn_light_entity(hass, entity_config, config_entry)

4
homeassistant/components/lcn/manifest.json
View File

@ -2,9 +2,9 @@
"domain": "lcn",
"name": "LCN",
"codeowners": ["@alengwenus"],
"config_flow": false,
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/lcn",
"iot_class": "local_push",
"loggers": ["pypck"],
"requirements": ["pypck==0.7.21"]
"requirements": ["pypck==0.7.21", "lcn-frontend==0.1.5"]
}

5
homeassistant/components/lcn/scene.py
View File

@ -15,10 +15,12 @@ from homeassistant.helpers.typing import ConfigType
from . import LcnEntity
from .const import (
ADD_ENTITIES_CALLBACKS,
CONF_DOMAIN_DATA,
CONF_OUTPUTS,
CONF_REGISTER,
CONF_TRANSITION,
DOMAIN,
OUTPUT_PORTS,
)
from .helpers import DeviceConnectionType, get_device_connection
@ -43,6 +45,9 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
{DOMAIN_SCENE: (async_add_entities, create_lcn_scene_entity)}
)
async_add_entities(
create_lcn_scene_entity(hass, entity_config, config_entry)

47
homeassistant/components/lcn/schemas.py
View File

@ -58,6 +58,8 @@ from .const import (
)
from .helpers import has_unique_host_names, is_address
ADDRESS_SCHEMA = vol.Coerce(tuple)
#
# Domain data
#
@ -169,23 +171,32 @@ CONNECTION_SCHEMA = vol.Schema(
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_CONNECTIONS): vol.All(
cv.ensure_list, has_unique_host_names, [CONNECTION_SCHEMA]
),
vol.Optional(CONF_BINARY_SENSORS): vol.All(
cv.ensure_list, [BINARY_SENSORS_SCHEMA]
),
vol.Optional(CONF_CLIMATES): vol.All(cv.ensure_list, [CLIMATES_SCHEMA]),
vol.Optional(CONF_COVERS): vol.All(cv.ensure_list, [COVERS_SCHEMA]),
vol.Optional(CONF_LIGHTS): vol.All(cv.ensure_list, [LIGHTS_SCHEMA]),
vol.Optional(CONF_SCENES): vol.All(cv.ensure_list, [SCENES_SCHEMA]),
vol.Optional(CONF_SENSORS): vol.All(cv.ensure_list, [SENSORS_SCHEMA]),
vol.Optional(CONF_SWITCHES): vol.All(cv.ensure_list, [SWITCHES_SCHEMA]),
}
)
},
vol.All(
cv.deprecated(DOMAIN),
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_CONNECTIONS): vol.All(
cv.ensure_list, has_unique_host_names, [CONNECTION_SCHEMA]
),
vol.Optional(CONF_BINARY_SENSORS): vol.All(
cv.ensure_list, [BINARY_SENSORS_SCHEMA]
),
vol.Optional(CONF_CLIMATES): vol.All(
cv.ensure_list, [CLIMATES_SCHEMA]
),
vol.Optional(CONF_COVERS): vol.All(cv.ensure_list, [COVERS_SCHEMA]),
vol.Optional(CONF_LIGHTS): vol.All(cv.ensure_list, [LIGHTS_SCHEMA]),
vol.Optional(CONF_SCENES): vol.All(cv.ensure_list, [SCENES_SCHEMA]),
vol.Optional(CONF_SENSORS): vol.All(
cv.ensure_list, [SENSORS_SCHEMA]
),
vol.Optional(CONF_SWITCHES): vol.All(
cv.ensure_list, [SWITCHES_SCHEMA]
),
},
)
},
),
extra=vol.ALLOW_EXTRA,
)

5
homeassistant/components/lcn/sensor.py
View File

@ -22,7 +22,9 @@ from homeassistant.helpers.typing import ConfigType
from . import LcnEntity
from .const import (
ADD_ENTITIES_CALLBACKS,
CONF_DOMAIN_DATA,
DOMAIN,
LED_PORTS,
S0_INPUTS,
SETPOINTS,
@ -56,6 +58,9 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
{DOMAIN_SENSOR: (async_add_entities, create_lcn_sensor_entity)}
)
async_add_entities(
create_lcn_sensor_entity(hass, entity_config, config_entry)

52
homeassistant/components/lcn/strings.json
View File

@ -14,6 +14,58 @@
"level": "Level"
}
},
"config": {
"step": {
"user": {
"title": "Setup LCN host",
"description": "Set up new connection to LCN host.",
"data": {
"host": "[%key:common::config_flow::data::name%]",
"ip_address": "[%key:common::config_flow::data::ip%]",
"port": "[%key:common::config_flow::data::port%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"sk_num_tries": "Segment coupler scan attempts",
"dim_mode": "Dimming mode"
}
},
"reconfigure": {
"title": "Reconfigure LCN host",
"description": "Reconfigure connection to LCN host.",
"data": {
"ip_address": "[%key:common::config_flow::data::ip%]",
"port": "[%key:common::config_flow::data::port%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"sk_num_tries": "Segment coupler scan attempts",
"dim_mode": "Dimming mode"
}
}
},
"error": {
"authentication_error": "Authentication failed. Wrong username or password.",
"license_error": "Maximum number of connections was reached. An additional licence key is required.",
"connection_refused": "Unable to connect to PCHK. Check IP and port.",
"already_configured": "PCHK connection using the same ip address/port is already configured."
},
"abort": {
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
}
},
"issues": {
"authentication_error": {
"title": "Authentication failed.",
"description": "Configuring LCN using YAML is being removed but there was an error importing your YAML configuration.\n\nEnsure username and password are correct.\n\nConsider removing the LCN YAML configuration from your configuration.yaml file and continue to [set up the integration]({url}) manually."
},
"license_error": {
"title": "Maximum number of connections was reached.",
"description": "Configuring LCN using YAML is being removed but there was an error importing your YAML configuration.\n\nEnsure sufficient PCHK licenses are registered and restart Home Assistant.\n\nConsider removing the LCN YAML configuration from your configuration.yaml file and continue to [set up the integration]({url}) manually."
},
"connection_refused": {
"title": "Unable to connect to PCHK.",
"description": "Configuring LCN using YAML is being removed but there was an error importing your YAML configuration.\n\nEnsure the connection (IP and port) to the LCN bus coupler is correct.\n\nConsider removing the LCN YAML configuration from your configuration.yaml file and continue to [set up the integration]({url}) manually."
}
},
"services": {
"output_abs": {
"name": "Output absolute brightness",

11
homeassistant/components/lcn/switch.py
View File

@ -14,7 +14,13 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType
from . import LcnEntity
from .const import CONF_DOMAIN_DATA, CONF_OUTPUT, OUTPUT_PORTS
from .const import (
ADD_ENTITIES_CALLBACKS,
CONF_DOMAIN_DATA,
CONF_OUTPUT,
DOMAIN,
OUTPUT_PORTS,
)
from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0
@ -40,6 +46,9 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
{DOMAIN_SWITCH: (async_add_entities, create_lcn_switch_entity)}
)
async_add_entities(
create_lcn_switch_entity(hass, entity_config, config_entry)

450
homeassistant/components/lcn/websocket.py Normal file
View File

@ -0,0 +1,450 @@
"""LCN Websocket API."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from functools import wraps
from typing import TYPE_CHECKING, Any, Final
import lcn_frontend as lcn_panel
import voluptuous as vol
from homeassistant.components import panel_custom, websocket_api
from homeassistant.components.websocket_api import AsyncWebSocketCommandHandler
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_ADDRESS,
CONF_DEVICES,
CONF_DOMAIN,
CONF_ENTITIES,
CONF_ENTITY_ID,
CONF_NAME,
CONF_RESOURCE,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
import homeassistant.helpers.config_validation as cv
from .const import (
ADD_ENTITIES_CALLBACKS,
CONF_DOMAIN_DATA,
CONF_HARDWARE_SERIAL,
CONF_HARDWARE_TYPE,
CONF_SOFTWARE_SERIAL,
CONNECTION,
DOMAIN,
)
from .helpers import (
DeviceConnectionType,
async_update_device_config,
generate_unique_id,
get_device_config,
get_device_connection,
get_resource,
purge_device_registry,
purge_entity_registry,
register_lcn_address_devices,
)
from .schemas import (
ADDRESS_SCHEMA,
DOMAIN_DATA_BINARY_SENSOR,
DOMAIN_DATA_CLIMATE,
DOMAIN_DATA_COVER,
DOMAIN_DATA_LIGHT,
DOMAIN_DATA_SCENE,
DOMAIN_DATA_SENSOR,
DOMAIN_DATA_SWITCH,
)
if TYPE_CHECKING:
from homeassistant.components.websocket_api import ActiveConnection
type AsyncLcnWebSocketCommandHandler = Callable[
[HomeAssistant, ActiveConnection, dict[str, Any], ConfigEntry], Awaitable[None]
]
URL_BASE: Final = "/lcn_static"
async def register_panel_and_ws_api(hass: HomeAssistant) -> None:
"""Register the LCN Panel and Websocket API."""
websocket_api.async_register_command(hass, websocket_get_device_configs)
websocket_api.async_register_command(hass, websocket_get_entity_configs)
websocket_api.async_register_command(hass, websocket_scan_devices)
websocket_api.async_register_command(hass, websocket_add_device)
websocket_api.async_register_command(hass, websocket_delete_device)
websocket_api.async_register_command(hass, websocket_add_entity)
websocket_api.async_register_command(hass, websocket_delete_entity)
if DOMAIN not in hass.data.get("frontend_panels", {}):
hass.http.register_static_path(
URL_BASE,
path=lcn_panel.locate_dir(),
cache_headers=lcn_panel.is_prod_build,
)
await panel_custom.async_register_panel(
hass=hass,
frontend_url_path=DOMAIN,
webcomponent_name=lcn_panel.webcomponent_name,
config_panel_domain=DOMAIN,
module_url=f"{URL_BASE}/{lcn_panel.entrypoint_js}",
embed_iframe=True,
require_admin=True,
)
def get_config_entry(
func: AsyncLcnWebSocketCommandHandler,
) -> AsyncWebSocketCommandHandler:
"""Websocket decorator to ensure the config_entry exists and return it."""
@callback
@wraps(func)
async def get_entry(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
) -> None:
"""Get config_entry."""
if not (config_entry := hass.config_entries.async_get_entry(msg["entry_id"])):
connection.send_result(msg["id"], False)
else:
await func(hass, connection, msg, config_entry)
return get_entry
@websocket_api.require_admin
@websocket_api.websocket_command(
{vol.Required("type"): "lcn/devices", vol.Required("entry_id"): cv.string}
)
@websocket_api.async_response
@get_config_entry
async def websocket_get_device_configs(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
config_entry: ConfigEntry,
) -> None:
"""Get device configs."""
connection.send_result(msg["id"], config_entry.data[CONF_DEVICES])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "lcn/entities",
vol.Required("entry_id"): cv.string,
vol.Optional(CONF_ADDRESS): ADDRESS_SCHEMA,
}
)
@websocket_api.async_response
@get_config_entry
async def websocket_get_entity_configs(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
config_entry: ConfigEntry,
) -> None:
"""Get entities configs."""
if CONF_ADDRESS in msg:
entity_configs = [
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if tuple(entity_config[CONF_ADDRESS]) == msg[CONF_ADDRESS]
]
else:
entity_configs = config_entry.data[CONF_ENTITIES]
entity_registry = er.async_get(hass)
for entity_config in entity_configs:
entity_unique_id = generate_unique_id(
config_entry.entry_id,
entity_config[CONF_ADDRESS],
entity_config[CONF_RESOURCE],
)
entity_id = entity_registry.async_get_entity_id(
entity_config[CONF_DOMAIN], DOMAIN, entity_unique_id
)
entity_config[CONF_ENTITY_ID] = entity_id
connection.send_result(msg["id"], entity_configs)
@websocket_api.require_admin
@websocket_api.websocket_command(
{vol.Required("type"): "lcn/devices/scan", vol.Required("entry_id"): cv.string}
)
@websocket_api.async_response
@get_config_entry
async def websocket_scan_devices(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
config_entry: ConfigEntry,
) -> None:
"""Scan for new devices."""
host_connection = hass.data[DOMAIN][config_entry.entry_id][CONNECTION]
await host_connection.scan_modules()
for device_connection in host_connection.address_conns.values():
if not device_connection.is_group:
await async_create_or_update_device_in_config_entry(
hass, device_connection, config_entry
)
# create/update devices in device registry
register_lcn_address_devices(hass, config_entry)
connection.send_result(msg["id"], config_entry.data[CONF_DEVICES])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "lcn/devices/add",
vol.Required("entry_id"): cv.string,
vol.Required(CONF_ADDRESS): ADDRESS_SCHEMA,
}
)
@websocket_api.async_response
@get_config_entry
async def websocket_add_device(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
config_entry: ConfigEntry,
) -> None:
"""Add a device."""
if get_device_config(msg[CONF_ADDRESS], config_entry):
connection.send_result(
msg["id"], False
) # device_config already in config_entry
return
device_config = {
CONF_ADDRESS: msg[CONF_ADDRESS],
CONF_NAME: "",
CONF_HARDWARE_SERIAL: -1,
CONF_SOFTWARE_SERIAL: -1,
CONF_HARDWARE_TYPE: -1,
}
# update device info from LCN
device_connection = get_device_connection(hass, msg[CONF_ADDRESS], config_entry)
await async_update_device_config(device_connection, device_config)
# add device_config to config_entry
device_configs = [*config_entry.data[CONF_DEVICES], device_config]
data = {**config_entry.data, CONF_DEVICES: device_configs}
hass.config_entries.async_update_entry(config_entry, data=data)
# create/update devices in device registry
register_lcn_address_devices(hass, config_entry)
connection.send_result(msg["id"], True)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "lcn/devices/delete",
vol.Required("entry_id"): cv.string,
vol.Required(CONF_ADDRESS): ADDRESS_SCHEMA,
}
)
@websocket_api.async_response
@get_config_entry
async def websocket_delete_device(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
config_entry: ConfigEntry,
) -> None:
"""Delete a device."""
device_config = get_device_config(msg[CONF_ADDRESS], config_entry)
device_registry = dr.async_get(hass)
identifiers = {
(DOMAIN, generate_unique_id(config_entry.entry_id, msg[CONF_ADDRESS]))
}
device = device_registry.async_get_device(identifiers, set())
if not (device and device_config):
connection.send_result(msg["id"], False)
return
# remove module/group device from config_entry data
device_configs = [
dc for dc in config_entry.data[CONF_DEVICES] if dc != device_config
]
data = {**config_entry.data, CONF_DEVICES: device_configs}
hass.config_entries.async_update_entry(config_entry, data=data)
# remove all child devices (and entities) from config_entry data
for entity_config in data[CONF_ENTITIES][:]:
if tuple(entity_config[CONF_ADDRESS]) == msg[CONF_ADDRESS]:
data[CONF_ENTITIES].remove(entity_config)
hass.config_entries.async_update_entry(config_entry, data=data)
# cleanup registries
purge_entity_registry(hass, config_entry.entry_id, data)
purge_device_registry(hass, config_entry.entry_id, data)
# return the device config, not all devices !!!
connection.send_result(msg["id"])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "lcn/entities/add",
vol.Required("entry_id"): cv.string,
vol.Required(CONF_ADDRESS): ADDRESS_SCHEMA,
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_DOMAIN): cv.string,
vol.Required(CONF_DOMAIN_DATA): vol.Any(
DOMAIN_DATA_BINARY_SENSOR,
DOMAIN_DATA_SENSOR,
DOMAIN_DATA_SWITCH,
DOMAIN_DATA_LIGHT,
DOMAIN_DATA_CLIMATE,
DOMAIN_DATA_COVER,
DOMAIN_DATA_SCENE,
),
}
)
@websocket_api.async_response
@get_config_entry
async def websocket_add_entity(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
config_entry: ConfigEntry,
) -> None:
"""Add an entity."""
if not (device_config := get_device_config(msg[CONF_ADDRESS], config_entry)):
connection.send_result(msg["id"], False)
return
domain_name = msg[CONF_DOMAIN]
domain_data = msg[CONF_DOMAIN_DATA]
resource = get_resource(domain_name, domain_data).lower()
unique_id = generate_unique_id(
config_entry.entry_id,
device_config[CONF_ADDRESS],
resource,
)
entity_registry = er.async_get(hass)
if entity_registry.async_get_entity_id(msg[CONF_DOMAIN], DOMAIN, unique_id):
connection.send_result(msg["id"], False)
return
entity_config = {
CONF_ADDRESS: msg[CONF_ADDRESS],
CONF_NAME: msg[CONF_NAME],
CONF_RESOURCE: resource,
CONF_DOMAIN: domain_name,
CONF_DOMAIN_DATA: domain_data,
}
# Create new entity and add to corresponding component
callbacks = hass.data[DOMAIN][msg["entry_id"]][ADD_ENTITIES_CALLBACKS]
async_add_entities, create_lcn_entity = callbacks[msg[CONF_DOMAIN]]
entity = create_lcn_entity(hass, entity_config, config_entry)
async_add_entities([entity])
# Add entity config to config_entry
entity_configs = [*config_entry.data[CONF_ENTITIES], entity_config]
data = {**config_entry.data, CONF_ENTITIES: entity_configs}
# schedule config_entry for save
hass.config_entries.async_update_entry(config_entry, data=data)
connection.send_result(msg["id"], True)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "lcn/entities/delete",
vol.Required("entry_id"): cv.string,
vol.Required(CONF_ADDRESS): ADDRESS_SCHEMA,
vol.Required(CONF_DOMAIN): cv.string,
vol.Required(CONF_RESOURCE): cv.string,
}
)
@websocket_api.async_response
@get_config_entry
async def websocket_delete_entity(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
config_entry: ConfigEntry,
) -> None:
"""Delete an entity."""
entity_config = next(
(
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if (
tuple(entity_config[CONF_ADDRESS]) == msg[CONF_ADDRESS]
and entity_config[CONF_DOMAIN] == msg[CONF_DOMAIN]
and entity_config[CONF_RESOURCE] == msg[CONF_RESOURCE]
)
),
None,
)
if entity_config is None:
connection.send_result(msg["id"], False)
return
entity_configs = [
ec for ec in config_entry.data[CONF_ENTITIES] if ec != entity_config
]
data = {**config_entry.data, CONF_ENTITIES: entity_configs}
hass.config_entries.async_update_entry(config_entry, data=data)
# cleanup registries
purge_entity_registry(hass, config_entry.entry_id, data)
purge_device_registry(hass, config_entry.entry_id, data)
connection.send_result(msg["id"])
async def async_create_or_update_device_in_config_entry(
hass: HomeAssistant,
device_connection: DeviceConnectionType,
config_entry: ConfigEntry,
) -> None:
"""Create or update device in config_entry according to given device_connection."""
address = (
device_connection.seg_id,
device_connection.addr_id,
device_connection.is_group,
)
device_configs = [*config_entry.data[CONF_DEVICES]]
data = {**config_entry.data, CONF_DEVICES: device_configs}
for device_config in data[CONF_DEVICES]:
if tuple(device_config[CONF_ADDRESS]) == address:
break # device already in config_entry
else:
# create new device_entry
device_config = {
CONF_ADDRESS: address,
CONF_NAME: "",
CONF_HARDWARE_SERIAL: -1,
CONF_SOFTWARE_SERIAL: -1,
CONF_HARDWARE_TYPE: -1,
}
data[CONF_DEVICES].append(device_config)
# update device_entry
await async_update_device_config(device_connection, device_config)
hass.config_entries.async_update_entry(config_entry, data=data)

1
homeassistant/generated/config_flows.py
View File

@ -310,6 +310,7 @@ FLOWS = {
"lastfm",
"launch_library",
"laundrify",
"lcn",
"ld2410_ble",
"leaone",
"led_ble",

2
homeassistant/generated/integrations.json
View File

@ -3178,7 +3178,7 @@
"lcn": {
"name": "LCN",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "local_push"
},
"ld2410_ble": {

3
requirements_all.txt
View File

@ -1242,6 +1242,9 @@ lakeside==0.13
# homeassistant.components.laundrify
laundrify-aio==1.2.2
# homeassistant.components.lcn
lcn-frontend==0.1.5
# homeassistant.components.ld2410_ble
ld2410-ble==0.1.1

3
requirements_test_all.txt
View File

@ -1035,6 +1035,9 @@ lacrosse-view==1.0.2
# homeassistant.components.laundrify
laundrify-aio==1.2.2
# homeassistant.components.lcn
lcn-frontend==0.1.5
# homeassistant.components.ld2410_ble
ld2410-ble==0.1.1

11
tests/components/lcn/conftest.py
View File

@ -3,7 +3,7 @@
from collections.abc import AsyncGenerator
import json
from typing import Any
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, Mock, patch
import pypck
from pypck.connection import PchkConnectionManager
@ -13,7 +13,7 @@ import pytest
from homeassistant.components.lcn.const import DOMAIN
from homeassistant.components.lcn.helpers import AddressType, generate_unique_id
from homeassistant.const import CONF_HOST
from homeassistant.const import CONF_ADDRESS, CONF_DEVICES, CONF_ENTITIES, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
@ -60,6 +60,7 @@ class MockPchkConnectionManager(PchkConnectionManager):
"""Get LCN address connection."""
return super().get_address_conn(addr, request_serials)
scan_modules = AsyncMock()
send_command = AsyncMock()
@ -67,6 +68,11 @@ def create_config_entry(name: str) -> MockConfigEntry:
"""Set up config entries with configuration data."""
fixture_filename = f"lcn/config_entry_{name}.json"
entry_data = json.loads(load_fixture(fixture_filename))
for device in entry_data[CONF_DEVICES]:
device[CONF_ADDRESS] = tuple(device[CONF_ADDRESS])
for entity in entry_data[CONF_ENTITIES]:
entity[CONF_ADDRESS] = tuple(entity[CONF_ADDRESS])
options = {}
title = entry_data[CONF_HOST]
@ -97,6 +103,7 @@ async def init_integration(
hass: HomeAssistant, entry: MockConfigEntry
) -> AsyncGenerator[MockPchkConnectionManager]:
"""Set up the LCN integration in Home Assistant."""
hass.http = Mock() # needs to be mocked as hass.http.register_static_path is called when registering the frontend
lcn_connection = None
def lcn_connection_factory(*args, **kwargs):

4
tests/components/lcn/fixtures/config_entry_pchk.json
View File

@ -9,14 +9,14 @@
"devices": [
{
"address": [0, 7, false],
"name": "",
"name": "TestModule",
"hardware_serial": -1,
"software_serial": -1,
"hardware_type": -1
},
{
"address": [0, 5, true],
"name": "",
"name": "TestGroup",
"hardware_serial": -1,
"software_serial": -1,
"hardware_type": -1

181
tests/components/lcn/test_config_flow.py
View File

@ -5,9 +5,11 @@ from unittest.mock import patch
from pypck.connection import PchkAuthenticationError, PchkLicenseError
import pytest
from homeassistant import config_entries
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.lcn.config_flow import LcnFlowHandler, validate_connection
from homeassistant.components.lcn.const import CONF_DIM_MODE, CONF_SK_NUM_TRIES, DOMAIN
from homeassistant.const import (
CONF_BASE,
CONF_DEVICES,
CONF_ENTITIES,
CONF_HOST,
@ -16,25 +18,33 @@ from homeassistant.const import (
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import issue_registry as ir
from tests.common import MockConfigEntry
IMPORT_DATA = {
CONF_HOST: "pchk",
CONFIG_DATA = {
CONF_IP_ADDRESS: "127.0.0.1",
CONF_PORT: 4114,
CONF_PORT: 1234,
CONF_USERNAME: "lcn",
CONF_PASSWORD: "lcn",
CONF_SK_NUM_TRIES: 0,
CONF_DIM_MODE: "STEPS200",
}
CONNECTION_DATA = {CONF_HOST: "pchk", **CONFIG_DATA}
IMPORT_DATA = {
**CONNECTION_DATA,
CONF_DEVICES: [],
CONF_ENTITIES: [],
}
async def test_step_import(hass: HomeAssistant) -> None:
async def test_step_import(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Test for import step."""
with (
@ -46,14 +56,18 @@ async def test_step_import(hass: HomeAssistant) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=data
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "pchk"
assert result["data"] == IMPORT_DATA
assert issue_registry.async_get_issue(
HOMEASSISTANT_DOMAIN, f"deprecated_yaml_{DOMAIN}"
)
async def test_step_import_existing_host(hass: HomeAssistant) -> None:
async def test_step_import_existing_host(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Test for update of config_entry if imported host already exists."""
# Create config entry and add it to hass
@ -67,13 +81,15 @@ async def test_step_import_existing_host(hass: HomeAssistant) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=imported_data
)
await hass.async_block_till_done()
# Check if config entry was updated
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "existing_configuration_updated"
assert mock_entry.source == config_entries.SOURCE_IMPORT
assert mock_entry.data == IMPORT_DATA
assert issue_registry.async_get_issue(
HOMEASSISTANT_DOMAIN, f"deprecated_yaml_{DOMAIN}"
)
@pytest.mark.parametrize(
@ -81,10 +97,12 @@ async def test_step_import_existing_host(hass: HomeAssistant) -> None:
[
(PchkAuthenticationError, "authentication_error"),
(PchkLicenseError, "license_error"),
(TimeoutError, "connection_timeout"),
(TimeoutError, "connection_refused"),
],
)
async def test_step_import_error(hass: HomeAssistant, error, reason) -> None:
async def test_step_import_error(
hass: HomeAssistant, issue_registry: ir.IssueRegistry, error, reason
) -> None:
"""Test for error in import is handled correctly."""
with patch(
"pypck.connection.PchkConnectionManager.async_connect", side_effect=error
@ -94,7 +112,146 @@ async def test_step_import_error(hass: HomeAssistant, error, reason) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=data
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == reason
assert issue_registry.async_get_issue(DOMAIN, reason)
async def test_show_form(hass: HomeAssistant) -> None:
"""Test that the form is served with no input."""
flow = LcnFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=None)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
async def test_step_user(hass):
"""Test for user step."""
with (
patch("pypck.connection.PchkConnectionManager.async_connect"),
patch("homeassistant.components.lcn.async_setup", return_value=True),
patch("homeassistant.components.lcn.async_setup_entry", return_value=True),
):
data = CONNECTION_DATA.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=data
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == CONNECTION_DATA[CONF_HOST]
assert result["data"] == {
**CONNECTION_DATA,
CONF_DEVICES: [],
CONF_ENTITIES: [],
}
async def test_step_user_existing_host(hass, entry):
"""Test for user defined host already exists."""
entry.add_to_hass(hass)
with patch("pypck.connection.PchkConnectionManager.async_connect"):
config_data = entry.data.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=config_data
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {CONF_BASE: "already_configured"}
@pytest.mark.parametrize(
("error", "errors"),
[
(PchkAuthenticationError, {CONF_BASE: "authentication_error"}),
(PchkLicenseError, {CONF_BASE: "license_error"}),
(TimeoutError, {CONF_BASE: "connection_refused"}),
],
)
async def test_step_user_error(hass, error, errors):
"""Test for error in user step is handled correctly."""
with patch(
"pypck.connection.PchkConnectionManager.async_connect", side_effect=error
):
data = CONNECTION_DATA.copy()
data.update({CONF_HOST: "pchk"})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=data
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == errors
async def test_step_reconfigure(hass, entry):
"""Test for reconfigure step."""
entry.add_to_hass(hass)
old_entry_data = entry.data.copy()
with (
patch("pypck.connection.PchkConnectionManager.async_connect"),
patch("homeassistant.components.lcn.async_setup", return_value=True),
patch("homeassistant.components.lcn.async_setup_entry", return_value=True),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_RECONFIGURE,
"entry_id": entry.entry_id,
},
data=CONFIG_DATA.copy(),
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
entry = hass.config_entries.async_get_entry(entry.entry_id)
assert entry.title == CONNECTION_DATA[CONF_HOST]
assert entry.data == {**old_entry_data, **CONFIG_DATA}
@pytest.mark.parametrize(
("error", "errors"),
[
(PchkAuthenticationError, {CONF_BASE: "authentication_error"}),
(PchkLicenseError, {CONF_BASE: "license_error"}),
(TimeoutError, {CONF_BASE: "connection_refused"}),
],
)
async def test_step_reconfigure_error(hass, entry, error, errors):
"""Test for error in reconfigure step is handled correctly."""
entry.add_to_hass(hass)
with patch(
"pypck.connection.PchkConnectionManager.async_connect", side_effect=error
):
data = {**CONNECTION_DATA, CONF_HOST: "pchk"}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_RECONFIGURE,
"entry_id": entry.entry_id,
},
data=data,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == errors
async def test_validate_connection():
"""Test the connection validation."""
data = CONNECTION_DATA.copy()
with (
patch("pypck.connection.PchkConnectionManager.async_connect") as async_connect,
patch("pypck.connection.PchkConnectionManager.async_close") as async_close,
):
result = await validate_connection(data=data)
assert async_connect.is_called
assert async_close.is_called
assert result is None

3
tests/components/lcn/test_init.py
View File

@ -1,6 +1,6 @@
"""Test init of LCN integration."""
from unittest.mock import patch
from unittest.mock import Mock, patch
from pypck.connection import (
PchkAuthenticationError,
@ -31,6 +31,7 @@ async def test_async_setup_entry(hass: HomeAssistant, entry, lcn_connection) ->
async def test_async_setup_multiple_entries(hass: HomeAssistant, entry, entry2) -> None:
"""Test a successful setup and unload of multiple entries."""
hass.http = Mock()
with patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager):
for config_entry in (entry, entry2):
config_entry.add_to_hass(hass)

303
tests/components/lcn/test_websocket.py Normal file
View File

@ -0,0 +1,303 @@
"""LCN Websocket Tests."""
from pypck.lcn_addr import LcnAddr
import pytest
from homeassistant.components.lcn.const import CONF_DOMAIN_DATA
from homeassistant.components.lcn.helpers import get_device_config, get_resource
from homeassistant.const import (
CONF_ADDRESS,
CONF_DEVICES,
CONF_DOMAIN,
CONF_ENTITIES,
CONF_NAME,
CONF_RESOURCE,
CONF_TYPE,
)
from homeassistant.core import HomeAssistant
from tests.typing import WebSocketGenerator
DEVICES_PAYLOAD = {CONF_TYPE: "lcn/devices", "entry_id": ""}
ENTITIES_PAYLOAD = {
CONF_TYPE: "lcn/entities",
"entry_id": "",
}
SCAN_PAYLOAD = {CONF_TYPE: "lcn/devices/scan", "entry_id": ""}
DEVICES_ADD_PAYLOAD = {
CONF_TYPE: "lcn/devices/add",
"entry_id": "",
CONF_ADDRESS: (0, 10, False),
}
DEVICES_DELETE_PAYLOAD = {
CONF_TYPE: "lcn/devices/delete",
"entry_id": "",
CONF_ADDRESS: (0, 7, False),
}
ENTITIES_ADD_PAYLOAD = {
CONF_TYPE: "lcn/entities/add",
"entry_id": "",
CONF_ADDRESS: (0, 7, False),
CONF_NAME: "test_switch",
CONF_DOMAIN: "switch",
CONF_DOMAIN_DATA: {"output": "RELAY5"},
}
ENTITIES_DELETE_PAYLOAD = {
CONF_TYPE: "lcn/entities/delete",
"entry_id": "",
CONF_ADDRESS: (0, 7, False),
CONF_DOMAIN: "switch",
CONF_RESOURCE: "relay1",
}
async def test_lcn_devices_command(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entry, lcn_connection
) -> None:
"""Test lcn/devices command."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({**DEVICES_PAYLOAD, "entry_id": entry.entry_id})
res = await client.receive_json()
assert res["success"], res
assert len(res["result"]) == len(entry.data[CONF_DEVICES])
assert all(
{**result, CONF_ADDRESS: tuple(result[CONF_ADDRESS])}
in entry.data[CONF_DEVICES]
for result in res["result"]
)
@pytest.mark.parametrize(
"payload",
[
ENTITIES_PAYLOAD,
{**ENTITIES_PAYLOAD, CONF_ADDRESS: (0, 7, False)},
],
)
async def test_lcn_entities_command(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
entry,
lcn_connection,
payload,
) -> None:
"""Test lcn/entities command."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
**payload,
"entry_id": entry.entry_id,
}
)
res = await client.receive_json()
assert res["success"], res
entities = [
entity
for entity in entry.data[CONF_ENTITIES]
if CONF_ADDRESS not in payload or entity[CONF_ADDRESS] == payload[CONF_ADDRESS]
]
assert len(res["result"]) == len(entities)
assert all(
{**result, CONF_ADDRESS: tuple(result[CONF_ADDRESS])} in entities
for result in res["result"]
)
async def test_lcn_devices_scan_command(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entry, lcn_connection
) -> None:
"""Test lcn/devices/scan command."""
# add new module which is not stored in config_entry
lcn_connection.get_address_conn(LcnAddr(0, 10, False))
client = await hass_ws_client(hass)
await client.send_json_auto_id({**SCAN_PAYLOAD, "entry_id": entry.entry_id})
res = await client.receive_json()
assert res["success"], res
lcn_connection.scan_modules.assert_awaited()
assert len(res["result"]) == len(entry.data[CONF_DEVICES])
assert all(
{**result, CONF_ADDRESS: tuple(result[CONF_ADDRESS])}
in entry.data[CONF_DEVICES]
for result in res["result"]
)
async def test_lcn_devices_add_command(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entry, lcn_connection
) -> None:
"""Test lcn/devices/add command."""
client = await hass_ws_client(hass)
assert get_device_config((0, 10, False), entry) is None
await client.send_json_auto_id({**DEVICES_ADD_PAYLOAD, "entry_id": entry.entry_id})
res = await client.receive_json()
assert res["success"], res
assert get_device_config((0, 10, False), entry)
async def test_lcn_devices_delete_command(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entry, lcn_connection
) -> None:
"""Test lcn/devices/delete command."""
client = await hass_ws_client(hass)
assert get_device_config((0, 7, False), entry)
await client.send_json_auto_id(
{**DEVICES_DELETE_PAYLOAD, "entry_id": entry.entry_id}
)
res = await client.receive_json()
assert res["success"], res
assert get_device_config((0, 7, False), entry) is None
async def test_lcn_entities_add_command(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entry, lcn_connection
) -> None:
"""Test lcn/entities/add command."""
client = await hass_ws_client(hass)
entity_config = {
key: ENTITIES_ADD_PAYLOAD[key]
for key in (CONF_ADDRESS, CONF_NAME, CONF_DOMAIN, CONF_DOMAIN_DATA)
}
resource = get_resource(
ENTITIES_ADD_PAYLOAD[CONF_DOMAIN], ENTITIES_ADD_PAYLOAD[CONF_DOMAIN_DATA]
).lower()
assert {**entity_config, CONF_RESOURCE: resource} not in entry.data[CONF_ENTITIES]
await client.send_json_auto_id({**ENTITIES_ADD_PAYLOAD, "entry_id": entry.entry_id})
res = await client.receive_json()
assert res["success"], res
assert {**entity_config, CONF_RESOURCE: resource} in entry.data[CONF_ENTITIES]
async def test_lcn_entities_delete_command(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entry, lcn_connection
) -> None:
"""Test lcn/entities/delete command."""
client = await hass_ws_client(hass)
assert (
len(
[
entity
for entity in entry.data[CONF_ENTITIES]
if entity[CONF_ADDRESS] == ENTITIES_DELETE_PAYLOAD[CONF_ADDRESS]
and entity[CONF_DOMAIN] == ENTITIES_DELETE_PAYLOAD[CONF_DOMAIN]
and entity[CONF_RESOURCE] == ENTITIES_DELETE_PAYLOAD[CONF_RESOURCE]
]
)
== 1
)
await client.send_json_auto_id(
{**ENTITIES_DELETE_PAYLOAD, "entry_id": entry.entry_id}
)
res = await client.receive_json()
assert res["success"], res
assert (
len(
[
entity
for entity in entry.data[CONF_ENTITIES]
if entity[CONF_ADDRESS] == ENTITIES_DELETE_PAYLOAD[CONF_ADDRESS]
and entity[CONF_DOMAIN] == ENTITIES_DELETE_PAYLOAD[CONF_DOMAIN]
and entity[CONF_RESOURCE] == ENTITIES_DELETE_PAYLOAD[CONF_RESOURCE]
]
)
== 0
)
@pytest.mark.parametrize(
("payload", "entity_id", "result"),
[
(DEVICES_PAYLOAD, "12345", False),
(ENTITIES_PAYLOAD, "12345", False),
(SCAN_PAYLOAD, "12345", False),
(DEVICES_ADD_PAYLOAD, "12345", False),
(DEVICES_DELETE_PAYLOAD, "12345", False),
(ENTITIES_ADD_PAYLOAD, "12345", False),
(ENTITIES_DELETE_PAYLOAD, "12345", False),
],
)
async def test_lcn_command_host_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
lcn_connection,
payload,
entity_id,
result,
) -> None:
"""Test lcn commands for unknown host."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({**payload, "entry_id": entity_id})
res = await client.receive_json()
assert res["success"], res
assert res["result"] == result
@pytest.mark.parametrize(
("payload", "address", "result"),
[
(DEVICES_ADD_PAYLOAD, (0, 7, False), False), # device already existing
(DEVICES_DELETE_PAYLOAD, (0, 42, False), False),
(ENTITIES_ADD_PAYLOAD, (0, 42, False), False),
(ENTITIES_DELETE_PAYLOAD, (0, 42, 0), False),
],
)
async def test_lcn_command_address_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
entry,
lcn_connection,
payload,
address,
result,
) -> None:
"""Test lcn commands for address error."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{**payload, "entry_id": entry.entry_id, CONF_ADDRESS: address}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"] == result
async def test_lcn_entities_add_existing_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
entry,
lcn_connection,
) -> None:
"""Test lcn commands for address error."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
**ENTITIES_ADD_PAYLOAD,
"entry_id": entry.entry_id,
CONF_DOMAIN_DATA: {"output": "RELAY1"},
}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"] is False