"""Tests for the Home Connect actions.""" from collections.abc import Awaitable, Callable from typing import Any from unittest.mock import AsyncMock, MagicMock from aiohomeconnect.model import ( HomeAppliance, Option, OptionKey, Program, ProgramKey, SettingKey, ) from aiohomeconnect.model.error import HomeConnectError, NoProgramActiveError import pytest from syrupy.assertion import SnapshotAssertion from voluptuous.error import MultipleInvalid from homeassistant.components import home_connect from homeassistant.components.home_connect.const import ( DOMAIN, PROGRAM_ENUM_OPTIONS, TRANSLATION_KEYS_PROGRAMS_MAP, ) from homeassistant.components.home_connect.services import PROGRAM_OPTIONS from homeassistant.components.home_connect.utils import bsh_key_to_translation_key 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 from homeassistant.util.yaml import load_yaml_dict from tests.common import MockConfigEntry 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_APPLIANCE_METHOD_MAPPING = { "change_setting": "set_setting", } SERVICE_VALIDATION_ERROR_MAPPING = { "change_setting": r"Error.*assigning.*value.*setting.*", } 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, }, ] def test_services_yaml_set_program_and_options_program_keys() -> None: """Test that all program keys in services.yaml exist in the translation map.""" services = load_yaml_dict(f"{home_connect.__path__[0]}/services.yaml") yaml_programs = set( services["set_program_and_options"]["fields"]["program"]["selector"]["select"][ "options" ] ) assert yaml_programs <= set(TRANSLATION_KEYS_PROGRAMS_MAP.keys()) def test_services_yaml_set_program_and_options_option_keys() -> None: """Test that all program keys in services.yaml exist in the translation map.""" services = load_yaml_dict(f"{home_connect.__path__[0]}/services.yaml") groups = services["set_program_and_options"]["fields"] groups.pop("device_id") groups.pop("affects_to") groups.pop("program") for group in groups.values(): for option, option_data in group["fields"].items(): assert option in PROGRAM_ENUM_OPTIONS or option in PROGRAM_OPTIONS, ( f"{option} is missing from both" " PROGRAM_ENUM_OPTIONS and PROGRAM_OPTIONS" ) if option in PROGRAM_ENUM_OPTIONS: enum_values = set(PROGRAM_ENUM_OPTIONS[option][1]) assert enum_values == set( option_data["selector"]["select"]["options"] ), ( f"Options for {option} do not match between" " services.yaml and constants.py" ) assert "example" in option_data, ( f"Example value for {option} is missing" ) assert option_data["example"] in enum_values, ( f"Example value for {option} is not a valid option" ) @pytest.mark.parametrize("appliance", ["Washer"], indirect=True) @pytest.mark.parametrize("service_call", SERVICE_KV_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", "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", ["Dishwasher"], indirect=True) @pytest.mark.parametrize( "additional_service_data", [ {}, { "b_s_h_common_option_start_in_relative": 1200, "b_s_h_common_option_finish_in_relative": 1200, }, { "b_s_h_common_option_start_in_relative": 1200, }, { "b_s_h_common_option_finish_in_relative": 1200, }, ], ) @pytest.mark.parametrize( "options_already_set", [ None, [ Option( key=OptionKey.DISHCARE_DISHWASHER_HALF_LOAD, value=True, ) ], ], ) @pytest.mark.parametrize( ("get_active_program_side_effect", "get_selected_program_call_count"), [(None, 0), (NoProgramActiveError("error.key"), 1)], ) async def test_start_selected_program( hass: HomeAssistant, device_registry: dr.DeviceRegistry, client: MagicMock, config_entry: MockConfigEntry, integration_setup: Callable[[MagicMock], Awaitable[bool]], appliance: HomeAppliance, additional_service_data: dict[str, Any], options_already_set: list[Option] | None, get_active_program_side_effect: NoProgramActiveError | None, get_selected_program_call_count: int, snapshot: SnapshotAssertion, ) -> None: """Test starting the selected program with optional parameter overrides.""" client.get_active_program = AsyncMock( return_value=Program( key=ProgramKey.DISHCARE_DISHWASHER_ECO_50, options=options_already_set, ), side_effect=get_active_program_side_effect, ) client.get_selected_program = AsyncMock( return_value=Program( key=ProgramKey.DISHCARE_DISHWASHER_ECO_50, options=options_already_set, ) ) 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)}, ) await hass.services.async_call( domain=DOMAIN, service="start_selected_program", service_data={ "device_id": device_entry.id, **additional_service_data, }, blocking=True, ) await hass.async_block_till_done() client.get_active_program.assert_awaited_once_with(appliance.ha_id) assert client.get_selected_program.call_count == get_selected_program_call_count for call_args in client.start_program.call_args_list: assert call_args[0][0] == appliance.ha_id assert client.start_program.call_count == 1 assert client.start_program.call_args == snapshot @pytest.mark.parametrize("appliance", ["Dishwasher"], indirect=True) @pytest.mark.parametrize( ("mock_attr", "error_regex", "get_active_program_side_effect"), [ ( "get_active_program", r"Error.*obtaining.*program.*", None, ), ( "get_selected_program", r"Error.*obtaining.*program.*", NoProgramActiveError("error.key"), ), ("start_program", r"Error.*starting.*program.*", None), ], ) async def test_start_selected_program_and_options_exceptions( hass: HomeAssistant, device_registry: dr.DeviceRegistry, client: MagicMock, config_entry: MockConfigEntry, integration_setup: Callable[[MagicMock], Awaitable[bool]], appliance: HomeAppliance, mock_attr: str, error_regex: str, get_active_program_side_effect: NoProgramActiveError | None, ) -> None: """Test error handling when starting the selected program.""" client.get_active_program = AsyncMock( return_value=Program( key=ProgramKey.DISHCARE_DISHWASHER_ECO_50, ), side_effect=get_active_program_side_effect, ) client.get_selected_program = AsyncMock( return_value=Program( key=ProgramKey.DISHCARE_DISHWASHER_ECO_50, ) ) getattr(client, mock_attr).side_effect = HomeConnectError("error.key") 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)}, ) with pytest.raises(HomeAssistantError, match=error_regex): await hass.services.async_call( domain=DOMAIN, service="start_selected_program", service_data={ "device_id": device_entry.id, }, blocking=True, ) @pytest.mark.parametrize( "get_active_program_side_effect", [None, NoProgramActiveError("error.key")], ) async def test_no_program_error( hass: HomeAssistant, device_registry: dr.DeviceRegistry, client: MagicMock, config_entry: MockConfigEntry, integration_setup: Callable[[MagicMock], Awaitable[bool]], get_active_program_side_effect: NoProgramActiveError | None, ) -> None: """Test handling of no program active error.""" 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, "HA_ID")}, ) client.get_active_program = AsyncMock( return_value=Program( key=None, ), side_effect=get_active_program_side_effect, ) client.get_selected_program = AsyncMock( return_value=Program( key=None, ) ) with pytest.raises(HomeAssistantError, match="No program to start"): await hass.services.async_call( domain=DOMAIN, service="start_selected_program", service_data={ "device_id": device_entry.id, }, blocking=True, ) @pytest.mark.parametrize( "service_call", [ SERVICE_KV_CALL_PARAMS[0], { "domain": DOMAIN, "service": "start_selected_program", "service_data": {}, "blocking": True, }, ], ) async def test_services_appliance_not_found( hass: HomeAssistant, device_registry: dr.DeviceRegistry, client: MagicMock, config_entry: MockConfigEntry, integration_setup: Callable[[MagicMock], Awaitable[bool]], service_call: dict[str, Any], ) -> 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_call.copy() # To avoid mutating the original test data service_call.setdefault("service_data", {}) 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, ) 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) async def test_not_possible_to_use_favorite_program( hass: HomeAssistant, device_registry: dr.DeviceRegistry, client: MagicMock, config_entry: MockConfigEntry, integration_setup: Callable[[MagicMock], Awaitable[bool]], ) -> None: """Raise a MultipleInvalid when trying to use a favorite program.""" 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, "HA_ID")}, ) with pytest.raises(MultipleInvalid): await hass.services.async_call( DOMAIN, "set_program_and_options", { "device_id": device_entry.id, "affects_to": "selected_program", "program": bsh_key_to_translation_key( ProgramKey.BSH_COMMON_FAVORITE_001.value ), }, blocking=True, )