[core][cover] Add register_apply_condition for conditions that only test the parent (#19540)

tests/components/template/common-base.yaml

@@ -551,11 +551,17 @@ text:
     update_interval: 1s
     lambda: |
       return std::string{"Hello!"};
+    # Exercise a register_apply_condition inside a trigger with a std::string
+    # argument, so ApplyCondition<std::string> is compiled.
     set_action:
       then:
-        - logger.log:
-            format: Template Text set to %s
-            args: ["x.c_str()"]
+        - if:
+            condition:
+              cover.is_open: template_cover_with_triggers
+            then:
+              - logger.log:
+                  format: Template Text set to %s
+                  args: ["x.c_str()"]
 
 alarm_control_panel:
   - platform: template

tests/unit_tests/test_automation.py

@@ -10,6 +10,7 @@ import pytest
 from esphome.automation import (
     ApplyAction,
     ApplyCall,
+    ApplyCondition,
     ApplyField,
     CallbackAutomation,
     TriggerForwarder,
@@ -19,6 +20,7 @@ from esphome.automation import (
     has_non_synchronous_actions,
     maybe_simple_id,
     register_apply_action,
+    register_apply_condition,
     register_bare_action,
     register_bare_condition,
     register_parented_action,
@@ -602,6 +604,20 @@ def test_shared_builders_keep_synchronous_flag(
     assert actions["my.parented"].synchronous is synchronous
 
 
+async def _run_entry(
+    entry: RegistryEntry,
+    config: dict[str, object],
+    args: list[tuple[object, str]] | None,
+    platform: str,
+) -> RegistryEntry:
+    """Run a registered builder with the given config, trigger args and platform."""
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: platform}
+    args = args or []
+    template_arg = cg.TemplateArguments(*(t for t, _ in args))
+    await entry.fun({CONF_ID: PARENT_ID, **config}, ID("obj_1"), template_arg, args)
+    return entry
+
+
 async def _run_apply_action(
     registries: tuple[Registry, Registry],
     fields: tuple[ApplyField | ApplyCall, ...],
@@ -611,14 +627,22 @@ async def _run_apply_action(
     platform: str = "esp32",
 ) -> RegistryEntry:
     """Register an apply action and run its builder with the given config."""
-    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: platform}
     actions, _ = registries
     register_apply_action("my.apply", None, *fields, call=call)
-    entry = actions["my.apply"]
-    args = args or []
-    template_arg = cg.TemplateArguments(*(t for t, _ in args))
-    await entry.fun({CONF_ID: PARENT_ID, **config}, ID("obj_1"), template_arg, args)
-    return entry
+    return await _run_entry(actions["my.apply"], config, args, platform)
+
+
+async def _run_apply_condition(
+    registries: tuple[Registry, Registry],
+    check: str | ApplyCall,
+    config: dict[str, object],
+    args: list[tuple[object, str]] | None = None,
+    platform: str = "esp32",
+) -> RegistryEntry:
+    """Register an apply condition and run its builder with the given config."""
+    _, conditions = registries
+    register_apply_condition("my.check", None, check)
+    return await _run_entry(conditions["my.check"], config, args, platform)
 
 
 def _apply_lambda(mock_cg: MockCodegen) -> str:
@@ -774,6 +798,13 @@ def test_apply_registration_checks(registries: tuple[Registry, Registry]) -> Non
     register_apply_action("my.ok", schema, ApplyField("kp", "set_kp", cg.float_))
     with pytest.raises(ValueError, match="'kd' is not in the schema"):
         register_apply_action("my.bad", schema, ApplyField("kd", "set_kd", cg.float_))
+    register_apply_condition(
+        "my.is", schema, ApplyCall("kp == {}", (("kp", cg.float_),))
+    )
+    with pytest.raises(ValueError, match="'kd' is not in the schema"):
+        register_apply_condition(
+            "my.bad_is", schema, ApplyCall("kd == {}", (("kd", cg.float_),))
+        )
     either = cv.Any(schema, cv.Schema({cv.Optional("kd"): cv.float_}))
     register_apply_action("my.any", either, ApplyField("kd", "set_kd", cg.float_))
     for wrapped in (
@@ -805,3 +836,67 @@ async def test_apply_string_constant_stays_in_flash_on_esp8266(
     assert f'::{PARENT_OBJ}->play(progmem_string(ESPHOME_F("a:b")));' in _apply_lambda(
         mock_cg
     )
+
+
+@pytest.mark.asyncio
+async def test_register_apply_condition_predicate(
+    registries: tuple[Registry, Registry], mock_cg: MockCodegen
+) -> None:
+    entry = await _run_apply_condition(
+        registries, "is_playing()", {}, args=[(cg.int32, "x")]
+    )
+    assert entry.type_id is ApplyCondition
+    condition_id, template_arg, check = mock_cg.new_pvariable.call_args.args
+    assert condition_id == ID("obj_1")
+    assert str(template_arg) == "<int32_t>"
+    text = str(check)
+    assert text.startswith("[](const std::remove_cvref_t<int32_t> & x) -> bool {")
+    assert f"return ::{PARENT_OBJ}->is_playing();" in text
+
+
+@pytest.mark.asyncio
+async def test_apply_condition_compares_config_value(
+    registries: tuple[Registry, Registry], mock_cg: MockCodegen
+) -> None:
+    check = ApplyCall("state == {}", (("state", cg.bool_),))
+    await _run_apply_condition(registries, check, {"state": True})
+    assert f"return ::{PARENT_OBJ}->state == true;" in _apply_lambda(mock_cg)
+
+    with pytest.raises(EsphomeError, match="needs all of"):
+        await _run_apply_condition(registries, check, {})
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("platform", ["esp32", "esp8266"])
+async def test_apply_condition_string_constant_is_a_plain_literal(
+    registries: tuple[Registry, Registry], mock_cg: MockCodegen, platform: str
+) -> None:
+    check = ApplyCall("state == {}", (("state", cg.std_string),))
+    await _run_apply_condition(registries, check, {"state": "two"}, platform=platform)
+    assert f'return ::{PARENT_OBJ}->state == "two";' in _apply_lambda(mock_cg)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("body", "expected", "called"),
+    [
+        ("return x;", "->state == (x);", False),
+        ('return x.empty() ? "e" : x;', '->state == (x.empty() ? "e" : x);', False),
+        ('if (x.empty()) return "e";\nreturn x;', "}(x);", True),
+    ],
+)
+async def test_apply_condition_string_lambda_paths(
+    registries: tuple[Registry, Registry],
+    mock_cg: MockCodegen,
+    body: str,
+    expected: str,
+    called: bool,
+) -> None:
+    """A single return is inlined with no copy; a longer body is a called std::string lambda."""
+    check = ApplyCall("state == {}", (("state", cg.std_string),))
+    await _run_apply_condition(
+        registries, check, {"state": Lambda(body)}, args=[(cg.std_string, "x")]
+    )
+    text = _apply_lambda(mock_cg)
+    assert expected in text
+    assert ("-> std::string {" in text) is called