mirror of
https://github.com/esphome/esphome.git
synced 2026-09-23 04:58:38 +00:00
J. Nick Koston
424f1c7b0a
Drop the trailing underscore from any_component_has_status_flag now that the method is public. Trailing underscores in the codebase are reserved for protected/private members per clang-tidy naming rules, which caused a CI failure on the previously-named public helper. Add integration tests covering: - Single-component status_set/clear for warning and error - Multi-component OR semantics (both clear orders) - Warning and error independence - End-to-end proof that status_led_light::loop() reads App.app_state_ and writes its output when the bits are set (via a fake template output whose write_action bumps a counter exposed as a sensor)
180 lines
8.4 KiB
Python
180 lines
8.4 KiB
Python
"""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
|
|
|
|
|
|
@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, record the LED write count ----
|
|
await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
|
|
baseline_led_count = await snapshot_led_writes()
|
|
|
|
# ================================================================
|
|
# 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(0.3)
|
|
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(0.3)
|
|
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)
|
|
|
|
# Sanity: baseline snapshot used earlier isn't stale, counter is monotonic
|
|
assert count_after_error >= baseline_led_count
|