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

Add and delete Home Connect devices on CONNECTED/PAIRED and DEPAIRED events (#136952)

Browse Source 
* Add and delete devices on CONNECT/PAIRED and DEPAIRED events

* Simplify device depairing

* small fixes

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Add always the devices

* kind of revert changes

to simplify the entity fetch and removing on connected/paired and depaired

* cache `ha_id`

* Fix typo

* Remove unnecessary device info at HomeConnectEntity

* Move common code of each platform to `common.py`

* Added docstring to clarify usage

* Apply suggestions

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

---------

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
J. Diego Rodríguez Royo 2025-02-02 02:02:45 +01:00 committed by GitHub
parent 147b5f549f
commit 30314ca32b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 1234 additions and 189 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
homeassistant/components/home_connect/binary_sensor.py
View File

@ -21,6 +21,7 @@ from homeassistant.helpers.issue_registry import (
async_delete_issue,
)
from .common import setup_home_connect_entry
from .const import (
BSH_DOOR_STATE_CLOSED,
BSH_DOOR_STATE_LOCKED,
@ -113,24 +114,33 @@ BINARY_SENSORS = (
)
def _get_entities_for_appliance(
entry: HomeConnectConfigEntry,
appliance: HomeConnectApplianceData,
) -> list[HomeConnectEntity]:
"""Get a list of entities."""
entities: list[HomeConnectEntity] = []
entities.extend(
HomeConnectBinarySensor(entry.runtime_data, appliance, description)
for description in BINARY_SENSORS
if description.key in appliance.status
)
if StatusKey.BSH_COMMON_DOOR_STATE in appliance.status:
entities.append(HomeConnectDoorBinarySensor(entry.runtime_data, appliance))
return entities
async def async_setup_entry(
hass: HomeAssistant,
entry: HomeConnectConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Home Connect binary sensor."""
entities: list[BinarySensorEntity] = []
for appliance in entry.runtime_data.data.values():
entities.extend(
HomeConnectBinarySensor(entry.runtime_data, appliance, description)
for description in BINARY_SENSORS
if description.key in appliance.status
)
if StatusKey.BSH_COMMON_DOOR_STATE in appliance.status:
entities.append(HomeConnectDoorBinarySensor(entry.runtime_data, appliance))
async_add_entities(entities)
setup_home_connect_entry(
entry,
_get_entities_for_appliance,
async_add_entities,
)
class HomeConnectBinarySensor(HomeConnectEntity, BinarySensorEntity):

99
homeassistant/components/home_connect/common.py Normal file
View File

@ -0,0 +1,99 @@
"""Common callbacks for all Home Connect platforms."""
from collections.abc import Callable
from functools import partial
from typing import cast
from aiohomeconnect.model import EventKey
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .coordinator import HomeConnectApplianceData, HomeConnectConfigEntry
from .entity import HomeConnectEntity
def _handle_paired_or_connected_appliance(
entry: HomeConnectConfigEntry,
known_entity_unique_ids: dict[str, str],
get_entities_for_appliance: Callable[
[HomeConnectConfigEntry, HomeConnectApplianceData], list[HomeConnectEntity]
],
async_add_entities: AddEntitiesCallback,
) -> None:
"""Handle a new paired appliance or an appliance that has been connected.
This function is used to handle connected events also, because some appliances
don't report any data while they are off because they disconnect themselves
when they are turned off, so we need to check if the entities have been added
already or it is the first time we see them when the appliance is connected.
"""
entities: list[HomeConnectEntity] = []
for appliance in entry.runtime_data.data.values():
entities_to_add = [
entity
for entity in get_entities_for_appliance(entry, appliance)
if entity.unique_id not in known_entity_unique_ids
]
known_entity_unique_ids.update(
{
cast(str, entity.unique_id): appliance.info.ha_id
for entity in entities_to_add
}
)
entities.extend(entities_to_add)
async_add_entities(entities)
def _handle_depaired_appliance(
entry: HomeConnectConfigEntry,
known_entity_unique_ids: dict[str, str],
) -> None:
"""Handle a removed appliance."""
for entity_unique_id, appliance_id in known_entity_unique_ids.copy().items():
if appliance_id not in entry.runtime_data.data:
known_entity_unique_ids.pop(entity_unique_id, None)
def setup_home_connect_entry(
entry: HomeConnectConfigEntry,
get_entities_for_appliance: Callable[
[HomeConnectConfigEntry, HomeConnectApplianceData], list[HomeConnectEntity]
],
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the callbacks for paired and depaired appliances."""
known_entity_unique_ids: dict[str, str] = {}
entities: list[HomeConnectEntity] = []
for appliance in entry.runtime_data.data.values():
entities_to_add = get_entities_for_appliance(entry, appliance)
known_entity_unique_ids.update(
{
cast(str, entity.unique_id): appliance.info.ha_id
for entity in entities_to_add
}
)
entities.extend(entities_to_add)
async_add_entities(entities)
entry.async_on_unload(
entry.runtime_data.async_add_special_listener(
partial(
_handle_paired_or_connected_appliance,
entry,
known_entity_unique_ids,
get_entities_for_appliance,
async_add_entities,
),
(
EventKey.BSH_COMMON_APPLIANCE_PAIRED,
EventKey.BSH_COMMON_APPLIANCE_CONNECTED,
),
)
)
entry.async_on_unload(
entry.runtime_data.async_add_special_listener(
partial(_handle_depaired_appliance, entry, known_entity_unique_ids),
(EventKey.BSH_COMMON_APPLIANCE_DEPAIRED,),
)
)

242
homeassistant/components/home_connect/coordinator.py
View File

@ -3,7 +3,7 @@
import asyncio
from collections import defaultdict
from collections.abc import Callable
from dataclasses import dataclass, field
from dataclasses import dataclass
import logging
from typing import Any
@ -30,6 +30,7 @@ from propcache.api import cached_property
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import APPLIANCES_WITH_PROGRAMS, DOMAIN
@ -46,9 +47,9 @@ EVENT_STREAM_RECONNECT_DELAY = 30
class HomeConnectApplianceData:
"""Class to hold Home Connect appliance data."""
events: dict[EventKey, Event] = field(default_factory=dict)
events: dict[EventKey, Event]
info: HomeAppliance
programs: list[EnumerateProgram] = field(default_factory=list)
programs: list[EnumerateProgram]
settings: dict[SettingKey, GetSetting]
status: dict[StatusKey, Status]
@ -83,6 +84,10 @@ class HomeConnectCoordinator(
name=config_entry.entry_id,
)
self.client = client
self._special_listeners: dict[
CALLBACK_TYPE, tuple[CALLBACK_TYPE, tuple[EventKey, ...]]
] = {}
self.device_registry = dr.async_get(self.hass)
@cached_property
def context_listeners(self) -> dict[tuple[str, EventKey], list[CALLBACK_TYPE]]:
@ -107,6 +112,28 @@ class HomeConnectCoordinator(
return remove_listener_and_invalidate_context_listeners
@callback
def async_add_special_listener(
self,
update_callback: CALLBACK_TYPE,
context: tuple[EventKey, ...],
) -> Callable[[], None]:
"""Listen for special data updates.
These listeners will not be called on refresh.
"""
@callback
def remove_listener() -> None:
"""Remove update listener."""
self._special_listeners.pop(remove_listener)
if not self._special_listeners:
self._unschedule_refresh()
self._special_listeners[remove_listener] = (update_callback, context)
return remove_listener
@callback
def start_event_listener(self) -> None:
"""Start event listener."""
@ -161,18 +188,49 @@ class HomeConnectCoordinator(
events[event.key] = event
self._call_event_listener(event_message)
case EventType.CONNECTED:
self.data[event_message_ha_id].info.connected = True
self._call_all_event_listeners_for_appliance(
case EventType.CONNECTED | EventType.PAIRED:
appliance_info = await self.client.get_specific_appliance(
event_message_ha_id
)
appliance_data = await self._get_appliance_data(
appliance_info, self.data.get(appliance_info.ha_id)
)
if event_message_ha_id in self.data:
self.data[event_message_ha_id].update(appliance_data)
else:
self.data[event_message_ha_id] = appliance_data
for listener, context in list(
self._special_listeners.values()
) + list(self._listeners.values()):
assert isinstance(context, tuple)
if (
EventKey.BSH_COMMON_APPLIANCE_DEPAIRED
not in context
):
listener()
case EventType.DISCONNECTED:
self.data[event_message_ha_id].info.connected = False
self._call_all_event_listeners_for_appliance(
event_message_ha_id
)
case EventType.DEPAIRED:
device = self.device_registry.async_get_device(
identifiers={(DOMAIN, event_message_ha_id)}
)
if device:
self.device_registry.async_update_device(
device_id=device.id,
remove_config_entry_id=self.config_entry.entry_id,
)
self.data.pop(event_message_ha_id, None)
for listener, context in self._special_listeners.values():
assert isinstance(context, tuple)
if EventKey.BSH_COMMON_APPLIANCE_DEPAIRED in context:
listener()
except (EventStreamInterruptedError, HomeConnectRequestError) as error:
_LOGGER.debug(
"Non-breaking error (%s) while listening for events,"
@ -217,81 +275,101 @@ class HomeConnectCoordinator(
translation_placeholders=get_dict_from_home_connect_error(error),
) from error
appliances_data = self.data or {}
for appliance in appliances.homeappliances:
try:
settings = {
setting.key: setting
for setting in (
await self.client.get_settings(appliance.ha_id)
).settings
}
except HomeConnectError as error:
_LOGGER.debug(
"Error fetching settings for %s: %s",
appliance.ha_id,
error
if isinstance(error, HomeConnectApiError)
else type(error).__name__,
)
settings = {}
try:
status = {
status.key: status
for status in (await self.client.get_status(appliance.ha_id)).status
}
except HomeConnectError as error:
_LOGGER.debug(
"Error fetching status for %s: %s",
appliance.ha_id,
error
if isinstance(error, HomeConnectApiError)
else type(error).__name__,
)
status = {}
appliance_data = HomeConnectApplianceData(
info=appliance, settings=settings, status=status
return {
appliance.ha_id: await self._get_appliance_data(
appliance, self.data.get(appliance.ha_id) if self.data else None
)
if appliance.ha_id in appliances_data:
appliances_data[appliance.ha_id].update(appliance_data)
appliance_data = appliances_data[appliance.ha_id]
else:
appliances_data[appliance.ha_id] = appliance_data
if appliance.type in APPLIANCES_WITH_PROGRAMS:
try:
all_programs = await self.client.get_all_programs(appliance.ha_id)
except HomeConnectError as error:
_LOGGER.debug(
"Error fetching programs for %s: %s",
appliance.ha_id,
error
if isinstance(error, HomeConnectApiError)
else type(error).__name__,
)
else:
appliance_data.programs.extend(all_programs.programs)
for program, event_key in (
(
all_programs.active,
EventKey.BSH_COMMON_ROOT_ACTIVE_PROGRAM,
),
(
all_programs.selected,
EventKey.BSH_COMMON_ROOT_SELECTED_PROGRAM,
),
):
if program and program.key:
appliance_data.events.update(
{
event_key: Event(
event_key,
event_key.value,
0,
"",
"",
program.key,
)
}
)
for appliance in appliances.homeappliances
}
return appliances_data
async def _get_appliance_data(
self,
appliance: HomeAppliance,
appliance_data_to_update: HomeConnectApplianceData | None = None,
) -> HomeConnectApplianceData:
"""Get appliance data."""
self.device_registry.async_get_or_create(
config_entry_id=self.config_entry.entry_id,
identifiers={(DOMAIN, appliance.ha_id)},
manufacturer=appliance.brand,
name=appliance.name,
model=appliance.vib,
)
try:
settings = {
setting.key: setting
for setting in (
await self.client.get_settings(appliance.ha_id)
).settings
}
except HomeConnectError as error:
_LOGGER.debug(
"Error fetching settings for %s: %s",
appliance.ha_id,
error
if isinstance(error, HomeConnectApiError)
else type(error).__name__,
)
settings = {}
try:
status = {
status.key: status
for status in (await self.client.get_status(appliance.ha_id)).status
}
except HomeConnectError as error:
_LOGGER.debug(
"Error fetching status for %s: %s",
appliance.ha_id,
error
if isinstance(error, HomeConnectApiError)
else type(error).__name__,
)
status = {}
programs = []
events = {}
if appliance.type in APPLIANCES_WITH_PROGRAMS:
try:
all_programs = await self.client.get_all_programs(appliance.ha_id)
except HomeConnectError as error:
_LOGGER.debug(
"Error fetching programs for %s: %s",
appliance.ha_id,
error
if isinstance(error, HomeConnectApiError)
else type(error).__name__,
)
else:
programs.extend(all_programs.programs)
for program, event_key in (
(
all_programs.active,
EventKey.BSH_COMMON_ROOT_ACTIVE_PROGRAM,
),
(
all_programs.selected,
EventKey.BSH_COMMON_ROOT_SELECTED_PROGRAM,
),
):
if program and program.key:
events[event_key] = Event(
event_key,
event_key.value,
0,
"",
"",
program.key,
)
appliance_data = HomeConnectApplianceData(
events=events,
info=appliance,
programs=programs,
settings=settings,
status=status,
)
if appliance_data_to_update:
appliance_data_to_update.update(appliance_data)
appliance_data = appliance_data_to_update
return appliance_data

3
homeassistant/components/home_connect/entity.py
View File

@ -35,9 +35,6 @@ class HomeConnectEntity(CoordinatorEntity[HomeConnectCoordinator]):
self._attr_unique_id = f"{appliance.info.ha_id}-{desc.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, appliance.info.ha_id)},
manufacturer=appliance.info.brand,
model=appliance.info.vib,
name=appliance.info.name,
)
self.update_native_value()

25
homeassistant/components/home_connect/light.py
View File

@ -20,6 +20,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import color as color_util
from .common import setup_home_connect_entry
from .const import (
BSH_AMBIENT_LIGHT_COLOR_CUSTOM_COLOR,
DOMAIN,
@ -78,20 +79,28 @@ LIGHTS: tuple[HomeConnectLightEntityDescription, ...] = (
)
def _get_entities_for_appliance(
entry: HomeConnectConfigEntry,
appliance: HomeConnectApplianceData,
) -> list[HomeConnectEntity]:
"""Get a list of entities."""
return [
HomeConnectLight(entry.runtime_data, appliance, description)
for description in LIGHTS
if description.key in appliance.settings
]
async def async_setup_entry(
hass: HomeAssistant,
entry: HomeConnectConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Home Connect light."""
async_add_entities(
[
HomeConnectLight(entry.runtime_data, appliance, description)
for description in LIGHTS
for appliance in entry.runtime_data.data.values()
if description.key in appliance.settings
],
setup_home_connect_entry(
entry,
_get_entities_for_appliance,
async_add_entities,
)

26
homeassistant/components/home_connect/number.py
View File

@ -15,6 +15,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .common import setup_home_connect_entry
from .const import (
DOMAIN,
SVE_TRANSLATION_KEY_SET_SETTING,
@ -22,7 +23,7 @@ from .const import (
SVE_TRANSLATION_PLACEHOLDER_KEY,
SVE_TRANSLATION_PLACEHOLDER_VALUE,
)
from .coordinator import HomeConnectConfigEntry
from .coordinator import HomeConnectApplianceData, HomeConnectConfigEntry
from .entity import HomeConnectEntity
from .utils import get_dict_from_home_connect_error
@ -78,19 +79,28 @@ NUMBERS = (
)
def _get_entities_for_appliance(
entry: HomeConnectConfigEntry,
appliance: HomeConnectApplianceData,
) -> list[HomeConnectEntity]:
"""Get a list of entities."""
return [
HomeConnectNumberEntity(entry.runtime_data, appliance, description)
for description in NUMBERS
if description.key in appliance.settings
]
async def async_setup_entry(
hass: HomeAssistant,
entry: HomeConnectConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Home Connect number."""
async_add_entities(
[
HomeConnectNumberEntity(entry.runtime_data, appliance, description)
for description in NUMBERS
for appliance in entry.runtime_data.data.values()
if description.key in appliance.settings
],
setup_home_connect_entry(
entry,
_get_entities_for_appliance,
async_add_entities,
)

26
homeassistant/components/home_connect/select.py
View File

@ -14,6 +14,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .common import setup_home_connect_entry
from .const import APPLIANCES_WITH_PROGRAMS, DOMAIN, SVE_TRANSLATION_PLACEHOLDER_PROGRAM
from .coordinator import (
HomeConnectApplianceData,
@ -69,18 +70,31 @@ PROGRAM_SELECT_ENTITY_DESCRIPTIONS = (
)
def _get_entities_for_appliance(
entry: HomeConnectConfigEntry,
appliance: HomeConnectApplianceData,
) -> list[HomeConnectEntity]:
"""Get a list of entities."""
return (
[
HomeConnectProgramSelectEntity(entry.runtime_data, appliance, desc)
for desc in PROGRAM_SELECT_ENTITY_DESCRIPTIONS
]
if appliance.info.type in APPLIANCES_WITH_PROGRAMS
else []
)
async def async_setup_entry(
hass: HomeAssistant,
entry: HomeConnectConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Home Connect select entities."""
async_add_entities(
HomeConnectProgramSelectEntity(entry.runtime_data, appliance, desc)
for appliance in entry.runtime_data.data.values()
for desc in PROGRAM_SELECT_ENTITY_DESCRIPTIONS
if appliance.info.type in APPLIANCES_WITH_PROGRAMS
setup_home_connect_entry(
entry,
_get_entities_for_appliance,
async_add_entities,
)

58
homeassistant/components/home_connect/sensor.py
View File

@ -17,13 +17,14 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import dt as dt_util, slugify
from .common import setup_home_connect_entry
from .const import (
APPLIANCES_WITH_PROGRAMS,
BSH_OPERATION_STATE_FINISHED,
BSH_OPERATION_STATE_PAUSE,
BSH_OPERATION_STATE_RUN,
)
from .coordinator import HomeConnectConfigEntry
from .coordinator import HomeConnectApplianceData, HomeConnectConfigEntry
from .entity import HomeConnectEntity
EVENT_OPTIONS = ["confirmed", "off", "present"]
@ -243,37 +244,42 @@ EVENT_SENSORS = (
)
def _get_entities_for_appliance(
entry: HomeConnectConfigEntry,
appliance: HomeConnectApplianceData,
) -> list[HomeConnectEntity]:
"""Get a list of entities."""
return [
*[
HomeConnectEventSensor(entry.runtime_data, appliance, description)
for description in EVENT_SENSORS
if description.appliance_types
and appliance.info.type in description.appliance_types
],
*[
HomeConnectProgramSensor(entry.runtime_data, appliance, desc)
for desc in BSH_PROGRAM_SENSORS
if desc.appliance_types and appliance.info.type in desc.appliance_types
],
*[
HomeConnectSensor(entry.runtime_data, appliance, description)
for description in SENSORS
if description.key in appliance.status
],
]
async def async_setup_entry(
hass: HomeAssistant,
entry: HomeConnectConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Home Connect sensor."""
entities: list[SensorEntity] = []
for appliance in entry.runtime_data.data.values():
entities.extend(
HomeConnectEventSensor(
entry.runtime_data,
appliance,
description,
)
for description in EVENT_SENSORS
if description.appliance_types
and appliance.info.type in description.appliance_types
)
entities.extend(
HomeConnectProgramSensor(entry.runtime_data, appliance, desc)
for desc in BSH_PROGRAM_SENSORS
if desc.appliance_types and appliance.info.type in desc.appliance_types
)
entities.extend(
HomeConnectSensor(entry.runtime_data, appliance, description)
for description in SENSORS
if description.key in appliance.status
)
async_add_entities(entities)
setup_home_connect_entry(
entry,
_get_entities_for_appliance,
async_add_entities,
)
class HomeConnectSensor(HomeConnectEntity, SensorEntity):

53
homeassistant/components/home_connect/switch.py
View File

@ -21,6 +21,7 @@ from homeassistant.helpers.issue_registry import (
)
from homeassistant.helpers.typing import UNDEFINED, UndefinedType
from .common import setup_home_connect_entry
from .const import (
BSH_POWER_OFF,
BSH_POWER_ON,
@ -100,33 +101,43 @@ POWER_SWITCH_DESCRIPTION = SwitchEntityDescription(
)
def _get_entities_for_appliance(
entry: HomeConnectConfigEntry,
appliance: HomeConnectApplianceData,
) -> list[HomeConnectEntity]:
"""Get a list of entities."""
entities: list[HomeConnectEntity] = []
entities.extend(
HomeConnectProgramSwitch(entry.runtime_data, appliance, program)
for program in appliance.programs
if program.key != ProgramKey.UNKNOWN
)
if SettingKey.BSH_COMMON_POWER_STATE in appliance.settings:
entities.append(
HomeConnectPowerSwitch(
entry.runtime_data, appliance, POWER_SWITCH_DESCRIPTION
)
)
entities.extend(
HomeConnectSwitch(entry.runtime_data, appliance, description)
for description in SWITCHES
if description.key in appliance.settings
)
return entities
async def async_setup_entry(
hass: HomeAssistant,
entry: HomeConnectConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Home Connect switch."""
entities: list[SwitchEntity] = []
for appliance in entry.runtime_data.data.values():
entities.extend(
HomeConnectProgramSwitch(entry.runtime_data, appliance, program)
for program in appliance.programs
if program.key != ProgramKey.UNKNOWN
)
if SettingKey.BSH_COMMON_POWER_STATE in appliance.settings:
entities.append(
HomeConnectPowerSwitch(
entry.runtime_data, appliance, POWER_SWITCH_DESCRIPTION
)
)
entities.extend(
HomeConnectSwitch(entry.runtime_data, appliance, description)
for description in SWITCHES
if description.key in appliance.settings
)
async_add_entities(entities)
setup_home_connect_entry(
entry,
_get_entities_for_appliance,
async_add_entities,
)
class HomeConnectSwitch(HomeConnectEntity, SwitchEntity):

27
homeassistant/components/home_connect/time.py
View File

@ -11,6 +11,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .common import setup_home_connect_entry
from .const import (
DOMAIN,
SVE_TRANSLATION_KEY_SET_SETTING,
@ -18,7 +19,7 @@ from .const import (
SVE_TRANSLATION_PLACEHOLDER_KEY,
SVE_TRANSLATION_PLACEHOLDER_VALUE,
)
from .coordinator import HomeConnectConfigEntry
from .coordinator import HomeConnectApplianceData, HomeConnectConfigEntry
from .entity import HomeConnectEntity
from .utils import get_dict_from_home_connect_error
@ -30,20 +31,28 @@ TIME_ENTITIES = (
)
def _get_entities_for_appliance(
entry: HomeConnectConfigEntry,
appliance: HomeConnectApplianceData,
) -> list[HomeConnectEntity]:
"""Get a list of entities."""
return [
HomeConnectTimeEntity(entry.runtime_data, appliance, description)
for description in TIME_ENTITIES
if description.key in appliance.settings
]
async def async_setup_entry(
hass: HomeAssistant,
entry: HomeConnectConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Home Connect switch."""
async_add_entities(
[
HomeConnectTimeEntity(entry.runtime_data, appliance, description)
for description in TIME_ENTITIES
for appliance in entry.runtime_data.data.values()
if description.key in appliance.settings
],
setup_home_connect_entry(
entry,
_get_entities_for_appliance,
async_add_entities,
)

37
tests/components/home_connect/conftest.py
View File

@ -18,8 +18,11 @@ from aiohomeconnect.model import (
EventKey,
EventMessage,
EventType,
GetSetting,
HomeAppliance,
Option,
Program,
SettingKey,
)
from aiohomeconnect.model.error import HomeConnectApiError, HomeConnectError
from aiohomeconnect.model.program import EnumerateProgram
@ -145,6 +148,14 @@ async def mock_integration_setup(
return run
def _get_specific_appliance_side_effect(ha_id: str) -> HomeAppliance:
"""Get specific appliance side effect."""
for appliance in copy.deepcopy(MOCK_APPLIANCES).homeappliances:
if appliance.ha_id == ha_id:
return appliance
raise HomeConnectApiError("error.key", "error description")
def _get_set_program_side_effect(
event_queue: asyncio.Queue[list[EventMessage]], event_key: EventKey
):
@ -253,6 +264,24 @@ async def _get_settings_side_effect(ha_id: str) -> ArrayOfSettings:
)
async def _get_setting_side_effect(ha_id: str, setting_key: SettingKey):
"""Get setting."""
for appliance in MOCK_APPLIANCES.homeappliances:
if appliance.ha_id == ha_id:
settings = MOCK_SETTINGS.get(
next(
appliance
for appliance in MOCK_APPLIANCES.homeappliances
if appliance.ha_id == ha_id
).type,
{},
).get("data", {"settings": []})
for setting_dict in cast(list[dict], settings["settings"]):
if setting_dict["key"] == setting_key:
return GetSetting.from_dict(setting_dict)
raise HomeConnectApiError("error.key", "error description")
@pytest.fixture(name="client")
def mock_client(request: pytest.FixtureRequest) -> MagicMock:
"""Fixture to mock Client from HomeConnect."""
@ -274,7 +303,10 @@ def mock_client(request: pytest.FixtureRequest) -> MagicMock:
for event in await event_queue.get():
yield event
mock.get_home_appliances = AsyncMock(return_value=MOCK_APPLIANCES)
mock.get_home_appliances = AsyncMock(return_value=copy.deepcopy(MOCK_APPLIANCES))
mock.get_specific_appliance = AsyncMock(
side_effect=_get_specific_appliance_side_effect
)
mock.stream_all_events = stream_all_events
mock.start_program = AsyncMock(
side_effect=_get_set_program_side_effect(
@ -296,6 +328,7 @@ def mock_client(request: pytest.FixtureRequest) -> MagicMock:
side_effect=_get_set_key_value_side_effect(event_queue, "setting_key"),
)
mock.get_settings = AsyncMock(side_effect=_get_settings_side_effect)
mock.get_setting = AsyncMock(side_effect=_get_setting_side_effect)
mock.get_status = AsyncMock(return_value=copy.deepcopy(MOCK_STATUS))
mock.get_all_programs = AsyncMock(side_effect=_get_all_programs_side_effect)
mock.put_command = AsyncMock()
@ -323,7 +356,7 @@ def mock_client_with_exception(request: pytest.FixtureRequest) -> MagicMock:
for event in await event_queue.get():
yield event
mock.get_home_appliances = AsyncMock(return_value=MOCK_APPLIANCES)
mock.get_home_appliances = AsyncMock(return_value=copy.deepcopy(MOCK_APPLIANCES))
mock.stream_all_events = stream_all_events
mock.start_program = AsyncMock(side_effect=exception)

108
tests/components/home_connect/test_binary_sensor.py
View File

@ -1,9 +1,10 @@
"""Tests for home_connect binary_sensor entities."""
from collections.abc import Awaitable, Callable
from unittest.mock import MagicMock
from unittest.mock import AsyncMock, MagicMock
from aiohomeconnect.model import ArrayOfEvents, Event, EventKey, EventMessage, EventType
from aiohomeconnect.model.error import HomeConnectApiError
import pytest
from homeassistant.components import automation, script
@ -26,6 +27,7 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
import homeassistant.helpers.issue_registry as ir
from homeassistant.setup import async_setup_component
@ -50,6 +52,110 @@ async def test_binary_sensors(
assert config_entry.state == ConfigEntryState.LOADED
async def test_paired_depaired_devices_flow(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that removed devices are correctly removed from and added to hass on API events."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert entity_entries
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.DEPAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert not device
for entity_entry in entity_entries:
assert not entity_registry.async_get(entity_entry.entity_id)
# Now that all everything related to the device is removed, pair it again
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.PAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
assert device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
for entity_entry in entity_entries:
assert entity_registry.async_get(entity_entry.entity_id)
async def test_connected_devices(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that devices reconnected.
Specifically those devices whose settings, status, etc. could
not be obtained while disconnected and once connected, the entities are added.
"""
get_status_original_mock = client.get_status
def get_status_side_effect(ha_id: str):
if ha_id == appliance_ha_id:
raise HomeConnectApiError(
"SDK.Error.HomeAppliance.Connection.Initialization.Failed"
)
return get_status_original_mock.return_value
client.get_status = AsyncMock(side_effect=get_status_side_effect)
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
client.get_status = get_status_original_mock
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.CONNECTED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
new_entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert len(new_entity_entries) > len(entity_entries)
async def test_binary_sensors_entity_availabilty(
hass: HomeAssistant,
config_entry: MockConfigEntry,

124
tests/components/home_connect/test_light.py
View File

@ -2,7 +2,7 @@
from collections.abc import Awaitable, Callable
from typing import Any
from unittest.mock import MagicMock, call
from unittest.mock import AsyncMock, MagicMock, call
from aiohomeconnect.model import (
ArrayOfEvents,
@ -14,11 +14,12 @@ from aiohomeconnect.model import (
GetSetting,
SettingKey,
)
from aiohomeconnect.model.error import HomeConnectError
from aiohomeconnect.model.error import HomeConnectApiError, HomeConnectError
import pytest
from homeassistant.components.home_connect.const import (
BSH_AMBIENT_LIGHT_COLOR_CUSTOM_COLOR,
DOMAIN,
)
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.config_entries import ConfigEntryState
@ -32,6 +33,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry
@ -56,6 +58,124 @@ async def test_light(
assert config_entry.state == ConfigEntryState.LOADED
@pytest.mark.parametrize("appliance_ha_id", ["Hood"], indirect=True)
async def test_paired_depaired_devices_flow(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that removed devices are correctly removed from and added to hass on API events."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert entity_entries
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.DEPAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert not device
for entity_entry in entity_entries:
assert not entity_registry.async_get(entity_entry.entity_id)
# Now that all everything related to the device is removed, pair it again
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.PAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
assert device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
for entity_entry in entity_entries:
assert entity_registry.async_get(entity_entry.entity_id)
@pytest.mark.parametrize("appliance_ha_id", ["Hood"], indirect=True)
async def test_connected_devices(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that devices reconnected.
Specifically those devices whose settings, status, etc. could
not be obtained while disconnected and once connected, the entities are added.
"""
get_settings_original_mock = client.get_settings
get_available_programs_mock = client.get_available_programs
async def get_settings_side_effect(ha_id: str):
if ha_id == appliance_ha_id:
raise HomeConnectApiError(
"SDK.Error.HomeAppliance.Connection.Initialization.Failed"
)
return await get_settings_original_mock.side_effect(ha_id)
async def get_available_programs_side_effect(ha_id: str):
if ha_id == appliance_ha_id:
raise HomeConnectApiError(
"SDK.Error.HomeAppliance.Connection.Initialization.Failed"
)
return await get_available_programs_mock.side_effect(ha_id)
client.get_settings = AsyncMock(side_effect=get_settings_side_effect)
client.get_available_programs = AsyncMock(
side_effect=get_available_programs_side_effect
)
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
client.get_settings = get_settings_original_mock
client.get_available_programs = get_available_programs_mock
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.CONNECTED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
new_entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert len(new_entity_entries) > len(entity_entries)
@pytest.mark.parametrize("appliance_ha_id", ["Hood"], indirect=True)
async def test_light_availabilty(
hass: HomeAssistant,

110
tests/components/home_connect/test_number.py
View File

@ -12,10 +12,11 @@ from aiohomeconnect.model import (
GetSetting,
SettingKey,
)
from aiohomeconnect.model.error import HomeConnectError
from aiohomeconnect.model.error import HomeConnectApiError, HomeConnectError
from aiohomeconnect.model.setting import SettingConstraints
import pytest
from homeassistant.components.home_connect.const import DOMAIN
from homeassistant.components.number import (
ATTR_VALUE as SERVICE_ATTR_VALUE,
DEFAULT_MAX_VALUE,
@ -27,6 +28,7 @@ from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry
@ -49,6 +51,112 @@ async def test_number(
assert config_entry.state is ConfigEntryState.LOADED
@pytest.mark.parametrize("appliance_ha_id", ["FridgeFreezer"], indirect=True)
async def test_paired_depaired_devices_flow(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that removed devices are correctly removed from and added to hass on API events."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert entity_entries
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.DEPAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert not device
for entity_entry in entity_entries:
assert not entity_registry.async_get(entity_entry.entity_id)
# Now that all everything related to the device is removed, pair it again
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.PAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
assert device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
for entity_entry in entity_entries:
assert entity_registry.async_get(entity_entry.entity_id)
@pytest.mark.parametrize("appliance_ha_id", ["FridgeFreezer"], indirect=True)
async def test_connected_devices(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that devices reconnected.
Specifically those devices whose settings, status, etc. could
not be obtained while disconnected and once connected, the entities are added.
"""
get_settings_original_mock = client.get_settings
def get_settings_side_effect(ha_id: str):
if ha_id == appliance_ha_id:
raise HomeConnectApiError(
"SDK.Error.HomeAppliance.Connection.Initialization.Failed"
)
return get_settings_original_mock.return_value
client.get_settings = AsyncMock(side_effect=get_settings_side_effect)
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
client.get_settings = get_settings_original_mock
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.CONNECTED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
new_entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert len(new_entity_entries) > len(entity_entries)
@pytest.mark.parametrize("appliance_ha_id", ["FridgeFreezer"], indirect=True)
async def test_number_entity_availabilty(
hass: HomeAssistant,

96
tests/components/home_connect/test_select.py
View File

@ -20,6 +20,7 @@ from aiohomeconnect.model.program import (
)
import pytest
from homeassistant.components.home_connect.const import DOMAIN
from homeassistant.components.select import (
ATTR_OPTION,
ATTR_OPTIONS,
@ -35,7 +36,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry
@ -58,6 +59,99 @@ async def test_select(
assert config_entry.state is ConfigEntryState.LOADED
async def test_paired_depaired_devices_flow(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that removed devices are correctly removed from and added to hass on API events."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert entity_entries
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.DEPAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert not device
for entity_entry in entity_entries:
assert not entity_registry.async_get(entity_entry.entity_id)
# Now that all everything related to the device is removed, pair it again
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.PAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
assert device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
for entity_entry in entity_entries:
assert entity_registry.async_get(entity_entry.entity_id)
async def test_connected_devices(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that devices reconnected.
Specifically those devices whose settings, status, etc. could
not be obtained while disconnected and once connected, the entities are added.
"""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.CONNECTED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert entity_entries
async def test_select_entity_availabilty(
hass: HomeAssistant,
config_entry: MockConfigEntry,

109
tests/components/home_connect/test_sensor.py
View File

@ -1,7 +1,7 @@
"""Tests for home_connect sensor entities."""
from collections.abc import Awaitable, Callable
from unittest.mock import MagicMock
from unittest.mock import AsyncMock, MagicMock
from aiohomeconnect.model import (
ArrayOfEvents,
@ -12,6 +12,7 @@ from aiohomeconnect.model import (
Status,
StatusKey,
)
from aiohomeconnect.model.error import HomeConnectApiError
from freezegun.api import FrozenDateTimeFactory
import pytest
@ -22,10 +23,12 @@ from homeassistant.components.home_connect.const import (
BSH_EVENT_PRESENT_STATE_CONFIRMED,
BSH_EVENT_PRESENT_STATE_OFF,
BSH_EVENT_PRESENT_STATE_PRESENT,
DOMAIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry
@ -94,6 +97,110 @@ async def test_sensors(
assert config_entry.state == ConfigEntryState.LOADED
async def test_paired_depaired_devices_flow(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that removed devices are correctly removed from and added to hass on API events."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert entity_entries
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.DEPAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert not device
for entity_entry in entity_entries:
assert not entity_registry.async_get(entity_entry.entity_id)
# Now that all everything related to the device is removed, pair it again
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.PAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
assert device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
for entity_entry in entity_entries:
assert entity_registry.async_get(entity_entry.entity_id)
async def test_connected_devices(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that devices reconnected.
Specifically those devices whose settings, status, etc. could
not be obtained while disconnected and once connected, the entities are added.
"""
get_status_original_mock = client.get_status
def get_status_side_effect(ha_id: str):
if ha_id == appliance_ha_id:
raise HomeConnectApiError(
"SDK.Error.HomeAppliance.Connection.Initialization.Failed"
)
return get_status_original_mock.return_value
client.get_status = AsyncMock(side_effect=get_status_side_effect)
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
client.get_status = get_status_original_mock
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.CONNECTED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
new_entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert len(new_entity_entries) > len(entity_entries)
@pytest.mark.parametrize("appliance_ha_id", [TEST_HC_APP], indirect=True)
async def test_sensor_entity_availabilty(
hass: HomeAssistant,

124
tests/components/home_connect/test_switch.py
View File

@ -13,7 +13,7 @@ from aiohomeconnect.model import (
ProgramKey,
SettingKey,
)
from aiohomeconnect.model.error import HomeConnectError
from aiohomeconnect.model.error import HomeConnectApiError, HomeConnectError
from aiohomeconnect.model.event import ArrayOfEvents, EventType
from aiohomeconnect.model.program import ArrayOfPrograms, EnumerateProgram
from aiohomeconnect.model.setting import SettingConstraints
@ -41,7 +41,11 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers import (
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
@ -66,6 +70,122 @@ async def test_switches(
assert config_entry.state == ConfigEntryState.LOADED
async def test_paired_depaired_devices_flow(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that removed devices are correctly removed from and added to hass on API events."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert entity_entries
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.DEPAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert not device
for entity_entry in entity_entries:
assert not entity_registry.async_get(entity_entry.entity_id)
# Now that all everything related to the device is removed, pair it again
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.PAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
assert device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
for entity_entry in entity_entries:
assert entity_registry.async_get(entity_entry.entity_id)
async def test_connected_devices(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that devices reconnected.
Specifically those devices whose settings, status, etc. could
not be obtained while disconnected and once connected, the entities are added.
"""
get_settings_original_mock = client.get_settings
get_available_programs_mock = client.get_available_programs
async def get_settings_side_effect(ha_id: str):
if ha_id == appliance_ha_id:
raise HomeConnectApiError(
"SDK.Error.HomeAppliance.Connection.Initialization.Failed"
)
return await get_settings_original_mock.side_effect(ha_id)
async def get_available_programs_side_effect(ha_id: str):
if ha_id == appliance_ha_id:
raise HomeConnectApiError(
"SDK.Error.HomeAppliance.Connection.Initialization.Failed"
)
return await get_available_programs_mock.side_effect(ha_id)
client.get_settings = AsyncMock(side_effect=get_settings_side_effect)
client.get_available_programs = AsyncMock(
side_effect=get_available_programs_side_effect
)
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
client.get_settings = get_settings_original_mock
client.get_available_programs = get_available_programs_mock
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.CONNECTED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
new_entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert len(new_entity_entries) > len(entity_entries)
@pytest.mark.parametrize("appliance_ha_id", ["Dishwasher"], indirect=True)
async def test_switch_entity_availabilty(
hass: HomeAssistant,

122
tests/components/home_connect/test_time.py
View File

@ -2,18 +2,26 @@
from collections.abc import Awaitable, Callable
from datetime import time
from unittest.mock import MagicMock
from unittest.mock import AsyncMock, MagicMock
from aiohomeconnect.model import ArrayOfSettings, EventMessage, GetSetting, SettingKey
from aiohomeconnect.model.error import HomeConnectError
from aiohomeconnect.model.event import ArrayOfEvents, EventType
from aiohomeconnect.model import (
ArrayOfEvents,
ArrayOfSettings,
EventMessage,
EventType,
GetSetting,
SettingKey,
)
from aiohomeconnect.model.error import HomeConnectApiError, HomeConnectError
import pytest
from homeassistant.components.home_connect.const import DOMAIN
from homeassistant.components.time import DOMAIN as TIME_DOMAIN, SERVICE_SET_VALUE
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID, ATTR_TIME, STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry
@ -36,6 +44,112 @@ async def test_time(
assert config_entry.state is ConfigEntryState.LOADED
@pytest.mark.parametrize("appliance_ha_id", ["Oven"], indirect=True)
async def test_paired_depaired_devices_flow(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that removed devices are correctly removed from and added to hass on API events."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert entity_entries
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.DEPAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert not device
for entity_entry in entity_entries:
assert not entity_registry.async_get(entity_entry.entity_id)
# Now that all everything related to the device is removed, pair it again
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.PAIRED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
assert device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
for entity_entry in entity_entries:
assert entity_registry.async_get(entity_entry.entity_id)
@pytest.mark.parametrize("appliance_ha_id", ["Oven"], indirect=True)
async def test_connected_devices(
appliance_ha_id: str,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that devices reconnected.
Specifically those devices whose settings, status, etc. could
not be obtained while disconnected and once connected, the entities are added.
"""
get_settings_original_mock = client.get_settings
async def get_settings_side_effect(ha_id: str):
if ha_id == appliance_ha_id:
raise HomeConnectApiError(
"SDK.Error.HomeAppliance.Connection.Initialization.Failed"
)
return await get_settings_original_mock.side_effect(ha_id)
client.get_settings = AsyncMock(side_effect=get_settings_side_effect)
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
client.get_settings = get_settings_original_mock
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.CONNECTED,
data=ArrayOfEvents([]),
)
]
)
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, appliance_ha_id)})
assert device
new_entity_entries = entity_registry.entities.get_entries_for_device_id(device.id)
assert len(new_entity_entries) > len(entity_entries)
@pytest.mark.parametrize("appliance_ha_id", ["Oven"], indirect=True)
async def test_time_entity_availabilty(
hass: HomeAssistant,