[tests] Use the shared press_and_wait and next_state in the action and light tests

tests/integration/README.md

@@ -58,8 +58,30 @@ The `InitialStateHelper` class solves a common problem in integration tests: whe
 - Provides debug logging to track state reception (use `--log-cli-level=DEBUG`)
 - Safe for concurrent use with multiple entity types
 
-**Future work:**
-Consider converting existing integration tests to use `InitialStateHelper` for more reliable state tracking and to eliminate race conditions related to initial state broadcasts.
+#### Subscribing after the initial states (`state_utils.py`)
+
+`subscribe_after_initial_states()` does the `InitialStateHelper` setup in one call: it subscribes, waits
+for every entity's connect time state, and fails the test with the entities still missing on timeout.
+It returns a `StateSubscription` with `initial_states` (by key) and a `waiter`, a `StateWaiter` that sees
+every later state. `waiter.next_state(key=..., state_type=...)` arms a wait for the next matching state;
+call it before sending the command. `press_and_wait()` presses a button and returns the next state of a
+type, from one entity when `key=` is given:
+
+```python
+from .state_utils import press_and_wait, subscribe_after_initial_states
+
+entities, _ = await client.list_entities_services()
+sub = await subscribe_after_initial_states(client, entities)
+
+state = await press_and_wait(client, entities, sub.waiter, "set_position", CoverState)
+assert state.position == pytest.approx(0.5, abs=0.01)
+
+next_light = sub.waiter.next_state(key=light.key, state_type=LightState)
+client.light_command(key=light.key, state=True)
+assert (await next_light).state is True
+```
+
+Pass `on_state=` as well when the test also needs to collect every state.
 
 #### Entity Lookup Helpers (`state_utils.py`)
 
