Files
esphome/tests/script/test_build_language_schema.py
Jesse Hills 63e1adfd5f [schema] Keep the field's own range when spreading a named schema
Address esphbot review: emitting min/max means named scalar schemas now carry
bounds (e.g. core.uint8_t is 0..255), and shrink()'s spread copied those over a
tighter bound the field defined itself. cv.All(cv.uint8_t, cv.Range(min=1,
max=6)) dumped 0..255 instead of 1..6, and All(uint32_t, Range(min=1)) lost its
min. The spread now reconciles the two to the intersection (larger min, smaller
max); non-numeric bounds keep the field's own value.
2026-09-15 07:13:29 +12:00

568 lines
20 KiB
Python

"""Unit tests for script/build_language_schema.py."""
from __future__ import annotations
import ast
from collections.abc import Callable
import importlib.util
import json
from pathlib import Path
import subprocess
import sys
from typing import Any
import pytest
from esphome import config_validation as cv
SCRIPT_PATH = (
Path(__file__).resolve().parent.parent.parent
/ "script"
/ "build_language_schema.py"
)
def _load_script_module():
spec = importlib.util.spec_from_file_location("build_language_schema", SCRIPT_PATH)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _extract_sort_obj():
# ``sort_obj`` is pure and self-contained; pulling it via AST avoids
# exercising the module-level component-loading state for these tests.
tree = ast.parse(SCRIPT_PATH.read_text())
for node in tree.body:
if isinstance(node, ast.FunctionDef) and node.name == "sort_obj":
namespace: dict = {"S_TYPE": "type"}
module = ast.Module(body=[node], type_ignores=[])
exec(compile(module, str(SCRIPT_PATH), "exec"), namespace)
return namespace["sort_obj"]
raise AssertionError("sort_obj not found in build_language_schema.py")
sort_obj = _extract_sort_obj()
_bls = _load_script_module()
def test_sort_obj_sorts_dict_keys() -> None:
result = sort_obj({"b": 1, "a": 2, "c": 3})
assert list(result.keys()) == ["a", "b", "c"]
def test_sort_obj_sorts_nested_dicts() -> None:
result = sort_obj({"outer": {"z": 1, "a": 2}})
assert list(result["outer"].keys()) == ["a", "z"]
def test_sort_obj_preserves_enum_values_order() -> None:
config = {
"type": "enum",
"values": {
"2MB": None,
"4MB": None,
"8MB": None,
"16MB": None,
"32MB": None,
},
}
result = sort_obj(config)
assert list(result["values"].keys()) == ["2MB", "4MB", "8MB", "16MB", "32MB"]
def test_sort_obj_sorts_non_enum_values_key() -> None:
config = {"type": "schema", "values": {"z": 1, "a": 2}}
result = sort_obj(config)
assert list(result["values"].keys()) == ["a", "z"]
def test_sort_obj_sorts_other_keys_in_enum() -> None:
config = {
"type": "enum",
"default": "4MB",
"key": "Optional",
"values": {"2MB": None, "4MB": None},
}
result = sort_obj(config)
assert list(result.keys()) == ["default", "key", "type", "values"]
assert list(result["values"].keys()) == ["2MB", "4MB"]
def test_sort_obj_recurses_into_enum_value_entries() -> None:
config = {
"type": "enum",
"values": {
"esp32": {"name": "ESP32", "docs": "Original"},
"esp32-c3": {"name": "ESP32-C3", "docs": "RISC-V"},
},
}
result = sort_obj(config)
assert list(result["values"].keys()) == ["esp32", "esp32-c3"]
assert list(result["values"]["esp32"].keys()) == ["docs", "name"]
def test_sort_obj_handles_lists() -> None:
result = sort_obj([{"b": 1, "a": 2}, {"d": 3, "c": 4}])
assert list(result[0].keys()) == ["a", "b"]
assert list(result[1].keys()) == ["c", "d"]
def test_sort_obj_passes_through_scalars() -> None:
assert sort_obj("hello") == "hello"
assert sort_obj(42) == 42
assert sort_obj(None) is None
assert sort_obj(True) is True
def test_convert_emits_explicit_sensitive_marker() -> None:
config_var: dict = {}
_bls.convert(cv.sensitive(cv.string), config_var, "/test")
assert config_var["sensitive"] is True
assert config_var["sensitive_source"] == "explicit"
assert config_var["type"] == "string"
def test_convert_walks_callable_schema_extractor() -> None:
"""A callable schema tagged for "schema" extraction is resolved and walked."""
from esphome.schema_extractors import SCHEMA_EXTRACT, schema_extractor
@schema_extractor("schema")
def dynamic_schema(value):
if value is SCHEMA_EXTRACT:
return cv.Schema({cv.Required("foo"): cv.string})
return value
config_var: dict = {}
_bls.convert(dynamic_schema, config_var, "/test")
assert config_var["type"] == "schema"
assert "foo" in config_var["schema"]["config_vars"]
def test_convert_emits_variant_enum() -> None:
"""A per-variant enum is dumped with each value tagged by its variants."""
from esphome.components.esp32 import (
VARIANT_ESP32,
VARIANT_ESP32S3,
variant_filtered_enum,
)
validator = variant_filtered_enum(
{VARIANT_ESP32: ("quad",), VARIANT_ESP32S3: ("quad", "octal")},
lower=True,
)
config_var: dict = {}
_bls.convert(validator, config_var, "/test")
assert config_var["type"] == "enum"
assert config_var["values"] == {
"quad": {"variants": [VARIANT_ESP32, VARIANT_ESP32S3]},
"octal": {"variants": [VARIANT_ESP32S3]},
}
def test_convert_keys_emits_heuristic_sensitive_marker() -> None:
converted: dict = {}
_bls.convert_keys(converted, {cv.Optional("password"): cv.string}, "/root")
entry = converted["schema"]["config_vars"]["password"]
assert entry["sensitive"] is True
assert entry["sensitive_source"] == "heuristic"
assert entry["type"] == "string"
def test_convert_keys_explicit_beats_heuristic() -> None:
# Key name matches a fragment but the validator is explicitly wrapped;
# the explicit branch should win and emit ``sensitive_source: explicit``.
converted: dict = {}
_bls.convert_keys(
converted, {cv.Optional("password"): cv.sensitive(cv.string)}, "/root"
)
entry = converted["schema"]["config_vars"]["password"]
assert entry["sensitive"] is True
assert entry["sensitive_source"] == "explicit"
def test_convert_keys_no_heuristic_for_non_string_leaves() -> None:
# Even though the key contains a fragment, a non-string leaf must not
# be flagged. Prevents false positives on unrelated fields whose name
# happens to embed a substring like "token".
converted: dict = {}
_bls.convert_keys(converted, {cv.Optional("password"): cv.boolean}, "/root")
entry = converted["schema"]["config_vars"]["password"]
assert "sensitive" not in entry
assert "sensitive_source" not in entry
def test_convert_keys_no_marker_for_non_sensitive_field() -> None:
converted: dict = {}
_bls.convert_keys(converted, {cv.Optional("hostname"): cv.string}, "/root")
entry = converted["schema"]["config_vars"]["hostname"]
assert "sensitive" not in entry
assert "sensitive_source" not in entry
def _wildcard_validator(value: Any) -> Any:
return value
def test_convert_keys_marker_wrapped_callable_key_normalizes() -> None:
converted: dict = {}
_bls.convert_keys(converted, {cv.Optional(_wildcard_validator): cv.string}, "/root")
config_vars = converted["schema"]["config_vars"]
assert set(config_vars) == {"string"}
assert config_vars["string"]["key"] == "Optional"
assert config_vars["string"]["key_type"] == "_wildcard_validator"
def test_convert_keys_marker_wrapped_callable_beside_fixed_keys() -> None:
converted: dict = {}
_bls.convert_keys(
converted,
{cv.Required("id"): cv.string, cv.Optional(_wildcard_validator): cv.string},
"/root",
)
assert set(converted["schema"]["config_vars"]) == {"id", "string"}
def test_convert_keys_bare_callable_dotted_qualname() -> None:
def make_validator() -> Callable[[Any], Any]:
def validator(value: Any) -> Any:
return value
return validator
converted: dict = {}
_bls.convert_keys(converted, {make_validator(): cv.string}, "/root")
assert converted["key"] == "String"
assert converted["key_type"].endswith("make_validator.<locals>.validator")
assert "at 0x" not in converted["key_type"]
assert set(converted["schema"]["config_vars"]) == {"string"}
# ---------------------------------------------------------------------------
# Regression tests for the lvgl schema dump.
#
# lvgl's CONFIG_SCHEMA is a callable closure and its widget/style schemas are
# built lazily at validation time, so the static dumper used to emit an empty
# `lvgl:` schema, no widget completion, and an inlined ~80-property STYLE_SCHEMA
# duplicated at every widget x part x state (a 17 MB lvgl.json). These exercise
# the full `build_schema()` and assert the generated lvgl.json carries the data
# the schema_extractor hooks added.
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def lvgl_schema(tmp_path_factory: pytest.TempPathFactory) -> dict:
"""Run the full language-schema build once and return parsed lvgl.json.
The build must run in a fresh interpreter: ``build_language_schema.py``
enables schema extraction *before* importing any esphome component, and the
extraction hooks are no-ops if the components were already imported (as they
are inside the pytest session). Running it as a subprocess mirrors how CI
generates the schema and keeps this test isolated from import order.
"""
out_dir = tmp_path_factory.mktemp("language_schema")
subprocess.run(
[sys.executable, str(SCRIPT_PATH), "--output-path", str(out_dir)],
check=True,
capture_output=True,
text=True,
)
return json.loads((out_dir / "lvgl.json").read_text())
def _lvgl_config_vars(lvgl_schema: dict) -> dict:
config_schema = lvgl_schema["lvgl"]["schemas"]["CONFIG_SCHEMA"]
# Previously empty (`{}`); the schema_extractor on lvgl_config_schema now
# hands the dumper the composed top-level schema.
assert config_schema["type"] == "schema"
return config_schema["schema"]["config_vars"]
def test_lvgl_top_level_schema_is_exposed(lvgl_schema: dict) -> None:
config_vars = _lvgl_config_vars(lvgl_schema)
# Was 0 config_vars before LVGL_TOP_LEVEL_SCHEMA was exposed.
assert len(config_vars) > 100
# A representative spread of top-level options the runtime validates.
for key in ("displays", "pages", "default_font", "on_idle", "touchscreens"):
assert key in config_vars, f"missing top-level lvgl option: {key}"
def test_lvgl_widgets_key_enumerated(lvgl_schema: dict) -> None:
config_vars = _lvgl_config_vars(lvgl_schema)
# The widgets: list is assembled per-value at runtime; the extractor
# enumerates every registered widget type into a named WIDGET_TYPES schema
# which the widgets: list references (recursive, so widgets can nest).
assert "widgets" in config_vars
widgets = config_vars["widgets"]
assert widgets["is_list"] is True
assert widgets["schema"]["extends"] == ["lvgl.WIDGET_TYPES"]
widget_types = lvgl_schema["lvgl"]["schemas"]["WIDGET_TYPES"]["schema"][
"config_vars"
]
# Every registered widget type should appear as an optional key.
for name in ("obj", "label", "button", "slider", "switch", "arc"):
assert name in widget_types, f"widget type not enumerated: {name}"
# Each enumerated widget carries its own property schema, not an empty stub.
assert widget_types["label"]["type"] == "schema"
assert len(widget_types["label"]["schema"]["config_vars"]) > 0
# Each widget can contain child widgets, via the same named ref — so the
# tree is recursive and the dump stays finite.
nested = widget_types["obj"]["schema"]["config_vars"]["widgets"]
assert nested["is_list"] is True
assert nested["schema"]["extends"] == ["lvgl.WIDGET_TYPES"]
def test_lvgl_style_schemas_are_named_and_deduped(lvgl_schema: dict) -> None:
schemas = lvgl_schema["lvgl"]["schemas"]
# Importing these into the lvgl __init__ namespace lets the dumper register
# them as named schemas and emit `extends` refs instead of inlining them.
for name in ("STYLE_SCHEMA", "STATE_SCHEMA", "SET_STATE_SCHEMA"):
assert name in schemas, f"style schema not registered as named: {name}"
# STYLE_SCHEMA must be referenced via `extends`, not inlined at every use
# site. Count the references to prove the dedup actually happened.
refs = 0
def _count(node: object) -> None:
nonlocal refs
if isinstance(node, dict):
extends = node.get("extends")
if isinstance(extends, list) and "lvgl.STYLE_SCHEMA" in extends:
refs += 1
for value in node.values():
_count(value)
elif isinstance(node, list):
for value in node:
_count(value)
_count(lvgl_schema)
assert refs > 100, f"STYLE_SCHEMA should be referenced via extends, got {refs}"
# ---------------------------------------------------------------------------
# Typing of esphome.config_validation validators.
#
# These validators used to fall through convert() with no ``type``, leaving the
# visual editor / dashboard unable to tell what YAML the field accepts. They are
# now described either by identity (scalar leaf validators) or via a
# schema_extractor decorator (factory-produced closures like float_with_unit).
# ---------------------------------------------------------------------------
def _convert(validator: object) -> dict:
config_var: dict = {}
_bls.convert(validator, config_var, "/x")
return config_var
@pytest.mark.parametrize(
("validator", "expected"),
[
(cv.icon, "string"),
(cv.mac_address, "string"),
(cv.url, "string"),
(cv.uuid, "string"),
(cv.directory, "string"),
(cv.mqtt_qos, "integer"),
(cv.hex_int, "integer"),
(cv.update_interval, "time"),
(cv.time_period_str_colon, "time"),
(cv.lambda_, "lambda"),
(cv.returning_lambda, "lambda"),
],
)
def test_convert_types_scalar_cv_validators(validator: object, expected: str) -> None:
assert _convert(validator).get("type") == expected
@pytest.mark.parametrize(
("validator", "expected_type", "expected_unit"),
[
(cv.percentage, "float", "%"),
(cv.possibly_negative_percentage, "float", "%"),
(cv.temperature, "float", "°C"),
(cv.temperature_delta, "float", "°C"),
(cv.color_temperature, "float", "mireds"),
(cv.percentage_int, "integer", "%"),
(cv.validate_bytes, "integer", "B"),
],
)
def test_convert_types_unit_validators(
validator: object, expected_type: str, expected_unit: str
) -> None:
# Number-with-unit validators accept both a bare number and a unit-suffixed
# string, so they carry their base numeric type plus the canonical unit.
entry = _convert(validator)
assert entry.get("type") == expected_type
assert entry.get("unit") == expected_unit
def test_convert_dimensions_left_untyped() -> None:
# cv.dimensions accepts either a "WIDTHxHEIGHT" string or a [w, h] list;
# a single scalar type would reject one form, so it is left untyped.
entry = _convert(cv.dimensions)
assert "type" not in entry
def test_convert_entity_category_is_enum() -> None:
entry = _convert(cv.entity_category)
assert entry["type"] == "enum"
assert set(entry["values"]) == set(cv.ENTITY_CATEGORIES)
def test_convert_bind_key_is_sensitive_string() -> None:
entry = _convert(cv.bind_key)
assert entry["type"] == "string"
assert entry["sensitive"] is True
@pytest.mark.parametrize("scalar", ["string", "integer", "float", "time", "lambda"])
def test_convert_scalar_schema_extractor(scalar: str) -> None:
"""A validator that declares a scalar type via schema_extractor is typed.
Mirrors cv.float_with_unit / cv.date_time, whose decorated closures return
None for SCHEMA_EXTRACT and are keyed into hidden_schemas by repr.
"""
from esphome import schema_extractors as ejs
def decorated(value: object) -> None:
return None
ejs.hidden_schemas[repr(decorated)] = scalar
try:
assert _convert(decorated).get("type") == scalar
finally:
del ejs.hidden_schemas[repr(decorated)]
def test_convert_float_with_unit_reports_unit() -> None:
"""A float schema_extractor whose probe returns a unit types float + unit."""
import voluptuous as vol
from esphome import schema_extractors as ejs
def frequency_validator(value: object) -> object:
return "Hz" if value is ejs.SCHEMA_EXTRACT else value
ejs.hidden_schemas[repr(frequency_validator)] = "float"
try:
entry = _convert(frequency_validator)
assert entry.get("type") == "float"
assert entry.get("unit") == "Hz"
finally:
del ejs.hidden_schemas[repr(frequency_validator)]
# A float source with no unit is a bare float (no unit key).
def plain_float(value: object) -> object:
return None if value is ejs.SCHEMA_EXTRACT else value
ejs.hidden_schemas[repr(plain_float)] = "float"
try:
entry = _convert(plain_float)
assert entry.get("type") == "float"
assert "unit" not in entry
finally:
del ejs.hidden_schemas[repr(plain_float)]
# A vol.Range in the All contributes min/max next to type, not a type.
assert _convert(vol.Range(min=45.0, max=66.0)) == {"min": 45.0, "max": 66.0}
def test_convert_range_stringifies_non_numeric_bounds() -> None:
"""A time-period range keeps JSON-serializable string bounds."""
import voluptuous as vol
entry = _convert(vol.Range(min=cv.time_period("1s"), max=cv.time_period("10s")))
assert isinstance(entry["min"], str) and isinstance(entry["max"], str)
@pytest.fixture(scope="module")
def full_schema_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
"""Run the full build once (fresh interpreter, see ``lvgl_schema``).
PYTHONPATH points at this worktree so the subprocess imports the local
esphome (with the config_validation changes) rather than an editable install
that may resolve to a different checkout.
"""
import os
out_dir = tmp_path_factory.mktemp("cv_types_schema")
repo_root = SCRIPT_PATH.parent.parent
subprocess.run(
[sys.executable, str(SCRIPT_PATH), "--output-path", str(out_dir)],
check=True,
capture_output=True,
text=True,
cwd=str(repo_root),
env={**os.environ, "PYTHONPATH": str(repo_root)},
)
return out_dir
def test_cv_types_end_to_end(full_schema_dir: Path) -> None:
"""The full build types config_validation fields end-to-end.
Also covers the shrink() spread of a field that references two typed schemas
at once (hex_uint8_t + uint8_t), which previously tripped an assertion.
"""
core = json.loads((full_schema_dir / "esphome.json").read_text())["core"]
entity = core["schemas"]["ENTITY_BASE_SCHEMA"]["schema"]["config_vars"]
assert entity["icon"]["type"] == "string"
assert entity["entity_category"]["type"] == "enum"
climate = json.loads((full_schema_dir / "climate.json").read_text())["climate"]
visual = climate["schemas"]["_CLIMATE_SCHEMA"]["schema"]["config_vars"]["visual"]
min_temp = visual["schema"]["config_vars"]["min_temperature"]
assert min_temp["type"] == "float"
assert min_temp["unit"] == "°C"
# message_type references both hex_uint8_t and uint8_t; shrink() must spread
# it to integer instead of tripping the single-extends assertion.
remote = json.loads((full_schema_dir / "remote_receiver.json").read_text())
abbwelcome = remote["remote_receiver.binary_sensor"]["schemas"]["CONFIG_SCHEMA"][
"schema"
]["config_vars"]["abbwelcome"]["schema"]["config_vars"]
assert abbwelcome["message_type"]["type"] == "integer"
# cv.All(cv.frequency, cv.float_range(45, 66)) -> float + unit + min/max
# inline, next to type.
ade = json.loads((full_schema_dir / "ade7880.json").read_text())
freq = ade["ade7880.sensor"]["schemas"]["CONFIG_SCHEMA"]["schema"]["config_vars"][
"frequency"
]
assert freq["type"] == "float"
assert freq["unit"] == "Hz"
assert freq["min"] == 45.0
assert freq["max"] == 66.0
# cv.All(cv.uint8_t, cv.Range(min=1, max=6)): spreading the uint8_t named
# schema (0..255) must not clobber the field's own tighter 1..6 bound.
tm1637 = json.loads((full_schema_dir / "tm1637.json").read_text())
length = tm1637["tm1637.display"]["schemas"]["CONFIG_SCHEMA"]["schema"][
"config_vars"
]["length"]
assert length["type"] == "integer"
assert length["data_type"] == "uint8_t"
assert length["min"] == 1
assert length["max"] == 6
# positive_float = All(float_, Range(min=0)): a bounds-only named schema
# spreads its data_type name and its min onto the field.
light = json.loads((full_schema_dir / "light.json").read_text())["light"]
gamma = light["schemas"]["BRIGHTNESS_ONLY_LIGHT_SCHEMA"]["schema"]["config_vars"][
"gamma_correct"
]
assert gamma["data_type"] == "positive_float"
assert gamma["min"] == 0