Keep the state waiter apart from the states dict

tests/integration/test_api_homeassistant_action_no_subscriber.py

@@ -13,6 +13,7 @@ import asyncio
 from aioesphomeapi import ButtonInfo, HomeassistantServiceCall
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name, require_only_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
@@ -25,26 +26,9 @@ async def test_api_homeassistant_action_no_subscriber(
 ) -> None:
     """Undeliverable actions warn in the log; actions after subscribing arrive."""
     loop = asyncio.get_running_loop()
-
-    boot_warning_future = loop.create_future()
-    connected_warning_future = loop.create_future()
+    waiter = LineWaiter()
     button_action_future = loop.create_future()
 
-    def check_output(line: str) -> None:
-        if (
-            not boot_warning_future.done()
-            and "Home Assistant action 'test.boot_action' dropped; no client connected"
-            in line
-        ):
-            boot_warning_future.set_result(True)
-        if (
-            not connected_warning_future.done()
-            and "Home Assistant action 'test.connected_action' dropped; "
-            "client has not subscribed to actions (yet)"
-            in line
-        ):
-            connected_warning_future.set_result(True)
-
     service_calls: list[HomeassistantServiceCall] = []
 
     def on_service_call(service_call: HomeassistantServiceCall) -> None:
@@ -55,16 +39,23 @@ async def test_api_homeassistant_action_no_subscriber(
         ):
             button_action_future.set_result(service_call)
 
-    async with run_compiled(yaml_config, line_callback=check_output):
+    async with run_compiled(yaml_config, line_callback=waiter.callback):
         # The on_boot action fires with no client connected at all.
-        await asyncio.wait_for(boot_warning_future, timeout=10.0)
+        await waiter.wait_for(
+            "Home Assistant action 'test.boot_action' dropped; no client connected",
+            timeout=10.0,
+        )
 
         async with api_client_connected() as client:
             await assert_device_name(client, "test-ha-action-no-subscriber")
 
             # on_client_connected fired at authentication, before this client
             # subscribed to Home Assistant actions.
-            await asyncio.wait_for(connected_warning_future, timeout=5.0)
+            await waiter.wait_for(
+                "Home Assistant action 'test.connected_action' dropped; "
+                "client has not subscribed to actions (yet)",
+                timeout=5.0,
+            )
 
             # After subscribing, actions must be delivered normally (and the
             # dropped ones must not suddenly show up).

tests/integration/test_api_list_entities_backpressure.py

@@ -18,6 +18,7 @@ import asyncio
 from aioesphomeapi import api_pb2
 import pytest
 
