jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 04:07:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Create, update and delete KNX entities from UI / WS-commands (#104079)

Browse Source 
* knx entity CRUD - initial commit - switch

* platform dependent schema

* coerce empty GA-lists to None

* read entity configuration from WS

* use entity_id instead of unique_id for lookup

* Add device support

* Rename KNXEntityStore to KNXConfigStore

* fix test after rename

* Send schema options for creating / editing entities

* Return entity_id after entity creation

* remove device_class config in favour of more-info-dialog settings

* refactor group address schema for custom selector

* Rename GA keys and remove invalid keys from schema

* fix rebase

* Fix deleting devices and their entities

* Validate entity schema in extra step - return validation infos

* Use exception to signal validation error; return validated data

* Forward validation result when editing entities

* Get proper validation error message for optional GAs

* Add entity validation only WS command

* use ulid instead of uuid

* Fix error handling for edit unknown entity

* Remove unused optional group address sets from validated schema

* Add optional dpt field for ga_schema

* Move knx config things to sub-key

* Add light platform

* async_forward_entry_setups only once

* Test crate and remove devices

* Test removing entities of a removed device

* Test entity creation and storage

* Test deleting entities

* Test unsuccessful entity creation

* Test updating entity data

* Test get entity config

* Test validate entity

* Update entity data by entity_id instead of unique_id

* Remove unnecessary uid unique check

* remove schema_options

* test fixture for entity creation

* clean up group address schema

class can be used to add custom serializer later

* Revert: Add light platfrom

* remove unused optional_ga_schema

* Test GASelector

* lint tests

* Review

* group entities before adding

* fix / ignore mypy

* always has_entity_name

* Entity name: check for empty string when no device

* use constants instead of strings in schema

* Fix mypy errors for voluptuous schemas

---------

Co-authored-by: Marc Mueller <30130371+cdce8p@users.noreply.github.com>
This commit is contained in:
Matthias Alphart 2024-07-21 20:01:48 +02:00 committed by GitHub
parent 890b54e36f
commit d418a40856
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
24 changed files with 1557 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
homeassistant/components/knx/__init__.py
View File

@ -31,6 +31,7 @@ from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.helpers.reload import async_integration_yaml_config
from homeassistant.helpers.storage import STORAGE_DIR
from homeassistant.helpers.typing import ConfigType
@ -62,7 +63,8 @@ from .const import (
DATA_KNX_CONFIG,
DOMAIN,
KNX_ADDRESS,
SUPPORTED_PLATFORMS,
SUPPORTED_PLATFORMS_UI,
SUPPORTED_PLATFORMS_YAML,
TELEGRAM_LOG_DEFAULT,
)
from .device import KNXInterfaceDevice
@ -90,6 +92,7 @@ from .schema import (
WeatherSchema,
)
from .services import register_knx_services
from .storage.config_store import KNXConfigStore
from .telegrams import STORAGE_KEY as TELEGRAMS_STORAGE_KEY, Telegrams
from .websocket import register_panel
@ -190,10 +193,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
knx_module.exposures.append(
create_knx_exposure(hass, knx_module.xknx, expose_config)
)
# always forward sensor for system entities (telegram counter, etc.)
platforms = {platform for platform in SUPPORTED_PLATFORMS if platform in config}
platforms.add(Platform.SENSOR)
await hass.config_entries.async_forward_entry_setups(entry, platforms)
await hass.config_entries.async_forward_entry_setups(
entry,
{
Platform.SENSOR, # always forward sensor for system entities (telegram counter, etc.)
*SUPPORTED_PLATFORMS_UI, # forward all platforms that support UI entity management
*{ # forward yaml-only managed platforms on demand
platform for platform in SUPPORTED_PLATFORMS_YAML if platform in config
},
},
)
# set up notify service for backwards compatibility - remove 2024.11
if NotifySchema.PLATFORM in config:
@ -220,15 +229,15 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
unload_ok = await hass.config_entries.async_unload_platforms(
entry,
[
{
Platform.SENSOR, # always unload system entities (telegram counter, etc.)
*[
*SUPPORTED_PLATFORMS_UI, # unload all platforms that support UI entity management
*{ # unload yaml-only managed platforms if configured
platform
for platform in SUPPORTED_PLATFORMS
for platform in SUPPORTED_PLATFORMS_YAML
if platform in hass.data[DATA_KNX_CONFIG]
and platform is not Platform.SENSOR
],
],
},
},
)
if unload_ok:
await knx_module.stop()
@ -263,6 +272,22 @@ async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
await hass.async_add_executor_job(remove_files, storage_dir, knxkeys_filename)
async def async_remove_config_entry_device(
hass: HomeAssistant, config_entry: ConfigEntry, device_entry: DeviceEntry
) -> bool:
"""Remove a config entry from a device."""
knx_module: KNXModule = hass.data[DOMAIN]
if not device_entry.identifiers.isdisjoint(
knx_module.interface_device.device_info["identifiers"]
):
# can not remove interface device
return False
for entity in knx_module.config_store.get_entity_entries():
if entity.device_id == device_entry.id:
await knx_module.config_store.delete_entity(entity.entity_id)
return True
class KNXModule:
"""Representation of KNX Object."""
@ -278,6 +303,7 @@ class KNXModule:
self.entry = entry
self.project = KNXProject(hass=hass, entry=entry)
self.config_store = KNXConfigStore(hass=hass, entry=entry)
self.xknx = XKNX(
connection_config=self.connection_config(),
@ -309,6 +335,7 @@ class KNXModule:
async def start(self) -> None:
"""Start XKNX object. Connect to tunneling or Routing device."""
await self.project.load_project()
await self.config_store.load_data()
await self.telegrams.load_history()
await self.xknx.start()

2
homeassistant/components/knx/config_flow.py
View File

@ -62,7 +62,7 @@ from .const import (
TELEGRAM_LOG_MAX,
KNXConfigEntryData,
)
from .helpers.keyring import DEFAULT_KNX_KEYRING_FILENAME, save_uploaded_knxkeys_file
from .storage.keyring import DEFAULT_KNX_KEYRING_FILENAME, save_uploaded_knxkeys_file
from .validation import ia_validator, ip_v4_validator
CONF_KNX_GATEWAY: Final = "gateway"

13
homeassistant/components/knx/const.py
View File

@ -127,12 +127,13 @@ class KNXConfigEntryData(TypedDict, total=False):
class ColorTempModes(Enum):
"""Color temperature modes for config validation."""
ABSOLUTE = "DPT-7.600"
ABSOLUTE_FLOAT = "DPT-9"
RELATIVE = "DPT-5.001"
# YAML uses Enum.name (with vol.Upper), UI uses Enum.value for lookup
ABSOLUTE = "7.600"
ABSOLUTE_FLOAT = "9"
RELATIVE = "5.001"
SUPPORTED_PLATFORMS: Final = [
SUPPORTED_PLATFORMS_YAML: Final = {
Platform.BINARY_SENSOR,
Platform.BUTTON,
Platform.CLIMATE,
@ -150,7 +151,9 @@ SUPPORTED_PLATFORMS: Final = [
Platform.TEXT,
Platform.TIME,
Platform.WEATHER,
]
}
SUPPORTED_PLATFORMS_UI: Final = {Platform.SWITCH}
# Map KNX controller modes to HA modes. This list might not be complete.
CONTROLLER_MODES: Final = {

0
homeassistant/components/knx/helpers/__init__.py → homeassistant/components/knx/storage/__init__.py
View File

145
homeassistant/components/knx/storage/config_store.py Normal file
View File

@ -0,0 +1,145 @@
"""KNX entity configuration store."""
from collections.abc import Callable
import logging
from typing import TYPE_CHECKING, Any, Final, TypedDict
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PLATFORM, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.storage import Store
from homeassistant.util.ulid import ulid_now
from ..const import DOMAIN
from .const import CONF_DATA
if TYPE_CHECKING:
from ..knx_entity import KnxEntity
_LOGGER = logging.getLogger(__name__)
STORAGE_VERSION: Final = 1
STORAGE_KEY: Final = f"{DOMAIN}/config_store.json"
KNXPlatformStoreModel = dict[str, dict[str, Any]] # unique_id: configuration
KNXEntityStoreModel = dict[
str, KNXPlatformStoreModel
] # platform: KNXPlatformStoreModel
class KNXConfigStoreModel(TypedDict):
"""Represent KNX configuration store data."""
entities: KNXEntityStoreModel
class KNXConfigStore:
"""Manage KNX config store data."""
def __init__(
self,
hass: HomeAssistant,
entry: ConfigEntry,
) -> None:
"""Initialize config store."""
self.hass = hass
self._store = Store[KNXConfigStoreModel](hass, STORAGE_VERSION, STORAGE_KEY)
self.data = KNXConfigStoreModel(entities={})
# entities and async_add_entity are filled by platform setups
self.entities: dict[str, KnxEntity] = {} # unique_id as key
self.async_add_entity: dict[
Platform, Callable[[str, dict[str, Any]], None]
] = {}
async def load_data(self) -> None:
"""Load config store data from storage."""
if data := await self._store.async_load():
self.data = KNXConfigStoreModel(**data)
_LOGGER.debug(
"Loaded KNX config data from storage. %s entity platforms",
len(self.data["entities"]),
)
async def create_entity(
self, platform: Platform, data: dict[str, Any]
) -> str | None:
"""Create a new entity."""
if platform not in self.async_add_entity:
raise ConfigStoreException(f"Entity platform not ready: {platform}")
unique_id = f"knx_es_{ulid_now()}"
self.async_add_entity[platform](unique_id, data)
# store data after entity was added to be sure config didn't raise exceptions
self.data["entities"].setdefault(platform, {})[unique_id] = data
await self._store.async_save(self.data)
entity_registry = er.async_get(self.hass)
return entity_registry.async_get_entity_id(platform, DOMAIN, unique_id)
@callback
def get_entity_config(self, entity_id: str) -> dict[str, Any]:
"""Return KNX entity configuration."""
entity_registry = er.async_get(self.hass)
if (entry := entity_registry.async_get(entity_id)) is None:
raise ConfigStoreException(f"Entity not found: {entity_id}")
try:
return {
CONF_PLATFORM: entry.domain,
CONF_DATA: self.data["entities"][entry.domain][entry.unique_id],
}
except KeyError as err:
raise ConfigStoreException(f"Entity data not found: {entity_id}") from err
async def update_entity(
self, platform: Platform, entity_id: str, data: dict[str, Any]
) -> None:
"""Update an existing entity."""
if platform not in self.async_add_entity:
raise ConfigStoreException(f"Entity platform not ready: {platform}")
entity_registry = er.async_get(self.hass)
if (entry := entity_registry.async_get(entity_id)) is None:
raise ConfigStoreException(f"Entity not found: {entity_id}")
unique_id = entry.unique_id
if (
platform not in self.data["entities"]
or unique_id not in self.data["entities"][platform]
):
raise ConfigStoreException(
f"Entity not found in storage: {entity_id} - {unique_id}"
)
await self.entities.pop(unique_id).async_remove()
self.async_add_entity[platform](unique_id, data)
# store data after entity is added to make sure config doesn't raise exceptions
self.data["entities"][platform][unique_id] = data
await self._store.async_save(self.data)
async def delete_entity(self, entity_id: str) -> None:
"""Delete an existing entity."""
entity_registry = er.async_get(self.hass)
if (entry := entity_registry.async_get(entity_id)) is None:
raise ConfigStoreException(f"Entity not found: {entity_id}")
try:
del self.data["entities"][entry.domain][entry.unique_id]
except KeyError as err:
raise ConfigStoreException(
f"Entity not found in {entry.domain}: {entry.unique_id}"
) from err
try:
del self.entities[entry.unique_id]
except KeyError:
_LOGGER.warning("Entity not initialized when deleted: %s", entity_id)
entity_registry.async_remove(entity_id)
await self._store.async_save(self.data)
def get_entity_entries(self) -> list[er.RegistryEntry]:
"""Get entity_ids of all configured entities by platform."""
return [
entity.registry_entry
for entity in self.entities.values()
if entity.registry_entry is not None
]
class ConfigStoreException(Exception):
"""KNX config store exception."""

14
homeassistant/components/knx/storage/const.py Normal file
View File

@ -0,0 +1,14 @@
"""Constants used in KNX config store."""
from typing import Final
CONF_DATA: Final = "data"
CONF_ENTITY: Final = "entity"
CONF_DEVICE_INFO: Final = "device_info"
CONF_GA_WRITE: Final = "write"
CONF_GA_STATE: Final = "state"
CONF_GA_PASSIVE: Final = "passive"
CONF_DPT: Final = "dpt"
CONF_GA_SWITCH: Final = "ga_switch"

94
homeassistant/components/knx/storage/entity_store_schema.py Normal file
View File

@ -0,0 +1,94 @@
"""KNX entity store schema."""
import voluptuous as vol
from homeassistant.const import (
CONF_ENTITY_CATEGORY,
CONF_ENTITY_ID,
CONF_NAME,
CONF_PLATFORM,
Platform,
)
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity import ENTITY_CATEGORIES_SCHEMA
from homeassistant.helpers.typing import VolDictType, VolSchemaType
from ..const import (
CONF_INVERT,
CONF_RESPOND_TO_READ,
CONF_SYNC_STATE,
DOMAIN,
SUPPORTED_PLATFORMS_UI,
)
from ..validation import sync_state_validator
from .const import CONF_DATA, CONF_DEVICE_INFO, CONF_ENTITY, CONF_GA_SWITCH
from .knx_selector import GASelector
BASE_ENTITY_SCHEMA = vol.All(
{
vol.Optional(CONF_NAME, default=None): vol.Maybe(str),
vol.Optional(CONF_DEVICE_INFO, default=None): vol.Maybe(str),
vol.Optional(CONF_ENTITY_CATEGORY, default=None): vol.Any(
ENTITY_CATEGORIES_SCHEMA, vol.SetTo(None)
),
},
vol.Any(
vol.Schema(
{
vol.Required(CONF_NAME): vol.All(str, vol.IsTrue()),
},
extra=vol.ALLOW_EXTRA,
),
vol.Schema(
{
vol.Required(CONF_DEVICE_INFO): str,
},
extra=vol.ALLOW_EXTRA,
),
msg="One of `Device` or `Name` is required",
),
)
SWITCH_SCHEMA = vol.Schema(
{
vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA,
vol.Required(DOMAIN): {
vol.Optional(CONF_INVERT, default=False): bool,
vol.Required(CONF_GA_SWITCH): GASelector(write_required=True),
vol.Optional(CONF_RESPOND_TO_READ, default=False): bool,
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
},
}
)
ENTITY_STORE_DATA_SCHEMA: VolSchemaType = vol.All(
vol.Schema(
{
vol.Required(CONF_PLATFORM): vol.All(
vol.Coerce(Platform),
vol.In(SUPPORTED_PLATFORMS_UI),
),
vol.Required(CONF_DATA): dict,
},
extra=vol.ALLOW_EXTRA,
),
cv.key_value_schemas(
CONF_PLATFORM,
{
Platform.SWITCH: vol.Schema(
{vol.Required(CONF_DATA): SWITCH_SCHEMA}, extra=vol.ALLOW_EXTRA
),
},
),
)
CREATE_ENTITY_BASE_SCHEMA: VolDictType = {
vol.Required(CONF_PLATFORM): str,
vol.Required(CONF_DATA): dict, # validated by ENTITY_STORE_DATA_SCHEMA for platform
}
UPDATE_ENTITY_BASE_SCHEMA = {
vol.Required(CONF_ENTITY_ID): str,
**CREATE_ENTITY_BASE_SCHEMA,
}

69
homeassistant/components/knx/storage/entity_store_validation.py Normal file
View File

@ -0,0 +1,69 @@
"""KNX Entity Store Validation."""
from typing import Literal, TypedDict
import voluptuous as vol
from .entity_store_schema import ENTITY_STORE_DATA_SCHEMA
class _ErrorDescription(TypedDict):
path: list[str] | None
error_message: str
error_class: str
class EntityStoreValidationError(TypedDict):
"""Negative entity store validation result."""
success: Literal[False]
error_base: str
errors: list[_ErrorDescription]
class EntityStoreValidationSuccess(TypedDict):
"""Positive entity store validation result."""
success: Literal[True]
entity_id: str | None
def parse_invalid(exc: vol.Invalid) -> _ErrorDescription:
"""Parse a vol.Invalid exception."""
return _ErrorDescription(
path=[str(path) for path in exc.path], # exc.path: str | vol.Required
error_message=exc.msg,
error_class=type(exc).__name__,
)
def validate_entity_data(entity_data: dict) -> dict:
"""Validate entity data. Return validated data or raise EntityStoreValidationException."""
try:
# return so defaults are applied
return ENTITY_STORE_DATA_SCHEMA(entity_data) # type: ignore[no-any-return]
except vol.MultipleInvalid as exc:
raise EntityStoreValidationException(
validation_error={
"success": False,
"error_base": str(exc),
"errors": [parse_invalid(invalid) for invalid in exc.errors],
}
) from exc
except vol.Invalid as exc:
raise EntityStoreValidationException(
validation_error={
"success": False,
"error_base": str(exc),
"errors": [parse_invalid(exc)],
}
) from exc
class EntityStoreValidationException(Exception):
"""Entity store validation exception."""
def __init__(self, validation_error: EntityStoreValidationError) -> None:
"""Initialize."""
super().__init__(validation_error)
self.validation_error = validation_error

0
homeassistant/components/knx/helpers/keyring.py → homeassistant/components/knx/storage/keyring.py
View File

81
homeassistant/components/knx/storage/knx_selector.py Normal file
View File

@ -0,0 +1,81 @@
"""Selectors for KNX."""
from enum import Enum
from typing import Any
import voluptuous as vol
from ..validation import ga_validator, maybe_ga_validator
from .const import CONF_DPT, CONF_GA_PASSIVE, CONF_GA_STATE, CONF_GA_WRITE
class GASelector:
"""Selector for a KNX group address structure."""
schema: vol.Schema
def __init__(
self,
write: bool = True,
state: bool = True,
passive: bool = True,
write_required: bool = False,
state_required: bool = False,
dpt: type[Enum] | None = None,
) -> None:
"""Initialize the group address selector."""
self.write = write
self.state = state
self.passive = passive
self.write_required = write_required
self.state_required = state_required
self.dpt = dpt
self.schema = self.build_schema()
def __call__(self, data: Any) -> Any:
"""Validate the passed data."""
return self.schema(data)
def build_schema(self) -> vol.Schema:
"""Create the schema based on configuration."""
schema: dict[vol.Marker, Any] = {} # will be modified in-place
self._add_group_addresses(schema)
self._add_passive(schema)
self._add_dpt(schema)
return vol.Schema(schema)
def _add_group_addresses(self, schema: dict[vol.Marker, Any]) -> None:
"""Add basic group address items to the schema."""
def add_ga_item(key: str, allowed: bool, required: bool) -> None:
"""Add a group address item validator to the schema."""
if not allowed:
schema[vol.Remove(key)] = object
return
if required:
schema[vol.Required(key)] = ga_validator
else:
schema[vol.Optional(key, default=None)] = maybe_ga_validator
add_ga_item(CONF_GA_WRITE, self.write, self.write_required)
add_ga_item(CONF_GA_STATE, self.state, self.state_required)
def _add_passive(self, schema: dict[vol.Marker, Any]) -> None:
"""Add passive group addresses validator to the schema."""
if self.passive:
schema[vol.Optional(CONF_GA_PASSIVE, default=list)] = vol.Any(
[ga_validator],
vol.All( # Coerce `None` to an empty list if passive is allowed
vol.IsFalse(), vol.SetTo(list)
),
)
else:
schema[vol.Remove(CONF_GA_PASSIVE)] = object
def _add_dpt(self, schema: dict[vol.Marker, Any]) -> None:
"""Add DPT validator to the schema."""
if self.dpt is not None:
schema[vol.Required(CONF_DPT)] = vol.In([item.value for item in self.dpt])
else:
schema[vol.Remove(CONF_DPT)] = object

114
homeassistant/components/knx/switch.py
View File

@ -18,14 +18,30 @@ from homeassistant.const import (
STATE_UNKNOWN,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType
from .const import CONF_RESPOND_TO_READ, DATA_KNX_CONFIG, DOMAIN, KNX_ADDRESS
from . import KNXModule
from .const import (
CONF_INVERT,
CONF_RESPOND_TO_READ,
DATA_KNX_CONFIG,
DOMAIN,
KNX_ADDRESS,
)
from .knx_entity import KnxEntity
from .schema import SwitchSchema
from .storage.const import (
CONF_DEVICE_INFO,
CONF_ENTITY,
CONF_GA_PASSIVE,
CONF_GA_STATE,
CONF_GA_SWITCH,
CONF_GA_WRITE,
)
async def async_setup_entry(
@ -34,33 +50,35 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up switch(es) for KNX platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][Platform.SWITCH]
knx_module: KNXModule = hass.data[DOMAIN]
async_add_entities(KNXSwitch(xknx, entity_config) for entity_config in config)
entities: list[KnxEntity] = []
if yaml_config := hass.data[DATA_KNX_CONFIG].get(Platform.SWITCH):
entities.extend(
KnxYamlSwitch(knx_module.xknx, entity_config)
for entity_config in yaml_config
)
if ui_config := knx_module.config_store.data["entities"].get(Platform.SWITCH):
entities.extend(
KnxUiSwitch(knx_module, unique_id, config)
for unique_id, config in ui_config.items()
)
if entities:
async_add_entities(entities)
@callback
def add_new_ui_switch(unique_id: str, config: dict[str, Any]) -> None:
"""Add KNX entity at runtime."""
async_add_entities([KnxUiSwitch(knx_module, unique_id, config)])
knx_module.config_store.async_add_entity[Platform.SWITCH] = add_new_ui_switch
class KNXSwitch(KnxEntity, SwitchEntity, RestoreEntity):
"""Representation of a KNX switch."""
class _KnxSwitch(KnxEntity, SwitchEntity, RestoreEntity):
"""Base class for a KNX switch."""
_device: XknxSwitch
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
"""Initialize of KNX switch."""
super().__init__(
device=XknxSwitch(
xknx,
name=config[CONF_NAME],
group_address=config[KNX_ADDRESS],
group_address_state=config.get(SwitchSchema.CONF_STATE_ADDRESS),
respond_to_read=config[CONF_RESPOND_TO_READ],
invert=config[SwitchSchema.CONF_INVERT],
)
)
self._attr_entity_category = config.get(CONF_ENTITY_CATEGORY)
self._attr_device_class = config.get(CONF_DEVICE_CLASS)
self._attr_unique_id = str(self._device.switch.group_address)
async def async_added_to_hass(self) -> None:
"""Restore last state."""
await super().async_added_to_hass()
@ -82,3 +100,53 @@ class KNXSwitch(KnxEntity, SwitchEntity, RestoreEntity):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
await self._device.set_off()
class KnxYamlSwitch(_KnxSwitch):
"""Representation of a KNX switch configured from YAML."""
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
"""Initialize of KNX switch."""
super().__init__(
device=XknxSwitch(
xknx,
name=config[CONF_NAME],
group_address=config[KNX_ADDRESS],
group_address_state=config.get(SwitchSchema.CONF_STATE_ADDRESS),
respond_to_read=config[CONF_RESPOND_TO_READ],
invert=config[SwitchSchema.CONF_INVERT],
)
)
self._attr_entity_category = config.get(CONF_ENTITY_CATEGORY)
self._attr_device_class = config.get(CONF_DEVICE_CLASS)
self._attr_unique_id = str(self._device.switch.group_address)
class KnxUiSwitch(_KnxSwitch):
"""Representation of a KNX switch configured from UI."""
_attr_has_entity_name = True
def __init__(
self, knx_module: KNXModule, unique_id: str, config: dict[str, Any]
) -> None:
"""Initialize of KNX switch."""
super().__init__(
device=XknxSwitch(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
group_address=config[DOMAIN][CONF_GA_SWITCH][CONF_GA_WRITE],
group_address_state=[
config[DOMAIN][CONF_GA_SWITCH][CONF_GA_STATE],
*config[DOMAIN][CONF_GA_SWITCH][CONF_GA_PASSIVE],
],
respond_to_read=config[DOMAIN][CONF_RESPOND_TO_READ],
invert=config[DOMAIN][CONF_INVERT],
)
)
self._attr_entity_category = config[CONF_ENTITY][CONF_ENTITY_CATEGORY]
self._attr_unique_id = unique_id
if device_info := config[CONF_ENTITY].get(CONF_DEVICE_INFO):
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, device_info)})
knx_module.config_store.entities[unique_id] = self

15
homeassistant/components/knx/validation.py
View File

@ -50,12 +50,27 @@ def ga_validator(value: Any) -> str | int:
return value
def maybe_ga_validator(value: Any) -> str | int | None:
"""Validate a group address or None."""
# this is a version of vol.Maybe(ga_validator) that delivers the
# error message of ga_validator if validation fails.
return ga_validator(value) if value is not None else None
ga_list_validator = vol.All(
cv.ensure_list,
[ga_validator],
vol.IsTrue("value must be a group address or a list containing group addresses"),
)
ga_list_validator_optional = vol.Maybe(
vol.All(
cv.ensure_list,
[ga_validator],
vol.Any(vol.IsTrue(), vol.SetTo(None)), # avoid empty lists -> None
)
)
ia_validator = vol.Any(
vol.All(str, str.strip, cv.matches_regex(IndividualAddress.ADDRESS_RE.pattern)),
vol.All(vol.Coerce(int), vol.Range(min=1, max=65535)),

220
homeassistant/components/knx/websocket.py
View File

@ -10,9 +10,24 @@ from xknxproject.exceptions import XknxProjectException
from homeassistant.components import panel_custom, websocket_api
from homeassistant.components.http import StaticPathConfig
from homeassistant.const import CONF_ENTITY_ID, CONF_PLATFORM
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import UNDEFINED
from homeassistant.util.ulid import ulid_now
from .const import DOMAIN
from .storage.config_store import ConfigStoreException
from .storage.const import CONF_DATA
from .storage.entity_store_schema import (
CREATE_ENTITY_BASE_SCHEMA,
UPDATE_ENTITY_BASE_SCHEMA,
)
from .storage.entity_store_validation import (
EntityStoreValidationException,
EntityStoreValidationSuccess,
validate_entity_data,
)
from .telegrams import TelegramDict
if TYPE_CHECKING:
@ -30,6 +45,13 @@ async def register_panel(hass: HomeAssistant) -> None:
websocket_api.async_register_command(hass, ws_group_monitor_info)
websocket_api.async_register_command(hass, ws_subscribe_telegram)
websocket_api.async_register_command(hass, ws_get_knx_project)
websocket_api.async_register_command(hass, ws_validate_entity)
websocket_api.async_register_command(hass, ws_create_entity)
websocket_api.async_register_command(hass, ws_update_entity)
websocket_api.async_register_command(hass, ws_delete_entity)
websocket_api.async_register_command(hass, ws_get_entity_config)
websocket_api.async_register_command(hass, ws_get_entity_entries)
websocket_api.async_register_command(hass, ws_create_device)
if DOMAIN not in hass.data.get("frontend_panels", {}):
await hass.http.async_register_static_paths(
@ -213,3 +235,201 @@ def ws_subscribe_telegram(
name="KNX GroupMonitor subscription",
)
connection.send_result(msg["id"])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/validate_entity",
**CREATE_ENTITY_BASE_SCHEMA,
}
)
@callback
def ws_validate_entity(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Validate entity data."""
try:
validate_entity_data(msg)
except EntityStoreValidationException as exc:
connection.send_result(msg["id"], exc.validation_error)
return
connection.send_result(
msg["id"], EntityStoreValidationSuccess(success=True, entity_id=None)
)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/create_entity",
**CREATE_ENTITY_BASE_SCHEMA,
}
)
@websocket_api.async_response
async def ws_create_entity(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Create entity in entity store and load it."""
try:
validated_data = validate_entity_data(msg)
except EntityStoreValidationException as exc:
connection.send_result(msg["id"], exc.validation_error)
return
knx: KNXModule = hass.data[DOMAIN]
try:
entity_id = await knx.config_store.create_entity(
# use validation result so defaults are applied
validated_data[CONF_PLATFORM],
validated_data[CONF_DATA],
)
except ConfigStoreException as err:
connection.send_error(
msg["id"], websocket_api.const.ERR_HOME_ASSISTANT_ERROR, str(err)
)
return
connection.send_result(
msg["id"], EntityStoreValidationSuccess(success=True, entity_id=entity_id)
)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/update_entity",
**UPDATE_ENTITY_BASE_SCHEMA,
}
)
@websocket_api.async_response
async def ws_update_entity(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Update entity in entity store and reload it."""
try:
validated_data = validate_entity_data(msg)
except EntityStoreValidationException as exc:
connection.send_result(msg["id"], exc.validation_error)
return
knx: KNXModule = hass.data[DOMAIN]
try:
await knx.config_store.update_entity(
validated_data[CONF_PLATFORM],
validated_data[CONF_ENTITY_ID],
validated_data[CONF_DATA],
)
except ConfigStoreException as err:
connection.send_error(
msg["id"], websocket_api.const.ERR_HOME_ASSISTANT_ERROR, str(err)
)
return
connection.send_result(
msg["id"], EntityStoreValidationSuccess(success=True, entity_id=None)
)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/delete_entity",
vol.Required(CONF_ENTITY_ID): str,
}
)
@websocket_api.async_response
async def ws_delete_entity(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Delete entity from entity store and remove it."""
knx: KNXModule = hass.data[DOMAIN]
try:
await knx.config_store.delete_entity(msg[CONF_ENTITY_ID])
except ConfigStoreException as err:
connection.send_error(
msg["id"], websocket_api.const.ERR_HOME_ASSISTANT_ERROR, str(err)
)
return
connection.send_result(msg["id"])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/get_entity_entries",
}
)
@callback
def ws_get_entity_entries(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Get entities configured from entity store."""
knx: KNXModule = hass.data[DOMAIN]
entity_entries = [
entry.extended_dict for entry in knx.config_store.get_entity_entries()
]
connection.send_result(msg["id"], entity_entries)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/get_entity_config",
vol.Required(CONF_ENTITY_ID): str,
}
)
@callback
def ws_get_entity_config(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Get entity configuration from entity store."""
knx: KNXModule = hass.data[DOMAIN]
try:
config_info = knx.config_store.get_entity_config(msg[CONF_ENTITY_ID])
except ConfigStoreException as err:
connection.send_error(
msg["id"], websocket_api.const.ERR_HOME_ASSISTANT_ERROR, str(err)
)
return
connection.send_result(msg["id"], config_info)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/create_device",
vol.Required("name"): str,
vol.Optional("area_id"): str,
}
)
@callback
def ws_create_device(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Create a new KNX device."""
knx: KNXModule = hass.data[DOMAIN]
identifier = f"knx_vdev_{ulid_now()}"
device_registry = dr.async_get(hass)
_device = device_registry.async_get_or_create(
config_entry_id=knx.entry.entry_id,
manufacturer="KNX",
name=msg["name"],
identifiers={(DOMAIN, identifier)},
)
device_registry.async_update_device(
_device.id,
area_id=msg.get("area_id") or UNDEFINED,
configuration_url=f"homeassistant://knx/entities/view?device_id={_device.id}",
)
connection.send_result(msg["id"], _device.dict_repr)

3
tests/components/knx/README.md
View File

@ -24,9 +24,10 @@ All outgoing telegrams are pushed to an assertion queue. Assert them in order th
Asserts that no telegram was sent (assertion queue is empty).
- `knx.assert_telegram_count(count: int)`
Asserts that `count` telegrams were sent.
- `knx.assert_read(group_address: str)`
- `knx.assert_read(group_address: str, response: int | tuple[int, ...] | None = None)`
Asserts that a GroupValueRead telegram was sent to `group_address`.
The telegram will be removed from the assertion queue.
Optionally inject incoming GroupValueResponse telegram after reception to clear the value reader waiting task. This can also be done manually with `knx.receive_response`.
- `knx.assert_response(group_address: str, payload: int | tuple[int, ...])`
Asserts that a GroupValueResponse telegram with `payload` was sent to `group_address`.
The telegram will be removed from the assertion queue.

6
tests/components/knx/__init__.py
View File

@ -1 +1,7 @@
"""Tests for the KNX integration."""
from collections.abc import Awaitable, Callable
from homeassistant.helpers import entity_registry as er
KnxEntityGenerator = Callable[..., Awaitable[er.RegistryEntry]]

75
tests/components/knx/conftest.py
View File

@ -3,7 +3,6 @@
from __future__ import annotations
import asyncio
import json
from typing import Any
from unittest.mock import DEFAULT, AsyncMock, Mock, patch
@ -30,13 +29,22 @@ from homeassistant.components.knx.const import (
DOMAIN as KNX_DOMAIN,
)
from homeassistant.components.knx.project import STORAGE_KEY as KNX_PROJECT_STORAGE_KEY
from homeassistant.components.knx.storage.config_store import (
STORAGE_KEY as KNX_CONFIG_STORAGE_KEY,
)
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, load_fixture
from . import KnxEntityGenerator
FIXTURE_PROJECT_DATA = json.loads(load_fixture("project.json", KNX_DOMAIN))
from tests.common import MockConfigEntry, load_json_object_fixture
from tests.typing import WebSocketGenerator
FIXTURE_PROJECT_DATA = load_json_object_fixture("project.json", KNX_DOMAIN)
FIXTURE_CONFIG_STORAGE_DATA = load_json_object_fixture("config_store.json", KNX_DOMAIN)
class KNXTestKit:
@ -166,9 +174,16 @@ class KNXTestKit:
telegram.payload.value.value == payload # type: ignore[attr-defined]
), f"Payload mismatch in {telegram} - Expected: {payload}"
async def assert_read(self, group_address: str) -> None:
"""Assert outgoing GroupValueRead telegram. One by one in timely order."""
async def assert_read(
self, group_address: str, response: int | tuple[int, ...] | None = None
) -> None:
"""Assert outgoing GroupValueRead telegram. One by one in timely order.
Optionally inject incoming GroupValueResponse telegram after reception.
"""
await self.assert_telegram(group_address, None, GroupValueRead)
if response is not None:
await self.receive_response(group_address, response)
async def assert_response(
self, group_address: str, payload: int | tuple[int, ...]
@ -280,3 +295,53 @@ def load_knxproj(hass_storage: dict[str, Any]) -> None:
"version": 1,
"data": FIXTURE_PROJECT_DATA,
}
@pytest.fixture
def load_config_store(hass_storage: dict[str, Any]) -> None:
"""Mock KNX config store data."""
hass_storage[KNX_CONFIG_STORAGE_KEY] = FIXTURE_CONFIG_STORAGE_DATA
@pytest.fixture
async def create_ui_entity(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
hass_storage: dict[str, Any],
) -> KnxEntityGenerator:
"""Return a helper to create a KNX entities via WS.
The KNX integration must be set up before using the helper.
"""
ws_client = await hass_ws_client(hass)
async def _create_ui_entity(
platform: Platform,
knx_data: dict[str, Any],
entity_data: dict[str, Any] | None = None,
) -> er.RegistryEntry:
"""Create a KNX entity from WS with given configuration."""
if entity_data is None:
entity_data = {"name": "Test"}
await ws_client.send_json_auto_id(
{
"type": "knx/create_entity",
"platform": platform,
"data": {
"entity": entity_data,
"knx": knx_data,
},
}
)
res = await ws_client.receive_json()
assert res["success"], res
assert res["result"]["success"] is True
entity_id = res["result"]["entity_id"]
entity = entity_registry.async_get(entity_id)
assert entity
return entity
return _create_ui_entity

29
tests/components/knx/fixtures/config_store.json Normal file
View File

@ -0,0 +1,29 @@
{
"version": 1,
"minor_version": 1,
"key": "knx/config_store.json",
"data": {
"entities": {
"switch": {
"knx_es_9d97829f47f1a2a3176a7c5b4216070c": {
"entity": {
"entity_category": null,
"name": "test",
"device_info": "knx_vdev_4c80a564f5fe5da701ed293966d6384d"
},
"knx": {
"ga_switch": {
"write": "1/1/45",
"state": "1/0/45",
"passive": []
},
"invert": false,
"sync_state": true,
"respond_to_read": false
}
}
},
"light": {}
}
}
}

4
tests/components/knx/test_config_flow.py
View File

@ -76,10 +76,10 @@ def patch_file_upload(return_value=FIXTURE_KEYRING, side_effect=None):
"""Patch file upload. Yields the Keyring instance (return_value)."""
with (
patch(
"homeassistant.components.knx.helpers.keyring.process_uploaded_file"
"homeassistant.components.knx.storage.keyring.process_uploaded_file"
) as file_upload_mock,
patch(
"homeassistant.components.knx.helpers.keyring.sync_load_keyring",
"homeassistant.components.knx.storage.keyring.sync_load_keyring",
return_value=return_value,
side_effect=side_effect,
),

412
tests/components/knx/test_config_store.py Normal file
View File

@ -0,0 +1,412 @@
"""Test KNX config store."""
from typing import Any
import pytest
from homeassistant.components.knx.storage.config_store import (
STORAGE_KEY as KNX_CONFIG_STORAGE_KEY,
)
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import KnxEntityGenerator
from .conftest import KNXTestKit
from tests.typing import WebSocketGenerator
async def test_create_entity(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
hass_storage: dict[str, Any],
create_ui_entity: KnxEntityGenerator,
) -> None:
"""Test entity creation."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
test_name = "Test no device"
test_entity = await create_ui_entity(
platform=Platform.SWITCH,
knx_data={"ga_switch": {"write": "1/2/3"}},
entity_data={"name": test_name},
)
# Test if entity is correctly stored in registry
await client.send_json_auto_id({"type": "knx/get_entity_entries"})
res = await client.receive_json()
assert res["success"], res
assert res["result"] == [
test_entity.extended_dict,
]
# Test if entity is correctly stored in config store
test_storage_data = next(
iter(
hass_storage[KNX_CONFIG_STORAGE_KEY]["data"]["entities"]["switch"].values()
)
)
assert test_storage_data == {
"entity": {
"name": test_name,
"device_info": None,
"entity_category": None,
},
"knx": {
"ga_switch": {"write": "1/2/3", "state": None, "passive": []},
"invert": False,
"respond_to_read": False,
"sync_state": True,
},
}
async def test_create_entity_error(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test unsuccessful entity creation."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
# create entity with invalid platform
await client.send_json_auto_id(
{
"type": "knx/create_entity",
"platform": "invalid_platform",
"data": {
"entity": {"name": "Test invalid platform"},
"knx": {"ga_switch": {"write": "1/2/3"}},
},
}
)
res = await client.receive_json()
assert res["success"], res
assert not res["result"]["success"]
assert res["result"]["errors"][0]["path"] == ["platform"]
assert res["result"]["error_base"].startswith("expected Platform or one of")
# create entity with unsupported platform
await client.send_json_auto_id(
{
"type": "knx/create_entity",
"platform": Platform.TTS, # "tts" is not a supported platform (and is unlikely to ever be)
"data": {
"entity": {"name": "Test invalid platform"},
"knx": {"ga_switch": {"write": "1/2/3"}},
},
}
)
res = await client.receive_json()
assert res["success"], res
assert not res["result"]["success"]
assert res["result"]["errors"][0]["path"] == ["platform"]
assert res["result"]["error_base"].startswith("value must be one of")
async def test_update_entity(
hass: HomeAssistant,
knx: KNXTestKit,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
hass_storage: dict[str, Any],
create_ui_entity: KnxEntityGenerator,
) -> None:
"""Test entity update."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
test_entity = await create_ui_entity(
platform=Platform.SWITCH,
knx_data={"ga_switch": {"write": "1/2/3"}},
entity_data={"name": "Test"},
)
test_entity_id = test_entity.entity_id
# update entity
new_name = "Updated name"
new_ga_switch_write = "4/5/6"
await client.send_json_auto_id(
{
"type": "knx/update_entity",
"platform": Platform.SWITCH,
"entity_id": test_entity_id,
"data": {
"entity": {"name": new_name},
"knx": {"ga_switch": {"write": new_ga_switch_write}},
},
}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"]["success"]
entity = entity_registry.async_get(test_entity_id)
assert entity
assert entity.original_name == new_name
assert (
hass_storage[KNX_CONFIG_STORAGE_KEY]["data"]["entities"]["switch"][
test_entity.unique_id
]["knx"]["ga_switch"]["write"]
== new_ga_switch_write
)
async def test_update_entity_error(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
create_ui_entity: KnxEntityGenerator,
) -> None:
"""Test entity update."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
test_entity = await create_ui_entity(
platform=Platform.SWITCH,
knx_data={"ga_switch": {"write": "1/2/3"}},
entity_data={"name": "Test"},
)
# update unsupported platform
new_name = "Updated name"
new_ga_switch_write = "4/5/6"
await client.send_json_auto_id(
{
"type": "knx/update_entity",
"platform": Platform.TTS,
"entity_id": test_entity.entity_id,
"data": {
"entity": {"name": new_name},
"knx": {"ga_switch": {"write": new_ga_switch_write}},
},
}
)
res = await client.receive_json()
assert res["success"], res
assert not res["result"]["success"]
assert res["result"]["errors"][0]["path"] == ["platform"]
assert res["result"]["error_base"].startswith("value must be one of")
# entity not found
await client.send_json_auto_id(
{
"type": "knx/update_entity",
"platform": Platform.SWITCH,
"entity_id": "non_existing_entity_id",
"data": {
"entity": {"name": new_name},
"knx": {"ga_switch": {"write": new_ga_switch_write}},
},
}
)
res = await client.receive_json()
assert not res["success"], res
assert res["error"]["code"] == "home_assistant_error"
assert res["error"]["message"].startswith("Entity not found:")
# entity not in storage
await client.send_json_auto_id(
{
"type": "knx/update_entity",
"platform": Platform.SWITCH,
# `sensor` isn't yet supported, but we only have sensor entities automatically
# created with no configuration - it doesn't ,atter for the test though
"entity_id": "sensor.knx_interface_individual_address",
"data": {
"entity": {"name": new_name},
"knx": {"ga_switch": {"write": new_ga_switch_write}},
},
}
)
res = await client.receive_json()
assert not res["success"], res
assert res["error"]["code"] == "home_assistant_error"
assert res["error"]["message"].startswith("Entity not found in storage")
async def test_delete_entity(
hass: HomeAssistant,
knx: KNXTestKit,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
hass_storage: dict[str, Any],
create_ui_entity: KnxEntityGenerator,
) -> None:
"""Test entity deletion."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
test_entity = await create_ui_entity(
platform=Platform.SWITCH,
knx_data={"ga_switch": {"write": "1/2/3"}},
entity_data={"name": "Test"},
)
test_entity_id = test_entity.entity_id
# delete entity
await client.send_json_auto_id(
{
"type": "knx/delete_entity",
"entity_id": test_entity_id,
}
)
res = await client.receive_json()
assert res["success"], res
assert not entity_registry.async_get(test_entity_id)
assert not hass_storage[KNX_CONFIG_STORAGE_KEY]["data"]["entities"].get("switch")
async def test_delete_entity_error(
hass: HomeAssistant,
knx: KNXTestKit,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
hass_storage: dict[str, Any],
) -> None:
"""Test unsuccessful entity deletion."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
# delete unknown entity
await client.send_json_auto_id(
{
"type": "knx/delete_entity",
"entity_id": "switch.non_existing_entity",
}
)
res = await client.receive_json()
assert not res["success"], res
assert res["error"]["code"] == "home_assistant_error"
assert res["error"]["message"].startswith("Entity not found")
# delete entity not in config store
test_entity_id = "sensor.knx_interface_individual_address"
assert entity_registry.async_get(test_entity_id)
await client.send_json_auto_id(
{
"type": "knx/delete_entity",
"entity_id": test_entity_id,
}
)
res = await client.receive_json()
assert not res["success"], res
assert res["error"]["code"] == "home_assistant_error"
assert res["error"]["message"].startswith("Entity not found")
async def test_get_entity_config(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
create_ui_entity: KnxEntityGenerator,
) -> None:
"""Test entity config retrieval."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
test_entity = await create_ui_entity(
platform=Platform.SWITCH,
knx_data={"ga_switch": {"write": "1/2/3"}},
entity_data={"name": "Test"},
)
await client.send_json_auto_id(
{
"type": "knx/get_entity_config",
"entity_id": test_entity.entity_id,
}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"]["platform"] == Platform.SWITCH
assert res["result"]["data"] == {
"entity": {
"name": "Test",
"device_info": None,
"entity_category": None,
},
"knx": {
"ga_switch": {"write": "1/2/3", "passive": [], "state": None},
"respond_to_read": False,
"invert": False,
"sync_state": True,
},
}
@pytest.mark.parametrize(
("test_entity_id", "error_message_start"),
[
("switch.non_existing_entity", "Entity not found"),
("sensor.knx_interface_individual_address", "Entity data not found"),
],
)
async def test_get_entity_config_error(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
test_entity_id: str,
error_message_start: str,
) -> None:
"""Test entity config retrieval errors."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "knx/get_entity_config",
"entity_id": test_entity_id,
}
)
res = await client.receive_json()
assert not res["success"], res
assert res["error"]["code"] == "home_assistant_error"
assert res["error"]["message"].startswith(error_message_start)
async def test_validate_entity(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test entity validation."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "knx/validate_entity",
"platform": Platform.SWITCH,
"data": {
"entity": {"name": "test_name"},
"knx": {"ga_switch": {"write": "1/2/3"}},
},
}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"]["success"] is True
# invalid data
await client.send_json_auto_id(
{
"type": "knx/validate_entity",
"platform": Platform.SWITCH,
"data": {
"entity": {"name": "test_name"},
"knx": {"ga_switch": {}},
},
}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"]["success"] is False
assert res["result"]["errors"][0]["path"] == ["data", "knx", "ga_switch", "write"]
assert res["result"]["errors"][0]["error_message"] == "required key not provided"
assert res["result"]["error_base"].startswith("required key not provided")

77
tests/components/knx/test_device.py Normal file
View File

@ -0,0 +1,77 @@
"""Test KNX devices."""
from typing import Any
from homeassistant.components.knx.const import DOMAIN
from homeassistant.components.knx.storage.config_store import (
STORAGE_KEY as KNX_CONFIG_STORAGE_KEY,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component
from .conftest import KNXTestKit
from tests.typing import WebSocketGenerator
async def test_create_device(
hass: HomeAssistant,
knx: KNXTestKit,
device_registry: dr.DeviceRegistry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test device creation."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "knx/create_device",
"name": "Test Device",
}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"]["name"] == "Test Device"
assert res["result"]["manufacturer"] == "KNX"
assert res["result"]["identifiers"]
assert res["result"]["config_entries"][0] == knx.mock_config_entry.entry_id
device_identifier = res["result"]["identifiers"][0][1]
assert device_registry.async_get_device({(DOMAIN, device_identifier)})
device_id = res["result"]["id"]
assert device_registry.async_get(device_id)
async def test_remove_device(
hass: HomeAssistant,
knx: KNXTestKit,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
load_config_store: None,
hass_storage: dict[str, Any],
) -> None:
"""Test device removal."""
assert await async_setup_component(hass, "config", {})
await knx.setup_integration({})
client = await hass_ws_client(hass)
await knx.assert_read("1/0/45", response=True)
assert hass_storage[KNX_CONFIG_STORAGE_KEY]["data"]["entities"].get("switch")
test_device = device_registry.async_get_device(
{(DOMAIN, "knx_vdev_4c80a564f5fe5da701ed293966d6384d")}
)
device_id = test_device.id
device_entities = entity_registry.entities.get_entries_for_device_id(device_id)
assert len(device_entities) == 1
response = await client.remove_device(device_id, knx.mock_config_entry.entry_id)
assert response["success"]
assert not device_registry.async_get_device(
{(DOMAIN, "knx_vdev_4c80a564f5fe5da701ed293966d6384d")}
)
assert not entity_registry.entities.get_entries_for_device_id(device_id)
assert not hass_storage[KNX_CONFIG_STORAGE_KEY]["data"]["entities"].get("switch")

31
tests/components/knx/test_interface_device.py
View File

@ -1,4 +1,4 @@
"""Test KNX scene."""
"""Test KNX interface device."""
from unittest.mock import patch
@ -8,12 +8,14 @@ from xknx.telegram import IndividualAddress
from homeassistant.components.knx.sensor import SCAN_INTERVAL
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .conftest import KNXTestKit
from tests.common import async_capture_events, async_fire_time_changed
from tests.typing import WebSocketGenerator
async def test_diagnostic_entities(
@ -111,3 +113,28 @@ async def test_removed_entity(
)
await hass.async_block_till_done()
unregister_mock.assert_called_once()
async def test_remove_interface_device(
hass: HomeAssistant,
knx: KNXTestKit,
device_registry: dr.DeviceRegistry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test device removal."""
assert await async_setup_component(hass, "config", {})
await knx.setup_integration({})
client = await hass_ws_client(hass)
knx_devices = device_registry.devices.get_devices_for_config_entry_id(
knx.mock_config_entry.entry_id
)
assert len(knx_devices) == 1
assert knx_devices[0].name == "KNX Interface"
device_id = knx_devices[0].id
# interface device can't be removed
res = await client.remove_device(device_id, knx.mock_config_entry.entry_id)
assert not res["success"]
assert (
res["error"]["message"]
== "Failed to remove device entry, rejected by integration"
)

122
tests/components/knx/test_knx_selectors.py Normal file
View File

@ -0,0 +1,122 @@
"""Test KNX selectors."""
import pytest
import voluptuous as vol
from homeassistant.components.knx.const import ColorTempModes
from homeassistant.components.knx.storage.knx_selector import GASelector
INVALID = "invalid"
@pytest.mark.parametrize(
("selector_config", "data", "expected"),
[
(
{},
{},
{"write": None, "state": None, "passive": []},
),
(
{},
{"write": "1/2/3"},
{"write": "1/2/3", "state": None, "passive": []},
),
(
{},
{"state": "1/2/3"},
{"write": None, "state": "1/2/3", "passive": []},
),
(
{},
{"passive": ["1/2/3"]},
{"write": None, "state": None, "passive": ["1/2/3"]},
),
(
{},
{"write": "1", "state": 2, "passive": ["1/2/3"]},
{"write": "1", "state": 2, "passive": ["1/2/3"]},
),
(
{"write": False},
{"write": "1/2/3"},
{"state": None, "passive": []},
),
(
{"write": False},
{"state": "1/2/3"},
{"state": "1/2/3", "passive": []},
),
(
{"write": False},
{"passive": ["1/2/3"]},
{"state": None, "passive": ["1/2/3"]},
),
(
{"passive": False},
{"passive": ["1/2/3"]},
{"write": None, "state": None},
),
(
{"passive": False},
{"write": "1/2/3"},
{"write": "1/2/3", "state": None},
),
# required keys
(
{"write_required": True},
{},
INVALID,
),
(
{"state_required": True},
{},
INVALID,
),
(
{"write_required": True},
{"write": "1/2/3"},
{"write": "1/2/3", "state": None, "passive": []},
),
(
{"state_required": True},
{"state": "1/2/3"},
{"write": None, "state": "1/2/3", "passive": []},
),
(
{"write_required": True},
{"state": "1/2/3"},
INVALID,
),
(
{"state_required": True},
{"write": "1/2/3"},
INVALID,
),
# dpt key
(
{"dpt": ColorTempModes},
{"write": "1/2/3"},
INVALID,
),
(
{"dpt": ColorTempModes},
{"write": "1/2/3", "dpt": "7.600"},
{"write": "1/2/3", "state": None, "passive": [], "dpt": "7.600"},
),
(
{"dpt": ColorTempModes},
{"write": "1/2/3", "state": None, "passive": [], "dpt": "invalid"},
INVALID,
),
],
)
def test_ga_selector(selector_config, data, expected):
"""Test GASelector."""
selector = GASelector(**selector_config)
if expected == INVALID:
with pytest.raises(vol.Invalid):
selector(data)
else:
result = selector(data)
assert result == expected

27
tests/components/knx/test_switch.py
View File

@ -6,9 +6,10 @@ from homeassistant.components.knx.const import (
KNX_ADDRESS,
)
from homeassistant.components.knx.schema import SwitchSchema
from homeassistant.const import CONF_NAME, STATE_OFF, STATE_ON
from homeassistant.const import CONF_NAME, STATE_OFF, STATE_ON, Platform
from homeassistant.core import HomeAssistant, State
from . import KnxEntityGenerator
from .conftest import KNXTestKit
from tests.common import mock_restore_cache
@ -146,3 +147,27 @@ async def test_switch_restore_and_respond(hass: HomeAssistant, knx) -> None:
# respond to new state
await knx.receive_read(_ADDRESS)
await knx.assert_response(_ADDRESS, False)
async def test_switch_ui_create(
hass: HomeAssistant,
knx: KNXTestKit,
create_ui_entity: KnxEntityGenerator,
) -> None:
"""Test creating a switch."""
await knx.setup_integration({})
await create_ui_entity(
platform=Platform.SWITCH,
entity_data={"name": "test"},
knx_data={
"ga_switch": {"write": "1/1/1", "state": "2/2/2"},
"respond_to_read": True,
"sync_state": True,
"invert": False,
},
)
# created entity sends read-request to KNX bus
await knx.assert_read("2/2/2")
await knx.receive_response("2/2/2", True)
state = hass.states.get("switch.test")
assert state.state is STATE_ON

9
tests/components/knx/test_websocket.py
View File

@ -4,6 +4,7 @@ from typing import Any
from unittest.mock import patch
from homeassistant.components.knx import DOMAIN, KNX_ADDRESS, SwitchSchema
from homeassistant.components.knx.project import STORAGE_KEY as KNX_PROJECT_STORAGE_KEY
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
@ -87,6 +88,7 @@ async def test_knx_project_file_process(
assert res["success"], res
assert hass.data[DOMAIN].project.loaded
assert hass_storage[KNX_PROJECT_STORAGE_KEY]["data"] == _parse_result
async def test_knx_project_file_process_error(
@ -126,19 +128,20 @@ async def test_knx_project_file_remove(
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
load_knxproj: None,
hass_storage: dict[str, Any],
) -> None:
"""Test knx/project_file_remove command."""
await knx.setup_integration({})
assert hass_storage[KNX_PROJECT_STORAGE_KEY]
client = await hass_ws_client(hass)
assert hass.data[DOMAIN].project.loaded
await client.send_json({"id": 6, "type": "knx/project_file_remove"})
with patch("homeassistant.helpers.storage.Store.async_remove") as remove_mock:
res = await client.receive_json()
remove_mock.assert_called_once_with()
res = await client.receive_json()
assert res["success"], res
assert not hass.data[DOMAIN].project.loaded
assert not hass_storage.get(KNX_PROJECT_STORAGE_KEY)
async def test_knx_get_project(