From 98f6e7438537a25da47267459cdbfce0cf10bd11 Mon Sep 17 00:00:00 2001 From: Jesse Hills <3060199+jesserockz@users.noreply.github.com> Date: Wed, 15 Jul 2026 14:44:28 +1200 Subject: [PATCH] [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. --- esphome/config_validation.py | 8 + script/build_language_schema.py | 139 +++++++++++++++- tests/script/test_build_language_schema.py | 174 +++++++++++++++++++++ tests/unit_tests/test_config_validation.py | 28 ++++ 4 files changed, 342 insertions(+), 7 deletions(-) diff --git a/esphome/config_validation.py b/esphome/config_validation.py index 16f0a63aa0..7191ee50d0 100644 --- a/esphome/config_validation.py +++ b/esphome/config_validation.py @@ -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) diff --git a/script/build_language_schema.py b/script/build_language_schema.py index f6dcf00851..fe01d4e3c2 100755 --- a/script/build_language_schema.py +++ b/script/build_language_schema.py @@ -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") diff --git a/tests/script/test_build_language_schema.py b/tests/script/test_build_language_schema.py index 8bbaa2773a..83d084bc89 100644 --- a/tests/script/test_build_language_schema.py +++ b/tests/script/test_build_language_schema.py @@ -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 diff --git a/tests/unit_tests/test_config_validation.py b/tests/unit_tests/test_config_validation.py index 17dfaad9b8..c60f9a040a 100644 --- a/tests/unit_tests/test_config_validation.py +++ b/tests/unit_tests/test_config_validation.py @@ -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)