Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-07-13 01:27:57 +01:00
Code Activity

iaqualink: complete test coverage, bump to silver (#168268)

Browse Source
This commit is contained in:
Florent Thoumie
2026-05-26 12:40:54 -07:00
committed by GitHub
parent 6a18e05bda
commit b6fa89c032
25 changed files with 2095 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/iaqualink/__init__.py
+12 -5
View File
@@ -35,6 +35,7 @@ from homeassistant.util.ssl import SSL_ALPN_HTTP11_HTTP2
from .const import DOMAIN
from .coordinator import AqualinkDataUpdateCoordinator
from .entity import AqualinkEntity
from .utils import error_detail
_LOGGER = logging.getLogger(__name__)
@@ -86,7 +87,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: AqualinkConfigEntry) ->
except (AqualinkServiceException, TimeoutError, httpx.HTTPError) as aio_exception:
await aqualink.close()
raise ConfigEntryNotReady(
f"Error while attempting login: {aio_exception}"
f"Error while attempting login: {error_detail(aio_exception)}"
) from aio_exception
try:
@@ -96,10 +97,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: AqualinkConfigEntry) ->
raise ConfigEntryAuthFailed(
"Invalid credentials for iAquaLink"
) from auth_exception
except AqualinkServiceException as svc_exception:
except (AqualinkServiceException, TimeoutError, httpx.HTTPError) as svc_exception:
await aqualink.close()
raise ConfigEntryNotReady(
f"Error while attempting to retrieve systems list: {svc_exception}"
"Error while attempting to retrieve systems list: "
f"{error_detail(svc_exception)}"
) from svc_exception
systems_list = list(systems.values())
@@ -132,10 +134,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: AqualinkConfigEntry) ->
raise ConfigEntryAuthFailed(
"Invalid credentials for iAquaLink"
) from auth_exception
except AqualinkServiceException as svc_exception:
except (
AqualinkServiceException,
TimeoutError,
httpx.HTTPError,
) as svc_exception:
await aqualink.close()
raise ConfigEntryNotReady(
f"Error while attempting to retrieve devices list: {svc_exception}"
"Error while attempting to retrieve devices list: "
f"{error_detail(svc_exception)}"
) from svc_exception
device_registry = dr.async_get(hass)
homeassistant/components/iaqualink/climate.py
+11 -3
View File
@@ -74,9 +74,13 @@ class HassAqualinkThermostat(AqualinkEntity[AqualinkThermostat], ClimateEntity):
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Turn the underlying heater switch on or off."""
if hvac_mode == HVACMode.HEAT:
await await_or_reraise(self.dev.turn_on())
await await_or_reraise(
self.hass, self.coordinator.config_entry, self.dev.turn_on()
)
elif hvac_mode == HVACMode.OFF:
await await_or_reraise(self.dev.turn_off())
await await_or_reraise(
self.hass, self.coordinator.config_entry, self.dev.turn_off()
)
else:
_LOGGER.warning("Unknown operation mode: %s", hvac_mode)
@@ -98,7 +102,11 @@ class HassAqualinkThermostat(AqualinkEntity[AqualinkThermostat], ClimateEntity):
@refresh_system
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
await await_or_reraise(self.dev.set_temperature(int(kwargs[ATTR_TEMPERATURE])))
await await_or_reraise(
self.hass,
self.coordinator.config_entry,
self.dev.set_temperature(int(kwargs[ATTR_TEMPERATURE])),
)
@property
def current_temperature(self) -> float | None:
homeassistant/components/iaqualink/config_flow.py
+3 -3
View File
@@ -1,4 +1,4 @@
"""Config flow to configure zone component."""
"""Config flow for iAquaLink."""
from collections.abc import Mapping
from typing import Any
@@ -31,7 +31,7 @@ CREDENTIALS_DATA_SCHEMA = vol.Schema(
class AqualinkFlowHandler(ConfigFlow, domain=DOMAIN):
"""Aqualink config flow."""
"""iAquaLink config flow."""
VERSION = 1
@@ -50,7 +50,7 @@ class AqualinkFlowHandler(ConfigFlow, domain=DOMAIN):
pass
except AqualinkServiceUnauthorizedException:
return {"base": "invalid_auth"}
except AqualinkServiceException, httpx.HTTPError:
except AqualinkServiceException, TimeoutError, httpx.HTTPError:
return {"base": "cannot_connect"}
return {}
homeassistant/components/iaqualink/coordinator.py
+4 -2
View File
@@ -16,6 +16,7 @@ from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, UPDATE_INTERVAL_BY_SYSTEM_TYPE, UPDATE_INTERVAL_DEFAULT
from .utils import error_detail
_LOGGER = logging.getLogger(__name__)
@@ -51,9 +52,10 @@ class AqualinkDataUpdateCoordinator(DataUpdateCoordinator[None]):
self.system.serial,
)
return
except (AqualinkServiceException, httpx.HTTPError) as err:
except (AqualinkServiceException, TimeoutError, httpx.HTTPError) as err:
raise UpdateFailed(
f"Unable to update iAquaLink system {self.system.serial}: {err}"
"Unable to update iAquaLink system "
f"{self.system.serial}: {error_detail(err)}"
) from err
if self.system.online is not True:
raise UpdateFailed(f"iAquaLink system {self.system.serial} is offline")
homeassistant/components/iaqualink/light.py
+15 -5
View File
@@ -67,18 +67,28 @@ class HassAqualinkLight(AqualinkEntity[AqualinkLight], LightEntity):
"""
# For now I'm assuming lights support either effects or brightness.
if effect_name := kwargs.get(ATTR_EFFECT):
await await_or_reraise(self.dev.set_effect_by_name(effect_name))
await await_or_reraise(
self.hass,
self.coordinator.config_entry,
self.dev.set_effect_by_name(effect_name),
)
elif brightness := kwargs.get(ATTR_BRIGHTNESS):
# Aqualink supports percentages in 25% increments.
pct = round(brightness * 4.0 / 255) * 25
await await_or_reraise(self.dev.set_brightness(pct))
await await_or_reraise(
self.hass, self.coordinator.config_entry, self.dev.set_brightness(pct)
)
else:
await await_or_reraise(self.dev.turn_on())
await await_or_reraise(
self.hass, self.coordinator.config_entry, self.dev.turn_on()
)
@refresh_system
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the light."""
await await_or_reraise(self.dev.turn_off())
await await_or_reraise(
self.hass, self.coordinator.config_entry, self.dev.turn_off()
)
@property
def brightness(self) -> int:
@@ -86,7 +96,7 @@ class HassAqualinkLight(AqualinkEntity[AqualinkLight], LightEntity):
The scale needs converting between 0-100 and 0-255.
"""
return self.dev.brightness * 255 / 100
return round(self.dev.brightness * 255 / 100)
@property
def effect(self) -> str:
homeassistant/components/iaqualink/manifest.json
+1 -1
View File
@@ -8,7 +8,7 @@
"integration_type": "hub",
"iot_class": "cloud_polling",
"loggers": ["iaqualink"],
"quality_scale": "bronze",
"quality_scale": "silver",
"requirements": ["iaqualink==0.7.0", "h2==4.3.0"],
"single_config_entry": true
}
homeassistant/components/iaqualink/quality_scale.yaml
+2 -2
View File
@@ -24,7 +24,7 @@ rules:
unique-config-entry: done
# Silver
action-exceptions: todo
action-exceptions: done
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
@@ -35,7 +35,7 @@ rules:
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: done
test-coverage: todo
test-coverage: done
# Gold
devices: done
homeassistant/components/iaqualink/switch.py
+6 -2
View File
@@ -56,9 +56,13 @@ class HassAqualinkSwitch(AqualinkEntity[AqualinkSwitch], SwitchEntity):
@refresh_system
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
await await_or_reraise(self.dev.turn_on())
await await_or_reraise(
self.hass, self.coordinator.config_entry, self.dev.turn_on()
)
@refresh_system
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
await await_or_reraise(self.dev.turn_off())
await await_or_reraise(
self.hass, self.coordinator.config_entry, self.dev.turn_off()
)
homeassistant/components/iaqualink/utils.py
+32 -4
View File
@@ -3,14 +3,42 @@
from collections.abc import Awaitable
import httpx
from iaqualink.exception import AqualinkServiceException
from iaqualink.exception import (
AqualinkServiceException,
AqualinkServiceUnauthorizedException,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
async def await_or_reraise(awaitable: Awaitable) -> None:
def error_detail(err: Exception) -> str:
"""Return a non-empty error detail for iaqualink exceptions."""
if detail := str(err):
return detail
return type(err).__name__
async def await_or_reraise(
hass: HomeAssistant,
config_entry: ConfigEntry | None,
awaitable: Awaitable,
) -> None:
"""Execute API call while catching service exceptions."""
try:
await awaitable
except AqualinkServiceUnauthorizedException as auth_exception:
if config_entry is not None:
config_entry.async_start_reauth(hass)
raise ConfigEntryAuthFailed(
"Invalid credentials for iAquaLink"
) from auth_exception
except TimeoutError as timeout_exception:
raise HomeAssistantError(
f"Aqualink error: {error_detail(timeout_exception)}"
) from timeout_exception
except (AqualinkServiceException, httpx.HTTPError) as svc_exception:
raise HomeAssistantError(f"Aqualink error: {svc_exception}") from svc_exception
raise HomeAssistantError(
f"Aqualink error: {error_detail(svc_exception)}"
) from svc_exception
tests/components/iaqualink/conftest.py
+141 -10
View File
@@ -1,15 +1,26 @@
"""Configuration for iAquaLink tests."""
import random
from dataclasses import dataclass
from unittest.mock import AsyncMock, PropertyMock, patch
from iaqualink.client import AqualinkClient
from iaqualink.device import AqualinkDevice
from iaqualink.system import AqualinkSystem
from iaqualink.systems.iaqua.device import (
IaquaAuxSwitch,
IaquaBinarySensor,
IaquaLightSwitch,
IaquaSensor,
IaquaThermostat,
)
from iaqualink.systems.iaqua.system import IaquaSystem
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.iaqualink import DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
@@ -18,14 +29,18 @@ MOCK_PASSWORD = "password"
MOCK_DATA = {CONF_USERNAME: MOCK_USERNAME, CONF_PASSWORD: MOCK_PASSWORD}
def async_returns(x):
"""Return value-returning async mock."""
return AsyncMock(return_value=x)
@dataclass
class SystemDevices:
"""All devices created by setup_integration."""
def async_raises(x):
"""Return exception-raising async mock."""
return AsyncMock(side_effect=x)
system: IaquaSystem
switch: IaquaAuxSwitch
light: IaquaLightSwitch
binary_sensor: IaquaBinarySensor
sensor: IaquaSensor
thermostat: IaquaThermostat
heater: IaquaAuxSwitch
pool_temp: IaquaSensor
@pytest.fixture(name="client")
@@ -42,13 +57,129 @@ def get_aqualink_system(aqualink, cls=None, data=None):
if data is None:
data = {}
num = random.randint(0, 99999)
data["name"] = "Pool"
data["serial_number"] = f"SN{num:05}"
data["serial_number"] = "SN00001"
return cls(aqualink=aqualink, data=data)
async def setup_integration(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> SystemDevices:
"""Set up the iAquaLink integration with all platform entities."""
system = get_aqualink_system(
client,
cls=IaquaSystem,
data={"home_screen": [{}, {}, {}, {"temp_scale": "F"}]},
)
system.online = True
async def update() -> None:
system.temp_unit = "F"
system.update = AsyncMock(side_effect=update)
switch = get_aqualink_device(
system, name="aux_1", cls=IaquaAuxSwitch, data={"state": "1", "aux": "1"}
)
light = get_aqualink_device(
system,
name="aux_2",
cls=IaquaLightSwitch,
data={"state": "1", "aux": "2", "label": "Pool Light"},
)
binary_sensor = get_aqualink_device(
system, name="freeze_protection", cls=IaquaBinarySensor, data={"state": "1"}
)
sensor = get_aqualink_device(
system, name="ph", cls=IaquaSensor, data={"state": "7.2"}
)
thermostat = get_aqualink_device(
system, name="pool_set_point", cls=IaquaThermostat, data={"state": "84"}
)
heater = get_aqualink_device(
system, name="pool_heater", cls=IaquaAuxSwitch, data={"state": "1", "aux": "3"}
)
pool_temp = get_aqualink_device(
system, name="pool_temp", cls=IaquaSensor, data={"state": "80"}
)
platform_devices = {
d.name: d for d in (switch, light, binary_sensor, sensor, thermostat)
}
system.devices = {
**platform_devices,
heater.name: heater,
pool_temp.name: pool_temp,
}
system.get_devices = AsyncMock(return_value=platform_devices)
system.set_aux = AsyncMock()
system.set_temps = AsyncMock()
system.set_light = AsyncMock()
await setup_entry(hass, config_entry, system)
return SystemDevices(
system=system,
switch=switch,
light=light,
binary_sensor=binary_sensor,
sensor=sensor,
thermostat=thermostat,
heater=heater,
pool_temp=pool_temp,
)
async def setup_entry(
hass: HomeAssistant,
config_entry: MockConfigEntry,
system: AqualinkSystem,
) -> None:
"""Set up a config entry with a pre-built system."""
config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
return_value=None,
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
return_value={system.serial: system},
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async def assert_platform_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
domain: str,
) -> None:
"""Assert all entities for a given platform domain are set up correctly."""
await setup_integration(hass, config_entry, client)
entity_entries = [
e
for e in er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
)
if e.domain == domain
]
assert entity_entries
for entity_entry in entity_entries:
assert entity_entry == snapshot(name=f"{entity_entry.entity_id}-entry")
assert hass.states.get(entity_entry.entity_id) == snapshot(
name=f"{entity_entry.entity_id}-state"
)
def get_aqualink_device(system, name, cls=None, data=None):
"""Create aqualink device."""
if cls is None:
tests/components/iaqualink/snapshots/test_binary_sensor.ambr
+52
View File
@@ -0,0 +1,52 @@
# serializer version: 1
# name: test_setup[binary_sensor.freeze_protection-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.freeze_protection',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.COLD: 'cold'>,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'SN00001_freeze_protection',
'unit_of_measurement': None,
})
# ---
# name: test_setup[binary_sensor.freeze_protection-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'cold',
'friendly_name': 'Freeze Protection',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.freeze_protection',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
tests/components/iaqualink/snapshots/test_climate.ambr
+68
View File
@@ -0,0 +1,68 @@
# serializer version: 1
# name: test_setup[climate.pool_set_point-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.HEAT: 'heat'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 40.0,
'min_temp': 1.1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.pool_set_point',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 385>,
'translation_key': None,
'unique_id': 'SN00001_pool_set_point',
'unit_of_measurement': None,
})
# ---
# name: test_setup[climate.pool_set_point-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 26.7,
'friendly_name': 'Pool Set Point',
'hvac_action': <HVACAction.HEATING: 'heating'>,
'hvac_modes': list([
<HVACMode.HEAT: 'heat'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 40.0,
'min_temp': 1.1,
'supported_features': <ClimateEntityFeature: 385>,
'temperature': 28.9,
}),
'context': <ANY>,
'entity_id': 'climate.pool_set_point',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'heat',
})
# ---
tests/components/iaqualink/snapshots/test_entity.ambr
+278
View File
@@ -0,0 +1,278 @@
# serializer version: 1
# name: test_setup[binary_sensor.freeze_protection-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.freeze_protection',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.COLD: 'cold'>,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'SN00001_freeze_protection',
'unit_of_measurement': None,
})
# ---
# name: test_setup[binary_sensor.freeze_protection-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'cold',
'friendly_name': 'Freeze Protection',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.freeze_protection',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_setup[climate.pool_set_point-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.HEAT: 'heat'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 40.0,
'min_temp': 1.1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.pool_set_point',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 385>,
'translation_key': None,
'unique_id': 'SN00001_pool_set_point',
'unit_of_measurement': None,
})
# ---
# name: test_setup[climate.pool_set_point-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 26.7,
'friendly_name': 'Pool Set Point',
'hvac_action': <HVACAction.HEATING: 'heating'>,
'hvac_modes': list([
<HVACMode.HEAT: 'heat'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 40.0,
'min_temp': 1.1,
'supported_features': <ClimateEntityFeature: 385>,
'temperature': 28.9,
}),
'context': <ANY>,
'entity_id': 'climate.pool_set_point',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'heat',
})
# ---
# name: test_setup[light.pool_light-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'supported_color_modes': list([
<ColorMode.ONOFF: 'onoff'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'light',
'entity_category': None,
'entity_id': 'light.pool_light',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'SN00001_aux_2',
'unit_of_measurement': None,
})
# ---
# name: test_setup[light.pool_light-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'color_mode': <ColorMode.ONOFF: 'onoff'>,
'friendly_name': 'Pool Light',
'supported_color_modes': list([
<ColorMode.ONOFF: 'onoff'>,
]),
'supported_features': <LightEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'light.pool_light',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_setup[sensor.ph-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ph',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'SN00001_ph',
'unit_of_measurement': None,
})
# ---
# name: test_setup[sensor.ph-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ph',
}),
'context': <ANY>,
'entity_id': 'sensor.ph',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '7.2',
})
# ---
# name: test_setup[switch.aux_1-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.aux_1',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'SN00001_aux_1',
'unit_of_measurement': None,
})
# ---
# name: test_setup[switch.aux_1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Aux 1',
}),
'context': <ANY>,
'entity_id': 'switch.aux_1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
tests/components/iaqualink/snapshots/test_light.ambr
+60
View File
@@ -0,0 +1,60 @@
# serializer version: 1
# name: test_setup[light.pool_light-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'supported_color_modes': list([
<ColorMode.ONOFF: 'onoff'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'light',
'entity_category': None,
'entity_id': 'light.pool_light',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'SN00001_aux_2',
'unit_of_measurement': None,
})
# ---
# name: test_setup[light.pool_light-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'color_mode': <ColorMode.ONOFF: 'onoff'>,
'friendly_name': 'Pool Light',
'supported_color_modes': list([
<ColorMode.ONOFF: 'onoff'>,
]),
'supported_features': <LightEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'light.pool_light',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
tests/components/iaqualink/snapshots/test_sensor.ambr
+51
View File
@@ -0,0 +1,51 @@
# serializer version: 1
# name: test_setup[sensor.ph-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ph',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'SN00001_ph',
'unit_of_measurement': None,
})
# ---
# name: test_setup[sensor.ph-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ph',
}),
'context': <ANY>,
'entity_id': 'sensor.ph',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '7.2',
})
# ---
tests/components/iaqualink/snapshots/test_switch.ambr
+51
View File
@@ -0,0 +1,51 @@
# serializer version: 1
# name: test_setup[switch.aux_1-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.aux_1',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'iaqualink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'SN00001_aux_1',
'unit_of_measurement': None,
})
# ---
# name: test_setup[switch.aux_1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Aux 1',
}),
'context': <ANY>,
'entity_id': 'switch.aux_1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
tests/components/iaqualink/test_binary_sensor.py
+117
View File
@@ -0,0 +1,117 @@
"""Binary sensor platform tests for iAquaLink."""
from __future__ import annotations
from unittest.mock import AsyncMock
from iaqualink.client import AqualinkClient
from iaqualink.systems.iaqua.device import IaquaBinarySensor
from iaqualink.systems.iaqua.system import IaquaSystem
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.binary_sensor import (
DOMAIN as BINARY_SENSOR_DOMAIN,
BinarySensorDeviceClass,
)
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .conftest import (
assert_platform_setup,
get_aqualink_device,
get_aqualink_system,
setup_entry,
)
from tests.common import MockConfigEntry
async def test_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all binary sensor entities are created correctly."""
await assert_platform_setup(
hass, config_entry, client, entity_registry, snapshot, BINARY_SENSOR_DOMAIN
)
async def _setup_binary_sensor(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
*,
name: str,
state: str,
) -> str:
"""Set up the integration with a single binary sensor entity."""
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
sensor = get_aqualink_device(
system, name=name, cls=IaquaBinarySensor, data={"state": state}
)
system.get_devices = AsyncMock(return_value={sensor.name: sensor})
await setup_entry(hass, config_entry, system)
entity_ids = hass.states.async_entity_ids(BINARY_SENSOR_DOMAIN)
assert len(entity_ids) == 1
return entity_ids[0]
@pytest.mark.parametrize(
("name", "expected_class"),
[
pytest.param("freeze_protection", BinarySensorDeviceClass.COLD, id="freeze"),
pytest.param("auxiliary", None, id="default"),
],
)
async def test_binary_sensor_device_class(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
entity_registry: er.EntityRegistry,
name: str,
expected_class: BinarySensorDeviceClass | None,
) -> None:
"""Test binary sensor device class mapping."""
entity_id = await _setup_binary_sensor(
hass,
config_entry,
client,
name=name,
state="1",
)
entry = entity_registry.async_get(entity_id)
assert entry is not None
assert entry.original_device_class == expected_class
assert entry.has_entity_name is True
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_ON
async def test_binary_sensor_off_state(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test binary sensor off state is exposed through Home Assistant."""
entity_id = await _setup_binary_sensor(
hass,
config_entry,
client,
name="freeze_protection",
state="0",
)
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_OFF
tests/components/iaqualink/test_climate.py
+355
View File
@@ -0,0 +1,355 @@
"""Climate platform tests for iAquaLink."""
from __future__ import annotations
import logging
from typing import Any
from unittest.mock import AsyncMock
import httpx
from iaqualink.client import AqualinkClient
from iaqualink.exception import (
AqualinkServiceException,
AqualinkServiceUnauthorizedException,
)
from iaqualink.systems.iaqua.device import (
AqualinkState,
IaquaAuxSwitch,
IaquaSensor,
IaquaThermostat,
)
from iaqualink.systems.iaqua.system import IaquaSystem
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
DOMAIN as CLIMATE_DOMAIN,
SERVICE_SET_HVAC_MODE,
SERVICE_SET_TEMPERATURE,
HVACAction,
HVACMode,
)
from homeassistant.const import ATTR_ENTITY_ID, ATTR_TEMPERATURE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.util.unit_system import US_CUSTOMARY_SYSTEM
from .conftest import (
assert_platform_setup,
get_aqualink_device,
get_aqualink_system,
setup_entry,
)
from tests.common import MockConfigEntry
async def test_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all climate entities are created correctly."""
await assert_platform_setup(
hass, config_entry, client, entity_registry, snapshot, CLIMATE_DOMAIN
)
async def _setup_thermostat(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
*,
temp_unit: str = "F",
heater_state: str = AqualinkState.ON.value,
target_temperature: str = "84",
current_temperature: str = "80",
) -> tuple[IaquaSystem, object, object, object, str, object]:
"""Set up the integration with a single thermostat entity."""
if temp_unit == "F":
hass.config.units = US_CUSTOMARY_SYSTEM
system = get_aqualink_system(
client,
cls=IaquaSystem,
data={"home_screen": [{}, {}, {}, {"temp_scale": temp_unit}]},
)
system.online = True
async def update() -> None:
system.temp_unit = temp_unit
system.update = AsyncMock(side_effect=update)
heater = get_aqualink_device(
system,
name="pool_heater",
cls=IaquaAuxSwitch,
data={"state": heater_state, "aux": "1"},
)
sensor = get_aqualink_device(
system,
name="pool_temp",
cls=IaquaSensor,
data={"state": current_temperature},
)
thermostat = get_aqualink_device(
system,
name="pool_set_point",
cls=IaquaThermostat,
data={"state": target_temperature},
)
system.devices = {
heater.name: heater,
sensor.name: sensor,
thermostat.name: thermostat,
}
system.get_devices = AsyncMock(return_value={thermostat.name: thermostat})
system.set_aux = AsyncMock()
system.set_temps = AsyncMock()
await setup_entry(hass, config_entry, system)
entity_ids = hass.states.async_entity_ids(CLIMATE_DOMAIN)
assert len(entity_ids) == 1
entity_id = entity_ids[0]
entity_state = hass.states.get(entity_id)
assert entity_state is not None
return system, heater, sensor, thermostat, entity_id, entity_state
@pytest.mark.parametrize(
("heater_state", "expected_action", "expected_mode"),
[
pytest.param(
AqualinkState.ON.value, HVACAction.HEATING, HVACMode.HEAT, id="heating"
),
pytest.param(
AqualinkState.ENABLED.value, HVACAction.IDLE, HVACMode.OFF, id="idle"
),
pytest.param(AqualinkState.OFF.value, HVACAction.OFF, HVACMode.OFF, id="off"),
],
)
async def test_thermostat_properties(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
heater_state: str,
expected_action: HVACAction,
expected_mode: HVACMode,
) -> None:
"""Test thermostat properties and HVAC action mapping."""
_, _, _, _, _, entity_state = await _setup_thermostat(
hass,
config_entry,
client,
heater_state=heater_state,
)
assert entity_state.state == expected_mode
assert entity_state.attributes["hvac_action"] == expected_action
assert entity_state.attributes["temperature"] == 84.0
assert entity_state.attributes["current_temperature"] == 80.0
async def test_thermostat_current_temperature_none(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test thermostat current temperature handles empty values."""
_, _, _, _, _, entity_state = await _setup_thermostat(
hass,
config_entry,
client,
temp_unit="C",
heater_state=AqualinkState.OFF.value,
target_temperature="24",
current_temperature="",
)
assert entity_state.state == HVACMode.OFF
assert entity_state.attributes["current_temperature"] is None
@pytest.mark.parametrize(
("hvac_mode", "initial_state", "expected_state"),
[
pytest.param(HVACMode.HEAT, AqualinkState.OFF.value, HVACMode.HEAT, id="heat"),
pytest.param(HVACMode.OFF, AqualinkState.ON.value, HVACMode.OFF, id="off"),
],
)
async def test_thermostat_set_hvac_mode(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
hvac_mode: HVACMode,
initial_state: str,
expected_state: HVACMode,
) -> None:
"""Test thermostat HVAC mode service updates Home Assistant state."""
system, heater, _, _, entity_id, _ = await _setup_thermostat(
hass,
config_entry,
client,
heater_state=initial_state,
)
async def set_aux(_: str) -> None:
heater.data["state"] = (
AqualinkState.ON.value
if hvac_mode == HVACMode.HEAT
else AqualinkState.OFF.value
)
system.set_aux = AsyncMock(side_effect=set_aux)
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_HVAC_MODE,
{ATTR_ENTITY_ID: entity_id, ATTR_HVAC_MODE: hvac_mode},
blocking=True,
)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == expected_state
async def test_thermostat_set_temperature(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test thermostat target temperature service updates Home Assistant state."""
system, _, _, thermostat, entity_id, _ = await _setup_thermostat(
hass,
config_entry,
client,
)
async def set_temps(data: dict[str, str]) -> None:
thermostat.data["state"] = data["temp1"]
system.set_temps = AsyncMock(side_effect=set_temps)
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: entity_id, ATTR_TEMPERATURE: 84.9},
blocking=True,
)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.attributes["temperature"] == 84.0
@pytest.mark.parametrize(
("service", "service_data", "system_attr"),
[
pytest.param(
SERVICE_SET_HVAC_MODE,
{ATTR_HVAC_MODE: HVACMode.HEAT},
"set_aux",
id="set-hvac-mode",
),
pytest.param(
SERVICE_SET_TEMPERATURE,
{ATTR_TEMPERATURE: 84},
"set_temps",
id="set-temperature",
),
],
)
@pytest.mark.parametrize(
("raised_exception", "expected_exception", "match"),
[
pytest.param(
AqualinkServiceException,
HomeAssistantError,
"Aqualink error: AqualinkServiceException",
id="service",
),
pytest.param(
TimeoutError(),
HomeAssistantError,
"Aqualink error: TimeoutError",
id="timeout",
),
pytest.param(
httpx.HTTPError("boom"),
HomeAssistantError,
"Aqualink error: boom",
id="http",
),
pytest.param(
AqualinkServiceUnauthorizedException,
ConfigEntryAuthFailed,
"Invalid credentials for iAquaLink",
id="unauthorized",
),
],
)
async def test_climate_action_errors_leave_state_unchanged(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
service: str,
service_data: dict[str, Any],
system_attr: str,
raised_exception: Exception | type[Exception],
expected_exception: type[Exception],
match: str,
) -> None:
"""Test climate action errors are surfaced through the service call."""
system, _, _, _, entity_id, entity_state = await _setup_thermostat(
hass,
config_entry,
client,
heater_state=AqualinkState.OFF.value,
)
initial_state = entity_state.state
setattr(system, system_attr, AsyncMock(side_effect=raised_exception))
with pytest.raises(expected_exception, match=match):
await hass.services.async_call(
CLIMATE_DOMAIN,
service,
{ATTR_ENTITY_ID: entity_id, **service_data},
blocking=True,
)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == initial_state
async def test_thermostat_set_unknown_hvac_mode_logs_warning(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test unsupported HVAC modes log a warning and keep state unchanged."""
system, _, _, _, entity_id, _ = await _setup_thermostat(
hass,
config_entry,
client,
heater_state=AqualinkState.OFF.value,
)
climate_component = hass.data[CLIMATE_DOMAIN]
entity = climate_component.get_entity(entity_id)
assert entity is not None
with caplog.at_level(logging.WARNING):
await entity.async_set_hvac_mode(HVACMode.COOL)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == HVACMode.OFF
assert "Unknown operation mode" in caplog.text
system.set_aux.assert_not_called()
tests/components/iaqualink/test_config_flow.py
+55 -25
View File
@@ -1,11 +1,14 @@
"""Tests for iAquaLink config flow."""
from typing import Any
from unittest.mock import patch
import httpx
from iaqualink.exception import (
AqualinkServiceException,
AqualinkServiceUnauthorizedException,
)
import pytest
from homeassistant.components.iaqualink import DOMAIN, config_flow
from homeassistant.config_entries import SOURCE_DHCP, SOURCE_USER
@@ -23,12 +26,15 @@ DHCP_DISCOVERY = DhcpServiceInfo(
)
async def test_already_configured(
async def _async_mock_login(self: Any) -> None:
"""Mock a successful login."""
async def test_single_instance_allowed(
hass: HomeAssistant,
config_entry: MockConfigEntry,
config_data: dict[str, str],
) -> None:
"""Test config flow when iaqualink component is already setup."""
"""Test config flow aborts when an entry already exists."""
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
@@ -70,7 +76,7 @@ async def test_dhcp_discovery_starts_user_flow(
with (
patch(
"homeassistant.components.iaqualink.config_flow.AqualinkClient.login",
return_value=None,
_async_mock_login,
),
patch(
"homeassistant.components.iaqualink.async_setup_entry",
@@ -107,16 +113,26 @@ async def test_with_invalid_credentials(
assert result["errors"] == {"base": "invalid_auth"}
async def test_service_exception(
hass: HomeAssistant, config_data: dict[str, str]
@pytest.mark.parametrize(
"raised_exception",
[
pytest.param(AqualinkServiceException, id="service"),
pytest.param(TimeoutError, id="timeout"),
pytest.param(httpx.HTTPError("boom"), id="http"),
],
)
async def test_cannot_connect_exception(
hass: HomeAssistant,
config_data: dict[str, str],
raised_exception: Exception | type[Exception],
) -> None:
"""Test config flow encountering service exception."""
"""Test config flow encountering a connection-related exception."""
flow = config_flow.AqualinkFlowHandler()
flow.hass = hass
with patch(
"homeassistant.components.iaqualink.config_flow.AqualinkClient.login",
side_effect=AqualinkServiceException,
side_effect=raised_exception,
):
result = await flow.async_step_user(config_data)
@@ -131,10 +147,11 @@ async def test_with_existing_config(
"""Test config flow with existing configuration."""
flow = config_flow.AqualinkFlowHandler()
flow.hass = hass
flow.context = {}
with patch(
"homeassistant.components.iaqualink.config_flow.AqualinkClient.login",
return_value=None,
_async_mock_login,
):
result = await flow.async_step_user(config_data)
@@ -169,8 +186,6 @@ async def test_reauth_success(hass: HomeAssistant, config_data: dict[str, str])
)
entry.add_to_hass(hass)
new_username = "updated@example.com"
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
@@ -180,7 +195,7 @@ async def test_reauth_success(hass: HomeAssistant, config_data: dict[str, str])
with (
patch(
"homeassistant.components.iaqualink.config_flow.AqualinkClient.login",
return_value=None,
_async_mock_login,
),
patch(
"homeassistant.config_entries.ConfigEntries.async_reload",
@@ -189,16 +204,18 @@ async def test_reauth_success(hass: HomeAssistant, config_data: dict[str, str])
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_USERNAME: new_username, CONF_PASSWORD: "new_password"},
{
CONF_USERNAME: config_data[CONF_USERNAME],
CONF_PASSWORD: "new_password",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert entry.title == new_username
assert entry.title == config_data[CONF_USERNAME]
assert dict(entry.data) == {
**config_data,
CONF_USERNAME: new_username,
CONF_PASSWORD: "new_password",
}
@@ -214,8 +231,6 @@ async def test_reconfigure_success(
)
entry.add_to_hass(hass)
new_username = "updated@example.com"
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
@@ -225,7 +240,7 @@ async def test_reconfigure_success(
with (
patch(
"homeassistant.components.iaqualink.config_flow.AqualinkClient.login",
return_value=None,
_async_mock_login,
),
patch(
"homeassistant.config_entries.ConfigEntries.async_reload",
@@ -234,16 +249,15 @@ async def test_reconfigure_success(
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_USERNAME: new_username, CONF_PASSWORD: "new_password"},
{CONF_USERNAME: config_data[CONF_USERNAME], CONF_PASSWORD: "new_password"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.title == new_username
assert entry.title == config_data[CONF_USERNAME]
assert dict(entry.data) == {
**config_data,
CONF_USERNAME: new_username,
CONF_PASSWORD: "new_password",
}
@@ -252,7 +266,10 @@ async def test_reauth_invalid_auth(
hass: HomeAssistant, config_data: dict[str, str]
) -> None:
"""Test reauthentication with invalid credentials."""
entry = MockConfigEntry(domain=DOMAIN, data=config_data)
entry = MockConfigEntry(
domain=DOMAIN,
data=config_data,
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
@@ -271,18 +288,31 @@ async def test_reauth_invalid_auth(
assert result["errors"] == {"base": "invalid_auth"}
@pytest.mark.parametrize(
"raised_exception",
[
pytest.param(AqualinkServiceException, id="service"),
pytest.param(TimeoutError, id="timeout"),
pytest.param(httpx.HTTPError("boom"), id="http"),
],
)
async def test_reauth_cannot_connect(
hass: HomeAssistant, config_data: dict[str, str]
hass: HomeAssistant,
config_data: dict[str, str],
raised_exception: Exception | type[Exception],
) -> None:
"""Test reauthentication when the service cannot be reached."""
entry = MockConfigEntry(domain=DOMAIN, data=config_data)
entry = MockConfigEntry(
domain=DOMAIN,
data=config_data,
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
with patch(
"homeassistant.components.iaqualink.config_flow.AqualinkClient.login",
side_effect=AqualinkServiceException,
side_effect=raised_exception,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
tests/components/iaqualink/test_entity.py
+33
View File
@@ -0,0 +1,33 @@
"""Entity base tests for iAquaLink."""
from __future__ import annotations
from iaqualink.client import AqualinkClient
from syrupy.assertion import SnapshotAssertion
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .conftest import setup_integration
from tests.common import MockConfigEntry
async def test_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all entities are created by setup_integration."""
await setup_integration(hass, config_entry, client)
entity_entries = er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
)
assert entity_entries
for entity_entry in entity_entries:
assert entity_entry == snapshot(name=f"{entity_entry.entity_id}-entry")
state = hass.states.get(entity_entry.entity_id)
assert state == snapshot(name=f"{entity_entry.entity_id}-state")
tests/components/iaqualink/test_init.py
+105 -215
View File
@@ -3,19 +3,14 @@
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import httpx
from iaqualink.client import AqualinkClient
from iaqualink.exception import (
AqualinkServiceException,
AqualinkServiceThrottledException,
AqualinkServiceUnauthorizedException,
)
from iaqualink.systems.iaqua.device import (
IaquaAuxSwitch,
IaquaBinarySensor,
IaquaLightSwitch,
IaquaSensor,
IaquaThermostat,
)
from iaqualink.systems.iaqua.device import IaquaLightSwitch
from iaqualink.systems.iaqua.system import IaquaSystem
import pytest
@@ -39,7 +34,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from .conftest import get_aqualink_device, get_aqualink_system
from .conftest import get_aqualink_device, get_aqualink_system, setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed
@@ -55,37 +50,23 @@ async def _advance_coordinator_time(
await hass.async_block_till_done(wait_background_tasks=True)
@pytest.mark.parametrize(
"raised_exception",
[
pytest.param(AqualinkServiceException, id="service"),
pytest.param(TimeoutError, id="timeout"),
pytest.param(httpx.HTTPError("boom"), id="http"),
],
)
async def test_system_refresh_failure_marks_entities_unavailable(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
freezer: FrozenDateTimeFactory,
raised_exception: Exception | type[Exception],
) -> None:
"""Test a system refresh failure marks attached entities unavailable."""
config_entry.add_to_hass(hass)
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
systems = {system.serial: system}
light = get_aqualink_device(
system, name="aux_1", cls=IaquaLightSwitch, data={"state": "1"}
)
devices = {light.name: light}
system.get_devices = AsyncMock(return_value=devices)
with (
patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
return_value=None,
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
return_value=systems,
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
devices = await setup_integration(hass, config_entry, client)
entity_ids = hass.states.async_entity_ids(LIGHT_DOMAIN)
assert len(entity_ids) == 1
@@ -96,10 +77,10 @@ async def test_system_refresh_failure_marks_entities_unavailable(
assert state.state == STATE_ON
async def fail_update() -> None:
system.online = None
raise AqualinkServiceException
devices.system.online = None
raise raised_exception
system.update = AsyncMock(side_effect=fail_update)
devices.system.update = AsyncMock(side_effect=fail_update)
await _advance_coordinator_time(hass, freezer)
@@ -165,39 +146,16 @@ async def test_light_service_calls_update_entity_state(
client: AqualinkClient,
) -> None:
"""Test light service calls update entity state from device properties."""
config_entry.add_to_hass(hass)
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
systems = {system.serial: system}
light = get_aqualink_device(
system, name="aux_1", cls=IaquaLightSwitch, data={"state": "1"}
)
devices = {light.name: light}
system.get_devices = AsyncMock(return_value=devices)
devices = await setup_integration(hass, config_entry, client)
async def turn_off() -> None:
light.data["state"] = "0"
devices.light.data["state"] = "0"
async def turn_on() -> None:
light.data["state"] = "1"
devices.light.data["state"] = "1"
light.turn_off = AsyncMock(side_effect=turn_off)
light.turn_on = AsyncMock(side_effect=turn_on)
with (
patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
return_value=None,
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
return_value=systems,
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
devices.light.turn_off = AsyncMock(side_effect=turn_off)
devices.light.turn_on = AsyncMock(side_effect=turn_on)
entity_ids = hass.states.async_entity_ids(LIGHT_DOMAIN)
assert len(entity_ids) == 1
@@ -228,15 +186,25 @@ async def test_light_service_calls_update_entity_state(
assert state.state == STATE_ON
async def test_setup_login_exception(
hass: HomeAssistant, config_entry: MockConfigEntry
@pytest.mark.parametrize(
"raised_exception",
[
pytest.param(AqualinkServiceException, id="service"),
pytest.param(TimeoutError, id="timeout"),
pytest.param(httpx.HTTPError("boom"), id="http"),
],
)
async def test_setup_login_retry_exceptions(
hass: HomeAssistant,
config_entry: MockConfigEntry,
raised_exception: Exception | type[Exception],
) -> None:
"""Test setup encountering a login exception."""
"""Test setup retries on connection-related login exceptions."""
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
side_effect=AqualinkServiceException,
side_effect=raised_exception,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@@ -264,24 +232,18 @@ async def test_setup_login_unauthorized(
assert flows[0]["context"]["source"] == SOURCE_REAUTH
async def test_setup_login_timeout(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test setup encountering a timeout while logging in."""
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
side_effect=TimeoutError,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
@pytest.mark.parametrize(
"raised_exception",
[
pytest.param(AqualinkServiceException, id="service"),
pytest.param(TimeoutError, id="timeout"),
pytest.param(httpx.HTTPError("boom"), id="http"),
],
)
async def test_setup_systems_exception(
hass: HomeAssistant, config_entry: MockConfigEntry
hass: HomeAssistant,
config_entry: MockConfigEntry,
raised_exception: Exception | type[Exception],
) -> None:
"""Test setup encountering an exception while retrieving systems."""
config_entry.add_to_hass(hass)
@@ -293,7 +255,7 @@ async def test_setup_systems_exception(
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
side_effect=AqualinkServiceException,
side_effect=raised_exception,
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
@@ -385,15 +347,25 @@ async def test_setup_no_systems_recognized(
assert config_entry.state is ConfigEntryState.SETUP_ERROR
@pytest.mark.parametrize(
"raised_exception",
[
pytest.param(AqualinkServiceException, id="service"),
pytest.param(TimeoutError, id="timeout"),
pytest.param(httpx.HTTPError("boom"), id="http"),
],
)
async def test_setup_devices_exception(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
raised_exception: Exception | type[Exception],
) -> None:
"""Test setup encountering an exception while retrieving devices."""
config_entry.add_to_hass(hass)
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
systems = {system.serial: system}
@@ -411,13 +383,51 @@ async def test_setup_devices_exception(
"get_devices",
) as mock_get_devices,
):
mock_get_devices.side_effect = AqualinkServiceException
mock_get_devices.side_effect = raised_exception
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_setup_devices_unauthorized(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test setup encountering an unauthorized exception while retrieving devices."""
config_entry.add_to_hass(hass)
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
systems = {system.serial: system}
with (
patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
return_value=None,
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
return_value=systems,
),
patch.object(
system,
"get_devices",
side_effect=AqualinkServiceUnauthorizedException,
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
assert flows[0]["context"]["source"] == SOURCE_REAUTH
async def test_setup_all_good_no_recognized_devices(
hass: HomeAssistant,
config_entry: MockConfigEntry,
@@ -473,56 +483,7 @@ async def test_setup_all_good_all_device_types(
entity_registry: er.EntityRegistry,
) -> None:
"""Test setup ending in one device of each type recognized."""
config_entry.add_to_hass(hass)
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
systems = {system.serial: system}
devices = [
get_aqualink_device(
system, name="aux_1", cls=IaquaAuxSwitch, data={"state": "0"}
),
get_aqualink_device(
system, name="freeze_protection", cls=IaquaBinarySensor, data={"state": "0"}
),
get_aqualink_device(
system, name="aux_2", cls=IaquaLightSwitch, data={"state": "0"}
),
get_aqualink_device(system, name="ph", cls=IaquaSensor, data={"state": "7.2"}),
get_aqualink_device(
system, name="pool_set_point", cls=IaquaThermostat, data={"state": "0"}
),
]
devices = {d.name: d for d in devices}
pool_heater = get_aqualink_device(
system, name="pool_heater", cls=IaquaAuxSwitch, data={"state": "0"}
)
pool_temp = get_aqualink_device(
system, name="pool_temp", cls=IaquaSensor, data={"state": "72"}
)
system.devices = {
**{d.name: d for d in devices.values()},
pool_heater.name: pool_heater,
pool_temp.name: pool_temp,
}
system.get_devices = AsyncMock(return_value=devices)
with (
patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
return_value=None,
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
return_value=systems,
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await setup_integration(hass, config_entry, client)
assert config_entry.state is ConfigEntryState.LOADED
@@ -557,32 +518,8 @@ async def test_multiple_updates(
freezer: FrozenDateTimeFactory,
) -> None:
"""Test all possible results of online status transition after update."""
config_entry.add_to_hass(hass)
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
systems = {system.serial: system}
light = get_aqualink_device(
system, name="aux_1", cls=IaquaLightSwitch, data={"state": "1"}
)
devices = {light.name: light}
system.get_devices = AsyncMock(return_value=devices)
with (
patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
return_value=None,
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
return_value=systems,
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
devices = await setup_integration(hass, config_entry, client)
system = devices.system
assert config_entry.state is ConfigEntryState.LOADED
@@ -683,31 +620,7 @@ async def test_entity_assumed_and_available(
freezer: FrozenDateTimeFactory,
) -> None:
"""Test assumed_state and_available properties for all values of online."""
config_entry.add_to_hass(hass)
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
systems = {system.serial: system}
light = get_aqualink_device(
system, name="aux_1", cls=IaquaLightSwitch, data={"state": "1"}
)
devices = {light.name: light}
system.get_devices = AsyncMock(return_value=devices)
system.update = AsyncMock()
with (
patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
return_value=None,
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
return_value=systems,
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
devices = await setup_integration(hass, config_entry, client)
entity_ids = hass.states.async_entity_ids(LIGHT_DOMAIN)
assert len(entity_ids) == 1
@@ -715,19 +628,19 @@ async def test_entity_assumed_and_available(
name = entity_ids[0]
# None means maybe.
light.system.online = None
devices.light.system.online = None
await _advance_coordinator_time(hass, freezer)
state = hass.states.get(name)
assert state.state == STATE_UNAVAILABLE
assert state.attributes.get(ATTR_ASSUMED_STATE) is True
light.system.online = False
devices.light.system.online = False
await _advance_coordinator_time(hass, freezer)
state = hass.states.get(name)
assert state.state == STATE_UNAVAILABLE
assert state.attributes.get(ATTR_ASSUMED_STATE) is True
light.system.online = True
devices.light.system.online = True
await _advance_coordinator_time(hass, freezer)
state = hass.states.get(name)
assert state.state == STATE_ON
@@ -741,34 +654,11 @@ async def test_system_refresh_unauthorized_triggers_reauth(
freezer: FrozenDateTimeFactory,
) -> None:
"""Test an unauthorized refresh starts reauthentication."""
config_entry.add_to_hass(hass)
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
systems = {system.serial: system}
light = get_aqualink_device(
system, name="aux_1", cls=IaquaLightSwitch, data={"state": "1"}
)
system.get_devices = AsyncMock(return_value={light.name: light})
with (
patch(
"homeassistant.components.iaqualink.AqualinkClient.login",
return_value=None,
),
patch(
"homeassistant.components.iaqualink.AqualinkClient.get_systems",
return_value=systems,
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
devices = await setup_integration(hass, config_entry, client)
assert config_entry.state is ConfigEntryState.LOADED
system.update = AsyncMock(side_effect=AqualinkServiceUnauthorizedException)
devices.system.update = AsyncMock(side_effect=AqualinkServiceUnauthorizedException)
await _advance_coordinator_time(hass, freezer)
tests/components/iaqualink/test_light.py
+281
View File
@@ -0,0 +1,281 @@
"""Light platform tests for iAquaLink."""
from __future__ import annotations
from unittest.mock import AsyncMock
import httpx
from iaqualink.client import AqualinkClient
from iaqualink.exception import (
AqualinkServiceException,
AqualinkServiceUnauthorizedException,
)
from iaqualink.systems.iaqua.device import (
IaquaColorLightJC,
IaquaDimmableLight,
IaquaLightSwitch,
)
from iaqualink.systems.iaqua.system import IaquaSystem
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_EFFECT,
DOMAIN as LIGHT_DOMAIN,
ColorMode,
LightEntityFeature,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers import entity_registry as er
from .conftest import (
assert_platform_setup,
get_aqualink_device,
get_aqualink_system,
setup_entry,
)
from tests.common import MockConfigEntry
async def test_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all light entities are created correctly."""
await assert_platform_setup(
hass, config_entry, client, entity_registry, snapshot, LIGHT_DOMAIN
)
async def _setup_light(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
cls: type,
data: dict[str, str],
) -> tuple[IaquaSystem, object, str, object]:
"""Set up the integration with a single light entity."""
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
light = get_aqualink_device(system, name="aux_1", cls=cls, data=data)
system.get_devices = AsyncMock(return_value={light.name: light})
system.set_aux = AsyncMock()
system.set_light = AsyncMock()
await setup_entry(hass, config_entry, system)
entity_ids = hass.states.async_entity_ids(LIGHT_DOMAIN)
assert len(entity_ids) == 1
entity_id = entity_ids[0]
entity_state = hass.states.get(entity_id)
assert entity_state is not None
return system, light, entity_id, entity_state
async def test_effect_light_setup_exposes_effect_features(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test color lights expose effect-related state through Home Assistant."""
_, _, _, entity_state = await _setup_light(
hass,
config_entry,
client,
IaquaColorLightJC,
{"state": "1", "aux": "1", "subtype": "1", "label": "pool light"},
)
assert entity_state.state == STATE_ON
assert entity_state.attributes["color_mode"] == ColorMode.ONOFF.value
assert entity_state.attributes["supported_features"] == LightEntityFeature.EFFECT
assert "Alpine White" in entity_state.attributes["effect_list"]
async def test_dimmable_light_setup_exposes_brightness(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test dimmable lights expose brightness through Home Assistant."""
_, _, _, entity_state = await _setup_light(
hass,
config_entry,
client,
IaquaDimmableLight,
{"state": "1", "aux": "1", "subtype": "50", "label": "pool light"},
)
assert entity_state.state == STATE_ON
assert entity_state.attributes["color_mode"] == ColorMode.BRIGHTNESS.value
assert entity_state.attributes["brightness"] == 128
async def test_light_turn_on_with_effect_updates_state(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test turning on a color light with an effect uses the effect branch."""
system, light, entity_id, _ = await _setup_light(
hass,
config_entry,
client,
IaquaColorLightJC,
{"state": "0", "aux": "1", "subtype": "1", "label": "pool light"},
)
light_component = hass.data[LIGHT_DOMAIN]
entity = light_component.get_entity(entity_id)
assert entity is not None
async def set_light(data: dict[str, str]) -> None:
light.data["state"] = data["light"]
system.set_light = AsyncMock(side_effect=set_light)
await entity.async_turn_on(**{ATTR_EFFECT: "Alpine White"})
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == STATE_ON
async def test_light_turn_on_with_brightness_updates_state(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test turning on a dimmable light updates Home Assistant state."""
system, light, entity_id, _ = await _setup_light(
hass,
config_entry,
client,
IaquaDimmableLight,
{"state": "0", "aux": "1", "subtype": "0", "label": "pool light"},
)
async def set_light(data: dict[str, str]) -> None:
light.data["state"] = "1" if data["light"] != "0" else "0"
light.data["subtype"] = data["light"]
system.set_light = AsyncMock(side_effect=set_light)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 128},
blocking=True,
)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == STATE_ON
assert entity_state.attributes["brightness"] == 128
async def test_light_turn_on_without_attributes_updates_state(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
) -> None:
"""Test turning on a basic light uses the default turn_on path."""
system, light, entity_id, _ = await _setup_light(
hass,
config_entry,
client,
IaquaLightSwitch,
{"state": "0", "aux": "1", "label": "pool light"},
)
async def set_aux(_: str) -> None:
light.data["state"] = "1"
system.set_aux = AsyncMock(side_effect=set_aux)
light_component = hass.data[LIGHT_DOMAIN]
entity = light_component.get_entity(entity_id)
assert entity is not None
await entity.async_turn_on()
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == STATE_ON
@pytest.mark.parametrize(
("raised_exception", "expected_exception", "match"),
[
pytest.param(
AqualinkServiceException,
HomeAssistantError,
"Aqualink error: AqualinkServiceException",
id="service",
),
pytest.param(
TimeoutError(),
HomeAssistantError,
"Aqualink error: TimeoutError",
id="timeout",
),
pytest.param(
httpx.HTTPError("boom"),
HomeAssistantError,
"Aqualink error: boom",
id="http",
),
pytest.param(
AqualinkServiceUnauthorizedException,
ConfigEntryAuthFailed,
"Invalid credentials for iAquaLink",
id="unauthorized",
),
pytest.param(
Exception("Test exception"),
Exception,
"Test exception",
id="unexpected",
),
],
)
async def test_light_turn_off_errors_leave_state_unchanged(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
raised_exception: Exception | type[Exception],
expected_exception: type[Exception],
match: str,
) -> None:
"""Test turn-off errors are surfaced through the light service call."""
system, _, entity_id, _ = await _setup_light(
hass,
config_entry,
client,
IaquaLightSwitch,
{"state": "1", "aux": "1", "label": "pool light"},
)
system.set_aux = AsyncMock(side_effect=raised_exception)
with pytest.raises(expected_exception, match=match):
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == STATE_ON
tests/components/iaqualink/test_sensor.py
+148
View File
@@ -0,0 +1,148 @@
"""Sensor platform tests for iAquaLink."""
from __future__ import annotations
from unittest.mock import AsyncMock
from iaqualink.client import AqualinkClient
from iaqualink.systems.iaqua.device import IaquaSensor
from iaqualink.systems.iaqua.system import IaquaSystem
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN, SensorDeviceClass
from homeassistant.const import STATE_UNKNOWN, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util.unit_system import US_CUSTOMARY_SYSTEM
from .conftest import (
assert_platform_setup,
get_aqualink_device,
get_aqualink_system,
setup_entry,
)
from tests.common import MockConfigEntry
async def test_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all sensor entities are created correctly."""
await assert_platform_setup(
hass, config_entry, client, entity_registry, snapshot, SENSOR_DOMAIN
)
async def _setup_sensor(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
*,
name: str,
state: str,
temp_unit: str = "F",
) -> tuple[str, object]:
"""Set up the integration with a single sensor entity."""
system = get_aqualink_system(
client,
cls=IaquaSystem,
data={"home_screen": [{}, {}, {}, {"temp_scale": temp_unit}]},
)
system.online = True
async def update() -> None:
system.temp_unit = temp_unit
system.update = AsyncMock(side_effect=update)
sensor = get_aqualink_device(
system, name=name, cls=IaquaSensor, data={"state": state}
)
system.get_devices = AsyncMock(return_value={sensor.name: sensor})
await setup_entry(hass, config_entry, system)
entity_ids = hass.states.async_entity_ids(SENSOR_DOMAIN)
assert len(entity_ids) == 1
entity_id = entity_ids[0]
entity_state = hass.states.get(entity_id)
assert entity_state is not None
return entity_id, entity_state
@pytest.mark.parametrize(
("name", "temp_unit", "expected_class", "expected_unit"),
[
pytest.param(
"pool_temp",
"F",
SensorDeviceClass.TEMPERATURE,
UnitOfTemperature.FAHRENHEIT,
id="fahrenheit-temperature",
),
pytest.param(
"spa_temp",
"C",
SensorDeviceClass.TEMPERATURE,
UnitOfTemperature.CELSIUS,
id="celsius-temperature",
),
pytest.param("ph", "F", None, None, id="non-temperature"),
],
)
async def test_sensor_initialization(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
name: str,
temp_unit: str,
expected_class: SensorDeviceClass | None,
expected_unit: UnitOfTemperature | None,
) -> None:
"""Test sensor initialization for temperature and generic sensors."""
if expected_unit is UnitOfTemperature.FAHRENHEIT:
hass.config.units = US_CUSTOMARY_SYSTEM
_, entity_state = await _setup_sensor(
hass,
config_entry,
client,
name=name,
temp_unit=temp_unit,
state="72",
)
assert entity_state.attributes.get("device_class") == expected_class
assert entity_state.attributes.get("unit_of_measurement") == expected_unit
@pytest.mark.parametrize(
("state", "expected_state"),
[
pytest.param("", STATE_UNKNOWN, id="empty"),
pytest.param("72", "72", id="int"),
pytest.param("7.2", "7.2", id="float"),
],
)
async def test_sensor_native_value(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
state: str,
expected_state: str,
) -> None:
"""Test sensor state parsing."""
_, entity_state = await _setup_sensor(
hass,
config_entry,
client,
name="ph",
state=state,
)
assert entity_state.state == expected_state
tests/components/iaqualink/test_switch.py
+214
View File
@@ -0,0 +1,214 @@
"""Switch platform tests for iAquaLink."""
from __future__ import annotations
from unittest.mock import AsyncMock
import httpx
from iaqualink.client import AqualinkClient
from iaqualink.exception import (
AqualinkServiceException,
AqualinkServiceUnauthorizedException,
)
from iaqualink.systems.iaqua.device import IaquaAuxSwitch
from iaqualink.systems.iaqua.system import IaquaSystem
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers import entity_registry as er
from .conftest import (
assert_platform_setup,
get_aqualink_device,
get_aqualink_system,
setup_entry,
)
from tests.common import MockConfigEntry
async def test_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all switch entities are created correctly."""
await assert_platform_setup(
hass, config_entry, client, entity_registry, snapshot, SWITCH_DOMAIN
)
async def _setup_switch(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
*,
label: str,
state: str,
) -> tuple[IaquaSystem, object, str, object]:
"""Set up the integration with a single switch entity."""
system = get_aqualink_system(client, cls=IaquaSystem)
system.online = True
system.update = AsyncMock()
switch = get_aqualink_device(
system,
name=label.lower().replace(" ", "_"),
cls=IaquaAuxSwitch,
data={"state": state, "aux": "1", "label": label},
)
system.get_devices = AsyncMock(return_value={switch.name: switch})
system.set_aux = AsyncMock()
await setup_entry(hass, config_entry, system)
entity_ids = hass.states.async_entity_ids(SWITCH_DOMAIN)
assert len(entity_ids) == 1
entity_id = entity_ids[0]
entity_state = hass.states.get(entity_id)
assert entity_state is not None
return system, switch, entity_id, entity_state
@pytest.mark.parametrize(
("label", "expected_icon"),
[
pytest.param("Cleaner", "mdi:robot-vacuum", id="cleaner"),
pytest.param("Waterfall", "mdi:fountain", id="waterfall"),
pytest.param("Spa Dscnt", "mdi:fountain", id="descent"),
pytest.param("Filter Pump", "mdi:fan", id="pump"),
pytest.param("Spa Blower", "mdi:fan", id="blower"),
pytest.param("Pool Heater", "mdi:radiator", id="heater"),
pytest.param("Auxiliary", None, id="default"),
],
)
async def test_switch_icons(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
label: str,
expected_icon: str | None,
) -> None:
"""Test switch icons are derived from the device label."""
_, _, _, entity_state = await _setup_switch(
hass,
config_entry,
client,
label=label,
state="1",
)
assert entity_state.attributes.get("icon") == expected_icon
assert entity_state.state == STATE_ON
@pytest.mark.parametrize(
("service", "initial_state", "expected_state"),
[
pytest.param(SERVICE_TURN_ON, "0", STATE_ON, id="turn-on"),
pytest.param(SERVICE_TURN_OFF, "1", STATE_OFF, id="turn-off"),
],
)
async def test_switch_actions(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
service: str,
initial_state: str,
expected_state: str,
) -> None:
"""Test switch services update Home Assistant state."""
system, switch, entity_id, _ = await _setup_switch(
hass,
config_entry,
client,
label="Auxiliary",
state=initial_state,
)
async def set_aux(_: str) -> None:
switch.data["state"] = "1" if service == SERVICE_TURN_ON else "0"
system.set_aux = AsyncMock(side_effect=set_aux)
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == expected_state
@pytest.mark.parametrize(
("raised_exception", "expected_exception", "match"),
[
pytest.param(
AqualinkServiceException,
HomeAssistantError,
"Aqualink error: AqualinkServiceException",
id="service",
),
pytest.param(
TimeoutError(),
HomeAssistantError,
"Aqualink error: TimeoutError",
id="timeout",
),
pytest.param(
httpx.HTTPError("boom"),
HomeAssistantError,
"Aqualink error: boom",
id="http",
),
pytest.param(
AqualinkServiceUnauthorizedException,
ConfigEntryAuthFailed,
"Invalid credentials for iAquaLink",
id="unauthorized",
),
],
)
async def test_switch_turn_off_errors_leave_state_unchanged(
hass: HomeAssistant,
config_entry: MockConfigEntry,
client: AqualinkClient,
raised_exception: Exception | type[Exception],
expected_exception: type[Exception],
match: str,
) -> None:
"""Test turn-off errors are surfaced through the switch service call."""
system, _, entity_id, _ = await _setup_switch(
hass,
config_entry,
client,
label="Auxiliary",
state="1",
)
system.set_aux = AsyncMock(side_effect=raised_exception)
with pytest.raises(expected_exception, match=match):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
assert entity_state.state == STATE_ON
tests/components/iaqualink/test_utils.py
-24
View File
@@ -1,24 +0,0 @@
"""Tests for iAquaLink integration utility functions."""
from iaqualink.exception import AqualinkServiceException
import pytest
from homeassistant.components.iaqualink.utils import await_or_reraise
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .conftest import async_raises, async_returns
async def test_await_or_reraise(hass: HomeAssistant) -> None:
"""Test await_or_reraise for all values of awaitable."""
async_noop = async_returns(None)
await await_or_reraise(async_noop())
with pytest.raises(Exception) as exc_info:
await await_or_reraise(async_raises(Exception("Test exception"))())
assert str(exc_info.value) == "Test exception"
async_ex = async_raises(AqualinkServiceException)
with pytest.raises(HomeAssistantError):
await await_or_reraise(async_ex())