jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Shelly RPC device trigger and logbook platforms (#56428)

Browse Source 
* Add RPC device trigger and logbook platforms

* Single input event for Block and RPC

* Add device generation to shelly.click
This commit is contained in:
Shay Levy 2021-09-21 00:09:44 +03:00 committed by GitHub
parent 542f637ac4
commit 47340802b3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 409 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
homeassistant/components/shelly/__init__.py
View File

@ -31,6 +31,7 @@ from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
ATTR_GENERATION,
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION,
BLOCK,
CONF_COAP_PORT,
@ -44,6 +45,7 @@ from .const import (
REST,
REST_SENSORS_UPDATE_INTERVAL,
RPC,
RPC_INPUTS_EVENTS_TYPES,
RPC_RECONNECT_INTERVAL,
SHBTN_MODELS,
SLEEP_PERIOD_MULTIPLIER,
@ -250,8 +252,8 @@ class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator):
self.entry = entry
self.device = device
self._async_remove_device_updates_handler = self.async_add_listener(
self._async_device_updates_handler
entry.async_on_unload(
self.async_add_listener(self._async_device_updates_handler)
)
self._last_input_events_count: dict = {}
@ -306,6 +308,7 @@ class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator):
ATTR_DEVICE: self.device.settings["device"]["hostname"],
ATTR_CHANNEL: channel,
ATTR_CLICK_TYPE: INPUTS_EVENTS_DICT[event_type],
ATTR_GENERATION: 1,
},
)
else:
@ -356,7 +359,6 @@ class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator):
def shutdown(self) -> None:
"""Shutdown the wrapper."""
self.device.shutdown()
self._async_remove_device_updates_handler()
@callback
def _handle_ha_stop(self, _event: Event) -> None:
@ -435,27 +437,40 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok
def get_device_wrapper(
def get_block_device_wrapper(
hass: HomeAssistant, device_id: str
) -> BlockDeviceWrapper | RpcDeviceWrapper | None:
"""Get a Shelly device wrapper for the given device id."""
) -> BlockDeviceWrapper | None:
"""Get a Shelly block device wrapper for the given device id."""
if not hass.data.get(DOMAIN):
return None
for config_entry in hass.data[DOMAIN][DATA_CONFIG_ENTRY]:
block_wrapper: BlockDeviceWrapper | None = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
config_entry
].get(BLOCK)
dev_reg = device_registry.async_get(hass)
if device := dev_reg.async_get(device_id):
for config_entry in device.config_entries:
if not hass.data[DOMAIN][DATA_CONFIG_ENTRY].get(config_entry):
continue
if block_wrapper and block_wrapper.device_id == device_id:
return block_wrapper
if wrapper := hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry].get(BLOCK):
return cast(BlockDeviceWrapper, wrapper)
rpc_wrapper: RpcDeviceWrapper | None = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
config_entry
].get(RPC)
return None
if rpc_wrapper and rpc_wrapper.device_id == device_id:
return rpc_wrapper
def get_rpc_device_wrapper(
hass: HomeAssistant, device_id: str
) -> RpcDeviceWrapper | None:
"""Get a Shelly RPC device wrapper for the given device id."""
if not hass.data.get(DOMAIN):
return None
dev_reg = device_registry.async_get(hass)
if device := dev_reg.async_get(device_id):
for config_entry in device.config_entries:
if not hass.data[DOMAIN][DATA_CONFIG_ENTRY].get(config_entry):
continue
if wrapper := hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry].get(RPC):
return cast(RpcDeviceWrapper, wrapper)
return None
@ -479,10 +494,42 @@ class RpcDeviceWrapper(update_coordinator.DataUpdateCoordinator):
self.entry = entry
self.device = device
entry.async_on_unload(
self.async_add_listener(self._async_device_updates_handler)
)
self._last_event: dict[str, Any] | None = None
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
)
@callback
def _async_device_updates_handler(self) -> None:
"""Handle device updates."""
if (
not self.device.initialized
or not self.device.event
or self.device.event == self._last_event
):
return
self._last_event = self.device.event
for event in self.device.event["events"]:
if event.get("event") not in RPC_INPUTS_EVENTS_TYPES:
continue
self.hass.bus.async_fire(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: self.device_id,
ATTR_DEVICE: self.device.hostname,
ATTR_CHANNEL: event["id"] + 1,
ATTR_CLICK_TYPE: event["event"],
ATTR_GENERATION: 2,
},
)
async def _async_update_data(self) -> None:
"""Fetch data."""
if self.device.connected:

