Compare commits

..

1 Commits

Author SHA1 Message Date
J. Nick Koston
3bbd941939 Add additional text_sensor filter tests 2026-01-23 02:32:02 -10:00
4 changed files with 254 additions and 11 deletions

View File

@@ -533,17 +533,6 @@ static void set_json_id(JsonObject &root, EntityBase *obj, const char *prefix, J
root[ESPHOME_F("id")] = id_buf;
// Add legacy_id for backward compatibility with third-party integrations
// Old format: {prefix}-{object_id} (e.g., "cover-garage_door")
// Remove before 2026.7.0 when object_id support is fully removed
char legacy_buf[ESPHOME_DOMAIN_MAX_LEN + 1 + OBJECT_ID_MAX_LEN];
char *lp = legacy_buf;
memcpy(lp, prefix, prefix_len);
lp += prefix_len;
*lp++ = '-';
obj->write_object_id_to(lp, sizeof(legacy_buf) - (lp - legacy_buf));
root[ESPHOME_F("legacy_id")] = legacy_buf;
if (start_config == DETAIL_ALL) {
root[ESPHOME_F("domain")] = prefix;
root[ESPHOME_F("name")] = name;

View File

@@ -64,3 +64,16 @@ text_sensor:
- suffix -> SUFFIX
- map:
- PREFIX text SUFFIX -> mapped
- platform: template
name: "Test Lambda Filter"
id: test_lambda_filter
filters:
- lambda: |-
return {"[" + x + "]"};
- to_upper
- lambda: |-
if (x.length() > 10) {
return {x.substr(0, 10) + "..."};
}
return {x};

View File

@@ -56,6 +56,36 @@ text_sensor:
- prepend: "["
- append: "]"
- platform: template
name: "To Lower Sensor"
id: to_lower_sensor
filters:
- to_lower
- platform: template
name: "Lambda Sensor"
id: lambda_sensor
filters:
- lambda: |-
return {"[" + x + "]"};
- platform: template
name: "Lambda Raw State Sensor"
id: lambda_raw_state_sensor
filters:
- lambda: |-
return {x + " MODIFIED"};
- platform: template
name: "Lambda Skip Sensor"
id: lambda_skip_sensor
filters:
- lambda: |-
if (x == "skip") {
return {};
}
return {x + " passed"};
# Button to publish values and log raw_state vs state
button:
- platform: template
@@ -179,3 +209,73 @@ button:
format: "CHAINED: state='%s'"
args:
- id(chained_sensor).state.c_str()
- platform: template
name: "Test To Lower Button"
id: test_to_lower_button
on_press:
- text_sensor.template.publish:
id: to_lower_sensor
state: "HELLO WORLD"
- delay: 50ms
- logger.log:
format: "TO_LOWER: state='%s'"
args:
- id(to_lower_sensor).state.c_str()
- platform: template
name: "Test Lambda Button"
id: test_lambda_button
on_press:
- text_sensor.template.publish:
id: lambda_sensor
state: "test"
- delay: 50ms
- logger.log:
format: "LAMBDA: state='%s'"
args:
- id(lambda_sensor).state.c_str()
- platform: template
name: "Test Lambda Pass Button"
id: test_lambda_pass_button
on_press:
- text_sensor.template.publish:
id: lambda_skip_sensor
state: "value"
- delay: 50ms
- logger.log:
format: "LAMBDA_PASS: state='%s'"
args:
- id(lambda_skip_sensor).state.c_str()
- platform: template
name: "Test Lambda Skip Button"
id: test_lambda_skip_button
on_press:
- text_sensor.template.publish:
id: lambda_skip_sensor
state: "skip"
- delay: 50ms
# When lambda returns {}, the value should NOT be published
# so state should remain from previous publish (or empty if first)
- logger.log:
format: "LAMBDA_SKIP: state='%s'"
args:
- id(lambda_skip_sensor).state.c_str()
- platform: template
name: "Test Lambda Raw State Button"
id: test_lambda_raw_state_button
on_press:
- text_sensor.template.publish:
id: lambda_raw_state_sensor
state: "original"
- delay: 50ms
# Verify raw_state is preserved (not mutated) after lambda filter
# state should be "original MODIFIED", raw_state should be "original"
- logger.log:
format: "LAMBDA_RAW_STATE: state='%s' raw_state='%s'"
args:
- id(lambda_raw_state_sensor).state.c_str()
- id(lambda_raw_state_sensor).get_raw_state().c_str()

View File

@@ -42,6 +42,11 @@ async def test_text_sensor_raw_state(
map_off_future: asyncio.Future[str] = loop.create_future()
map_unknown_future: asyncio.Future[str] = loop.create_future()
chained_future: asyncio.Future[str] = loop.create_future()
to_lower_future: asyncio.Future[str] = loop.create_future()
lambda_future: asyncio.Future[str] = loop.create_future()
lambda_pass_future: asyncio.Future[str] = loop.create_future()
lambda_skip_future: asyncio.Future[str] = loop.create_future()
lambda_raw_state_future: asyncio.Future[tuple[str, str]] = loop.create_future()
# Patterns to match log output
# NO_FILTER: state='hello world' raw_state='hello world'
@@ -58,6 +63,13 @@ async def test_text_sensor_raw_state(
map_off_pattern = re.compile(r"MAP_OFF: state='([^']*)'")
map_unknown_pattern = re.compile(r"MAP_UNKNOWN: state='([^']*)'")
chained_pattern = re.compile(r"CHAINED: state='([^']*)'")
to_lower_pattern = re.compile(r"TO_LOWER: state='([^']*)'")
lambda_pattern = re.compile(r"LAMBDA: state='([^']*)'")
lambda_pass_pattern = re.compile(r"LAMBDA_PASS: state='([^']*)'")
lambda_skip_pattern = re.compile(r"LAMBDA_SKIP: state='([^']*)'")
lambda_raw_state_pattern = re.compile(
r"LAMBDA_RAW_STATE: state='([^']*)' raw_state='([^']*)'"
)
def check_output(line: str) -> None:
"""Check log output for expected messages."""
@@ -92,6 +104,27 @@ async def test_text_sensor_raw_state(
if not chained_future.done() and (match := chained_pattern.search(line)):
chained_future.set_result(match.group(1))
if not to_lower_future.done() and (match := to_lower_pattern.search(line)):
to_lower_future.set_result(match.group(1))
if not lambda_future.done() and (match := lambda_pattern.search(line)):
lambda_future.set_result(match.group(1))
if not lambda_pass_future.done() and (
match := lambda_pass_pattern.search(line)
):
lambda_pass_future.set_result(match.group(1))
if not lambda_skip_future.done() and (
match := lambda_skip_pattern.search(line)
):
lambda_skip_future.set_result(match.group(1))
if not lambda_raw_state_future.done() and (
match := lambda_raw_state_pattern.search(line)
):
lambda_raw_state_future.set_result((match.group(1), match.group(2)))
async with (
run_compiled(yaml_config, line_callback=check_output),
api_client_connected() as client,
@@ -272,3 +305,111 @@ async def test_text_sensor_raw_state(
pytest.fail("Timeout waiting for CHAINED log message")
assert state == "[value]", f"Chained failed: expected '[value]', got '{state}'"
# Test 10: to_lower filter
# "HELLO WORLD" -> "hello world"
to_lower_button = next(
(e for e in entities if "test_to_lower_button" in e.object_id.lower()),
None,
)
assert to_lower_button is not None, "Test To Lower Button not found"
client.button_command(to_lower_button.key)
try:
state = await asyncio.wait_for(to_lower_future, timeout=5.0)
except TimeoutError:
pytest.fail("Timeout waiting for TO_LOWER log message")
assert state == "hello world", (
f"to_lower failed: expected 'hello world', got '{state}'"
)
# Test 11: Lambda filter
# "test" -> "[test]"
lambda_button = next(
(e for e in entities if "test_lambda_button" in e.object_id.lower()),
None,
)
assert lambda_button is not None, "Test Lambda Button not found"
client.button_command(lambda_button.key)
try:
state = await asyncio.wait_for(lambda_future, timeout=5.0)
except TimeoutError:
pytest.fail("Timeout waiting for LAMBDA log message")
assert state == "[test]", f"Lambda failed: expected '[test]', got '{state}'"
# Test 12: Lambda filter - value passes through
# "value" -> "value passed"
lambda_pass_button = next(
(e for e in entities if "test_lambda_pass_button" in e.object_id.lower()),
None,
)
assert lambda_pass_button is not None, "Test Lambda Pass Button not found"
client.button_command(lambda_pass_button.key)
try:
state = await asyncio.wait_for(lambda_pass_future, timeout=5.0)
except TimeoutError:
pytest.fail("Timeout waiting for LAMBDA_PASS log message")
assert state == "value passed", (
f"Lambda pass failed: expected 'value passed', got '{state}'"
)
# Test 13: Lambda filter - skip publishing (return {})
# "skip" -> no publish, state remains "value passed" from previous test
lambda_skip_button = next(
(e for e in entities if "test_lambda_skip_button" in e.object_id.lower()),
None,
)
assert lambda_skip_button is not None, "Test Lambda Skip Button not found"
client.button_command(lambda_skip_button.key)
try:
state = await asyncio.wait_for(lambda_skip_future, timeout=5.0)
except TimeoutError:
pytest.fail("Timeout waiting for LAMBDA_SKIP log message")
# When lambda returns {}, value should NOT be published
# State remains from previous successful publish ("value passed")
assert state == "value passed", (
f"Lambda skip failed: expected 'value passed' (unchanged), got '{state}'"
)
# Test 14: Lambda filter - verify raw_state is preserved (not mutated)
# This is critical to verify the in-place mutation optimization is safe
# "original" -> state="original MODIFIED", raw_state="original"
lambda_raw_state_button = next(
(
e
for e in entities
if "test_lambda_raw_state_button" in e.object_id.lower()
),
None,
)
assert lambda_raw_state_button is not None, (
"Test Lambda Raw State Button not found"
)
client.button_command(lambda_raw_state_button.key)
try:
state, raw_state = await asyncio.wait_for(
lambda_raw_state_future, timeout=5.0
)
except TimeoutError:
pytest.fail("Timeout waiting for LAMBDA_RAW_STATE log message")
assert state == "original MODIFIED", (
f"Lambda raw_state test failed: expected state='original MODIFIED', "
f"got '{state}'"
)
assert raw_state == "original", (
f"Lambda raw_state test failed: raw_state was mutated! "
f"Expected 'original', got '{raw_state}'"
)
assert state != raw_state, (
f"Lambda filter should modify state but preserve raw_state. "
f"state='{state}', raw_state='{raw_state}'"
)