Move Home Connect service actions to a services.py (#141100)

* Move Home Connect service actions to a actions.py

* Rename actions.py to services.py

* Move more fuctions to module level
This commit is contained in:
J. Diego Rodríguez Royo 2025-03-22 12:35:46 +01:00 committed by GitHub
parent c08cbf3763
commit 9d9b352631
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 1052 additions and 1020 deletions

View File

@ -2,192 +2,29 @@
from __future__ import annotations
from collections.abc import Awaitable
import logging
from typing import Any, cast
from typing import Any
from aiohomeconnect.client import Client as HomeConnectClient
from aiohomeconnect.model import (
ArrayOfOptions,
CommandKey,
Option,
OptionKey,
ProgramKey,
SettingKey,
)
from aiohomeconnect.model.error import HomeConnectError
import aiohttp
import voluptuous as vol
from homeassistant.const import ATTR_DEVICE_ID, Platform
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
HomeAssistantError,
ServiceValidationError,
)
from homeassistant.helpers import (
config_entry_oauth2_flow,
config_validation as cv,
device_registry as dr,
)
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_entry_oauth2_flow, config_validation as cv
from homeassistant.helpers.entity_registry import RegistryEntry, async_migrate_entries
from homeassistant.helpers.issue_registry import (
IssueSeverity,
async_create_issue,
async_delete_issue,
)
from homeassistant.helpers.issue_registry import async_delete_issue
from homeassistant.helpers.typing import ConfigType
from .api import AsyncConfigEntryAuth
from .const import (
AFFECTS_TO_ACTIVE_PROGRAM,
AFFECTS_TO_SELECTED_PROGRAM,
ATTR_AFFECTS_TO,
ATTR_KEY,
ATTR_PROGRAM,
ATTR_UNIT,
ATTR_VALUE,
DOMAIN,
OLD_NEW_UNIQUE_ID_SUFFIX_MAP,
PROGRAM_ENUM_OPTIONS,
SERVICE_OPTION_ACTIVE,
SERVICE_OPTION_SELECTED,
SERVICE_PAUSE_PROGRAM,
SERVICE_RESUME_PROGRAM,
SERVICE_SELECT_PROGRAM,
SERVICE_SET_PROGRAM_AND_OPTIONS,
SERVICE_SETTING,
SERVICE_START_PROGRAM,
TRANSLATION_KEYS_PROGRAMS_MAP,
)
from .const import DOMAIN, OLD_NEW_UNIQUE_ID_SUFFIX_MAP
from .coordinator import HomeConnectConfigEntry, HomeConnectCoordinator
from .utils import bsh_key_to_translation_key, get_dict_from_home_connect_error
from .services import register_actions
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
PROGRAM_OPTIONS = {
bsh_key_to_translation_key(key): (
key,
value,
)
for key, value in {
OptionKey.BSH_COMMON_DURATION: int,
OptionKey.BSH_COMMON_START_IN_RELATIVE: int,
OptionKey.BSH_COMMON_FINISH_IN_RELATIVE: int,
OptionKey.CONSUMER_PRODUCTS_COFFEE_MAKER_FILL_QUANTITY: int,
OptionKey.CONSUMER_PRODUCTS_COFFEE_MAKER_MULTIPLE_BEVERAGES: bool,
OptionKey.DISHCARE_DISHWASHER_INTENSIV_ZONE: bool,
OptionKey.DISHCARE_DISHWASHER_BRILLIANCE_DRY: bool,
OptionKey.DISHCARE_DISHWASHER_VARIO_SPEED_PLUS: bool,
OptionKey.DISHCARE_DISHWASHER_SILENCE_ON_DEMAND: bool,
OptionKey.DISHCARE_DISHWASHER_HALF_LOAD: bool,
OptionKey.DISHCARE_DISHWASHER_EXTRA_DRY: bool,
OptionKey.DISHCARE_DISHWASHER_HYGIENE_PLUS: bool,
OptionKey.DISHCARE_DISHWASHER_ECO_DRY: bool,
OptionKey.DISHCARE_DISHWASHER_ZEOLITE_DRY: bool,
OptionKey.COOKING_OVEN_SETPOINT_TEMPERATURE: int,
OptionKey.COOKING_OVEN_FAST_PRE_HEAT: bool,
OptionKey.LAUNDRY_CARE_WASHER_I_DOS_1_ACTIVE: bool,
OptionKey.LAUNDRY_CARE_WASHER_I_DOS_2_ACTIVE: bool,
}.items()
}
SERVICE_SETTING_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_KEY): vol.All(
vol.Coerce(SettingKey),
vol.NotIn([SettingKey.UNKNOWN]),
),
vol.Required(ATTR_VALUE): vol.Any(str, int, bool),
}
)
# DEPRECATED: Remove in 2025.9.0
SERVICE_OPTION_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_KEY): vol.All(
vol.Coerce(OptionKey),
vol.NotIn([OptionKey.UNKNOWN]),
),
vol.Required(ATTR_VALUE): vol.Any(str, int, bool),
vol.Optional(ATTR_UNIT): str,
}
)
# DEPRECATED: Remove in 2025.9.0
SERVICE_PROGRAM_SCHEMA = vol.Any(
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_PROGRAM): vol.All(
vol.Coerce(ProgramKey),
vol.NotIn([ProgramKey.UNKNOWN]),
),
vol.Required(ATTR_KEY): vol.All(
vol.Coerce(OptionKey),
vol.NotIn([OptionKey.UNKNOWN]),
),
vol.Required(ATTR_VALUE): vol.Any(int, str),
vol.Optional(ATTR_UNIT): str,
},
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_PROGRAM): vol.All(
vol.Coerce(ProgramKey),
vol.NotIn([ProgramKey.UNKNOWN]),
),
},
)
def _require_program_or_at_least_one_option(data: dict) -> dict:
if ATTR_PROGRAM not in data and not any(
option_key in data for option_key in (PROGRAM_ENUM_OPTIONS | PROGRAM_OPTIONS)
):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="required_program_or_one_option_at_least",
)
return data
SERVICE_PROGRAM_AND_OPTIONS_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_AFFECTS_TO): vol.In(
[AFFECTS_TO_ACTIVE_PROGRAM, AFFECTS_TO_SELECTED_PROGRAM]
),
vol.Optional(ATTR_PROGRAM): vol.In(TRANSLATION_KEYS_PROGRAMS_MAP.keys()),
}
)
.extend(
{
vol.Optional(translation_key): vol.In(allowed_values.keys())
for translation_key, (
key,
allowed_values,
) in PROGRAM_ENUM_OPTIONS.items()
}
)
.extend(
{
vol.Optional(translation_key): schema
for translation_key, (key, schema) in PROGRAM_OPTIONS.items()
}
),
_require_program_or_at_least_one_option,
)
SERVICE_COMMAND_SCHEMA = vol.Schema({vol.Required(ATTR_DEVICE_ID): str})
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.BUTTON,
@ -200,402 +37,9 @@ PLATFORMS = [
]
async def _get_client_and_ha_id(
hass: HomeAssistant, device_id: str
) -> tuple[HomeConnectClient, str]:
device_registry = dr.async_get(hass)
device_entry = device_registry.async_get(device_id)
if device_entry is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="device_entry_not_found",
translation_placeholders={
"device_id": device_id,
},
)
entry: HomeConnectConfigEntry | None = None
for entry_id in device_entry.config_entries:
_entry = hass.config_entries.async_get_entry(entry_id)
assert _entry
if _entry.domain == DOMAIN:
entry = cast(HomeConnectConfigEntry, _entry)
break
if entry is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="config_entry_not_found",
translation_placeholders={
"device_id": device_id,
},
)
ha_id = next(
(
identifier[1]
for identifier in device_entry.identifiers
if identifier[0] == DOMAIN
),
None,
)
if ha_id is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="appliance_not_found",
translation_placeholders={
"device_id": device_id,
},
)
return entry.runtime_data.client, ha_id
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: # noqa: C901
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Home Connect component."""
async def _async_service_program(call: ServiceCall, start: bool) -> None:
"""Execute calls to services taking a program."""
program = call.data[ATTR_PROGRAM]
client, ha_id = await _get_client_and_ha_id(hass, call.data[ATTR_DEVICE_ID])
option_key = call.data.get(ATTR_KEY)
options = (
[
Option(
option_key,
call.data[ATTR_VALUE],
unit=call.data.get(ATTR_UNIT),
)
]
if option_key is not None
else None
)
async_create_issue(
hass,
DOMAIN,
"deprecated_set_program_and_option_actions",
breaks_in_ha_version="2025.9.0",
is_fixable=True,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="deprecated_set_program_and_option_actions",
translation_placeholders={
"new_action_key": SERVICE_SET_PROGRAM_AND_OPTIONS,
"remove_release": "2025.9.0",
"deprecated_action_yaml": "\n".join(
[
"```yaml",
f"action: {DOMAIN}.{SERVICE_START_PROGRAM if start else SERVICE_SELECT_PROGRAM}",
"data:",
f" {ATTR_DEVICE_ID}: DEVICE_ID",
f" {ATTR_PROGRAM}: {program}",
*([f" {ATTR_KEY}: {options[0].key}"] if options else []),
*([f" {ATTR_VALUE}: {options[0].value}"] if options else []),
*(
[f" {ATTR_UNIT}: {options[0].unit}"]
if options and options[0].unit
else []
),
"```",
]
),
"new_action_yaml": "\n ".join(
[
"```yaml",
f"action: {DOMAIN}.{SERVICE_SET_PROGRAM_AND_OPTIONS}",
"data:",
f" {ATTR_DEVICE_ID}: DEVICE_ID",
f" {ATTR_AFFECTS_TO}: {AFFECTS_TO_ACTIVE_PROGRAM if start else AFFECTS_TO_SELECTED_PROGRAM}",
f" {ATTR_PROGRAM}: {bsh_key_to_translation_key(program.value)}",
*(
[
f" {bsh_key_to_translation_key(options[0].key)}: {options[0].value}"
]
if options
else []
),
"```",
]
),
"repo_link": "[aiohomeconnect](https://github.com/MartinHjelmare/aiohomeconnect)",
},
)
try:
if start:
await client.start_program(ha_id, program_key=program, options=options)
else:
await client.set_selected_program(
ha_id, program_key=program, options=options
)
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="start_program" if start else "select_program",
translation_placeholders={
**get_dict_from_home_connect_error(err),
"program": program,
},
) from err
async def _async_service_set_program_options(
call: ServiceCall, active: bool
) -> None:
"""Execute calls to services taking a program."""
option_key = call.data[ATTR_KEY]
value = call.data[ATTR_VALUE]
unit = call.data.get(ATTR_UNIT)
client, ha_id = await _get_client_and_ha_id(hass, call.data[ATTR_DEVICE_ID])
async_create_issue(
hass,
DOMAIN,
"deprecated_set_program_and_option_actions",
breaks_in_ha_version="2025.9.0",
is_fixable=True,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="deprecated_set_program_and_option_actions",
translation_placeholders={
"new_action_key": SERVICE_SET_PROGRAM_AND_OPTIONS,
"remove_release": "2025.9.0",
"deprecated_action_yaml": "\n".join(
[
"```yaml",
f"action: {DOMAIN}.{SERVICE_OPTION_ACTIVE if active else SERVICE_OPTION_SELECTED}",
"data:",
f" {ATTR_DEVICE_ID}: DEVICE_ID",
f" {ATTR_KEY}: {option_key}",
f" {ATTR_VALUE}: {value}",
*([f" {ATTR_UNIT}: {unit}"] if unit else []),
"```",
]
),
"new_action_yaml": "\n ".join(
[
"```yaml",
f"action: {DOMAIN}.{SERVICE_SET_PROGRAM_AND_OPTIONS}",
"data:",
f" {ATTR_DEVICE_ID}: DEVICE_ID",
f" {ATTR_AFFECTS_TO}: {AFFECTS_TO_ACTIVE_PROGRAM if active else AFFECTS_TO_SELECTED_PROGRAM}",
f" {bsh_key_to_translation_key(option_key)}: {value}",
"```",
]
),
"repo_link": "[aiohomeconnect](https://github.com/MartinHjelmare/aiohomeconnect)",
},
)
try:
if active:
await client.set_active_program_option(
ha_id,
option_key=option_key,
value=value,
unit=unit,
)
else:
await client.set_selected_program_option(
ha_id,
option_key=option_key,
value=value,
unit=unit,
)
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="set_options_active_program"
if active
else "set_options_selected_program",
translation_placeholders={
**get_dict_from_home_connect_error(err),
"key": option_key,
"value": str(value),
},
) from err
async def _async_service_command(
call: ServiceCall, command_key: CommandKey
) -> None:
"""Execute calls to services executing a command."""
client, ha_id = await _get_client_and_ha_id(hass, call.data[ATTR_DEVICE_ID])
async_create_issue(
hass,
DOMAIN,
"deprecated_command_actions",
breaks_in_ha_version="2025.9.0",
is_fixable=True,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="deprecated_command_actions",
)
try:
await client.put_command(ha_id, command_key=command_key, value=True)
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="execute_command",
translation_placeholders={
**get_dict_from_home_connect_error(err),
"command": command_key.value,
},
) from err
async def async_service_option_active(call: ServiceCall) -> None:
"""Service for setting an option for an active program."""
await _async_service_set_program_options(call, True)
async def async_service_option_selected(call: ServiceCall) -> None:
"""Service for setting an option for a selected program."""
await _async_service_set_program_options(call, False)
async def async_service_setting(call: ServiceCall) -> None:
"""Service for changing a setting."""
key = call.data[ATTR_KEY]
value = call.data[ATTR_VALUE]
client, ha_id = await _get_client_and_ha_id(hass, call.data[ATTR_DEVICE_ID])
try:
await client.set_setting(ha_id, setting_key=key, value=value)
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="set_setting",
translation_placeholders={
**get_dict_from_home_connect_error(err),
"key": key,
"value": str(value),
},
) from err
async def async_service_pause_program(call: ServiceCall) -> None:
"""Service for pausing a program."""
await _async_service_command(call, CommandKey.BSH_COMMON_PAUSE_PROGRAM)
async def async_service_resume_program(call: ServiceCall) -> None:
"""Service for resuming a paused program."""
await _async_service_command(call, CommandKey.BSH_COMMON_RESUME_PROGRAM)
async def async_service_select_program(call: ServiceCall) -> None:
"""Service for selecting a program."""
await _async_service_program(call, False)
async def async_service_set_program_and_options(call: ServiceCall) -> None:
"""Service for setting a program and options."""
data = dict(call.data)
program = data.pop(ATTR_PROGRAM, None)
affects_to = data.pop(ATTR_AFFECTS_TO)
client, ha_id = await _get_client_and_ha_id(hass, data.pop(ATTR_DEVICE_ID))
options: list[Option] = []
for option, value in data.items():
if option in PROGRAM_ENUM_OPTIONS:
options.append(
Option(
PROGRAM_ENUM_OPTIONS[option][0],
PROGRAM_ENUM_OPTIONS[option][1][value],
)
)
elif option in PROGRAM_OPTIONS:
option_key = PROGRAM_OPTIONS[option][0]
options.append(Option(option_key, value))
method_call: Awaitable[Any]
exception_translation_key: str
if program:
program = (
program
if isinstance(program, ProgramKey)
else TRANSLATION_KEYS_PROGRAMS_MAP[program]
)
if affects_to == AFFECTS_TO_ACTIVE_PROGRAM:
method_call = client.start_program(
ha_id, program_key=program, options=options
)
exception_translation_key = "start_program"
elif affects_to == AFFECTS_TO_SELECTED_PROGRAM:
method_call = client.set_selected_program(
ha_id, program_key=program, options=options
)
exception_translation_key = "select_program"
else:
array_of_options = ArrayOfOptions(options)
if affects_to == AFFECTS_TO_ACTIVE_PROGRAM:
method_call = client.set_active_program_options(
ha_id, array_of_options=array_of_options
)
exception_translation_key = "set_options_active_program"
else:
# affects_to is AFFECTS_TO_SELECTED_PROGRAM
method_call = client.set_selected_program_options(
ha_id, array_of_options=array_of_options
)
exception_translation_key = "set_options_selected_program"
try:
await method_call
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key=exception_translation_key,
translation_placeholders={
**get_dict_from_home_connect_error(err),
**({"program": program} if program else {}),
},
) from err
async def async_service_start_program(call: ServiceCall) -> None:
"""Service for starting a program."""
await _async_service_program(call, True)
hass.services.async_register(
DOMAIN,
SERVICE_OPTION_ACTIVE,
async_service_option_active,
schema=SERVICE_OPTION_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_OPTION_SELECTED,
async_service_option_selected,
schema=SERVICE_OPTION_SCHEMA,
)
hass.services.async_register(
DOMAIN, SERVICE_SETTING, async_service_setting, schema=SERVICE_SETTING_SCHEMA
)
hass.services.async_register(
DOMAIN,
SERVICE_PAUSE_PROGRAM,
async_service_pause_program,
schema=SERVICE_COMMAND_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_RESUME_PROGRAM,
async_service_resume_program,
schema=SERVICE_COMMAND_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_SELECT_PROGRAM,
async_service_select_program,
schema=SERVICE_PROGRAM_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_START_PROGRAM,
async_service_start_program,
schema=SERVICE_PROGRAM_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_SET_PROGRAM_AND_OPTIONS,
async_service_set_program_and_options,
schema=SERVICE_PROGRAM_AND_OPTIONS_SCHEMA,
)
register_actions(hass)
return True

View File

@ -0,0 +1,572 @@
"""Custom actions (previously known as services) for the Home Connect integration."""
from __future__ import annotations
from collections.abc import Awaitable
from typing import Any, cast
from aiohomeconnect.client import Client as HomeConnectClient
from aiohomeconnect.model import (
ArrayOfOptions,
CommandKey,
Option,
OptionKey,
ProgramKey,
SettingKey,
)
from aiohomeconnect.model.error import HomeConnectError
import voluptuous as vol
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from .const import (
AFFECTS_TO_ACTIVE_PROGRAM,
AFFECTS_TO_SELECTED_PROGRAM,
ATTR_AFFECTS_TO,
ATTR_KEY,
ATTR_PROGRAM,
ATTR_UNIT,
ATTR_VALUE,
DOMAIN,
PROGRAM_ENUM_OPTIONS,
SERVICE_OPTION_ACTIVE,
SERVICE_OPTION_SELECTED,
SERVICE_PAUSE_PROGRAM,
SERVICE_RESUME_PROGRAM,
SERVICE_SELECT_PROGRAM,
SERVICE_SET_PROGRAM_AND_OPTIONS,
SERVICE_SETTING,
SERVICE_START_PROGRAM,
TRANSLATION_KEYS_PROGRAMS_MAP,
)
from .coordinator import HomeConnectConfigEntry
from .utils import bsh_key_to_translation_key, get_dict_from_home_connect_error
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
PROGRAM_OPTIONS = {
bsh_key_to_translation_key(key): (
key,
value,
)
for key, value in {
OptionKey.BSH_COMMON_DURATION: int,
OptionKey.BSH_COMMON_START_IN_RELATIVE: int,
OptionKey.BSH_COMMON_FINISH_IN_RELATIVE: int,
OptionKey.CONSUMER_PRODUCTS_COFFEE_MAKER_FILL_QUANTITY: int,
OptionKey.CONSUMER_PRODUCTS_COFFEE_MAKER_MULTIPLE_BEVERAGES: bool,
OptionKey.DISHCARE_DISHWASHER_INTENSIV_ZONE: bool,
OptionKey.DISHCARE_DISHWASHER_BRILLIANCE_DRY: bool,
OptionKey.DISHCARE_DISHWASHER_VARIO_SPEED_PLUS: bool,
OptionKey.DISHCARE_DISHWASHER_SILENCE_ON_DEMAND: bool,
OptionKey.DISHCARE_DISHWASHER_HALF_LOAD: bool,
OptionKey.DISHCARE_DISHWASHER_EXTRA_DRY: bool,
OptionKey.DISHCARE_DISHWASHER_HYGIENE_PLUS: bool,
OptionKey.DISHCARE_DISHWASHER_ECO_DRY: bool,
OptionKey.DISHCARE_DISHWASHER_ZEOLITE_DRY: bool,
OptionKey.COOKING_OVEN_SETPOINT_TEMPERATURE: int,
OptionKey.COOKING_OVEN_FAST_PRE_HEAT: bool,
OptionKey.LAUNDRY_CARE_WASHER_I_DOS_1_ACTIVE: bool,
OptionKey.LAUNDRY_CARE_WASHER_I_DOS_2_ACTIVE: bool,
}.items()
}
SERVICE_SETTING_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_KEY): vol.All(
vol.Coerce(SettingKey),
vol.NotIn([SettingKey.UNKNOWN]),
),
vol.Required(ATTR_VALUE): vol.Any(str, int, bool),
}
)
# DEPRECATED: Remove in 2025.9.0
SERVICE_OPTION_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_KEY): vol.All(
vol.Coerce(OptionKey),
vol.NotIn([OptionKey.UNKNOWN]),
),
vol.Required(ATTR_VALUE): vol.Any(str, int, bool),
vol.Optional(ATTR_UNIT): str,
}
)
# DEPRECATED: Remove in 2025.9.0
SERVICE_PROGRAM_SCHEMA = vol.Any(
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_PROGRAM): vol.All(
vol.Coerce(ProgramKey),
vol.NotIn([ProgramKey.UNKNOWN]),
),
vol.Required(ATTR_KEY): vol.All(
vol.Coerce(OptionKey),
vol.NotIn([OptionKey.UNKNOWN]),
),
vol.Required(ATTR_VALUE): vol.Any(int, str),
vol.Optional(ATTR_UNIT): str,
},
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_PROGRAM): vol.All(
vol.Coerce(ProgramKey),
vol.NotIn([ProgramKey.UNKNOWN]),
),
},
)
def _require_program_or_at_least_one_option(data: dict) -> dict:
if ATTR_PROGRAM not in data and not any(
option_key in data for option_key in (PROGRAM_ENUM_OPTIONS | PROGRAM_OPTIONS)
):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="required_program_or_one_option_at_least",
)
return data
SERVICE_PROGRAM_AND_OPTIONS_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): str,
vol.Required(ATTR_AFFECTS_TO): vol.In(
[AFFECTS_TO_ACTIVE_PROGRAM, AFFECTS_TO_SELECTED_PROGRAM]
),
vol.Optional(ATTR_PROGRAM): vol.In(TRANSLATION_KEYS_PROGRAMS_MAP.keys()),
}
)
.extend(
{
vol.Optional(translation_key): vol.In(allowed_values.keys())
for translation_key, (
key,
allowed_values,
) in PROGRAM_ENUM_OPTIONS.items()
}
)
.extend(
{
vol.Optional(translation_key): schema
for translation_key, (key, schema) in PROGRAM_OPTIONS.items()
}
),
_require_program_or_at_least_one_option,
)
SERVICE_COMMAND_SCHEMA = vol.Schema({vol.Required(ATTR_DEVICE_ID): str})
async def _get_client_and_ha_id(
hass: HomeAssistant, device_id: str
) -> tuple[HomeConnectClient, str]:
device_registry = dr.async_get(hass)
device_entry = device_registry.async_get(device_id)
if device_entry is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="device_entry_not_found",
translation_placeholders={
"device_id": device_id,
},
)
entry: HomeConnectConfigEntry | None = None
for entry_id in device_entry.config_entries:
_entry = hass.config_entries.async_get_entry(entry_id)
assert _entry
if _entry.domain == DOMAIN:
entry = cast(HomeConnectConfigEntry, _entry)
break
if entry is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="config_entry_not_found",
translation_placeholders={
"device_id": device_id,
},
)
ha_id = next(
(
identifier[1]
for identifier in device_entry.identifiers
if identifier[0] == DOMAIN
),
None,
)
if ha_id is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="appliance_not_found",
translation_placeholders={
"device_id": device_id,
},
)
return entry.runtime_data.client, ha_id
async def _async_service_program(call: ServiceCall, start: bool) -> None:
"""Execute calls to services taking a program."""
program = call.data[ATTR_PROGRAM]
client, ha_id = await _get_client_and_ha_id(call.hass, call.data[ATTR_DEVICE_ID])
option_key = call.data.get(ATTR_KEY)
options = (
[
Option(
option_key,
call.data[ATTR_VALUE],
unit=call.data.get(ATTR_UNIT),
)
]
if option_key is not None
else None
)
async_create_issue(
call.hass,
DOMAIN,
"deprecated_set_program_and_option_actions",
breaks_in_ha_version="2025.9.0",
is_fixable=True,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="deprecated_set_program_and_option_actions",
translation_placeholders={
"new_action_key": SERVICE_SET_PROGRAM_AND_OPTIONS,
"remove_release": "2025.9.0",
"deprecated_action_yaml": "\n".join(
[
"```yaml",
f"action: {DOMAIN}.{SERVICE_START_PROGRAM if start else SERVICE_SELECT_PROGRAM}",
"data:",
f" {ATTR_DEVICE_ID}: DEVICE_ID",
f" {ATTR_PROGRAM}: {program}",
*([f" {ATTR_KEY}: {options[0].key}"] if options else []),
*([f" {ATTR_VALUE}: {options[0].value}"] if options else []),
*(
[f" {ATTR_UNIT}: {options[0].unit}"]
if options and options[0].unit
else []
),
"```",
]
),
"new_action_yaml": "\n ".join(
[
"```yaml",
f"action: {DOMAIN}.{SERVICE_SET_PROGRAM_AND_OPTIONS}",
"data:",
f" {ATTR_DEVICE_ID}: DEVICE_ID",
f" {ATTR_AFFECTS_TO}: {AFFECTS_TO_ACTIVE_PROGRAM if start else AFFECTS_TO_SELECTED_PROGRAM}",
f" {ATTR_PROGRAM}: {bsh_key_to_translation_key(program.value)}",
*(
[
f" {bsh_key_to_translation_key(options[0].key)}: {options[0].value}"
]
if options
else []
),
"```",
]
),
"repo_link": "[aiohomeconnect](https://github.com/MartinHjelmare/aiohomeconnect)",
},
)
try:
if start:
await client.start_program(ha_id, program_key=program, options=options)
else:
await client.set_selected_program(
ha_id, program_key=program, options=options
)
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="start_program" if start else "select_program",
translation_placeholders={
**get_dict_from_home_connect_error(err),
"program": program,
},
) from err
async def _async_service_set_program_options(call: ServiceCall, active: bool) -> None:
"""Execute calls to services taking a program."""
option_key = call.data[ATTR_KEY]
value = call.data[ATTR_VALUE]
unit = call.data.get(ATTR_UNIT)
client, ha_id = await _get_client_and_ha_id(call.hass, call.data[ATTR_DEVICE_ID])
async_create_issue(
call.hass,
DOMAIN,
"deprecated_set_program_and_option_actions",
breaks_in_ha_version="2025.9.0",
is_fixable=True,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="deprecated_set_program_and_option_actions",
translation_placeholders={
"new_action_key": SERVICE_SET_PROGRAM_AND_OPTIONS,
"remove_release": "2025.9.0",
"deprecated_action_yaml": "\n".join(
[
"```yaml",
f"action: {DOMAIN}.{SERVICE_OPTION_ACTIVE if active else SERVICE_OPTION_SELECTED}",
"data:",
f" {ATTR_DEVICE_ID}: DEVICE_ID",
f" {ATTR_KEY}: {option_key}",
f" {ATTR_VALUE}: {value}",
*([f" {ATTR_UNIT}: {unit}"] if unit else []),
"```",
]
),
"new_action_yaml": "\n ".join(
[
"```yaml",
f"action: {DOMAIN}.{SERVICE_SET_PROGRAM_AND_OPTIONS}",
"data:",
f" {ATTR_DEVICE_ID}: DEVICE_ID",
f" {ATTR_AFFECTS_TO}: {AFFECTS_TO_ACTIVE_PROGRAM if active else AFFECTS_TO_SELECTED_PROGRAM}",
f" {bsh_key_to_translation_key(option_key)}: {value}",
"```",
]
),
"repo_link": "[aiohomeconnect](https://github.com/MartinHjelmare/aiohomeconnect)",
},
)
try:
if active:
await client.set_active_program_option(
ha_id,
option_key=option_key,
value=value,
unit=unit,
)
else:
await client.set_selected_program_option(
ha_id,
option_key=option_key,
value=value,
unit=unit,
)
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="set_options_active_program"
if active
else "set_options_selected_program",
translation_placeholders={
**get_dict_from_home_connect_error(err),
"key": option_key,
"value": str(value),
},
) from err
async def _async_service_command(call: ServiceCall, command_key: CommandKey) -> None:
"""Execute calls to services executing a command."""
client, ha_id = await _get_client_and_ha_id(call.hass, call.data[ATTR_DEVICE_ID])
async_create_issue(
call.hass,
DOMAIN,
"deprecated_command_actions",
breaks_in_ha_version="2025.9.0",
is_fixable=True,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="deprecated_command_actions",
)
try:
await client.put_command(ha_id, command_key=command_key, value=True)
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="execute_command",
translation_placeholders={
**get_dict_from_home_connect_error(err),
"command": command_key.value,
},
) from err
async def async_service_option_active(call: ServiceCall) -> None:
"""Service for setting an option for an active program."""
await _async_service_set_program_options(call, True)
async def async_service_option_selected(call: ServiceCall) -> None:
"""Service for setting an option for a selected program."""
await _async_service_set_program_options(call, False)
async def async_service_setting(call: ServiceCall) -> None:
"""Service for changing a setting."""
key = call.data[ATTR_KEY]
value = call.data[ATTR_VALUE]
client, ha_id = await _get_client_and_ha_id(call.hass, call.data[ATTR_DEVICE_ID])
try:
await client.set_setting(ha_id, setting_key=key, value=value)
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="set_setting",
translation_placeholders={
**get_dict_from_home_connect_error(err),
"key": key,
"value": str(value),
},
) from err
async def async_service_pause_program(call: ServiceCall) -> None:
"""Service for pausing a program."""
await _async_service_command(call, CommandKey.BSH_COMMON_PAUSE_PROGRAM)
async def async_service_resume_program(call: ServiceCall) -> None:
"""Service for resuming a paused program."""
await _async_service_command(call, CommandKey.BSH_COMMON_RESUME_PROGRAM)
async def async_service_select_program(call: ServiceCall) -> None:
"""Service for selecting a program."""
await _async_service_program(call, False)
async def async_service_set_program_and_options(call: ServiceCall) -> None:
"""Service for setting a program and options."""
data = dict(call.data)
program = data.pop(ATTR_PROGRAM, None)
affects_to = data.pop(ATTR_AFFECTS_TO)
client, ha_id = await _get_client_and_ha_id(call.hass, data.pop(ATTR_DEVICE_ID))
options: list[Option] = []
for option, value in data.items():
if option in PROGRAM_ENUM_OPTIONS:
options.append(
Option(
PROGRAM_ENUM_OPTIONS[option][0],
PROGRAM_ENUM_OPTIONS[option][1][value],
)
)
elif option in PROGRAM_OPTIONS:
option_key = PROGRAM_OPTIONS[option][0]
options.append(Option(option_key, value))
method_call: Awaitable[Any]
exception_translation_key: str
if program:
program = (
program
if isinstance(program, ProgramKey)
else TRANSLATION_KEYS_PROGRAMS_MAP[program]
)
if affects_to == AFFECTS_TO_ACTIVE_PROGRAM:
method_call = client.start_program(
ha_id, program_key=program, options=options
)
exception_translation_key = "start_program"
elif affects_to == AFFECTS_TO_SELECTED_PROGRAM:
method_call = client.set_selected_program(
ha_id, program_key=program, options=options
)
exception_translation_key = "select_program"
else:
array_of_options = ArrayOfOptions(options)
if affects_to == AFFECTS_TO_ACTIVE_PROGRAM:
method_call = client.set_active_program_options(
ha_id, array_of_options=array_of_options
)
exception_translation_key = "set_options_active_program"
else:
# affects_to is AFFECTS_TO_SELECTED_PROGRAM
method_call = client.set_selected_program_options(
ha_id, array_of_options=array_of_options
)
exception_translation_key = "set_options_selected_program"
try:
await method_call
except HomeConnectError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key=exception_translation_key,
translation_placeholders={
**get_dict_from_home_connect_error(err),
**({"program": program} if program else {}),
},
) from err
async def async_service_start_program(call: ServiceCall) -> None:
"""Service for starting a program."""
await _async_service_program(call, True)
def register_actions(hass: HomeAssistant) -> None:
"""Register custom actions."""
hass.services.async_register(
DOMAIN,
SERVICE_OPTION_ACTIVE,
async_service_option_active,
schema=SERVICE_OPTION_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_OPTION_SELECTED,
async_service_option_selected,
schema=SERVICE_OPTION_SCHEMA,
)
hass.services.async_register(
DOMAIN, SERVICE_SETTING, async_service_setting, schema=SERVICE_SETTING_SCHEMA
)
hass.services.async_register(
DOMAIN,
SERVICE_PAUSE_PROGRAM,
async_service_pause_program,
schema=SERVICE_COMMAND_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_RESUME_PROGRAM,
async_service_resume_program,
schema=SERVICE_COMMAND_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_SELECT_PROGRAM,
async_service_select_program,
schema=SERVICE_PROGRAM_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_START_PROGRAM,
async_service_start_program,
schema=SERVICE_PROGRAM_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_SET_PROGRAM_AND_OPTIONS,
async_service_set_program_and_options,
schema=SERVICE_PROGRAM_AND_OPTIONS_SCHEMA,
)

View File

@ -1,12 +1,11 @@
"""Test the integration init functionality."""
from collections.abc import Awaitable, Callable
from http import HTTPStatus
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from aiohomeconnect.const import OAUTH2_TOKEN
from aiohomeconnect.model import OptionKey, ProgramKey, SettingKey, StatusKey
from aiohomeconnect.model import SettingKey, StatusKey
from aiohomeconnect.model.error import (
HomeConnectError,
TooManyRequestsError,
@ -14,7 +13,6 @@ from aiohomeconnect.model.error import (
)
import aiohttp
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.home_connect.const import DOMAIN
@ -25,9 +23,8 @@ from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import device_registry as dr, entity_registry as er
import homeassistant.helpers.issue_registry as ir
from script.hassfest.translations import RE_TRANSLATION_KEY
from .conftest import (
@ -40,157 +37,6 @@ from .conftest import (
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
DEPRECATED_SERVICE_KV_CALL_PARAMS = [
{
"domain": DOMAIN,
"service": "set_option_active",
"service_data": {
"device_id": "DEVICE_ID",
"key": OptionKey.BSH_COMMON_FINISH_IN_RELATIVE.value,
"value": 43200,
"unit": "seconds",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_option_selected",
"service_data": {
"device_id": "DEVICE_ID",
"key": OptionKey.LAUNDRY_CARE_WASHER_TEMPERATURE.value,
"value": "LaundryCare.Washer.EnumType.Temperature.GC40",
},
"blocking": True,
},
]
SERVICE_KV_CALL_PARAMS = [
*DEPRECATED_SERVICE_KV_CALL_PARAMS,
{
"domain": DOMAIN,
"service": "change_setting",
"service_data": {
"device_id": "DEVICE_ID",
"key": SettingKey.BSH_COMMON_CHILD_LOCK.value,
"value": True,
},
"blocking": True,
},
]
SERVICE_COMMAND_CALL_PARAMS = [
{
"domain": DOMAIN,
"service": "pause_program",
"service_data": {
"device_id": "DEVICE_ID",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "resume_program",
"service_data": {
"device_id": "DEVICE_ID",
},
"blocking": True,
},
]
SERVICE_PROGRAM_CALL_PARAMS = [
{
"domain": DOMAIN,
"service": "select_program",
"service_data": {
"device_id": "DEVICE_ID",
"program": ProgramKey.LAUNDRY_CARE_WASHER_COTTON.value,
"key": OptionKey.LAUNDRY_CARE_WASHER_TEMPERATURE.value,
"value": "LaundryCare.Washer.EnumType.Temperature.GC40",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "start_program",
"service_data": {
"device_id": "DEVICE_ID",
"program": ProgramKey.LAUNDRY_CARE_WASHER_COTTON.value,
"key": OptionKey.BSH_COMMON_FINISH_IN_RELATIVE.value,
"value": 43200,
"unit": "seconds",
},
"blocking": True,
},
]
SERVICE_APPLIANCE_METHOD_MAPPING = {
"set_option_active": "set_active_program_option",
"set_option_selected": "set_selected_program_option",
"change_setting": "set_setting",
"pause_program": "put_command",
"resume_program": "put_command",
"select_program": "set_selected_program",
"start_program": "start_program",
}
SERVICE_VALIDATION_ERROR_MAPPING = {
"set_option_active": r"Error.*setting.*options.*active.*program.*",
"set_option_selected": r"Error.*setting.*options.*selected.*program.*",
"change_setting": r"Error.*assigning.*value.*setting.*",
"pause_program": r"Error.*executing.*command.*",
"resume_program": r"Error.*executing.*command.*",
"select_program": r"Error.*selecting.*program.*",
"start_program": r"Error.*starting.*program.*",
}
SERVICES_SET_PROGRAM_AND_OPTIONS = [
{
"domain": DOMAIN,
"service": "set_program_and_options",
"service_data": {
"device_id": "DEVICE_ID",
"affects_to": "selected_program",
"program": "dishcare_dishwasher_program_eco_50",
"b_s_h_common_option_start_in_relative": 1800,
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_program_and_options",
"service_data": {
"device_id": "DEVICE_ID",
"affects_to": "active_program",
"program": "consumer_products_coffee_maker_program_beverage_coffee",
"consumer_products_coffee_maker_option_bean_amount": "consumer_products_coffee_maker_enum_type_bean_amount_normal",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_program_and_options",
"service_data": {
"device_id": "DEVICE_ID",
"affects_to": "active_program",
"consumer_products_coffee_maker_option_coffee_milk_ratio": "consumer_products_coffee_maker_enum_type_coffee_milk_ratio_50_percent",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_program_and_options",
"service_data": {
"device_id": "DEVICE_ID",
"affects_to": "selected_program",
"consumer_products_coffee_maker_option_fill_quantity": 35,
},
"blocking": True,
},
]
async def test_entry_setup(
@ -401,197 +247,6 @@ async def test_client_rate_limit_error(
asyncio_sleep_mock.assert_called_once_with(retry_after)
@pytest.mark.parametrize(
"service_call",
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
)
async def test_key_value_services(
service_call: dict[str, Any],
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
appliance_ha_id: str,
) -> None:
"""Create and test services."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_name = service_call["service"]
service_call["service_data"]["device_id"] = device_entry.id
await hass.services.async_call(**service_call)
await hass.async_block_till_done()
assert (
getattr(client, SERVICE_APPLIANCE_METHOD_MAPPING[service_name]).call_count == 1
)
@pytest.mark.parametrize(
("service_call", "issue_id"),
[
*zip(
DEPRECATED_SERVICE_KV_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
["deprecated_set_program_and_option_actions"]
* (
len(DEPRECATED_SERVICE_KV_CALL_PARAMS)
+ len(SERVICE_PROGRAM_CALL_PARAMS)
),
strict=True,
),
*zip(
SERVICE_COMMAND_CALL_PARAMS,
["deprecated_command_actions"] * len(SERVICE_COMMAND_CALL_PARAMS),
strict=True,
),
],
)
async def test_programs_and_options_actions_deprecation(
service_call: dict[str, Any],
issue_id: str,
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
appliance_ha_id: str,
issue_registry: ir.IssueRegistry,
hass_client: ClientSessionGenerator,
) -> None:
"""Test deprecated service keys."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
await hass.services.async_call(**service_call)
await hass.async_block_till_done()
assert len(issue_registry.issues) == 1
issue = issue_registry.async_get_issue(DOMAIN, issue_id)
assert issue
_client = await hass_client()
resp = await _client.post(
"/api/repairs/issues/fix",
json={"handler": DOMAIN, "issue_id": issue.issue_id},
)
assert resp.status == HTTPStatus.OK
flow_id = (await resp.json())["flow_id"]
resp = await _client.post(f"/api/repairs/issues/fix/{flow_id}")
assert not issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 0
await hass.services.async_call(**service_call)
await hass.async_block_till_done()
assert len(issue_registry.issues) == 1
assert issue_registry.async_get_issue(DOMAIN, issue_id)
await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
# Assert the issue is no longer present
assert not issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 0
@pytest.mark.parametrize(
("service_call", "called_method"),
zip(
SERVICES_SET_PROGRAM_AND_OPTIONS,
[
"set_selected_program",
"start_program",
"set_active_program_options",
"set_selected_program_options",
],
strict=True,
),
)
async def test_set_program_and_options(
service_call: dict[str, Any],
called_method: str,
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
appliance_ha_id: str,
snapshot: SnapshotAssertion,
) -> None:
"""Test recognized options."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
await hass.services.async_call(**service_call)
await hass.async_block_till_done()
method_mock: MagicMock = getattr(client, called_method)
assert method_mock.call_count == 1
assert method_mock.call_args == snapshot
@pytest.mark.parametrize(
("service_call", "error_regex"),
zip(
SERVICES_SET_PROGRAM_AND_OPTIONS,
[
r"Error.*selecting.*program.*",
r"Error.*starting.*program.*",
r"Error.*setting.*options.*active.*program.*",
r"Error.*setting.*options.*selected.*program.*",
],
strict=True,
),
)
async def test_set_program_and_options_exceptions(
service_call: dict[str, Any],
error_regex: str,
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client_with_exception: MagicMock,
appliance_ha_id: str,
) -> None:
"""Test recognized options."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client_with_exception)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
with pytest.raises(HomeAssistantError, match=error_regex):
await hass.services.async_call(**service_call)
async def test_required_program_or_at_least_an_option(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
@ -626,113 +281,6 @@ async def test_required_program_or_at_least_an_option(
)
@pytest.mark.parametrize(
"service_call",
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
)
async def test_services_exception_device_id(
service_call: dict[str, Any],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client_with_exception: MagicMock,
appliance_ha_id: str,
device_registry: dr.DeviceRegistry,
) -> None:
"""Raise a HomeAssistantError when there is an API error."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client_with_exception)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
with pytest.raises(HomeAssistantError):
await hass.services.async_call(**service_call)
async def test_services_appliance_not_found(
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
) -> None:
"""Raise a ServiceValidationError when device id does not match."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
service_call = SERVICE_KV_CALL_PARAMS[0]
service_call["service_data"]["device_id"] = "DOES_NOT_EXISTS"
with pytest.raises(ServiceValidationError, match=r"Device entry.*not found"):
await hass.services.async_call(**service_call)
unrelated_config_entry = MockConfigEntry(
domain="TEST",
)
unrelated_config_entry.add_to_hass(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=unrelated_config_entry.entry_id,
identifiers={("RANDOM", "ABCD")},
)
service_call["service_data"]["device_id"] = device_entry.id
with pytest.raises(ServiceValidationError, match=r"Config entry.*not found"):
await hass.services.async_call(**service_call)
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={("RANDOM", "ABCD")},
)
service_call["service_data"]["device_id"] = device_entry.id
with pytest.raises(ServiceValidationError, match=r"Appliance.*not found"):
await hass.services.async_call(**service_call)
@pytest.mark.parametrize(
"service_call",
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
)
async def test_services_exception(
service_call: dict[str, Any],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client_with_exception: MagicMock,
appliance_ha_id: str,
device_registry: dr.DeviceRegistry,
) -> None:
"""Raise a ValueError when device id does not match."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client_with_exception)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
service_name = service_call["service"]
with pytest.raises(
HomeAssistantError,
match=SERVICE_VALIDATION_ERROR_MAPPING[service_name],
):
await hass.services.async_call(**service_call)
async def test_entity_migration(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,

View File

@ -0,0 +1,468 @@
"""Tests for the Home Connect actions."""
from collections.abc import Awaitable, Callable
from http import HTTPStatus
from typing import Any
from unittest.mock import MagicMock
from aiohomeconnect.model import OptionKey, ProgramKey, SettingKey
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.home_connect.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr
import homeassistant.helpers.issue_registry as ir
from tests.common import MockConfigEntry
from tests.typing import ClientSessionGenerator
DEPRECATED_SERVICE_KV_CALL_PARAMS = [
{
"domain": DOMAIN,
"service": "set_option_active",
"service_data": {
"device_id": "DEVICE_ID",
"key": OptionKey.BSH_COMMON_FINISH_IN_RELATIVE.value,
"value": 43200,
"unit": "seconds",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_option_selected",
"service_data": {
"device_id": "DEVICE_ID",
"key": OptionKey.LAUNDRY_CARE_WASHER_TEMPERATURE.value,
"value": "LaundryCare.Washer.EnumType.Temperature.GC40",
},
"blocking": True,
},
]
SERVICE_KV_CALL_PARAMS = [
*DEPRECATED_SERVICE_KV_CALL_PARAMS,
{
"domain": DOMAIN,
"service": "change_setting",
"service_data": {
"device_id": "DEVICE_ID",
"key": SettingKey.BSH_COMMON_CHILD_LOCK.value,
"value": True,
},
"blocking": True,
},
]
SERVICE_COMMAND_CALL_PARAMS = [
{
"domain": DOMAIN,
"service": "pause_program",
"service_data": {
"device_id": "DEVICE_ID",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "resume_program",
"service_data": {
"device_id": "DEVICE_ID",
},
"blocking": True,
},
]
SERVICE_PROGRAM_CALL_PARAMS = [
{
"domain": DOMAIN,
"service": "select_program",
"service_data": {
"device_id": "DEVICE_ID",
"program": ProgramKey.LAUNDRY_CARE_WASHER_COTTON.value,
"key": OptionKey.LAUNDRY_CARE_WASHER_TEMPERATURE.value,
"value": "LaundryCare.Washer.EnumType.Temperature.GC40",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "start_program",
"service_data": {
"device_id": "DEVICE_ID",
"program": ProgramKey.LAUNDRY_CARE_WASHER_COTTON.value,
"key": OptionKey.BSH_COMMON_FINISH_IN_RELATIVE.value,
"value": 43200,
"unit": "seconds",
},
"blocking": True,
},
]
SERVICE_APPLIANCE_METHOD_MAPPING = {
"set_option_active": "set_active_program_option",
"set_option_selected": "set_selected_program_option",
"change_setting": "set_setting",
"pause_program": "put_command",
"resume_program": "put_command",
"select_program": "set_selected_program",
"start_program": "start_program",
}
SERVICE_VALIDATION_ERROR_MAPPING = {
"set_option_active": r"Error.*setting.*options.*active.*program.*",
"set_option_selected": r"Error.*setting.*options.*selected.*program.*",
"change_setting": r"Error.*assigning.*value.*setting.*",
"pause_program": r"Error.*executing.*command.*",
"resume_program": r"Error.*executing.*command.*",
"select_program": r"Error.*selecting.*program.*",
"start_program": r"Error.*starting.*program.*",
}
SERVICES_SET_PROGRAM_AND_OPTIONS = [
{
"domain": DOMAIN,
"service": "set_program_and_options",
"service_data": {
"device_id": "DEVICE_ID",
"affects_to": "selected_program",
"program": "dishcare_dishwasher_program_eco_50",
"b_s_h_common_option_start_in_relative": 1800,
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_program_and_options",
"service_data": {
"device_id": "DEVICE_ID",
"affects_to": "active_program",
"program": "consumer_products_coffee_maker_program_beverage_coffee",
"consumer_products_coffee_maker_option_bean_amount": "consumer_products_coffee_maker_enum_type_bean_amount_normal",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_program_and_options",
"service_data": {
"device_id": "DEVICE_ID",
"affects_to": "active_program",
"consumer_products_coffee_maker_option_coffee_milk_ratio": "consumer_products_coffee_maker_enum_type_coffee_milk_ratio_50_percent",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_program_and_options",
"service_data": {
"device_id": "DEVICE_ID",
"affects_to": "selected_program",
"consumer_products_coffee_maker_option_fill_quantity": 35,
},
"blocking": True,
},
]
@pytest.mark.parametrize(
"service_call",
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
)
async def test_key_value_services(
service_call: dict[str, Any],
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
appliance_ha_id: str,
) -> None:
"""Create and test services."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_name = service_call["service"]
service_call["service_data"]["device_id"] = device_entry.id
await hass.services.async_call(**service_call)
await hass.async_block_till_done()
assert (
getattr(client, SERVICE_APPLIANCE_METHOD_MAPPING[service_name]).call_count == 1
)
@pytest.mark.parametrize(
("service_call", "issue_id"),
[
*zip(
DEPRECATED_SERVICE_KV_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
["deprecated_set_program_and_option_actions"]
* (
len(DEPRECATED_SERVICE_KV_CALL_PARAMS)
+ len(SERVICE_PROGRAM_CALL_PARAMS)
),
strict=True,
),
*zip(
SERVICE_COMMAND_CALL_PARAMS,
["deprecated_command_actions"] * len(SERVICE_COMMAND_CALL_PARAMS),
strict=True,
),
],
)
async def test_programs_and_options_actions_deprecation(
service_call: dict[str, Any],
issue_id: str,
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
appliance_ha_id: str,
issue_registry: ir.IssueRegistry,
hass_client: ClientSessionGenerator,
) -> None:
"""Test deprecated service keys."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
await hass.services.async_call(**service_call)
await hass.async_block_till_done()
assert len(issue_registry.issues) == 1
issue = issue_registry.async_get_issue(DOMAIN, issue_id)
assert issue
_client = await hass_client()
resp = await _client.post(
"/api/repairs/issues/fix",
json={"handler": DOMAIN, "issue_id": issue.issue_id},
)
assert resp.status == HTTPStatus.OK
flow_id = (await resp.json())["flow_id"]
resp = await _client.post(f"/api/repairs/issues/fix/{flow_id}")
assert not issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 0
await hass.services.async_call(**service_call)
await hass.async_block_till_done()
assert len(issue_registry.issues) == 1
assert issue_registry.async_get_issue(DOMAIN, issue_id)
await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
# Assert the issue is no longer present
assert not issue_registry.async_get_issue(DOMAIN, issue_id)
assert len(issue_registry.issues) == 0
@pytest.mark.parametrize(
("service_call", "called_method"),
zip(
SERVICES_SET_PROGRAM_AND_OPTIONS,
[
"set_selected_program",
"start_program",
"set_active_program_options",
"set_selected_program_options",
],
strict=True,
),
)
async def test_set_program_and_options(
service_call: dict[str, Any],
called_method: str,
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
appliance_ha_id: str,
snapshot: SnapshotAssertion,
) -> None:
"""Test recognized options."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
await hass.services.async_call(**service_call)
await hass.async_block_till_done()
method_mock: MagicMock = getattr(client, called_method)
assert method_mock.call_count == 1
assert method_mock.call_args == snapshot
@pytest.mark.parametrize(
("service_call", "error_regex"),
zip(
SERVICES_SET_PROGRAM_AND_OPTIONS,
[
r"Error.*selecting.*program.*",
r"Error.*starting.*program.*",
r"Error.*setting.*options.*active.*program.*",
r"Error.*setting.*options.*selected.*program.*",
],
strict=True,
),
)
async def test_set_program_and_options_exceptions(
service_call: dict[str, Any],
error_regex: str,
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client_with_exception: MagicMock,
appliance_ha_id: str,
) -> None:
"""Test recognized options."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client_with_exception)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
with pytest.raises(HomeAssistantError, match=error_regex):
await hass.services.async_call(**service_call)
@pytest.mark.parametrize(
"service_call",
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
)
async def test_services_exception_device_id(
service_call: dict[str, Any],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client_with_exception: MagicMock,
appliance_ha_id: str,
device_registry: dr.DeviceRegistry,
) -> None:
"""Raise a HomeAssistantError when there is an API error."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client_with_exception)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
with pytest.raises(HomeAssistantError):
await hass.services.async_call(**service_call)
async def test_services_appliance_not_found(
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
device_registry: dr.DeviceRegistry,
) -> None:
"""Raise a ServiceValidationError when device id does not match."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client)
assert config_entry.state == ConfigEntryState.LOADED
service_call = SERVICE_KV_CALL_PARAMS[0]
service_call["service_data"]["device_id"] = "DOES_NOT_EXISTS"
with pytest.raises(ServiceValidationError, match=r"Device entry.*not found"):
await hass.services.async_call(**service_call)
unrelated_config_entry = MockConfigEntry(
domain="TEST",
)
unrelated_config_entry.add_to_hass(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=unrelated_config_entry.entry_id,
identifiers={("RANDOM", "ABCD")},
)
service_call["service_data"]["device_id"] = device_entry.id
with pytest.raises(ServiceValidationError, match=r"Config entry.*not found"):
await hass.services.async_call(**service_call)
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={("RANDOM", "ABCD")},
)
service_call["service_data"]["device_id"] = device_entry.id
with pytest.raises(ServiceValidationError, match=r"Appliance.*not found"):
await hass.services.async_call(**service_call)
@pytest.mark.parametrize(
"service_call",
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
)
async def test_services_exception(
service_call: dict[str, Any],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client_with_exception: MagicMock,
appliance_ha_id: str,
device_registry: dr.DeviceRegistry,
) -> None:
"""Raise a ValueError when device id does not match."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup(client_with_exception)
assert config_entry.state == ConfigEntryState.LOADED
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance_ha_id)},
)
service_call["service_data"]["device_id"] = device_entry.id
service_name = service_call["service"]
with pytest.raises(
HomeAssistantError,
match=SERVICE_VALIDATION_ERROR_MAPPING[service_name],
):
await hass.services.async_call(**service_call)