22
homeassistant/components/shelly/const.py
View File

@ -64,18 +64,28 @@ INPUTS_EVENTS_DICT: Final = {
# List of battery devices that maintain a permanent WiFi connection
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION: Final = ["SHMOS-01"]
# Button/Click events for Block & RPC devices
EVENT_SHELLY_CLICK: Final = "shelly.click"
ATTR_CLICK_TYPE: Final = "click_type"
ATTR_CHANNEL: Final = "channel"
ATTR_DEVICE: Final = "device"
ATTR_GENERATION: Final = "generation"
CONF_SUBTYPE: Final = "subtype"
BASIC_INPUTS_EVENTS_TYPES: Final = {"single", "long"}
SHBTN_INPUTS_EVENTS_TYPES: Final = {"single", "double", "triple", "long"}
SUPPORTED_INPUTS_EVENTS_TYPES: Final = {
RPC_INPUTS_EVENTS_TYPES: Final = {
"btn_down",
"btn_up",
"single_push",
"double_push",
"long_push",
}
BLOCK_INPUTS_EVENTS_TYPES: Final = {
"single",
"double",
"triple",
@ -84,9 +94,15 @@ SUPPORTED_INPUTS_EVENTS_TYPES: Final = {
"long_single",
}
SHIX3_1_INPUTS_EVENTS_TYPES = SUPPORTED_INPUTS_EVENTS_TYPES
SHIX3_1_INPUTS_EVENTS_TYPES = BLOCK_INPUTS_EVENTS_TYPES
INPUTS_EVENTS_SUBTYPES: Final = {"button": 1, "button1": 1, "button2": 2, "button3": 3}
INPUTS_EVENTS_SUBTYPES: Final = {
"button": 1,
"button1": 1,
"button2": 2,
"button3": 3,
"button4": 4,
}
SHBTN_MODELS: Final = ["SHBTN-1", "SHBTN-2"]

112
homeassistant/components/shelly/device_trigger.py
View File

@ -25,28 +25,52 @@ from homeassistant.const import (
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers.typing import ConfigType
from . import RpcDeviceWrapper, get_device_wrapper
from . import get_block_device_wrapper, get_rpc_device_wrapper
from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
BLOCK_INPUTS_EVENTS_TYPES,
CONF_SUBTYPE,
DOMAIN,
EVENT_SHELLY_CLICK,
INPUTS_EVENTS_SUBTYPES,
SHBTN_INPUTS_EVENTS_TYPES,
RPC_INPUTS_EVENTS_TYPES,
SHBTN_MODELS,
SUPPORTED_INPUTS_EVENTS_TYPES,
)
from .utils import get_input_triggers
from .utils import (
get_block_input_triggers,
get_rpc_input_triggers,
get_shbtn_input_triggers,
)
TRIGGER_SCHEMA: Final = DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In(SUPPORTED_INPUTS_EVENTS_TYPES),
vol.Required(CONF_TYPE): vol.In(
RPC_INPUTS_EVENTS_TYPES | BLOCK_INPUTS_EVENTS_TYPES
),
vol.Required(CONF_SUBTYPE): vol.In(INPUTS_EVENTS_SUBTYPES),
}
)
def append_input_triggers(
triggers: list[dict[str, Any]],
input_triggers: list[tuple[str, str]],
device_id: str,
) -> None:
"""Add trigger to triggers list."""
for trigger, subtype in input_triggers:
triggers.append(
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
)
async def async_validate_trigger_config(
hass: HomeAssistant, config: dict[str, Any]
) -> dict[str, Any]:
@ -54,23 +78,29 @@ async def async_validate_trigger_config(
config = TRIGGER_SCHEMA(config)
# if device is available verify parameters against device capabilities
wrapper = get_device_wrapper(hass, config[CONF_DEVICE_ID])
if isinstance(wrapper, RpcDeviceWrapper):
return config
if not wrapper or not wrapper.device.initialized:
return config
trigger = (config[CONF_TYPE], config[CONF_SUBTYPE])
assert wrapper.device.blocks
if config[CONF_TYPE] in RPC_INPUTS_EVENTS_TYPES:
rpc_wrapper = get_rpc_device_wrapper(hass, config[CONF_DEVICE_ID])
if not rpc_wrapper or not rpc_wrapper.device.initialized:
return config
for block in wrapper.device.blocks:
input_triggers = get_input_triggers(wrapper.device, block)
input_triggers = get_rpc_input_triggers(rpc_wrapper.device)
if trigger in input_triggers:
return config
elif config[CONF_TYPE] in BLOCK_INPUTS_EVENTS_TYPES:
block_wrapper = get_block_device_wrapper(hass, config[CONF_DEVICE_ID])
if not block_wrapper or not block_wrapper.device.initialized:
return config
assert block_wrapper.device.blocks
for block in block_wrapper.device.blocks:
input_triggers = get_block_input_triggers(block_wrapper.device, block)
if trigger in input_triggers:
return config
raise InvalidDeviceAutomationConfig(
f"Invalid ({CONF_TYPE},{CONF_SUBTYPE}): {trigger}"
)
@ -80,45 +110,28 @@ async def async_get_triggers(
hass: HomeAssistant, device_id: str
) -> list[dict[str, Any]]:
"""List device triggers for Shelly devices."""
wrapper = get_device_wrapper(hass, device_id)
if not wrapper:
raise InvalidDeviceAutomationConfig(f"Device not found: {device_id}")
triggers: list[dict[str, Any]] = []
if isinstance(wrapper, RpcDeviceWrapper):
return []
triggers = []
if wrapper.model in SHBTN_MODELS:
for trigger in SHBTN_INPUTS_EVENTS_TYPES:
triggers.append(
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: trigger,
CONF_SUBTYPE: "button",
}
)
if rpc_wrapper := get_rpc_device_wrapper(hass, device_id):
input_triggers = get_rpc_input_triggers(rpc_wrapper.device)
append_input_triggers(triggers, input_triggers, device_id)
return triggers
assert wrapper.device.blocks
if block_wrapper := get_block_device_wrapper(hass, device_id):
if block_wrapper.model in SHBTN_MODELS:
input_triggers = get_shbtn_input_triggers()
append_input_triggers(triggers, input_triggers, device_id)
return triggers
for block in wrapper.device.blocks:
input_triggers = get_input_triggers(wrapper.device, block)
assert block_wrapper.device.blocks
for trigger, subtype in input_triggers:
triggers.append(
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
)
for block in block_wrapper.device.blocks:
input_triggers = get_block_input_triggers(block_wrapper.device, block)
append_input_triggers(triggers, input_triggers, device_id)
return triggers
return triggers
raise InvalidDeviceAutomationConfig(f"Device not found: {device_id}")
async def async_attach_trigger(
@ -137,6 +150,7 @@ async def async_attach_trigger(
ATTR_CLICK_TYPE: config[CONF_TYPE],
},
}
event_config = event_trigger.TRIGGER_SCHEMA(event_config)
return await event_trigger.async_attach_trigger(
hass, event_config, action, automation_info, platform_type="device"

36
homeassistant/components/shelly/logbook.py
View File

@ -7,15 +7,17 @@ from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.typing import EventType
from . import RpcDeviceWrapper, get_device_wrapper
from . import get_block_device_wrapper, get_rpc_device_wrapper
from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
BLOCK_INPUTS_EVENTS_TYPES,
DOMAIN,
EVENT_SHELLY_CLICK,
RPC_INPUTS_EVENTS_TYPES,
)
from .utils import get_block_device_name
from .utils import get_block_device_name, get_rpc_entity_name
@callback
@ -27,23 +29,27 @@ def async_describe_events(
@callback
def async_describe_shelly_click_event(event: EventType) -> dict[str, str]:
"""Describe shelly.click logbook event."""
wrapper = get_device_wrapper(hass, event.data[ATTR_DEVICE_ID])
if isinstance(wrapper, RpcDeviceWrapper):
return {}
if wrapper and wrapper.device.initialized:
device_name = get_block_device_name(wrapper.device)
else:
device_name = event.data[ATTR_DEVICE]
channel = event.data[ATTR_CHANNEL]
"""Describe shelly.click logbook event (block device)."""
device_id = event.data[ATTR_DEVICE_ID]
click_type = event.data[ATTR_CLICK_TYPE]
channel = event.data[ATTR_CHANNEL]
input_name = f"{event.data[ATTR_DEVICE]} channel {channel}"
if click_type in RPC_INPUTS_EVENTS_TYPES:
rpc_wrapper = get_rpc_device_wrapper(hass, device_id)
if rpc_wrapper and rpc_wrapper.device.initialized:
key = f"input:{channel-1}"
input_name = get_rpc_entity_name(rpc_wrapper.device, key)
elif click_type in BLOCK_INPUTS_EVENTS_TYPES:
block_wrapper = get_block_device_wrapper(hass, device_id)
if block_wrapper and block_wrapper.device.initialized:
device_name = get_block_device_name(block_wrapper.device)
input_name = f"{device_name} channel {channel}"
return {
"name": "Shelly",
"message": f"'{click_type}' click event for {device_name} channel {channel} was fired.",
"message": f"'{click_type}' click event for {input_name} Input was fired.",
}
async_describe_event(DOMAIN, EVENT_SHELLY_CLICK, async_describe_shelly_click_event)

10
homeassistant/components/shelly/strings.json
View File

@ -33,7 +33,8 @@
"button": "Button",
"button1": "First button",
"button2": "Second button",
"button3": "Third button"
"button3": "Third button",
"button4": "Fourth button"
},
"trigger_type": {
"single": "{subtype} single clicked",
@ -41,7 +42,12 @@
"triple": "{subtype} triple clicked",
"long": " {subtype} long clicked",
"single_long": "{subtype} single clicked and then long clicked",
"long_single": "{subtype} long clicked and then single clicked"
"long_single": "{subtype} long clicked and then single clicked",
"btn_down": "{subtype} button down",
"btn_up": "{subtype} button up",
"single_push": "{subtype} single push",
"double_push": "{subtype} double push",
"long_push": " {subtype} long push"
}
}
}

