mirror of https://github.com/home-assistant/core
1432 lines
42 KiB
Python
1432 lines
42 KiB
Python
"""Test the Switch config flow."""
|
|
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from pytest_unordered import unordered
|
|
|
|
from homeassistant import config_entries
|
|
from homeassistant.components.template import DOMAIN, async_setup_entry
|
|
from homeassistant.const import STATE_UNAVAILABLE
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.data_entry_flow import FlowResultType
|
|
from homeassistant.helpers import device_registry as dr
|
|
|
|
from tests.common import MockConfigEntry
|
|
from tests.typing import WebSocketGenerator
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"state_template",
|
|
"template_state",
|
|
"input_states",
|
|
"input_attributes",
|
|
"extra_input",
|
|
"extra_options",
|
|
"extra_attrs",
|
|
),
|
|
[
|
|
(
|
|
"alarm_control_panel",
|
|
{"value_template": "{{ states('alarm_control_panel.one') }}"},
|
|
"armed_away",
|
|
{"one": "armed_away", "two": "disarmed"},
|
|
{},
|
|
{},
|
|
{"code_arm_required": True, "code_format": "number"},
|
|
{},
|
|
),
|
|
(
|
|
"binary_sensor",
|
|
{
|
|
"state": "{{ states('binary_sensor.one') == 'on' or states('binary_sensor.two') == 'on' }}"
|
|
},
|
|
"on",
|
|
{"one": "on", "two": "off"},
|
|
{},
|
|
{},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"sensor",
|
|
{
|
|
"state": "{{ float(states('sensor.one')) + float(states('sensor.two')) }}"
|
|
},
|
|
"50.0",
|
|
{"one": "30.0", "two": "20.0"},
|
|
{},
|
|
{},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"button",
|
|
{},
|
|
"unknown",
|
|
{"one": "30.0", "two": "20.0"},
|
|
{},
|
|
{
|
|
"device_class": "restart",
|
|
"press": [
|
|
{
|
|
"action": "input_boolean.toggle",
|
|
"target": {"entity_id": "input_boolean.test"},
|
|
"data": {},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"device_class": "restart",
|
|
"press": [
|
|
{
|
|
"action": "input_boolean.toggle",
|
|
"target": {"entity_id": "input_boolean.test"},
|
|
"data": {},
|
|
}
|
|
],
|
|
},
|
|
{},
|
|
),
|
|
(
|
|
"image",
|
|
{"url": "{{ states('sensor.one') }}"},
|
|
"2024-07-09T00:00:00+00:00",
|
|
{"one": "http://www.test.com", "two": ""},
|
|
{},
|
|
{"verify_ssl": True},
|
|
{"verify_ssl": True},
|
|
{},
|
|
),
|
|
(
|
|
"number",
|
|
{"state": "{{ states('number.one') }}"},
|
|
"30.0",
|
|
{"one": "30.0", "two": "20.0"},
|
|
{},
|
|
{
|
|
"min": "0",
|
|
"max": "100",
|
|
"step": "0.1",
|
|
"unit_of_measurement": "cm",
|
|
"set_value": {
|
|
"action": "input_number.set_value",
|
|
"target": {"entity_id": "input_number.test"},
|
|
"data": {"value": "{{ value }}"},
|
|
},
|
|
},
|
|
{
|
|
"min": 0,
|
|
"max": 100,
|
|
"step": 0.1,
|
|
"unit_of_measurement": "cm",
|
|
"set_value": {
|
|
"action": "input_number.set_value",
|
|
"target": {"entity_id": "input_number.test"},
|
|
"data": {"value": "{{ value }}"},
|
|
},
|
|
},
|
|
{},
|
|
),
|
|
(
|
|
"select",
|
|
{"state": "{{ states('select.one') }}"},
|
|
"on",
|
|
{"one": "on", "two": "off"},
|
|
{},
|
|
{"options": "{{ ['off', 'on', 'auto'] }}"},
|
|
{"options": "{{ ['off', 'on', 'auto'] }}"},
|
|
{},
|
|
),
|
|
(
|
|
"switch",
|
|
{"value_template": "{{ states('switch.one') }}"},
|
|
"on",
|
|
{"one": "on", "two": "off"},
|
|
{},
|
|
{},
|
|
{},
|
|
{},
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.freeze_time("2024-07-09 00:00:00+00:00")
|
|
async def test_config_flow(
|
|
hass: HomeAssistant,
|
|
template_type: str,
|
|
state_template: dict[str, Any],
|
|
template_state: str,
|
|
input_states: dict[str, Any],
|
|
input_attributes: dict[str, Any],
|
|
extra_input: dict[str, Any],
|
|
extra_options: dict[str, Any],
|
|
extra_attrs: dict[str, Any],
|
|
) -> None:
|
|
"""Test the config flow."""
|
|
input_entities = ["one", "two"]
|
|
for input_entity in input_entities:
|
|
hass.states.async_set(
|
|
f"{template_type}.{input_entity}",
|
|
input_states[input_entity],
|
|
input_attributes.get(input_entity, {}),
|
|
)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": template_type},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
|
|
with patch(
|
|
"homeassistant.components.template.async_setup_entry", wraps=async_setup_entry
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
"name": "My template",
|
|
**state_template,
|
|
**extra_input,
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "My template"
|
|
assert result["data"] == {}
|
|
assert result["options"] == {
|
|
"name": "My template",
|
|
"template_type": template_type,
|
|
**state_template,
|
|
**extra_options,
|
|
}
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
|
|
assert config_entry.data == {}
|
|
assert config_entry.options == {
|
|
"name": "My template",
|
|
"template_type": template_type,
|
|
**state_template,
|
|
**extra_options,
|
|
}
|
|
|
|
state = hass.states.get(f"{template_type}.my_template")
|
|
assert state.state == template_state
|
|
for key, value in extra_attrs.items():
|
|
assert state.attributes[key] == value
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"state_template",
|
|
"extra_input",
|
|
"extra_options",
|
|
),
|
|
[
|
|
(
|
|
"sensor",
|
|
{"state": "{{ 15 }}"},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"binary_sensor",
|
|
{"state": "{{ false }}"},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"switch",
|
|
{"value_template": "{{ false }}"},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"button",
|
|
{},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"image",
|
|
{
|
|
"url": "{{ states('sensor.one') }}",
|
|
},
|
|
{"verify_ssl": True},
|
|
{"verify_ssl": True},
|
|
),
|
|
(
|
|
"number",
|
|
{"state": "{{ states('number.one') }}"},
|
|
{
|
|
"min": "0",
|
|
"max": "100",
|
|
"step": "0.1",
|
|
},
|
|
{
|
|
"min": 0,
|
|
"max": 100,
|
|
"step": 0.1,
|
|
},
|
|
),
|
|
(
|
|
"alarm_control_panel",
|
|
{"value_template": "{{ states('alarm_control_panel.one') }}"},
|
|
{"code_arm_required": True, "code_format": "number"},
|
|
{"code_arm_required": True, "code_format": "number"},
|
|
),
|
|
(
|
|
"select",
|
|
{"state": "{{ states('select.one') }}"},
|
|
{"options": "{{ ['off', 'on', 'auto'] }}"},
|
|
{"options": "{{ ['off', 'on', 'auto'] }}"},
|
|
),
|
|
],
|
|
)
|
|
async def test_config_flow_device(
|
|
hass: HomeAssistant,
|
|
template_type: str,
|
|
state_template: dict[str, Any],
|
|
extra_input: dict[str, Any],
|
|
extra_options: dict[str, Any],
|
|
device_registry: dr.DeviceRegistry,
|
|
) -> None:
|
|
"""Test remove the device registry configuration entry when the device changes."""
|
|
|
|
# Configure a device registry
|
|
entry_device = MockConfigEntry()
|
|
entry_device.add_to_hass(hass)
|
|
device = device_registry.async_get_or_create(
|
|
config_entry_id=entry_device.entry_id,
|
|
identifiers={("test", "identifier_test1")},
|
|
connections={("mac", "20:31:32:33:34:01")},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
device_id = device.id
|
|
assert device_id is not None
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": template_type},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
|
|
with patch(
|
|
"homeassistant.components.template.async_setup_entry", wraps=async_setup_entry
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
"name": "My template",
|
|
"device_id": device_id,
|
|
**state_template,
|
|
**extra_input,
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "My template"
|
|
assert result["data"] == {}
|
|
assert result["options"] == {
|
|
"name": "My template",
|
|
"template_type": template_type,
|
|
"device_id": device_id,
|
|
**state_template,
|
|
**extra_options,
|
|
}
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
|
|
assert config_entry.data == {}
|
|
assert config_entry.options == {
|
|
"name": "My template",
|
|
"template_type": template_type,
|
|
"device_id": device_id,
|
|
**state_template,
|
|
**extra_options,
|
|
}
|
|
|
|
|
|
def get_suggested(schema, key):
|
|
"""Get suggested value for key in voluptuous schema."""
|
|
for k in schema:
|
|
if k == key:
|
|
if k.description is None or "suggested_value" not in k.description:
|
|
return None
|
|
return k.description["suggested_value"]
|
|
# If the desired key is missing from the schema, return None
|
|
return None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"old_state_template",
|
|
"new_state_template",
|
|
"template_state",
|
|
"input_states",
|
|
"extra_options",
|
|
"options_options",
|
|
"key_template",
|
|
),
|
|
[
|
|
(
|
|
"binary_sensor",
|
|
{
|
|
"state": "{{ states('binary_sensor.one') == 'on' or states('binary_sensor.two') == 'on' }}"
|
|
},
|
|
{
|
|
"state": "{{ states('binary_sensor.one') == 'on' and states('binary_sensor.two') == 'on' }}"
|
|
},
|
|
["on", "off"],
|
|
{"one": "on", "two": "off"},
|
|
{},
|
|
{},
|
|
"state",
|
|
),
|
|
(
|
|
"sensor",
|
|
{
|
|
"state": "{{ float(states('sensor.one')) + float(states('sensor.two')) }}"
|
|
},
|
|
{
|
|
"state": "{{ float(states('sensor.one')) - float(states('sensor.two')) }}"
|
|
},
|
|
["50.0", "10.0"],
|
|
{"one": "30.0", "two": "20.0"},
|
|
{},
|
|
{},
|
|
"state",
|
|
),
|
|
(
|
|
"button",
|
|
{},
|
|
{},
|
|
["unknown", "unknown"],
|
|
{"one": "30.0", "two": "20.0"},
|
|
{
|
|
"device_class": "restart",
|
|
"press": [
|
|
{
|
|
"action": "input_boolean.toggle",
|
|
"target": {"entity_id": "input_boolean.test"},
|
|
"data": {},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"press": [
|
|
{
|
|
"action": "input_boolean.toggle",
|
|
"target": {"entity_id": "input_boolean.test"},
|
|
"data": {},
|
|
}
|
|
],
|
|
},
|
|
"state",
|
|
),
|
|
(
|
|
"image",
|
|
{
|
|
"url": "{{ states('sensor.one') }}",
|
|
},
|
|
{
|
|
"url": "{{ states('sensor.two') }}",
|
|
},
|
|
["2024-07-09T00:00:00+00:00", "2024-07-09T00:00:00+00:00"],
|
|
{"one": "http://www.test.com", "two": "http://www.test2.com"},
|
|
{"verify_ssl": True},
|
|
{
|
|
"url": "{{ states('sensor.two') }}",
|
|
"verify_ssl": True,
|
|
},
|
|
"url",
|
|
),
|
|
(
|
|
"number",
|
|
{"state": "{{ states('number.one') }}"},
|
|
{"state": "{{ states('number.two') }}"},
|
|
["30.0", "20.0"],
|
|
{"one": "30.0", "two": "20.0"},
|
|
{
|
|
"min": 0,
|
|
"max": 100,
|
|
"step": 0.1,
|
|
"unit_of_measurement": "cm",
|
|
"set_value": {
|
|
"action": "input_number.set_value",
|
|
"target": {"entity_id": "input_number.test"},
|
|
"data": {"value": "{{ value }}"},
|
|
},
|
|
},
|
|
{
|
|
"min": 0,
|
|
"max": 100,
|
|
"step": 0.1,
|
|
"unit_of_measurement": "cm",
|
|
"set_value": {
|
|
"action": "input_number.set_value",
|
|
"target": {"entity_id": "input_number.test"},
|
|
"data": {"value": "{{ value }}"},
|
|
},
|
|
},
|
|
"state",
|
|
),
|
|
(
|
|
"alarm_control_panel",
|
|
{"value_template": "{{ states('alarm_control_panel.one') }}"},
|
|
{"value_template": "{{ states('alarm_control_panel.two') }}"},
|
|
["armed_away", "disarmed"],
|
|
{"one": "armed_away", "two": "disarmed"},
|
|
{"code_arm_required": True, "code_format": "number"},
|
|
{"code_arm_required": True, "code_format": "number"},
|
|
"value_template",
|
|
),
|
|
(
|
|
"select",
|
|
{"state": "{{ states('select.one') }}"},
|
|
{"state": "{{ states('select.two') }}"},
|
|
["on", "off"],
|
|
{"one": "on", "two": "off"},
|
|
{"options": "{{ ['off', 'on', 'auto'] }}"},
|
|
{"options": "{{ ['off', 'on', 'auto'] }}"},
|
|
"state",
|
|
),
|
|
(
|
|
"switch",
|
|
{"value_template": "{{ states('switch.one') }}"},
|
|
{"value_template": "{{ states('switch.two') }}"},
|
|
["on", "off"],
|
|
{"one": "on", "two": "off"},
|
|
{},
|
|
{},
|
|
"value_template",
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.freeze_time("2024-07-09 00:00:00+00:00")
|
|
async def test_options(
|
|
hass: HomeAssistant,
|
|
template_type: str,
|
|
old_state_template: dict[str, Any],
|
|
new_state_template: dict[str, Any],
|
|
template_state: list[str],
|
|
input_states: dict[str, Any],
|
|
extra_options: dict[str, Any],
|
|
options_options: dict[str, Any],
|
|
key_template: str,
|
|
) -> None:
|
|
"""Test reconfiguring."""
|
|
input_entities = ["one", "two"]
|
|
|
|
for input_entity in input_entities:
|
|
hass.states.async_set(
|
|
f"{template_type}.{input_entity}", input_states[input_entity], {}
|
|
)
|
|
|
|
template_config_entry = MockConfigEntry(
|
|
data={},
|
|
domain=DOMAIN,
|
|
options={
|
|
"name": "My template",
|
|
"template_type": template_type,
|
|
**old_state_template,
|
|
**extra_options,
|
|
},
|
|
title="My template",
|
|
)
|
|
template_config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(template_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
state = hass.states.get(f"{template_type}.my_template")
|
|
assert state.state == template_state[0]
|
|
|
|
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
|
|
|
|
result = await hass.config_entries.options.async_init(config_entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
assert get_suggested(
|
|
result["data_schema"].schema, key_template
|
|
) == old_state_template.get(key_template)
|
|
assert "name" not in result["data_schema"].schema
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
**new_state_template,
|
|
**options_options,
|
|
},
|
|
)
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"] == {
|
|
"name": "My template",
|
|
"template_type": template_type,
|
|
**new_state_template,
|
|
**extra_options,
|
|
}
|
|
assert config_entry.data == {}
|
|
assert config_entry.options == {
|
|
"name": "My template",
|
|
"template_type": template_type,
|
|
**new_state_template,
|
|
**extra_options,
|
|
}
|
|
assert config_entry.title == "My template"
|
|
|
|
# Check config entry is reloaded with new options
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get(f"{template_type}.my_template")
|
|
assert state.state == template_state[1]
|
|
|
|
# Check we don't get suggestions from another entry
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": template_type},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
|
|
assert get_suggested(result["data_schema"].schema, "name") is None
|
|
assert get_suggested(result["data_schema"].schema, key_template) is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"state_template",
|
|
"extra_user_input",
|
|
"input_states",
|
|
"template_states",
|
|
"extra_attributes",
|
|
"listeners",
|
|
),
|
|
[
|
|
(
|
|
"binary_sensor",
|
|
"{{ states.binary_sensor.one.state == 'on' or states.binary_sensor.two.state == 'on' }}",
|
|
{},
|
|
{"one": "on", "two": "off"},
|
|
["off", "on"],
|
|
[{}, {}],
|
|
[["one", "two"], ["one"]],
|
|
),
|
|
(
|
|
"sensor",
|
|
"{{ float(states('sensor.one'), default='') + float(states('sensor.two'), default='') }}",
|
|
{},
|
|
{"one": "30.0", "two": "20.0"},
|
|
["", STATE_UNAVAILABLE, "50.0"],
|
|
[{}, {}],
|
|
[["one", "two"], ["one", "two"]],
|
|
),
|
|
],
|
|
)
|
|
async def test_config_flow_preview(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
template_type: str,
|
|
state_template: str,
|
|
extra_user_input: dict[str, Any],
|
|
input_states: dict[str, Any],
|
|
template_states: str,
|
|
extra_attributes: list[dict[str, Any]],
|
|
listeners: list[list[str]],
|
|
) -> None:
|
|
"""Test the config flow preview."""
|
|
client = await hass_ws_client(hass)
|
|
|
|
input_entities = ["one", "two"]
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": template_type},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
assert result["errors"] is None
|
|
assert result["preview"] == "template"
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "template/start_preview",
|
|
"flow_id": result["flow_id"],
|
|
"flow_type": "config_flow",
|
|
"user_input": {"name": "My template", "state": state_template}
|
|
| extra_user_input,
|
|
}
|
|
)
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] is None
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["event"] == {
|
|
"attributes": {"friendly_name": "My template"} | extra_attributes[0],
|
|
"listeners": {
|
|
"all": False,
|
|
"domains": [],
|
|
"entities": unordered([f"{template_type}.{_id}" for _id in listeners[0]]),
|
|
"time": False,
|
|
},
|
|
"state": template_states[0],
|
|
}
|
|
|
|
for input_entity in input_entities:
|
|
hass.states.async_set(
|
|
f"{template_type}.{input_entity}", input_states[input_entity], {}
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
for template_state in template_states[1:]:
|
|
msg = await client.receive_json()
|
|
assert msg["event"] == {
|
|
"attributes": {"friendly_name": "My template"}
|
|
| extra_attributes[0]
|
|
| extra_attributes[1],
|
|
"listeners": {
|
|
"all": False,
|
|
"domains": [],
|
|
"entities": unordered(
|
|
[f"{template_type}.{_id}" for _id in listeners[1]]
|
|
),
|
|
"time": False,
|
|
},
|
|
"state": template_state,
|
|
}
|
|
assert len(hass.states.async_all()) == 2
|
|
|
|
|
|
EARLY_END_ERROR = "invalid template (TemplateSyntaxError: unexpected 'end of template')"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("template_type", "state_template", "extra_user_input", "error"),
|
|
[
|
|
("binary_sensor", "{{", {}, {"state": EARLY_END_ERROR}),
|
|
("sensor", "{{", {}, {"state": EARLY_END_ERROR}),
|
|
(
|
|
"sensor",
|
|
"",
|
|
{"device_class": "aqi", "unit_of_measurement": "cats"},
|
|
{
|
|
"unit_of_measurement": (
|
|
"'cats' is not a valid unit for device class 'aqi'; "
|
|
"expected no unit of measurement"
|
|
),
|
|
},
|
|
),
|
|
(
|
|
"sensor",
|
|
"",
|
|
{"device_class": "temperature", "unit_of_measurement": "cats"},
|
|
{
|
|
"unit_of_measurement": (
|
|
"'cats' is not a valid unit for device class 'temperature'; "
|
|
"expected one of 'K', '°C', '°F'"
|
|
),
|
|
},
|
|
),
|
|
(
|
|
"sensor",
|
|
"",
|
|
{"device_class": "timestamp", "state_class": "measurement"},
|
|
{
|
|
"state_class": (
|
|
"'measurement' is not a valid state class for device class "
|
|
"'timestamp'; expected no state class"
|
|
),
|
|
},
|
|
),
|
|
(
|
|
"sensor",
|
|
"",
|
|
{"device_class": "aqi", "state_class": "total"},
|
|
{
|
|
"state_class": (
|
|
"'total' is not a valid state class for device class "
|
|
"'aqi'; expected 'measurement'"
|
|
),
|
|
},
|
|
),
|
|
(
|
|
"sensor",
|
|
"",
|
|
{"device_class": "energy", "state_class": "measurement"},
|
|
{
|
|
"state_class": (
|
|
"'measurement' is not a valid state class for device class "
|
|
"'energy'; expected one of 'total', 'total_increasing'"
|
|
),
|
|
"unit_of_measurement": (
|
|
"'None' is not a valid unit for device class 'energy'; "
|
|
"expected one of 'cal', 'Gcal', 'GJ', 'GWh', 'J', 'kcal', 'kJ', 'kWh', 'Mcal', 'MJ', 'MWh', 'TWh', 'Wh'"
|
|
),
|
|
},
|
|
),
|
|
],
|
|
)
|
|
async def test_config_flow_preview_bad_input(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
template_type: str,
|
|
state_template: str,
|
|
extra_user_input: dict[str, str],
|
|
error: dict[str, str],
|
|
) -> None:
|
|
"""Test the config flow preview."""
|
|
client = await hass_ws_client(hass)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": template_type},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
assert result["errors"] is None
|
|
assert result["preview"] == "template"
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "template/start_preview",
|
|
"flow_id": result["flow_id"],
|
|
"flow_type": "config_flow",
|
|
"user_input": {"name": "My template", "state": state_template}
|
|
| extra_user_input,
|
|
}
|
|
)
|
|
msg = await client.receive_json()
|
|
assert not msg["success"]
|
|
assert msg["error"] == {
|
|
"code": "invalid_user_input",
|
|
"message": error,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"state_template",
|
|
"input_states",
|
|
"template_states",
|
|
"error_events",
|
|
),
|
|
[
|
|
(
|
|
"sensor",
|
|
"{{ float(states('sensor.one')) + float(states('sensor.two')) }}",
|
|
{"one": "30.0", "two": "20.0"},
|
|
["unavailable", "50.0"],
|
|
[
|
|
(
|
|
"ValueError: Template error: float got invalid input 'unknown' "
|
|
"when rendering template '{{ float(states('sensor.one')) + "
|
|
"float(states('sensor.two')) }}' but no default was specified"
|
|
)
|
|
],
|
|
),
|
|
],
|
|
)
|
|
async def test_config_flow_preview_template_startup_error(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
template_type: str,
|
|
state_template: str,
|
|
input_states: dict[str, str],
|
|
template_states: list[str],
|
|
error_events: list[str],
|
|
) -> None:
|
|
"""Test the config flow preview."""
|
|
client = await hass_ws_client(hass)
|
|
|
|
input_entities = ["one", "two"]
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": template_type},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
assert result["errors"] is None
|
|
assert result["preview"] == "template"
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "template/start_preview",
|
|
"flow_id": result["flow_id"],
|
|
"flow_type": "config_flow",
|
|
"user_input": {"name": "My template", "state": state_template},
|
|
}
|
|
)
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "result"
|
|
assert msg["success"]
|
|
|
|
for error_event in error_events:
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "event"
|
|
assert msg["event"] == {"error": error_event}
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "event"
|
|
assert msg["event"]["state"] == template_states[0]
|
|
|
|
for input_entity in input_entities:
|
|
hass.states.async_set(
|
|
f"{template_type}.{input_entity}", input_states[input_entity], {}
|
|
)
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "event"
|
|
assert msg["event"]["state"] == template_states[1]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"state_template",
|
|
"input_states",
|
|
"template_states",
|
|
"error_events",
|
|
),
|
|
[
|
|
(
|
|
"sensor",
|
|
"{{ float(states('sensor.one')) > 30 and undefined_function() }}",
|
|
[{"one": "30.0", "two": "20.0"}, {"one": "35.0", "two": "20.0"}],
|
|
["False", "unavailable"],
|
|
["'undefined_function' is undefined"],
|
|
),
|
|
],
|
|
)
|
|
async def test_config_flow_preview_template_error(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
template_type: str,
|
|
state_template: str,
|
|
input_states: list[dict[str, str]],
|
|
template_states: list[str],
|
|
error_events: list[str],
|
|
) -> None:
|
|
"""Test the config flow preview."""
|
|
client = await hass_ws_client(hass)
|
|
|
|
input_entities = ["one", "two"]
|
|
|
|
for input_entity in input_entities:
|
|
hass.states.async_set(
|
|
f"{template_type}.{input_entity}", input_states[0][input_entity], {}
|
|
)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": template_type},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
assert result["errors"] is None
|
|
assert result["preview"] == "template"
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "template/start_preview",
|
|
"flow_id": result["flow_id"],
|
|
"flow_type": "config_flow",
|
|
"user_input": {"name": "My template", "state": state_template},
|
|
}
|
|
)
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "result"
|
|
assert msg["success"]
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "event"
|
|
assert msg["event"]["state"] == template_states[0]
|
|
|
|
for input_entity in input_entities:
|
|
hass.states.async_set(
|
|
f"{template_type}.{input_entity}", input_states[1][input_entity], {}
|
|
)
|
|
|
|
for error_event in error_events:
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "event"
|
|
assert msg["event"] == {"error": error_event}
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "event"
|
|
assert msg["event"]["state"] == template_states[1]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"state_template",
|
|
"extra_user_input",
|
|
),
|
|
[
|
|
(
|
|
"sensor",
|
|
"{{ states('sensor.one') }}",
|
|
{"unit_of_measurement": "°C"},
|
|
),
|
|
],
|
|
)
|
|
async def test_config_flow_preview_bad_state(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
template_type: str,
|
|
state_template: str,
|
|
extra_user_input: dict[str, Any],
|
|
) -> None:
|
|
"""Test the config flow preview."""
|
|
client = await hass_ws_client(hass)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": template_type},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == template_type
|
|
assert result["errors"] is None
|
|
assert result["preview"] == "template"
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "template/start_preview",
|
|
"flow_id": result["flow_id"],
|
|
"flow_type": "config_flow",
|
|
"user_input": {"name": "My template", "state": state_template}
|
|
| extra_user_input,
|
|
}
|
|
)
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] is None
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["event"] == {
|
|
"error": (
|
|
"Sensor None has device class 'None', state class 'None' unit '°C' "
|
|
"and suggested precision 'None' thus indicating it has a numeric "
|
|
"value; however, it has the non-numeric value: 'unknown' (<class "
|
|
"'str'>)"
|
|
),
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"old_state_template",
|
|
"new_state_template",
|
|
"extra_config_flow_data",
|
|
"extra_user_input",
|
|
"input_states",
|
|
"template_state",
|
|
"extra_attributes",
|
|
"listeners",
|
|
),
|
|
[
|
|
(
|
|
"binary_sensor",
|
|
"{{ states('binary_sensor.one') == 'on' or states('binary_sensor.two') == 'on' }}",
|
|
"{{ states('binary_sensor.one') == 'on' and states('binary_sensor.two') == 'on' }}",
|
|
{},
|
|
{},
|
|
{"one": "on", "two": "off"},
|
|
"off",
|
|
{},
|
|
["one", "two"],
|
|
),
|
|
(
|
|
"sensor",
|
|
"{{ float(states('sensor.one')) + float(states('sensor.two')) }}",
|
|
"{{ float(states('sensor.one')) - float(states('sensor.two')) }}",
|
|
{},
|
|
{},
|
|
{"one": "30.0", "two": "20.0"},
|
|
"10.0",
|
|
{},
|
|
["one", "two"],
|
|
),
|
|
],
|
|
)
|
|
async def test_option_flow_preview(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
template_type: str,
|
|
old_state_template: str,
|
|
new_state_template: str,
|
|
extra_config_flow_data: dict[str, Any],
|
|
extra_user_input: dict[str, Any],
|
|
input_states: dict[str, Any],
|
|
template_state: str,
|
|
extra_attributes: dict[str, Any],
|
|
listeners: list[str],
|
|
) -> None:
|
|
"""Test the option flow preview."""
|
|
client = await hass_ws_client(hass)
|
|
|
|
input_entities = ["one", "two"]
|
|
|
|
# Setup the config entry
|
|
config_entry = MockConfigEntry(
|
|
data={},
|
|
domain=DOMAIN,
|
|
options={
|
|
"name": "My template",
|
|
"state": old_state_template,
|
|
"template_type": template_type,
|
|
}
|
|
| extra_config_flow_data,
|
|
title="My template",
|
|
)
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
result = await hass.config_entries.options.async_init(config_entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["errors"] is None
|
|
assert result["preview"] == "template"
|
|
|
|
for input_entity in input_entities:
|
|
hass.states.async_set(
|
|
f"{template_type}.{input_entity}", input_states[input_entity], {}
|
|
)
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "template/start_preview",
|
|
"flow_id": result["flow_id"],
|
|
"flow_type": "options_flow",
|
|
"user_input": {"state": new_state_template} | extra_user_input,
|
|
}
|
|
)
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] is None
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["event"] == {
|
|
"attributes": {"friendly_name": "My template"} | extra_attributes,
|
|
"listeners": {
|
|
"all": False,
|
|
"domains": [],
|
|
"entities": unordered([f"{template_type}.{_id}" for _id in listeners]),
|
|
"time": False,
|
|
},
|
|
"state": template_state,
|
|
}
|
|
assert len(hass.states.async_all()) == 3
|
|
|
|
|
|
async def test_option_flow_sensor_preview_config_entry_removed(
|
|
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
|
|
) -> None:
|
|
"""Test the option flow preview where the config entry is removed."""
|
|
client = await hass_ws_client(hass)
|
|
|
|
# Setup the config entry
|
|
config_entry = MockConfigEntry(
|
|
data={},
|
|
domain=DOMAIN,
|
|
options={
|
|
"name": "My template",
|
|
"state": "Hello!",
|
|
"template_type": "sensor",
|
|
},
|
|
title="My template",
|
|
)
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
result = await hass.config_entries.options.async_init(config_entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["errors"] is None
|
|
assert result["preview"] == "template"
|
|
|
|
await hass.config_entries.async_remove(config_entry.entry_id)
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "template/start_preview",
|
|
"flow_id": result["flow_id"],
|
|
"flow_type": "options_flow",
|
|
"user_input": {"state": "Goodbye!"},
|
|
}
|
|
)
|
|
msg = await client.receive_json()
|
|
assert not msg["success"]
|
|
assert msg["error"] == {"code": "home_assistant_error", "message": "Unknown error"}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"template_type",
|
|
"state_template",
|
|
"extra_input",
|
|
"extra_options",
|
|
),
|
|
[
|
|
(
|
|
"sensor",
|
|
{"state": "{{ 15 }}"},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"binary_sensor",
|
|
{"state": "{{ false }}"},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"button",
|
|
{},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"image",
|
|
{
|
|
"url": "{{ states('sensor.one') }}",
|
|
"verify_ssl": True,
|
|
},
|
|
{},
|
|
{},
|
|
),
|
|
(
|
|
"number",
|
|
{"state": "{{ states('number.one') }}"},
|
|
{
|
|
"min": 0,
|
|
"max": 100,
|
|
"step": 0.1,
|
|
},
|
|
{
|
|
"min": 0,
|
|
"max": 100,
|
|
"step": 0.1,
|
|
},
|
|
),
|
|
(
|
|
"alarm_control_panel",
|
|
{"value_template": "{{ states('alarm_control_panel.one') }}"},
|
|
{"code_arm_required": True, "code_format": "number"},
|
|
{"code_arm_required": True, "code_format": "number"},
|
|
),
|
|
(
|
|
"select",
|
|
{"state": "{{ states('select.one') }}"},
|
|
{"options": "{{ ['off', 'on', 'auto'] }}"},
|
|
{"options": "{{ ['off', 'on', 'auto'] }}"},
|
|
),
|
|
(
|
|
"switch",
|
|
{"value_template": "{{ false }}"},
|
|
{},
|
|
{},
|
|
),
|
|
],
|
|
)
|
|
async def test_options_flow_change_device(
|
|
hass: HomeAssistant,
|
|
template_type: str,
|
|
state_template: dict[str, Any],
|
|
extra_input: dict[str, Any],
|
|
extra_options: dict[str, Any],
|
|
device_registry: dr.DeviceRegistry,
|
|
) -> None:
|
|
"""Test remove the device registry configuration entry when the device changes."""
|
|
|
|
# Configure a device registry
|
|
entry_device1 = MockConfigEntry()
|
|
entry_device1.add_to_hass(hass)
|
|
device1 = device_registry.async_get_or_create(
|
|
config_entry_id=entry_device1.entry_id,
|
|
identifiers={("test", "identifier_test1")},
|
|
connections={("mac", "20:31:32:33:34:01")},
|
|
)
|
|
entry_device2 = MockConfigEntry()
|
|
entry_device2.add_to_hass(hass)
|
|
device2 = device_registry.async_get_or_create(
|
|
config_entry_id=entry_device1.entry_id,
|
|
identifiers={("test", "identifier_test2")},
|
|
connections={("mac", "20:31:32:33:34:02")},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
device_id1 = device1.id
|
|
assert device_id1 is not None
|
|
|
|
device_id2 = device2.id
|
|
assert device_id2 is not None
|
|
|
|
# Setup the config entry with device 1
|
|
template_config_entry = MockConfigEntry(
|
|
data={},
|
|
domain=DOMAIN,
|
|
options={
|
|
"template_type": template_type,
|
|
"name": "My template",
|
|
"device_id": device_id1,
|
|
**state_template,
|
|
**extra_options,
|
|
},
|
|
title="Template",
|
|
)
|
|
template_config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(template_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
# Change to link to device 2
|
|
result = await hass.config_entries.options.async_init(
|
|
template_config_entry.entry_id
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
"device_id": device_id2,
|
|
**state_template,
|
|
**extra_input,
|
|
},
|
|
)
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"] == {
|
|
"template_type": template_type,
|
|
"name": "My template",
|
|
"device_id": device_id2,
|
|
**state_template,
|
|
**extra_input,
|
|
}
|
|
assert template_config_entry.data == {}
|
|
assert template_config_entry.options == {
|
|
"template_type": template_type,
|
|
"name": "My template",
|
|
"device_id": device_id2,
|
|
**state_template,
|
|
**extra_options,
|
|
}
|
|
|
|
# Remove link with device
|
|
result = await hass.config_entries.options.async_init(
|
|
template_config_entry.entry_id
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
**state_template,
|
|
**extra_input,
|
|
},
|
|
)
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"] == {
|
|
"template_type": template_type,
|
|
"name": "My template",
|
|
**state_template,
|
|
**extra_input,
|
|
}
|
|
assert template_config_entry.data == {}
|
|
assert template_config_entry.options == {
|
|
"template_type": template_type,
|
|
"name": "My template",
|
|
**state_template,
|
|
**extra_options,
|
|
}
|
|
|
|
# Change to link to device 1
|
|
result = await hass.config_entries.options.async_init(
|
|
template_config_entry.entry_id
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
"device_id": device_id1,
|
|
**state_template,
|
|
**extra_input,
|
|
},
|
|
)
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"] == {
|
|
"template_type": template_type,
|
|
"name": "My template",
|
|
"device_id": device_id1,
|
|
**state_template,
|
|
**extra_input,
|
|
}
|
|
assert template_config_entry.data == {}
|
|
assert template_config_entry.options == {
|
|
"template_type": template_type,
|
|
"name": "My template",
|
|
"device_id": device_id1,
|
|
**state_template,
|
|
**extra_options,
|
|
}
|