Compare commits

..

1 Commits

Author SHA1 Message Date
mib1185
c019331de1 make trigger_behavior selector translations common 2025-12-11 17:36:54 +00:00
44 changed files with 341 additions and 713 deletions

View File

@@ -78,9 +78,9 @@
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -23,9 +23,9 @@
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -7,5 +7,5 @@
"integration_type": "hub",
"iot_class": "local_polling",
"loggers": ["aioasuswrt", "asusrouter", "asyncssh"],
"requirements": ["aioasuswrt==1.5.2", "asusrouter==1.21.3"]
"requirements": ["aioasuswrt==1.5.1", "asusrouter==1.21.3"]
}

View File

@@ -125,7 +125,6 @@ _EXPERIMENTAL_TRIGGER_PLATFORMS = {
"alarm_control_panel",
"assist_satellite",
"binary_sensor",
"button",
"climate",
"cover",
"fan",

View File

@@ -324,9 +324,9 @@
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -7,7 +7,7 @@
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["bsblan"],
"requirements": ["python-bsblan==3.1.4"],
"requirements": ["python-bsblan==3.1.3"],
"zeroconf": [
{
"name": "bsb-lan*",

View File

@@ -17,10 +17,5 @@
"press": {
"service": "mdi:gesture-tap-button"
}
},
"triggers": {
"pressed": {
"trigger": "mdi:gesture-tap-button"
}
}
}

View File

@@ -27,11 +27,5 @@
"name": "Press"
}
},
"title": "Button",
"triggers": {
"pressed": {
"description": "Triggers after one or several buttons were pressed.",
"name": "Button pressed"
}
}
"title": "Button"
}

View File

@@ -1,42 +0,0 @@
"""Provides triggers for buttons."""
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.trigger import (
ENTITY_STATE_TRIGGER_SCHEMA,
EntityTriggerBase,
Trigger,
)
from . import DOMAIN
class ButtonPressedTrigger(EntityTriggerBase):
"""Trigger for button entity presses."""
_domain = DOMAIN
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state is not an expected target states."""
# UNKNOWN is a valid from_state, otherwise the first time the button is pressed
# would not trigger
if from_state.state == STATE_UNAVAILABLE:
return False
return from_state.state != to_state.state
def is_valid_state(self, state: State) -> bool:
"""Check if the new state is not invalid."""
return state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)
TRIGGERS: dict[str, type[Trigger]] = {
"pressed": ButtonPressedTrigger,
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for buttons."""
return TRIGGERS

View File

@@ -1,4 +0,0 @@
pressed:
target:
entity:
domain: button

View File

@@ -194,9 +194,9 @@
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -1,8 +1,8 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
"already_configured": "This device is already configured.",
"reauth_successful": "Reauthentication successful."
},
"create_entry": {
"add_sensor_mapping_hint": "You can now add mappings from any sensor in Home Assistant to {integration_name} using the '+ add sensor mapping' button."

View File

@@ -73,9 +73,9 @@
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -10,7 +10,7 @@ from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
from . import oauth2
from .const import DOMAIN
from .coordinator import HomeLinkConfigEntry, HomeLinkCoordinator
from .coordinator import HomeLinkConfigEntry, HomeLinkCoordinator, HomeLinkData
PLATFORMS: list[Platform] = [Platform.EVENT]
@@ -44,7 +44,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) ->
)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
entry.runtime_data = HomeLinkData(
provider=provider, coordinator=coordinator, last_update_id=None
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
@@ -52,5 +54,5 @@ async def async_setup_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) ->
async def async_unload_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) -> bool:
"""Unload a config entry."""
await entry.runtime_data.async_on_unload(None)
await entry.runtime_data.coordinator.async_on_unload(None)
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -3,6 +3,7 @@
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from functools import partial
import logging
from typing import TypedDict
@@ -16,10 +17,19 @@ from homeassistant.util.ssl import get_default_context
_LOGGER = logging.getLogger(__name__)
type HomeLinkConfigEntry = ConfigEntry[HomeLinkCoordinator]
type HomeLinkConfigEntry = ConfigEntry[HomeLinkData]
type EventCallback = Callable[[HomeLinkEventData], None]
@dataclass
class HomeLinkData:
"""Class for HomeLink integration runtime data."""
provider: MQTTProvider
coordinator: HomeLinkCoordinator
last_update_id: str | None
class HomeLinkEventData(TypedDict):
"""Data for a single event."""
@@ -58,7 +68,7 @@ class HomeLinkCoordinator:
self._listeners[target_event_id] = update_callback
return partial(self.__async_remove_listener_internal, target_event_id)
def __async_remove_listener_internal(self, listener_id: str) -> None:
def __async_remove_listener_internal(self, listener_id: str):
del self._listeners[listener_id]
@callback
@@ -82,7 +92,7 @@ class HomeLinkCoordinator:
await self.discover_devices()
self.provider.listen(self.on_message)
async def discover_devices(self) -> None:
async def discover_devices(self):
"""Discover devices and build the Entities."""
self.device_data = await self.provider.discover()