@@ -203,46 +225,24 @@ async def test_my_sensor(
 
 ##### State Subscription Pattern
 
-**Recommended: Using InitialStateHelper**
+**Recommended: Using subscribe_after_initial_states**
 
-When an API client connects, ESPHome automatically sends the current state of all entities. The `InitialStateHelper` (from `state_utils.py`) handles this by swallowing these initial states and only forwarding subsequent state changes to your test callback:
+When an API client connects, ESPHome automatically sends the current state of all entities.
+`subscribe_after_initial_states()` (from `state_utils.py`) swallows these initial states and only forwards
+subsequent state changes, to your callback and to the returned `waiter`:
 
 ```python
-from .state_utils import InitialStateHelper
+from .state_utils import subscribe_after_initial_states
 
-# Track state changes with futures
-loop = asyncio.get_running_loop()
-states: dict[int, EntityState] = {}
-state_future: asyncio.Future[EntityState] = loop.create_future()
-
-
-def on_state(state: EntityState) -> None:
-    """This callback only receives NEW state changes, not initial states."""
-    states[state.key] = state
-    # Check for specific condition using isinstance
-    if isinstance(state, SensorState) and state.state == expected_value:
-        if not state_future.done():
-            state_future.set_result(state)
-
-
-# Get entities and set up state synchronization
 entities, services = await client.list_entities_services()
-initial_state_helper = InitialStateHelper(entities)
+sub = await subscribe_after_initial_states(client, entities)
 
-# Subscribe with the wrapper that filters initial states
-client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
-# Wait for all initial states to be broadcast; on timeout this fails the test
-# and lists the entities that never sent one
-await initial_state_helper.wait_for_initial_states()
-
-# Now perform your test actions - on_state will only receive new changes
-# ... trigger state changes ...
-
-# Wait for expected state; on timeout the test fails and shows the states received
-result = await wait_or_fail(
-    state_future, 5.0, "Expected state not received", states=states
+# Arm the wait before the action that produces the state
+expected = sub.waiter.expect(
+    lambda s: isinstance(s, SensorState) and s.state == expected_value
 )
+# ... trigger state changes ...
+result = await expected
 ```
 
 **Legacy: Manual State Tracking**
@@ -386,38 +386,34 @@ async with run_compiled(yaml_config, line_callback=on_log_line):
     # Test implementation
 ```
 
-Example using futures for specific log patterns:
+To wait for log lines, use `LineWaiter` (`log_utils.py`) instead of a hand written
+callback with one future per pattern:
 ```python
 import re
 
-loop = asyncio.get_running_loop()
-connected_future = loop.create_future()
-service_future = loop.create_future()
+from .log_utils import LineWaiter
 
-# Patterns to match
-connected_pattern = re.compile(r"Client .* connected from")
-service_pattern = re.compile(r"Service called")
+log = LineWaiter()
+# Arm expectations before the action that logs them; any number may be pending.
+connected = log.expect(re.compile(r"Client .* connected from"))
+value = log.expect(re.compile(r"value=(\d+)"))
 
-
-def check_output(line: str) -> None:
-    """Check log output for expected messages."""
-    if not connected_future.done() and connected_pattern.search(line):
-        connected_future.set_result(True)
-    elif not service_future.done() and service_pattern.search(line):
-        service_future.set_result(True)
-
-
-async with run_compiled(yaml_config, line_callback=check_output):
+async with run_compiled(yaml_config, line_callback=log.callback):
     async with api_client_connected() as client:
-        # Wait for specific log message
-        await asyncio.wait_for(connected_future, timeout=5.0)
-
-        # Do test actions...
-
-        # Wait for service log
-        await asyncio.wait_for(service_future, timeout=5.0)
+        await log.wait(connected)
+        match = await log.wait(value, timeout=5.0, msg="no value logged")
+        assert match.group(1) == "42"
+        assert log.count("Service called") == 1
 ```
 
+- `expect(pattern)` only matches lines that arrive after the call. A `str` matches as a
+  substring and resolves to the line; a compiled regex is searched and resolves to its match.
+  Each expectation resolves on the next matching line, so two armed for the same text both
+  resolve on its first occurrence: arm the second after the first resolves, or use `count`.
+- `wait(future)` fails the test on timeout with the pattern and the last 20 log lines.
+- `wait_for(*needles)` also matches lines seen before the call, but only one may be pending.
+- `count(pattern)` counts the lines seen so far.
+
 **Note**: Tests that monitor log messages typically have fewer race conditions compared to state-based testing, making them more reliable. However, be aware that the host platform currently does not have a thread-safe logger, so logging from threads will not work correctly.
 
 ##### Timeout Handling

tests/integration/log_utils.py

@@ -3,23 +3,47 @@
 from __future__ import annotations
 
 import asyncio
+import re
+from typing import Any, overload
+
+from .wait_utils import wait_or_fail
+
+type LinePattern = str | re.Pattern[str]
+type LineMatch = str | re.Match[str]
+
+
+def _match(pattern: LinePattern, line: str) -> LineMatch | None:
+    """Return ``line`` if a str pattern is in it, or the search match of a regex."""
+    if isinstance(pattern, str):
+        return line if pattern in line else None
+    return pattern.search(line)
 
 
 class LineWaiter:
-    """Collects log lines and lets a test await one containing all needles.
+    """Collects log lines and lets a test await lines that match.
 
     Pass ``callback`` as ``run_compiled``'s ``line_callback``; the callback runs
-    on the test's own event loop, so futures are resolved directly. Only one
-    ``wait_for`` may be outstanding at a time (tests await sequentially).
+    on the test's own event loop, so futures are resolved directly. ``wait_for``
+    also matches lines seen before the call, and only one may be outstanding at a
+    time. ``expect`` matches only later lines, and any number may be armed at once;
+    each resolves on the next matching line, so two armed for the same text both
+    resolve on its first occurrence. Arm the second after the first, or use ``count``.
     """
 
     def __init__(self) -> None:
         self.lines: list[str] = []
         self._needles: tuple[str, ...] = ()
         self._future: asyncio.Future | None = None
+        self._expectations: dict[asyncio.Future[Any], LinePattern] = {}
 
     def callback(self, line: str) -> None:
         self.lines.append(line)
+        for future, pattern in list(self._expectations.items()):
+            if future.done():
+                del self._expectations[future]
+            elif (match := _match(pattern, line)) is not None:
+                future.set_result(match)
+                del self._expectations[future]
         if (
             self._future is not None
             and not self._future.done()
@@ -46,3 +70,38 @@ class LineWaiter:
         finally:
             self._future = None
             self._needles = ()
+
+    @overload
+    def expect(self, pattern: str) -> asyncio.Future[str]: ...  # noqa: E704
+
+    @overload
+    def expect(self, pattern: re.Pattern[str]) -> asyncio.Future[re.Match[str]]: ...  # noqa: E704
+
+    def expect(self, pattern: LinePattern) -> asyncio.Future[LineMatch]:
+        """Arm a wait for the next line matching ``pattern``; earlier lines never match.
+
+        A str matches as a substring and resolves to the line; a compiled regex is
+        searched and resolves to its match. Arm it before the action that logs the line.
+        """
+        future: asyncio.Future[LineMatch] = asyncio.get_running_loop().create_future()
+        self._expectations[future] = pattern
+        return future
+
+    async def wait[T: str | re.Match[str]](
+        self,
+        future: asyncio.Future[T],
+        timeout: float = 5.0,
+        msg: str = "log line not seen",
+    ) -> T:
+        """Await a future from ``expect``; on timeout fail with its pattern and the last lines."""
+        return await wait_or_fail(
+            future,
+            timeout,
+            msg,
+            pattern=self._expectations.get(future),
+            lines=self.lines,
+        )
+
+    def count(self, pattern: LinePattern) -> int:
+        """Return how many lines seen so far match ``pattern``."""
+        return sum(_match(pattern, line) is not None for line in self.lines)

tests/integration/state_utils.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
 import logging
 from typing import TypeVar
 
@@ -100,6 +101,22 @@ class StateWaiter:
         self._waiters.append(entry)
         return self._wait(entry, timeout, label)
 
+    def next_state[T: EntityState](
+        self,
+        key: int | None = None,
+        state_type: type[T] = EntityState,
+        timeout: float = 5.0,
+    ) -> Awaitable[T]:
+        """Arm a wait for the next ``state_type`` state, for entity ``key`` if given.
+
+        Like :meth:`expect`, call it before sending the command and await the result afterwards.
+        """
+        return self.expect(
+            lambda s: isinstance(s, state_type) and (key is None or s.key == key),
+            timeout,
+            state_type.__name__ if key is None else f"{state_type.__name__} key={key}",
+        )
+
     async def _wait(
         self,
         entry: tuple[Callable[[EntityState], bool], asyncio.Future[EntityState]],
@@ -402,6 +419,58 @@ class InitialStateHelper:
         )
 
 
+@dataclass(frozen=True, slots=True)
+class StateSubscription:
+    """A state subscription that is past the connect time states."""
+
+    initial_states: dict[int, EntityState]
+    waiter: StateWaiter
+
+
+async def subscribe_after_initial_states(
+    client: APIClient,
+    entities: list[EntityInfo],
+    on_state: Callable[[EntityState], None] | None = None,
+    timeout: float = 5.0,
+) -> StateSubscription:
+    """Subscribe to states and wait until every entity has sent its connect time state.
+
+    Later states go to ``on_state`` (if given) and to the returned ``waiter``. The connect time
+    states are kept in ``initial_states`` by key. Fails the test listing the entities still missing.
+    """
+    helper = InitialStateHelper(entities)
+    waiter = StateWaiter()
+
+    def forward(state: EntityState) -> None:
+        waiter.on_state(state)
+        if on_state is not None:
+            on_state(state)
+
+    client.subscribe_states(helper.on_state_wrapper(forward))
+    await helper.wait_for_initial_states(timeout)
+    return StateSubscription(helper.initial_states, waiter)
+
+
+async def press_and_wait[T: EntityState](
+    client: APIClient,
+    entities: list[EntityInfo],
+    waiter: StateWaiter,
+    button: str,
+    state_type: type[T],
+    *,
+    key: int | None = None,
+    timeout: float = 5.0,
+) -> T:
+    """Press the button whose object_id contains ``button`` and return the next ``state_type`` state.
+
+    Pass the entity's ``key`` when the fixture has more than one entity of that type.
+    """
+    btn = require_entity(entities, button, ButtonInfo)
+    next_state = waiter.next_state(key=key, state_type=state_type, timeout=timeout)
+    client.button_command(btn.key)
+    return await next_state
+
+
 class SensorStateCollector:
     """Collects sensor, binary sensor, and text sensor state updates with wait helpers.
 

tests/integration/test_addressable_light_transition.py

@@ -23,7 +23,7 @@ import asyncio
 from aioesphomeapi import LightInfo, SensorInfo, SensorState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -52,11 +52,9 @@ async def test_addressable_light_transition(
                 return
             samples.append((loop.time(), state.state))
 
-        # InitialStateHelper swallows the first state ESPHome sends per entity
+        # Swallow the first state ESPHome sends per entity
         # on subscribe, so on_state only sees real post-subscribe updates.
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Start transition: off -> full white over 1 second. This is the
         # scenario from the bug report, compressed in time.

tests/integration/test_alarm_control_panel_state_transitions.py

@@ -15,7 +15,11 @@ from aioesphomeapi import (
 )
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity, require_only_entity
+from .state_utils import (
+    require_entity,
+    require_only_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -99,15 +103,11 @@ async def test_alarm_control_panel_state_transitions(
                 states_received.append(state.state)
                 state_event.set()
 
-        # Use InitialStateHelper to handle initial state broadcast
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial states from all entities
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Verify alarm panel started in DISARMED state
-        initial_alarm_state = initial_state_helper.initial_states.get(alarm_info.key)
+        initial_alarm_state = sub.initial_states.get(alarm_info.key)
         assert initial_alarm_state is not None, "No initial alarm state received"
         assert isinstance(initial_alarm_state, AlarmControlPanelEntityState)
         assert initial_alarm_state.state == AlarmControlPanelState.DISARMED
@@ -193,7 +193,7 @@ async def test_alarm_control_panel_state_transitions(
         )
         await wait_for_state(AlarmControlPanelState.DISARMED)
 
-        # Verify basic state sequence (initial DISARMED is handled by InitialStateHelper)
+        # Verify basic state sequence (initial DISARMED is swallowed on subscribe)
         expected_states = [
             AlarmControlPanelState.ARMING,  # Arm away
             AlarmControlPanelState.ARMED_AWAY,

tests/integration/test_api_cover_toggle.py

@@ -18,7 +18,7 @@ from .raw_api_client import (
     decode_fields,
     tag,
 )
-from .state_utils import InitialStateHelper, StateWaiter, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 COVER_COMMAND = MESSAGE_TYPE_OF[api_pb2.CoverCommandRequest]
@@ -48,12 +48,10 @@ async def test_api_cover_toggle(
         plain_cover = require_entity(entities, "plain_cover", CoverInfo, exact=True)
 
         positions: dict[int, list[float]] = {toggle_cover.key: [], plain_cover.key: []}
-        waiter = StateWaiter()
 
         def on_state(state: EntityState) -> None:
             if isinstance(state, CoverState) and state.key in positions:
                 positions[state.key].append(state.position)
-            waiter.on_state(state)
 
         def position_is(key: int, value: float) -> Callable[[EntityState], bool]:
             return lambda s: (
@@ -62,9 +60,8 @@ async def test_api_cover_toggle(
                 and s.position == pytest.approx(value, abs=0.01)
             )
 
-        initial = InitialStateHelper(entities)
-        client.subscribe_states(initial.on_state_wrapper(on_state))
-        await initial.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
+        waiter = sub.waiter
 
         # Capability discovery: only the cover with a toggle_action advertises toggle
         await raw.connect()

tests/integration/test_api_decode_wire_types.py

@@ -25,7 +25,7 @@ from .raw_api_client import (
     encode_varint,
     tag,
 )
-from .state_utils import InitialStateHelper, StateWaiter, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 SWITCH_COMMAND = MESSAGE_TYPE_OF[api_pb2.SwitchCommandRequest]
@@ -52,12 +52,10 @@ async def test_api_decode_wire_types(
         on, off = tag(2, WIRE_VARINT) + b"\x01", tag(2, WIRE_VARINT) + b"\x00"
 
         switch_states: list[bool] = []
-        waiter = StateWaiter()
 
         def on_state(state: EntityState) -> None:
             if isinstance(state, SwitchState) and state.key == switch.key:
                 switch_states.append(state.state)
-            waiter.on_state(state)
 
         def switch_is(value: bool) -> Callable[[EntityState], bool]:
             return lambda s: (
@@ -69,9 +67,8 @@ async def test_api_decode_wire_types(
                 isinstance(s, NumberState) and s.key == number.key and s.state == value
             )
 
-        initial = InitialStateHelper(entities)
-        client.subscribe_states(initial.on_state_wrapper(on_state))
-        await initial.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
+        waiter = sub.waiter
         await raw.connect()
 
         # A well formed command: fixed32 key, varint state

tests/integration/test_api_encode_boundaries.py

@@ -14,7 +14,7 @@ from aioesphomeapi import (
 )
 import pytest
 
-from .state_utils import InitialStateHelper, StateWaiter, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 LONG_OPTION = (
@@ -44,16 +44,14 @@ async def test_api_encode_boundaries(
         number = require_entity(entities, "negative_number")
         button = require_entity(entities, "publish_values")
 
-        initial = InitialStateHelper(entities)
-        waiter = StateWaiter()
-        client.subscribe_states(initial.on_state_wrapper(waiter.on_state))
-        await initial.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
+        waiter = sub.waiter
 
         # A float of exactly zero is skipped on the wire and must still read as 0.0, not missing
-        first = initial.initial_states[sensor.key]
+        first = sub.initial_states[sensor.key]
         assert isinstance(first, SensorState)
         assert first.state == 0.0 and not first.missing_state
-        first_number = initial.initial_states[number.key]
+        first_number = sub.initial_states[number.key]
         assert isinstance(first_number, NumberState)
         assert first_number.state == -123.5
 

tests/integration/test_binary_sensor_autorepeat_filter.py

@@ -18,10 +18,10 @@ import asyncio
 import pytest
 
 from .state_utils import (
-    InitialStateHelper,
     SensorStateCollector,
     assert_device_name,
     require_entity,
+    subscribe_after_initial_states,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -53,13 +53,10 @@ async def test_binary_sensor_autorepeat_filter(
             entities, "release", description="Release button"
         )
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
 
-        await initial_state_helper.wait_for_initial_states()
-
         autorepeat_states = collector.binary_states["autorepeat_sensor"]
 
         # Press: source becomes true, autorepeat passes the initial true through

tests/integration/test_binary_sensor_invalidate_state.py

@@ -9,14 +9,15 @@ Regression test for: https://github.com/esphome/esphome/issues/12252
 
 from __future__ import annotations
 
-import asyncio
-
-from aioesphomeapi import BinarySensorInfo, BinarySensorState, EntityState
+from aioesphomeapi import BinarySensorInfo, BinarySensorState
 import pytest
 
-from .state_utils import InitialStateHelper, assert_device_name, require_entity
+from .state_utils import (
+    assert_device_name,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -33,19 +34,6 @@ async def test_binary_sensor_invalidate_state(
     3. Publishing false sets missing_state=False and state=False
     4. Invalidating state sets missing_state=True (unknown state)
     """
-    loop = asyncio.get_running_loop()
-
-    # Track state changes
-    states_received: list[BinarySensorState] = []
-    state_future: asyncio.Future[BinarySensorState] = loop.create_future()
-
-    def on_state(state: EntityState) -> None:
-        """Track binary sensor state changes."""
-        if isinstance(state, BinarySensorState):
-            states_received.append(state)
-            if not state_future.done():
-                state_future.set_result(state)
-
     async with (
         run_compiled(yaml_config),
         api_client_connected() as client,
@@ -68,15 +56,11 @@ async def test_binary_sensor_invalidate_state(
             entities, "invalidate", description="Invalidate button"
         )
 
-        # Set up initial state helper to handle the initial state broadcast
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial states
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # Check initial state - should be missing (unknown)
-        initial_state = initial_state_helper.initial_states.get(binary_sensor.key)
+        initial_state = sub.initial_states.get(binary_sensor.key)
         assert initial_state is not None, "No initial state received for binary sensor"
         assert isinstance(initial_state, BinarySensorState)
         assert initial_state.missing_state is True, (
@@ -84,11 +68,9 @@ async def test_binary_sensor_invalidate_state(
         )
 
         # Test 1: Set state to true
-        states_received.clear()
-        state_future = loop.create_future()
+        next_state = sub.waiter.next_state(binary_sensor.key, BinarySensorState)
         client.button_command(set_true_button.key)
-
-        state = await wait_or_fail(state_future, 5.0, "Timeout waiting for state=true")
+        state = await next_state
 
         assert state.missing_state is False, (
             f"After setting true, missing_state should be False, got {state}"
@@ -96,11 +78,9 @@ async def test_binary_sensor_invalidate_state(
         assert state.state is True, f"Expected state=True, got {state}"
 
         # Test 2: Set state to false
-        states_received.clear()
-        state_future = loop.create_future()
+        next_state = sub.waiter.next_state(binary_sensor.key, BinarySensorState)
         client.button_command(set_false_button.key)
-
-        state = await wait_or_fail(state_future, 5.0, "Timeout waiting for state=false")
+        state = await next_state
 
         assert state.missing_state is False, (
             f"After setting false, missing_state should be False, got {state}"
@@ -109,17 +89,10 @@ async def test_binary_sensor_invalidate_state(
 
         # Test 3: Invalidate state (set to unknown)
         # This is the critical test for the bug fix
-        states_received.clear()
-        state_future = loop.create_future()
+        # A timeout here means invalidate_state() is not reported to the API (issue #12252)
+        next_state = sub.waiter.next_state(binary_sensor.key, BinarySensorState)
         client.button_command(invalidate_button.key)
-
-        state = await wait_or_fail(
-            state_future,
-            5.0,
-            "Timeout waiting for invalidated state - "
-            "binary_sensor.invalidate_state() may not be reporting to the API. "
-            "See issue #12252.",
-        )
+        state = await next_state
 
         assert state.missing_state is True, (
             f"After invalidate_state(), missing_state should be True (unknown), "

tests/integration/test_climate_control_action.py

@@ -7,18 +7,12 @@ multiple field combinations and the lambda path.
 
 from __future__ import annotations
 
-import asyncio
+from functools import partial
 
-from aioesphomeapi import (
-    ButtonInfo,
-    ClimateInfo,
-    ClimateMode,
-    ClimateState,
-    EntityState,
-)
+from aioesphomeapi import ClimateInfo, ClimateMode, ClimateState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import press_and_wait, require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -29,56 +23,34 @@ async def test_climate_control_action(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test climate ControlAction with constants and lambdas."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        climate_state_future: asyncio.Future[ClimateState] | None = None
-
-        def on_state(state: EntityState) -> None:
-            if (
-                isinstance(state, ClimateState)
-                and climate_state_future is not None
-                and not climate_state_future.done()
-            ):
-                climate_state_future.set_result(state)
-
-        async def wait_for_climate_state(timeout: float = 5.0) -> ClimateState:
-            nonlocal climate_state_future
-            climate_state_future = loop.create_future()
-            try:
-                return await asyncio.wait_for(climate_state_future, timeout)
-            finally:
-                climate_state_future = None
-
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         require_entity(entities, "test_climate", ClimateInfo)
 
-        async def press_and_wait(name: str) -> ClimateState:
-            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
-            client.button_command(btn.key)
-            return await wait_for_climate_state()
+        press = partial(
+            press_and_wait, client, entities, sub.waiter, state_type=ClimateState
+        )
 
         # mode only — set HEAT
-        state = await press_and_wait("Set Mode Heat")
+        state = await press("set_mode_heat")
         assert state.mode == ClimateMode.HEAT
 
         # mode + target_temperature_low + target_temperature_high
-        state = await press_and_wait("Set Mode Temps")
+        state = await press("set_mode_temps")
         assert state.mode == ClimateMode.HEAT_COOL
         assert state.target_temperature_low == pytest.approx(19.0, abs=0.5)
         assert state.target_temperature_high == pytest.approx(23.0, abs=0.5)
 
         # target_temperature_low only
-        state = await press_and_wait("Set Low Only")
+        state = await press("set_low_only")
         assert state.target_temperature_low == pytest.approx(17.5, abs=0.5)
 
         # lambda path: target_temperature_high computed at runtime
-        state = await press_and_wait("Lambda High")
+        state = await press("lambda_high")
         assert state.target_temperature_high == pytest.approx(21.5, abs=0.5)
 
         # mode only — turn off via mode
-        state = await press_and_wait("Set Off")
+        state = await press("set_off")
         assert state.mode == ClimateMode.OFF

tests/integration/test_climate_custom_modes.py

@@ -2,15 +2,12 @@
 
 from __future__ import annotations
 
-import asyncio
-
 import aioesphomeapi
-from aioesphomeapi import ClimateInfo, ClimatePreset, EntityState
+from aioesphomeapi import ClimateInfo, ClimatePreset
 import pytest
 
-from .state_utils import InitialStateHelper, require_only_entity
+from .state_utils import require_only_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -20,42 +17,13 @@ async def test_climate_custom_fan_modes_and_presets(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test that custom presets are properly exposed and can be changed."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        states: dict[int, EntityState] = {}
-        super_saver_future: asyncio.Future[EntityState] = loop.create_future()
-        vacation_future: asyncio.Future[EntityState] = loop.create_future()
-
-        def on_state(state: EntityState) -> None:
-            states[state.key] = state
-            if isinstance(state, aioesphomeapi.ClimateState):
-                # Wait for Super Saver preset
-                if (
-                    state.custom_preset == "Super Saver"
-                    and state.target_temperature_low == 20.0
-                    and state.target_temperature_high == 24.0
-                    and not super_saver_future.done()
-                ):
-                    super_saver_future.set_result(state)
-                # Wait for Vacation Mode preset
-                elif (
-                    state.custom_preset == "Vacation Mode"
-                    and state.target_temperature_low == 15.0
-                    and state.target_temperature_high == 18.0
-                    and not vacation_future.done()
-                ):
-                    vacation_future.set_result(state)
-
         # Get entities and set up state synchronization
         entities, services = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
 
-        # Subscribe with the wrapper that filters initial states
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for all initial states to be broadcast
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # Verify enum presets are exposed (from preset: config map)
         assert ClimatePreset.AWAY in test_climate.supported_presets, (
@@ -76,7 +44,7 @@ async def test_climate_custom_fan_modes_and_presets(
         )
 
         # Get initial state and verify default preset
-        initial_state = initial_state_helper.initial_states.get(test_climate.key)
+        initial_state = sub.initial_states.get(test_climate.key)
         assert initial_state is not None, "Climate initial state not found"
         assert isinstance(initial_state, aioesphomeapi.ClimateState)
         assert initial_state.custom_preset == "Eco Plus", (
@@ -90,13 +58,17 @@ async def test_climate_custom_fan_modes_and_presets(
         )
 
         # Test changing to "Super Saver" custom preset
-        client.climate_command(test_climate.key, custom_preset="Super Saver")
-
-        super_saver_state = await wait_or_fail(
-            super_saver_future,
-            5.0,
-            "Super Saver preset change not received within 5 seconds",
+        super_saver_seen = sub.waiter.expect(
+            lambda s: (
+                isinstance(s, aioesphomeapi.ClimateState)
+                and s.custom_preset == "Super Saver"
+                and s.target_temperature_low == 20.0
+                and s.target_temperature_high == 24.0
+            ),
+            label="Super Saver preset change",
         )
+        client.climate_command(test_climate.key, custom_preset="Super Saver")
+        super_saver_state = await super_saver_seen
 
         assert isinstance(super_saver_state, aioesphomeapi.ClimateState)
         assert super_saver_state.custom_preset == "Super Saver"
@@ -104,13 +76,17 @@ async def test_climate_custom_fan_modes_and_presets(
         assert super_saver_state.target_temperature_high == 24.0
 
         # Test changing to "Vacation Mode" custom preset
-        client.climate_command(test_climate.key, custom_preset="Vacation Mode")
-
-        vacation_state = await wait_or_fail(
-            vacation_future,
-            5.0,
-            "Vacation Mode preset change not received within 5 seconds",
+        vacation_seen = sub.waiter.expect(
+            lambda s: (
+                isinstance(s, aioesphomeapi.ClimateState)
+                and s.custom_preset == "Vacation Mode"
+                and s.target_temperature_low == 15.0
+                and s.target_temperature_high == 18.0
+            ),
+            label="Vacation Mode preset change",
         )
+        client.climate_command(test_climate.key, custom_preset="Vacation Mode")
+        vacation_state = await vacation_seen
 
         assert isinstance(vacation_state, aioesphomeapi.ClimateState)
         assert vacation_state.custom_preset == "Vacation Mode"

tests/integration/test_cover_control_action.py

@@ -7,12 +7,12 @@ implementation. Exercises multiple field combinations and the lambda path.
 
 from __future__ import annotations
 
-import asyncio
+from functools import partial
 
-from aioesphomeapi import ButtonInfo, CoverInfo, CoverState, EntityState
+from aioesphomeapi import CoverInfo, CoverState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import press_and_wait, require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -23,80 +23,58 @@ async def test_cover_control_action(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test cover.control and cover.template.publish with constants and lambdas."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        cover_state_future: asyncio.Future[CoverState] | None = None
-
-        def on_state(state: EntityState) -> None:
-            if (
-                isinstance(state, CoverState)
-                and cover_state_future is not None
-                and not cover_state_future.done()
-            ):
-                cover_state_future.set_result(state)
-
-        async def wait_for_cover_state(timeout: float = 5.0) -> CoverState:
-            nonlocal cover_state_future
-            cover_state_future = loop.create_future()
-            try:
-                return await asyncio.wait_for(cover_state_future, timeout)
-            finally:
-                cover_state_future = None
-
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         require_entity(entities, "test_cover", CoverInfo)
 
-        async def press_and_wait(name: str) -> CoverState:
-            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
-            client.button_command(btn.key)
-            return await wait_for_cover_state()
+        press = partial(
+            press_and_wait, client, entities, sub.waiter, state_type=CoverState
+        )
 
         # cover.control: position only
-        state = await press_and_wait("Set Position")
+        state = await press("set_position")
         assert state.position == pytest.approx(0.5, abs=0.01)
 
         # cover.control: tilt only
-        state = await press_and_wait("Set Tilt")
+        state = await press("set_tilt")
         assert state.tilt == pytest.approx(0.75, abs=0.01)
 
         # cover.control: position + tilt
-        state = await press_and_wait("Set Pos Tilt")
+        state = await press("set_pos_tilt")
         assert state.position == pytest.approx(0.25, abs=0.01)
         assert state.tilt == pytest.approx(0.30, abs=0.01)
 
         # cover.control: state alias for position 1.0
-        state = await press_and_wait("Open State")
+        state = await press("open_state")
         assert state.position == pytest.approx(1.0, abs=0.01)
 
         # cover.control: lambda position (test_position global = 0.42)
-        state = await press_and_wait("Lambda Position")
+        state = await press("lambda_position")
         assert state.position == pytest.approx(0.42, abs=0.01)
 
         # cover.template.publish: position only
-        state = await press_and_wait("Publish Pos")
+        state = await press("publish_pos")
         assert state.position == pytest.approx(0.6, abs=0.01)
 
         # cover.template.publish: current_operation only
-        state = await press_and_wait("Publish Op")
+        state = await press("publish_op")
         # CoverOperation.OPENING == 1
         assert state.current_operation == 1
 
         # cover.control: stop only — template cover's stop_action publishes
         # current_operation: IDLE.
-        state = await press_and_wait("Stop Cover")
+        state = await press("stop_cover")
         # CoverOperation.IDLE == 0
         assert state.current_operation == 0
 
         # stop_tilt_action publishes tilt 0.4
-        state = await press_and_wait("Stop Tilt")
+        state = await press("stop_tilt")
         assert state.tilt == pytest.approx(0.4, abs=0.01)
 
-        state = await press_and_wait("Set Tilt")
+        state = await press("set_tilt")
         assert state.tilt == pytest.approx(0.75, abs=0.01)
 
-        state = await press_and_wait("Control Stop Tilt")
+        state = await press("control_stop_tilt")
         assert state.tilt == pytest.approx(0.4, abs=0.01)

tests/integration/test_cover_stop_tilt_cancel.py

@@ -7,7 +7,7 @@ import asyncio
 from aioesphomeapi import ButtonInfo, CoverInfo, CoverState, EntityState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -30,9 +30,7 @@ async def test_cover_stop_tilt_cancel(
                 tilts.append(state.tilt)
 
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         cover = require_entity(entities, "test_cover", CoverInfo)
         button = require_entity(entities, "stop_tilt", ButtonInfo)

tests/integration/test_entity_missing_state.py

@@ -15,7 +15,6 @@ from aioesphomeapi import (
     ClimateMode,
     ClimateState,
     EntityInfo,
-    EntityState,
     SwitchInfo,
     SwitchState,
     WaterHeaterInfo,
@@ -24,9 +23,8 @@ from aioesphomeapi import (
 )
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -42,15 +40,6 @@ async def test_entity_missing_state(
     2. Publishing clears missing_state and reports the published value
     3. A first value that happens to equal the default still publishes
     """
-    loop = asyncio.get_running_loop()
-    futures: dict[int, asyncio.Future[EntityState]] = {}
-
-    def on_state(state: EntityState) -> None:
-        """Resolve the pending future for the entity that changed."""
-        future = futures.get(state.key)
-        if future is not None and not future.done():
-            future.set_result(state)
-
     async with (
         run_compiled(yaml_config),
         api_client_connected() as client,
@@ -76,14 +65,11 @@ async def test_entity_missing_state(
             lambda_water_heater_info,
         ]
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # Nothing has published yet, so all of them must report unknown
         for info in stateful:
-            state = initial_state_helper.initial_states.get(info.key)
+            state = sub.initial_states.get(info.key)
             assert state is not None, f"No initial state received for {info.object_id}"
             assert state.missing_state is True, (
                 f"Initial state for {info.object_id} should have "
@@ -91,41 +77,34 @@ async def test_entity_missing_state(
             )
 
         # A configured initial_state must not start out unknown
-        initial_climate_state = initial_state_helper.initial_states.get(
-            initial_climate_info.key
-        )
+        initial_climate_state = sub.initial_states.get(initial_climate_info.key)
         assert isinstance(initial_climate_state, ClimateState)
         assert initial_climate_state.missing_state is False
         assert initial_climate_state.mode is ClimateMode.HEAT
 
         # Publishing a state on each one clears missing_state
-        futures = {info.key: loop.create_future() for info in stateful}
+        waits = {info.key: sub.waiter.next_state(info.key) for info in stateful}
         client.button_command(publish_button.key)
+        published = dict(zip(waits, await asyncio.gather(*waits.values()), strict=True))
 
-        await wait_or_fail(
-            asyncio.gather(*futures.values()),
-            5.0,
-            "Timeout waiting for published states",
-        )
-
-        switch_state = futures[switch_info.key].result()
+        switch_state = published[switch_info.key]
         assert isinstance(switch_state, SwitchState)
         assert switch_state.missing_state is False
         assert switch_state.state is True
 
-        climate_state = futures[climate_info.key].result()
+        climate_state = published[climate_info.key]
         assert isinstance(climate_state, ClimateState)
         assert climate_state.missing_state is False
         assert climate_state.mode is ClimateMode.HEAT
 
-        water_heater_state = futures[water_heater_info.key].result()
+        water_heater_state = published[water_heater_info.key]
         assert isinstance(water_heater_state, WaterHeaterState)
         assert water_heater_state.missing_state is False
         assert water_heater_state.mode is WaterHeaterMode.ECO
 
         # Its first value is OFF, which is also the default the entity starts on,
         # so it only leaves unknown if the first value publishes regardless
-        lambda_state = futures[lambda_water_heater_info.key].result()
+        lambda_state = published[lambda_water_heater_info.key]
         assert isinstance(lambda_state, WaterHeaterState)
         assert lambda_state.missing_state is False
         assert lambda_state.mode is WaterHeaterMode.OFF

tests/integration/test_fan_turn_on_action.py

@@ -6,12 +6,12 @@ field combinations and the lambda path.
 
 from __future__ import annotations
 
-import asyncio
+from functools import partial
 
-from aioesphomeapi import ButtonInfo, EntityState, FanDirection, FanInfo, FanState
+from aioesphomeapi import FanDirection, FanInfo, FanState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import press_and_wait, require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -22,54 +22,32 @@ async def test_fan_turn_on_action(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test fan TurnOnAction with constants and a lambda."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        fan_state_future: asyncio.Future[FanState] | None = None
-
-        def on_state(state: EntityState) -> None:
-            if (
-                isinstance(state, FanState)
-                and fan_state_future is not None
-                and not fan_state_future.done()
-            ):
-                fan_state_future.set_result(state)
-
-        async def wait_for_fan_state(timeout: float = 5.0) -> FanState:
-            nonlocal fan_state_future
-            fan_state_future = loop.create_future()
-            try:
-                return await asyncio.wait_for(fan_state_future, timeout)
-            finally:
-                fan_state_future = None
-
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         require_entity(entities, "test_fan", FanInfo)
 
-        async def press_and_wait(name: str) -> FanState:
-            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
-            client.button_command(btn.key)
-            return await wait_for_fan_state()
+        press = partial(
+            press_and_wait, client, entities, sub.waiter, state_type=FanState
+        )
 
         # speed only
-        state = await press_and_wait("Set Speed")
+        state = await press("set_speed")
         assert state.state is True
         assert state.speed_level == 3
 
         # oscillating + direction
-        state = await press_and_wait("Set Oscillate Direction")
+        state = await press("set_oscillate_direction")
         assert state.oscillating is True
         assert state.direction == FanDirection.REVERSE
 
         # all three fields
-        state = await press_and_wait("Set All Fields")
+        state = await press("set_all_fields")
         assert state.oscillating is False
         assert state.speed_level == 4
         assert state.direction == FanDirection.FORWARD
 
         # lambda path: speed computed at runtime (test_speed global = 2)
-        state = await press_and_wait("Lambda Speed")
+        state = await press("lambda_speed")
         assert state.speed_level == 2

tests/integration/test_host_mode_climate_control.py

@@ -2,15 +2,12 @@
 
 from __future__ import annotations
 
-import asyncio
-
 import aioesphomeapi
-from aioesphomeapi import ClimateInfo, ClimateMode, EntityState
+from aioesphomeapi import ClimateInfo, ClimateMode
 import pytest
 
-from .state_utils import InitialStateHelper, require_only_entity
+from .state_utils import require_only_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -20,35 +17,25 @@ async def test_host_mode_climate_control(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test climate mode control."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        states: dict[int, EntityState] = {}
-        climate_future: asyncio.Future[EntityState] = loop.create_future()
-
-        def on_state(state: EntityState) -> None:
-            states[state.key] = state
-            if (
-                isinstance(state, aioesphomeapi.ClimateState)
-                and state.mode == ClimateMode.HEAT
-                and state.target_temperature_low == 21.5
-                and state.target_temperature_high == 26.5
-                and not climate_future.done()
-            ):
-                climate_future.set_result(state)
-
         # Get entities and set up state synchronization
         entities, services = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
         assert test_climate.name == "Dual-mode Thermostat"
 
-        # Subscribe with the wrapper that filters initial states
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for all initial states to be broadcast
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # Adjust setpoints
+        climate_seen = sub.waiter.expect(
+            lambda s: (
+                isinstance(s, aioesphomeapi.ClimateState)
+                and s.mode == ClimateMode.HEAT
+                and s.target_temperature_low == 21.5
+                and s.target_temperature_high == 26.5
+            ),
+            label="Climate state",
+        )
         client.climate_command(
             test_climate.key,
             mode=ClimateMode.HEAT,
@@ -56,9 +43,7 @@ async def test_host_mode_climate_control(
             target_temperature_high=26.5,
         )
 
-        climate_state = await wait_or_fail(
-            climate_future, 5.0, "Climate state not received within 5 seconds"
-        )
+        climate_state = await climate_seen
 
         assert isinstance(climate_state, aioesphomeapi.ClimateState)
         assert climate_state.mode == ClimateMode.HEAT

tests/integration/test_light_binary_effect_off_phase.py

@@ -23,10 +23,10 @@ import asyncio
 import re
 from typing import Any
 
-from aioesphomeapi import EntityState, LightState
+from aioesphomeapi import LightState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 OUTPUT_PATTERN = re.compile(r"BINARY_OUTPUT:(YES|NO)")
@@ -66,25 +66,15 @@ async def test_light_binary_effect_off_phase(
         entities, _ = await client.list_entities_services()
         light = require_entity(entities, "test_binary_light", exact=True)
 
-        state_futures: dict[int, asyncio.Future[LightState]] = {}
-
-        def on_state(state: EntityState) -> None:
-            if isinstance(state, LightState) and state.key in state_futures:
-                future = state_futures[state.key]
-                if not future.done():
-                    future.set_result(state)
-
         # ESPHome sends the current state of every entity right after connecting; drain
         # that initial burst so it can't be mistaken for the response to a command below.
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
-        async def send_and_wait(timeout: float = 5.0, **kwargs: Any) -> LightState:
-            """Send a light command and wait for the matching state response."""
-            state_futures[light.key] = loop.create_future()
+        async def send_and_wait(**kwargs: Any) -> LightState:
+            """Send a light command and return the light's next state."""
+            next_light = sub.waiter.next_state(light.key, LightState)
             client.light_command(key=light.key, **kwargs)
-            return await asyncio.wait_for(state_futures[light.key], timeout=timeout)
+            return await next_light
 
         # A plain turn-on must drive the output on -- brightness defaults to 100% and
         # must not be mistaken for a dark phase.
@@ -165,22 +155,13 @@ async def test_light_binary_zero_brightness_is_recoverable(
         entities, _ = await client.list_entities_services()
         light = require_entity(entities, "test_binary_light", exact=True)
 
-        state_futures: dict[int, asyncio.Future[LightState]] = {}
+        sub = await subscribe_after_initial_states(client, entities)
 
-        def on_state(state: EntityState) -> None:
-            if isinstance(state, LightState) and state.key in state_futures:
-                future = state_futures[state.key]
-                if not future.done():
-                    future.set_result(state)
-
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
-
-        async def send_and_wait(timeout: float = 5.0, **kwargs: Any) -> LightState:
-            state_futures[light.key] = loop.create_future()
+        async def send_and_wait(**kwargs: Any) -> LightState:
+            """Send a light command and return the light's next state."""
+            next_light = sub.waiter.next_state(light.key, LightState)
             client.light_command(key=light.key, **kwargs)
-            return await asyncio.wait_for(state_futures[light.key], timeout=timeout)
+            return await next_light
 
         output = arm_output()
         state = await send_and_wait(state=True)

tests/integration/test_light_constant_brightness.py

@@ -12,17 +12,15 @@ gamma after the balancing formula instead of before it.
 
 from __future__ import annotations
 
-import asyncio
 import re
 from typing import Any
 
-from aioesphomeapi import EntityState, LightInfo, LightState
+from aioesphomeapi import LightInfo, LightState
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 CB_CW_RE = re.compile(r"(?<!N)CB_CW_OUTPUT:([\d.]+)")
 CB_WW_RE = re.compile(r"(?<!N)CB_WW_OUTPUT:([\d.]+)")
@@ -46,8 +44,6 @@ async def test_light_constant_brightness(
                 return float(match[1])
         return 0.0
 
-    loop = asyncio.get_running_loop()
-
     async with (
         run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
@@ -56,33 +52,13 @@ async def test_light_constant_brightness(
         cb_light = require_entity(entities, "test_cb_light", LightInfo, exact=True)
         ncb_light = require_entity(entities, "test_ncb_light", LightInfo, exact=True)
 
-        # Use InitialStateHelper to wait for initial state broadcast
-        initial_state_helper = InitialStateHelper(entities)
+        sub = await subscribe_after_initial_states(client, entities)
 
-        # Track state changes per light key
-        state_futures: dict[int, asyncio.Future[EntityState]] = {}
-
-        def on_state(state: EntityState) -> None:
-            if isinstance(state, LightState) and state.key in state_futures:
-                future = state_futures[state.key]
-                if not future.done():
-                    future.set_result(state)
-
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
-        await initial_state_helper.wait_for_initial_states()
-
-        async def send_and_wait(
-            light_key: int, timeout: float = 5.0, **kwargs: Any
-        ) -> LightState:
-            """Send a light command and wait for the state response."""
-            state_futures[light_key] = loop.create_future()
+        async def send_and_wait(light_key: int, **kwargs: Any) -> LightState:
+            """Send a light command and return the light's next state."""
+            next_light = sub.waiter.next_state(light_key, LightState)
             client.light_command(key=light_key, **kwargs)
-            return await wait_or_fail(
-                state_futures[light_key],
-                timeout,
-                f"Timeout waiting for light state after command: {kwargs}",
-            )
+            return await next_light
 
         # --- Test constant_brightness: true ---
 

tests/integration/test_light_control_action.py

@@ -5,12 +5,12 @@ work correctly with the compact per-field union storage. Exercises both constant
 value and lambda paths.
 """
 
-import asyncio
-from typing import Any
+from functools import partial
 
+from aioesphomeapi import LightState
 import pytest
 
-from .state_utils import require_entity
+from .state_utils import press_and_wait, require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -22,36 +22,20 @@ async def test_light_control_action(
 ) -> None:
     """Test light.turn_on, light.turn_off and light.control with constants and lambdas."""
     async with run_compiled(yaml_config), api_client_connected() as client:
-        state_futures: dict[int, asyncio.Future[Any]] = {}
-
-        def on_state(state: Any) -> None:
-            if state.key in state_futures and not state_futures[state.key].done():
-                state_futures[state.key].set_result(state)
-
-        client.subscribe_states(on_state)
-
-        # Get entities
-        entities = await client.list_entities_services()
-        light = require_entity(entities[0], "test_light", exact=True)
-        buttons = {e.name: e for e in entities[0] if hasattr(e, "name")}
-
-        async def wait_for_state(key: int, timeout: float = 5.0) -> Any:
-            """Wait for a state change for the given entity key."""
-            loop = asyncio.get_running_loop()
-            state_futures[key] = loop.create_future()
-            try:
-                return await asyncio.wait_for(state_futures[key], timeout)
-            finally:
-                state_futures.pop(key, None)
-
-        async def press_and_wait(button_name: str) -> Any:
-            """Press a button and wait for light state change."""
-            btn = buttons[button_name]
-            client.button_command(btn.key)
-            return await wait_for_state(light.key)
+        entities, _ = await client.list_entities_services()
+        light = require_entity(entities, "test_light", exact=True)
+        sub = await subscribe_after_initial_states(client, entities)
+        press = partial(
+            press_and_wait,
+            client,
+            entities,
+            sub.waiter,
+            state_type=LightState,
+            key=light.key,
+        )
 
         # Test 1: light.turn_on with RGB constants
-        state = await press_and_wait("Turn On RGB")
+        state = await press("turn_on_rgb")
         assert state.state is True
         assert state.brightness == pytest.approx(1.0)
         assert state.red == pytest.approx(0.0, abs=0.01)
@@ -59,31 +43,31 @@ async def test_light_control_action(
         assert state.blue == pytest.approx(1.0, abs=0.01)
 
         # Test 2: light.turn_off
-        state = await press_and_wait("Turn Off")
+        state = await press("turn_off")
         assert state.state is False
 
         # Test 3: light.turn_on with color_temperature
-        state = await press_and_wait("Turn On CT")
+        state = await press("turn_on_ct")
         assert state.state is True
         assert state.brightness == pytest.approx(0.8)
         assert state.color_temperature == pytest.approx(250.0)  # 4000K = 250 mireds
 
         # Test 4: light.turn_on with effect
-        state = await press_and_wait("Turn On Effect")
+        state = await press("turn_on_effect")
         assert state.effect == "Test Effect"
 
         # Test 5: Clear effect
-        state = await press_and_wait("Clear Effect")
+        state = await press("clear_effect")
         assert state.effect == "None"
 
         # Test 6: light.control with cold/warm white
-        state = await press_and_wait("Control CW")
+        state = await press("control_cw")
         assert state.cold_white == pytest.approx(0.9, abs=0.1)
         assert state.warm_white == pytest.approx(0.1, abs=0.1)
 
         # Test 7: light.turn_on with lambda brightness
         # The global test_brightness is 0.75
-        state = await press_and_wait("Lambda Brightness")
+        state = await press("lambda_brightness")
         assert state.state is True
         assert state.brightness == pytest.approx(0.75, abs=0.05)
         assert state.red == pytest.approx(1.0, abs=0.01)
@@ -91,6 +75,6 @@ async def test_light_control_action(
         assert state.blue == pytest.approx(0.0, abs=0.01)
 
         # Test 8: light.turn_on with transition_length and brightness
-        state = await press_and_wait("Turn On Transition")
+        state = await press("turn_on_transition")
         assert state.state is True
         assert state.brightness == pytest.approx(0.5)

tests/integration/test_light_dim_relative_action.py

@@ -6,12 +6,12 @@ DimRelativeAction<HasTransitionLength=true> instantiations.
 
 from __future__ import annotations
 
-import asyncio
+from functools import partial
 
-from aioesphomeapi import ButtonInfo, EntityState, LightInfo, LightState
+from aioesphomeapi import LightInfo, LightState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import press_and_wait, require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -22,51 +22,29 @@ async def test_light_dim_relative_action(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test light.dim_relative with and without transition_length."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        light_state_future: asyncio.Future[LightState] | None = None
-
-        def on_state(state: EntityState) -> None:
-            if (
-                isinstance(state, LightState)
-                and light_state_future is not None
-                and not light_state_future.done()
-            ):
-                light_state_future.set_result(state)
-
-        async def wait_for_light_state(timeout: float = 5.0) -> LightState:
-            nonlocal light_state_future
-            light_state_future = loop.create_future()
-            try:
-                return await asyncio.wait_for(light_state_future, timeout)
-            finally:
-                light_state_future = None
-
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         require_entity(entities, "test_light", LightInfo)
 
-        async def press_and_wait(name: str) -> LightState:
-            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
-            client.button_command(btn.key)
-            return await wait_for_light_state()
+        press = partial(
+            press_and_wait, client, entities, sub.waiter, state_type=LightState
+        )
 
         # Setup: turn on at 50%
-        state = await press_and_wait("Setup")
+        state = await press("setup")
         assert state.state is True
         assert state.brightness == pytest.approx(0.5, abs=0.05)
 
         # Test 1: dim_relative without transition_length: 50% + 25% = 75%
-        state = await press_and_wait("Dim Up")
+        state = await press("dim_up")
         assert state.brightness == pytest.approx(0.75, abs=0.05)
 
         # Test 2: dim_relative with transition_length: 75% - 10% = 65%
-        state = await press_and_wait("Dim Down")
+        state = await press("dim_down")
         assert state.brightness == pytest.approx(0.65, abs=0.05)
 
         # Test 3: dim_relative with max_brightness limit: 65% + 50% clamped to 80%
-        state = await press_and_wait("Dim Clamp")
+        state = await press("dim_clamp")
         assert state.brightness == pytest.approx(0.80, abs=0.05)

tests/integration/test_light_effect_zero_brightness.py

@@ -17,10 +17,10 @@ import asyncio
 import re
 from typing import Any
 
-from aioesphomeapi import EntityState, LightState
+from aioesphomeapi import LightState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -46,25 +46,15 @@ async def test_light_effect_zero_brightness(
         entities, _ = await client.list_entities_services()
         light = require_entity(entities, "test_pulse_light", exact=True)
 
-        state_futures: dict[int, asyncio.Future[LightState]] = {}
-
-        def on_state(state: EntityState) -> None:
-            if isinstance(state, LightState) and state.key in state_futures:
-                future = state_futures[state.key]
-                if not future.done():
-                    future.set_result(state)
-
         # ESPHome sends the current state of every entity right after connecting; drain
         # that initial burst so it can't be mistaken for the response to a command below.
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
-        async def send_and_wait(timeout: float = 5.0, **kwargs: Any) -> LightState:
-            """Send a light command and wait for the matching state response."""
-            state_futures[light.key] = asyncio.get_running_loop().create_future()
+        async def send_and_wait(**kwargs: Any) -> LightState:
+            """Send a light command and return the light's next state."""
+            next_light = sub.waiter.next_state(light.key, LightState)
             client.light_command(key=light.key, **kwargs)
-            return await asyncio.wait_for(state_futures[light.key], timeout=timeout)
+            return await next_light
 
         # Turn the light on first so the effect starts from a known, visible state.
         state = await send_and_wait(state=True, brightness=1.0)

tests/integration/test_light_initial_state.py

@@ -7,7 +7,7 @@ that the restore logic uses as a fallback.
 
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -22,11 +22,9 @@ async def test_light_initial_state(
         entities, _ = await client.list_entities_services()
         light = require_entity(entities, "test_light")
 
-        helper = InitialStateHelper(entities)
-        client.subscribe_states(helper.on_state_wrapper(lambda s: None))
-        await helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
-        state = helper.initial_states[light.key]
+        state = sub.initial_states[light.key]
 
         # restore_mode: ALWAYS_OFF overrides state to false
         assert state.state is False
@@ -40,14 +38,14 @@ async def test_light_initial_state(
         # Regression test: RESTORE_AND_ON always forces the light on at boot, even when
         # the recovered/initial brightness was 0 -- it must never come up on-but-invisible.
         restore_and_on_light = require_entity(entities, "test_restore_and_on_light")
-        restore_and_on_state = helper.initial_states[restore_and_on_light.key]
+        restore_and_on_state = sub.initial_states[restore_and_on_light.key]
         assert restore_and_on_state.state is True
         assert restore_and_on_state.brightness == pytest.approx(1.0)
 
         # initial_state may also name the effect to start with. The light must come up
         # with that effect already active, without an on_boot automation.
         effect_light = require_entity(entities, "test_effect_light")
-        effect_state = helper.initial_states[effect_light.key]
+        effect_state = sub.initial_states[effect_light.key]
         assert effect_state.state is True
         assert effect_state.effect == "Test Pulse"
 
@@ -55,7 +53,7 @@ async def test_light_initial_state(
         # be honored as-is -- unlike every explicit restore_mode: value, omitting both
         # keys entirely must not force the light off.
         no_restore_key_light = require_entity(entities, "test_no_restore_key_light")
-        no_restore_key_state = helper.initial_states[no_restore_key_light.key]
+        no_restore_key_state = sub.initial_states[no_restore_key_light.key]
         assert no_restore_key_state.state is True
         assert no_restore_key_state.brightness == pytest.approx(0.6, abs=0.01)
 
@@ -63,7 +61,7 @@ async def test_light_initial_state(
         # light comes up red rather than the default white). RGB values are normalised to
         # proportions, so 30% red on its own is pure red.
         inferred_light = require_entity(entities, "test_inferred_color_mode_light")
-        inferred_state = helper.initial_states[inferred_light.key]
+        inferred_state = sub.initial_states[inferred_light.key]
         assert inferred_state.state is True
         assert inferred_state.red == pytest.approx(1.0, abs=0.01)
         assert inferred_state.green == pytest.approx(0.0, abs=0.01)

tests/integration/test_light_repeat_effect.py

@@ -14,10 +14,10 @@ import asyncio
 import re
 from typing import Any
 
-from aioesphomeapi import EntityState, LightState
+from aioesphomeapi import LightState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -47,23 +47,13 @@ async def test_light_repeat_effect(
         entities, _ = await client.list_entities_services()
         light = require_entity(entities, "test_rgb_light", exact=True)
 
-        state_futures: dict[int, asyncio.Future[LightState]] = {}
+        sub = await subscribe_after_initial_states(client, entities)
 
-        def on_state(state: EntityState) -> None:
-            if isinstance(state, LightState) and state.key in state_futures:
-                future = state_futures[state.key]
-                if not future.done():
-                    future.set_result(state)
-
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
-
-        async def send_and_wait(timeout: float = 5.0, **kwargs: Any) -> LightState:
-            """Send a light command and wait for the matching state response."""
-            state_futures[light.key] = asyncio.get_running_loop().create_future()
+        async def send_and_wait(**kwargs: Any) -> LightState:
+            """Send a light command and return the light's next state."""
+            next_light = sub.waiter.next_state(light.key, LightState)
             client.light_command(key=light.key, **kwargs)
-            return await asyncio.wait_for(state_futures[light.key], timeout=timeout)
+            return await next_light
 
         # Plain turn-on leaves the published colour at the white default
         state = await send_and_wait(state=True, transition_length=0.0)

tests/integration/test_light_restore_state.py

@@ -10,16 +10,14 @@ Tests that:
 
 from __future__ import annotations
 
-import asyncio
 import socket
-from typing import Any
 
 from aioesphomeapi import LightInfo, LightState
 import pytest
 
 from .conftest import run_binary_and_wait_for_port, wait_and_connect_api_client
 from .host_prefs import clear_host_prefs
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import CompileFunction, ConfigWriter
 
 DEVICE_NAME = "host-light-restore-state-test"
@@ -54,38 +52,26 @@ async def test_light_restore_state(
         # A single subscription serves both initial-state capture and later
         # state-change waits -- a second subscribe_states call would restart the
         # device's initial-state iterator and could resolve a wait on replayed data.
-        loop = asyncio.get_running_loop()
-        state_futures: dict[int, asyncio.Future[Any]] = {}
+        sub = await subscribe_after_initial_states(client, entities)
 
-        def on_state(state: Any) -> None:
-            if state.key in state_futures and not state_futures[state.key].done():
-                state_futures[state.key].set_result(state)
-
-        helper = InitialStateHelper(entities)
-        client.subscribe_states(helper.on_state_wrapper(on_state))
-        await helper.wait_for_initial_states()
-
-        keep_initial = helper.initial_states[keep_light.key]
+        keep_initial = sub.initial_states[keep_light.key]
         assert isinstance(keep_initial, LightState)
         assert keep_initial.state is False
 
-        override_initial = helper.initial_states[override_light.key]
+        override_initial = sub.initial_states[override_light.key]
         assert isinstance(override_initial, LightState)
         assert override_initial.state is False
 
         # Set both lights to a known, saved state
-        async def wait_for_state(key: int, timeout: float = 2.0) -> Any:
-            state_futures[key] = loop.create_future()
-            try:
-                return await asyncio.wait_for(state_futures[key], timeout)
-            finally:
-                state_futures.pop(key, None)
-
+        next_keep = sub.waiter.next_state(keep_light.key, LightState, timeout=2.0)
         client.light_command(key=keep_light.key, state=True, brightness=0.64)
-        await wait_for_state(keep_light.key)
+        await next_keep
 
+        next_override = sub.waiter.next_state(
+            override_light.key, LightState, timeout=2.0
+        )
         client.light_command(key=override_light.key, state=True, brightness=0.3)
-        await wait_for_state(override_light.key)
+        await next_override
 
     # --- Second run: same binary, same prefs file ---
     async with (
@@ -96,19 +82,17 @@ async def test_light_restore_state(
         keep_light = require_entity(entities, "test_light_keep", LightInfo)
         override_light = require_entity(entities, "test_light_override", LightInfo)
 
-        helper = InitialStateHelper(entities)
-        client.subscribe_states(helper.on_state_wrapper(lambda s: None))
-        await helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # restore_state: {} -- comes back exactly as saved
-        keep_state = helper.initial_states[keep_light.key]
+        keep_state = sub.initial_states[keep_light.key]
         assert isinstance(keep_state, LightState)
         assert keep_state.state is True
         assert keep_state.brightness == pytest.approx(0.64, abs=0.01)
 
         # restore_state: with overrides -- state inverted, brightness forced to 100%
         # regardless of what was actually saved (0.3)
-        override_state = helper.initial_states[override_light.key]
+        override_state = sub.initial_states[override_light.key]
         assert isinstance(override_state, LightState)
         assert override_state.state is False
         assert override_state.brightness == pytest.approx(1.0, abs=0.01)

tests/integration/test_light_resume_effect.py

@@ -7,13 +7,12 @@ parameters.
 
 from __future__ import annotations
 
-import asyncio
 from typing import Any
 
-from aioesphomeapi import ColorMode, EntityState, LightInfo, LightState
+from aioesphomeapi import ColorMode, LightInfo, LightState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -32,26 +31,14 @@ async def test_light_resume_effect(
         light_resume = require_entity(entities, "test_light_resume", exact=True)
         light_no_resume = require_entity(entities, "test_light_no_resume", exact=True)
 
-        state_futures: dict[int, asyncio.Future[LightState]] = {}
-
-        def on_state(state: EntityState) -> None:
-            if isinstance(state, LightState) and state.key in state_futures:
-                future = state_futures[state.key]
-                if not future.done():
-                    future.set_result(state)
-
         # Drain initial state burst
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
-        async def send_and_wait(
-            light: LightInfo, timeout: float = 5.0, **kwargs: Any
-        ) -> LightState:
-            """Send a light command and wait for the matching state response."""
-            state_futures[light.key] = asyncio.get_running_loop().create_future()
+        async def send_and_wait(light: LightInfo, **kwargs: Any) -> LightState:
+            """Send a light command and return the light's next state."""
+            next_light = sub.waiter.next_state(light.key, LightState)
             client.light_command(key=light.key, **kwargs)
-            return await asyncio.wait_for(state_futures[light.key], timeout=timeout)
+            return await next_light
 
         # Test 1: Start with Pulse effect
         state = await send_and_wait(light_resume, state=True, effect="Pulse Effect")

tests/integration/test_light_toggle_action.py

@@ -2,12 +2,12 @@
 
 from __future__ import annotations
 
-import asyncio
+from functools import partial
 
-from aioesphomeapi import ButtonInfo, EntityState, LightInfo, LightState
+from aioesphomeapi import LightInfo, LightState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import press_and_wait, require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -18,46 +18,24 @@ async def test_light_toggle_action(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test light.toggle with and without transition_length."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        light_state_future: asyncio.Future[LightState] | None = None
-
-        def on_state(state: EntityState) -> None:
-            if (
-                isinstance(state, LightState)
-                and light_state_future is not None
-                and not light_state_future.done()
-            ):
-                light_state_future.set_result(state)
-
-        async def wait_for_light_state(timeout: float = 5.0) -> LightState:
-            nonlocal light_state_future
-            light_state_future = loop.create_future()
-            try:
-                return await asyncio.wait_for(light_state_future, timeout)
-            finally:
-                light_state_future = None
-
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         require_entity(entities, "test_light", LightInfo)
 
-        async def press_and_wait(name: str) -> LightState:
-            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
-            client.button_command(btn.key)
-            return await wait_for_light_state()
+        press = partial(
+            press_and_wait, client, entities, sub.waiter, state_type=LightState
+        )
 
         # Test 1: toggle without transition_length flips off->on
-        state = await press_and_wait("Toggle")
+        state = await press("toggle")
         assert state.state is True
 
         # Test 2: toggle with transition_length flips on->off
-        state = await press_and_wait("Toggle With Trans")
+        state = await press("toggle_with_trans")
         assert state.state is False
 
         # Test 3: toggle without transition_length flips off->on again
-        state = await press_and_wait("Toggle")
+        state = await press("toggle")
         assert state.state is True

tests/integration/test_light_transition_state_publish_interval.py

@@ -16,7 +16,7 @@ from aioesphomeapi import (
 )
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity, wait_for_state
+from .state_utils import require_entity, subscribe_after_initial_states, wait_for_state
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 Timeline = list[tuple[float, LightState]]
@@ -75,10 +75,8 @@ class _Recorder:
 async def _subscribe(client: APIClient) -> tuple[list[EntityInfo], _Recorder]:
     """List entities and attach a recorder once the initial states have arrived."""
     entities, _ = await client.list_entities_services()
-    helper = InitialStateHelper(entities)
     recorder = _Recorder()
-    client.subscribe_states(helper.on_state_wrapper(recorder.on_state))
-    await helper.wait_for_initial_states()
+    await subscribe_after_initial_states(client, entities, on_state=recorder.on_state)
     return entities, recorder
 
 

tests/integration/test_preference_key_stability.py

@@ -30,7 +30,11 @@ from esphome.helpers import fnv1_hash, fnv1_hash_name, fnv1_hash_object_id
 
 from .conftest import run_binary_and_wait_for_port, wait_and_connect_api_client
 from .host_prefs import clear_host_prefs, write_host_prefs
-from .state_utils import InitialStateHelper, assert_device_name, require_entity
+from .state_utils import (
+    assert_device_name,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import CompileFunction, ConfigWriter
 
 DEVICE_NAME = "host-pref-key-stability"
@@ -102,15 +106,11 @@ async def test_preference_key_stability(
             )
             text_entity = require_entity(entities, "test_text", TextInfo, "Test Text")
 
-            initial_state_helper = InitialStateHelper(entities)
-            client.subscribe_states(
-                initial_state_helper.on_state_wrapper(lambda s: None)
-            )
-            await initial_state_helper.wait_for_initial_states()
+            sub = await subscribe_after_initial_states(client, entities)
 
-            switch_state = initial_state_helper.initial_states[switch_entity.key]
-            number_state = initial_state_helper.initial_states[number_entity.key]
-            text_state = initial_state_helper.initial_states[text_entity.key]
+            switch_state = sub.initial_states[switch_entity.key]
+            number_state = sub.initial_states[number_entity.key]
+            text_state = sub.initial_states[text_entity.key]
             assert isinstance(switch_state, SwitchState)
             assert isinstance(number_state, NumberState)
             assert isinstance(text_state, TextState)

tests/integration/test_sensor_filters_delta.py

@@ -8,7 +8,7 @@ import math
 from aioesphomeapi import ButtonInfo, EntityState, SensorState
 import pytest
 
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping
+from .state_utils import build_key_to_entity_mapping, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -109,14 +109,8 @@ async def test_sensor_filters_delta(
             },
         )
 
-        # Set up initial state helper with all entities
-        initial_state_helper = InitialStateHelper(entities)
-
-        # Subscribe to state changes with wrapper
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial states
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Find all buttons
         button_name_map = {

tests/integration/test_sensor_filters_rh_correction.py

@@ -5,7 +5,11 @@ from __future__ import annotations
 from aioesphomeapi import ButtonInfo
 import pytest
 
-from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .state_utils import (
+    SensorStateCollector,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -29,11 +33,9 @@ async def test_sensor_filters_rh_correction(
     ):
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
-        await initial_state_helper.wait_for_initial_states()
 
         done = collector.add_waiter(
             lambda: all(

tests/integration/test_sensor_filters_ring_buffer.py

@@ -7,7 +7,11 @@ import asyncio
 from aioesphomeapi import EntityState, SensorState
 import pytest
 
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping, require_entity
+from .state_utils import (
+    build_key_to_entity_mapping,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -77,14 +81,8 @@ async def test_sensor_filters_ring_buffer(
             ],
         )
 
-        # Set up initial state helper with all entities
-        initial_state_helper = InitialStateHelper(entities)
-
-        # Subscribe to state changes with wrapper
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial states to be sent before pressing button
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Find the publish button
         publish_button = require_entity(entities, "publish_values_button")

tests/integration/test_sensor_filters_sliding_window.py

@@ -7,7 +7,11 @@ import asyncio
 from aioesphomeapi import EntityState, SensorState
 import pytest
 
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping, require_entity
+from .state_utils import (
+    build_key_to_entity_mapping,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -109,14 +113,8 @@ async def test_sensor_filters_sliding_window(
             ],
         )
 
-        # Set up initial state helper with all entities
-        initial_state_helper = InitialStateHelper(entities)
-
-        # Subscribe to state changes with wrapper
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial states to be sent before pressing button
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Find the publish button
         publish_button = require_entity(entities, "publish_values_button")
@@ -231,14 +229,8 @@ async def test_sensor_filters_nan_handling(
         # Build key-to-sensor mapping
         key_to_sensor = build_key_to_entity_mapping(entities, ["min_nan", "max_nan"])
 
-        # Set up initial state helper with all entities
-        initial_state_helper = InitialStateHelper(entities)
-
-        # Subscribe to state changes with wrapper
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial states
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Find the publish button
         publish_button = require_entity(entities, "publish_nan_values_button")
@@ -323,14 +315,8 @@ async def test_sensor_filters_ring_buffer_wraparound(
         # Build key-to-sensor mapping
         key_to_sensor = build_key_to_entity_mapping(entities, ["wraparound_min"])
 
-        # Set up initial state helper with all entities
-        initial_state_helper = InitialStateHelper(entities)
-
-        # Subscribe to state changes with wrapper
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial state
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Find the publish button
         publish_button = require_entity(entities, "publish_wraparound_button")

tests/integration/test_sensor_filters_value_list.py

@@ -8,7 +8,7 @@ import math
 from aioesphomeapi import ButtonInfo, EntityState, SensorState
 import pytest
 
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping
+from .state_utils import build_key_to_entity_mapping, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -116,14 +116,8 @@ async def test_sensor_filters_value_list(
             },
         )
 
-        # Set up initial state helper with all entities
-        initial_state_helper = InitialStateHelper(entities)
-
-        # Subscribe to state changes with wrapper
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial states
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Find all buttons
         button_name_map = {

tests/integration/test_sensor_timeout_filter.py

@@ -7,7 +7,11 @@ import asyncio
 from aioesphomeapi import EntityState, SensorState
 import pytest
 
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping, require_entity
+from .state_utils import (
+    build_key_to_entity_mapping,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -87,10 +91,7 @@ async def test_sensor_timeout_filter(
             ],
         )
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Helper to find buttons by object_id substring
         def find_button(object_id_substring: str) -> int:

tests/integration/test_sensor_timeout_throttle_filter.py

@@ -8,7 +8,11 @@ import math
 from aioesphomeapi import ButtonInfo, EntityState, SensorState
 import pytest
 
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping, require_entity
+from .state_utils import (
+    build_key_to_entity_mapping,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -39,9 +43,7 @@ async def test_sensor_timeout_throttle_filter(
         key_to_sensor = build_key_to_entity_mapping(
             entities, {"merged": "Merged", "reference": "Reference"}
         )
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         button = require_entity(entities, "drive", ButtonInfo, exact=True)
         client.button_command(button.key)

tests/integration/test_status_flags.py

@@ -19,7 +19,11 @@ import asyncio
 
 import pytest
 
-from .state_utils import InitialStateHelper, SensorTracker, build_key_to_entity_mapping
+from .state_utils import (
+    SensorTracker,
+    build_key_to_entity_mapping,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 # Time to let the host-mode main loop run so status_led_light.loop() can
@@ -69,9 +73,9 @@ async def test_status_flags(
 
         # 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))
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(
+            client, entities, on_state=tracker.on_state
+        )
 
         async def call(name: str) -> None:
             await client.execute_service(svc[name], {})

tests/integration/test_status_momentary.py

@@ -8,7 +8,7 @@ from aioesphomeapi import BinarySensorState, EntityState
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper
+from .state_utils import subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -37,11 +37,9 @@ async def test_status_momentary(
             if isinstance(state, BinarySensorState):
                 changes.put_nowait((state.key, state.state, loop.time()))
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
-        assert initial_state_helper.initial_states[warning_key].state is False
-        assert initial_state_helper.initial_states[error_key].state is False
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
+        assert sub.initial_states[warning_key].state is False
+        assert sub.initial_states[error_key].state is False
 
         async def next_change(key: int) -> tuple[bool, float]:
             while True:

tests/integration/test_strftime_to.py

@@ -7,7 +7,11 @@ import asyncio
 from aioesphomeapi import EntityState, TextSensorState
 import pytest
 
-from .state_utils import InitialStateHelper, assert_device_name, require_entity
+from .state_utils import (
+    assert_device_name,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -39,7 +43,7 @@ async def test_strftime_to(
             entities, "time_error_format", description="Time Error Format sensor"
         )
 
-        # Set up state tracking with InitialStateHelper
+        # Set up state tracking
         loop = asyncio.get_running_loop()
         states: dict[int, TextSensorState] = {}
         all_received = loop.create_future()
@@ -49,7 +53,6 @@ async def test_strftime_to(
             string_format.key,
             error_format.key,
         }
-        initial_state_helper = InitialStateHelper(entities)
 
         def on_state(state: EntityState) -> None:
             if isinstance(state, TextSensorState) and not state.missing_state:
@@ -57,11 +60,8 @@ async def test_strftime_to(
                 if expected_keys <= states.keys() and not all_received.done():
                     all_received.set_result(True)
 
-        # Subscribe with the wrapper that filters initial states
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for initial states to be broadcast
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Wait for all expected states
         await wait_or_fail(

tests/integration/test_template_alarm_control_panel_many_sensors.py

@@ -6,7 +6,7 @@ import aioesphomeapi
 from aioesphomeapi.model import APIIntEnum
 import pytest
 
-from .state_utils import InitialStateHelper, require_only_entity
+from .state_utils import require_only_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -66,20 +66,11 @@ async def test_template_alarm_control_panel_many_sensors(
             f"Got: {actual_sensor_names}"
         )
 
-        # Use InitialStateHelper to wait for all initial states
-        state_helper = InitialStateHelper(entities)
-
-        def on_state(state: aioesphomeapi.EntityState) -> None:
-            # We'll receive subsequent states here after initial states
-            pass
-
-        client.subscribe_states(state_helper.on_state_wrapper(on_state))
-
         # Wait for all initial states
-        await state_helper.wait_for_initial_states(timeout=5.0)
+        sub = await subscribe_after_initial_states(client, entities)
 
         # Verify the alarm state is disarmed initially
-        alarm_state = state_helper.initial_states.get(alarm_info.key)
+        alarm_state = sub.initial_states.get(alarm_info.key)
         assert alarm_state is not None, "Alarm control panel initial state not received"
         assert isinstance(alarm_state, aioesphomeapi.AlarmControlPanelEntityState)
         assert alarm_state.state == aioesphomeapi.AlarmControlPanelState.DISARMED, (
@@ -88,7 +79,7 @@ async def test_template_alarm_control_panel_many_sensors(
 
         # Verify all 10 binary sensors have initial states
         binary_sensor_states = [
-            state_helper.initial_states.get(sensor.key) for sensor in binary_sensors
+            sub.initial_states.get(sensor.key) for sensor in binary_sensors
         ]
         assert all(state is not None for state in binary_sensor_states), (
             "Not all binary sensors have initial states"

tests/integration/test_template_climate_basic.py

@@ -27,9 +27,9 @@ import pytest
 
 from .host_prefs import clear_host_prefs
 from .state_utils import (
-    InitialStateHelper,
     require_entity,
     require_only_entity,
+    subscribe_after_initial_states,
     wait_for_state,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -65,7 +65,6 @@ async def test_template_climate_basic(
             )
 
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
 
         # Advertised capabilities come straight from the supported_*/custom_* config lists.
@@ -85,12 +84,9 @@ async def test_template_climate_basic(
 
         report_button = require_entity(entities, "simulate_device_report", ButtonInfo)
 
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(lambda state: None)
-        )
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
-        initial = initial_state_helper.initial_states.get(test_climate.key)
+        initial = sub.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"
         assert isinstance(initial, aioesphomeapi.ClimateState)
         assert initial.current_temperature == pytest.approx(22.5, abs=0.1)

tests/integration/test_template_climate_custom_modes.py

@@ -13,9 +13,9 @@ import pytest
 from .host_prefs import clear_host_prefs
 from .log_utils import LineWaiter
 from .state_utils import (
-    InitialStateHelper,
     require_entity,
     require_only_entity,
+    subscribe_after_initial_states,
     wait_for_state,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -47,7 +47,6 @@ async def test_template_climate_custom_modes(
             )
 
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
 
         assert set(test_climate.supported_custom_fan_modes) == {
@@ -63,12 +62,9 @@ async def test_template_climate_custom_modes(
 
         report_button = require_entity(entities, "simulate_device_report", ButtonInfo)
 
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(lambda state: None)
-        )
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
-        initial = initial_state_helper.initial_states.get(test_climate.key)
+        initial = sub.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"
         assert isinstance(initial, aioesphomeapi.ClimateState)
         assert initial.custom_fan_mode == ""

tests/integration/test_template_climate_nonoptimistic.py

@@ -23,9 +23,9 @@ import pytest
 from .host_prefs import clear_host_prefs
 from .log_utils import LineWaiter
 from .state_utils import (
-    InitialStateHelper,
     require_entity,
     require_only_entity,
+    subscribe_after_initial_states,
     wait_for_state,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -55,17 +55,15 @@ async def test_template_climate_nonoptimistic(
                 state_updates.append(state)
 
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
 
         confirm_button = require_entity(
             entities, "simulate_device_confirmation", ButtonInfo
         )
 
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
 
-        initial = initial_state_helper.initial_states.get(test_climate.key)
+        initial = sub.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"
         assert isinstance(initial, aioesphomeapi.ClimateState)
         assert initial.mode == ClimateMode.OFF

tests/integration/test_template_climate_on_control_ordering.py

@@ -13,7 +13,11 @@ import pytest
 
 from .host_prefs import clear_host_prefs
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, require_only_entity, wait_for_state
+from .state_utils import (
+    require_only_entity,
+    subscribe_after_initial_states,
+    wait_for_state,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 DEVICE_NAME = "tmpl-clim-oc-order"
@@ -43,13 +47,9 @@ async def test_template_climate_on_control_ordering(
             )
 
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
 
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(lambda state: None)
-        )
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities)
 
         client.climate_command(test_climate.key, mode=ClimateMode.HEAT)
         state = await wait_for_climate_state()

tests/integration/test_template_climate_publish_all_fields.py

@@ -26,9 +26,9 @@ import pytest
 from .host_prefs import clear_host_prefs
 from .log_utils import LineWaiter
 from .state_utils import (
-    InitialStateHelper,
     require_entity,
     require_only_entity,
+    subscribe_after_initial_states,
     wait_for_state,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -58,13 +58,11 @@ async def test_template_climate_publish_all_fields(
                 state_updates.append(state)
 
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         require_only_entity(entities, ClimateInfo)
 
         publish_button = require_entity(entities, "publish_all", ButtonInfo)
 
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         client.button_command(publish_button.key)
         state = await wait_for_state(

tests/integration/test_template_climate_sensor_push.py

@@ -16,9 +16,9 @@ import pytest
 
 from .host_prefs import clear_host_prefs
 from .state_utils import (
-    InitialStateHelper,
     require_entity,
     require_only_entity,
+    subscribe_after_initial_states,
     wait_for_state,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -47,7 +47,6 @@ async def test_template_climate_sensor_push(
                 state_updates.append(state)
 
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
 
         publish_temp = require_entity(entities, "publish_temperature", ButtonInfo)
@@ -56,10 +55,9 @@ async def test_template_climate_sensor_push(
         )
         publish_humidity = require_entity(entities, "publish_humidity", ButtonInfo)
 
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
 
-        initial = initial_state_helper.initial_states.get(test_climate.key)
+        initial = sub.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"
         assert isinstance(initial, aioesphomeapi.ClimateState)
         # Neither backing sensor has published anything yet.

tests/integration/test_template_climate_set_actions.py

@@ -20,9 +20,9 @@ import pytest
 from .host_prefs import clear_host_prefs
 from .log_utils import LineWaiter
 from .state_utils import (
-    InitialStateHelper,
     require_entity,
     require_only_entity,
+    subscribe_after_initial_states,
     wait_for_state,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -49,7 +49,6 @@ async def test_template_climate_set_actions(
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
 
         report_button = require_entity(entities, "report_device_state", ButtonInfo)
@@ -57,10 +56,7 @@ async def test_template_climate_set_actions(
             entities, "report_unsupported_mode", ButtonInfo
         )
 
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(lambda state: None)
-        )
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities)
 
         # Both traits are derived from the low/high and humidity set actions, not declared.
         assert test_climate.supports_two_point_target_temperature

tests/integration/test_template_climate_two_point_temperature.py

@@ -15,9 +15,9 @@ import pytest
 from .host_prefs import clear_host_prefs
 from .log_utils import LineWaiter
 from .state_utils import (
-    InitialStateHelper,
     require_entity,
     require_only_entity,
+    subscribe_after_initial_states,
     wait_for_state,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -49,7 +49,6 @@ async def test_template_climate_two_point_temperature(
             )
 
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_climate = require_only_entity(entities, ClimateInfo)
         assert test_climate.name == "Test Two-Point Heatpump"
         assert test_climate.supports_two_point_target_temperature
@@ -57,13 +56,9 @@ async def test_template_climate_two_point_temperature(
 
         report_button = require_entity(entities, "simulate_device_report", ButtonInfo)
 
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(lambda state: None)
-        )
+        sub = await subscribe_after_initial_states(client, entities)
 
-        await initial_state_helper.wait_for_initial_states()
-
-        initial = initial_state_helper.initial_states.get(test_climate.key)
+        initial = sub.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"
         assert isinstance(initial, aioesphomeapi.ClimateState)
         # Nothing has been published yet: settable fields have no sensor to seed them from, so

tests/integration/test_template_text_save.py

@@ -10,7 +10,6 @@ from __future__ import annotations
 
 import asyncio
 import socket
-from typing import Any
 
 from aioesphomeapi import TextInfo, TextState
 import pytest
@@ -18,7 +17,11 @@ import pytest
 from .conftest import run_binary_and_wait_for_port, wait_and_connect_api_client
 from .host_prefs import clear_host_prefs
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, assert_device_name, require_entity
+from .state_utils import (
+    assert_device_name,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import CompileFunction, ConfigWriter
 
 DEVICE_NAME = "host-template-text-save-test"
@@ -60,33 +63,17 @@ async def test_template_text_save(
             entities, "test_text_restore", TextInfo, "Test Text Restore"
         )
 
-        # Set up state tracking
-        loop = asyncio.get_running_loop()
-        state_futures: dict[int, asyncio.Future[Any]] = {}
-
-        def on_state(state: Any) -> None:
-            if state.key in state_futures and not state_futures[state.key].done():
-                state_futures[state.key].set_result(state)
-
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # Verify initial value from config
-        initial = initial_state_helper.initial_states[text_entity.key]
+        initial = sub.initial_states[text_entity.key]
         assert isinstance(initial, TextState)
         assert initial.state == "hello"
 
-        async def wait_for_state(key: int, timeout: float = 2.0) -> Any:
-            state_futures[key] = loop.create_future()
-            try:
-                return await asyncio.wait_for(state_futures[key], timeout)
-            finally:
-                state_futures.pop(key, None)
-
         # Set a new value that fits within max_length
+        next_text = sub.waiter.next_state(text_entity.key, TextState, timeout=2.0)
         client.text_command(key=text_entity.key, state="world")
-        state = await wait_for_state(text_entity.key)
+        state = await next_text
         assert state.state == "world"
 
         # Set the same value again - should NOT produce "too long" warning
@@ -112,12 +99,10 @@ async def test_template_text_save(
             entities, "test_text_restore", TextInfo, "Test Text Restore"
         )
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(lambda s: None))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # The value should be "world" - restored from preferences
-        restored = initial_state_helper.initial_states[text_entity.key]
+        restored = sub.initial_states[text_entity.key]
         assert isinstance(restored, TextState)
         assert restored.state == "world", (
             f"Expected restored value 'world', got '{restored.state}'"

tests/integration/test_template_time.py

@@ -7,7 +7,7 @@ import asyncio
 from aioesphomeapi import EntityState, TextSensorState
 import pytest
 
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping
+from .state_utils import build_key_to_entity_mapping, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -39,9 +39,7 @@ async def test_template_time(
                 if name is not None and state.state == EXPECTED_STATES[name]:
                     events[name].set()
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         for name, event in events.items():
             await wait_or_fail(

tests/integration/test_text_command.py

@@ -7,13 +7,14 @@ transmitted via the API.
 
 from __future__ import annotations
 
-import asyncio
-from typing import Any
-
 from aioesphomeapi import TextInfo, TextState
 import pytest
 
-from .state_utils import InitialStateHelper, assert_device_name, require_entity
+from .state_utils import (
+    assert_device_name,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -24,7 +25,6 @@ async def test_text_command(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test text command with various string values including edge cases."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
         # Verify we can get device info
         await assert_device_name(client, "host-text-command-test")
@@ -41,81 +41,57 @@ async def test_text_command(
             entities, "test_text_long", TextInfo, "Test Text Long entity"
         )
 
-        # Track state changes
-        states: dict[int, Any] = {}
-        state_futures: dict[int, asyncio.Future[Any]] = {}
-
-        def on_state(state: Any) -> None:
-            states[state.key] = state
-            if state.key in state_futures and not state_futures[state.key].done():
-                state_futures[state.key].set_result(state)
-
-        # Set up InitialStateHelper to swallow initial state broadcasts
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for all initial states to be received
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # Verify initial states were received
-        assert test_text.key in initial_state_helper.initial_states
-        initial_text_state = initial_state_helper.initial_states[test_text.key]
+        assert test_text.key in sub.initial_states
+        initial_text_state = sub.initial_states[test_text.key]
         assert isinstance(initial_text_state, TextState)
         assert initial_text_state.state == "initial"
 
-        async def wait_for_state_change(key: int, timeout: float = 2.0) -> Any:
-            """Wait for a state change for the given entity key."""
-            state_futures[key] = loop.create_future()
-            try:
-                return await asyncio.wait_for(state_futures[key], timeout)
-            finally:
-                state_futures.pop(key, None)
+        async def send_and_wait(key: int, value: str) -> TextState:
+            """Send a text command and return the entity's next state."""
+            next_text = sub.waiter.next_state(key, TextState, timeout=2.0)
+            client.text_command(key=key, state=value)
+            return await next_text
 
         # Test 1: Simple text value
-        client.text_command(key=test_text.key, state="hello world")
-        state = await wait_for_state_change(test_text.key)
+        state = await send_and_wait(test_text.key, "hello world")
         assert state.state == "hello world"
 
         # Test 2: Empty string (edge case for zero-copy)
-        client.text_command(key=test_text.key, state="")
-        state = await wait_for_state_change(test_text.key)
+        state = await send_and_wait(test_text.key, "")
         assert state.state == ""
 
         # Test 3: Single character
-        client.text_command(key=test_text.key, state="x")
-        state = await wait_for_state_change(test_text.key)
+        state = await send_and_wait(test_text.key, "x")
         assert state.state == "x"
 
         # Test 4: String with special characters
-        client.text_command(key=test_text.key, state="hello\tworld\n!")
-        state = await wait_for_state_change(test_text.key)
+        state = await send_and_wait(test_text.key, "hello\tworld\n!")
         assert state.state == "hello\tworld\n!"
 
         # Test 5: Unicode characters
-        client.text_command(key=test_text.key, state="hello 世界 🌍")
-        state = await wait_for_state_change(test_text.key)
+        state = await send_and_wait(test_text.key, "hello 世界 🌍")
         assert state.state == "hello 世界 🌍"
 
         # Test 6: Long string (tests buffer handling)
         long_text = "a" * 200
-        client.text_command(key=test_text_long.key, state=long_text)
-        state = await wait_for_state_change(test_text_long.key)
+        state = await send_and_wait(test_text_long.key, long_text)
         assert state.state == long_text
         assert len(state.state) == 200
 
         # Test 7: Password field (same mechanism, different mode)
-        client.text_command(key=test_password.key, state="newpassword123")
-        state = await wait_for_state_change(test_password.key)
+        state = await send_and_wait(test_password.key, "newpassword123")
         assert state.state == "newpassword123"
 
         # Test 8: String with null bytes embedded (edge case)
         # Note: protobuf strings should handle this but it's good to verify
-        client.text_command(key=test_text.key, state="before\x00after")
-        state = await wait_for_state_change(test_text.key)
+        state = await send_and_wait(test_text.key, "before\x00after")
         assert state.state == "before\x00after"
 
         # Test 9: Rapid successive commands (tests buffer reuse)
         for i in range(5):
-            client.text_command(key=test_text.key, state=f"rapid_{i}")
-            state = await wait_for_state_change(test_text.key)
+            state = await send_and_wait(test_text.key, f"rapid_{i}")
             assert state.state == f"rapid_{i}"

tests/integration/test_uart_mock_ld2410.py

@@ -22,7 +22,11 @@ from aioesphomeapi import ButtonInfo
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .state_utils import (
+    SensorStateCollector,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -70,14 +74,10 @@ async def test_uart_mock_ld2410(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        # Set up initial state helper
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
 
-        await initial_state_helper.wait_for_initial_states()
-
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
@@ -188,13 +188,10 @@ async def test_uart_mock_ld2410_engineering(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
 
-        await initial_state_helper.wait_for_initial_states()
-
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_uart_mock_ld2410_query_ack.py

@@ -21,7 +21,11 @@ from aioesphomeapi import (
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping, require_entity
+from .state_utils import (
+    build_key_to_entity_mapping,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -113,10 +117,7 @@ async def test_uart_mock_ld2410_query_ack(
             f"Sensor entities missing, found: {sorted(sensor_key_to_name.values())}"
         )
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)

tests/integration/test_uart_mock_ld2410s.py

@@ -41,7 +41,11 @@ import re
 from aioesphomeapi import APIClient, ButtonInfo, EntityState, NumberInfo, NumberState
 import pytest
 
-from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .state_utils import (
+    SensorStateCollector,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -115,9 +119,7 @@ async def _start_scenario(client: APIClient, collector: SensorStateCollector) ->
     """Subscribe the collector past the initial states, then start the mock scenario."""
     entities, _ = await client.list_entities_services()
     collector.build_key_mapping(entities)
-    initial_state_helper = InitialStateHelper(entities)
-    client.subscribe_states(initial_state_helper.on_state_wrapper(collector.on_state))
-    await initial_state_helper.wait_for_initial_states()
+    await subscribe_after_initial_states(client, entities, on_state=collector.on_state)
     start_btn = require_entity(entities, "start_scenario", ButtonInfo)
     client.button_command(start_btn.key)
 
@@ -194,12 +196,10 @@ async def test_uart_mock_ld2410s(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
         max_distance_key = require_entity(entities, "max_distance", NumberInfo).key
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
         # The values read from the module: gate 16 and 10 s
         no_delay_key = require_entity(entities, "no_delay", NumberInfo).key
-        initial = initial_state_helper.initial_states
+        initial = sub.initial_states
         assert initial[max_distance_key].state == pytest.approx(11.2)
         assert initial[no_delay_key].state == pytest.approx(10)
 

tests/integration/test_uart_mock_ld2412.py

@@ -32,22 +32,15 @@ import asyncio
 from collections.abc import Callable
 import math
 
-from aioesphomeapi import (
-    APIClient,
-    ButtonInfo,
-    EntityInfo,
-    EntityState,
-    NumberInfo,
-    NumberState,
-)
+from aioesphomeapi import ButtonInfo, EntityInfo, EntityState, NumberInfo, NumberState
 import pytest
 
 from .log_utils import LineWaiter
 from .state_utils import (
-    InitialStateHelper,
     SensorStateCollector,
     StateWaiter,
     require_entity,
+    subscribe_after_initial_states,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -82,18 +75,6 @@ def _threshold_numbers(entities: list[EntityInfo]) -> dict[str, NumberInfo]:
     }
 
 
-async def _wait_initial(
-    client: APIClient,
-    entities: list[EntityInfo],
-    on_state: Callable[[EntityState], None] = lambda s: None,
-) -> InitialStateHelper:
-    """Subscribe to states and wait until every entity has sent its initial state."""
-    initial_state_helper = InitialStateHelper(entities)
-    client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-    await initial_state_helper.wait_for_initial_states()
-    return initial_state_helper
-
-
 def _number_state_of(key: int) -> Callable[[EntityState], bool]:
     """Return a predicate that matches a number state with a value for ``key``."""
     return lambda s: isinstance(s, NumberState) and not s.missing_state and s.key == key
@@ -175,7 +156,9 @@ async def test_uart_mock_ld2412(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        await _wait_initial(client, entities, collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
+        )
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
@@ -275,7 +258,9 @@ async def test_uart_mock_ld2412_engineering(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        await _wait_initial(client, entities, collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
+        )
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
@@ -362,7 +347,9 @@ async def test_uart_mock_ld2412_engineering_truncated(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        await _wait_initial(client, entities, collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
+        )
 
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
@@ -423,7 +410,7 @@ async def test_uart_mock_ld2412_footer_in_payload(
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
-        initial_state_helper = await _wait_initial(client, entities)
+        sub = await subscribe_after_initial_states(client, entities)
         numbers = _threshold_numbers(entities)
 
         # Gate 5 move is the first footer byte inside the motion answer
@@ -433,7 +420,7 @@ async def test_uart_mock_ld2412_footer_in_payload(
             "gate_0_still_threshold": 0.0,
             "gate_5_still_threshold": 25.0,
         }.items():
-            state = initial_state_helper.initial_states[numbers[name].key]
+            state = sub.initial_states[numbers[name].key]
             assert state.state == pytest.approx(value), name
 
         motion_written = events.next_line_with(expected_motion)
@@ -523,10 +510,10 @@ async def test_uart_mock_ld2412_truncated_answer(
                 states.setdefault(state.key, []).append(state.state)
             state_waiter.on_state(state)
 
-        initial_state_helper = await _wait_initial(client, entities, record)
+        sub = await subscribe_after_initial_states(client, entities, on_state=record)
 
         # The late answer may already have been parsed when the client connected
-        for state in initial_state_helper.initial_states.values():
+        for state in sub.initial_states.values():
             record(state)
 
         try:
@@ -573,17 +560,16 @@ async def test_uart_mock_ld2412_requery_after_giveup(
         gate_5_key = numbers["gate_5_move_threshold"].key
 
         gate_5_values: list[float] = []
-        waiter = StateWaiter()
 
         def on_state(state: EntityState) -> None:
             if _number_state_of(gate_5_key)(state):
                 gate_5_values.append(state.state)
-            waiter.on_state(state)
 
-        initial_state_helper = await _wait_initial(client, entities, on_state)
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
+        waiter = sub.waiter
 
         for key in (gate_0_key, gate_5_key):
-            assert initial_state_helper.initial_states[key].missing_state
+            assert sub.initial_states[key].missing_state
 
         held_back = events.next_line_with(MOTION_HELD_BACK)
         config_mode_left = events.next_line_with(CONFIG_MODE_LEFT)

tests/integration/test_uart_mock_ld2420.py

@@ -23,7 +23,11 @@ from aioesphomeapi import ButtonInfo
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .state_utils import (
+    SensorStateCollector,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -54,14 +58,10 @@ async def test_uart_mock_ld2420(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        # Set up initial state helper
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
 
-        await initial_state_helper.wait_for_initial_states()
-
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
@@ -151,13 +151,10 @@ async def test_uart_mock_ld2420_simple(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
 
-        await initial_state_helper.wait_for_initial_states()
-
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_uart_mock_ld2420_config_exit.py

@@ -8,7 +8,11 @@ from aioesphomeapi import ButtonInfo
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .state_utils import (
+    SensorStateCollector,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -47,11 +51,9 @@ async def test_uart_mock_ld2420_config_exit(
     ):
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
-        await initial_state_helper.wait_for_initial_states()
 
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_uart_mock_ld2450.py

@@ -21,7 +21,11 @@ from aioesphomeapi import ButtonInfo
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .state_utils import (
+    SensorStateCollector,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -83,14 +87,10 @@ async def test_uart_mock_ld2450(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        # Set up initial state helper
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
 
-        await initial_state_helper.wait_for_initial_states()
-
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_uart_mock_ld2450_has_target_only.py

@@ -12,7 +12,7 @@ import asyncio
 from aioesphomeapi import BinarySensorInfo, BinarySensorState, ButtonInfo, EntityState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 # The fixture sends one frame with a target, then 2 s of frames without one
@@ -41,9 +41,7 @@ async def test_uart_mock_ld2450_has_target_only(
                 if state.state and not turned_on.done():
                     turned_on.set_result(None)
 
-        helper = InitialStateHelper(entities)
-        client.subscribe_states(helper.on_state_wrapper(on_state))
-        await helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_uart_mock_ld2450_polygon_zone_restore.py

@@ -7,7 +7,6 @@ Tests:
 
 from __future__ import annotations
 
-import asyncio
 import socket
 import struct
 
@@ -17,7 +16,7 @@ import pytest
 from .conftest import run_binary_and_wait_for_port, wait_and_connect_api_client
 from .host_prefs import clear_host_prefs, read_host_prefs, write_host_prefs
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import CompileFunction, ConfigWriter
 
 DEVICE_NAME = "uart-mock-ld2450-zone-restore"
@@ -53,10 +52,8 @@ async def test_uart_mock_ld2450_polygon_zone_restore(
         ):
             entities, _ = await client.list_entities_services()
             polygon = require_entity(entities, "restore_zone_polygon", TextInfo)
-            helper = InitialStateHelper(entities)
-            client.subscribe_states(helper.on_state_wrapper(lambda state: None))
-            await helper.wait_for_initial_states()
-            state = helper.initial_states[polygon.key]
+            sub = await subscribe_after_initial_states(client, entities)
+            state = sub.initial_states[polygon.key]
             assert isinstance(state, TextState)
             return state.state
 
@@ -67,25 +64,18 @@ async def test_uart_mock_ld2450_polygon_zone_restore(
     ):
         entities, _ = await client.list_entities_services()
         polygon = require_entity(entities, "restore_zone_polygon", TextInfo)
-        loop = asyncio.get_running_loop()
-        published = loop.create_future()
-
-        def on_state(state: object) -> None:
-            if (
-                isinstance(state, TextState)
-                and state.key == polygon.key
-                and state.state == ZONE_POLYGON
-                and not published.done()
-            ):
-                published.set_result(None)
-
-        helper = InitialStateHelper(entities)
-        client.subscribe_states(helper.on_state_wrapper(on_state))
-        await helper.wait_for_initial_states()
-        assert helper.initial_states[polygon.key].state == ""
+        sub = await subscribe_after_initial_states(client, entities)
+        assert sub.initial_states[polygon.key].state == ""
 
+        published = sub.waiter.expect(
+            lambda s: (
+                isinstance(s, TextState)
+                and s.key == polygon.key
+                and s.state == ZONE_POLYGON
+            )
+        )
         client.text_command(polygon.key, ZONE_POLYGON)
-        await asyncio.wait_for(published, timeout=5.0)
+        await published
 
     # Second run: the polygon is restored
     assert await initial_polygon() == ZONE_POLYGON

tests/integration/test_uart_mock_ld2450_polygon_zone_timeout.py

@@ -22,7 +22,7 @@ from aioesphomeapi import (
 import pytest
 
 from .host_prefs import clear_host_prefs
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 DEVICE_NAME = "uart-mock-ld2450-zone-timeout"
@@ -76,9 +76,7 @@ async def test_uart_mock_ld2450_polygon_zone_timeout(
             ):
                 timeout_set.set_result(None)
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
 
         client.text_command(polygon.key, ZONE_POLYGON)
         client.number_command(timeout.key, PRESENCE_TIMEOUT_S)

tests/integration/test_uart_mock_ld2450_polygon_zones.py

@@ -28,7 +28,7 @@ import pytest
 from esphome.components.ld2450.text import POLYGON_MAX_LENGTH, POLYGON_PATTERN
 
 from .host_prefs import clear_host_prefs
-from .state_utils import InitialStateHelper, StateWaiter, require_entity
+from .state_utils import require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 DEVICE_NAME = "uart-mock-ld2450-zones"
@@ -64,28 +64,22 @@ async def test_uart_mock_ld2450_polygon_zones(
         presence_history: dict[str, list[bool | None]] = {
             name: [] for name in presences
         }
-        waiter = StateWaiter()
-        expect = waiter.expect
 
         def on_state(state: EntityState) -> None:
             if isinstance(state, BinarySensorState) and state.key in presence_names:
                 presence_history[presence_names[state.key]].append(
                     None if state.missing_state else state.state
                 )
-            waiter.on_state(state)
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
+        expect = sub.waiter.expect
 
         for name, polygon in polygons.items():
             # Home Assistant validates input and the zone card finds zones with these
             assert polygon.pattern == POLYGON_PATTERN
             assert polygon.max_length == POLYGON_MAX_LENGTH
-            assert initial_state_helper.initial_states[polygon.key].state == ""
-            assert initial_state_helper.initial_states[
-                presences[name].key
-            ].missing_state
+            assert sub.initial_states[polygon.key].state == ""
+            assert sub.initial_states[presences[name].key].missing_state
 
         async def set_polygon(name: str, value: str, expected: str) -> None:
             key = polygons[name].key

tests/integration/test_uart_mock_ld2460.py

@@ -18,7 +18,11 @@ from aioesphomeapi import ButtonInfo
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .state_utils import (
+    SensorStateCollector,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -72,11 +76,9 @@ async def test_uart_mock_ld2460(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
-        await initial_state_helper.wait_for_initial_states()
 
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_uart_mock_pylontech.py

@@ -17,7 +17,11 @@ from aioesphomeapi import ButtonInfo
 import pytest
 
 from .log_utils import LineWaiter
-from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .state_utils import (
+    SensorStateCollector,
+    require_entity,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
 
@@ -88,11 +92,9 @@ async def test_uart_mock_pylontech(
         entities, _ = await client.list_entities_services()
         collector.build_key_mapping(entities)
 
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(
-            initial_state_helper.on_state_wrapper(collector.on_state)
+        await subscribe_after_initial_states(
+            client, entities, on_state=collector.on_state
         )
-        await initial_state_helper.wait_for_initial_states()
 
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_valve_control_action.py

@@ -6,12 +6,12 @@ field combinations and the lambda path.
 
 from __future__ import annotations
 
-import asyncio
+from functools import partial
 
-from aioesphomeapi import ButtonInfo, EntityState, ValveInfo, ValveOperation, ValveState
+from aioesphomeapi import ValveInfo, ValveOperation, ValveState
 import pytest
 
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import press_and_wait, require_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -22,51 +22,29 @@ async def test_valve_control_action(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test valve ControlAction with constants and a lambda."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        valve_state_future: asyncio.Future[ValveState] | None = None
-
-        def on_state(state: EntityState) -> None:
-            if (
-                isinstance(state, ValveState)
-                and valve_state_future is not None
-                and not valve_state_future.done()
-            ):
-                valve_state_future.set_result(state)
-
-        async def wait_for_valve_state(timeout: float = 5.0) -> ValveState:
-            nonlocal valve_state_future
-            valve_state_future = loop.create_future()
-            try:
-                return await asyncio.wait_for(valve_state_future, timeout)
-            finally:
-                valve_state_future = None
-
         entities, _ = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         require_entity(entities, "test_valve", ValveInfo)
 
-        async def press_and_wait(name: str) -> ValveState:
-            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
-            client.button_command(btn.key)
-            return await wait_for_valve_state()
+        press = partial(
+            press_and_wait, client, entities, sub.waiter, state_type=ValveState
+        )
 
         # valve.control: position only
-        state = await press_and_wait("Set Position")
+        state = await press("set_position")
         assert state.position == pytest.approx(0.5, abs=0.01)
 
         # valve.control: state alias for position 1.0
-        state = await press_and_wait("Open State")
+        state = await press("open_state")
         assert state.position == pytest.approx(1.0, abs=0.01)
 
         # valve.control: lambda position (test_position global = 0.42)
-        state = await press_and_wait("Lambda Position")
+        state = await press("lambda_position")
         assert state.position == pytest.approx(0.42, abs=0.01)
 
         # valve.control: stop only — template valve's stop_action publishes
         # current_operation: IDLE.
-        state = await press_and_wait("Stop Valve")
+        state = await press("stop_valve")
         assert state.current_operation == ValveOperation.IDLE

tests/integration/test_water_heater_template.py

@@ -14,7 +14,7 @@ from aioesphomeapi import (
 )
 import pytest
 
-from .state_utils import InitialStateHelper, require_only_entity
+from .state_utils import require_only_entity, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -25,32 +25,9 @@ async def test_water_heater_template(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test template water heater basic state and mode changes."""
-    loop = asyncio.get_running_loop()
     async with run_compiled(yaml_config), api_client_connected() as client:
-        states: dict[int, aioesphomeapi.EntityState] = {}
-        state_future: asyncio.Future[WaterHeaterState] | None = None
-
-        def on_state(state: aioesphomeapi.EntityState) -> None:
-            states[state.key] = state
-            if (
-                isinstance(state, WaterHeaterState)
-                and state_future is not None
-                and not state_future.done()
-            ):
-                state_future.set_result(state)
-
-        async def wait_for_state(timeout: float = 5.0) -> WaterHeaterState:
-            """Wait for next water heater state change."""
-            nonlocal state_future
-            state_future = loop.create_future()
-            try:
-                return await asyncio.wait_for(state_future, timeout)
-            finally:
-                state_future = None
-
         # Get entities and set up state synchronization
         entities, services = await client.list_entities_services()
-        initial_state_helper = InitialStateHelper(entities)
         test_water_heater = require_only_entity(entities, WaterHeaterInfo)
 
         # Verify water heater entity info
@@ -72,14 +49,11 @@ async def test_water_heater_template(
             f"Expected 4 supported modes, got {len(supported_modes)}: {supported_modes}"
         )
 
-        # Subscribe with the wrapper that filters initial states
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
         # Wait for all initial states to be broadcast
-        await initial_state_helper.wait_for_initial_states()
+        sub = await subscribe_after_initial_states(client, entities)
 
         # Get initial state and verify
-        initial_state = initial_state_helper.initial_states.get(test_water_heater.key)
+        initial_state = sub.initial_states.get(test_water_heater.key)
         assert initial_state is not None, "Water heater initial state not found"
         assert isinstance(initial_state, WaterHeaterState)
         # Initial mode is OFF (default) since we don't have a mode lambda
@@ -115,37 +89,43 @@ async def test_water_heater_template(
         )
 
         # Test turning on away mode
+        next_state = sub.waiter.next_state(test_water_heater.key, WaterHeaterState)
         client.water_heater_command(test_water_heater.key, away=True)
-        away_on_state = await wait_for_state()
+        away_on_state = await next_state
         assert (away_on_state.state & WaterHeaterStateFlag.AWAY) != 0
         # ON flag should still be set (is_on lambda returns true)
         assert (away_on_state.state & WaterHeaterStateFlag.ON) != 0
 
         # Test turning off away mode
+        next_state = sub.waiter.next_state(test_water_heater.key, WaterHeaterState)
         client.water_heater_command(test_water_heater.key, away=False)
-        away_off_state = await wait_for_state()
+        away_off_state = await next_state
         assert (away_off_state.state & WaterHeaterStateFlag.AWAY) == 0
         assert (away_off_state.state & WaterHeaterStateFlag.ON) != 0
 
         # Test turning off (on=False)
+        next_state = sub.waiter.next_state(test_water_heater.key, WaterHeaterState)
         client.water_heater_command(test_water_heater.key, on=False)
-        off_state = await wait_for_state()
+        off_state = await next_state
         assert (off_state.state & WaterHeaterStateFlag.ON) == 0
         assert (off_state.state & WaterHeaterStateFlag.AWAY) == 0
 
         # Test turning back on (on=True)
+        next_state = sub.waiter.next_state(test_water_heater.key, WaterHeaterState)
         client.water_heater_command(test_water_heater.key, on=True)
-        on_state = await wait_for_state()
+        on_state = await next_state
         assert (on_state.state & WaterHeaterStateFlag.ON) != 0
 
         # Test changing to GAS mode
+        next_state = sub.waiter.next_state(test_water_heater.key, WaterHeaterState)
         client.water_heater_command(test_water_heater.key, mode=WaterHeaterMode.GAS)
-        gas_state = await wait_for_state()
+        gas_state = await next_state
         assert gas_state.mode == WaterHeaterMode.GAS
 
         # Test changing to ECO mode (from GAS)
+        next_state = sub.waiter.next_state(test_water_heater.key, WaterHeaterState)
         client.water_heater_command(test_water_heater.key, mode=WaterHeaterMode.ECO)
-        eco_state = await wait_for_state()
+        eco_state = await next_state
         assert eco_state.mode == WaterHeaterMode.ECO
 
 

tests/unit_tests/test_integration_log_utils.py

@@ -0,0 +1,81 @@
+"""Tests for the integration test LineWaiter."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+import pytest
+
+from tests.integration.log_utils import LineWaiter
+
+
+@pytest.mark.asyncio
+async def test_expect_matches_only_later_lines() -> None:
+    waiter = LineWaiter()
+    waiter.callback("[I][app]: ready")
+    ready = waiter.expect("ready")
+    value = waiter.expect(re.compile(r"value=(\d+)"))
+    assert not ready.done()
+
+    waiter.callback("[D][x]: value=42")
+    waiter.callback("[I][app]: ready again")
+    assert await waiter.wait(ready) == "[I][app]: ready again"
+    match = await waiter.wait(value)
+    assert isinstance(match, re.Match)
+    assert match.group(1) == "42"
+    # wait_for still matches lines seen before the call.
+    assert await waiter.wait_for("app", "ready") == "[I][app]: ready"
+
+
+@pytest.mark.asyncio
+async def test_wait_timeout_reports_pattern_and_lines() -> None:
+    waiter = LineWaiter()
+    for i in range(30):
+        waiter.callback(f"line {i}")
+    future = waiter.expect("never")
+    with pytest.raises(pytest.fail.Exception) as err:
+        await waiter.wait(future, 0.01, "boot message missing")
+    message = str(err.value)
+    assert message.startswith("boot message missing (after 0.01s)")
+    assert "pattern: 'never'" in message
+    assert "lines (last 20 of 30):" in message
+    assert message.endswith("line 29")
+    assert future.cancelled()
+    # The cancelled expectation is dropped on the next line.
+    waiter.callback("never")
+    assert not waiter._expectations
+
+
+def test_count() -> None:
+    waiter = LineWaiter()
+    for line in ("tick 1", "tock", "tick 2"):
+        waiter.callback(line)
+    assert waiter.count("tick") == 2
+    assert waiter.count(re.compile(r"t\w+k$")) == 1
+    assert waiter.count("missing") == 0
+
+
+@pytest.mark.asyncio
+async def test_expect_resolves_from_callback_while_waiting() -> None:
+    waiter = LineWaiter()
+    first = waiter.expect("first")
+    second = waiter.expect("second")
+    loop = asyncio.get_running_loop()
+    loop.call_soon(waiter.callback, "second")
+    loop.call_soon(waiter.callback, "first")
+    assert await waiter.wait(first) == "first"
+    assert await waiter.wait(second) == "second"
+
+
+@pytest.mark.asyncio
+async def test_line_waiter_expectations_share_a_matching_line() -> None:
+    waiter = LineWaiter()
+    first = waiter.expect("tick")
+    second = waiter.expect("tick")
+    waiter.callback("tick 1")
+    assert first.result() == second.result() == "tick 1"
+    third = waiter.expect("tick")
+    waiter.callback("tick 2")
+    assert third.result() == "tick 2"
+    assert waiter.count("tick") == 2

tests/unit_tests/test_integration_state_subscribe.py

@@ -0,0 +1,108 @@
+"""Tests for the state subscription helpers in tests/integration/state_utils.py."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+
+from aioesphomeapi import ButtonInfo, EntityState, SensorInfo, SensorState, SwitchState
+import pytest
+
+from tests.integration.state_utils import (
+    StateSubscription,
+    StateWaiter,
+    press_and_wait,
+    subscribe_after_initial_states,
+)
+
+ENTITIES = [
+    SensorInfo(object_id="temp", key=1),
+    SensorInfo(object_id="humidity", key=2),
+    ButtonInfo(object_id="set_temp", key=3),
+]
+
+
+class FakeClient:
+    """Records the state callback and button presses."""
+
+    def __init__(self) -> None:
+        self.callback: Callable[[EntityState], None] | None = None
+        self.pressed: list[int] = []
+
+    def subscribe_states(self, callback: Callable[[EntityState], None]) -> None:
+        self.callback = callback
+
+    def button_command(self, key: int) -> None:
+        self.pressed.append(key)
+        self.callback(SensorState(key=1, state=42.0))
+
+
+async def _subscribe(client: FakeClient, **kwargs: object) -> StateSubscription:
+    task = asyncio.create_task(
+        subscribe_after_initial_states(client, ENTITIES, **kwargs)
+    )
+    await asyncio.sleep(0)
+    client.callback(SensorState(key=1, state=1.0))
+    client.callback(SensorState(key=2, state=2.0))
+    return await task
+
+
+@pytest.mark.asyncio
+async def test_subscribe_swallows_initial_states_and_forwards_later_ones() -> None:
+    client = FakeClient()
+    received: list[EntityState] = []
+    sub = await _subscribe(client, on_state=received.append)
+    assert sub.initial_states[1].state == 1.0
+    assert sub.initial_states[2].state == 2.0
+    assert received == []
+
+    next_state = sub.waiter.next_state(key=2, state_type=SensorState)
+    client.callback(SensorState(key=1, state=3.0))
+    client.callback(SensorState(key=2, state=4.0))
+    assert (await next_state).state == 4.0
+    assert [s.state for s in received] == [3.0, 4.0]
+
+
+@pytest.mark.asyncio
+async def test_subscribe_timeout_lists_missing_entities() -> None:
+    client = FakeClient()
+    task = asyncio.create_task(
+        subscribe_after_initial_states(client, ENTITIES, timeout=0.01)
+    )
+    await asyncio.sleep(0)
+    client.callback(SensorState(key=1, state=1.0))
+    with pytest.raises(pytest.fail.Exception, match=r"still_waiting[\s\S]*humidity"):
+        await task
+
+
+@pytest.mark.asyncio
+async def test_next_state_filters_by_type_and_times_out() -> None:
+    waiter = StateWaiter()
+    next_switch = waiter.next_state(state_type=SwitchState)
+    waiter.on_state(SensorState(key=1, state=1.0))
+    waiter.on_state(SwitchState(key=5, state=True))
+    assert (await next_switch).key == 5
+    with pytest.raises(TimeoutError, match="SwitchState key=9"):
+        await waiter.next_state(key=9, state_type=SwitchState, timeout=0.01)
+
+
+@pytest.mark.asyncio
+async def test_press_and_wait_presses_button_and_returns_state() -> None:
+    client = FakeClient()
+    sub = await _subscribe(client)
+    state = await press_and_wait(client, ENTITIES, sub.waiter, "set_temp", SensorState)
+    assert client.pressed == [3]
+    assert state.state == 42.0
+
+
+@pytest.mark.asyncio
+async def test_press_and_wait_waits_for_the_given_key() -> None:
+    client = FakeClient()
+    sub = await _subscribe(client)
+    pending = asyncio.ensure_future(
+        press_and_wait(client, ENTITIES, sub.waiter, "set_temp", SensorState, key=2)
+    )
+    await asyncio.sleep(0)
+    assert not pending.done()  # the press published key 1
+    client.callback(SensorState(key=2, state=55.0))
+    assert (await pending).state == 55.0