[core] Add cv.only_when and cv.key_requires key validators (#21234)

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"
 )

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":

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 (

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
 # ---------------------------------------------------------------------------