Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

New template merge_response (#114204) 

* New template merge_response

* Extending

* Extend comment

* Update

* Fixes

* Fix comments

* Mods

* snapshots

* Fixes from discussion
pull/125060/head
G Johansson 2024-09-02 09:13:10 +03:00 committed by GitHub
parent 9fff3a13a5
commit 78cf7dc873
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 654 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
homeassistant/helpers/template.py
View File

@ -51,6 +51,7 @@ from homeassistant.const import (
from homeassistant.core import ( from homeassistant.core import (
Context, Context,
HomeAssistant, HomeAssistant,
ServiceResponse,
State, State,
callback, callback,
split_entity_id, split_entity_id,
@ -2118,6 +2119,62 @@ def as_timedelta(value: str) -> timedelta | None:
return dt_util.parse_duration(value) return dt_util.parse_duration(value)
def merge_response(value: ServiceResponse) -> list[Any]:
"""Merge action responses into single list.
Checks that the input is a correct service response:
{
"entity_id": {str: dict[str, Any]},
}
If response is a single list, it will extend the list with the items
and add the entity_id and value_key to each dictionary for reference.
If response is a dictionary or multiple lists,
it will append the dictionary/lists to the list
and add the entity_id to each dictionary for reference.
"""
if not isinstance(value, dict):
raise TypeError("Response is not a dictionary")
if not value:
# Bail out early if response is an empty dictionary
return []
is_single_list = False
response_items: list = []
for entity_id, entity_response in value.items(): # pylint: disable=too-many-nested-blocks
if not isinstance(entity_response, dict):
raise TypeError("Response is not a dictionary")
for value_key, type_response in entity_response.items():
if len(entity_response) == 1 and isinstance(type_response, list):
# Provides special handling for responses such as calendar events
# and weather forecasts where the response contains a single list with multiple
# dictionaries inside.
is_single_list = True
for dict_in_list in type_response:
if isinstance(dict_in_list, dict):
if ATTR_ENTITY_ID in dict_in_list:
raise ValueError(
f"Response dictionary already contains key '{ATTR_ENTITY_ID}'"
)
dict_in_list[ATTR_ENTITY_ID] = entity_id
dict_in_list["value_key"] = value_key
response_items.extend(type_response)
else:
# Break the loop if not a single list as the logic is then managed in the outer loop
# which handles both dictionaries and in the case of multiple lists.
break
if not is_single_list:
_response = entity_response.copy()
if ATTR_ENTITY_ID in _response:
raise ValueError(
f"Response dictionary already contains key '{ATTR_ENTITY_ID}'"
)
_response[ATTR_ENTITY_ID] = entity_id
response_items.append(_response)
return response_items
def strptime(string, fmt, default=_SENTINEL): def strptime(string, fmt, default=_SENTINEL):
"""Parse a time string to datetime.""" """Parse a time string to datetime."""
try: try:
@ -2833,6 +2890,7 @@ class TemplateEnvironment(ImmutableSandboxedEnvironment):
self.globals["as_timedelta"] = as_timedelta self.globals["as_timedelta"] = as_timedelta
self.globals["as_timestamp"] = forgiving_as_timestamp self.globals["as_timestamp"] = forgiving_as_timestamp
self.globals["timedelta"] = timedelta self.globals["timedelta"] = timedelta
self.globals["merge_response"] = merge_response
self.globals["strptime"] = strptime self.globals["strptime"] = strptime
self.globals["urlencode"] = urlencode self.globals["urlencode"] = urlencode
self.globals["average"] = average self.globals["average"] = average

337
tests/helpers/snapshots/test_template.ambr Normal file
View File

@ -0,0 +1,337 @@
# serializer version: 1
# name: test_merge_response[calendar][a_response]
dict({
'calendar.local_furry_events': dict({
'events': list([
]),
}),
'calendar.sports': dict({
'events': list([
dict({
'description': '',
'end': '2024-02-27T18:00:00-06:00',
'start': '2024-02-27T17:00:00-06:00',
'summary': 'Basketball vs. Rockets',
}),
]),
}),
'calendar.yap_house_schedules': dict({
'events': list([
dict({
'description': '',
'end': '2024-02-26T09:00:00-06:00',
'start': '2024-02-26T08:00:00-06:00',
'summary': 'Dr. Appt',
}),
dict({
'description': 'something good',
'end': '2024-02-28T21:00:00-06:00',
'start': '2024-02-28T20:00:00-06:00',
'summary': 'Bake a cake',
}),
]),
}),
})
# ---
# name: test_merge_response[calendar][b_rendered]
Wrapper([
dict({
'description': '',
'end': '2024-02-27T18:00:00-06:00',
'entity_id': 'calendar.sports',
'start': '2024-02-27T17:00:00-06:00',
'summary': 'Basketball vs. Rockets',
'value_key': 'events',
}),
dict({
'description': '',
'end': '2024-02-26T09:00:00-06:00',
'entity_id': 'calendar.yap_house_schedules',
'start': '2024-02-26T08:00:00-06:00',
'summary': 'Dr. Appt',
'value_key': 'events',
}),
dict({
'description': 'something good',
'end': '2024-02-28T21:00:00-06:00',
'entity_id': 'calendar.yap_house_schedules',
'start': '2024-02-28T20:00:00-06:00',
'summary': 'Bake a cake',
'value_key': 'events',
}),
])
# ---
# name: test_merge_response[vacuum][a_response]
dict({
'vacuum.deebot_n8_plus_1': dict({
'header': dict({
'ver': '0.0.1',
}),
'payloadType': 'j',
'resp': dict({
'body': dict({
'msg': 'ok',
}),
}),
}),
'vacuum.deebot_n8_plus_2': dict({
'header': dict({
'ver': '0.0.1',
}),
'payloadType': 'j',
'resp': dict({
'body': dict({
'msg': 'ok',
}),
}),
}),
})
# ---
# name: test_merge_response[vacuum][b_rendered]
Wrapper([
dict({
'entity_id': 'vacuum.deebot_n8_plus_1',
'header': dict({
'ver': '0.0.1',
}),
'payloadType': 'j',
'resp': dict({
'body': dict({
'msg': 'ok',
}),
}),
}),
dict({
'entity_id': 'vacuum.deebot_n8_plus_2',
'header': dict({
'ver': '0.0.1',
}),
'payloadType': 'j',
'resp': dict({
'body': dict({
'msg': 'ok',
}),
}),
}),
])
# ---
# name: test_merge_response[weather][a_response]
dict({
'weather.forecast_home': dict({
'forecast': list([
dict({
'condition': 'cloudy',
'datetime': '2024-03-31T10:00:00+00:00',
'humidity': 71,
'precipitation': 0,
'precipitation_probability': 6.6,
'temperature': 10.9,
'templow': 6.5,
'wind_bearing': 71.8,
'wind_gust_speed': 24.1,
'wind_speed': 13.7,
}),
dict({
'condition': 'cloudy',
'datetime': '2024-04-01T10:00:00+00:00',
'humidity': 79,
'precipitation': 0,
'precipitation_probability': 8,
'temperature': 10.2,
'templow': 3.4,
'wind_bearing': 350.6,
'wind_gust_speed': 38.2,
'wind_speed': 21.6,
}),
dict({
'condition': 'snowy',
'datetime': '2024-04-02T10:00:00+00:00',
'humidity': 77,
'precipitation': 2.3,
'precipitation_probability': 67.4,
'temperature': 3,
'templow': 0,
'wind_bearing': 24.5,
'wind_gust_speed': 64.8,
'wind_speed': 37.4,
}),
]),
}),
'weather.smhi_home': dict({
'forecast': list([
dict({
'cloud_coverage': 100,
'condition': 'cloudy',
'datetime': '2024-03-31T16:00:00',
'humidity': 87,
'precipitation': 0.2,
'pressure': 998,
'temperature': 10,
'templow': 4,
'wind_bearing': 79,
'wind_gust_speed': 21.6,
'wind_speed': 11.88,
}),
dict({
'cloud_coverage': 100,
'condition': 'rainy',
'datetime': '2024-04-01T12:00:00',
'humidity': 88,
'precipitation': 2.2,
'pressure': 999,
'temperature': 6,
'templow': 1,
'wind_bearing': 17,
'wind_gust_speed': 20.52,
'wind_speed': 8.64,
}),
dict({
'cloud_coverage': 100,
'condition': 'cloudy',
'datetime': '2024-04-02T12:00:00',
'humidity': 71,
'precipitation': 1.3,
'pressure': 1003,
'temperature': 0,
'templow': -3,
'wind_bearing': 17,
'wind_gust_speed': 57.24,
'wind_speed': 30.6,
}),
]),
}),
})
# ---
# name: test_merge_response[weather][b_rendered]
Wrapper([
dict({
'cloud_coverage': 100,
'condition': 'cloudy',
'datetime': '2024-03-31T16:00:00',
'entity_id': 'weather.smhi_home',
'humidity': 87,
'precipitation': 0.2,
'pressure': 998,
'temperature': 10,
'templow': 4,
'value_key': 'forecast',
'wind_bearing': 79,
'wind_gust_speed': 21.6,
'wind_speed': 11.88,
}),
dict({
'cloud_coverage': 100,
'condition': 'rainy',
'datetime': '2024-04-01T12:00:00',
'entity_id': 'weather.smhi_home',
'humidity': 88,
'precipitation': 2.2,
'pressure': 999,
'temperature': 6,
'templow': 1,
'value_key': 'forecast',
'wind_bearing': 17,
'wind_gust_speed': 20.52,
'wind_speed': 8.64,
}),
dict({
'cloud_coverage': 100,
'condition': 'cloudy',
'datetime': '2024-04-02T12:00:00',
'entity_id': 'weather.smhi_home',
'humidity': 71,
'precipitation': 1.3,
'pressure': 1003,
'temperature': 0,
'templow': -3,
'value_key': 'forecast',
'wind_bearing': 17,
'wind_gust_speed': 57.24,
'wind_speed': 30.6,
}),
dict({
'condition': 'cloudy',
'datetime': '2024-03-31T10:00:00+00:00',
'entity_id': 'weather.forecast_home',
'humidity': 71,
'precipitation': 0,
'precipitation_probability': 6.6,
'temperature': 10.9,
'templow': 6.5,
'value_key': 'forecast',
'wind_bearing': 71.8,
'wind_gust_speed': 24.1,
'wind_speed': 13.7,
}),
dict({
'condition': 'cloudy',
'datetime': '2024-04-01T10:00:00+00:00',
'entity_id': 'weather.forecast_home',
'humidity': 79,
'precipitation': 0,
'precipitation_probability': 8,
'temperature': 10.2,
'templow': 3.4,
'value_key': 'forecast',
'wind_bearing': 350.6,
'wind_gust_speed': 38.2,
'wind_speed': 21.6,
}),
dict({
'condition': 'snowy',
'datetime': '2024-04-02T10:00:00+00:00',
'entity_id': 'weather.forecast_home',
'humidity': 77,
'precipitation': 2.3,
'precipitation_probability': 67.4,
'temperature': 3,
'templow': 0,
'value_key': 'forecast',
'wind_bearing': 24.5,
'wind_gust_speed': 64.8,
'wind_speed': 37.4,
}),
])
# ---
# name: test_merge_response[workday][a_response]
dict({
'binary_sensor.workday': dict({
'workday': True,
}),
'binary_sensor.workday2': dict({
'workday': False,
}),
})
# ---
# name: test_merge_response[workday][b_rendered]
Wrapper([
dict({
'entity_id': 'binary_sensor.workday',
'workday': True,
}),
dict({
'entity_id': 'binary_sensor.workday2',
'workday': False,
}),
])
# ---
# name: test_merge_response_with_empty_response[a_response]
dict({
'calendar.local_furry_events': dict({
'events': list([
]),
}),
'calendar.sports': dict({
'events': list([
]),
}),
'calendar.yap_house_schedules': dict({
'events': list([
]),
}),
})
# ---
# name: test_merge_response_with_empty_response[b_rendered]
Wrapper([
])
# ---

259
tests/helpers/test_template.py
View File

@ -15,6 +15,7 @@ from unittest.mock import patch
from freezegun import freeze_time from freezegun import freeze_time
import orjson import orjson
import pytest import pytest
from syrupy import SnapshotAssertion
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
@ -6288,3 +6289,261 @@ def test_template_output_exceeds_maximum_size(hass: HomeAssistant) -> None:
tpl = template.Template("{{ 'a' * 1024 * 257 }}", hass) tpl = template.Template("{{ 'a' * 1024 * 257 }}", hass)
with pytest.raises(TemplateError): with pytest.raises(TemplateError):
tpl.async_render() tpl.async_render()
@pytest.mark.parametrize(
("service_response"),
[
{
"calendar.sports": {
"events": [
{
"start": "2024-02-27T17:00:00-06:00",
"end": "2024-02-27T18:00:00-06:00",
"summary": "Basketball vs. Rockets",
"description": "",
}
]
},
"calendar.local_furry_events": {"events": []},
"calendar.yap_house_schedules": {
"events": [
{
"start": "2024-02-26T08:00:00-06:00",
"end": "2024-02-26T09:00:00-06:00",
"summary": "Dr. Appt",
"description": "",
},
{
"start": "2024-02-28T20:00:00-06:00",
"end": "2024-02-28T21:00:00-06:00",
"summary": "Bake a cake",
"description": "something good",
},
]
},
},
{
"binary_sensor.workday": {"workday": True},
"binary_sensor.workday2": {"workday": False},
},
{
"weather.smhi_home": {
"forecast": [
{
"datetime": "2024-03-31T16:00:00",
"condition": "cloudy",
"wind_bearing": 79,
"cloud_coverage": 100,
"temperature": 10,
"templow": 4,
"pressure": 998,
"wind_gust_speed": 21.6,
"wind_speed": 11.88,
"precipitation": 0.2,
"humidity": 87,
},
{
"datetime": "2024-04-01T12:00:00",
"condition": "rainy",
"wind_bearing": 17,
"cloud_coverage": 100,
"temperature": 6,
"templow": 1,
"pressure": 999,
"wind_gust_speed": 20.52,
"wind_speed": 8.64,
"precipitation": 2.2,
"humidity": 88,
},
{
"datetime": "2024-04-02T12:00:00",
"condition": "cloudy",
"wind_bearing": 17,
"cloud_coverage": 100,
"temperature": 0,
"templow": -3,
"pressure": 1003,
"wind_gust_speed": 57.24,
"wind_speed": 30.6,
"precipitation": 1.3,
"humidity": 71,
},
]
},
"weather.forecast_home": {
"forecast": [
{
"condition": "cloudy",
"precipitation_probability": 6.6,
"datetime": "2024-03-31T10:00:00+00:00",
"wind_bearing": 71.8,
"temperature": 10.9,
"templow": 6.5,
"wind_gust_speed": 24.1,
"wind_speed": 13.7,
"precipitation": 0,
"humidity": 71,
},
{
"condition": "cloudy",
"precipitation_probability": 8,
"datetime": "2024-04-01T10:00:00+00:00",
"wind_bearing": 350.6,
"temperature": 10.2,
"templow": 3.4,
"wind_gust_speed": 38.2,
"wind_speed": 21.6,
"precipitation": 0,
"humidity": 79,
},
{
"condition": "snowy",
"precipitation_probability": 67.4,
"datetime": "2024-04-02T10:00:00+00:00",
"wind_bearing": 24.5,
"temperature": 3,
"templow": 0,
"wind_gust_speed": 64.8,
"wind_speed": 37.4,
"precipitation": 2.3,
"humidity": 77,
},
]
},
},
{
"vacuum.deebot_n8_plus_1": {
"payloadType": "j",
"resp": {
"body": {
"msg": "ok",
}
},
"header": {
"ver": "0.0.1",
},
},
"vacuum.deebot_n8_plus_2": {
"payloadType": "j",
"resp": {
"body": {
"msg": "ok",
}
},
"header": {
"ver": "0.0.1",
},
},
},
],
ids=["calendar", "workday", "weather", "vacuum"],
)
async def test_merge_response(
hass: HomeAssistant,
service_response: dict,
snapshot: SnapshotAssertion,
) -> None:
"""Test the merge_response function/filter."""
_template = "{{ merge_response(" + str(service_response) + ") }}"
tpl = template.Template(_template, hass)
assert service_response == snapshot(name="a_response")
assert tpl.async_render() == snapshot(name="b_rendered")
async def test_merge_response_with_entity_id_in_response(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
) -> None:
"""Test the merge_response function/filter with empty lists."""
service_response = {
"test.response": {"some_key": True, "entity_id": "test.response"},
"test.response2": {"some_key": False, "entity_id": "test.response2"},
}
_template = "{{ merge_response(" + str(service_response) + ") }}"
with pytest.raises(
TemplateError,
match="ValueError: Response dictionary already contains key 'entity_id'",
):
template.Template(_template, hass).async_render()
service_response = {
"test.response": {
"happening": [
{
"start": "2024-02-27T17:00:00-06:00",
"end": "2024-02-27T18:00:00-06:00",
"summary": "Magic day",
"entity_id": "test.response",
}
]
}
}
_template = "{{ merge_response(" + str(service_response) + ") }}"
with pytest.raises(
TemplateError,
match="ValueError: Response dictionary already contains key 'entity_id'",
):
template.Template(_template, hass).async_render()
async def test_merge_response_with_empty_response(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
) -> None:
"""Test the merge_response function/filter with empty lists."""
service_response = {
"calendar.sports": {"events": []},
"calendar.local_furry_events": {"events": []},
"calendar.yap_house_schedules": {"events": []},
}
_template = "{{ merge_response(" + str(service_response) + ") }}"
tpl = template.Template(_template, hass)
assert service_response == snapshot(name="a_response")
assert tpl.async_render() == snapshot(name="b_rendered")
async def test_response_empty_dict(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
) -> None:
"""Test the merge_response function/filter with empty dict."""
service_response = {}
_template = "{{ merge_response(" + str(service_response) + ") }}"
tpl = template.Template(_template, hass)
assert tpl.async_render() == []
async def test_response_incorrect_value(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
) -> None:
"""Test the merge_response function/filter with incorrect response."""
service_response = "incorrect"
_template = "{{ merge_response(" + str(service_response) + ") }}"
with pytest.raises(TemplateError, match="TypeError: Response is not a dictionary"):
template.Template(_template, hass).async_render()
async def test_merge_response_with_incorrect_response(hass: HomeAssistant) -> None:
"""Test the merge_response function/filter with empty response should raise."""
service_response = {"calendar.sports": []}
_template = "{{ merge_response(" + str(service_response) + ") }}"
tpl = template.Template(_template, hass)
with pytest.raises(TemplateError, match="TypeError: Response is not a dictionary"):
tpl.async_render()
service_response = {
"binary_sensor.workday": [],
}
_template = "{{ merge_response(" + str(service_response) + ") }}"
tpl = template.Template(_template, hass)
with pytest.raises(TemplateError, match="TypeError: Response is not a dictionary"):
tpl.async_render()