evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix KNX device trigger passing info data (#95076)

Browse source
This commit is contained in:
Matthias Alphart 2023-06-22 21:13:10 +02:00 • committed by Franck Nijhof
parent 0fa954040e
commit 8f9425f09f
No known key found for this signature in database
GPG key ID: D62583BA8AB11CA3
2 changed files with 17 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
homeassistant/components/knx/device_trigger.py
View file

@ -84,6 +84,7 @@ async def async_attach_trigger(
trigger_info: TriggerInfo,
) -> CALLBACK_TYPE:
"""Attach a trigger."""
trigger_data = trigger_info["trigger_data"]
dst_addresses: list[str] = config.get(EXTRA_FIELD_DESTINATION, [])
job = HassJob(action, f"KNX device trigger {trigger_info}")
knx: KNXModule = hass.data[DOMAIN]
@ -95,7 +96,7 @@ async def async_attach_trigger(
return
hass.async_run_hass_job(
job,
{"trigger": telegram},
{"trigger": {**trigger_data, **telegram}},
)
return knx.telegrams.async_listen_telegram(

20
tests/components/knx/test_device_trigger.py
View file

@ -56,6 +56,7 @@ async def test_if_fires_on_telegram(
identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")}
)
# "id" field added to action to test if `trigger_data` passed correctly in `async_attach_trigger`
assert await async_setup_component(
hass,
automation.DOMAIN,
@ -71,7 +72,8 @@ async def test_if_fires_on_telegram(
"action": {
"service": "test.automation",
"data_template": {
"catch_all": ("telegram - {{ trigger.destination }}")
"catch_all": ("telegram - {{ trigger.destination }}"),
"id": (" {{ trigger.id }}"),
},
},
},
@ -82,11 +84,13 @@ async def test_if_fires_on_telegram(
"device_id": device_entry.id,
"type": "telegram",
"destination": ["1/2/3", "1/2/4"],
"id": "test-id",
},
"action": {
"service": "test.automation",
"data_template": {
"specific": ("telegram - {{ trigger.destination }}")
"specific": ("telegram - {{ trigger.destination }}"),
"id": (" {{ trigger.id }}"),
},
},
},
@ -96,12 +100,18 @@ async def test_if_fires_on_telegram(
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(calls) == 1
assert calls.pop().data["catch_all"] == "telegram - 0/0/1"
test_call = calls.pop()
assert test_call.data["catch_all"] == "telegram - 0/0/1"
assert test_call.data["id"] == 0
await knx.receive_write("1/2/4", (0x03, 0x2F))
assert len(calls) == 2
assert calls.pop().data["specific"] == "telegram - 1/2/4"
assert calls.pop().data["catch_all"] == "telegram - 1/2/4"
test_call = calls.pop()
assert test_call.data["specific"] == "telegram - 1/2/4"
assert test_call.data["id"] == "test-id"
test_call = calls.pop()
assert test_call.data["catch_all"] == "telegram - 1/2/4"
assert test_call.data["id"] == 0
async def test_remove_device_trigger(