View File

@@ -3,24 +3,25 @@
from __future__ import annotations
from homeassistant.components.event import EventDeviceClass, EventEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN, EVENT_PRESSED
from .coordinator import HomeLinkConfigEntry, HomeLinkCoordinator, HomeLinkEventData
from .coordinator import HomeLinkCoordinator, HomeLinkEventData
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeLinkConfigEntry,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Add the entities for the event platform."""
coordinator = config_entry.runtime_data
"""Add the entities for the binary sensor."""
coordinator = config_entry.runtime_data.coordinator
async_add_entities(
HomeLinkEventEntity(coordinator, button.id, button.name, device.id, device.name)
HomeLinkEventEntity(button.id, button.name, device.id, device.name, coordinator)
for device in coordinator.device_data
for button in device.buttons
)
@@ -39,11 +40,11 @@ class HomeLinkEventEntity(EventEntity):
def __init__(
self,
coordinator: HomeLinkCoordinator,
button_id: str,
param_name: str,
device_id: str,
device_name: str,
coordinator: HomeLinkCoordinator,
) -> None:
"""Initialize the event entity."""
@@ -73,4 +74,5 @@ class HomeLinkEventEntity(EventEntity):
if update_data["requestId"] != self.last_request_id:
self._trigger_event(EVENT_PRESSED)
self.last_request_id = update_data["requestId"]
self.async_write_ha_state()
self.async_write_ha_state()

View File

@@ -18,9 +18,9 @@
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -330,9 +330,9 @@
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -184,9 +184,9 @@
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -56,5 +56,4 @@ WIDGET_TO_WATER_HEATER_ENTITY = {
CONTROLLABLE_NAME_TO_WATER_HEATER_ENTITY = {
"modbuslink:AtlanticDomesticHotWaterProductionMBLComponent": AtlanticDomesticHotWaterProductionMBLComponent,
"io:AtlanticDomesticHotWaterProductionV2_CV4E_IOComponent": AtlanticDomesticHotWaterProductionV2IOComponent,
"io:AtlanticDomesticHotWaterProductionV2_CETHI_V4_IOComponent": AtlanticDomesticHotWaterProductionV2IOComponent,
}

View File

@@ -2,7 +2,7 @@
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
"reauth_successful": "Re-authentication was successful"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",

View File

@@ -6,5 +6,5 @@
"documentation": "https://www.home-assistant.io/integrations/smarttub",
"iot_class": "cloud_polling",
"loggers": ["smarttub"],
"requirements": ["python-smarttub==0.0.46"]
"requirements": ["python-smarttub==0.0.45"]
}

View File

@@ -17,6 +17,10 @@ class TextChangedTrigger(EntityTriggerBase):
_domain = DOMAIN
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the old and new states are different."""
return from_state.state != to_state.state
def is_valid_state(self, state: State) -> bool:
"""Check if the new state is not invalid."""
return state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)

View File

