diff --git a/esphome/config_validation.py b/esphome/config_validation.py index 32da6e41855..9af419919ba 100644 --- a/esphome/config_validation.py +++ b/esphome/config_validation.py @@ -1019,6 +1019,123 @@ def has_none_or_all_keys(*keys): return validate +def _as_tuple(value: typing.Any) -> tuple: + return tuple(value) if isinstance(value, (list, tuple)) else (value,) + + +def _quote_value(value: typing.Any) -> str: + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, (int, float)): + return str(value) + return f"'{value}'" + + +def _key_constraint( + record: dict[str, typing.Any], + passes: Callable[[dict], bool], + message: Callable[[dict], str], + hint: str | None, +) -> Callable[[typing.Any], typing.Any]: + """Reject each of ``record["keys"]`` that is set when *passes* fails. + + The schema dump appends *record* to the schema's ``constraints`` list. + """ + suffix = f". {hint}" if hint else "" + + @schema_extractor("key_constraint") + def validate(obj: typing.Any) -> typing.Any: + if obj is SCHEMA_EXTRACT: + return copy.deepcopy(record) + if not isinstance(obj, dict): + raise Invalid("expected dictionary") + if passes(obj): + return obj + if errors := [ + Invalid(f"'{key}' {message(obj)}{suffix}", path=[key]) + for key in record["keys"] + if key in obj + ]: + raise MultipleInvalid(errors) + return obj + + return validate + + +def only_when( + keys: str | list[str] | tuple[str, ...], + selector: str, + values: typing.Any, + *, + negate: bool = False, + hint: str | None = None, +) -> Callable[[typing.Any], typing.Any]: + """Allow *keys* only when *selector* is one of *values* (none of them with *negate*). + + Place it after the schema in ``cv.All`` so the selector's default counts. *hint* is + added to the error. Prefer ``cv.typed_schema`` when most of the schema depends on + the selector. + """ + keys = _as_tuple(keys) + values = _as_tuple(values) + assert all(isinstance(v, (str, int, float, bool)) for v in values) + shown = " or ".join(_quote_value(v) for v in values) + message = f"is only valid when '{selector}' is {'not ' if negate else ''}{shown}" + + # The bool tag keeps True from matching 1 + wanted = [(isinstance(v, bool), v) for v in values] + + def passes(obj: dict) -> bool: + actual = obj.get(selector) + return ((isinstance(actual, bool), actual) in wanted) != negate + + return _key_constraint( + { + "type": "only_when", + "keys": list(keys), + "selector": selector, + "values": list(values), + "negate": negate, + }, + passes, + lambda obj: message, + hint, + ) + + +def key_requires( + keys: str | list[str] | tuple[str, ...], + *required: str, + any_of: bool = False, + hint: str | None = None, +) -> Callable[[typing.Any], typing.Any]: + """Allow *keys* only when all of *required* are set (any one of them with *any_of*). + + Place it after the schema in ``cv.All``. *hint* is added to the error. + """ + assert required + check = any if any_of else all + + def message(obj: dict) -> str: + # Name only the missing keys when all are required + if any_of: + return f"requires {' or '.join(_quote_value(key) for key in required)}" + missing = (_quote_value(key) for key in required if key not in obj) + return f"requires {' and '.join(missing)}" + + return _key_constraint( + { + "type": "key_requires", + "keys": list(_as_tuple(keys)), + "required": list(required), + "any_of": any_of, + }, + lambda obj: check(key in obj for key in required), + message, + hint, + ) + + TIME_PERIOD_ERROR = ( "Time period {} should be format number + unit, for example 5ms, 5s, 5min, 5h" ) diff --git a/script/build_language_schema.py b/script/build_language_schema.py index d8f692de63a..881733215b6 100755 --- a/script/build_language_schema.py +++ b/script/build_language_schema.py @@ -575,12 +575,22 @@ def merge(source, destination): # get node or create one node = destination.setdefault(key, {}) merge(value, node) + elif key == "constraints" and isinstance(value, list): + destination[key] = value + destination.get(key, []) else: destination[key] = value return destination +def inline_schema(node: dict, base: dict) -> None: + """Copy *base* into *node*, keeping the key constraints of both.""" + local = node.get("constraints", []) + node |= base + if local: + node["constraints"] = base.get("constraints", []) + local + + def is_platform_schema(schema_name): # added mostly because of schema_name == "microphone.MICROPHONE_SCHEMA" # and "alarm_control_panel" @@ -626,11 +636,14 @@ def shrink(): else: print("expected extends here!" + x) arr_s = merge(key_s, arr_s) - if arr_s[S_TYPE] in ["enum", "typed"]: - arr_s.pop(S_SCHEMA) + # Only key constraints survive the collapse + if arr_s[S_TYPE] in ["enum", "typed"] and ( + constraints := arr_s.pop(S_SCHEMA).get("constraints") + ): + arr_s[S_SCHEMA] = {"constraints": constraints} else: arr_s.pop(S_EXTENDS) - arr_s |= key_s[S_SCHEMA] + inline_schema(arr_s, key_s[S_SCHEMA]) # simple types should be spread on each component, # for enums so far these are logger.is_log_level, cover.validate_cover_state and pulse_counter.sensor.COUNT_MODE_SCHEMA @@ -1068,6 +1081,10 @@ def convert(schema, config_var, path): # A callable CONFIG_SCHEMA that returned a representative schema # for extraction (model-driven components); walk it as usual. convert(data, config_var, path) + elif schema_type == "key_constraint": + config_var.setdefault(S_SCHEMA, {}).setdefault("constraints", []).append( + data + ) else: raise TypeError("Unknown extracted schema type") elif config_var.get("key") == "GeneratedID": diff --git a/tests/script/test_build_language_schema.py b/tests/script/test_build_language_schema.py index 121a84e5aa3..d53af0e40c2 100644 --- a/tests/script/test_build_language_schema.py +++ b/tests/script/test_build_language_schema.py @@ -141,6 +141,62 @@ def test_convert_walks_callable_schema_extractor() -> None: assert "foo" in config_var["schema"]["config_vars"] +def test_convert_emits_key_constraints() -> None: + base = cv.Schema({cv.Optional("transport", default="tcp"): cv.string}) + schema = cv.All( + base.extend({cv.Optional("path"): cv.string}), + cv.only_when("path", "transport", ("ws", "wss")), + cv.only_when(("path", "transport"), "mode", "tls", negate=True), + cv.key_requires("path", "a", "b", any_of=True), + ) + assert _bls.convert_config(schema, "/test")["schema"]["constraints"] == [ + { + "type": "only_when", + "keys": ["path"], + "selector": "transport", + "values": ["ws", "wss"], + "negate": False, + }, + { + "type": "only_when", + "keys": ["path", "transport"], + "selector": "mode", + "values": ["tls"], + "negate": True, + }, + { + "type": "key_requires", + "keys": ["path"], + "required": ["a", "b"], + "any_of": True, + }, + ] + + +def test_merge_keeps_base_and_local_constraints() -> None: + base = {"schema": {"constraints": [{"type": "key_requires", "keys": ["a"]}]}} + local = {"schema": {"constraints": [{"type": "only_when", "keys": ["b"]}]}} + assert _bls.merge(base, local)["schema"]["constraints"] == [ + {"type": "key_requires", "keys": ["a"]}, + {"type": "only_when", "keys": ["b"]}, + ] + + +def test_merge_enum_option_named_constraints() -> None: + base = {"values": {"constraints": None}} + assert _bls.merge(base, {"values": {}}) == {"values": {"constraints": None}} + + +def test_inline_schema_keeps_base_and_local_constraints() -> None: + node = {"constraints": [{"type": "only_when"}]} + base = {"config_vars": {"a": {}}, "constraints": [{"type": "key_requires"}]} + _bls.inline_schema(node, base) + assert node == { + "config_vars": {"a": {}}, + "constraints": [{"type": "key_requires"}, {"type": "only_when"}], + } + + 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 ( diff --git a/tests/unit_tests/test_config_validation.py b/tests/unit_tests/test_config_validation.py index 66732dd04db..da5d7c85512 100644 --- a/tests/unit_tests/test_config_validation.py +++ b/tests/unit_tests/test_config_validation.py @@ -1876,6 +1876,82 @@ def test_has_none_or_all_keys_none() -> None: assert cv.has_none_or_all_keys("a", "b")(obj) is obj +def test_only_when_not_dict() -> None: + with pytest.raises(Invalid, match="expected dictionary"): + cv.only_when("path", "transport", "ws")(5) + + +@pytest.mark.parametrize( + "obj", + [ + {"transport": "ws", "path": "/mqtt"}, + {"transport": "wss", "path": "/mqtt"}, + {"transport": "tcp"}, + {}, + ], +) +def test_only_when_ok(obj: dict) -> None: + assert cv.only_when("path", "transport", ("ws", "wss"))(obj) is obj + + +def test_only_when_rejects_each_key() -> None: + with pytest.raises(vol.MultipleInvalid) as exc: + cv.only_when(("a", "b"), "mode", "range")({"mode": "normal", "a": 1, "b": 2}) + assert [str(e) for e in exc.value.errors] == [ + "'a' is only valid when 'mode' is 'range' @ data['a']", + "'b' is only valid when 'mode' is 'range' @ data['b']", + ] + + +def test_only_when_missing_selector() -> None: + with pytest.raises( + Invalid, match="'path' is only valid when 'transport' is 'ws' or 'wss'" + ): + cv.only_when("path", "transport", ("ws", "wss"))({"path": "/mqtt"}) + + +@pytest.mark.parametrize( + ("value", "actual", "shown"), [(True, False, "true"), (1, True, "1")] +) +def test_only_when_bool_selector(value: object, actual: object, shown: str) -> None: + validator = cv.only_when("a", "mode", value) + assert validator({"mode": value, "a": 1}) + with pytest.raises(Invalid, match=f"when 'mode' is {shown}"): + validator({"mode": actual, "a": 1}) + + +def test_only_when_negate() -> None: + validator = cv.only_when("dc_pin", "bus_mode", "quad", negate=True) + assert validator({"bus_mode": "single", "dc_pin": 4}) + with pytest.raises(Invalid, match="when 'bus_mode' is not 'quad'"): + validator({"bus_mode": "quad", "dc_pin": 4}) + + +@pytest.mark.parametrize("obj", [{"a": 1, "b": 2, "c": 3}, {"b": 2}, {}]) +def test_key_requires_ok(obj: dict) -> None: + assert cv.key_requires("a", "b", "c")(obj) is obj + + +def test_key_requires_names_missing_keys() -> None: + validator = cv.key_requires("a", "b", "c", "d") + with pytest.raises(Invalid, match="^'a' requires 'c' and 'd'"): + validator({"a": 1, "b": 2}) + + +def test_hint_is_added_to_error() -> None: + with pytest.raises(Invalid, match="^'a' requires 'b'. Set b first"): + cv.key_requires("a", "b", hint="Set b first")({"a": 1}) + with pytest.raises(Invalid, match="^'a' is only valid when 'm' is 1. Use m 1"): + cv.only_when("a", "m", 1, hint="Use m 1")({"a": 1, "m": 2}) + + +def test_key_requires_any_of() -> None: + validator = cv.key_requires("a", "b", "c", any_of=True) + assert validator({"a": 1, "c": 3}) + with pytest.raises(Invalid, match="'a' requires 'b' or 'c'"): + validator({"a": 1}) + + # --------------------------------------------------------------------------- # time_period_str_colon / time_period_str_unit # ---------------------------------------------------------------------------