Add tests to Home Connect integration (#114214)

* Add tests to Home Connect integration

* Fix misspelling

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Changes to tests with properly setup fixtures.

* Consolidated api tests, patched library instead of code

* Consolidate sensor edge cases, switch mock assertion to call_count

* Adjust assertion

---------

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Robert Contreras 2024-04-08 23:52:39 -07:00 committed by GitHub
parent 4a7a22641e
commit af7d0187cb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 1192 additions and 3 deletions

View file

@ -542,12 +542,9 @@ omit =
homeassistant/components/hko/weather.py
homeassistant/components/hlk_sw16/__init__.py
homeassistant/components/hlk_sw16/switch.py
homeassistant/components/home_connect/__init__.py
homeassistant/components/home_connect/api.py
homeassistant/components/home_connect/binary_sensor.py
homeassistant/components/home_connect/entity.py
homeassistant/components/home_connect/light.py
homeassistant/components/home_connect/sensor.py
homeassistant/components/home_connect/switch.py
homeassistant/components/homematic/__init__.py
homeassistant/components/homematic/binary_sensor.py

View file

@ -0,0 +1,235 @@
"""Test fixtures for home_connect."""
from collections.abc import Awaitable, Callable, Generator
import time
from typing import Any
from unittest.mock import MagicMock, Mock, PropertyMock, patch
from homeconnect.api import HomeConnectAppliance, HomeConnectError
import pytest
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.home_connect import update_all_devices
from homeassistant.components.home_connect.const import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, load_json_object_fixture
MOCK_APPLIANCES_PROPERTIES = {
x["name"]: x
for x in load_json_object_fixture("home_connect/appliances.json")["data"][
"homeappliances"
]
}
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
FAKE_ACCESS_TOKEN = "some-access-token"
FAKE_REFRESH_TOKEN = "some-refresh-token"
FAKE_AUTH_IMPL = "conftest-imported-cred"
SERVER_ACCESS_TOKEN = {
"refresh_token": "server-refresh-token",
"access_token": "server-access-token",
"type": "Bearer",
"expires_in": 60,
}
@pytest.fixture(name="token_expiration_time")
def mock_token_expiration_time() -> float:
"""Fixture for expiration time of the config entry auth token."""
return time.time() + 86400
@pytest.fixture(name="token_entry")
def mock_token_entry(token_expiration_time: float) -> dict[str, Any]:
"""Fixture for OAuth 'token' data for a ConfigEntry."""
return {
"refresh_token": FAKE_REFRESH_TOKEN,
"access_token": FAKE_ACCESS_TOKEN,
"type": "Bearer",
"expires_at": token_expiration_time,
}
@pytest.fixture(name="config_entry")
def mock_config_entry(token_entry: dict[str, Any]) -> MockConfigEntry:
"""Fixture for a config entry."""
return MockConfigEntry(
domain=DOMAIN,
data={
"auth_implementation": FAKE_AUTH_IMPL,
"token": token_entry,
},
)
@pytest.fixture
async def setup_credentials(hass: HomeAssistant) -> None:
"""Fixture to setup credentials."""
assert await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass,
DOMAIN,
ClientCredential(CLIENT_ID, CLIENT_SECRET),
FAKE_AUTH_IMPL,
)
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture to specify platforms to test."""
return []
async def bypass_throttle(hass: HomeAssistant, config_entry: MockConfigEntry):
"""Add kwarg to disable throttle."""
await update_all_devices(hass, config_entry, no_throttle=True)
@pytest.fixture(name="bypass_throttle")
def mock_bypass_throttle():
"""Fixture to bypass the throttle decorator in __init__."""
with patch(
"homeassistant.components.home_connect.update_all_devices",
side_effect=lambda x, y: bypass_throttle(x, y),
):
yield
@pytest.fixture(name="integration_setup")
async def mock_integration_setup(
hass: HomeAssistant,
platforms: list[Platform],
config_entry: MockConfigEntry,
) -> Callable[[], Awaitable[bool]]:
"""Fixture to set up the integration."""
config_entry.add_to_hass(hass)
async def run() -> bool:
with patch("homeassistant.components.home_connect.PLATFORMS", platforms):
result = await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return result
return run
@pytest.fixture(name="get_appliances")
def mock_get_appliances() -> Generator[None, Any, None]:
"""Mock ConfigEntryAuth parent (HomeAssistantAPI) method."""
with patch(
"homeassistant.components.home_connect.api.ConfigEntryAuth.get_appliances",
) as mock:
yield mock
@pytest.fixture(name="appliance")
def mock_appliance(request) -> Mock:
"""Fixture to mock Appliance."""
app = "Washer"
if hasattr(request, "param") and request.param:
app = request.param
mock = MagicMock(
autospec=HomeConnectAppliance,
**MOCK_APPLIANCES_PROPERTIES.get(app),
)
mock.name = app
type(mock).status = PropertyMock(return_value={})
mock.get.return_value = {}
mock.get_programs_available.return_value = []
mock.get_status.return_value = {}
mock.get_settings.return_value = {}
return mock
@pytest.fixture(name="problematic_appliance")
def mock_problematic_appliance() -> Mock:
"""Fixture to mock a problematic Appliance."""
app = "Washer"
mock = Mock(
spec=HomeConnectAppliance,
**MOCK_APPLIANCES_PROPERTIES.get(app),
)
mock.name = app
setattr(mock, "status", {})
mock.get_programs_active.side_effect = HomeConnectError
mock.get_programs_available.side_effect = HomeConnectError
mock.start_program.side_effect = HomeConnectError
mock.stop_program.side_effect = HomeConnectError
mock.get_status.side_effect = HomeConnectError
mock.get_settings.side_effect = HomeConnectError
mock.set_setting.side_effect = HomeConnectError
return mock
def get_all_appliances():
"""Return a list of `HomeConnectAppliance` instances for all appliances."""
appliances = {}
data = load_json_object_fixture("home_connect/appliances.json").get("data")
programs_active = load_json_object_fixture("home_connect/programs-active.json")
programs_available = load_json_object_fixture(
"home_connect/programs-available.json"
)
def listen_callback(mock, callback):
callback["callback"](mock)
for home_appliance in data["homeappliances"]:
api_status = load_json_object_fixture("home_connect/status.json")
api_settings = load_json_object_fixture("home_connect/settings.json")
ha_id = home_appliance["haId"]
ha_type = home_appliance["type"]
appliance = MagicMock(spec=HomeConnectAppliance, **home_appliance)
appliance.name = home_appliance["name"]
appliance.listen_events.side_effect = (
lambda app=appliance, **x: listen_callback(app, x)
)
appliance.get_programs_active.return_value = programs_active.get(
ha_type, {}
).get("data", {})
appliance.get_programs_available.return_value = [
program["key"]
for program in programs_available.get(ha_type, {})
.get("data", {})
.get("programs", [])
]
appliance.get_status.return_value = HomeConnectAppliance.json2dict(
api_status.get("data", {}).get("status", [])
)
appliance.get_settings.return_value = HomeConnectAppliance.json2dict(
api_settings.get(ha_type, {}).get("data", {}).get("settings", [])
)
setattr(appliance, "status", {})
appliance.status.update(appliance.get_status.return_value)
appliance.status.update(appliance.get_settings.return_value)
appliance.set_setting.side_effect = (
lambda x, y, appliance=appliance: appliance.status.update({x: {"value": y}})
)
appliance.start_program.side_effect = (
lambda x, appliance=appliance: appliance.status.update(
{"BSH.Common.Root.ActiveProgram": {"value": x}}
)
)
appliance.stop_program.side_effect = (
lambda appliance=appliance: appliance.status.update(
{"BSH.Common.Root.ActiveProgram": {}}
)
)
appliances[ha_id] = appliance
return list(appliances.values())

View file

@ -0,0 +1,123 @@
{
"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"
}
]
}
}

View file

@ -0,0 +1,28 @@
{
"Oven": {
"data": {
"key": "Cooking.Oven.Program.HeatingMode.HotAir",
"name": "Hot air",
"options": [
{
"key": "Cooking.Oven.Option.SetpointTemperature",
"name": "Target temperature for the cavity",
"value": 230,
"unit": "°C"
},
{
"key": "BSH.Common.Option.Duration",
"name": "Adjust the duration",
"value": 1200,
"unit": "seconds"
}
]
}
},
"Washer": {
"data": {
"key": "BSH.Common.Root.ActiveProgram",
"value": "LaundryCare.Dryer.Program.Mix"
}
}
}

View file

@ -0,0 +1,185 @@
{
"Oven": {
"data": {
"programs": [
{
"key": "Cooking.Oven.Program.HeatingMode.HotAir",
"name": "Hot air",
"contraints": {
"execution": "selectandstart"
}
},
{
"key": "Cooking.Oven.Program.HeatingMode.TopBottomHeating",
"name": "Top/bottom heating",
"contraints": {
"execution": "none"
}
},
{
"key": "Cooking.Oven.Program.HeatingMode.PizzaSetting",
"name": "Pizza setting",
"contraints": {
"execution": "startonly"
}
}
]
}
},
"DishWasher": {
"data": {
"programs": [
{
"key": "Dishcare.Dishwasher.Program.Auto1",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "Dishcare.Dishwasher.Program.Auto2",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "Dishcare.Dishwasher.Program.Auto3",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "Dishcare.Dishwasher.Program.Eco50",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "Dishcare.Dishwasher.Program.Quick45",
"constraints": {
"execution": "selectandstart"
}
}
]
}
},
"Washer": {
"data": {
"programs": [
{
"key": "LaundryCare.Washer.Program.Cotton",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "LaundryCare.Washer.Program.EasyCare",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "LaundryCare.Washer.Program.Mix",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "LaundryCare.Washer.Program.DelicatesSilk",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "LaundryCare.Washer.Program.Wool",
"constraints": {
"execution": "selectandstart"
}
}
]
}
},
"Dryer": {
"data": {
"programs": [
{
"key": "LaundryCare.Dryer.Program.Cotton",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "LaundryCare.Dryer.Program.Synthetic",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "LaundryCare.Dryer.Program.Mix",
"constraints": {
"execution": "selectandstart"
}
}
]
}
},
"CoffeeMaker": {
"data": {
"programs": [
{
"key": "ConsumerProducts.CoffeeMaker.Program.Beverage.Espresso",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "ConsumerProducts.CoffeeMaker.Program.Beverage.EspressoMacchiato",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "ConsumerProducts.CoffeeMaker.Program.Beverage.Coffee",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "ConsumerProducts.CoffeeMaker.Program.Beverage.Cappuccino",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "ConsumerProducts.CoffeeMaker.Program.Beverage.LatteMacchiato",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "ConsumerProducts.CoffeeMaker.Program.Beverage.CaffeLatte",
"constraints": {
"execution": "selectandstart"
}
}
]
}
},
"WasherDryer": {
"data": {
"programs": [
{
"key": "LaundryCare.WasherDryer.Program.Mix",
"constraints": {
"execution": "selectandstart"
}
},
{
"key": "LaundryCare.Washer.Option.Temperature",
"constraints": {
"execution": "selectandstart"
}
}
]
}
}
}

View file

@ -0,0 +1,99 @@
{
"Dishwasher": {
"data": {
"settings": [
{
"key": "BSH.Common.Setting.AmbientLightEnabled",
"value": true,
"type": "Boolean"
},
{
"key": "BSH.Common.Setting.AmbientLightBrightness",
"value": 70,
"unit": "%",
"type": "Double"
},
{
"key": "BSH.Common.Setting.AmbientLightColor",
"value": "BSH.Common.EnumType.AmbientLightColor.Color43",
"type": "BSH.Common.EnumType.AmbientLightColor"
},
{
"key": "BSH.Common.Setting.AmbientLightCustomColor",
"value": "#4a88f8",
"type": "String"
},
{
"key": "BSH.Common.Setting.PowerState",
"value": "BSH.Common.EnumType.PowerState.On",
"type": "BSH.Common.EnumType.PowerState"
},
{
"key": "BSH.Common.Setting.ChildLock",
"value": false,
"type": "Boolean"
}
]
}
},
"Hood": {
"data": {
"settings": [
{
"key": "Cooking.Common.Setting.Lighting",
"value": true,
"type": "Boolean"
},
{
"key": "Cooking.Common.Setting.LightingBrightness",
"value": 70,
"unit": "%",
"type": "Double"
},
{
"key": "Cooking.Hood.Setting.ColorTemperaturePercent",
"value": 70,
"unit": "%",
"type": "Double"
},
{
"key": "BSH.Common.Setting.ColorTemperature",
"value": "Cooking.Hood.EnumType.ColorTemperature.warmToNeutral",
"type": "BSH.Common.EnumType.ColorTemperature"
},
{
"key": "BSH.Common.Setting.AmbientLightEnabled",
"value": true,
"type": "Boolean"
},
{
"key": "BSH.Common.Setting.AmbientLightBrightness",
"value": 70,
"unit": "%",
"type": "Double"
},
{
"key": "BSH.Common.Setting.AmbientLightColor",
"value": "BSH.Common.EnumType.AmbientLightColor.Color43",
"type": "BSH.Common.EnumType.AmbientLightColor"
},
{
"key": "BSH.Common.Setting.AmbientLightCustomColor",
"value": "#4a88f8",
"type": "String"
}
]
}
},
"Oven": {
"data": {
"settings": [
{
"key": "BSH.Common.Setting.PowerState",
"value": "BSH.Common.EnumType.PowerState.On",
"type": "BSH.Common.EnumType.PowerState"
}
]
}
}
}

View file

@ -0,0 +1,16 @@
{
"data": {
"status": [
{ "key": "BSH.Common.Status.RemoteControlActive", "value": true },
{ "key": "BSH.Common.Status.RemoteControlStartAllowed", "value": true },
{
"key": "BSH.Common.Status.OperationState",
"value": "BSH.Common.EnumType.OperationState.Ready"
},
{
"key": "BSH.Common.Status.DoorState",
"value": "BSH.Common.EnumType.DoorState.Closed"
}
]
}
}

View file

@ -0,0 +1,301 @@
"""Test the integration init functionality."""
from collections.abc import Awaitable, Callable, Generator
from typing import Any
from unittest.mock import MagicMock, Mock
import pytest
from requests import HTTPError
import requests_mock
from homeassistant.components.home_connect.const import DOMAIN, OAUTH2_TOKEN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .conftest import (
CLIENT_ID,
CLIENT_SECRET,
FAKE_ACCESS_TOKEN,
FAKE_REFRESH_TOKEN,
SERVER_ACCESS_TOKEN,
get_all_appliances,
)
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
SERVICE_KV_CALL_PARAMS = [
{
"domain": DOMAIN,
"service": "set_option_active",
"service_data": {
"device_id": "DEVICE_ID",
"key": "",
"value": "",
"unit": "",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "set_option_selected",
"service_data": {
"device_id": "DEVICE_ID",
"key": "",
"value": "",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "change_setting",
"service_data": {
"device_id": "DEVICE_ID",
"key": "",
"value": "",
},
"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": "",
"key": "",
"value": "",
},
"blocking": True,
},
{
"domain": DOMAIN,
"service": "start_program",
"service_data": {
"device_id": "DEVICE_ID",
"program": "",
"key": "",
"value": "",
"unit": "C",
},
"blocking": True,
},
]
SERVICE_APPLIANCE_METHOD_MAPPING = {
"set_option_active": "set_options_active_program",
"set_option_selected": "set_options_selected_program",
"change_setting": "set_setting",
"pause_program": "execute_command",
"resume_program": "execute_command",
"select_program": "select_program",
"start_program": "start_program",
}
async def test_api_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[], Awaitable[bool]],
setup_credentials: None,
get_appliances: MagicMock,
) -> None:
"""Test setup and unload."""
get_appliances.side_effect = get_all_appliances
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state == ConfigEntryState.NOT_LOADED
async def test_exception_handling(
bypass_throttle: Generator[None, Any, None],
hass: HomeAssistant,
integration_setup: Callable[[], Awaitable[bool]],
config_entry: MockConfigEntry,
setup_credentials: None,
get_appliances: MagicMock,
problematic_appliance: Mock,
) -> None:
"""Test exception handling."""
get_appliances.return_value = [problematic_appliance]
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
@pytest.mark.parametrize("token_expiration_time", [12345])
async def test_token_refresh_success(
bypass_throttle: Generator[None, Any, None],
integration_setup: Callable[[], Awaitable[bool]],
config_entry: MockConfigEntry,
aioclient_mock: AiohttpClientMocker,
requests_mock: requests_mock.Mocker,
setup_credentials: None,
) -> None:
"""Test where token is expired and the refresh attempt succeeds."""
assert config_entry.data["token"]["access_token"] == FAKE_ACCESS_TOKEN
requests_mock.post(OAUTH2_TOKEN, json=SERVER_ACCESS_TOKEN)
requests_mock.get("/api/homeappliances", json={"data": {"homeappliances": []}})
aioclient_mock.post(
OAUTH2_TOKEN,
json=SERVER_ACCESS_TOKEN,
)
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
# Verify token request
assert aioclient_mock.call_count == 1
assert aioclient_mock.mock_calls[0][2] == {
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"grant_type": "refresh_token",
"refresh_token": FAKE_REFRESH_TOKEN,
}
# Verify updated token
assert (
config_entry.data["token"]["access_token"]
== SERVER_ACCESS_TOKEN["access_token"]
)
async def test_setup(
hass: HomeAssistant,
integration_setup: Callable[[], Awaitable[bool]],
config_entry: MockConfigEntry,
setup_credentials: None,
) -> None:
"""Test setting up the integration."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state == ConfigEntryState.NOT_LOADED
async def test_update_throttle(
appliance: Mock,
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[], Awaitable[bool]],
setup_credentials: None,
platforms: list[Platform],
get_appliances: MagicMock,
) -> None:
"""Test to check Throttle functionality."""
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
assert get_appliances.call_count == 0
async def test_http_error(
bypass_throttle: Generator[None, Any, None],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[], Awaitable[bool]],
setup_credentials: None,
get_appliances: MagicMock,
) -> None:
"""Test HTTP errors during setup integration."""
get_appliances.side_effect = HTTPError(response=MagicMock())
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
assert get_appliances.call_count == 1
@pytest.mark.parametrize(
"service_call",
SERVICE_KV_CALL_PARAMS + SERVICE_COMMAND_CALL_PARAMS + SERVICE_PROGRAM_CALL_PARAMS,
)
async def test_services(
service_call: list[dict[str, Any]],
bypass_throttle: Generator[None, Any, None],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[], Awaitable[bool]],
setup_credentials: None,
get_appliances: MagicMock,
appliance: Mock,
) -> None:
"""Create and test services."""
get_appliances.return_value = [appliance]
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
device_registry = dr.async_get(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, appliance.haId)},
)
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(appliance, SERVICE_APPLIANCE_METHOD_MAPPING[service_name]).call_count
== 1
)
async def test_services_exception(
bypass_throttle: Generator[None, Any, None],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[], Awaitable[bool]],
setup_credentials: None,
get_appliances: MagicMock,
appliance: Mock,
) -> None:
"""Raise a ValueError when device id does not match."""
get_appliances.return_value = [appliance]
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
service_call = SERVICE_KV_CALL_PARAMS[0]
with pytest.raises(ValueError):
service_call["service_data"]["device_id"] = "DOES_NOT_EXISTS"
await hass.services.async_call(**service_call)
await hass.async_block_till