@@ -380,7 +380,7 @@ class _CustomDPCodeWrapper(DPCodeWrapper):
def read_device_status(self, device: CustomerDevice) -> bool | None:
"""Read the device value for the dpcode."""
if (raw_value := device.status.get(self.dpcode)) is None:
if (raw_value := self._read_device_status_raw(device)) is None:
return None
return raw_value in self._valid_values

View File

@@ -33,7 +33,7 @@ class _DPCodePercentageMappingWrapper(DPCodeIntegerWrapper):
return False
def read_device_status(self, device: CustomerDevice) -> float | None:
if (value := device.status.get(self.dpcode)) is None:
if (value := self._read_device_status_raw(device)) is None:
return None
return round(

View File

@@ -77,7 +77,7 @@ class _AlarmMessageWrapper(DPCodeStringWrapper, _DPCodeEventWrapper):
def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
"""Return the event attributes for the alarm message."""
if (raw_value := device.status.get(self.dpcode)) is None:
if (raw_value := self._read_device_status_raw(device)) is None:
return None
return {"message": b64decode(raw_value).decode("utf-8")}

View File

@@ -53,7 +53,7 @@ class _BrightnessWrapper(DPCodeIntegerWrapper):
def read_device_status(self, device: CustomerDevice) -> Any | None:
"""Return the brightness of this light between 0..255."""
if (brightness := device.status.get(self.dpcode)) is None:
if (brightness := self._read_device_status_raw(device)) is None:
return None
# Remap value to our scale
@@ -114,7 +114,7 @@ class _ColorTempWrapper(DPCodeIntegerWrapper):
def read_device_status(self, device: CustomerDevice) -> Any | None:
"""Return the color temperature value in Kelvin."""
if (temperature := device.status.get(self.dpcode)) is None:
if (temperature := self._read_device_status_raw(device)) is None:
return None
return color_util.color_temperature_mired_to_kelvin(

View File

@@ -46,6 +46,13 @@ class DPCodeWrapper(DeviceWrapper):
"""Init DPCodeWrapper."""
self.dpcode = dpcode
def _read_device_status_raw(self, device: CustomerDevice) -> Any | None:
"""Read the raw device status for the DPCode.
Private helper method for `read_device_status`.
"""
return device.status.get(self.dpcode)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value.
@@ -83,7 +90,7 @@ class DPCodeTypeInformationWrapper[T: TypeInformation](DPCodeWrapper):
def read_device_status(self, device: CustomerDevice) -> Any | None:
"""Read the device value for the dpcode."""
return self.type_information.process_raw_value(
device.status.get(self.dpcode), device
self._read_device_status_raw(device), device
)
@classmethod
@@ -190,7 +197,7 @@ class DPCodeBitmapBitWrapper(DPCodeWrapper):
def read_device_status(self, device: CustomerDevice) -> bool | None:
"""Read the device value for the dpcode."""
if (raw_value := device.status.get(self.dpcode)) is None:
if (raw_value := self._read_device_status_raw(device)) is None:
return None
return (raw_value & (1 << self._mask)) != 0

View File

@@ -76,7 +76,9 @@ class _WindDirectionWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]):
def read_device_status(self, device: CustomerDevice) -> float | None:
"""Read the device value for the dpcode."""
if (raw_value := device.status.get(self.dpcode)) in self.type_information.range:
if (
raw_value := self._read_device_status_raw(device)
) in self.type_information.range:
return self._WIND_DIRECTIONS.get(raw_value)
return None

View File

@@ -38,9 +38,9 @@
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
"any": "[%key:common::selector::trigger_behavior::options::any%]",
"first": "[%key:common::selector::trigger_behavior::options::first%]",
"last": "[%key:common::selector::trigger_behavior::options::last%]"
}
}
},

View File