10
homeassistant/components/shelly/translations/en.json
View File

@ -33,7 +33,8 @@
"button": "Button",
"button1": "First button",
"button2": "Second button",
"button3": "Third button"
"button3": "Third button",
"button4": "Fourth button"
},
"trigger_type": {
"double": "{subtype} double clicked",
@ -41,7 +42,12 @@
"long_single": "{subtype} long clicked and then single clicked",
"single": "{subtype} single clicked",
"single_long": "{subtype} single clicked and then long clicked",
"triple": "{subtype} triple clicked"
"triple": "{subtype} triple clicked",
"btn_down": "{subtype} button down",
"btn_up": "{subtype} button up",
"single_push": "{subtype} single push",
"double_push": "{subtype} double push",
"long_push": " {subtype} long push"
}
}
}

33
homeassistant/components/shelly/utils.py
View File

@ -22,6 +22,7 @@ from .const import (
DEFAULT_COAP_PORT,
DOMAIN,
MAX_RPC_KEY_INSTANCES,
RPC_INPUTS_EVENTS_TYPES,
SHBTN_INPUTS_EVENTS_TYPES,
SHBTN_MODELS,
SHIX3_1_INPUTS_EVENTS_TYPES,
@ -162,7 +163,9 @@ def get_device_uptime(uptime: float, last_uptime: str | None) -> str:
return last_uptime
def get_input_triggers(device: BlockDevice, block: Block) -> list[tuple[str, str]]:
def get_block_input_triggers(
device: BlockDevice, block: Block
) -> list[tuple[str, str]]:
"""Return list of input triggers for block."""
if "inputEvent" not in block.sensor_ids or "inputEventCnt" not in block.sensor_ids:
return []
@ -191,6 +194,16 @@ def get_input_triggers(device: BlockDevice, block: Block) -> list[tuple[str, str
return triggers
def get_shbtn_input_triggers() -> list[tuple[str, str]]:
"""Return list of input triggers for SHBTN models."""
triggers = []
for trigger_type in SHBTN_INPUTS_EVENTS_TYPES:
triggers.append((trigger_type, "button"))
return triggers
@singleton.singleton("shelly_coap")
async def get_coap_context(hass: HomeAssistant) -> COAP:
"""Get CoAP context to be used in all Shelly devices."""
@ -314,3 +327,21 @@ def is_rpc_channel_type_light(config: dict[str, Any], channel: int) -> bool:
"""Return true if rpc channel consumption type is set to light."""
con_types = config["sys"]["ui_data"].get("consumption_types")
return con_types is not None and con_types[channel].lower().startswith("light")
def get_rpc_input_triggers(device: RpcDevice) -> list[tuple[str, str]]:
"""Return list of input triggers for RPC device."""
triggers = []
key_ids = get_rpc_key_ids(device.config, "input")
for id_ in key_ids:
key = f"input:{id_}"
if not is_rpc_momentary_input(device.config, key):
continue
for trigger_type in RPC_INPUTS_EVENTS_TYPES:
subtype = f"button{id_+1}"
triggers.append((trigger_type, subtype))
return triggers

2
tests/components/shelly/conftest.py
View File

@ -56,6 +56,7 @@ MOCK_BLOCKS = [
]
MOCK_CONFIG = {
"input:0": {"id": 0, "type": "button"},
"switch:0": {"name": "test switch_0"},
"sys": {"ui_data": {}},
"wifi": {
@ -147,6 +148,7 @@ async def rpc_wrapper(hass):
device = Mock(
call_rpc=AsyncMock(),
config=MOCK_CONFIG,
event={},
shelly=MOCK_SHELLY,
status=MOCK_STATUS,
firmware_version="some fw string",

145
tests/components/shelly/test_device_trigger.py
View File

@ -30,8 +30,8 @@ from tests.common import (
)
async def test_get_triggers(hass, coap_wrapper):
"""Test we get the expected triggers from a shelly."""
async def test_get_triggers_block_device(hass, coap_wrapper):
"""Test we get the expected triggers from a shelly block device."""
assert coap_wrapper
expected_triggers = [
{
@ -57,6 +57,54 @@ async def test_get_triggers(hass, coap_wrapper):
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_rpc_device(hass, rpc_wrapper):
"""Test we get the expected triggers from a shelly RPC device."""
assert rpc_wrapper
expected_triggers = [
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "btn_down",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "btn_up",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "single_push",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "double_push",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "long_push",
CONF_SUBTYPE: "button1",
},
]
triggers = await async_get_device_automations(
hass, "trigger", rpc_wrapper.device_id
)
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_button(hass):
"""Test we get the expected triggers from a shelly button."""
await async_setup_component(hass, "shelly", {})
@ -136,8 +184,8 @@ async def test_get_triggers_for_invalid_device_id(hass, device_reg, coap_wrapper
await async_get_device_automations(hass, "trigger", invalid_device.id)
async def test_if_fires_on_click_event(hass, calls, coap_wrapper):
"""Test for click_event trigger firing."""
async def test_if_fires_on_click_event_block_device(hass, calls, coap_wrapper):
"""Test for click_event trigger firing for block device."""
assert coap_wrapper
await setup.async_setup_component(hass, "persistent_notification", {})
@ -175,8 +223,47 @@ async def test_if_fires_on_click_event(hass, calls, coap_wrapper):
assert calls[0].data["some"] == "test_trigger_single_click"
async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
"""Test for click_event with no device."""
async def test_if_fires_on_click_event_rpc_device(hass, calls, rpc_wrapper):
"""Test for click_event trigger firing for rpc device."""
assert rpc_wrapper
await setup.async_setup_component(hass, "persistent_notification", {})
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_TYPE: "single_push",
CONF_SUBTYPE: "button1",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_single_push"},
},
},
]
},
)
message = {
CONF_DEVICE_ID: rpc_wrapper.device_id,
ATTR_CLICK_TYPE: "single_push",
ATTR_CHANNEL: 1,
}
hass.bus.async_fire(EVENT_SHELLY_CLICK, message)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_single_push"
async def test_validate_trigger_block_device_not_ready(hass, calls, coap_wrapper):
"""Test validate trigger config when block device is not ready."""
assert coap_wrapper
await setup.async_setup_component(hass, "persistent_notification", {})
@ -189,7 +276,7 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: "no_device",
CONF_DEVICE_ID: "device_not_ready",
CONF_TYPE: "single",
CONF_SUBTYPE: "button1",
},
@ -201,7 +288,11 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
]
},
)
message = {CONF_DEVICE_ID: "no_device", ATTR_CLICK_TYPE: "single", ATTR_CHANNEL: 1}
message = {
CONF_DEVICE_ID: "device_not_ready",
ATTR_CLICK_TYPE: "single",
ATTR_CHANNEL: 1,
}
hass.bus.async_fire(EVENT_SHELLY_CLICK, message)
await hass.async_block_till_done()
@ -209,6 +300,44 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
assert calls[0].data["some"] == "test_trigger_single_click"
async def test_validate_trigger_rpc_device_not_ready(hass, calls, rpc_wrapper):
"""Test validate trigger config when RPC device is not ready."""
assert rpc_wrapper
await setup.async_setup_component(hass, "persistent_notification", {})
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: "device_not_ready",
CONF_TYPE: "single_push",
CONF_SUBTYPE: "button1",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_single_push"},
},
},
]
},
)
message = {
CONF_DEVICE_ID: "device_not_ready",
ATTR_CLICK_TYPE: "single_push",
ATTR_CHANNEL: 1,
}
hass.bus.async_fire(EVENT_SHELLY_CLICK, message)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_single_push"
async def test_validate_trigger_invalid_triggers(hass, coap_wrapper):
"""Test for click_event with invalid triggers."""
assert coap_wrapper