+from .log_utils import LineWaiter
 from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -71,15 +72,10 @@ async def test_api_list_entities_backpressure(
     config = yaml_config.replace("# GENERATED_ACTIONS", _generated_actions())
     config = config.replace("SERVER_SNDBUF", str(SERVER_SNDBUF))
 
-    pin_applied = asyncio.Event()
-
-    def _on_log_line(line: str) -> None:
-        if SNDBUF_PIN_LOG in line:
-            pin_applied.set()
-
-    async with run_compiled(config, line_callback=_on_log_line):
+    waiter = LineWaiter()
+    async with run_compiled(config, line_callback=waiter.callback):
         # Fails loudly if the pin never applied
-        await asyncio.wait_for(pin_applied.wait(), 10)
+        await waiter.wait_for(SNDBUF_PIN_LOG, timeout=10)
 
         async with RawApiClient(
             unused_tcp_port, recv_buffer_size=RECV_BUFFER_SIZE

tests/integration/test_crc8_helper.py

@@ -2,13 +2,11 @@
 
 from __future__ import annotations
 
-import asyncio
-
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -18,71 +16,30 @@ async def test_crc8_helper(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test the CRC8 helper function through integration testing."""
-    # Track test completion with asyncio.Event
-    test_complete = asyncio.Event()
+    waiter = LineWaiter()
 
-    # Track test results
-    test_results = {
-        "dallas_maxim": False,
-        "sensirion": False,
-        "pec": False,
-        "parameter_equivalence": False,
-        "edge_cases": False,
-        "component_compatibility": False,
-        "setup_started": False,
-    }
-
-    def on_log_line(line):
-        """Process log lines to track test progress and results."""
-        # Track test start
-        if "CRC8 Helper Function Integration Test Starting" in line:
-            test_results["setup_started"] = True
-
-        # Track test completion
-        elif "CRC8 Integration Test Complete" in line:
-            test_complete.set()
-
-        # Track individual test results
-        elif "ALL TESTS PASSED" in line:
-            if "Dallas/Maxim CRC8" in line:
-                test_results["dallas_maxim"] = True
-            elif "Sensirion CRC8" in line:
-                test_results["sensirion"] = True
-            elif "PEC CRC8" in line:
-                test_results["pec"] = True
-            elif "Parameter equivalence" in line:
-                test_results["parameter_equivalence"] = True
-            elif "Edge cases" in line:
-                test_results["edge_cases"] = True
-            elif "Component compatibility" in line:
-                test_results["component_compatibility"] = True
-
-        # Log failures for debugging
-        elif "TEST FAILED:" in line or "SUBTEST FAILED:" in line:
-            print(f"CRC8 Test Failure: {line}")
+    def passed(name: str) -> bool:
+        return any("ALL TESTS PASSED" in line and name in line for line in waiter.lines)
 
     # Compile and run the test
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         # Verify device info
         await assert_device_name(client, "crc8-helper-test")
 
         # Wait for tests to complete with timeout
-        await wait_or_fail(
-            test_complete, 5.0, "CRC8 integration test timed out after 5 seconds"
-        )
+        await waiter.wait_for("CRC8 Integration Test Complete", timeout=5.0)
 
         # Verify all tests passed
-        assert test_results["setup_started"], "CRC8 test setup never started"
-        assert test_results["dallas_maxim"], "Dallas/Maxim CRC8 test failed"
-        assert test_results["sensirion"], "Sensirion CRC8 test failed"
-        assert test_results["pec"], "PEC CRC8 test failed"
-        assert test_results["parameter_equivalence"], (
-            "Parameter equivalence test failed"
-        )
-        assert test_results["edge_cases"], "Edge cases test failed"
-        assert test_results["component_compatibility"], (
-            "Component compatibility test failed"
-        )
+        assert any(
+            "CRC8 Helper Function Integration Test Starting" in line
+            for line in waiter.lines
+        ), "CRC8 test setup never started"
+        assert passed("Dallas/Maxim CRC8"), "Dallas/Maxim CRC8 test failed"
+        assert passed("Sensirion CRC8"), "Sensirion CRC8 test failed"
+        assert passed("PEC CRC8"), "PEC CRC8 test failed"
+        assert passed("Parameter equivalence"), "Parameter equivalence test failed"
+        assert passed("Edge cases"), "Edge cases test failed"
+        assert passed("Component compatibility"), "Component compatibility test failed"

tests/integration/test_lock_automations.py

@@ -3,13 +3,11 @@
 Tests that on_lock and on_unlock triggers work correctly.
 """
 
-import asyncio
-
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -19,21 +17,10 @@ async def test_lock_automations(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test lock on_lock and on_unlock triggers."""
-    loop = asyncio.get_running_loop()
-
-    # Futures for log line detection
-    on_lock_future: asyncio.Future[bool] = loop.create_future()
-    on_unlock_future: asyncio.Future[bool] = loop.create_future()
-
-    def check_output(line: str) -> None:
-        """Check log output for trigger messages."""
-        if "TRIGGER: on_lock fired" in line and not on_lock_future.done():
-            on_lock_future.set_result(True)
-        elif "TRIGGER: on_unlock fired" in line and not on_unlock_future.done():
-            on_unlock_future.set_result(True)
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=check_output),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         # Import here to avoid import errors when aioesphomeapi is not installed
@@ -46,9 +33,9 @@ async def test_lock_automations(
         # Test 1: Lock - should trigger on_lock
         client.lock_command(key=lock.key, command=LockCommand.LOCK)
 
-        await wait_or_fail(on_lock_future, 5.0, "on_lock trigger did not fire")
+        await waiter.wait_for("TRIGGER: on_lock fired", timeout=5.0)
 
         # Test 2: Unlock - should trigger on_unlock
         client.lock_command(key=lock.key, command=LockCommand.UNLOCK)
 
-        await wait_or_fail(on_unlock_future, 5.0, "on_unlock trigger did not fire")
+        await waiter.wait_for("TRIGGER: on_unlock fired", timeout=5.0)

tests/integration/test_micros_to_millis.py

@@ -2,11 +2,9 @@
 
 from __future__ import annotations
 
-import asyncio
-import re
-
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -20,24 +18,22 @@ async def test_micros_to_millis(
 ) -> None:
     """Test that micros_to_millis matches reference uint64 division."""
 
-    all_passed = asyncio.Event()
-    failures: list[str] = []
+    waiter = LineWaiter()
 
-    def on_log_line(line: str) -> None:
-        clean_line = re.sub(r"\x1b\[[0-9;]*m", "", line)
-        if "ALL_PASSED" in clean_line:
-            all_passed.set()
-        elif "FAILED" in clean_line and "[MTM" in clean_line:
-            failures.append(clean_line)
+    def failures() -> list[str]:
+        return [line for line in waiter.lines if "FAILED" in line and "[MTM" in line]
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await assert_device_name(client, "micros-to-millis-test")
 
         await wait_or_fail(
-            all_passed, 2.0, "micros_to_millis test timed out", failures=failures
+            waiter.wait_for("ALL_PASSED"),
+            2.0,
+            "micros_to_millis test timed out",
+            failures=failures,
         )
 
-        assert not failures, f"micros_to_millis failures: {failures}"
+        assert not failures(), f"micros_to_millis failures: {failures()}"

tests/integration/test_online_image_stalled_download.py

@@ -7,6 +7,7 @@ import struct
 
 import pytest
 
+from .log_utils import LineWaiter
 from .online_image_utils import BMP_IMAGE
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
@@ -72,24 +73,7 @@ async def test_online_image_stalled_download(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Esphome should report an error when a download can never complete."""
-    loop = asyncio.get_running_loop()
-    expected_logs = {
-        "truncated_reason": "Download ended with 1 bytes the decoder could not use",
-        "truncated_error": "download error: truncated",
-        "large_header_reason": "Image does not fit in the 256 byte download buffer",
-        "large_header_error": "download error: large_header",
-        "empty_reason": "Server sent an empty response",
-        "empty_error": "download error: empty",
-    }
-    log_futures = {key: loop.create_future() for key in expected_logs}
-    finished_lines: list[str] = []
-
-    def check_output(line: str) -> None:
-        for key, text in expected_logs.items():
-            if text in line and not log_futures[key].done():
-                log_futures[key].set_result(True)
-        if "download finished" in line:
-            finished_lines.append(line)
+    waiter = LineWaiter()
 
     server = await asyncio.start_server(handle_http, "127.0.0.1", 0)
     http_server_port = server.sockets[0].getsockname()[1]
@@ -97,7 +81,7 @@ async def test_online_image_stalled_download(
 
     async with (
         server,
-        run_compiled(config, line_callback=check_output),
+        run_compiled(config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         _, services = await client.list_entities_services()
@@ -105,17 +89,23 @@ async def test_online_image_stalled_download(
 
         await client.execute_service(actions["fetch_truncated"], {})
         async with asyncio.timeout(2.0):
-            await log_futures["truncated_reason"]
-            await log_futures["truncated_error"]
+            await waiter.wait_for_each(
+                "Download ended with 1 bytes the decoder could not use",
+                "download error: truncated",
+            )
 
         await client.execute_service(actions["fetch_large_header"], {})
         async with asyncio.timeout(2.0):
-            await log_futures["large_header_reason"]
-            await log_futures["large_header_error"]
+            await waiter.wait_for_each(
+                "Image does not fit in the 256 byte download buffer",
+                "download error: large_header",
+            )
 
         await client.execute_service(actions["fetch_empty"], {})
         async with asyncio.timeout(2.0):
-            await log_futures["empty_reason"]
-            await log_futures["empty_error"]
+            await waiter.wait_for_each(
+                "Server sent an empty response", "download error: empty"
+            )
 
+    finished_lines = [line for line in waiter.lines if "download finished" in line]
     assert not finished_lines, f"Unexpected successful download: {finished_lines}"

tests/integration/test_scheduler_internal_id_no_collision.py

@@ -6,11 +6,9 @@ components) are in completely separate matching namespaces, even when
 the underlying uint32_t values are identical and on the same component.
 """
 
-import asyncio
-import re
-
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -23,103 +21,59 @@ async def test_scheduler_internal_id_no_collision(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test that internal and numeric IDs with same value don't collide."""
-    # Test 1: Both types fire independently with same ID
-    internal_timeout_0_fired = asyncio.Event()
-    numeric_timeout_0_fired = asyncio.Event()
+    waiter = LineWaiter()
 
-    # Test 2: Cancelling numeric doesn't cancel internal
-    internal_timeout_1_survived = asyncio.Event()
-    numeric_timeout_1_error = asyncio.Event()
-
-    # Test 3: Cancelling internal doesn't cancel numeric
-    numeric_timeout_2_survived = asyncio.Event()
-    internal_timeout_2_error = asyncio.Event()
-
-    # Test 4: Both interval types fire independently
-    internal_interval_3_done = asyncio.Event()
-    numeric_interval_3_done = asyncio.Event()
-
-    # Test 5: String name doesn't collide with internal ID
-    string_timeout_fired = asyncio.Event()
-    internal_timeout_10_fired = asyncio.Event()
-
-    # Completion
-    all_tests_complete = asyncio.Event()
-
-    def on_log_line(line: str) -> None:
-        clean_line = re.sub(r"\x1b\[[0-9;]*m", "", line)
-
-        if "Internal timeout 0 fired" in clean_line:
-            internal_timeout_0_fired.set()
-        elif "Numeric timeout 0 fired" in clean_line:
-            numeric_timeout_0_fired.set()
-        elif "Internal timeout 1 survived cancel" in clean_line:
-            internal_timeout_1_survived.set()
-        elif "ERROR: Numeric timeout 1 should have been cancelled" in clean_line:
-            numeric_timeout_1_error.set()
-        elif "Numeric timeout 2 survived cancel" in clean_line:
-            numeric_timeout_2_survived.set()
-        elif "ERROR: Internal timeout 2 should have been cancelled" in clean_line:
-            internal_timeout_2_error.set()
-        elif "Internal interval 3 fired twice" in clean_line:
-            internal_interval_3_done.set()
-        elif "Numeric interval 3 fired twice" in clean_line:
-            numeric_interval_3_done.set()
-        elif "String timeout collision_test fired" in clean_line:
-            string_timeout_fired.set()
-        elif "Internal timeout 10 fired" in clean_line:
-            internal_timeout_10_fired.set()
-        elif "All collision tests complete" in clean_line:
-            all_tests_complete.set()
+    def seen(text: str) -> bool:
+        return any(text in line for line in waiter.lines)
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await assert_device_name(client, "scheduler-internal-id-test")
 
         await wait_or_fail(
-            all_tests_complete,
+            waiter.wait_for("All collision tests complete"),
             5.0,
             "Not all collision tests completed within 5 seconds",
         )
 
         # Test 1: Both timeout types with same ID 0 must fire
-        assert internal_timeout_0_fired.is_set(), (
+        assert seen("Internal timeout 0 fired"), (
             "Internal timeout with ID 0 should have fired"
         )
-        assert numeric_timeout_0_fired.is_set(), (
+        assert seen("Numeric timeout 0 fired"), (
             "Numeric timeout with ID 0 should have fired"
         )
 
         # Test 2: Cancelling numeric ID must NOT cancel internal ID
-        assert internal_timeout_1_survived.is_set(), (
+        assert seen("Internal timeout 1 survived cancel"), (
             "Internal timeout 1 should survive cancellation of numeric timeout 1"
         )
-        assert not numeric_timeout_1_error.is_set(), (
+        assert not seen("ERROR: Numeric timeout 1 should have been cancelled"), (
             "Numeric timeout 1 should have been cancelled"
         )
 
         # Test 3: Cancelling internal ID must NOT cancel numeric ID
-        assert numeric_timeout_2_survived.is_set(), (
+        assert seen("Numeric timeout 2 survived cancel"), (
             "Numeric timeout 2 should survive cancellation of internal timeout 2"
         )
-        assert not internal_timeout_2_error.is_set(), (
+        assert not seen("ERROR: Internal timeout 2 should have been cancelled"), (
             "Internal timeout 2 should have been cancelled"
         )
 
         # Test 4: Both interval types with same ID must fire independently
-        assert internal_interval_3_done.is_set(), (
+        assert seen("Internal interval 3 fired twice"), (
             "Internal interval 3 should have fired at least twice"
         )
-        assert numeric_interval_3_done.is_set(), (
+        assert seen("Numeric interval 3 fired twice"), (
             "Numeric interval 3 should have fired at least twice"
         )
 
         # Test 5: String name and internal ID don't collide
-        assert string_timeout_fired.is_set(), (
+        assert seen("String timeout collision_test fired"), (
             "String timeout 'collision_test' should have fired"
         )
-        assert internal_timeout_10_fired.is_set(), (
+        assert seen("Internal timeout 10 fired"), (
             "Internal timeout 10 should have fired alongside string timeout"
         )

tests/integration/test_scheduler_interval_zero_coerced.py

@@ -13,13 +13,11 @@ and the device stays responsive to API calls.
 
 from __future__ import annotations
 
-import asyncio
-
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -29,18 +27,10 @@ async def test_scheduler_interval_zero_coerced(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """interval=0ms must be coerced to 1ms and not starve the main loop."""
-    loop = asyncio.get_running_loop()
-    reached_50: asyncio.Future[None] = loop.create_future()
-    coerce_warning: asyncio.Future[None] = loop.create_future()
-
-    def on_log_line(line: str) -> None:
-        if "ZERO_INTERVAL_50_FIRES_REACHED" in line and not reached_50.done():
-            reached_50.set_result(None)
-        if "would spin main loop" in line and not coerce_warning.done():
-            coerce_warning.set_result(None)
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         # The API-client connection itself is evidence that the main loop
@@ -49,19 +39,10 @@ async def test_scheduler_interval_zero_coerced(
         await assert_device_name(client, "sched-interval-zero")
 
         # Coerce warning must fire at registration
-        await wait_or_fail(
-            coerce_warning,
-            5.0,
-            "Expected coerce warning 'would spin main loop' not seen",
-        )
+        await waiter.wait_for("would spin main loop", timeout=5.0)
 
         # The coerced 1ms interval should fire 50 times quickly — this
         # confirms the callback actually runs (not just registered) and the
         # scheduler yields back to the main loop each time.
-        await wait_or_fail(
-            reached_50,
-            5.0,
-            "Coerced interval=0→1ms did not reach 50 fires within 5s, "
-            "which would indicate either the coercion failed or the "
-            "main loop is still being starved.",
-        )
+        # A timeout means the coercion failed or the main loop is still starved.
+        await waiter.wait_for("ZERO_INTERVAL_50_FIRES_REACHED", timeout=5.0)

tests/integration/test_scheduler_self_keyed.py

@@ -7,11 +7,9 @@ self-keyed and component-keyed namespaces are independent, and re-registering
 the same key replaces the existing timer.
 """
 
-import asyncio
-import re
-
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -24,74 +22,52 @@ async def test_scheduler_self_keyed(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test self-keyed scheduler API."""
-    self_a_fired = asyncio.Event()
-    self_b_error = asyncio.Event()
-    self_c_fired = asyncio.Event()
-    self_d_fired = asyncio.Event()
-    self_shared_fired = asyncio.Event()
-    component_7777_fired = asyncio.Event()
-    self_interval_done = asyncio.Event()
-    self_f_first_error = asyncio.Event()
-    self_f_replacement_fired = asyncio.Event()
-    all_tests_complete = asyncio.Event()
+    waiter = LineWaiter()
 
-    def on_log_line(line: str) -> None:
-        clean_line = re.sub(r"\x1b\[[0-9;]*m", "", line)
-
-        if "Self timeout A fired" in clean_line:
-            self_a_fired.set()
-        elif "ERROR: Self timeout B" in clean_line:
-            self_b_error.set()
-        elif "Self timeout C fired" in clean_line:
-            self_c_fired.set()
-        elif "Self timeout D fired" in clean_line:
-            self_d_fired.set()
-        elif "Self timeout shared fired" in clean_line:
-            self_shared_fired.set()
-        elif "Component timeout 7777 fired" in clean_line:
-            component_7777_fired.set()
-        elif "Self interval E fired twice" in clean_line:
-            self_interval_done.set()
-        elif "ERROR: Self timeout F first registration" in clean_line:
-            self_f_first_error.set()
-        elif "Self timeout F replacement fired" in clean_line:
-            self_f_replacement_fired.set()
-        elif "All self-keyed tests complete" in clean_line:
-            all_tests_complete.set()
+    def seen(text: str) -> bool:
+        return any(text in line for line in waiter.lines)
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await assert_device_name(client, "scheduler-self-keyed-test")
 
         await wait_or_fail(
-            all_tests_complete,
+            waiter.wait_for("All self-keyed tests complete"),
             5.0,
             "Not all self-keyed tests completed within 5 seconds",
         )
 
         # Test 1: self-keyed timeout fires
-        assert self_a_fired.is_set(), "Self timeout A should have fired"
+        assert seen("Self timeout A fired"), "Self timeout A should have fired"
 
         # Test 2: cancel_timeout(self) actually cancels
-        assert not self_b_error.is_set(), "Self timeout B should have been cancelled"
+        assert not seen("ERROR: Self timeout B"), (
+            "Self timeout B should have been cancelled"
+        )
 
         # Test 3: distinct self keys don't collide
-        assert self_c_fired.is_set(), "Self timeout C should have fired"
-        assert self_d_fired.is_set(), "Self timeout D should have fired"
+        assert seen("Self timeout C fired"), "Self timeout C should have fired"
+        assert seen("Self timeout D fired"), "Self timeout D should have fired"
 
         # Test 4: self-keyed and component-keyed namespaces are independent
-        assert self_shared_fired.is_set(), "Self timeout shared should have fired"
-        assert component_7777_fired.is_set(), "Component timeout 7777 should have fired"
+        assert seen("Self timeout shared fired"), (
+            "Self timeout shared should have fired"
+        )
+        assert seen("Component timeout 7777 fired"), (
+            "Component timeout 7777 should have fired"
+        )
 
         # Test 5: self-keyed interval fires repeatedly and cancels cleanly
-        assert self_interval_done.is_set(), "Self interval E should have fired twice"
+        assert seen("Self interval E fired twice"), (
+            "Self interval E should have fired twice"
+        )
 
         # Test 6: re-registering same self-key replaces the previous timer
-        assert not self_f_first_error.is_set(), (
+        assert not seen("ERROR: Self timeout F first registration"), (
             "Self timeout F first registration should have been replaced"
         )
-        assert self_f_replacement_fired.is_set(), (
+        assert seen("Self timeout F replacement fired"), (
             "Self timeout F replacement should have fired"
         )

tests/integration/test_script_on_stop_and_choose.py

@@ -7,6 +7,7 @@ import asyncio
 from aioesphomeapi import UserService
 import pytest
 
+from .log_utils import LineWaiter
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -17,25 +18,13 @@ async def test_script_on_stop_and_choose(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test on_stop, self-stopping scripts and choose."""
-    lines: list[str] = []
-    waiters: list[tuple[str, asyncio.Future[None]]] = []
-
-    def check_output(line: str) -> None:
-        lines.append(line)
-        for text, future in waiters:
-            if text in line and not future.done():
-                future.set_result(None)
-
-    def expect(text: str) -> asyncio.Future[None]:
-        future: asyncio.Future[None] = asyncio.get_running_loop().create_future()
-        waiters.append((text, future))
-        return future
+    waiter = LineWaiter()
 
     def count(text: str) -> int:
-        return sum(text in line for line in lines)
+        return sum(text in line for line in waiter.lines)
 
     async with (
-        run_compiled(yaml_config, line_callback=check_output),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         _, services = await client.list_entities_services()
@@ -49,9 +38,8 @@ async def test_script_on_stop_and_choose(
             (500, "large"),
             (5000, "default"),
         ):
-            done = expect(f"CHOOSE done {value}")
             await client.execute_service(service["run_choose"], {"value": value})
-            await asyncio.wait_for(done, timeout=5.0)
+            await waiter.wait_for(f"CHOOSE done {value}", timeout=5.0)
             assert count(f"CHOOSE {label} {value}") == 1
         assert count("CHOOSE small") == 1
         assert count("CHOOSE medium") == 1
@@ -59,44 +47,36 @@ async def test_script_on_stop_and_choose(
         assert count("CHOOSE default") == 1
 
         # A bare script.stop inside a script stops that script and fires on_stop
-        stopped = expect("SELF on_stop")
         await client.execute_service(service["run_self_stop"], {})
-        await asyncio.wait_for(stopped, timeout=5.0)
+        await waiter.wait_for("SELF on_stop", timeout=5.0)
 
         # Stopping a running script from outside fires on_stop
-        stopped = expect("LONG on_stop")
         await client.execute_service(service["run_external_stop"], {})
-        await asyncio.wait_for(stopped, timeout=5.0)
+        await waiter.wait_for("LONG on_stop", timeout=5.0)
 
         # Stopping a script that is not running does not fire on_stop
-        idle_done = expect("IDLE stop done")
         await client.execute_service(service["run_idle_stop"], {})
-        await asyncio.wait_for(idle_done, timeout=5.0)
+        await waiter.wait_for("IDLE stop done", timeout=5.0)
 
         # Restarting a running script stops the old run early
-        finished = expect("RESTART finished")
         await client.execute_service(service["run_restart"], {})
-        await asyncio.wait_for(finished, timeout=5.0)
+        await waiter.wait_for("RESTART finished", timeout=5.0)
 
         # A script that finishes normally does not fire on_stop
-        finished = expect("NORMAL finished")
         await client.execute_service(service["run_normal"], {})
-        await asyncio.wait_for(finished, timeout=5.0)
+        await waiter.wait_for("NORMAL finished", timeout=5.0)
 
         # A restart script whose on_stop executes it again ends up with one run
-        finished = expect("REENTER finished")
         await client.execute_service(service["run_reenter"], {})
-        await asyncio.wait_for(finished, timeout=5.0)
+        await waiter.wait_for("REENTER finished", timeout=5.0)
 
         # Stopping several parallel runs fires on_stop once
-        done = expect("PARALLEL stop done")
         await client.execute_service(service["run_parallel_stop"], {})
-        await asyncio.wait_for(done, timeout=5.0)
+        await waiter.wait_for("PARALLEL stop done", timeout=5.0)
 
         # Stopping a queued script fires on_stop once and drops the queued run
-        done = expect("QUEUED stop done")
         await client.execute_service(service["run_queued_stop"], {})
-        await asyncio.wait_for(done, timeout=5.0)
+        await waiter.wait_for("QUEUED stop done", timeout=5.0)
         await asyncio.sleep(0.2)
 
         assert count("SELF not reached") == 0

tests/integration/test_socket_ipv4_resolve.py

@@ -2,10 +2,9 @@
 
 from __future__ import annotations
 
-import asyncio
-
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -25,24 +24,29 @@ async def test_socket_ipv4_resolve(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Exercise Ipv4Resolve literals, forget, failure, and getaddrinfo on host."""
-    test_complete = asyncio.Event()
-    results: dict[str, bool] = {}
+    waiter = LineWaiter()
 
-    def on_log_line(line: str) -> None:
-        if "IPv4 resolve test complete" in line:
-            test_complete.set()
-            return
-        for check in CHECKS:
-            if f"{check}:" in line:
-                results[check] = "PASSED" in line
+    def passed(check: str) -> bool:
+        # The last report of a check wins
+        reports = [
+            line
+            for line in waiter.lines
+            if f"{check}:" in line and "IPv4 resolve test complete" not in line
+        ]
+        return bool(reports) and "PASSED" in reports[-1]
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await assert_device_name(client, "socket-ipv4-resolve-test")
 
-        await wait_or_fail(test_complete, 10.0, "IPv4 resolve test timed out")
+        # The outer bound carries the message; the inner one only has to outlast it
+        await wait_or_fail(
+            waiter.wait_for("IPv4 resolve test complete", timeout=20.0),
+            10.0,
+            "IPv4 resolve test timed out",
+        )
 
     for check in CHECKS:
-        assert results.get(check), f"{check} check failed or never ran"
+        assert passed(check), f"{check} check failed or never ran"

tests/integration/test_socket_wake_gate_tcp.py

@@ -12,14 +12,13 @@ loop_interval_ expired. With the fix, the gate opens immediately.
 
 from __future__ import annotations
 
-import asyncio
 import time
 
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name, require_service
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -30,25 +29,15 @@ async def test_socket_wake_gate_tcp(
 ) -> None:
     """TCP socket receive must open the component-phase gate immediately,
     even with loop_interval_ raised to 2000ms."""
-    loop = asyncio.get_running_loop()
-    boot_done: asyncio.Future[None] = loop.create_future()
-    pong: asyncio.Future[None] = loop.create_future()
-
-    def on_log_line(line: str) -> None:
-        if "BOOT_DONE" in line and not boot_done.done():
-            boot_done.set_result(None)
-        if "PONG" in line and not pong.done():
-            pong.set_result(None)
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await assert_device_name(client, "socket-wake-gate-tcp")
 
-        await wait_or_fail(
-            boot_done, 15.0, "BOOT_DONE never appeared — device did not complete boot"
-        )
+        await waiter.wait_for("BOOT_DONE", timeout=15.0)
 
         _, services = await client.list_entities_services()
         ping_service = require_service(services, "ping")
@@ -59,7 +48,7 @@ async def test_socket_wake_gate_tcp(
         t_send = time.monotonic()
         await client.execute_service(ping_service, {})
 
-        await wait_or_fail(pong, 5.0, "PONG never appeared — service did not execute")
+        await waiter.wait_for("PONG", timeout=5.0)
 
         elapsed_ms = (time.monotonic() - t_send) * 1000
         # Without the fix the gate stays closed for up to loop_interval_=2000ms.

tests/integration/test_socket_wake_gate_udp.py

@@ -2,14 +2,13 @@
 
 from __future__ import annotations
 
-import asyncio
 import socket
 import time
 
 import pytest
 
+from .log_utils import LineWaiter
 from .types import RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -26,19 +25,11 @@ async def test_socket_wake_gate_udp(
     listen_port = hold_sock.getsockname()[1]
     config = yaml_config.replace("UDP_LISTEN_PORT_PLACEHOLDER", str(listen_port))
 
-    loop = asyncio.get_running_loop()
-    boot_done: asyncio.Future[None] = loop.create_future()
-    pong: asyncio.Future[None] = loop.create_future()
-
-    def on_log_line(line: str) -> None:
-        if "BOOT_DONE" in line and not boot_done.done():
-            boot_done.set_result(None)
-        if "PONG" in line and not pong.done():
-            pong.set_result(None)
+    waiter = LineWaiter()
 
     try:
-        async with run_compiled(config, line_callback=on_log_line):
-            await wait_or_fail(boot_done, 15.0, "BOOT_DONE never appeared")
+        async with run_compiled(config, line_callback=waiter.callback):
+            await waiter.wait_for("BOOT_DONE", timeout=15.0)
 
             # Datagrams go to the more specific bind, so release the hold before sending
             hold_sock.close()
@@ -48,7 +39,7 @@ async def test_socket_wake_gate_udp(
             with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as send_sock:
                 send_sock.sendto(b"PING", ("127.0.0.1", listen_port))
 
-            await wait_or_fail(pong, 5.0, "PONG never appeared")
+            await waiter.wait_for("PONG", timeout=5.0)
 
             elapsed_ms = (time.monotonic() - t_send) * 1000
             # Without the wake the receive waits for loop_interval_; 500ms leaves CI headroom

tests/integration/test_status_momentary.py

@@ -7,6 +7,7 @@ import asyncio
 from aioesphomeapi import BinarySensorState, EntityState
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import InitialStateHelper
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
@@ -18,20 +19,10 @@ async def test_status_momentary(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     loop = asyncio.get_running_loop()
-    warning_logged = loop.create_future()
-    error_logged = loop.create_future()
-
-    def on_log_line(line: str) -> None:
-        if (
-            "set Warning flag: probe warning reason" in line
-            and not warning_logged.done()
-        ):
-            warning_logged.set_result(True)
-        if "set Error flag: probe error reason" in line and not error_logged.done():
-            error_logged.set_result(True)
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, services = await client.list_entities_services()
@@ -82,6 +73,6 @@ async def test_status_momentary(
 
         # A message replaces "unspecified" in the log line
         await client.execute_service(svc["momentary_warning_message"], {})
-        await asyncio.wait_for(warning_logged, timeout=5)
+        await waiter.wait_for("set Warning flag: probe warning reason", timeout=5)
         await client.execute_service(svc["momentary_error_message"], {})
-        await asyncio.wait_for(error_logged, timeout=5)
+        await waiter.wait_for("set Error flag: probe error reason", timeout=5)

tests/integration/test_template_climate_custom_modes.py

@@ -6,13 +6,12 @@ the enum-based mode/preset fields, but for the custom string variants.
 
 from __future__ import annotations
 
-import asyncio
-
 import aioesphomeapi
 from aioesphomeapi import ButtonInfo, ClimateInfo
 import pytest
 
 from .host_prefs import clear_host_prefs
+from .log_utils import LineWaiter
 from .state_utils import (
     InitialStateHelper,
     require_entity,
@@ -33,14 +32,10 @@ async def test_template_climate_custom_modes(
     """Custom fan mode/preset: traits, on_control forwarding, and publish precedence."""
     clear_host_prefs(DEVICE_NAME)
 
-    log_lines: list[str] = []
-
-    def on_log_line(line: str) -> None:
-        if "on_control " in line:
-            log_lines.append(line)
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
 
@@ -87,9 +82,11 @@ async def test_template_climate_custom_modes(
         state = await wait_for_climate_state()
         assert state.custom_preset == "power_save"
 
-        await asyncio.sleep(0.2)
-        assert any("on_control custom_fan_mode=turbo" in line for line in log_lines)
-        assert any("on_control custom_preset=power_save" in line for line in log_lines)
+        await waiter.wait_for_each(
+            "on_control custom_fan_mode=turbo",
+            "on_control custom_preset=power_save",
+            timeout=5.0,
+        )
 
         # The device's report is authoritative and overrides what was commanded above.
         client.button_command(report_button.key)

tests/integration/test_template_climate_nonoptimistic.py

@@ -21,6 +21,7 @@ from aioesphomeapi import (
 import pytest
 
 from .host_prefs import clear_host_prefs
+from .log_utils import LineWaiter
 from .state_utils import (
     InitialStateHelper,
     require_entity,
@@ -41,15 +42,11 @@ async def test_template_climate_nonoptimistic(
     """Nonoptimistic: a command doesn't change state until explicitly published."""
     clear_host_prefs(DEVICE_NAME)
 
-    log_lines: list[str] = []
+    waiter = LineWaiter()
     state_updates: list[aioesphomeapi.ClimateState] = []
 
-    def on_log_line(line: str) -> None:
-        if "on_control " in line:
-            log_lines.append(line)
-
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
 
@@ -84,13 +81,14 @@ async def test_template_climate_nonoptimistic(
             preset=ClimatePreset.AWAY,
         )
         await asyncio.sleep(0.3)
-        assert any(
-            "on_control mode=3" in line for line in log_lines
-        )  # CLIMATE_MODE_HEAT
-        assert any("on_control target_temperature=22.5" in line for line in log_lines)
-        assert any("on_control fan_mode=" in line for line in log_lines)
-        assert any("on_control swing_mode=" in line for line in log_lines)
-        assert any("on_control preset=" in line for line in log_lines)
+        await waiter.wait_for_each(
+            "on_control mode=3",  # CLIMATE_MODE_HEAT
+            "on_control target_temperature=22.5",
+            "on_control fan_mode=",
+            "on_control swing_mode=",
+            "on_control preset=",
+            timeout=5.0,
+        )
         assert not state_updates, (
             "optimistic: false must not publish a state until climate.template.publish reports it"
         )

tests/integration/test_template_climate_on_control_ordering.py

@@ -7,13 +7,12 @@ proving the firing order is on_control, then control(), then on_state.
 
 from __future__ import annotations
 
-import asyncio
-
 import aioesphomeapi
 from aioesphomeapi import ClimateInfo, ClimateMode
 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 .types import APIClientConnectedFactory, RunCompiledFunction
 
@@ -29,14 +28,10 @@ async def test_template_climate_on_control_ordering(
     """on_control sees the requested value while the entity's own state is still the old one."""
     clear_host_prefs(DEVICE_NAME)
 
-    log_lines: list[str] = []
-
-    def on_log_line(line: str) -> None:
-        if "on_control " in line or "on_state " in line:
-            log_lines.append(line)
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
 
@@ -60,19 +55,15 @@ async def test_template_climate_on_control_ordering(
         state = await wait_for_climate_state()
         assert state.mode == ClimateMode.HEAT
 
-        await asyncio.sleep(0.2)
-
         # on_control saw the new requested mode (3 == CLIMATE_MODE_HEAT) while the entity's own
         # state was still the old one (0 == CLIMATE_MODE_OFF) -- proving it fired before control().
-        assert any(
-            "on_control requested_mode=3 current_mode_before_apply=0" in line
-            for line in log_lines
+        await waiter.wait_for(
+            "on_control requested_mode=3 current_mode_before_apply=0", timeout=5.0
         )
         # on_state fired afterward, reporting the now-applied mode.
-        assert any("on_state mode=3" in line for line in log_lines)
+        await waiter.wait_for("on_state mode=3", timeout=5.0)
 
-        control_index = next(
-            i for i, line in enumerate(log_lines) if "on_control " in line
-        )
-        state_index = next(i for i, line in enumerate(log_lines) if "on_state " in line)
+        lines = waiter.lines
+        control_index = next(i for i, line in enumerate(lines) if "on_control " in line)
+        state_index = next(i for i, line in enumerate(lines) if "on_state " in line)
         assert control_index < state_index, "on_control must fire before on_state"

tests/integration/test_template_climate_set_actions.py

@@ -6,8 +6,6 @@ device's actual state back, so the actions are the only thing that reacts to a c
 
 from __future__ import annotations
 
-import asyncio
-
 import aioesphomeapi
 from aioesphomeapi import (
     ButtonInfo,
@@ -20,6 +18,7 @@ from aioesphomeapi import (
 import pytest
 
 from .host_prefs import clear_host_prefs
+from .log_utils import LineWaiter
 from .state_utils import (
     InitialStateHelper,
     require_entity,
@@ -40,17 +39,13 @@ async def test_template_climate_set_actions(
     """Every set_*_action fires with the requested value; state waits for a publish."""
     clear_host_prefs(DEVICE_NAME)
 
-    log_lines: list[str] = []
-
-    def on_log_line(line: str) -> None:
-        if "_action " in line or "Unsupported" in line:
-            log_lines.append(line)
+    waiter = LineWaiter()
 
     def logged(fragment: str) -> bool:
-        return any(fragment in line for line in log_lines)
+        return any(fragment in line for line in waiter.lines)
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -82,10 +77,7 @@ async def test_template_climate_set_actions(
         client.climate_command(test_climate.key, preset=ClimatePreset.ECO)
         client.climate_command(test_climate.key, custom_preset="eco_plus")
 
-        for _ in range(50):
-            await asyncio.sleep(0.1)
-            if logged("set_custom_preset_action eco_plus"):
-                break
+        await waiter.wait_for("set_custom_preset_action eco_plus", timeout=5.0)
 
         assert logged("set_mode_action 3")  # CLIMATE_MODE_HEAT
         assert logged("set_target_temperature_low_action 18.0")
@@ -107,8 +99,4 @@ async def test_template_climate_set_actions(
 
         # A publish naming a mode outside supported_modes warns instead of publishing it.
         client.button_command(unsupported_button.key)
-        for _ in range(50):
-            await asyncio.sleep(0.1)
-            if logged("Unsupported mode"):
-                break
-        assert logged("Unsupported mode")
+        await waiter.wait_for("Unsupported mode", timeout=5.0)

tests/integration/test_template_climate_two_point_temperature.py

@@ -8,13 +8,12 @@ changed via a physical remote).
 
 from __future__ import annotations
 
-import asyncio
-
 import aioesphomeapi
 from aioesphomeapi import ButtonInfo, ClimateInfo, ClimateMode
 import pytest
 
 from .host_prefs import clear_host_prefs
+from .log_utils import LineWaiter
 from .state_utils import (
     InitialStateHelper,
     require_entity,
@@ -35,14 +34,10 @@ async def test_template_climate_two_point_temperature(
     """Two-point target temperature + humidity: booleans, on_control, and publish precedence."""
     clear_host_prefs(DEVICE_NAME)
 
-    log_lines: list[str] = []
-
-    def on_log_line(line: str) -> None:
-        if "on_control " in line:
-            log_lines.append(line)
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=on_log_line),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
 
@@ -93,19 +88,16 @@ async def test_template_climate_two_point_temperature(
         state = await wait_for_climate_state()
         assert state.target_temperature_low == pytest.approx(19.0, abs=0.1)
         assert state.target_temperature_high == pytest.approx(25.0, abs=0.1)
-        await asyncio.sleep(0.2)
-        assert any(
-            "on_control target_temperature_low=19.0" in line for line in log_lines
-        )
-        assert any(
-            "on_control target_temperature_high=25.0" in line for line in log_lines
+        await waiter.wait_for_each(
+            "on_control target_temperature_low=19.0",
+            "on_control target_temperature_high=25.0",
+            timeout=5.0,
         )
 
         client.climate_command(test_climate.key, target_humidity=45.0)
         state = await wait_for_climate_state()
         assert state.target_humidity == pytest.approx(45.0, abs=0.1)
-        await asyncio.sleep(0.2)
-        assert any("on_control target_humidity=45.0" in line for line in log_lines)
+        await waiter.wait_for("on_control target_humidity=45.0", timeout=5.0)
 
         # The device's next report is authoritative and overrides whatever was optimistically
         # applied above -- this is the whole point of climate.template.publish: a device that owns

tests/integration/test_template_text_save.py

@@ -17,6 +17,7 @@ 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 .types import CompileFunction, ConfigWriter
 
@@ -44,15 +45,11 @@ async def test_template_text_save(
     port_socket.close()
 
     # --- First run: set a value and verify no spurious warnings ---
-    warning_lines: list[str] = []
-
-    def capture_warnings(line: str) -> None:
-        if "too long to save" in line.lower():
-            warning_lines.append(line)
+    waiter = LineWaiter()
 
     async with (
         run_binary_and_wait_for_port(
-            binary_path, "127.0.0.1", port, line_callback=capture_warnings
+            binary_path, "127.0.0.1", port, line_callback=waiter.callback
         ),
         wait_and_connect_api_client(port=port) as client,
     ):
@@ -98,6 +95,9 @@ async def test_template_text_save(
         await asyncio.sleep(0.5)
 
     # No warnings should have appeared
+    warning_lines = [
+        line for line in waiter.lines if "too long to save" in line.lower()
+    ]
     assert warning_lines == [], (
         f"Unexpected 'too long to save' warning(s): {warning_lines}"
     )

tests/integration/test_uart_mock_dfrobot_sen0609.py

@@ -11,9 +11,9 @@ import asyncio
 from aioesphomeapi import BinarySensorInfo, EntityState, NumberInfo, TextSensorInfo
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -23,13 +23,7 @@ async def test_uart_mock_dfrobot_sen0609(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Boot reads the module's settings; a number change is written and saved."""
-    loop = asyncio.get_running_loop()
-    saved = loop.create_future()
-
-    def line_callback(line: str) -> None:
-        if "Saved range 0.60 8.00" in line and not saved.done():
-            saved.set_result(True)
-
+    waiter = LineWaiter()
     states: dict[int, EntityState] = {}
 
     async def wait_for(condition, what: str) -> None:
@@ -41,7 +35,7 @@ async def test_uart_mock_dfrobot_sen0609(
             pytest.fail(f"Timeout waiting for {what}; states: {states}")
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -64,7 +58,7 @@ async def test_uart_mock_dfrobot_sen0609(
         assert state_of(min_range) == pytest.approx(0.6)
 
         client.number_command(max_range.key, 8.0)
-        await wait_or_fail(saved, 10.0, "The new range was not written and saved")
+        await waiter.wait_for("Saved range 0.60 8.00", timeout=10.0)
         await wait_for(
             lambda: state_of(max_range) == pytest.approx(8.0), "new maximum range"
         )

tests/integration/test_uart_mock_ld2410.py

@@ -18,11 +18,10 @@ test_uart_mock_ld2410_engineering (engineering mode):
 
 from __future__ import annotations
 
-import asyncio
-
 from aioesphomeapi import ButtonInfo
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -35,20 +34,7 @@ async def test_uart_mock_ld2410(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test LD2410 data parsing with happy path, garbage, overflow, and recovery."""
-    loop = asyncio.get_running_loop()
-
-    # Track overflow warning in logs
-    overflow_seen = loop.create_future()
-
-    # Track TX data logged by the mock for assertions
-    tx_log_lines: list[str] = []
-
-    def line_callback(line: str) -> None:
-        if "Max command length exceeded" in line and not overflow_seen.done():
-            overflow_seen.set_result(True)
-        # Capture all TX log lines from uart_mock
-        if "uart_mock" in line and "TX " in line:
-            tx_log_lines.append(line)
+    waiter = LineWaiter()
 
     collector = SensorStateCollector(
         sensor_names=[
@@ -78,7 +64,7 @@ async def test_uart_mock_ld2410(
     )
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -116,13 +102,16 @@ async def test_uart_mock_ld2410(
         )
 
         # Verify overflow warning was logged
-        assert overflow_seen.done(), (
+        assert any("Max command length exceeded" in line for line in waiter.lines), (
             "Expected 'Max command length exceeded' warning in logs"
         )
 
         # Verify LD2410 sent setup commands (TX logging)
         # LD2410 sends 7 commands during setup: FF (config on), A0 (version),
         # A5 (MAC), AB (distance res), AE (light), 61 (params), FE (config off)
+        tx_log_lines = [
+            line for line in waiter.lines if "uart_mock" in line and "TX " in line
+        ]
         assert len(tx_log_lines) > 0, "Expected TX log lines from uart_mock"
         tx_data = " ".join(tx_log_lines)
         assert "FD:FC:FB:FA" in tx_data, (

tests/integration/test_uart_mock_ld2410_query_ack.py

@@ -20,6 +20,7 @@ from aioesphomeapi import (
 )
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import InitialStateHelper, build_key_to_entity_mapping, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -55,16 +56,12 @@ async def test_uart_mock_ld2410_query_ack(
 ) -> None:
     """Test a data frame, then a query ACK with the footer bytes in its values."""
     loop = asyncio.get_running_loop()
-    ack_errors: list[str] = []
-    waited_for_frame_end = loop.create_future()
+    waiter = LineWaiter()
 
-    def line_callback(line: str) -> None:
-        if "[ld2410" not in line:
-            return
-        if "Invalid " in line:
-            ack_errors.append(line)
-        if "Ack Data incomplete" in line and not waited_for_frame_end.done():
-            waited_for_frame_end.set_result(True)
+    def ack_errors() -> list[str]:
+        return [
+            line for line in waiter.lines if "[ld2410" in line and "Invalid " in line
+        ]
 
     number_states: dict[str, list[float]] = {name: [] for name in EXPECTED}
     sensor_states: dict[str, list[float]] = {name: [] for name in SENSOR_EXPECTED}
@@ -93,7 +90,7 @@ async def test_uart_mock_ld2410_query_ack(
             final_received.set_result(True)
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -145,10 +142,7 @@ async def test_uart_mock_ld2410_query_ack(
         # Each number is published once per ACK: the seed value, then the value under test
         for name, (seed, value) in EXPECTED.items():
             assert number_states[name] == pytest.approx([seed, value]), name
-        # The log line comes over stdout, which can trail the API states
-        await wait_or_fail(
-            waited_for_frame_end,
-            2.0,
-            "Expected the parser to wait past the footer bytes in the values",
-        )
-        assert not ack_errors, f"Unexpected ACK errors: {ack_errors}"
+        # The parser waits past the footer bytes in the values; the log line
+        # comes over stdout, which can trail the API states
+        await waiter.wait_for("[ld2410", "Ack Data incomplete", timeout=2.0)
+        assert not ack_errors(), f"Unexpected ACK errors: {ack_errors()}"

tests/integration/test_uart_mock_ld2412.py

@@ -42,6 +42,7 @@ from aioesphomeapi import (
 )
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import (
     InitialStateHelper,
     SensorStateCollector,
@@ -139,20 +140,7 @@ async def test_uart_mock_ld2412(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test LD2412 data parsing with happy path, garbage, overflow, and recovery."""
-    loop = asyncio.get_running_loop()
-
-    # Track overflow warning in logs
-    overflow_seen = loop.create_future()
-
-    # Track TX data logged by the mock for assertions
-    tx_log_lines: list[str] = []
-
-    def line_callback(line: str) -> None:
-        if "Max command length exceeded" in line and not overflow_seen.done():
-            overflow_seen.set_result(True)
-        # Capture all TX log lines from uart_mock
-        if _is_tx(line):
-            tx_log_lines.append(line)
+    waiter = LineWaiter()
 
     collector = SensorStateCollector(
         sensor_names=[
@@ -181,7 +169,7 @@ async def test_uart_mock_ld2412(
     )
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -213,11 +201,12 @@ async def test_uart_mock_ld2412(
         )
 
         # Verify overflow warning was logged
-        assert overflow_seen.done(), (
+        assert any("Max command length exceeded" in line for line in waiter.lines), (
             "Expected 'Max command length exceeded' warning in logs"
         )
 
         # Verify LD2412 sent setup commands (TX logging)
+        tx_log_lines = [line for line in waiter.lines if _is_tx(line)]
         assert len(tx_log_lines) > 0, "Expected TX log lines from uart_mock"
         tx_data = " ".join(tx_log_lines)
         assert "FD:FC:FB:FA" in tx_data, (
@@ -337,17 +326,7 @@ async def test_uart_mock_ld2412_engineering_truncated(
     (e.g. frame footer bytes 0xF8=248 as gate energy).
     """
 
-    loop = asyncio.get_running_loop()
-
-    # Track the truncated frame warning
-    truncated_warning_seen = loop.create_future()
-
-    def line_callback(line: str) -> None:
-        if (
-            "Engineering mode packet too short" in line
-            and not truncated_warning_seen.done()
-        ):
-            truncated_warning_seen.set_result(True)
+    waiter = LineWaiter()
 
     collector = SensorStateCollector(
         sensor_names=[
@@ -377,7 +356,7 @@ async def test_uart_mock_ld2412_engineering_truncated(
     )
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -405,9 +384,9 @@ async def test_uart_mock_ld2412_engineering_truncated(
         )
 
         # Verify the truncated frame warning was logged
-        assert truncated_warning_seen.done(), (
-            "Expected 'Engineering mode packet too short' warning in logs"
-        )
+        assert any(
+            "Engineering mode packet too short" in line for line in waiter.lines
+        ), "Expected 'Engineering mode packet too short' warning in logs"
 
         # Phase 3 recovery: gate_0_move=50, light=42
         assert pytest.approx(50.0) in collector.sensor_states["gate_0_move_energy"]
@@ -516,34 +495,25 @@ async def test_uart_mock_ld2412_truncated_answer(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test that a threshold answer merged with the next frame is dropped and the parser recovers."""
-    loop = asyncio.get_running_loop()
-
-    dropped = loop.create_future()
-    warnings: list[str] = []
-
-    def line_callback(line: str) -> None:
-        if "[W][ld2412" in line:
-            warnings.append(line)
-        if "Dropping gate threshold answer" in line and not dropped.done():
-            dropped.set_result(True)
+    waiter = LineWaiter()
 
     # Gate bytes of the truncated answer
     merged_values = set(range(0x30, 0x3C))
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
         numbers = _threshold_numbers(entities)
 
         states: dict[int, list[float]] = {}
-        waiter = StateWaiter()
-        gate_0_read = waiter.expect(
+        state_waiter = StateWaiter()
+        gate_0_read = state_waiter.expect(
             _number_state_of(numbers["gate_0_move_threshold"].key),
             label="gate 0 move threshold",
         )
-        gate_5_read = waiter.expect(
+        gate_5_read = state_waiter.expect(
             _number_state_of(numbers["gate_5_move_threshold"].key),
             label="gate 5 move threshold",
         )
@@ -551,7 +521,7 @@ async def test_uart_mock_ld2412_truncated_answer(
         def record(state: EntityState) -> None:
             if isinstance(state, NumberState) and not state.missing_state:
                 states.setdefault(state.key, []).append(state.state)
-            waiter.on_state(state)
+            state_waiter.on_state(state)
 
         initial_state_helper = await _wait_initial(client, entities, record)
 
@@ -561,12 +531,14 @@ async def test_uart_mock_ld2412_truncated_answer(
 
         try:
             _, gate_0_state, gate_5_state = await asyncio.gather(
-                asyncio.wait_for(dropped, timeout=5.0), gate_0_read, gate_5_read
+                waiter.wait_for("Dropping gate threshold answer", timeout=5.0),
+                gate_0_read,
+                gate_5_read,
             )
         except TimeoutError:
             pytest.fail(
                 "Timeout waiting for the threshold answers.\n"
-                f"  ld2412 warnings: {warnings}\n"
+                f"  ld2412 warnings: {[line for line in waiter.lines if '[W][ld2412' in line]}\n"
                 f"  states: {states}"
             )
 

tests/integration/test_uart_mock_ld2420.py

@@ -19,11 +19,10 @@ test_uart_mock_ld2420_simple (simple mode):
 
 from __future__ import annotations
 
-import asyncio
-
 from aioesphomeapi import ButtonInfo
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -36,27 +35,7 @@ async def test_uart_mock_ld2420(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test LD2420 energy mode: happy path, truncated frame, overflow, and recovery."""
-    loop = asyncio.get_running_loop()
-
-    # Track overflow warning in logs
-    overflow_seen = loop.create_future()
-
-    # Track "Energy frame too short" warning (PR #14458 bug #3 fix)
-    # This message ONLY exists after the fix. Pre-fix, handle_energy_mode_
-    # silently reads past the buffer without any warning.
-    truncated_frame_warning_seen = loop.create_future()
-
-    # Track TX data logged by the mock for assertions
-    tx_log_lines: list[str] = []
-
-    def line_callback(line: str) -> None:
-        if "Max command length exceeded" in line and not overflow_seen.done():
-            overflow_seen.set_result(True)
-        if "Energy frame too short" in line and not truncated_frame_warning_seen.done():
-            truncated_frame_warning_seen.set_result(True)
-        # Capture all TX log lines from uart_mock
-        if "uart_mock" in line and "TX " in line:
-            tx_log_lines.append(line)
+    waiter = LineWaiter()
 
     collector = SensorStateCollector(
         sensor_names=["moving_distance"],
@@ -69,7 +48,7 @@ async def test_uart_mock_ld2420(
     )
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -104,20 +83,23 @@ async def test_uart_mock_ld2420(
         )
 
         # Verify overflow warning was logged
-        assert overflow_seen.done(), (
+        assert any("Max command length exceeded" in line for line in waiter.lines), (
             "Expected 'Max command length exceeded' warning in logs"
         )
 
         # Verify truncated frame warning was logged (PR #14458 bug #3)
         # This assertion FAILS before PR #14458 because the length check
         # and warning message did not exist.
-        assert truncated_frame_warning_seen.done(), (
+        assert any("Energy frame too short" in line for line in waiter.lines), (
             "Expected 'Energy frame too short' warning in logs. "
             "This indicates PR #14458 fix for handle_energy_mode_ length "
             "validation is missing."
         )
 
         # Verify LD2420 sent setup commands (TX logging)
+        tx_log_lines = [
+            line for line in waiter.lines if "uart_mock" in line and "TX " in line
+        ]
         assert len(tx_log_lines) > 0, "Expected TX log lines from uart_mock"
         tx_data = " ".join(tx_log_lines)
         # Verify command frame header appears (FD:FC:FB:FA)
@@ -147,14 +129,7 @@ async def test_uart_mock_ld2420_simple(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test LD2420 simple mode: happy path, overflow, and 16-digit bug trigger."""
-    loop = asyncio.get_running_loop()
-
-    # Track overflow warning in logs
-    overflow_seen = loop.create_future()
-
-    def line_callback(line: str) -> None:
-        if "Max command length exceeded" in line and not overflow_seen.done():
-            overflow_seen.set_result(True)
+    waiter = LineWaiter()
 
     collector = SensorStateCollector(
         sensor_names=["moving_distance"],
@@ -170,7 +145,7 @@ async def test_uart_mock_ld2420_simple(
     )
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -203,7 +178,7 @@ async def test_uart_mock_ld2420_simple(
         )
 
         # Verify overflow warning was logged
-        assert overflow_seen.done(), (
+        assert any("Max command length exceeded" in line for line in waiter.lines), (
             "Expected 'Max command length exceeded' warning in logs"
         )
 

tests/integration/test_uart_mock_ld2450.py

@@ -17,11 +17,10 @@ test_uart_mock_ld2450:
 
 from __future__ import annotations
 
-import asyncio
-
 from aioesphomeapi import ButtonInfo
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -34,20 +33,7 @@ async def test_uart_mock_ld2450(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test LD2450 data parsing with happy path, garbage, overflow, and recovery."""
-    loop = asyncio.get_running_loop()
-
-    # Track overflow warning in logs
-    overflow_seen = loop.create_future()
-
-    # Track TX data logged by the mock for assertions
-    tx_log_lines: list[str] = []
-
-    def line_callback(line: str) -> None:
-        if "Max command length exceeded" in line and not overflow_seen.done():
-            overflow_seen.set_result(True)
-        # Capture all TX log lines from uart_mock
-        if "uart_mock" in line and "TX " in line:
-            tx_log_lines.append(line)
+    waiter = LineWaiter()
 
     collector = SensorStateCollector(
         sensor_names=[
@@ -91,7 +77,7 @@ async def test_uart_mock_ld2450(
     )
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -156,11 +142,14 @@ async def test_uart_mock_ld2450(
         )
 
         # Verify overflow warning was logged
-        assert overflow_seen.done(), (
+        assert any("Max command length exceeded" in line for line in waiter.lines), (
             "Expected 'Max command length exceeded' warning in logs"
         )
 
         # Verify LD2450 sent setup commands (TX logging)
+        tx_log_lines = [
+            line for line in waiter.lines if "uart_mock" in line and "TX " in line
+        ]
         assert len(tx_log_lines) > 0, "Expected TX log lines from uart_mock"
         tx_data = " ".join(tx_log_lines)
         # Verify command frame header appears (FD:FC:FB:FA)

tests/integration/test_uart_mock_ld2450_polygon_zone_restore.py

@@ -16,6 +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 .types import CompileFunction, ConfigWriter
 
@@ -98,13 +99,10 @@ async def test_uart_mock_ld2450_polygon_zone_restore(
     prefs[keys[0]] = bytes(data)
     write_host_prefs(DEVICE_NAME, prefs)
 
-    warnings: list[str] = []
-
-    def capture_warning(line: str) -> None:
-        if "saved polygon is invalid" in line:
-            warnings.append(line)
-
-    assert await initial_polygon(capture_warning) == ""
-    assert warnings, "expected a warning about the invalid saved polygon"
+    waiter = LineWaiter()
+    assert await initial_polygon(waiter.callback) == ""
+    assert any("saved polygon is invalid" in line for line in waiter.lines), (
+        "expected a warning about the invalid saved polygon"
+    )
 
     clear_host_prefs(DEVICE_NAME)

tests/integration/test_uart_mock_ld2460.py

@@ -14,11 +14,10 @@ test_uart_mock_ld2460:
 
 from __future__ import annotations
 
-import asyncio
-
 from aioesphomeapi import ButtonInfo
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -31,15 +30,7 @@ async def test_uart_mock_ld2460(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test LD2460 frame parsing with the datasheet frames, garbage and overflow."""
-    loop = asyncio.get_running_loop()
-    overflow_seen = loop.create_future()
-    tx_log_lines: list[str] = []
-
-    def line_callback(line: str) -> None:
-        if "Max frame length exceeded" in line and not overflow_seen.done():
-            overflow_seen.set_result(True)
-        if "uart_mock" in line and "TX " in line:
-            tx_log_lines.append(line)
+    waiter = LineWaiter()
 
     collector = SensorStateCollector(
         sensor_names=[
@@ -75,7 +66,7 @@ async def test_uart_mock_ld2460(
     )
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -121,7 +112,9 @@ async def test_uart_mock_ld2460(
             "Timeout waiting for the recovery frame",
             sensor_states=collector.sensor_states,
         )
-        assert overflow_seen.done(), "Expected 'Max frame length exceeded' in the log"
+        assert any("Max frame length exceeded" in line for line in waiter.lines), (
+            "Expected 'Max frame length exceeded' in the log"
+        )
         assert (
             pytest.approx(3.606, abs=0.01)
             in collector.sensor_states["target_1_distance"]
@@ -140,7 +133,9 @@ async def test_uart_mock_ld2460(
         )
 
         # The hub sent the installation-mode query from the datasheet (table 11)
-        tx_data = " ".join(tx_log_lines)
+        tx_data = " ".join(
+            line for line in waiter.lines if "uart_mock" in line and "TX " in line
+        )
         assert "FD:FC:FB:FA:0A:0C:00:01:04:03:02:01" in tx_data, (
             f"Expected the installation-mode query in the TX log, got: {tx_data}"
         )

tests/integration/test_uart_mock_pylontech.py

@@ -16,6 +16,7 @@ import asyncio
 from aioesphomeapi import ButtonInfo
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -66,14 +67,7 @@ async def test_uart_mock_pylontech(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test pylontech parsing with happy path, garbage, overflow and recovery."""
-    overflow_lines: list[str] = []
-    rejected_lines: list[str] = []
-
-    def line_callback(line: str) -> None:
-        if "Line over 199 chars dropped" in line:
-            overflow_lines.append(line)
-        if "malformed lines" in line:
-            rejected_lines.append(line)
+    waiter = LineWaiter()
 
     collector = SensorStateCollector(
         sensor_names=SENSORS, text_sensor_names=TEXT_SENSORS
@@ -88,7 +82,7 @@ async def test_uart_mock_pylontech(
     )
 
     async with (
-        run_compiled(yaml_config, line_callback=line_callback),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         entities, _ = await client.list_entities_services()
@@ -119,6 +113,11 @@ async def test_uart_mock_pylontech(
             assert collector.text_sensor_states[name] == values, name
 
         # Phase 4 and the phase 5 header both overflow
+        overflow_lines = [
+            line for line in waiter.lines if "Line over 199 chars dropped" in line
+        ]
         assert len(overflow_lines) == 2, overflow_lines
         # The phase 3 cut line is counted and reported once the UART is idle
-        assert rejected_lines, "no rejected-line summary logged"
+        assert any("malformed lines" in line for line in waiter.lines), (
+            "no rejected-line summary logged"
+        )

tests/integration/test_udp.py

@@ -11,9 +11,9 @@ import socket
 
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name, require_service
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 
 @dataclass
@@ -111,13 +111,7 @@ async def test_udp_send_receive(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test UDP component can send and receive messages."""
-    log_lines: list[str] = []
-    receive_event = asyncio.Event()
-
-    def on_log_line(line: str) -> None:
-        log_lines.append(line)
-        if "Received UDP:" in line:
-            receive_event.set()
+    waiter = LineWaiter()
 
     async with udp_listener() as (broadcast_port, receiver):
         listen_port = _get_free_udp_port()
@@ -125,7 +119,7 @@ async def test_udp_send_receive(
         config = config.replace("UDP_BROADCAST_PORT_PLACEHOLDER", str(broadcast_port))
 
         async with (
-            run_compiled(config, line_callback=on_log_line),
+            run_compiled(config, line_callback=waiter.callback),
             api_client_connected() as client,
         ):
             # Verify device is running
@@ -165,7 +159,7 @@ async def test_udp_send_receive(
 
             # Verify dump_config logged all configured addresses
             # This tests that FixedVector<const char*> stores addresses correctly
-            log_text = "\n".join(log_lines)
+            log_text = "\n".join(waiter.lines)
             assert "Address: 127.0.0.1" in log_text, (
                 f"Address 127.0.0.1 not found in dump_config. Log: {log_text[-2000:]}"
             )
@@ -181,9 +175,4 @@ async def test_udp_send_receive(
             finally:
                 send_sock.close()
 
-            await wait_or_fail(
-                receive_event,
-                5.0,
-                "on_receive did not fire. Expected 'Received UDP:' in logs",
-                log_lines=log_lines,
-            )
+            await waiter.wait_for("Received UDP:", timeout=5.0)

tests/integration/test_udp_multicast.py

@@ -7,9 +7,9 @@ import socket
 
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 MULTICAST_GROUP = "239.1.2.3"
 
@@ -30,36 +30,25 @@ async def test_udp_multicast_join(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """The group is joined from the poll after setup, once, and then receives."""
-    log_lines: list[str] = []
-    joined_event = asyncio.Event()
-    receive_event = asyncio.Event()
-
-    def on_log_line(line: str) -> None:
-        log_lines.append(line)
-        if f"Joined multicast {MULTICAST_GROUP}" in line:
-            joined_event.set()
-        if "Received UDP:" in line:
-            receive_event.set()
+    waiter = LineWaiter()
 
     listen_port = _get_free_udp_port()
     config = yaml_config.replace("UDP_LISTEN_PORT_PLACEHOLDER", str(listen_port))
 
     async with (
-        run_compiled(config, line_callback=on_log_line),
+        run_compiled(config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await assert_device_name(client, "udp-multicast-test")
 
-        await wait_or_fail(
-            joined_event, 5.0, "Multicast group was not joined", log_lines=log_lines
-        )
+        await waiter.wait_for(f"Joined multicast {MULTICAST_GROUP}", timeout=5.0)
 
         # The poll cancels itself after the join, so a second poll period must not join again
         await asyncio.sleep(2.5)
-        join_lines = [line for line in log_lines if "Joined multicast" in line]
+        join_lines = [line for line in waiter.lines if "Joined multicast" in line]
         assert len(join_lines) == 1, f"Expected one join, got: {join_lines}"
-        assert not any("IP_ADD_MEMBERSHIP" in line for line in log_lines), (
-            f"Join reported a failure. Log: {log_lines[-20:]}"
+        assert not any("IP_ADD_MEMBERSHIP" in line for line in waiter.lines), (
+            f"Join reported a failure. Log: {waiter.lines[-20:]}"
         )
 
         send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@@ -68,9 +57,4 @@ async def test_udp_multicast_join(
         finally:
             send_sock.close()
 
-        await wait_or_fail(
-            receive_event,
-            5.0,
-            "Multicast datagram was not received",
-            log_lines=log_lines,
-        )
+        await waiter.wait_for("Received UDP:", timeout=5.0)

tests/integration/test_use_address_runtime.py

@@ -8,13 +8,11 @@ suffix, so it never matched the actual mDNS hostname.
 
 from __future__ import annotations
 
-import asyncio
-
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
-from .wait_utils import wait_or_fail
 
 # Host platform default MAC: 98:35:69:ab:f6:79 -> suffix "abf679"
 MAC_SUFFIX = "abf679"
@@ -27,21 +25,15 @@ async def test_use_address_runtime(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """The API dump_config logs "<name>.local" built from the device name."""
-    address_seen = asyncio.Event()
-
-    def check_output(line: str) -> None:
-        if "Address: use-address-runtime.local:" in line:
-            address_seen.set()
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=check_output),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await assert_device_name(client, "use-address-runtime")
 
-        await wait_or_fail(
-            address_seen, 10.0, "Did not log 'Address: use-address-runtime.local:'"
-        )
+        await waiter.wait_for("Address: use-address-runtime.local:", timeout=10.0)
 
 
 @pytest.mark.asyncio
@@ -51,17 +43,14 @@ async def test_use_address_runtime_mac_suffix(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """With name_add_mac_suffix the logged address includes the MAC suffix."""
-    address_seen = asyncio.Event()
-    expected = f"Address: use-address-mac-{MAC_SUFFIX}.local:"
-
-    def check_output(line: str) -> None:
-        if expected in line:
-            address_seen.set()
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config, line_callback=check_output),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await assert_device_name(client, f"use-address-mac-{MAC_SUFFIX}")
 
-        await wait_or_fail(address_seen, 10.0, f"Did not log '{expected}'")
+        await waiter.wait_for(
+            f"Address: use-address-mac-{MAC_SUFFIX}.local:", timeout=10.0
+        )