mirror of
https://github.com/esphome/esphome.git
synced 2026-10-08 12:04:03 +00:00
943 lines
28 KiB
Python
943 lines
28 KiB
Python
"""Tests for the api wizard: schema, final validation and generated tables."""
|
|
|
|
from collections.abc import Callable
|
|
import json
|
|
from pathlib import Path
|
|
import random
|
|
import re
|
|
import string
|
|
import textwrap
|
|
|
|
import pytest
|
|
|
|
from esphome.components.api import wizard
|
|
from esphome.components.homeassistant.switch import SUPPORTED_DOMAINS as SWITCH_DOMAINS
|
|
from esphome.config import load_config
|
|
from esphome.core import CORE
|
|
from esphome.helpers import fnv1_hash, fnv1a_32bit_hash
|
|
from tests.component_tests.helpers import get_define_value
|
|
|
|
ESP32_HEADER = """
|
|
esphome:
|
|
name: test
|
|
|
|
esp32:
|
|
variant: esp32
|
|
|
|
wifi:
|
|
ssid: test
|
|
password: testtest
|
|
|
|
logger:
|
|
"""
|
|
|
|
ESP8266_HEADER = """
|
|
esphome:
|
|
name: test
|
|
|
|
esp8266:
|
|
board: d1_mini
|
|
|
|
wifi:
|
|
ssid: test
|
|
password: testtest
|
|
|
|
logger:
|
|
"""
|
|
|
|
ENTITIES = """
|
|
switch:
|
|
- platform: template
|
|
id: sw
|
|
name: Switch
|
|
optimistic: true
|
|
- platform: template
|
|
id: internal_sw
|
|
name: Internal switch
|
|
internal: true
|
|
optimistic: true
|
|
- platform: template
|
|
id: unnamed_sw
|
|
optimistic: true
|
|
- platform: homeassistant
|
|
id: ha_switch
|
|
%(light)s
|
|
sensor:
|
|
- platform: template
|
|
id: template_sensor
|
|
name: Template sensor
|
|
- platform: homeassistant
|
|
id: ha_plain_sensor
|
|
entity_id: sensor.plain
|
|
- platform: homeassistant
|
|
id: ha_sensor
|
|
%(sensor)s
|
|
binary_sensor:
|
|
- platform: homeassistant
|
|
id: ha_binary
|
|
%(binary)s
|
|
text_sensor:
|
|
- platform: homeassistant
|
|
id: ha_text
|
|
%(text)s
|
|
number:
|
|
- platform: homeassistant
|
|
id: ha_number
|
|
%(number)s
|
|
text:
|
|
- platform: homeassistant
|
|
id: ha_txt
|
|
%(text_entity)s
|
|
select:
|
|
- platform: homeassistant
|
|
id: ha_select
|
|
%(select)s
|
|
button:
|
|
- platform: homeassistant
|
|
id: ha_button
|
|
%(button)s
|
|
"""
|
|
|
|
|
|
def build_entities(inputs: set[str]) -> str:
|
|
"""The test entities. The homeassistant ones that are wizard inputs set no entity_id, the others do."""
|
|
ids = {
|
|
"light": ("ha_switch", "entity_id: light.lamp"),
|
|
"sensor": ("ha_sensor", "entity_id: sensor.a"),
|
|
"binary": ("ha_binary", "entity_id: binary_sensor.a"),
|
|
"text": ("ha_text", "entity_id: sensor.b"),
|
|
"number": ("ha_number", "entity_id: number.a"),
|
|
"text_entity": ("ha_txt", "entity_id: text.a"),
|
|
"select": ("ha_select", "entity_id: select.a"),
|
|
"button": ("ha_button", "entity_id: button.a"),
|
|
}
|
|
return ENTITIES % {
|
|
key: "" if name in inputs else yaml for key, (name, yaml) in ids.items()
|
|
}
|
|
|
|
|
|
GOOD_PAGES = """
|
|
api:
|
|
wizard:
|
|
pages:
|
|
- title: Setup
|
|
description: "[%key:component::domain::section::name%]"
|
|
entities:
|
|
- id: sw
|
|
description: Enable
|
|
- id: sw
|
|
- inputs:
|
|
- entity: ha_sensor
|
|
description: Weather
|
|
target:
|
|
entity:
|
|
- integration: met
|
|
domain: sensor
|
|
device_class: [temperature, humidity]
|
|
supported_features: weather.WeatherEntityFeature.FORECAST_DAILY
|
|
- domain: [sensor, number]
|
|
- entity: ha_binary
|
|
- entity: ha_text
|
|
- entity: ha_number
|
|
- entity: ha_switch
|
|
- entity: ha_txt
|
|
- entity: ha_select
|
|
- entity: ha_button
|
|
- id: weather_input
|
|
target:
|
|
entity:
|
|
- domain: weather
|
|
- id: unset_input
|
|
"""
|
|
|
|
|
|
def write_config(
|
|
tmp_path: Path, header: str, api: str, yaml_entities: str | None = None
|
|
) -> Path:
|
|
"""Write a config. By default the homeassistant entities that the wizard does not use as inputs set an entity_id."""
|
|
path = tmp_path / "test.yaml"
|
|
if yaml_entities is None:
|
|
yaml_entities = build_entities(set(re.findall(r"entity: (\w+)", api)))
|
|
path.write_text(
|
|
textwrap.dedent(header) + textwrap.dedent(yaml_entities) + textwrap.dedent(api)
|
|
)
|
|
return path
|
|
|
|
|
|
def write_input_config(tmp_path: Path, header: str) -> Path:
|
|
"""Write a config with GOOD_PAGES."""
|
|
return write_config(tmp_path, header, GOOD_PAGES)
|
|
|
|
|
|
def config_errors(path: Path) -> list[str]:
|
|
CORE.config_path = path
|
|
return [str(err.msg) for err in load_config({}).errors]
|
|
|
|
|
|
def wizard_api(pages: str) -> str:
|
|
return "api:\n wizard:\n pages:\n" + textwrap.indent(
|
|
textwrap.dedent(pages).strip("\n"), " "
|
|
)
|
|
|
|
|
|
def test_no_wizard_emits_nothing(
|
|
tmp_path: Path, generate_main: Callable[[str | Path], str]
|
|
) -> None:
|
|
main_cpp = generate_main(
|
|
write_config(
|
|
tmp_path,
|
|
ESP32_HEADER,
|
|
"api:\n",
|
|
"sensor:\n - platform: homeassistant\n id: s\n entity_id: sensor.s\n",
|
|
)
|
|
)
|
|
|
|
assert "API_WIZARD_PAGES" not in main_cpp
|
|
assert 's->set_entity_id("sensor.s");' in main_cpp
|
|
assert not any(define.name == "USE_API_WIZARD" for define in CORE.defines)
|
|
|
|
|
|
def test_valid_wizard_has_no_errors(tmp_path: Path) -> None:
|
|
assert config_errors(write_input_config(tmp_path, ESP32_HEADER)) == []
|
|
|
|
|
|
def wizard_inputs(inputs: str) -> str:
|
|
return wizard_api("- inputs:\n" + textwrap.indent(textwrap.dedent(inputs), " "))
|
|
|
|
|
|
def test_switch_input_with_supported_domains_is_valid(tmp_path: Path) -> None:
|
|
api = wizard_inputs(
|
|
"""
|
|
- entity: ha_switch
|
|
target:
|
|
entity:
|
|
- domain: [light, fan]
|
|
"""
|
|
)
|
|
|
|
assert config_errors(write_config(tmp_path, ESP32_HEADER, api)) == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("api", "message"),
|
|
[
|
|
pytest.param(
|
|
wizard_inputs("- entity: template_sensor"),
|
|
"must be a homeassistant",
|
|
id="input-not-homeassistant",
|
|
),
|
|
pytest.param(
|
|
wizard_inputs("- entity: sw"),
|
|
"must be a homeassistant",
|
|
id="input-not-homeassistant-switch",
|
|
),
|
|
pytest.param(
|
|
wizard_inputs(
|
|
"""
|
|
- entity: ha_switch
|
|
target:
|
|
entity:
|
|
- domain: [light, valve]
|
|
"""
|
|
),
|
|
"does not support the domain(s) valve",
|
|
id="switch-unsupported-domain",
|
|
),
|
|
pytest.param(
|
|
wizard_inputs(
|
|
"""
|
|
- entity: ha_switch
|
|
target:
|
|
entity:
|
|
- integration: hue
|
|
"""
|
|
),
|
|
"must set domain",
|
|
id="switch-filter-without-domain",
|
|
),
|
|
pytest.param(
|
|
wizard_inputs("- entity: ha_sensor\n- entity: ha_sensor"),
|
|
"'ha_sensor' is used more than once",
|
|
id="duplicate-input",
|
|
),
|
|
],
|
|
)
|
|
def test_invalid_inputs_are_rejected(tmp_path: Path, api: str, message: str) -> None:
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, api))
|
|
|
|
assert any(message in error for error in errors), errors
|
|
|
|
|
|
def test_hash_collision_is_rejected(tmp_path: Path) -> None:
|
|
entities = "".join(
|
|
f"sensor:\n - platform: homeassistant\n id: {name}\n"
|
|
if index == 0
|
|
else f" - platform: homeassistant\n id: {name}\n"
|
|
for index, name in enumerate(("a179599", "a362382"))
|
|
)
|
|
api = wizard_inputs("- entity: a179599\n- entity: a362382")
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, api, entities))
|
|
|
|
assert any("have the same hash" in error for error in errors), errors
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"extra",
|
|
[
|
|
"",
|
|
" entity_id: sensor.default\n",
|
|
],
|
|
)
|
|
def test_entity_id_is_optional_only_for_wizard_inputs(
|
|
tmp_path: Path, extra: str
|
|
) -> None:
|
|
entities = f"sensor:\n - platform: homeassistant\n id: lonely\n{extra}"
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, "api:\n", entities))
|
|
|
|
if extra:
|
|
assert errors == []
|
|
else:
|
|
assert any(
|
|
"entity_id is required unless this entity is a wizard input" in error
|
|
for error in errors
|
|
), errors
|
|
|
|
|
|
def test_entity_id_is_optional_for_an_input_and_its_domain_is_checked(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
entities = "switch:\n - platform: homeassistant\n id: ha_switch\n entity_id: climate.heater\n"
|
|
errors = config_errors(
|
|
write_config(
|
|
tmp_path, ESP32_HEADER, wizard_inputs("- entity: ha_switch"), entities
|
|
)
|
|
)
|
|
|
|
# The YAML entity ID is still checked against the platform's domains
|
|
assert any("not supported by the switch platform" in error for error in errors)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("pages", "message"),
|
|
[
|
|
pytest.param(
|
|
"- title: Empty",
|
|
"at least one of entities, inputs",
|
|
id="page-without-fields",
|
|
),
|
|
pytest.param(
|
|
"""
|
|
- entities: []
|
|
""",
|
|
"length of value must be at least 1",
|
|
id="empty-entities-list",
|
|
),
|
|
pytest.param(
|
|
"""
|
|
- entities: [{id: internal_sw}]
|
|
""",
|
|
"Entity 'internal_sw' is internal",
|
|
id="internal-entity",
|
|
),
|
|
pytest.param(
|
|
"""
|
|
- entities: [{id: unnamed_sw}]
|
|
""",
|
|
"Entity 'unnamed_sw' is internal",
|
|
id="unnamed-entity-is-internal",
|
|
),
|
|
pytest.param(
|
|
"""
|
|
- inputs:
|
|
- entity: ha_sensor
|
|
target:
|
|
entity:
|
|
- {}
|
|
""",
|
|
"at least one of",
|
|
id="empty-filter",
|
|
),
|
|
pytest.param(
|
|
"""
|
|
- inputs:
|
|
- entity: ha_sensor
|
|
target:
|
|
entity: []
|
|
""",
|
|
"length of value must be at least 1",
|
|
id="no-filters",
|
|
),
|
|
pytest.param(
|
|
"""
|
|
- inputs:
|
|
- entity: ha_sensor
|
|
target:
|
|
entity:
|
|
- domain: []
|
|
""",
|
|
"length of value must be at least 1",
|
|
id="empty-domain-list",
|
|
),
|
|
pytest.param(
|
|
"""
|
|
- inputs:
|
|
- entity: ha_sensor
|
|
target:
|
|
entity:
|
|
- supported_features: FEATURE
|
|
""".replace("FEATURE", "x" * 128),
|
|
"length of value must be at most 127",
|
|
id="long-feature",
|
|
),
|
|
pytest.param(
|
|
"""
|
|
- description: TEXT
|
|
entities: [{id: sw}]
|
|
""".replace("TEXT", "x" * 256),
|
|
"length of value must be at most 255",
|
|
id="long-description",
|
|
),
|
|
],
|
|
)
|
|
def test_invalid_wizards_are_rejected(tmp_path: Path, pages: str, message: str) -> None:
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, wizard_api(pages)))
|
|
|
|
assert any(message in error for error in errors), errors
|
|
|
|
|
|
def test_no_pages_is_rejected(tmp_path: Path) -> None:
|
|
errors = config_errors(
|
|
write_config(tmp_path, ESP32_HEADER, "api:\n wizard:\n pages: []\n")
|
|
)
|
|
|
|
assert any("length of value must be at least 1" in error for error in errors), (
|
|
errors
|
|
)
|
|
|
|
|
|
def test_input_is_set_condition(
|
|
tmp_path: Path, generate_main: Callable[[str | Path], str]
|
|
) -> None:
|
|
interval = """
|
|
interval:
|
|
- interval: 1h
|
|
then:
|
|
- if:
|
|
condition:
|
|
api.wizard.input_is_set: weather_input
|
|
then:
|
|
- logger.log: set
|
|
- if:
|
|
condition:
|
|
api.wizard.input_is_set:
|
|
id: unset_input
|
|
then:
|
|
- logger.log: set
|
|
"""
|
|
path = write_input_config(tmp_path, ESP32_HEADER)
|
|
path.write_text(path.read_text() + textwrap.dedent(interval))
|
|
main_cpp = generate_main(path)
|
|
|
|
assert "weather_input->has_entity_id()" in main_cpp
|
|
assert "unset_input->has_entity_id()" in main_cpp
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("inputs", "message"),
|
|
[
|
|
pytest.param(
|
|
"- description: Neither",
|
|
"exactly one of id, entity",
|
|
id="neither-id-nor-entity",
|
|
),
|
|
pytest.param(
|
|
"- id: standalone\n entity: ha_sensor",
|
|
"more than one of id, entity",
|
|
id="both-id-and-entity",
|
|
),
|
|
pytest.param(
|
|
"- entity: ha_sensor\n entity_id: sensor.a",
|
|
"extra keys not allowed",
|
|
id="entity-id-on-linked-input",
|
|
),
|
|
pytest.param(
|
|
"- id: standalone\n entity_id: weather.home",
|
|
"extra keys not allowed",
|
|
id="entity-id-on-standalone-input",
|
|
),
|
|
pytest.param(
|
|
"- entity: ha_sensor\n- entity: ha_sensor",
|
|
"used more than once",
|
|
id="duplicate-linked-input",
|
|
),
|
|
],
|
|
)
|
|
def test_standalone_and_linked_input_rules(
|
|
tmp_path: Path, inputs: str, message: str
|
|
) -> None:
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, wizard_inputs(inputs)))
|
|
|
|
assert any(message in error for error in errors), errors
|
|
|
|
|
|
def test_standalone_input_is_valid(tmp_path: Path) -> None:
|
|
api = wizard_inputs("- id: standalone")
|
|
|
|
assert config_errors(write_config(tmp_path, ESP32_HEADER, api)) == []
|
|
|
|
|
|
WIZARD_DEFINES = {
|
|
"USE_API_WIZARD",
|
|
"USE_API_WIZARD_INPUTS",
|
|
"USE_API_WIZARD_LINKED_INPUTS",
|
|
"USE_API_WIZARD_STANDALONE_INPUTS",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("pages", "expected"),
|
|
[
|
|
pytest.param(
|
|
"- entities: [{id: sw}]",
|
|
{"USE_API_WIZARD"},
|
|
id="entity-only",
|
|
),
|
|
pytest.param(
|
|
"- inputs: [{id: standalone}]",
|
|
{
|
|
"USE_API_WIZARD",
|
|
"USE_API_WIZARD_INPUTS",
|
|
"USE_API_WIZARD_STANDALONE_INPUTS",
|
|
},
|
|
id="standalone-input-without-filter",
|
|
),
|
|
pytest.param(
|
|
"- inputs: [{id: standalone, target: {entity: [{domain: weather}]}}]",
|
|
{
|
|
"USE_API_WIZARD",
|
|
"USE_API_WIZARD_INPUTS",
|
|
"USE_API_WIZARD_STANDALONE_INPUTS",
|
|
},
|
|
id="standalone-input-with-filter",
|
|
),
|
|
pytest.param(
|
|
"- inputs: [{entity: ha_sensor}]",
|
|
{
|
|
"USE_API_WIZARD",
|
|
"USE_API_WIZARD_INPUTS",
|
|
"USE_API_WIZARD_LINKED_INPUTS",
|
|
},
|
|
id="linked-sensor-without-filter",
|
|
),
|
|
pytest.param(
|
|
"- inputs: [{entity: ha_sensor, target: {entity: [{domain: sensor}]}}]",
|
|
{
|
|
"USE_API_WIZARD",
|
|
"USE_API_WIZARD_INPUTS",
|
|
"USE_API_WIZARD_LINKED_INPUTS",
|
|
},
|
|
id="linked-sensor-with-filter",
|
|
),
|
|
pytest.param(
|
|
# A switch without a target gets the default filter
|
|
"- inputs: [{entity: ha_switch}]",
|
|
{
|
|
"USE_API_WIZARD",
|
|
"USE_API_WIZARD_INPUTS",
|
|
"USE_API_WIZARD_LINKED_INPUTS",
|
|
},
|
|
id="linked-switch-default-filter",
|
|
),
|
|
pytest.param(
|
|
"- entities: [{id: sw}]\n inputs: [{entity: ha_sensor, target: {entity: [{domain: sensor}]}}, {id: standalone}]",
|
|
WIZARD_DEFINES,
|
|
id="everything",
|
|
),
|
|
],
|
|
)
|
|
def test_only_the_defines_the_wizard_needs_are_emitted(
|
|
tmp_path: Path,
|
|
generate_main: Callable[[str | Path], str],
|
|
pages: str,
|
|
expected: set[str],
|
|
) -> None:
|
|
api = wizard_api(pages)
|
|
generate_main(write_config(tmp_path, ESP32_HEADER, api))
|
|
|
|
assert {d.name for d in CORE.defines} & WIZARD_DEFINES == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("entity", "bad_domain"),
|
|
[
|
|
("ha_txt", "select"),
|
|
("ha_select", "text"),
|
|
("ha_button", "switch"),
|
|
("ha_number", "sensor"),
|
|
],
|
|
)
|
|
def test_text_select_and_button_inputs_keep_to_their_domains(
|
|
tmp_path: Path, entity: str, bad_domain: str
|
|
) -> None:
|
|
api = wizard_inputs(
|
|
f"- entity: {entity}\n target:\n entity:\n - domain: {bad_domain}"
|
|
)
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, api))
|
|
|
|
assert any(
|
|
f"does not support the domain(s) {bad_domain}" in error for error in errors
|
|
), errors
|
|
|
|
|
|
@pytest.mark.parametrize("entity", ["ha_txt", "ha_select", "ha_button", "ha_number"])
|
|
def test_text_select_and_button_filters_must_set_a_domain(
|
|
tmp_path: Path, entity: str
|
|
) -> None:
|
|
api = wizard_inputs(
|
|
f"- entity: {entity}\n target:\n entity:\n - integration: hue"
|
|
)
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, api))
|
|
|
|
assert any("must set domain" in error for error in errors), errors
|
|
|
|
|
|
def test_number_input_accepts_input_number(tmp_path: Path) -> None:
|
|
api = wizard_inputs(
|
|
"- entity: ha_number\n target:\n entity:\n - domain: input_number"
|
|
)
|
|
|
|
assert config_errors(write_config(tmp_path, ESP32_HEADER, api)) == []
|
|
|
|
|
|
def test_text_select_and_button_default_to_their_domains(
|
|
tmp_path: Path, generate_main: Callable[[str | Path], str]
|
|
) -> None:
|
|
main_cpp = generate_main(write_input_config(tmp_path, ESP32_HEADER))
|
|
document = json.loads(wizard.zstd_module().decompress(blob_in(main_cpp)))
|
|
filters = {
|
|
entry["key"]: entry.get("entity_filters")
|
|
for page in document["pages"]
|
|
for entry in page.get("inputs", [])
|
|
}
|
|
|
|
# ha_txt and ha_button have no target, so they take every supported domain
|
|
assert filters[fnv1_hash("ha_txt")] == [{"domain": ["input_text", "text"]}]
|
|
assert filters[fnv1_hash("ha_button")] == [{"domain": ["button", "input_button"]}]
|
|
assert filters[fnv1_hash("ha_select")] == [{"domain": ["input_select", "select"]}]
|
|
# A number takes number and input_number entities, and the other sensors take anything
|
|
assert filters[fnv1_hash("ha_number")] == [{"domain": ["input_number", "number"]}]
|
|
assert filters[fnv1_hash("ha_binary")] is None
|
|
for entity in ("ha_txt", "ha_select", "ha_button"):
|
|
assert f"{entity}->set_entity_id(api_wizard_input_{entity});" in main_cpp
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"entity_yaml",
|
|
[
|
|
"text:\n - platform: homeassistant\n id: lonely\n",
|
|
"select:\n - platform: homeassistant\n id: lonely\n",
|
|
"button:\n - platform: homeassistant\n id: lonely\n",
|
|
],
|
|
)
|
|
def test_new_platforms_need_an_entity_id_unless_they_are_inputs(
|
|
tmp_path: Path, entity_yaml: str
|
|
) -> None:
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, "api:\n", entity_yaml))
|
|
|
|
assert any(
|
|
"entity_id is required unless this entity is a wizard input" in error
|
|
for error in errors
|
|
), errors
|
|
|
|
|
|
DEVICES_HEADER = """
|
|
esphome:
|
|
name: test
|
|
devices:
|
|
- id: kitchen_dev
|
|
name: Kitchen
|
|
|
|
esp32:
|
|
variant: esp32
|
|
|
|
wifi:
|
|
ssid: test
|
|
password: testtest
|
|
|
|
logger:
|
|
"""
|
|
|
|
JSON_ENTITIES = """
|
|
switch:
|
|
- platform: template
|
|
id: sw
|
|
name: Switch
|
|
optimistic: true
|
|
- platform: template
|
|
id: sw2
|
|
name: Kitchen Switch
|
|
device_id: kitchen_dev
|
|
optimistic: true
|
|
- platform: homeassistant
|
|
id: ha_switch
|
|
sensor:
|
|
- platform: homeassistant
|
|
id: ha_sensor
|
|
"""
|
|
|
|
JSON_PAGES = """
|
|
- title: Audio
|
|
description: Pick
|
|
entities:
|
|
- id: sw
|
|
description: Enable
|
|
- id: sw2
|
|
inputs:
|
|
- id: weather
|
|
description: Weather
|
|
target:
|
|
entity:
|
|
- integration: met
|
|
domain: weather
|
|
device_class: [temperature, humidity]
|
|
- domain: [weather, sensor]
|
|
- entity: ha_switch
|
|
- inputs: [{entity: ha_sensor}]
|
|
"""
|
|
|
|
|
|
def blob_in(main_cpp: str) -> bytes:
|
|
"""The compressed wizard that the generated code puts in flash."""
|
|
match = re.search(
|
|
r"const uint8_t esphome::api::API_WIZARD_DATA\[\] PROGMEM = \{([^}]*)\};",
|
|
main_cpp,
|
|
)
|
|
assert match is not None
|
|
return bytes(int(byte) for byte in match.group(1).split(", "))
|
|
|
|
|
|
def entity_hash(main_cpp: str, variable: str) -> int:
|
|
"""The key the generated code passes to configure_entity_, which ListEntities then sends."""
|
|
match = re.search(
|
|
rf'App\.register_switch\({variable}, "[^"]*", (\d+)UL, \d+\)', main_cpp
|
|
)
|
|
assert match is not None, variable
|
|
return int(match.group(1))
|
|
|
|
|
|
def test_the_blob_is_the_exact_json_document(
|
|
tmp_path: Path, generate_main: Callable[[str | Path], str]
|
|
) -> None:
|
|
main_cpp = generate_main(
|
|
write_config(tmp_path, DEVICES_HEADER, wizard_api(JSON_PAGES), JSON_ENTITIES)
|
|
)
|
|
blob = blob_in(main_cpp)
|
|
|
|
text = wizard.zstd_module().decompress(blob).decode("utf-8")
|
|
expected = {
|
|
"version": 1,
|
|
"pages": [
|
|
{
|
|
"title": "Audio",
|
|
"description": "Pick",
|
|
"entities": [
|
|
# No device_id for the main device, and no description when unset
|
|
{"key": entity_hash(main_cpp, "sw"), "description": "Enable"},
|
|
{
|
|
"key": entity_hash(main_cpp, "sw2"),
|
|
"device_id": fnv1a_32bit_hash("kitchen_dev"),
|
|
},
|
|
],
|
|
"inputs": [
|
|
{
|
|
"key": fnv1_hash("weather"),
|
|
"description": "Weather",
|
|
"entity_filters": [
|
|
{
|
|
"integration": "met",
|
|
"domain": ["weather"],
|
|
"device_class": ["temperature", "humidity"],
|
|
},
|
|
{"domain": ["weather", "sensor"]},
|
|
],
|
|
},
|
|
{
|
|
"key": fnv1_hash("ha_switch"),
|
|
# The default filter of a switch
|
|
"entity_filters": [{"domain": SWITCH_DOMAINS}],
|
|
},
|
|
],
|
|
},
|
|
{"inputs": [{"key": fnv1_hash("ha_sensor")}]},
|
|
],
|
|
}
|
|
assert text == json.dumps(
|
|
expected, separators=(",", ":"), sort_keys=True, ensure_ascii=False
|
|
)
|
|
# The device id is the one the generated code gives the device
|
|
assert f"set_device_id({fnv1a_32bit_hash('kitchen_dev')})" in main_cpp
|
|
assert get_define_value("API_WIZARD_DATA_SIZE") == str(len(blob))
|
|
|
|
|
|
def test_the_blob_is_deterministic_and_the_same_on_every_platform(
|
|
tmp_path: Path, generate_main: Callable[[str | Path], str]
|
|
) -> None:
|
|
api = wizard_api(JSON_PAGES)
|
|
first = blob_in(
|
|
generate_main(write_config(tmp_path, DEVICES_HEADER, api, JSON_ENTITIES))
|
|
)
|
|
# Compressing the same document again gives the same bytes
|
|
document = wizard.wizard_document(CORE.config["api"]["wizard"], CORE.config)
|
|
again = wizard.zstd_module().compress(
|
|
json.dumps(
|
|
document, separators=(",", ":"), sort_keys=True, ensure_ascii=False
|
|
).encode("utf-8"),
|
|
level=wizard.WIZARD_ZSTD_LEVEL,
|
|
)
|
|
|
|
assert again == first
|
|
assert first[:4] == b"\x28\xb5\x2f\xfd" # a zstd frame
|
|
|
|
|
|
def test_the_blob_is_in_flash_on_esp8266(
|
|
tmp_path: Path, generate_main: Callable[[str | Path], str]
|
|
) -> None:
|
|
main_cpp = generate_main(write_input_config(tmp_path, ESP8266_HEADER))
|
|
|
|
assert "esphome::api::API_WIZARD_DATA[] PROGMEM = {" in main_cpp
|
|
assert (
|
|
"const api::WizardInputEntry esphome::api::API_WIZARD_INPUTS[] PROGMEM = {"
|
|
in main_cpp
|
|
)
|
|
# Only the compressed data and the input table are emitted, no strings or row tables
|
|
assert "api_wizard_str" not in main_cpp
|
|
assert "WizardEntityRow" not in main_cpp
|
|
|
|
|
|
def test_inputs_have_buffers_hashes_and_a_table(
|
|
tmp_path: Path, generate_main: Callable[[str | Path], str]
|
|
) -> None:
|
|
main_cpp = generate_main(write_input_config(tmp_path, ESP32_HEADER))
|
|
|
|
# The buffer holds the YAML entity ID until the wizard sets another
|
|
assert 'static char api_wizard_input_ha_sensor[256] = "";' in main_cpp
|
|
entities = (
|
|
"ha_sensor",
|
|
"ha_binary",
|
|
"ha_text",
|
|
"ha_number",
|
|
"ha_switch",
|
|
"ha_txt",
|
|
"ha_select",
|
|
"ha_button",
|
|
"weather_input",
|
|
"unset_input",
|
|
)
|
|
for entity in entities:
|
|
assert f"{{{fnv1_hash(entity)}u, api_wizard_input_{entity}}}" in main_cpp
|
|
assert get_define_value("API_WIZARD_INPUT_COUNT") == str(len(entities))
|
|
for entity in entities[:-2]:
|
|
assert f"{entity}->set_entity_id(api_wizard_input_{entity});" in main_cpp
|
|
# The standalone inputs are objects that read the same buffer
|
|
assert "WizardInput(api_wizard_input_weather_input)" in main_cpp
|
|
assert "WizardInput(api_wizard_input_unset_input)" in main_cpp
|
|
# Not an input, so it keeps its constant
|
|
assert 'ha_plain_sensor->set_entity_id("sensor.plain");' in main_cpp
|
|
|
|
|
|
def test_an_entity_without_a_name_cannot_be_in_the_wizard(tmp_path: Path) -> None:
|
|
yaml_entities = 'switch:\n - platform: template\n id: nameless\n name: ""\n optimistic: true\n'
|
|
errors = config_errors(
|
|
write_config(
|
|
tmp_path,
|
|
ESP32_HEADER,
|
|
wizard_api("- entities: [{id: nameless}]"),
|
|
yaml_entities,
|
|
)
|
|
)
|
|
|
|
assert any("has no name of its own" in error for error in errors), errors
|
|
|
|
|
|
def test_zstd_falls_back_to_the_backport(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Before Python 3.14 the standard library has no zstd, so the backport is used."""
|
|
backport = object()
|
|
|
|
def import_module(name: str) -> object:
|
|
if name == "compression.zstd":
|
|
raise ImportError(name)
|
|
assert name == "backports.zstd"
|
|
return backport
|
|
|
|
monkeypatch.setattr(wizard.importlib, "import_module", import_module)
|
|
|
|
assert wizard.zstd_module() is backport
|
|
|
|
|
|
def test_a_wizard_too_big_for_one_message_is_rejected(tmp_path: Path) -> None:
|
|
# Random text does not compress, so this needs more than the limit even compressed
|
|
rng = random.Random(1)
|
|
alphabet = string.ascii_letters + string.digits
|
|
entities = ",\n".join(
|
|
"{id: sw, description: "
|
|
+ "".join(rng.choice(alphabet) for _ in range(255))
|
|
+ "}"
|
|
for _ in range(400)
|
|
)
|
|
api = wizard_api(f"- entities: [\n{entities}\n]")
|
|
errors = config_errors(write_config(tmp_path, ESP32_HEADER, api))
|
|
|
|
assert any(
|
|
"The compressed wizard is" in error and "bytes over the 65512 bytes" in error
|
|
for error in errors
|
|
), errors
|
|
|
|
|
|
def test_repeated_text_compresses_well_inside_the_limit(tmp_path: Path) -> None:
|
|
# 400 identical entities are far over the limit as JSON, but compress to very little
|
|
entities = ",\n".join(["{id: sw, description: " + "x" * 255 + "}"] * 400)
|
|
api = wizard_api(f"- entities: [\n{entities}\n]")
|
|
|
|
assert config_errors(write_config(tmp_path, ESP32_HEADER, api)) == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("platform", "entity_id"),
|
|
[
|
|
("sensor", "sensor.a"),
|
|
("switch", "light.a"),
|
|
("text", "text.a"),
|
|
("button", "button.a"),
|
|
],
|
|
)
|
|
def test_a_linked_entity_must_not_set_an_entity_id(
|
|
tmp_path: Path, platform: str, entity_id: str
|
|
) -> None:
|
|
yaml_entities = (
|
|
f"{platform}:\n - platform: homeassistant\n id: fixed\n"
|
|
f" entity_id: {entity_id}\n"
|
|
)
|
|
errors = config_errors(
|
|
write_config(
|
|
tmp_path, ESP32_HEADER, wizard_inputs("- entity: fixed"), yaml_entities
|
|
)
|
|
)
|
|
|
|
assert any(
|
|
"'fixed' has an entity_id set in its configuration" in error
|
|
and "Remove entity_id" in error
|
|
for error in errors
|
|
), errors
|
|
|
|
|
|
def test_an_entity_with_an_entity_id_that_is_no_input_is_valid(
|
|
tmp_path: Path, generate_main: Callable[[str | Path], str]
|
|
) -> None:
|
|
main_cpp = generate_main(write_input_config(tmp_path, ESP32_HEADER))
|
|
|
|
# Static entries keep their literal and are kept apart from the wizard buffers
|
|
assert 'ha_plain_sensor->set_entity_id("sensor.plain");' in main_cpp
|
|
assert "api_wizard_input_ha_plain_sensor" not in main_cpp
|
|
# Every buffer starts empty
|
|
assert '] = "sensor.' not in main_cpp.split("API_WIZARD_INPUTS")[0]
|