@@ -338,11 +338,8 @@ class EntityTriggerBase(Trigger):
self._target = config.target
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state is valid and the state has changed."""
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return False
return from_state.state != to_state.state
"""Check if the origin state is not an expected target states."""
return not self.is_valid_state(from_state)
@abc.abstractmethod
def is_valid_state(self, state: State) -> bool:
@@ -393,11 +390,12 @@ class EntityTriggerBase(Trigger):
from_state = event.data["old_state"]
to_state = event.data["new_state"]
if not from_state or not to_state:
# The trigger should never fire if the previous state was not a valid state
if not from_state or from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return
# The trigger should never fire if the new state is not valid
if not self.is_valid_state(to_state):
if not to_state or not self.is_valid_state(to_state):
return
# The trigger should never fire if the transition is not valid
@@ -448,9 +446,6 @@ class ConditionalEntityStateTriggerBase(EntityTriggerBase):
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state matches the expected ones."""
if not super().is_valid_transition(from_state, to_state):
return False
return from_state.state in self._from_states
def is_valid_state(self, state: State) -> bool:
@@ -464,15 +459,6 @@ class EntityStateAttributeTriggerBase(EntityTriggerBase):
_attribute: str
_attribute_to_state: str
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state is valid and the state has changed."""
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return False
return from_state.attributes.get(self._attribute) != to_state.attributes.get(
self._attribute
)
def is_valid_state(self, state: State) -> bool:
"""Check if the new state attribute matches the expected one."""
return state.attributes.get(self._attribute) == self._attribute_to_state

View File

@@ -223,6 +223,3 @@ gql<4.0.0
# Pin pytest-rerunfailures to prevent accidental breaks
pytest-rerunfailures==16.0.1
# pycares 5.x is not yet compatible with aiodns
pycares==4.11.0

View File

@@ -124,6 +124,15 @@
"model": "Model",
"ui_managed": "Managed via UI"
},
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"state": {
"active": "Active",
"auto": "Auto",

6
requirements_all.txt generated
View File

@@ -206,7 +206,7 @@ aioaquacell==0.2.0
aioaseko==1.0.0
# homeassistant.components.asuswrt
aioasuswrt==1.5.2
aioasuswrt==1.5.1
# homeassistant.components.husqvarna_automower
aioautomower==2.7.1
@@ -2475,7 +2475,7 @@ python-awair==0.2.5
python-blockchain-api==0.0.2
# homeassistant.components.bsblan
python-bsblan==3.1.4
python-bsblan==3.1.3
# homeassistant.components.citybikes
python-citybikes==0.3.3
@@ -2578,7 +2578,7 @@ python-ripple-api==0.0.3
python-roborock==3.12.2
# homeassistant.components.smarttub
python-smarttub==0.0.46
python-smarttub==0.0.45
# homeassistant.components.snoo
python-snoo==0.8.3

View File

@@ -197,7 +197,7 @@ aioaquacell==0.2.0
aioaseko==1.0.0
# homeassistant.components.asuswrt
aioasuswrt==1.5.2
aioasuswrt==1.5.1
# homeassistant.components.husqvarna_automower
aioautomower==2.7.1
@@ -2086,7 +2086,7 @@ python-MotionMount==2.3.0
python-awair==0.2.5
# homeassistant.components.bsblan
python-bsblan==3.1.4
python-bsblan==3.1.3
# homeassistant.components.ecobee
python-ecobee-api==0.3.2
@@ -2159,7 +2159,7 @@ python-rabbitair==0.0.8
python-roborock==3.12.2
# homeassistant.components.smarttub
python-smarttub==0.0.46
python-smarttub==0.0.45
# homeassistant.components.snoo
python-snoo==0.8.3

View File

@@ -214,9 +214,6 @@ gql<4.0.0
# Pin pytest-rerunfailures to prevent accidental breaks
pytest-rerunfailures==16.0.1
# pycares 5.x is not yet compatible with aiodns
pycares==4.11.0
"""
GENERATED_MESSAGE = (

View File

@@ -1,192 +0,0 @@
"""Test button trigger."""
from collections.abc import Generator
from unittest.mock import patch
import pytest
from homeassistant.const import (
ATTR_LABEL_ID,
CONF_ENTITY_ID,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, ServiceCall
from tests.components import (
StateDescription,
arm_trigger,
parametrize_target_entities,
set_or_remove_state,
target_entities,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture(name="enable_experimental_triggers_conditions")
def enable_experimental_triggers_conditions() -> Generator[None]:
"""Enable experimental triggers and conditions."""
with patch(
"homeassistant.components.labs.async_is_preview_feature_enabled",
return_value=True,
):
yield
@pytest.fixture
async def target_buttons(hass: HomeAssistant) -> list[str]:
"""Create multiple button entities associated with different targets."""
return (await target_entities(hass, "button"))["included"]
@pytest.mark.parametrize("trigger_key", ["button.pressed"])
async def test_button_triggers_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, trigger_key: str
) -> None:
"""Test the button triggers are gated by the labs flag."""
await arm_trigger(hass, trigger_key, None, {ATTR_LABEL_ID: "test_label"})
assert (
"Unnamed automation failed to setup triggers and has been disabled: Trigger "
f"'{trigger_key}' requires the experimental 'New triggers and conditions' "
"feature to be enabled in Home Assistant Labs settings (feature flag: "
"'new_triggers_conditions')"
) in caplog.text
@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("button"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
(
"button.pressed",
[
{"included": {"state": None, "attributes": {}}, "count": 0},
{
"included": {
"state": "2021-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 0,
},
{
"included": {
"state": "2022-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
],
),
(
"button.pressed",
[
{"included": {"state": "foo", "attributes": {}}, "count": 0},
{
"included": {
"state": "2021-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
{
"included": {
"state": "2022-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
],
),
(
"button.pressed",
[
{
"included": {"state": STATE_UNAVAILABLE, "attributes": {}},
"count": 0,
},
{
"included": {
"state": "2021-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 0,
},
{
"included": {
"state": "2022-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
{
"included": {"state": STATE_UNAVAILABLE, "attributes": {}},
"count": 0,
},
],
),
(
"button.pressed",
[
{"included": {"state": STATE_UNKNOWN, "attributes": {}}, "count": 0},
{
"included": {
"state": "2021-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
{
"included": {
"state": "2022-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
{"included": {"state": STATE_UNKNOWN, "attributes": {}}, "count": 0},
],
),
],
)
async def test_button_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_buttons: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the button state trigger fires when any button state changes to a specific state."""
other_entity_ids = set(target_buttons) - {entity_id}
# Set all buttons, including the tested button, to the initial state
for eid in target_buttons:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, None, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other buttons also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()

View File

@@ -1,32 +1 @@
"""Tests for the homelink integration."""
from typing import Any
from unittest.mock import AsyncMock
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, entry: MockConfigEntry) -> None:
"""Set up the homelink integration for testing."""
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
async def update_callback(
hass: HomeAssistant, mock: AsyncMock, update_type: str, data: dict[str, Any]
) -> None:
"""Invoke the MQTT provider's message callback with the specified update type and data."""
for call in mock.listen.call_args_list:
call[0][0](
"topic",
{
"type": update_type,
"data": data,
},
)
await hass.async_block_till_done()
await hass.async_block_till_done()

View File

@@ -3,14 +3,8 @@
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from homelink.model.button import Button
import homelink.model.device
import pytest
from homeassistant.components.gentex_homelink import DOMAIN
from tests.common import MockConfigEntry
@pytest.fixture
def mock_srp_auth() -> Generator[AsyncMock]:
@@ -30,50 +24,6 @@ def mock_srp_auth() -> Generator[AsyncMock]:
yield instance
@pytest.fixture
def mock_mqtt_provider(mock_device: AsyncMock) -> Generator[AsyncMock]:
"""Mock MQTT provider."""
with patch(
"homeassistant.components.gentex_homelink.MQTTProvider", autospec=True
) as mock_mqtt_provider:
instance = mock_mqtt_provider.return_value
instance.discover.return_value = [mock_device]
yield instance
@pytest.fixture
def mock_device() -> AsyncMock:
"""Mock Device instance."""
device = AsyncMock(spec=homelink.model.device.Device, autospec=True)
buttons = [
Button(id="1", name="Button 1", device=device),
Button(id="2", name="Button 2", device=device),
Button(id="3", name="Button 3", device=device),
]
device.id = "TestDevice"
device.name = "TestDevice"
device.buttons = buttons
return device
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock setup entry."""
return MockConfigEntry(
domain=DOMAIN,
data={
"auth_implementation": "gentex_homelink",
"token": {
"access_token": "access",
"refresh_token": "refresh",
"expires_in": 3600,
"token_type": "bearer",
"expires_at": 1234567890,
},
},
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Mock setup entry."""

View File

@@ -1,172 +0,0 @@
# serializer version: 1
# name: test_entities[event.testdevice_button_1-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'Pressed',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.testdevice_button_1',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.BUTTON: 'button'>,
'original_icon': None,
'original_name': 'Button 1',
'platform': 'gentex_homelink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '1',
'unit_of_measurement': None,
})
# ---
# name: test_entities[event.testdevice_button_1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'button',
'event_type': None,
'event_types': list([
'Pressed',
]),
'friendly_name': 'TestDevice Button 1',
}),
'context': <ANY>,
'entity_id': 'event.testdevice_button_1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_entities[event.testdevice_button_2-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'Pressed',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.testdevice_button_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.BUTTON: 'button'>,
'original_icon': None,
'original_name': 'Button 2',
'platform': 'gentex_homelink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '2',
'unit_of_measurement': None,
})
# ---
# name: test_entities[event.testdevice_button_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'button',
'event_type': None,
'event_types': list([
'Pressed',
]),
'friendly_name': 'TestDevice Button 2',
}),
'context': <ANY>,
'entity_id': 'event.testdevice_button_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_entities[event.testdevice_button_3-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'Pressed',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.testdevice_button_3',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.BUTTON: 'button'>,
'original_icon': None,
'original_name': 'Button 3',
'platform': 'gentex_homelink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '3',
'unit_of_measurement': None,
})
# ---
# name: test_entities[event.testdevice_button_3-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'button',
'event_type': None,
'event_types': list([
'Pressed',
]),
'friendly_name': 'TestDevice Button 3',
}),
'context': <ANY>,
'entity_id': 'event.testdevice_button_3',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@@ -1,32 +0,0 @@
# serializer version: 1
# name: test_device
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'gentex_homelink',
'TestDevice',
),
}),
'labels': set({
}),
'manufacturer': None,
'model': None,
'model_id': None,
'name': 'TestDevice',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'sw_version': None,
'via_device_id': None,
})
# ---