View file

@ -0,0 +1,205 @@
"""Tests for home_connect sensor entities."""
from collections.abc import Awaitable, Callable, Generator
from typing import Any
from unittest.mock import MagicMock, Mock
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_component import async_update_entity
from tests.common import MockConfigEntry
TEST_HC_APP = "Dishwasher"
EVENT_PROG_DELAYED_START = {
"BSH.Common.Status.OperationState": {
"value": "BSH.Common.EnumType.OperationState.Delayed"
},
}
EVENT_PROG_REMAIN_NO_VALUE = {
"BSH.Common.Option.RemainingProgramTime": {},
"BSH.Common.Status.OperationState": {
"value": "BSH.Common.EnumType.OperationState.Delayed"
},
}
EVENT_PROG_RUN = {
"BSH.Common.Option.RemainingProgramTime": {"value": "0"},
"BSH.Common.Option.ProgramProgress": {"value": "60"},
"BSH.Common.Status.OperationState": {
"value": "BSH.Common.EnumType.OperationState.Run"
},
}
EVENT_PROG_UPDATE_1 = {
"BSH.Common.Option.RemainingProgramTime": {"value": "0"},
"BSH.Common.Option.ProgramProgress": {"value": "80"},
"BSH.Common.Status.OperationState": {
"value": "BSH.Common.EnumType.OperationState.Run"
},
}
EVENT_PROG_UPDATE_2 = {
"BSH.Common.Option.RemainingProgramTime": {"value": "20"},
"BSH.Common.Option.ProgramProgress": {"value": "99"},
"BSH.Common.Status.OperationState": {
"value": "BSH.Common.EnumType.OperationState.Run"
},
}
EVENT_PROG_END = {
"BSH.Common.Status.OperationState": {
"value": "BSH.Common.EnumType.OperationState.Ready"
},
}
@pytest.fixture
def platforms() -> list[str]:
"""Fixture to specify platforms to test."""
return [Platform.SENSOR]
async def test_sensors(
bypass_throttle: Generator[None, Any, None],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[], Awaitable[bool]],
setup_credentials: None,
get_appliances: MagicMock,
appliance: Mock,
) -> None:
"""Test sensor entities."""
get_appliances.return_value = [appliance]
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
# Appliance program sequence with a delayed start.
PROGRAM_SEQUENCE_EVENTS = (
EVENT_PROG_DELAYED_START,
EVENT_PROG_RUN,
EVENT_PROG_UPDATE_1,
EVENT_PROG_UPDATE_2,
EVENT_PROG_END,
)
# Entity mapping to expected state at each program sequence.
ENTITY_ID_STATES = {
"sensor.dishwasher_operation_state": (
"Delayed",
"Run",
"Run",
"Run",
"Ready",
),
"sensor.dishwasher_remaining_program_time": (
"unavailable",
"2021-01-09T12:00:00+00:00",
"2021-01-09T12:00:00+00:00",
"2021-01-09T12:00:20+00:00",
"unavailable",
),
"sensor.dishwasher_program_progress": (
"unavailable",
"60",
"80",
"99",
"unavailable",
),
}
@pytest.mark.parametrize("appliance", [TEST_HC_APP], indirect=True)
@pytest.mark.parametrize(
("states", "event_run"),
list(zip(list(zip(*ENTITY_ID_STATES.values())), PROGRAM_SEQUENCE_EVENTS)),
)
async def test_event_sensors(
appliance: Mock,
states: tuple,
event_run: dict,
freezer: FrozenDateTimeFactory,
bypass_throttle: Generator[None, Any, None],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[], Awaitable[bool]],
setup_credentials: None,
get_appliances: MagicMock,
) -> None:
"""Test sequence for sensors that are only available after an event happens."""
entity_ids = ENTITY_ID_STATES.keys()
time_to_freeze = "2021-01-09 12:00:00+00:00"
freezer.move_to(time_to_freeze)
get_appliances.return_value = [appliance]
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
appliance.status.update(event_run)
for entity_id, state in zip(entity_ids, states):
await async_update_entity(hass, entity_id)
await hass.async_block_till_done()
assert hass.states.is_state(entity_id, state)
# Program sequence for SensorDeviceClass.TIMESTAMP edge cases.
PROGRAM_SEQUENCE_EDGE_CASE = [
EVENT_PROG_REMAIN_NO_VALUE,
EVENT_PROG_RUN,
EVENT_PROG_END,
EVENT_PROG_END,
]
# Expected state at each sequence.
ENTITY_ID_EDGE_CASE_STATES = [
"unavailable",
"2021-01-09T12:00:01+00:00",
"unavailable",
"unavailable",
]
@pytest.mark.parametrize("appliance", [TEST_HC_APP], indirect=True)
async def test_remaining_prog_time_edge_cases(
appliance: Mock,
freezer: FrozenDateTimeFactory,
bypass_throttle: Generator[None, Any, None],
hass: HomeAssistant,
config_entry: MockConfigEntry,
integration_setup: Callable[[], Awaitable[bool]],
setup_credentials: None,
get_appliances: MagicMock,
) -> None:
"""Run program sequence to test edge cases for the remaining_prog_time entity."""
get_appliances.return_value = [appliance]
entity_id = "sensor.dishwasher_remaining_program_time"
time_to_freeze = "2021-01-09 12:00:00+00:00"
freezer.move_to(time_to_freeze)
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert await integration_setup()
assert config_entry.state == ConfigEntryState.LOADED
for (
event,
expected_state,
) in zip(PROGRAM_SEQUENCE_EDGE_CASE, ENTITY_ID_EDGE_CASE_STATES):
appliance.status.update(event)
await async_update_entity(hass, entity_id)
await hass.async_block_till_done()
freezer.tick()
assert hass.states.is_state(entity_id, expected_state)