Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 00:20:30 +01:00
Code Activity

Get program from base program option at Home Connect (#164885)

Browse Source
This commit is contained in:
J. Diego Rodríguez Royo
2026-03-25 09:12:51 +01:00
committed by GitHub
parent 78e2514b46
commit c055972887
4 changed files with 217 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -312,6 +312,7 @@ class HomeConnectApplianceCoordinator(DataUpdateCoordinator[HomeConnectAppliance
case EventType.NOTIFY:
settings = self.data.settings
events = self.data.events
program_update_event_value = None
for event in event_message.data.items:
event_key = event.key
if event_key in SettingKey.__members__.values(): # type: ignore[comparison-overlap]
@@ -330,11 +331,13 @@ class HomeConnectApplianceCoordinator(DataUpdateCoordinator[HomeConnectAppliance
EventKey.BSH_COMMON_ROOT_ACTIVE_PROGRAM,
EventKey.BSH_COMMON_ROOT_SELECTED_PROGRAM,
) and isinstance(event_value, str):
await self.update_options(
event_key,
ProgramKey(event_value),
)
program_update_event_value = ProgramKey(event_value)
events[event_key] = event
# Process program update after all events to ensure
# BSH_COMMON_OPTION_BASE_PROGRAM event is available for
# favorite program resolution
if program_update_event_value:
await self.update_options(program_update_event_value)
self._call_event_listener(event_message)
case EventType.EVENT:
@@ -493,7 +496,7 @@ class HomeConnectApplianceCoordinator(DataUpdateCoordinator[HomeConnectAppliance
programs = []
events = {}
options = {}
if appliance.type in APPLIANCES_WITH_PROGRAMS:
if appliance.type in APPLIANCES_WITH_PROGRAMS: # pylint: disable=too-many-nested-blocks
try:
all_programs = await self.client.get_all_programs(appliance.ha_id)
except TooManyRequestsError:
@@ -529,6 +532,17 @@ class HomeConnectApplianceCoordinator(DataUpdateCoordinator[HomeConnectAppliance
)
current_program_key = program.key
program_options = program.options
if (
current_program_key == ProgramKey.BSH_COMMON_FAVORITE_001
and program_options
):
# The API doesn't allow to fetch the options from the favorite program.
# We can attempt to get the base program and get the options
for option in program_options:
if option.key == OptionKey.BSH_COMMON_BASE_PROGRAM:
current_program_key = ProgramKey(option.value)
break
if current_program_key:
options = await self.get_options_definitions(current_program_key)
for option in program_options or []:
@@ -595,15 +609,24 @@ class HomeConnectApplianceCoordinator(DataUpdateCoordinator[HomeConnectAppliance
)
return {}
async def update_options(
self, event_key: EventKey, program_key: ProgramKey
) -> None:
async def update_options(self, program_key: ProgramKey) -> None:
"""Update options for appliance."""
options = self.data.options
events = self.data.events
options_to_notify = options.copy()
options.clear()
options.update(await self.get_options_definitions(program_key))
if (
program_key == ProgramKey.BSH_COMMON_FAVORITE_001
and (event := events.get(EventKey.BSH_COMMON_OPTION_BASE_PROGRAM))
and isinstance(event.value, str)
):
# The API doesn't allow to fetch the options from the favorite program.
# We can attempt to get the base program and get the options
resolved_program_key = ProgramKey(event.value)
else:
resolved_program_key = program_key
options.update(await self.get_options_definitions(resolved_program_key))
for option in options.values():
option_value = option.constraints.default if option.constraints else None

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

@@ -430,11 +430,24 @@ class HomeConnectProgramSelectEntity(HomeConnectEntity, SelectEntity):
def update_native_value(self) -> None:
"""Set the program value."""
event = self.appliance.events.get(cast(EventKey, self.bsh_key))
self._attr_current_option = (
PROGRAMS_TRANSLATION_KEYS_MAP.get(ProgramKey(event_value))
program_key = (
ProgramKey(event_value)
if event and isinstance(event_value := event.value, str)
else None
)
if (
program_key == ProgramKey.BSH_COMMON_FAVORITE_001
and (
base_program_event := self.appliance.events.get(
EventKey.BSH_COMMON_OPTION_BASE_PROGRAM
)
)
and isinstance(base_program_event.value, str)
):
program_key = ProgramKey(base_program_event.value)
self._attr_current_option = (
PROGRAMS_TRANSLATION_KEYS_MAP.get(program_key) if program_key else None
)
async def async_select_option(self, option: str) -> None:
"""Select new program."""

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

@@ -9,6 +9,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
from aiohomeconnect.model import (
ArrayOfEvents,
ArrayOfHomeAppliances,
ArrayOfPrograms,
ArrayOfSettings,
ArrayOfStatus,
Event,
@@ -27,6 +28,7 @@ from aiohomeconnect.model.error import (
TooManyRequestsError,
UnauthorizedError,
)
from aiohomeconnect.model.program import Option, OptionKey, Program, ProgramKey
from freezegun.api import FrozenDateTimeFactory
import pytest
@@ -197,11 +199,7 @@ async def test_coordinator_failure_refresh_and_stream(
assert state.state != STATE_UNAVAILABLE
@pytest.mark.parametrize(
"appliance",
["Dishwasher"],
indirect=True,
)
@pytest.mark.parametrize("appliance", ["Dishwasher"], indirect=True)
async def test_coordinator_not_fetching_on_disconnected_appliance(
client: MagicMock,
config_entry: MockConfigEntry,
@@ -888,3 +886,103 @@ async def test_other_errors_while_updating_appliance(
record.levelname == log_level and re.search(string_in_log, record.message)
for record in caplog.records
)
@pytest.mark.parametrize("appliance", ["Dishwasher"], indirect=True)
@pytest.mark.parametrize("array_of_programs_param", ["active", "selected"])
async def test_fetch_base_program_options_when_active_favorite_program(
client: MagicMock,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
appliance: HomeAppliance,
array_of_programs_param: str,
) -> None:
"""Test usage of base program option.
Test that when the favorite program is active or selected,
the options are fetched from the base program.
"""
client.get_all_programs = AsyncMock(
return_value=ArrayOfPrograms(
programs=[],
**{
array_of_programs_param: Program(
key=ProgramKey.BSH_COMMON_FAVORITE_001,
options=[
Option(
OptionKey.BSH_COMMON_BASE_PROGRAM,
ProgramKey.DISHCARE_DISHWASHER_ECO_50.value,
)
],
),
},
)
)
assert await integration_setup(client)
assert config_entry.state is ConfigEntryState.LOADED
client.get_available_program.assert_awaited_once_with(
appliance.ha_id, program_key=ProgramKey.DISHCARE_DISHWASHER_ECO_50
)
@pytest.mark.parametrize("appliance", ["Dishwasher"], indirect=True)
@pytest.mark.parametrize(
"event_key",
[
EventKey.BSH_COMMON_ROOT_ACTIVE_PROGRAM,
EventKey.BSH_COMMON_ROOT_SELECTED_PROGRAM,
],
)
async def test_fetch_base_program_options_when_favorite_program_event(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
appliance: HomeAppliance,
event_key: EventKey,
) -> None:
"""Test usage of base program option on event.
Test that when a program event does report favorite program,
the options are fetched from the base program.
"""
appliance_ha_id = appliance.ha_id
assert await integration_setup(client)
assert config_entry.state is ConfigEntryState.LOADED
client.get_available_program.reset_mock()
await client.add_events(
[
EventMessage(
appliance_ha_id,
EventType.NOTIFY,
data=ArrayOfEvents(
[
Event(
key=event_key,
raw_key=event_key.value,
timestamp=0,
level="",
handling="",
value=ProgramKey.BSH_COMMON_FAVORITE_001.value,
),
Event(
key=EventKey.BSH_COMMON_OPTION_BASE_PROGRAM,
raw_key=EventKey.BSH_COMMON_OPTION_BASE_PROGRAM.value,
timestamp=0,
level="",
handling="",
value=ProgramKey.DISHCARE_DISHWASHER_ECO_50.value,
),
]
),
)
]
)
await hass.async_block_till_done()
client.get_available_program.assert_awaited_once_with(
appliance.ha_id, program_key=ProgramKey.DISHCARE_DISHWASHER_ECO_50
)

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

@@ -29,6 +29,8 @@ from aiohomeconnect.model.program import (
EnumerateProgram,
EnumerateProgramConstraints,
Execution,
Option,
Program,
ProgramDefinitionConstraints,
ProgramDefinitionOption,
)
@@ -1174,3 +1176,68 @@ async def test_favorite_001_program_not_exposed_as_option(
entity_state = hass.states.get("select.dishwasher_selected_program")
assert entity_state
assert entity_state.attributes[ATTR_OPTIONS] == []
@pytest.mark.parametrize("appliance", ["Dishwasher"], indirect=True)
@pytest.mark.parametrize(
("array_of_programs_param", "entity_id"),
[
("active", "select.dishwasher_active_program"),
("selected", "select.dishwasher_selected_program"),
],
)
@pytest.mark.parametrize(
("options", "expected_state"),
[
(
[
Option(
OptionKey.BSH_COMMON_BASE_PROGRAM,
ProgramKey.DISHCARE_DISHWASHER_ECO_50.value,
)
],
"dishcare_dishwasher_program_eco_50",
),
(None, STATE_UNKNOWN),
],
)
async def test_use_base_program_on_favorite_program(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
array_of_programs_param: str,
entity_id: str,
options: list[Option] | None,
expected_state: str,
) -> None:
"""Test that base program is used.
Assert that when the favorite program is active/selected,
use the base program if present to set the value of the entity;
if not present, the state should be unknown.
"""
client.get_all_programs = AsyncMock(
return_value=ArrayOfPrograms(
programs=[
EnumerateProgram(
key=ProgramKey.DISHCARE_DISHWASHER_ECO_50,
raw_key=ProgramKey.DISHCARE_DISHWASHER_ECO_50.value,
constraints=EnumerateProgramConstraints(
execution=Execution.SELECT_AND_START,
),
),
],
**{
array_of_programs_param: Program(
key=ProgramKey.BSH_COMMON_FAVORITE_001,
options=options,
),
},
)
)
assert await integration_setup(client)
assert config_entry.state is ConfigEntryState.LOADED
assert hass.states.is_state(entity_id, expected_state)