View File

@@ -0,0 +1,160 @@
"""Tests for the homelink coordinator."""
import asyncio
import time
from unittest.mock import patch
from homelink.model.button import Button
from homelink.model.device import Device
import pytest
from homeassistant.components.gentex_homelink import async_setup_entry
from homeassistant.components.gentex_homelink.const import EVENT_PRESSED
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
import homeassistant.helpers.entity_registry as er
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
DOMAIN = "gentex_homelink"
deviceInst = Device(id="TestDevice", name="TestDevice")
deviceInst.buttons = [
Button(id="Button 1", name="Button 1", device=deviceInst),
Button(id="Button 2", name="Button 2", device=deviceInst),
Button(id="Button 3", name="Button 3", device=deviceInst),
]
async def test_get_state_updates(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test state updates.
Tests that get_state calls are called by home assistant, and the homeassistant components respond appropriately to the data returned.
"""
with patch(
"homeassistant.components.gentex_homelink.MQTTProvider", autospec=True
) as MockProvider:
instance = MockProvider.return_value
instance.discover.return_value = [deviceInst]
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=None,
version=1,
data={
"auth_implementation": "gentex_homelink",
"token": {"expires_at": time.time() + 10000, "access_token": ""},
"last_update_id": None,
},
state=ConfigEntryState.LOADED,
)
config_entry.add_to_hass(hass)
result = await async_setup_entry(hass, config_entry)
# Assert configuration worked without errors
assert result
provider = config_entry.runtime_data.provider
state_data = {
"type": "state",
"data": {
"Button 1": {"requestId": "rid1", "timestamp": time.time()},
"Button 2": {"requestId": "rid2", "timestamp": time.time()},
"Button 3": {"requestId": "rid3", "timestamp": time.time()},
},
}
# Test successful setup and first data fetch. The buttons should be unknown at the start
await hass.async_block_till_done(wait_background_tasks=True)
states = hass.states.async_all()
assert states, "No states were loaded"
assert all(state != STATE_UNAVAILABLE for state in states), (
"At least one state was not initialized as STATE_UNAVAILABLE"
)
buttons_unknown = [s.state == "unknown" for s in states]
assert buttons_unknown and all(buttons_unknown), (
"At least one button state was not initialized to unknown"
)
provider.listen.mock_calls[0].args[0](None, state_data)
await hass.async_block_till_done(wait_background_tasks=True)
await asyncio.gather(*asyncio.all_tasks() - {asyncio.current_task()})
await asyncio.sleep(0.01)
states = hass.states.async_all()
assert all(state != STATE_UNAVAILABLE for state in states), (
"Some button became unavailable"
)
buttons_pressed = [s.attributes["event_type"] == EVENT_PRESSED for s in states]
assert buttons_pressed and all(buttons_pressed), (
"At least one button was not pressed"
)
async def test_request_sync(hass: HomeAssistant) -> None:
"""Test that the config entry is reloaded when a requestSync request is sent."""
updatedDeviceInst = Device(id="TestDevice", name="TestDevice")
updatedDeviceInst.buttons = [
Button(id="Button 1", name="New Button 1", device=deviceInst),
Button(id="Button 2", name="New Button 2", device=deviceInst),
Button(id="Button 3", name="New Button 3", device=deviceInst),
]
with patch(
"homeassistant.components.gentex_homelink.MQTTProvider", autospec=True
) as MockProvider:
instance = MockProvider.return_value
instance.discover.side_effect = [[deviceInst], [updatedDeviceInst]]
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=None,
version=1,
data={
"auth_implementation": "gentex_homelink",
"token": {"expires_at": time.time() + 10000, "access_token": ""},
"last_update_id": None,
},
state=ConfigEntryState.LOADED,
)
config_entry.add_to_hass(hass)
result = await async_setup_entry(hass, config_entry)
# Assert configuration worked without errors
assert result
# Check to see if the correct buttons names were loaded
comp = er.async_get(hass)
button_names = {"Button 1", "Button 2", "Button 3"}
registered_button_names = {b.original_name for b in comp.entities.values()}
assert button_names == registered_button_names, (
"Expect button names to be correct for the initial config"
)
provider = config_entry.runtime_data.provider
coordinator = config_entry.runtime_data.coordinator
with patch.object(
coordinator.hass.config_entries, "async_reload"
) as async_reload_mock:
# Mimic request sync event
state_data = {
"type": "requestSync",
}
# async reload should not be called yet
async_reload_mock.assert_not_called()
# Send the request sync
provider.listen.mock_calls[0].args[0](None, state_data)
# Wait for the request to be processed
await hass.async_block_till_done(wait_background_tasks=True)
await asyncio.gather(*asyncio.all_tasks() - {asyncio.current_task()})
await asyncio.sleep(0.01)
# Now async reload should have been called
async_reload_mock.assert_called()

View File

@@ -1,55 +1,77 @@
"""Test that the devices and entities are correctly configured."""
import time
from unittest.mock import AsyncMock
from unittest.mock import patch
from homelink.model.button import Button
from homelink.model.device import Device
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_UNKNOWN
from homeassistant.components.gentex_homelink import async_setup_entry
from homeassistant.components.gentex_homelink.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
import homeassistant.helpers.device_registry as dr
import homeassistant.helpers.entity_registry as er
from . import setup_integration, update_callback
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
from tests.common import MockConfigEntry, snapshot_platform
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
TEST_CONFIG_ENTRY_ID = "ABC123"
"""Mock classes for testing."""
async def test_entities(
deviceInst = Device(id="TestDevice", name="TestDevice")
deviceInst.buttons = [
Button(id="1", name="Button 1", device=deviceInst),
Button(id="2", name="Button 2", device=deviceInst),
Button(id="3", name="Button 3", device=deviceInst),
]
@pytest.fixture
async def test_setup_config(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_mqtt_provider: AsyncMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Setup config entry."""
with patch(
"homeassistant.components.gentex_homelink.MQTTProvider", autospec=True
) as MockProvider:
instance = MockProvider.return_value
instance.discover.return_value = [deviceInst]
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=None,
version=1,
data={"auth_implementation": "gentex_homelink"},
state=ConfigEntryState.LOADED,
)
config_entry.add_to_hass(hass)
result = await async_setup_entry(hass, config_entry)
# Assert configuration worked without errors
assert result
async def test_device_registered(hass: HomeAssistant, test_setup_config) -> None:
"""Check if a device is registered."""
# Assert we got a device with the test ID
device_registry = dr.async_get(hass)
device = device_registry.async_get_device([(DOMAIN, "TestDevice")])
assert device
assert device.name == "TestDevice"
def test_entities_registered(hass: HomeAssistant, test_setup_config) -> None:
"""Check if the entities are registered."""
await setup_integration(hass, mock_config_entry)
comp = er.async_get(hass)
button_names = {"Button 1", "Button 2", "Button 3"}
registered_button_names = {b.original_name for b in comp.entities.values()}
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.freeze_time("2021-07-30")
async def test_entities_update(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_mqtt_provider: AsyncMock,
) -> None:
"""Check if the entities are updated."""
await setup_integration(hass, mock_config_entry)
assert hass.states.get("event.testdevice_button_1").state == STATE_UNKNOWN
await update_callback(
hass,
mock_mqtt_provider,
"state",
{
"1": {"requestId": "rid1", "timestamp": time.time()},
"2": {"requestId": "rid2", "timestamp": time.time()},
"3": {"requestId": "rid3", "timestamp": time.time()},
},
)
assert (
hass.states.get("event.testdevice_button_1").state
== "2021-07-30T00:00:00.000+00:00"
)
assert button_names == registered_button_names