59
tests/components/shelly/test_logbook.py
View File

@ -13,8 +13,8 @@ from homeassistant.setup import async_setup_component
from tests.components.logbook.test_init import MockLazyEventPartialState
async def test_humanify_shelly_click_event(hass, coap_wrapper):
"""Test humanifying Shelly click event."""
async def test_humanify_shelly_click_event_block_device(hass, coap_wrapper):
"""Test humanifying Shelly click event for block device."""
assert coap_wrapper
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
@ -51,12 +51,63 @@ async def test_humanify_shelly_click_event(hass, coap_wrapper):
assert event1["name"] == "Shelly"
assert event1["domain"] == DOMAIN
assert (
event1["message"] == "'single' click event for Test name channel 1 was fired."
event1["message"]
== "'single' click event for Test name channel 1 Input was fired."
)
assert event2["name"] == "Shelly"
assert event2["domain"] == DOMAIN
assert (
event2["message"]
== "'long' click event for shellyswitch25-12345678 channel 2 was fired."
== "'long' click event for shellyswitch25-12345678 channel 2 Input was fired."
)
async def test_humanify_shelly_click_event_rpc_device(hass, rpc_wrapper):
"""Test humanifying Shelly click event for rpc device."""
assert rpc_wrapper
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
entity_attr_cache = logbook.EntityAttributeCache(hass)
event1, event2 = list(
logbook.humanify(
hass,
[
MockLazyEventPartialState(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: rpc_wrapper.device_id,
ATTR_DEVICE: "shellyplus1pm-12345678",
ATTR_CLICK_TYPE: "single_push",
ATTR_CHANNEL: 1,
},
),
MockLazyEventPartialState(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: "no_device_id",
ATTR_DEVICE: "shellypro4pm-12345678",
ATTR_CLICK_TYPE: "btn_down",
ATTR_CHANNEL: 2,
},
),
],
entity_attr_cache,
{},
)
)
assert event1["name"] == "Shelly"
assert event1["domain"] == DOMAIN
assert (
event1["message"]
== "'single_push' click event for test switch_0 Input was fired."
)
assert event2["name"] == "Shelly"
assert event2["domain"] == DOMAIN
assert (
event2["message"]
== "'btn_down' click event for shellypro4pm-12345678 channel 2 Input was fired."
)