Compare commits

...
2 changed files with 88 additions and 8 deletions
+10 -4
View File
@@ -524,7 +524,8 @@ class SubEntities:
"""Create the entity from ``conf``; ``setter`` may be None when the hub keeps no pointer.
With ``apply`` the entity is the domain's ``Apply*`` class and its control runs
``parent->apply``, with ``publish`` ("before"/"after") placing the domain's publish.
``parent->apply``, or ``apply`` as a namespaced free statement when no parent is bound, with
``publish`` ("before"/"after") placing the domain's publish.
"""
entity_id: ID = conf[CONF_ID]
is_apply_class = self.apply_entity is not None and str(entity_id.type) == str(
@@ -536,8 +537,13 @@ class SubEntities:
raise ValueError(
f"{entity_id}: apply= needs the domain's Apply class, not {entity_id.type}"
)
if apply is not None and self.parent is None:
raise ValueError(f"{entity_id}: apply= needs a parent")
# A parentless statement is a namespaced free function, so a hub platform that forgot
# parent= fails here instead of in the generated C++
if apply is not None and self.parent is None and "::" not in apply:
raise ValueError(
f"{entity_id}: apply= without a parent must call a namespaced free function "
f"(ns::fn()); bind parent= for a hub method"
)
if publish is not None and (apply is None or self.apply_entity.publish is None):
raise ValueError(f"{entity_id}: publish= needs apply= on a stateful domain")
if publish not in (None, "before", "after"):
@@ -562,7 +568,7 @@ class SubEntities:
if isinstance(parent, ID):
parent = await get_variable(parent)
apply_entity = self.apply_entity
body = [f"{parent_ref(parent)}->{apply};"]
body = [f"{apply};" if parent is None else f"{parent_ref(parent)}->{apply};"]
if publish == "before":
body.insert(0, apply_entity.publish)
elif publish == "after":
+78 -4
View File
@@ -1534,13 +1534,87 @@ async def test_sub_entities_apply_errors(
@pytest.mark.asyncio
async def test_sub_entities_apply_needs_parent(
async def test_sub_entities_apply_without_parent_needs_namespaced_call(
setup_test_environment: list[str],
apply_functions: list[tuple[str, str, str]],
) -> None:
"""A hub method with no parent bound fails in Python, not as an undeclared name in C++."""
calls: list[tuple[Any, ...]] = []
conf = _apply_conf("k_id", _BUTTON_APPLY.cls)
sub = _apply_sub(calls, {"k": conf}, _BUTTON_APPLY, parent=None)
with pytest.raises(ValueError, match="namespaced free function"):
await sub("k", None, apply="restart()")
assert calls == []
assert apply_functions == []
@pytest.mark.asyncio
@pytest.mark.parametrize(
("apply_entity", "type_", "publish", "params", "body"),
[
(
_BUTTON_APPLY,
_BUTTON_APPLY.cls,
None,
"button::Button * entity",
"restart::restart_device();",
),
(
_SWITCH_APPLY,
_APPLY_SWITCH_CLS,
"after",
"switch_::Switch * entity, bool state",
"restart::restart_device();\nentity->publish_state(state);",
),
],
)
async def test_sub_entities_apply_without_parent(
monkeypatch: pytest.MonkeyPatch,
setup_test_environment: list[str],
apply_functions: list[tuple[str, str, str]],
apply_entity: entity_helpers.ApplyEntity,
type_: MockObjClass,
publish: str | None,
params: str,
body: str,
) -> None:
"""Without a parent the statement runs as written, with no parent lookup or deref."""
async def fail(*args: Any) -> None:
raise AssertionError("must not be called")
monkeypatch.setattr(entity_helpers.cg, "register_parented", fail)
monkeypatch.setattr(entity_helpers, "get_variable", fail)
calls: list[tuple[Any, ...]] = []
sub = _apply_sub(calls, {}, apply_entity, parent=None)
var = await sub.create(
_apply_conf("rst", type_),
None,
apply="restart::restart_device()",
publish=publish,
)
assert str(var) == "rst"
assert calls == [("rst", "<esphome__rst__fn>")]
assert apply_functions == [("esphome__rst__fn", params, body)]
assert setup_test_environment == []
@pytest.mark.asyncio
async def test_sub_entities_apply_without_parent_shares_equal_functions(
setup_test_environment: list[str], apply_functions: list[tuple[str, str, str]]
) -> None:
"""Parentless entities with the same statement share one function."""
calls: list[tuple[Any, ...]] = []
sub = _apply_sub(calls, {"k": _apply_conf("k_id")}, parent=None)
with pytest.raises(ValueError, match="needs a parent"):
await sub("k", None, apply="x()")
sub = _apply_sub(calls, {}, _BUTTON_APPLY, parent=None)
for name in ("a_btn", "b_btn"):
await sub.create(_apply_conf(name, _BUTTON_APPLY.cls), None, apply="x::go()")
await sub.create(_apply_conf("c_btn", _BUTTON_APPLY.cls), None, apply="x::stop()")
assert calls == [
("a_btn", "<esphome__a_btn__fn>"),
("b_btn", "<esphome__a_btn__fn>"),
("c_btn", "<esphome__c_btn__fn>"),
]
assert [body for _, _, body in apply_functions] == ["x::go();", "x::stop();"]
@pytest.mark.parametrize(