github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-09-25 14:00:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

[core] Fix app_state_ status bits clobbered for non-looping components (#15658)

Browse Source
This commit is contained in:
J. Nick Koston
2026-04-14 07:50:11 -10:00
committed by GitHub
parent cf01163c8c
commit da9fbb8044
6 changed files with 416 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/fixtures/status_flags.yaml
+141
View File
@@ -0,0 +1,141 @@
esphome:
name: status-flags-test
host:
api:
actions:
# Warning flag services for sensor_a
- action: set_warning_a
then:
- lambda: "id(sensor_a)->status_set_warning();"
- component.update: app_warning_bit
- component.update: app_error_bit
- action: clear_warning_a
then:
- lambda: "id(sensor_a)->status_clear_warning();"
- component.update: app_warning_bit
- component.update: app_error_bit
# Warning flag services for sensor_b
- action: set_warning_b
then:
- lambda: "id(sensor_b)->status_set_warning();"
- component.update: app_warning_bit
- component.update: app_error_bit
- action: clear_warning_b
then:
- lambda: "id(sensor_b)->status_clear_warning();"
- component.update: app_warning_bit
- component.update: app_error_bit
# Error flag services for sensor_a
- action: set_error_a
then:
- lambda: "id(sensor_a)->status_set_error();"
- component.update: app_warning_bit
- component.update: app_error_bit
- action: clear_error_a
then:
- lambda: "id(sensor_a)->status_clear_error();"
- component.update: app_warning_bit
- component.update: app_error_bit
# Error flag services for sensor_b
- action: set_error_b
then:
- lambda: "id(sensor_b)->status_set_error();"
- component.update: app_warning_bit
- component.update: app_error_bit
- action: clear_error_b
then:
- lambda: "id(sensor_b)->status_clear_error();"
- component.update: app_warning_bit
- component.update: app_error_bit
# Snapshot of the status_led_light's output state for observation.
- action: snapshot_led
then:
- component.update: status_led_writes
- component.update: status_led_last_state
logger:
# Tracks each write to the fake status_led output.
globals:
- id: status_led_write_count
type: uint32_t
restore_value: no
initial_value: "0"
- id: status_led_last_write
type: bool
restore_value: no
initial_value: "false"
# Fake binary output — status_led_light writes to this instead of a pin.
# Every write bumps a counter and records the last value, both of which
# are exposed below so the test can verify status_led_light's loop is
# actually reading App.get_app_state() and responding.
output:
- platform: template
id: fake_status_led
type: binary
write_action:
- globals.set:
id: status_led_write_count
value: !lambda "return id(status_led_write_count) + 1;"
- globals.set:
id: status_led_last_write
value: !lambda "return state;"
# Actual status_led_light component under test.
light:
- platform: status_led
name: Status LED
id: status_led_light_id
output: fake_status_led
sensor:
# Two components that the test will toggle warning/error flags on.
- platform: template
name: Sensor A
id: sensor_a
update_interval: 24h
lambda: return 1.0;
- platform: template
name: Sensor B
id: sensor_b
update_interval: 24h
lambda: return 2.0;
# Expose App.app_state_'s STATUS_LED_WARNING / STATUS_LED_ERROR bits
# as 0.0 / 1.0. force_update ensures every manual component.update
# publishes even if the value is unchanged.
- platform: template
name: App Warning Bit
id: app_warning_bit
update_interval: 24h
force_update: true
lambda: |-
return (App.get_app_state() & STATUS_LED_WARNING) != 0 ? 1.0 : 0.0;
- platform: template
name: App Error Bit
id: app_error_bit
update_interval: 24h
force_update: true
lambda: |-
return (App.get_app_state() & STATUS_LED_ERROR) != 0 ? 1.0 : 0.0;
# Observables for the fake status_led output.
- platform: template
name: Status LED Writes
id: status_led_writes
update_interval: 24h
force_update: true
lambda: return id(status_led_write_count);
- platform: template
name: Status LED Last State
id: status_led_last_state
update_interval: 24h
force_update: true
lambda: |-
return id(status_led_last_write) ? 1.0 : 0.0;
tests/integration/test_status_flags.py
+209
View File
@@ -0,0 +1,209 @@
"""Integration tests for Component::status_set/clear_warning/error propagation.
Verifies that toggling STATUS_LED_WARNING / STATUS_LED_ERROR on individual
components correctly updates the app-wide bits on Application::app_state_,
AND that the status_led_light component actually responds to those bits
by writing to its output (the full chain from component.status_set_warning
→ App.app_state_ → status_led_light.loop() reading get_app_state()).
Exercises the multi-component OR semantics (the app bit stays set while
any component still has the flag, and only clears when the last component
clears its bit), the independence of warning and error, and the actual
status_led_light read of the bits via a fake template output that counts
writes.
"""
from __future__ import annotations
import asyncio
import pytest
from .state_utils import InitialStateHelper, SensorTracker, build_key_to_entity_mapping
from .types import APIClientConnectedFactory, RunCompiledFunction
# Time to let the host-mode main loop run so status_led_light.loop() can
# execute enough iterations to produce measurable write-count changes on
# the fake template output. 300 ms is well above the minimum needed.
STATUS_LED_SETTLE_S = 0.3
@pytest.mark.asyncio
async def test_status_flags(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
async with run_compiled(yaml_config), api_client_connected() as client:
entities, services = await client.list_entities_services()
# Map every custom API service by name for the test to execute.
svc = {s.name: s for s in services}
for name in (
"set_warning_a",
"clear_warning_a",
"set_warning_b",
"clear_warning_b",
"set_error_a",
"clear_error_a",
"set_error_b",
"clear_error_b",
"snapshot_led",
):
assert name in svc, f"service {name} not registered"
# Track every sensor we care about. SensorTracker gives us
# expect(value) / expect_any() futures that resolve when a
# matching state arrives; much simpler than manual bookkeeping.
tracker = SensorTracker(
[
"app_warning_bit",
"app_error_bit",
"status_led_writes",
"status_led_last_state",
]
)
tracker.key_to_sensor.update(
build_key_to_entity_mapping(entities, list(tracker.sensor_states.keys()))
)
# Swallow initial state broadcasts so the test only reacts to
# state changes triggered by our service calls.
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(initial_state_helper.on_state_wrapper(tracker.on_state))
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
async def call(name: str) -> None:
await client.execute_service(svc[name], {})
async def call_and_expect_bits(
service_name: str, *, warning: float, error: float
) -> None:
"""Execute a service and wait for both app bit sensors to match.
Each bit-toggling service calls component.update on both
app_warning_bit and app_error_bit, so both sensors publish.
"""
futures = tracker.expect_all(
{"app_warning_bit": warning, "app_error_bit": error}
)
await call(service_name)
await tracker.await_all(futures)
async def snapshot_led_writes() -> int:
"""Trigger a publish of the fake status_led output counter and return it."""
future = tracker.expect_any("status_led_writes")
await call("snapshot_led")
await tracker.await_change(future, "status_led_writes")
return int(tracker.sensor_states["status_led_writes"][-1])
# ---- Baseline: everything clean ----
await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
# ================================================================
# Part 1 — STATUS_LED_WARNING propagation to App.app_state_
# ================================================================
# Single component set/clear
await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
# Multi-component OR: both set, clear A, bit stays (B still has it), clear B, gone
await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
await call_and_expect_bits("set_warning_b", warning=1.0, error=0.0)
await call_and_expect_bits("clear_warning_a", warning=1.0, error=0.0)
await call_and_expect_bits("clear_warning_b", warning=0.0, error=0.0)
# Opposite clear order
await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
await call_and_expect_bits("set_warning_b", warning=1.0, error=0.0)
await call_and_expect_bits("clear_warning_b", warning=1.0, error=0.0)
await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
# ================================================================
# Part 2 — STATUS_LED_ERROR propagation (same scenarios)
# ================================================================
await call_and_expect_bits("set_error_a", warning=0.0, error=1.0)
await call_and_expect_bits("clear_error_a", warning=0.0, error=0.0)
await call_and_expect_bits("set_error_a", warning=0.0, error=1.0)
await call_and_expect_bits("set_error_b", warning=0.0, error=1.0)
await call_and_expect_bits("clear_error_a", warning=0.0, error=1.0)
await call_and_expect_bits("clear_error_b", warning=0.0, error=0.0)
# ================================================================
# Part 3 — warning and error are independent
# ================================================================
await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
await call_and_expect_bits("set_error_b", warning=1.0, error=1.0)
await call_and_expect_bits("clear_warning_a", warning=0.0, error=1.0)
await call_and_expect_bits("clear_error_b", warning=0.0, error=0.0)
# ================================================================
# Part 4 — status_led_light actually reads App.app_state_
# ================================================================
# The fake status_led_light output increments status_led_write_count
# on every write. status_led_light::loop() writes its output on every
# iteration while an error/warning bit is set, so after holding a
# warning for ~300 ms we should see the counter move significantly.
# This is the end-to-end proof that the bits we set above actually
# reach status_led_light and drive its behavior.
count_before_warning = await snapshot_led_writes()
await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
# Let status_led_light's loop run long enough to toggle the pin
# several times (it reads get_app_state() every main loop iteration).
await asyncio.sleep(STATUS_LED_SETTLE_S)
count_after_warning = await snapshot_led_writes()
assert count_after_warning > count_before_warning, (
"status_led_light did not respond to STATUS_LED_WARNING being set: "
f"write count stayed at {count_before_warning} → {count_after_warning}. "
"The full chain Component::status_set_warning → App.app_state_ → "
"status_led_light::loop reading get_app_state() is broken."
)
await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
# Same check for ERROR
count_before_error = await snapshot_led_writes()
await call_and_expect_bits("set_error_a", warning=0.0, error=1.0)
await asyncio.sleep(STATUS_LED_SETTLE_S)
count_after_error = await snapshot_led_writes()
assert count_after_error > count_before_error, (
"status_led_light did not respond to STATUS_LED_ERROR being set: "
f"write count stayed at {count_before_error} → {count_after_error}. "
)
await call_and_expect_bits("clear_error_a", warning=0.0, error=0.0)
# ---- Set → clear → re-set round-trip ----
# After clearing, status_led_light stops writing (steady state).
# Re-setting the flag must make it resume. This guards against a
# future idle optimization (e.g. #15642) where status_led disables
# its own loop when idle: if the re-enable path were broken, the
# second set would not produce writes.
#
# Snapshot AFTER the clear to avoid counting writes that were still
# in-flight from the error-set phase.
count_after_clear = await snapshot_led_writes()
await asyncio.sleep(STATUS_LED_SETTLE_S)
count_after_idle = await snapshot_led_writes()
assert count_after_idle - count_after_clear <= 5, (
"status_led_light kept writing after warning/error was cleared: "
f"count grew from {count_after_clear} to {count_after_idle}. "
"Expected it to stop writing once all status bits were clear."
)
# Re-set warning — writes must resume.
await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
await asyncio.sleep(STATUS_LED_SETTLE_S)
count_after_reset = await snapshot_led_writes()
assert count_after_reset > count_after_idle + 5, (
"status_led_light did not resume writing after re-setting "
f"STATUS_LED_WARNING: count went from {count_after_idle} to "
f"{count_after_reset}. If an idle optimization disabled the "
"loop, the re-enable path may be broken."
)
await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)