mirror of
https://github.com/esphome/esphome.git
synced 2026-09-30 16:30:22 +00:00
[schema] Type config_validation validators and range bounds in language schema dump
The language schema dump left many config_validation validators untyped (icon, mac_address, percentage, update_interval, lambdas, encryption keys, ...), so the visual editor and dashboard could not tell what YAML those fields accept. Type them via convert() and schema_extractor decorators, emit float_with_unit quantities (frequency, voltage, current, ...) as the field type, and attach min/max bounds detected from range validators next to the type.
This commit is contained in:
Jesse Hills
@@ -1196,7 +1196,10 @@ def date_time(date: bool, time: bool):
|
||||
}
|
||||
)
|
||||
|
||||
@schema_extractor("string")
|
||||
def validator(value):
|
||||
if value is SCHEMA_EXTRACT:
|
||||
return None
|
||||
if isinstance(value, dict):
|
||||
return schema(value)
|
||||
value = string(value)
|
||||
@@ -1365,7 +1368,12 @@ def float_with_unit(quantity, regex_suffix, optional_unit=False):
|
||||
f"^([-+]?[0-9]*\\.?[0-9]*)\\s*(\\w*?){regex_suffix}$", re.UNICODE
|
||||
)
|
||||
|
||||
@schema_extractor("float")
|
||||
def validator(value):
|
||||
if value is SCHEMA_EXTRACT:
|
||||
# Report the specific quantity (e.g. "frequency", "voltage") so the
|
||||
# schema dump can use it as the field type instead of a bare float.
|
||||
return quantity
|
||||
if optional_unit:
|
||||
try:
|
||||
return float_(value)
|
||||
|
||||
@@ -120,6 +120,61 @@ from esphome.util import Registry # noqa: E402
|
||||
|
||||
# pylint: enable=wrong-import-position
|
||||
|
||||
# Scalar ``cv.*`` validators the dumper describes by identity. Extending these
|
||||
# tuples -- rather than decorating each validator with a schema_extractor --
|
||||
# mirrors how ``cv.boolean`` / ``cv.string`` / ``cv.int_`` are already handled
|
||||
# in ``convert()`` and keeps runtime validation completely untouched. Grouped by
|
||||
# the ``type`` emitted into the schema dump so a language server / visual editor
|
||||
# knows what YAML each field accepts instead of treating it as free-form.
|
||||
_CV_STRING_VALIDATORS = (
|
||||
cv.icon,
|
||||
cv.mac_address,
|
||||
cv.url,
|
||||
cv.publish_topic,
|
||||
cv.subscribe_topic,
|
||||
cv.mqtt_payload,
|
||||
cv.uuid,
|
||||
cv.ssid,
|
||||
cv.domain,
|
||||
cv.domain_name,
|
||||
cv.hostname,
|
||||
cv.entity_id,
|
||||
cv.git_ref,
|
||||
cv.string_no_slash,
|
||||
cv.version_number,
|
||||
cv.validate_esphome_version,
|
||||
cv.validate_id_name,
|
||||
cv._validate_entity_name,
|
||||
cv.validate_source_shorthand,
|
||||
cv.ipv4address,
|
||||
cv.ipv6network,
|
||||
cv.ipv4address_multi_broadcast,
|
||||
cv.time_of_day,
|
||||
cv.directory,
|
||||
cv.file_,
|
||||
cv.dimensions,
|
||||
cv.none,
|
||||
)
|
||||
_CV_INTEGER_VALIDATORS = (
|
||||
cv.hex_int,
|
||||
cv.percentage_int,
|
||||
cv.mqtt_qos,
|
||||
cv.validate_bytes,
|
||||
)
|
||||
_CV_FLOAT_VALIDATORS = (
|
||||
cv.percentage,
|
||||
cv.possibly_negative_percentage,
|
||||
cv.temperature,
|
||||
cv.temperature_delta,
|
||||
cv.color_temperature,
|
||||
)
|
||||
_CV_TIME_VALIDATORS = (
|
||||
cv.update_interval,
|
||||
cv.time_period_str_unit,
|
||||
cv.time_period_str_colon,
|
||||
)
|
||||
_CV_LAMBDA_VALIDATORS = (cv.lambda_, cv.returning_lambda)
|
||||
|
||||
|
||||
def sort_obj(obj):
|
||||
if isinstance(obj, dict):
|
||||
@@ -625,22 +680,48 @@ def shrink():
|
||||
# then are all simple types, integer and strings
|
||||
for x, paths in referenced_schemas.items():
|
||||
key_s = get_str_path_schema(x)
|
||||
if key_s and key_s.get(S_TYPE) in ["enum", "registry", "integer", "string"]:
|
||||
if key_s[S_TYPE] == "registry":
|
||||
# Spread scalar leaf schemas (a single ``type`` with no nested schema or
|
||||
# config vars) onto each referencing field so the type is inline. This
|
||||
# covers enum/registry/integer/string plus float_with_unit quantities
|
||||
# (e.g. a ``core.frequency`` schema typed ``frequency``), time and
|
||||
# lambda -- but never structural schemas, which stay as references.
|
||||
key_type = key_s.get(S_TYPE) if key_s else None
|
||||
if (
|
||||
key_type is not None
|
||||
and key_type not in ("schema", "typed", "trigger", "pin", "use_id")
|
||||
and S_SCHEMA not in key_s
|
||||
and S_CONFIG_VARS not in key_s
|
||||
):
|
||||
if key_type == "registry":
|
||||
print("Spreading registry: " + x)
|
||||
for target in paths:
|
||||
target_s = get_arr_path_schema(target)
|
||||
if S_SCHEMA not in target_s:
|
||||
print("skipping simple spread for " + ".".join(target))
|
||||
continue
|
||||
assert target_s[S_SCHEMA][S_EXTENDS] == [x]
|
||||
extends = target_s[S_SCHEMA][S_EXTENDS]
|
||||
if x not in extends:
|
||||
# Already handled on an earlier visit (a field can list the
|
||||
# same schema reference more than once).
|
||||
continue
|
||||
if len(extends) > 1:
|
||||
# The field references several schemas at once (e.g. a value
|
||||
# that extends both hex_uint8_t and uint8_t). Drop this
|
||||
# reference and let the remaining one(s) describe the type,
|
||||
# rather than forcing a single-extends spread here.
|
||||
extends.remove(x)
|
||||
continue
|
||||
assert extends == [x]
|
||||
target_s.pop(S_SCHEMA)
|
||||
target_s |= key_s
|
||||
if key_s[S_TYPE] in ["integer", "string"]:
|
||||
target_s["data_type"] = x.split(".")[1]
|
||||
# remove this dangling again
|
||||
pop_str_path_schema(x)
|
||||
elif not key_s:
|
||||
elif not key_s or set(key_s) <= {"min", "max"}:
|
||||
# An untyped named schema, or one carrying only range bounds (e.g.
|
||||
# positive_float = All(float_, Range(min=0)) has no scalar type but a
|
||||
# min). Spread its data_type name and any bounds onto each field.
|
||||
for target in paths:
|
||||
target_s = get_arr_path_schema(target)
|
||||
if S_SCHEMA not in target_s:
|
||||
@@ -651,6 +732,7 @@ def shrink():
|
||||
target_s.pop(S_SCHEMA)
|
||||
target_s.pop(S_TYPE) # undefined
|
||||
target_s["data_type"] = x.split(".")[1]
|
||||
target_s.update(key_s) # carry min/max bounds, if any
|
||||
# remove this dangling again
|
||||
pop_str_path_schema(x)
|
||||
|
||||
@@ -897,7 +979,12 @@ def convert(schema, config_var, path):
|
||||
if isinstance(schema, cv.SensitiveValidator):
|
||||
config_var["sensitive"] = True
|
||||
config_var["sensitive_source"] = "explicit"
|
||||
convert(schema.inner, config_var, f"{path}/sensitive")
|
||||
if isinstance(schema, cv.BindKeyValidator):
|
||||
# Its inner is the bound ``_validate`` method (a hex-key string);
|
||||
# walking it yields no type, so describe it directly.
|
||||
config_var[S_TYPE] = "string"
|
||||
else:
|
||||
convert(schema.inner, config_var, f"{path}/sensitive")
|
||||
return
|
||||
|
||||
if isinstance(schema, cv.All):
|
||||
@@ -930,16 +1017,44 @@ def convert(schema, config_var, path):
|
||||
if DUMP_RAW:
|
||||
config_var["raw"] = repr_schema
|
||||
|
||||
# A numeric range constraint (from cv.int_range / cv.float_range / a bare
|
||||
# vol.Range in an All) contributes bounds, not a type. Attach them at the
|
||||
# config var level, next to ``type``, so editors can validate the range.
|
||||
if isinstance(schema, vol.Range):
|
||||
# min/max may be non-numeric (e.g. a TimePeriod for a time-period range);
|
||||
# keep numbers as-is and stringify anything else so the dump stays JSON
|
||||
# serializable.
|
||||
if schema.min is not None:
|
||||
config_var["min"] = (
|
||||
schema.min if isinstance(schema.min, (int, float)) else str(schema.min)
|
||||
)
|
||||
if schema.max is not None:
|
||||
config_var["max"] = (
|
||||
schema.max if isinstance(schema.max, (int, float)) else str(schema.max)
|
||||
)
|
||||
return
|
||||
|
||||
# pylint: disable=comparison-with-callable
|
||||
if schema == cv.boolean:
|
||||
config_var[S_TYPE] = "boolean"
|
||||
elif schema == automation.validate_potentially_and_condition:
|
||||
config_var[S_TYPE] = "registry"
|
||||
config_var["registry"] = "condition"
|
||||
elif schema in (cv.int_, cv.int_range):
|
||||
elif schema in (cv.int_, cv.int_range) or schema in _CV_INTEGER_VALIDATORS:
|
||||
config_var[S_TYPE] = "integer"
|
||||
elif schema in (cv.string, cv.string_strict, cv.valid_name):
|
||||
elif schema in (cv.string, cv.string_strict, cv.valid_name) or (
|
||||
schema in _CV_STRING_VALIDATORS
|
||||
):
|
||||
config_var[S_TYPE] = "string"
|
||||
elif schema in _CV_FLOAT_VALIDATORS:
|
||||
config_var[S_TYPE] = "float"
|
||||
elif schema in _CV_TIME_VALIDATORS:
|
||||
config_var[S_TYPE] = "time"
|
||||
elif schema in _CV_LAMBDA_VALIDATORS:
|
||||
config_var[S_TYPE] = "lambda"
|
||||
elif schema == cv.entity_category:
|
||||
config_var[S_TYPE] = "enum"
|
||||
config_var["values"] = dict.fromkeys(cv.ENTITY_CATEGORIES)
|
||||
|
||||
elif isinstance(schema, vol.Schema):
|
||||
# test: esphome/project
|
||||
@@ -1013,6 +1128,16 @@ def convert(schema, config_var, path):
|
||||
config_var[S_TYPE] = "registry"
|
||||
config_var["registry"] = "light.effects"
|
||||
config_var["filter"] = data[0]
|
||||
elif schema_type == "float":
|
||||
# cv.float_with_unit returns its quantity name (e.g. "frequency")
|
||||
# for SCHEMA_EXTRACT, so the field type is the specific quantity
|
||||
# rather than a bare "float". Other float sources return None.
|
||||
config_var[S_TYPE] = data if isinstance(data, str) else "float"
|
||||
elif schema_type in ("string", "integer", "time", "lambda"):
|
||||
# Scalar validators (e.g. cv.date_time) that declare their result
|
||||
# type via schema_extractor. ``data`` is unused (the decorated
|
||||
# validator returns None for SCHEMA_EXTRACT).
|
||||
config_var[S_TYPE] = schema_type
|
||||
elif schema_type == "templatable":
|
||||
config_var["templatable"] = True
|
||||
convert(data, config_var, path + "/templat")
|
||||
|
||||
@@ -305,3 +305,177 @@ def test_lvgl_style_schemas_are_named_and_deduped(lvgl_schema: dict) -> None:
|
||||
|
||||
_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.validate_bytes, "integer"),
|
||||
(cv.percentage, "float"),
|
||||
(cv.temperature, "float"),
|
||||
(cv.color_temperature, "float"),
|
||||
(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
|
||||
|
||||
|
||||
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_uses_quantity_as_type() -> None:
|
||||
"""A float schema_extractor whose probe returns a quantity types by quantity."""
|
||||
import voluptuous as vol
|
||||
|
||||
from esphome import schema_extractors as ejs
|
||||
|
||||
def frequency_validator(value: object) -> object:
|
||||
return "frequency" if value is ejs.SCHEMA_EXTRACT else value
|
||||
|
||||
ejs.hidden_schemas[repr(frequency_validator)] = "float"
|
||||
try:
|
||||
assert _convert(frequency_validator).get("type") == "frequency"
|
||||
finally:
|
||||
del ejs.hidden_schemas[repr(frequency_validator)]
|
||||
|
||||
# A float source that does not name a quantity falls back to "float".
|
||||
def plain_float(value: object) -> object:
|
||||
return None if value is ejs.SCHEMA_EXTRACT else value
|
||||
|
||||
ejs.hidden_schemas[repr(plain_float)] = "float"
|
||||
try:
|
||||
assert _convert(plain_float).get("type") == "float"
|
||||
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"]
|
||||
assert visual["schema"]["config_vars"]["min_temperature"]["type"] == "float"
|
||||
|
||||
# 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)) -> quantity type + 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"] == "frequency"
|
||||
assert freq["min"] == 45.0
|
||||
assert freq["max"] == 66.0
|
||||
|
||||
# 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
|
||||
|
||||
@@ -1641,6 +1641,34 @@ def test_templatable_schema_extract() -> None:
|
||||
assert cv.templatable(cv.int_)(SCHEMA_EXTRACT) is cv.int_
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("validator", "quantity"),
|
||||
[
|
||||
(cv.float_with_unit("frequency", "(Hz)?"), "frequency"),
|
||||
(cv.frequency, "frequency"),
|
||||
(cv.voltage, "voltage"),
|
||||
(cv.decibel, "decibel"),
|
||||
],
|
||||
)
|
||||
def test_float_with_unit_schema_extract(validator: object, quantity: str) -> None:
|
||||
# For the SCHEMA_EXTRACT sentinel the validator returns its quantity name
|
||||
# (not the parsed value) so build_language_schema can type the field by
|
||||
# quantity (e.g. "frequency") instead of a bare float.
|
||||
assert validator(SCHEMA_EXTRACT) == quantity
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"validator",
|
||||
[
|
||||
cv.date_time(date=True, time=False),
|
||||
cv.date_time(date=False, time=True),
|
||||
cv.date_time(date=True, time=True),
|
||||
],
|
||||
)
|
||||
def test_date_time_schema_extract(validator: object) -> None:
|
||||
assert validator(SCHEMA_EXTRACT) is None
|
||||
|
||||
|
||||
def test_templatable_lambda() -> None:
|
||||
result = cv.templatable(cv.int_)(Lambda("return 5;"))
|
||||
assert isinstance(result, Lambda)
|
||||
|
||||
Reference in New Issue
Block a user