View File

@@ -1,66 +1,32 @@
"""Test that the integration is initialized correctly."""
from unittest.mock import AsyncMock, patch
from syrupy.assertion import SnapshotAssertion
from unittest.mock import patch
from homeassistant.components import gentex_homelink
from homeassistant.components.gentex_homelink.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
import homeassistant.helpers.device_registry as dr
from . import setup_integration, update_callback
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
async def test_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_mqtt_provider: AsyncMock,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test device is registered correctly."""
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device(
identifiers={(DOMAIN, "TestDevice")},
)
assert device
assert device == snapshot
async def test_reload_sync(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_mqtt_provider: AsyncMock,
) -> None:
"""Test that the config entry is reloaded when a requestSync request is sent."""
await setup_integration(hass, mock_config_entry)
with patch.object(hass.config_entries, "async_reload") as async_reload_mock:
await update_callback(
hass,
mock_mqtt_provider,
"requestSync",
{},
)
async_reload_mock.assert_called_once_with(mock_config_entry.entry_id)
async def test_load_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_mqtt_provider: AsyncMock,
) -> None:
"""Test the entry can be loaded and unloaded."""
await setup_integration(hass, mock_config_entry)
with patch("homeassistant.components.gentex_homelink.MQTTProvider", autospec=True):
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=None,
version=1,
data={"auth_implementation": "gentex_homelink"},
)
entry.add_to_hass(hass)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await async_setup_component(hass, DOMAIN, {}) is True, (
"Component is not set up"
)
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
assert await gentex_homelink.async_unload_entry(hass, entry), (
"Component not unloaded"
)