461 lines
15 KiB
Python
461 lines
15 KiB
Python
"""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 HomeAppliance, 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("appliance", ["Washer"], indirect=True)
|
|
@pytest.mark.parametrize(
|
|
"service_call",
|
|
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
|
|
)
|
|
async def test_key_value_services(
|
|
hass: HomeAssistant,
|
|
device_registry: dr.DeviceRegistry,
|
|
client: MagicMock,
|
|
config_entry: MockConfigEntry,
|
|
integration_setup: Callable[[MagicMock], Awaitable[bool]],
|
|
appliance: HomeAppliance,
|
|
service_call: dict[str, Any],
|
|
) -> None:
|
|
"""Create and test services."""
|
|
assert await integration_setup(client)
|
|
assert config_entry.state is 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("appliance", ["Washer"], indirect=True)
|
|
@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(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
device_registry: dr.DeviceRegistry,
|
|
issue_registry: ir.IssueRegistry,
|
|
client: MagicMock,
|
|
config_entry: MockConfigEntry,
|
|
integration_setup: Callable[[MagicMock], Awaitable[bool]],
|
|
appliance: HomeAppliance,
|
|
service_call: dict[str, Any],
|
|
issue_id: str,
|
|
) -> None:
|
|
"""Test deprecated service keys."""
|
|
assert await integration_setup(client)
|
|
assert config_entry.state is 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("appliance", ["Washer"], indirect=True)
|
|
@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(
|
|
hass: HomeAssistant,
|
|
device_registry: dr.DeviceRegistry,
|
|
client: MagicMock,
|
|
config_entry: MockConfigEntry,
|
|
integration_setup: Callable[[MagicMock], Awaitable[bool]],
|
|
appliance: HomeAppliance,
|
|
service_call: dict[str, Any],
|
|
called_method: str,
|
|
snapshot: SnapshotAssertion,
|
|
) -> None:
|
|
"""Test recognized options."""
|
|
assert await integration_setup(client)
|
|
assert config_entry.state is 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("appliance", ["Washer"], indirect=True)
|
|
@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(
|
|
hass: HomeAssistant,
|
|
device_registry: dr.DeviceRegistry,
|
|
client_with_exception: MagicMock,
|
|
config_entry: MockConfigEntry,
|
|
integration_setup: Callable[[MagicMock], Awaitable[bool]],
|
|
appliance: HomeAppliance,
|
|
service_call: dict[str, Any],
|
|
error_regex: str,
|
|
) -> None:
|
|
"""Test recognized options."""
|
|
assert await integration_setup(client_with_exception)
|
|
assert config_entry.state is 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("appliance", ["Washer"], indirect=True)
|
|
@pytest.mark.parametrize(
|
|
"service_call",
|
|
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
|
|
)
|
|
async def test_services_exception_device_id(
|
|
hass: HomeAssistant,
|
|
device_registry: dr.DeviceRegistry,
|
|
client_with_exception: MagicMock,
|
|
config_entry: MockConfigEntry,
|
|
integration_setup: Callable[[MagicMock], Awaitable[bool]],
|
|
appliance: HomeAppliance,
|
|
service_call: dict[str, Any],
|
|
) -> None:
|
|
"""Raise a HomeAssistantError when there is an API error."""
|
|
assert await integration_setup(client_with_exception)
|
|
assert config_entry.state is 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,
|
|
device_registry: dr.DeviceRegistry,
|
|
client: MagicMock,
|
|
config_entry: MockConfigEntry,
|
|
integration_setup: Callable[[MagicMock], Awaitable[bool]],
|
|
) -> None:
|
|
"""Raise a ServiceValidationError when device id does not match."""
|
|
assert await integration_setup(client)
|
|
assert config_entry.state is 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("appliance", ["Washer"], indirect=True)
|
|
@pytest.mark.parametrize(
|
|
"service_call",
|
|
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
|
|
)
|
|
async def test_services_exception(
|
|
hass: HomeAssistant,
|
|
device_registry: dr.DeviceRegistry,
|
|
client_with_exception: MagicMock,
|
|
config_entry: MockConfigEntry,
|
|
integration_setup: Callable[[MagicMock], Awaitable[bool]],
|
|
appliance: HomeAppliance,
|
|
service_call: dict[str, Any],
|
|
) -> None:
|
|
"""Raise a ValueError when device id does not match."""
|
|
assert await integration_setup(client_with_exception)
|
|
assert config_entry.state is 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)
|