Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve Home Connect appliances test fixture (#139787) 

Improve Home Connect appliances fixture
pull/139801/head
Martin Hjelmare 2025-03-05 00:45:58 +01:00 committed by GitHub
parent 50ba93042b
commit c671862d3f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 267 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
tests/components/home_connect/__init__.py
View File

@ -2,13 +2,10 @@
from typing import Any
from aiohomeconnect.model import ArrayOfHomeAppliances, ArrayOfStatus
from aiohomeconnect.model import ArrayOfStatus
from tests.common import load_json_object_fixture
MOCK_APPLIANCES = ArrayOfHomeAppliances.from_dict(
load_json_object_fixture("home_connect/appliances.json")["data"] # type: ignore[arg-type]
)
MOCK_PROGRAMS: dict[str, Any] = load_json_object_fixture("home_connect/programs.json")
MOCK_SETTINGS: dict[str, Any] = load_json_object_fixture("home_connect/settings.json")
MOCK_STATUS = ArrayOfStatus.from_dict(

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

@ -11,6 +11,7 @@ from aiohomeconnect.client import Client as HomeConnectClient
from aiohomeconnect.model import (
ArrayOfCommands,
ArrayOfEvents,
ArrayOfHomeAppliances,
ArrayOfOptions,
ArrayOfPrograms,
ArrayOfSettings,
@ -39,15 +40,9 @@ from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from . import (
MOCK_APPLIANCES,
MOCK_AVAILABLE_COMMANDS,
MOCK_PROGRAMS,
MOCK_SETTINGS,
MOCK_STATUS,
)
from . import MOCK_AVAILABLE_COMMANDS, MOCK_PROGRAMS, MOCK_SETTINGS, MOCK_STATUS
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, load_fixture
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
@ -148,14 +143,6 @@ async def mock_integration_setup(
return run
def _get_specific_appliance_side_effect(ha_id: str) -> HomeAppliance:
"""Get specific appliance side effect."""
for appliance in copy.deepcopy(MOCK_APPLIANCES).homeappliances:
if appliance.ha_id == ha_id:
return appliance
raise HomeConnectApiError("error.key", "error description")
def _get_set_program_side_effect(
event_queue: asyncio.Queue[list[EventMessage]], event_key: EventKey
):
@ -271,68 +258,12 @@ def _get_set_program_options_side_effect(
return set_program_options_side_effect
async def _get_all_programs_side_effect(ha_id: str) -> ArrayOfPrograms:
"""Get all programs."""
appliance_type = next(
appliance
for appliance in MOCK_APPLIANCES.homeappliances
if appliance.ha_id == ha_id
).type
if appliance_type not in MOCK_PROGRAMS:
raise HomeConnectApiError("error.key", "error description")
return ArrayOfPrograms(
[
EnumerateProgram.from_dict(program)
for program in MOCK_PROGRAMS[appliance_type]["data"]["programs"]
],
Program.from_dict(MOCK_PROGRAMS[appliance_type]["data"]["programs"][0]),
Program.from_dict(MOCK_PROGRAMS[appliance_type]["data"]["programs"][0]),
)
async def _get_settings_side_effect(ha_id: str) -> ArrayOfSettings:
"""Get settings."""
return ArrayOfSettings.from_dict(
MOCK_SETTINGS.get(
next(
appliance
for appliance in MOCK_APPLIANCES.homeappliances
if appliance.ha_id == ha_id
).type,
{},
).get("data", {"settings": []})
)
async def _get_setting_side_effect(ha_id: str, setting_key: SettingKey):
"""Get setting."""
for appliance in MOCK_APPLIANCES.homeappliances:
if appliance.ha_id == ha_id:
settings = MOCK_SETTINGS.get(
next(
appliance
for appliance in MOCK_APPLIANCES.homeappliances
if appliance.ha_id == ha_id
).type,
{},
).get("data", {"settings": []})
for setting_dict in cast(list[dict], settings["settings"]):
if setting_dict["key"] == setting_key:
return GetSetting.from_dict(setting_dict)
raise HomeConnectApiError("error.key", "error description")
async def _get_available_commands_side_effect(ha_id: str) -> ArrayOfCommands:
"""Get available commands."""
for appliance in MOCK_APPLIANCES.homeappliances:
if appliance.ha_id == ha_id and appliance.type in MOCK_AVAILABLE_COMMANDS:
return ArrayOfCommands.from_dict(MOCK_AVAILABLE_COMMANDS[appliance.type])
raise HomeConnectApiError("error.key", "error description")
@pytest.fixture(name="client")
def mock_client(request: pytest.FixtureRequest) -> MagicMock:
def mock_client(
appliances: list[HomeAppliance],
appliance: HomeAppliance | None,
request: pytest.FixtureRequest,
) -> MagicMock:
"""Fixture to mock Client from HomeConnect."""
mock = MagicMock(
@ -369,17 +300,78 @@ def mock_client(request: pytest.FixtureRequest) -> MagicMock:
]
)
appliances = [appliance] if appliance else appliances
async def stream_all_events() -> AsyncGenerator[EventMessage]:
"""Mock stream_all_events."""
while True:
for event in await event_queue.get():
yield event
mock.get_home_appliances = AsyncMock(return_value=copy.deepcopy(MOCK_APPLIANCES))
mock.get_home_appliances = AsyncMock(return_value=ArrayOfHomeAppliances(appliances))
def _get_specific_appliance_side_effect(ha_id: str) -> HomeAppliance:
"""Get specific appliance side effect."""
for appliance_ in appliances:
if appliance_.ha_id == ha_id:
return appliance_
raise HomeConnectApiError("error.key", "error description")
mock.get_specific_appliance = AsyncMock(
side_effect=_get_specific_appliance_side_effect
)
mock.stream_all_events = stream_all_events
async def _get_all_programs_side_effect(ha_id: str) -> ArrayOfPrograms:
"""Get all programs."""
appliance_type = next(
appliance for appliance in appliances if appliance.ha_id == ha_id
).type
if appliance_type not in MOCK_PROGRAMS:
raise HomeConnectApiError("error.key", "error description")
return ArrayOfPrograms(
[
EnumerateProgram.from_dict(program)
for program in MOCK_PROGRAMS[appliance_type]["data"]["programs"]
],
Program.from_dict(MOCK_PROGRAMS[appliance_type]["data"]["programs"][0]),
Program.from_dict(MOCK_PROGRAMS[appliance_type]["data"]["programs"][0]),
)
async def _get_settings_side_effect(ha_id: str) -> ArrayOfSettings:
"""Get settings."""
return ArrayOfSettings.from_dict(
MOCK_SETTINGS.get(
next(
appliance for appliance in appliances if appliance.ha_id == ha_id
).type,
{},
).get("data", {"settings": []})
)
async def _get_setting_side_effect(ha_id: str, setting_key: SettingKey):
"""Get setting."""
for appliance_ in appliances:
if appliance_.ha_id == ha_id:
settings = MOCK_SETTINGS.get(
appliance_.type,
{},
).get("data", {"settings": []})
for setting_dict in cast(list[dict], settings["settings"]):
if setting_dict["key"] == setting_key:
return GetSetting.from_dict(setting_dict)
raise HomeConnectApiError("error.key", "error description")
async def _get_available_commands_side_effect(ha_id: str) -> ArrayOfCommands:
"""Get available commands."""
for appliance_ in appliances:
if appliance_.ha_id == ha_id and appliance_.type in MOCK_AVAILABLE_COMMANDS:
return ArrayOfCommands.from_dict(
MOCK_AVAILABLE_COMMANDS[appliance_.type]
)
raise HomeConnectApiError("error.key", "error description")
mock.start_program = AsyncMock(
side_effect=_get_set_program_side_effect(
event_queue, EventKey.BSH_COMMON_ROOT_ACTIVE_PROGRAM
@ -431,7 +423,11 @@ def mock_client(request: pytest.FixtureRequest) -> MagicMock:
@pytest.fixture(name="client_with_exception")
def mock_client_with_exception(request: pytest.FixtureRequest) -> MagicMock:
def mock_client_with_exception(
appliances: list[HomeAppliance],
appliance: HomeAppliance | None,
request: pytest.FixtureRequest,
) -> MagicMock:
"""Fixture to mock Client from HomeConnect that raise exceptions."""
mock = MagicMock(
autospec=HomeConnectClient,
@ -449,7 +445,8 @@ def mock_client_with_exception(request: pytest.FixtureRequest) -> MagicMock:
for event in await event_queue.get():
yield event
mock.get_home_appliances = AsyncMock(return_value=copy.deepcopy(MOCK_APPLIANCES))
appliances = [appliance] if appliance else appliances
mock.get_home_appliances = AsyncMock(return_value=ArrayOfHomeAppliances(appliances))
mock.stream_all_events = stream_all_events
mock.start_program = AsyncMock(side_effect=exception)
@ -477,12 +474,52 @@ def mock_client_with_exception(request: pytest.FixtureRequest) -> MagicMock:
@pytest.fixture(name="appliance_ha_id")
def mock_appliance_ha_id(request: pytest.FixtureRequest) -> str:
"""Fixture to mock Appliance."""
app = "Washer"
def mock_appliance_ha_id(
appliances: list[HomeAppliance], request: pytest.FixtureRequest
) -> str:
"""Fixture to get the ha_id of an appliance."""
appliance_type = "Washer"
if hasattr(request, "param") and request.param:
app = request.param
for appliance in MOCK_APPLIANCES.homeappliances:
if appliance.type == app:
appliance_type = request.param
for appliance in appliances:
if appliance.type == appliance_type:
return appliance.ha_id
raise ValueError(f"Appliance {app} not found")
raise ValueError(f"Appliance {appliance_type} not found")
@pytest.fixture(name="appliances")
def mock_appliances(
appliances_data: str, request: pytest.FixtureRequest
) -> list[HomeAppliance]:
"""Fixture to mock the returned appliances."""
appliances = ArrayOfHomeAppliances.from_json(appliances_data).homeappliances
appliance_types = {appliance.type for appliance in appliances}
if hasattr(request, "param") and request.param:
appliance_types = request.param
return [appliance for appliance in appliances if appliance.type in appliance_types]
@pytest.fixture(name="appliance")
def mock_appliance(
appliances_data: str, request: pytest.FixtureRequest
) -> HomeAppliance | None:
"""Fixture to mock a single specific appliance to return."""
appliance_type = None
if hasattr(request, "param") and request.param:
appliance_type = request.param
return next(
(
appliance
for appliance in ArrayOfHomeAppliances.from_json(
appliances_data
).homeappliances
if appliance.type == appliance_type
),
None,
)
@pytest.fixture(name="appliances_data")
def appliances_data_fixture() -> str:
"""Fixture to return a the string for an array of appliances."""
return load_fixture("appliances.json", integration=DOMAIN)

240
tests/components/home_connect/fixtures/appliances.json
View File

@ -1,123 +1,121 @@
{
"data": {
"homeappliances": [
{
"name": "FridgeFreezer",
"brand": "SIEMENS",
"vib": "HCS05FRF1",
"connected": true,
"type": "FridgeFreezer",
"enumber": "HCS05FRF1/03",
"haId": "SIEMENS-HCS05FRF1-304F4F9E541D"
},
{
"name": "Dishwasher",
"brand": "SIEMENS",
"vib": "HCS02DWH1",
"connected": true,
"type": "Dishwasher",
"enumber": "HCS02DWH1/03",
"haId": "SIEMENS-HCS02DWH1-6BE58C26DCC1"
},
{
"name": "Oven",
"brand": "BOSCH",
"vib": "HCS01OVN1",
"connected": true,
"type": "Oven",
"enumber": "HCS01OVN1/03",
"haId": "BOSCH-HCS01OVN1-43E0065FE245"
},
{
"name": "Washer",
"brand": "SIEMENS",
"vib": "HCS03WCH1",
"connected": true,
"type": "Washer",
"enumber": "HCS03WCH1/03",
"haId": "SIEMENS-HCS03WCH1-7BC6383CF794"
},
{
"name": "Dryer",
"brand": "BOSCH",
"vib": "HCS04DYR1",
"connected": true,
"type": "Dryer",
"enumber": "HCS04DYR1/03",
"haId": "BOSCH-HCS04DYR1-831694AE3C5A"
},
{
"name": "CoffeeMaker",
"brand": "BOSCH",
"vib": "HCS06COM1",
"connected": true,
"type": "CoffeeMaker",
"enumber": "HCS06COM1/03",
"haId": "BOSCH-HCS06COM1-D70390681C2C"
},
{
"name": "WasherDryer",
"brand": "BOSCH",
"vib": "HCS000001",
"connected": true,
"type": "WasherDryer",
"enumber": "HCS000000/01",
"haId": "BOSCH-HCS000000-D00000000001"
},
{
"name": "Refrigerator",
"brand": "BOSCH",
"vib": "HCS000002",
"connected": true,
"type": "Refrigerator",
"enumber": "HCS000000/02",
"haId": "BOSCH-HCS000000-D00000000002"
},
{
"name": "Freezer",
"brand": "BOSCH",
"vib": "HCS000003",
"connected": true,
"type": "Freezer",
"enumber": "HCS000000/03",
"haId": "BOSCH-HCS000000-D00000000003"
},
{
"name": "Hood",
"brand": "BOSCH",
"vib": "HCS000004",
"connected": true,
"type": "Hood",
"enumber": "HCS000000/04",
"haId": "BOSCH-HCS000000-D00000000004"
},
{
"name": "Hob",
"brand": "BOSCH",
"vib": "HCS000005",
"connected": true,
"type": "Hob",
"enumber": "HCS000000/05",
"haId": "BOSCH-HCS000000-D00000000005"
},
{
"name": "CookProcessor",
"brand": "BOSCH",
"vib": "HCS000006",
"connected": true,
"type": "CookProcessor",
"enumber": "HCS000000/06",
"haId": "BOSCH-HCS000000-D00000000006"
},
{
"name": "DNE",
"brand": "BOSCH",
"vib": "HCS000000",
"connected": true,
"type": "DNE",
"enumber": "HCS000000/00",
"haId": "BOSCH-000000000-000000000000"
}
]
}
"homeappliances": [
{
"name": "FridgeFreezer",
"brand": "SIEMENS",
"vib": "HCS05FRF1",
"connected": true,
"type": "FridgeFreezer",
"enumber": "HCS05FRF1/03",
"haId": "SIEMENS-HCS05FRF1-304F4F9E541D"
},
{
"name": "Dishwasher",
"brand": "SIEMENS",
"vib": "HCS02DWH1",
"connected": true,
"type": "Dishwasher",
"enumber": "HCS02DWH1/03",
"haId": "SIEMENS-HCS02DWH1-6BE58C26DCC1"
},
{
"name": "Oven",
"brand": "BOSCH",
"vib": "HCS01OVN1",
"connected": true,
"type": "Oven",
"enumber": "HCS01OVN1/03",
"haId": "BOSCH-HCS01OVN1-43E0065FE245"
},
{
"name": "Washer",
"brand": "SIEMENS",
"vib": "HCS03WCH1",
"connected": true,
"type": "Washer",
"enumber": "HCS03WCH1/03",
"haId": "SIEMENS-HCS03WCH1-7BC6383CF794"
},
{
"name": "Dryer",
"brand": "BOSCH",
"vib": "HCS04DYR1",
"connected": true,
"type": "Dryer",
"enumber": "HCS04DYR1/03",
"haId": "BOSCH-HCS04DYR1-831694AE3C5A"
},
{
"name": "CoffeeMaker",
"brand": "BOSCH",
"vib": "HCS06COM1",
"connected": true,
"type": "CoffeeMaker",
"enumber": "HCS06COM1/03",
"haId": "BOSCH-HCS06COM1-D70390681C2C"
},
{
"name": "WasherDryer",
"brand": "BOSCH",
"vib": "HCS000001",
"connected": true,
"type": "WasherDryer",
"enumber": "HCS000000/01",
"haId": "BOSCH-HCS000000-D00000000001"
},
{
"name": "Refrigerator",
"brand": "BOSCH",
"vib": "HCS000002",
"connected": true,
"type": "Refrigerator",
"enumber": "HCS000000/02",
"haId": "BOSCH-HCS000000-D00000000002"
},
{
"name": "Freezer",
"brand": "BOSCH",
"vib": "HCS000003",
"connected": true,
"type": "Freezer",
"enumber": "HCS000000/03",
"haId": "BOSCH-HCS000000-D00000000003"
},
{
"name": "Hood",
"brand": "BOSCH",
"vib": "HCS000004",
"connected": true,
"type": "Hood",
"enumber": "HCS000000/04",
"haId": "BOSCH-HCS000000-D00000000004"
},
{
"name": "Hob",
"brand": "BOSCH",
"vib": "HCS000005",
"connected": true,
"type": "Hob",
"enumber": "HCS000000/05",
"haId": "BOSCH-HCS000000-D00000000005"
},
{
"name": "CookProcessor",
"brand": "BOSCH",
"vib": "HCS000006",
"connected": true,
"type": "CookProcessor",
"enumber": "HCS000000/06",
"haId": "BOSCH-HCS000000-D00000000006"
},
{
"name": "DNE",
"brand": "BOSCH",
"vib": "HCS000000",
"connected": true,
"type": "DNE",
"enumber": "HCS000000/00",
"haId": "BOSCH-000000000-000000000000"
}
]
}

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

@ -1,19 +1,20 @@
"""Test for Home Connect coordinator."""
from collections.abc import Awaitable, Callable
import copy
from datetime import timedelta
from typing import Any
from typing import Any, cast
from unittest.mock import AsyncMock, MagicMock, patch
from aiohomeconnect.model import (
ArrayOfEvents,
ArrayOfHomeAppliances,
ArrayOfSettings,
ArrayOfStatus,
Event,
EventKey,
EventMessage,
EventType,
HomeAppliance,
)
from aiohomeconnect.model.error import (
EventStreamInterruptedError,
@ -41,8 +42,6 @@ from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from . import MOCK_APPLIANCES
from tests.common import MockConfigEntry, async_fire_time_changed
@ -81,16 +80,21 @@ async def test_coordinator_update_failing_get_appliances(
@pytest.mark.usefixtures("setup_credentials")
@pytest.mark.parametrize("platforms", [("binary_sensor",)])
@pytest.mark.parametrize("appliance_ha_id", ["Washer"], indirect=True)
@pytest.mark.parametrize("appliance", ["Washer"], indirect=True)
async def test_coordinator_failure_refresh_and_stream(
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
client: MagicMock,
freezer: FrozenDateTimeFactory,
appliance_ha_id: str,
appliance: HomeAppliance,
) -> None:
"""Test entity available state via coordinator refresh and event stream."""
appliance_data = (
cast(str, appliance.to_json())
.replace("ha_id", "haId")
.replace("e_number", "enumber")
)
entity_id_1 = "binary_sensor.washer_remote_control"
entity_id_2 = "binary_sensor.washer_remote_start"
await async_setup_component(hass, "homeassistant", {})
@ -121,7 +125,9 @@ async def test_coordinator_failure_refresh_and_stream(
# Test that the entity becomes available again after a successful update.
client.get_home_appliances.side_effect = None
client.get_home_appliances.return_value = copy.deepcopy(MOCK_APPLIANCES)
client.get_home_appliances.return_value = ArrayOfHomeAppliances(
[HomeAppliance.from_json(appliance_data)]
)
# Move time forward to pass the debounce time.
freezer.tick(timedelta(hours=1))
@ -166,11 +172,13 @@ async def test_coordinator_failure_refresh_and_stream(
# Now make the entity available again.
client.get_home_appliances.side_effect = None
client.get_home_appliances.return_value = copy.deepcopy(MOCK_APPLIANCES)
client.get_home_appliances.return_value = ArrayOfHomeAppliances(
[HomeAppliance.from_json(appliance_data)]
)
# One event should make all entities for this appliance available again.
event_message = EventMessage(
appliance_ha_id,
appliance.ha_id,
EventType.STATUS,
ArrayOfEvents(
[
@ -399,6 +407,9 @@ async def test_event_listener_error(
assert not config_entry._background_tasks
@pytest.mark.usefixtures("setup_credentials")
@pytest.mark.parametrize("platforms", [("sensor",)])
@pytest.mark.parametrize("appliance", ["Washer"], indirect=True)
@pytest.mark.parametrize(
"exception",
[HomeConnectRequestError(), EventStreamInterruptedError()],
@ -429,11 +440,10 @@ async def test_event_listener_resilience(
after_event_expected_state: str,
exception: HomeConnectError,
hass: HomeAssistant,
appliance: HomeAppliance,
client: MagicMock,
config_entry: MockConfigEntry,
integration_setup: Callable[[MagicMock], Awaitable[bool]],
setup_credentials: None,
client: MagicMock,
appliance_ha_id: str,
) -> None:
"""Test that the event listener is resilient to interruptions."""
future = hass.loop.create_future()
@ -467,7 +477,7 @@ async def test_event_listener_resilience(
await client.add_events(
[
EventMessage(
appliance_ha_id,
appliance.ha_id,
EventType.STATUS,
ArrayOfEvents(
[