apply automatic formatting fixes

tests/integration/README.md

@@ -239,11 +239,10 @@ 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
-try:
-    result = await asyncio.wait_for(state_future, timeout=5.0)
-except asyncio.TimeoutError:
-    pytest.fail(f"Expected state not received. Got: {list(states.values())}")
+# 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
+)
 ```
 
 **Legacy: Manual State Tracking**
@@ -268,10 +267,9 @@ def on_state(state: EntityState) -> None:
 client.subscribe_states(on_state)
 
 # Wait for state with timeout
-try:
-    result = await asyncio.wait_for(state_future, timeout=5.0)
-except asyncio.TimeoutError:
-    pytest.fail(f"Expected state not received. Got: {list(states.values())}")
+result = await wait_or_fail(
+    state_future, 5.0, "Expected state not received", states=states
+)
 ```
 
 ##### Service Execution Pattern

tests/integration/test_api_homeassistant.py

@@ -23,6 +23,7 @@ from .state_utils import (
     wait_for_state,
 )
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -251,173 +252,191 @@ async def test_api_homeassistant(
         await client.execute_service(trigger_service, {})
 
         # Wait for all tests to complete with appropriate timeouts
-        try:
-            # Templated service test - the main bug fix
-            computed_value = await asyncio.wait_for(lambda_computed_future, timeout=5.0)
-            # Verify the computed value is reasonable (75 * 255 / 100 = 191.25 -> 191)
-            assert computed_value in ["191", "192"], (
-                f"Unexpected computed value: {computed_value}"
-            )
+        # Shown when a log line or service call does not arrive
+        ctx = {
+            "logged": log_lines,
+            "calls": lambda: [call.service for call in ha_service_calls],
+        }
+        # Templated service test - the main bug fix
+        computed_value = await wait_or_fail(
+            lambda_computed_future, 5.0, "no templated value", **ctx
+        )
+        # Verify the computed value is reasonable (75 * 255 / 100 = 191.25 -> 191)
+        assert computed_value in ["191", "192"], (
+            f"Unexpected computed value: {computed_value}"
+        )
 
-            # Check state reads - verify we received the mocked values
-            temp_line = await asyncio.wait_for(ha_temp_state_future, timeout=5.0)
-            assert "Current HA Temperature: 22.5" in temp_line
+        # Check state reads - verify we received the mocked values
+        temp_line = await wait_or_fail(
+            ha_temp_state_future, 5.0, "no HA temperature", **ctx
+        )
+        assert "Current HA Temperature: 22.5" in temp_line
 
-            humidity_line = await asyncio.wait_for(
-                ha_humidity_state_future, timeout=5.0
-            )
-            assert "Current HA Humidity: 65.0" in humidity_line
+        humidity_line = await wait_or_fail(
+            ha_humidity_state_future, 5.0, "no HA humidity", **ctx
+        )
+        assert "Current HA Humidity: 65.0" in humidity_line
 
-            motion_line = await asyncio.wait_for(ha_motion_state_future, timeout=5.0)
-            assert "Current HA Motion: ON" in motion_line
+        motion_line = await wait_or_fail(
+            ha_motion_state_future, 5.0, "no HA motion", **ctx
+        )
+        assert "Current HA Motion: ON" in motion_line
 
-            weather_line = await asyncio.wait_for(ha_weather_state_future, timeout=5.0)
-            assert "Current HA Weather: sunny" in weather_line
+        weather_line = await wait_or_fail(
+            ha_weather_state_future, 5.0, "no HA weather", **ctx
+        )
+        assert "Current HA Weather: sunny" in weather_line
 
-            # Number test
-            number_value = await asyncio.wait_for(ha_number_future, timeout=5.0)
-            assert number_value == "42.5", f"Unexpected number value: {number_value}"
+        # Number test
+        number_value = await wait_or_fail(
+            ha_number_future, 5.0, "no HA number value", **ctx
+        )
+        assert number_value == "42.5", f"Unexpected number value: {number_value}"
 
-            # Long attribute test - verify 300 chars weren't truncated to 255
-            long_attr_len = await asyncio.wait_for(long_attr_future, timeout=5.0)
-            assert long_attr_len == 300, (
-                f"Long attribute was truncated! Expected 300 chars, got {long_attr_len}. "
-                "This indicates the 256-byte truncation bug."
-            )
+        # Long attribute test - verify 300 chars weren't truncated to 255
+        long_attr_len = await wait_or_fail(
+            long_attr_future, 5.0, "no long attribute", **ctx
+        )
+        assert long_attr_len == 300, (
+            f"Long attribute was truncated! Expected 300 chars, got {long_attr_len}. "
+            "This indicates the 256-byte truncation bug."
+        )
 
-            # Wait for completion
-            await asyncio.wait_for(tests_complete_future, timeout=5.0)
+        # Wait for completion
+        await wait_or_fail(tests_complete_future, 5.0, "tests did not complete", **ctx)
 
-            # Now verify the protobuf messages
-            # 1. Basic service call
-            basic_call = await asyncio.wait_for(
-                service_call_futures["light.turn_off"], timeout=2.0
-            )
-            assert basic_call.service == "light.turn_off"
-            assert "entity_id" in basic_call.data, (
-                f"entity_id not found in data: {basic_call.data}"
-            )
-            assert basic_call.data["entity_id"] == "light.test_light", (
-                f"Wrong entity_id: {basic_call.data['entity_id']}"
-            )
+        # Now verify the protobuf messages
+        # 1. Basic service call
+        basic_call = await wait_or_fail(
+            service_call_futures["light.turn_off"], 2.0, "no light.turn_off call", **ctx
+        )
+        assert basic_call.service == "light.turn_off"
+        assert "entity_id" in basic_call.data, (
+            f"entity_id not found in data: {basic_call.data}"
+        )
+        assert basic_call.data["entity_id"] == "light.test_light", (
+            f"Wrong entity_id: {basic_call.data['entity_id']}"
+        )
 
-            # 2. Templated service call - verify the temporary string issue is fixed
-            templated_call = await asyncio.wait_for(
-                service_call_futures["light.turn_on"], timeout=2.0
-            )
-            assert templated_call.service == "light.turn_on"
-            # Check the computed brightness value
-            assert "brightness" in templated_call.data
-            assert templated_call.data["brightness"] in ["191", "192"]  # 75 * 255 / 100
-            # Check data_template
-            assert "color_name" in templated_call.data_template
-            assert templated_call.data_template["color_name"] == "test_value"
-            # Check variables
-            assert "transition" in templated_call.variables
-            assert templated_call.variables["transition"] == "2.5"
+        # 2. Templated service call - verify the temporary string issue is fixed
+        templated_call = await wait_or_fail(
+            service_call_futures["light.turn_on"], 2.0, "no light.turn_on call", **ctx
+        )
+        assert templated_call.service == "light.turn_on"
+        # Check the computed brightness value
+        assert "brightness" in templated_call.data
+        assert templated_call.data["brightness"] in ["191", "192"]  # 75 * 255 / 100
+        # Check data_template
+        assert "color_name" in templated_call.data_template
+        assert templated_call.data_template["color_name"] == "test_value"
+        # Check variables
+        assert "transition" in templated_call.variables
+        assert templated_call.variables["transition"] == "2.5"
 
-            # 3. Empty string service call
-            empty_call = await asyncio.wait_for(
-                service_call_futures["notify.test"], timeout=2.0
-            )
-            assert empty_call.service == "notify.test"
-            # Verify empty strings are properly handled
-            assert "title" in empty_call.data and empty_call.data["title"] == ""
-            assert (
-                "target" in empty_call.data_template
-                and empty_call.data_template["target"] == ""
-            )
-            assert (
-                "sound" in empty_call.variables and empty_call.variables["sound"] == ""
-            )
+        # 3. Empty string service call
+        empty_call = await wait_or_fail(
+            service_call_futures["notify.test"], 2.0, "no notify.test call", **ctx
+        )
+        assert empty_call.service == "notify.test"
+        # Verify empty strings are properly handled
+        assert "title" in empty_call.data and empty_call.data["title"] == ""
+        assert (
+            "target" in empty_call.data_template
+            and empty_call.data_template["target"] == ""
+        )
+        assert "sound" in empty_call.variables and empty_call.variables["sound"] == ""
 
-            # 4. Multiple fields service call
-            multi_call = await asyncio.wait_for(
-                service_call_futures["climate.set_temperature"], timeout=2.0
-            )
-            assert multi_call.service == "climate.set_temperature"
-            assert multi_call.data["temperature"] == "22"
-            assert multi_call.data["hvac_mode"] == "heat"
-            assert multi_call.data_template["target_temp_high"] == "24"
-            assert multi_call.variables["preset_mode"] == "comfort"
+        # 4. Multiple fields service call
+        multi_call = await wait_or_fail(
+            service_call_futures["climate.set_temperature"],
+            2.0,
+            "no climate.set_temperature call",
+            **ctx,
+        )
+        assert multi_call.service == "climate.set_temperature"
+        assert multi_call.data["temperature"] == "22"
+        assert multi_call.data["hvac_mode"] == "heat"
+        assert multi_call.data_template["target_temp_high"] == "24"
+        assert multi_call.variables["preset_mode"] == "comfort"
 
-            # 5. Complex lambda service call
-            complex_call = await asyncio.wait_for(
-                service_call_futures["script.test_script"], timeout=2.0
-            )
-            assert complex_call.service == "script.test_script"
-            assert complex_call.data["entity_id"] == "light.living_room"
-            assert complex_call.data["brightness_pct"] == "99"  # 42 * 2.38 ≈ 99
-            # Check message includes sensor value
-            assert "message" in complex_call.data_template
-            assert "Sensor: 42.0" in complex_call.data_template["message"]
+        # 5. Complex lambda service call
+        complex_call = await wait_or_fail(
+            service_call_futures["script.test_script"],
+            2.0,
+            "no script.test_script call",
+            **ctx,
+        )
+        assert complex_call.service == "script.test_script"
+        assert complex_call.data["entity_id"] == "light.living_room"
+        assert complex_call.data["brightness_pct"] == "99"  # 42 * 2.38 ≈ 99
+        # Check message includes sensor value
+        assert "message" in complex_call.data_template
+        assert "Sensor: 42.0" in complex_call.data_template["message"]
 
-            # 6. All empty service call
-            all_empty_call = await asyncio.wait_for(
-                service_call_futures["test.empty"], timeout=2.0
-            )
-            assert all_empty_call.service == "test.empty"
-            # All fields should be empty strings
-            assert all(v == "" for v in all_empty_call.data.values())
-            assert all(v == "" for v in all_empty_call.data_template.values())
-            assert all(v == "" for v in all_empty_call.variables.values())
+        # 6. All empty service call
+        all_empty_call = await wait_or_fail(
+            service_call_futures["test.empty"], 2.0, "no test.empty call", **ctx
+        )
+        assert all_empty_call.service == "test.empty"
+        # All fields should be empty strings
+        assert all(v == "" for v in all_empty_call.data.values())
+        assert all(v == "" for v in all_empty_call.data_template.values())
+        assert all(v == "" for v in all_empty_call.variables.values())
 
-            # 7. HA Number service call
-            number_call = await asyncio.wait_for(
-                service_call_futures["input_number.set_value"], timeout=2.0
-            )
-            assert number_call.service == "input_number.set_value"
-            assert number_call.data["entity_id"] == "input_number.test_number"
-            # The value might be formatted with trailing zeros
-            assert float(number_call.data["value"]) == 42.5
+        # 7. HA Number service call
+        number_call = await wait_or_fail(
+            service_call_futures["input_number.set_value"],
+            2.0,
+            "no input_number.set_value call",
+            **ctx,
+        )
+        assert number_call.service == "input_number.set_value"
+        assert number_call.data["entity_id"] == "input_number.test_number"
+        # The value might be formatted with trailing zeros
+        assert float(number_call.data["value"]) == 42.5
 
-            # 7b. Real homeassistant numbers pick the action from the entity domain
-            real_number_call = await asyncio.wait_for(
-                ha_real_number_futures[("number.set_value", "number.test_real_number")],
-                timeout=2.0,
-            )
-            assert float(real_number_call.data["value"]) == 12.5
+        # 7b. Real homeassistant numbers pick the action from the entity domain
+        real_number_call = await wait_or_fail(
+            ha_real_number_futures[("number.set_value", "number.test_real_number")],
+            2.0,
+            "no number.set_value call",
+            **ctx,
+        )
+        assert float(real_number_call.data["value"]) == 12.5
 
-            real_input_number_call = await asyncio.wait_for(
-                ha_real_number_futures[
-                    ("input_number.set_value", "input_number.test_real_input_number")
-                ],
-                timeout=2.0,
-            )
-            assert float(real_input_number_call.data["value"]) == 7
+        real_input_number_call = await wait_or_fail(
+            ha_real_number_futures[
+                ("input_number.set_value", "input_number.test_real_input_number")
+            ],
+            2.0,
+            "no input_number.set_value call for the real input number",
+            **ctx,
+        )
+        assert float(real_input_number_call.data["value"]) == 7
 
-            # 8. HA Switch service calls
-            switch_on_call = await asyncio.wait_for(
-                service_call_futures["switch.turn_on"], timeout=2.0
-            )
-            assert switch_on_call.service == "switch.turn_on"
-            assert switch_on_call.data["entity_id"] == "switch.test_switch"
+        # 8. HA Switch service calls
+        switch_on_call = await wait_or_fail(
+            service_call_futures["switch.turn_on"], 2.0, "no switch.turn_on call", **ctx
+        )
+        assert switch_on_call.service == "switch.turn_on"
+        assert switch_on_call.data["entity_id"] == "switch.test_switch"
 
-            switch_off_call = await asyncio.wait_for(
-                service_call_futures["switch.turn_off"], timeout=2.0
-            )
-            assert switch_off_call.service == "switch.turn_off"
-            assert switch_off_call.data["entity_id"] == "switch.test_switch"
+        switch_off_call = await wait_or_fail(
+            service_call_futures["switch.turn_off"],
+            2.0,
+            "no switch.turn_off call",
+            **ctx,
+        )
+        assert switch_off_call.service == "switch.turn_off"
+        assert switch_off_call.data["entity_id"] == "switch.test_switch"
 
-            # 9. Action response error test (tests StringRef error message)
-            # The error response is sent automatically in on_service_call callback
-            # Wait for the error to be logged (proves StringRef -> std::string works)
-            error_log_line = await asyncio.wait_for(
-                action_error_received_future, timeout=2.0
-            )
-            test_error_message = "Test error: action not found"
-            assert test_error_message in error_log_line, (
-                f"Expected error message '{test_error_message}' not found in: {error_log_line}"
-            )
-
-        except TimeoutError as e:
-            # Show recent log lines for debugging
-            recent_logs = "\n".join(log_lines[-20:])
-            service_calls_summary = "\n".join(
-                f"- {call.service}" for call in ha_service_calls
-            )
-            pytest.fail(
-                f"Test timed out waiting for expected log pattern or service call. Error: {e}\n\n"
-                f"Recent log lines:\n{recent_logs}\n\n"
-                f"Received service calls:\n{service_calls_summary}"
-            )
+        # 9. Action response error test (tests StringRef error message)
+        # The error response is sent automatically in on_service_call callback
+        # Wait for the error to be logged (proves StringRef -> std::string works)
+        error_log_line = await wait_or_fail(
+            action_error_received_future, 2.0, "no action error logged", **ctx
+        )
+        test_error_message = "Test error: action not found"
+        assert test_error_message in error_log_line, (
+            f"Expected error message '{test_error_message}' not found in: {error_log_line}"
+        )

tests/integration/test_api_homeassistant_light.py

@@ -23,6 +23,7 @@ import pytest
 from .log_utils import LineWaiter
 from .state_utils import require_entity, require_service
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 ENTITY = "light.kitchen"
 SUBSCRIPTIONS = {
@@ -138,10 +139,12 @@ async def light_entity(
         client.subscribe_service_calls(calls.put_nowait)
         client.subscribe_states(on_state)
         client.subscribe_home_assistant_states(on_state_sub)
-        try:
-            await asyncio.wait_for(all_subscribed, timeout=5.0)
-        except TimeoutError:
-            pytest.fail(f"never subscribed: {SUBSCRIPTIONS - subscribed}")
+        await wait_or_fail(
+            all_subscribed,
+            5.0,
+            "never subscribed",
+            missing=lambda: SUBSCRIPTIONS - subscribed,
+        )
         ha = FakeHomeAssistant(client, key, services, waiter, calls, states)
         ha.report("supported_color_modes", modes)
         ha.report("min_color_temp_kelvin", "2000")

tests/integration/test_api_message_size_batching.py

@@ -9,6 +9,7 @@ import pytest
 
 from .state_utils import assert_device_name, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -128,15 +129,13 @@ async def test_api_message_size_batching(
         client.subscribe_states(on_state)
 
         # Wait for states with timeout
-        try:
-            await asyncio.wait_for(states_future, timeout=5.0)
-        except TimeoutError:
-            missing_keys = expected_keys - received_keys
-            pytest.fail(
-                f"Did not receive states from all entities within 5 seconds. "
-                f"Missing keys: {missing_keys}, "
-                f"Received {len(received_keys)} of {len(expected_keys)} expected states"
-            )
+        await wait_or_fail(
+            states_future,
+            5.0,
+            "Did not receive states from all entities",
+            missing_keys=lambda: expected_keys - received_keys,
+            received=lambda: f"{len(received_keys)} of {len(expected_keys)}",
+        )
 
         # Verify we received states from all entities
         assert expected_keys.issubset(received_keys)

tests/integration/test_device_id_in_state.py

@@ -27,6 +27,7 @@ from aioesphomeapi import (
 import pytest
 
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 # Mapping of entity name to device name for all entities with device_id
 ENTITY_TO_DEVICE = {
@@ -143,20 +144,17 @@ async def test_device_id_in_state(
         client.subscribe_states(on_state)
 
         # Wait for states
-        try:
-            await asyncio.wait_for(states_future, timeout=10.0)
-        except TimeoutError:
-            received_names = {e.name for e in all_entities if e.key in states}
-            missing_names = (
+        await wait_or_fail(
+            states_future,
+            10.0,
+            "Did not receive all entity states",
+            received=lambda: len(states),
+            missing=lambda: (
                 (set(ENTITY_TO_DEVICE) | NO_DEVICE_ENTITIES)
-                - received_names
+                - {e.name for e in all_entities if e.key in states}
                 - {"Doorbell"}
-            )
-            pytest.fail(
-                f"Did not receive all entity states within 10 seconds. "
-                f"Received {len(states)} states. "
-                f"Missing: {missing_names}"
-            )
+            ),
+        )
 
         # Verify each state has the correct device_id
         verified_count = 0

tests/integration/test_duplicate_entities.py

@@ -8,6 +8,7 @@ from aioesphomeapi import EntityInfo
 import pytest
 
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -174,10 +175,10 @@ async def test_duplicate_entities_on_different_devices(
         client.subscribe_states(on_state)
 
         # Wait for all entity states
-        try:
-            await asyncio.wait_for(states_future, timeout=10.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Did not receive all entity states within 10 seconds. "
-                f"Expected {expected_count}, received {state_count}"
-            )
+        await wait_or_fail(
+            states_future,
+            10.0,
+            "Did not receive all entity states",
+            expected=expected_count,
+            received=lambda: state_count,
+        )

tests/integration/test_host_mode_many_entities.py

@@ -22,6 +22,7 @@ import pytest
 
 from .state_utils import require_only_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -65,28 +66,19 @@ async def test_host_mode_many_entities(
         client.subscribe_states(on_state)
 
         # Wait for minimum states with timeout
-        try:
-            await asyncio.wait_for(minimum_states_future, timeout=10.0)
-        except TimeoutError:
-            sensor_states = [
-                s
-                for s in states.values()
-                if isinstance(s, SensorState) and isinstance(s.state, float)
-            ]
-            date_states = [s for s in states.values() if isinstance(s, DateState)]
-            time_states = [s for s in states.values() if isinstance(s, TimeState)]
-            datetime_states = [
-                s for s in states.values() if isinstance(s, DateTimeState)
-            ]
+        def count(kind: type) -> int:
+            return sum(isinstance(s, kind) for s in states.values())
 
-            pytest.fail(
-                f"Did not receive expected states within 10 seconds. "
-                f"Received: {len(sensor_states)} sensor states (expected >=50), "
-                f"{len(date_states)} date states (expected >=1), "
-                f"{len(time_states)} time states (expected >=1), "
-                f"{len(datetime_states)} datetime states (expected >=1). "
-                f"Total states: {len(states)}"
-            )
+        await wait_or_fail(
+            minimum_states_future,
+            10.0,
+            "Did not receive the expected states (>=50 sensor, >=1 date, time and datetime)",
+            sensor=lambda: count(SensorState),
+            date=lambda: count(DateState),
+            time=lambda: count(TimeState),
+            datetime=lambda: count(DateTimeState),
+            total=lambda: len(states),
+        )
 
         # Verify we received a good number of entity states
         assert len(states) >= 50, (

tests/integration/test_host_mode_many_entities_multiple_connections.py

@@ -8,6 +8,7 @@ from aioesphomeapi import EntityState
 import pytest
 
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -45,16 +46,13 @@ async def test_host_mode_many_entities_multiple_connections(
         client2.subscribe_states(on_state2)
 
         # Wait for both clients to receive states
-        try:
-            count1, count2 = await asyncio.gather(
-                asyncio.wait_for(client1_ready, timeout=10.0),
-                asyncio.wait_for(client2_ready, timeout=10.0),
-            )
-        except TimeoutError:
-            pytest.fail(
-                f"One or both clients did not receive enough states within 10 seconds. "
-                f"Client1: {len(states1)}, Client2: {len(states2)}"
-            )
+        count1, count2 = await wait_or_fail(
+            asyncio.gather(client1_ready, client2_ready),
+            10.0,
+            "One or both clients did not receive enough states",
+            client1=lambda: len(states1),
+            client2=lambda: len(states2),
+        )
 
         # Verify both clients received states successfully
         assert count1 >= 20, (

tests/integration/test_logger_buffered_recursion_guard.py

@@ -31,6 +31,7 @@ import pytest
 
 from .state_utils import assert_device_name, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 _ANSI = re.compile(r"\x1b\[[0-9;]*m")
 # THREADMSGnnn_DATA_xxxxxxxx where data is a deterministic checksum of the index
@@ -77,14 +78,14 @@ async def test_logger_buffered_recursion_guard(
         client.button_command(button.key)
 
         # Wait until every buffered thread message has been delivered over the API.
-        try:
-            await asyncio.wait_for(all_drained.wait(), timeout=30.0)
-        except TimeoutError:
-            received = sum(len(THREAD_MSG_PATTERN.findall(m)) for m in api_messages)
-            pytest.fail(
-                f"Only {received}/{NUM_MESSAGES} thread messages arrived before timeout; "
-                "device likely crashed or hung."
-            )
+        await wait_or_fail(
+            all_drained,
+            30.0,
+            f"Not all {NUM_MESSAGES} thread messages arrived; device likely crashed or hung",
+            received=lambda: sum(
+                len(THREAD_MSG_PATTERN.findall(m)) for m in api_messages
+            ),
+        )
 
     intact: set[int] = set()
     contaminated: list[str] = []

tests/integration/test_scheduler_defer_stress.py

@@ -7,6 +7,7 @@ import pytest
 
 from .state_utils import assert_device_name, require_service
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -76,15 +77,13 @@ async def test_scheduler_defer_stress(
         await client.execute_service(run_stress_test_service, {})
 
         # Wait for all defers to execute (should be quick)
-        try:
-            await asyncio.wait_for(test_complete_future, timeout=5.0)
-        except TimeoutError:
-            # Report how many we got
-            pytest.fail(
-                f"Stress test timed out. Only {len(executed_defers)} of "
-                f"1000 defers executed. Missing IDs: "
-                f"{sorted(set(range(1000)) - executed_defers)[:10]}..."
-            )
+        await wait_or_fail(
+            test_complete_future,
+            5.0,
+            "Stress test did not complete",
+            executed=lambda: f"{len(executed_defers)} of 1000 defers",
+            missing_ids=lambda: sorted(set(range(1000)) - executed_defers),
+        )
 
         # Verify all defers executed
         assert len(executed_defers) == 1000, (

tests/integration/test_scheduler_heap_stress.py

@@ -7,6 +7,7 @@ import pytest
 
 from .state_utils import assert_device_name, require_service
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -81,16 +82,13 @@ async def test_scheduler_heap_stress(
         await client.execute_service(run_stress_test_service, {})
 
         # Wait for all callbacks to execute (should be quick, but give more time for scheduling)
-        try:
-            await asyncio.wait_for(test_complete_future, timeout=10.0)
-        except TimeoutError:
-            # Report how many we got
-            missing_ids = sorted(set(range(1000)) - executed_callbacks)
-            pytest.fail(
-                f"Stress test timed out. Only {len(executed_callbacks)} of "
-                f"1000 callbacks executed. Missing IDs: "
-                f"{missing_ids[:20]}... (total missing: {len(missing_ids)})"
-            )
+        await wait_or_fail(
+            test_complete_future,
+            10.0,
+            "Stress test did not complete",
+            executed=lambda: f"{len(executed_callbacks)} of 1000 callbacks",
+            missing_ids=lambda: sorted(set(range(1000)) - executed_callbacks),
+        )
 
         # Verify all callbacks executed
         assert len(executed_callbacks) == 1000, (

tests/integration/test_scheduler_interval_reschedule.py

@@ -147,12 +147,12 @@ async def test_scheduler_interval_reschedule(
             replace_original_future, 5.0, "Replace-test original interval did not fire"
         )
 
-        try:
-            await asyncio.wait_for(replaced_stopped_future, timeout=5.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Replaced interval did not stop. Fired {replaced_fire_count} times"
-            )
+        await wait_or_fail(
+            replaced_stopped_future,
+            5.0,
+            "Replaced interval did not stop",
+            fired=lambda: replaced_fire_count,
+        )
 
         # Wait to ensure replacement doesn't fire again after cancellation
         await asyncio.sleep(0.3)

tests/integration/test_scheduler_pool.py

@@ -9,6 +9,7 @@ import pytest
 
 from .state_utils import assert_device_name, require_service
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -116,60 +117,55 @@ async def test_scheduler_pool(
         }
         complete_service = require_service(services, "run_complete")
 
-        try:
-            # Phase 1: Component lifecycle
-            await client.execute_service(phase_services[1], {})
-            await asyncio.wait_for(phase_futures[1], timeout=1.0)
-            await asyncio.sleep(0.05)  # Let timeouts complete
+        # Shown when a phase does not complete
+        pool = {
+            "phases_completed": lambda: [
+                n for n, fut in phase_futures.items() if fut.done()
+            ],
+            "reuse": lambda: pool_reuse_count,
+            "recycle": lambda: pool_recycle_count,
+            "pool_full": lambda: pool_full_count,
+            "new_alloc": lambda: new_alloc_count,
+            "logged": log_lines,
+        }
+        # Phase 1: Component lifecycle
+        await client.execute_service(phase_services[1], {})
+        await wait_or_fail(phase_futures[1], 1.0, "Phase 1 did not complete", **pool)
+        await asyncio.sleep(0.05)  # Let timeouts complete
 
-            # Phase 2: Sensor polling
-            await client.execute_service(phase_services[2], {})
-            await asyncio.wait_for(phase_futures[2], timeout=1.0)
-            await asyncio.sleep(0.1)  # Let intervals run a bit
+        # Phase 2: Sensor polling
+        await client.execute_service(phase_services[2], {})
+        await wait_or_fail(phase_futures[2], 1.0, "Phase 2 did not complete", **pool)
+        await asyncio.sleep(0.1)  # Let intervals run a bit
 
-            # Phase 3: Communication patterns
-            await client.execute_service(phase_services[3], {})
-            await asyncio.wait_for(phase_futures[3], timeout=1.0)
-            await asyncio.sleep(0.1)  # Let heartbeat run
+        # Phase 3: Communication patterns
+        await client.execute_service(phase_services[3], {})
+        await wait_or_fail(phase_futures[3], 1.0, "Phase 3 did not complete", **pool)
+        await asyncio.sleep(0.1)  # Let heartbeat run
 
-            # Phase 4: Defer patterns
-            await client.execute_service(phase_services[4], {})
-            await asyncio.wait_for(phase_futures[4], timeout=1.0)
-            await asyncio.sleep(0.2)  # Let everything settle and recycle
+        # Phase 4: Defer patterns
+        await client.execute_service(phase_services[4], {})
+        await wait_or_fail(phase_futures[4], 1.0, "Phase 4 did not complete", **pool)
+        await asyncio.sleep(0.2)  # Let everything settle and recycle
 
-            # Phase 5: Pool reuse verification
-            await client.execute_service(phase_services[5], {})
-            await asyncio.wait_for(phase_futures[5], timeout=1.0)
-            await asyncio.sleep(0.1)  # Let Phase 5 timeouts complete and recycle
+        # Phase 5: Pool reuse verification
+        await client.execute_service(phase_services[5], {})
+        await wait_or_fail(phase_futures[5], 1.0, "Phase 5 did not complete", **pool)
+        await asyncio.sleep(0.1)  # Let Phase 5 timeouts complete and recycle
 
-            # Phase 6: Full pool reuse verification
-            await client.execute_service(phase_services[6], {})
-            await asyncio.wait_for(phase_futures[6], timeout=1.0)
-            await asyncio.sleep(0.1)  # Let Phase 6 timeouts complete
+        # Phase 6: Full pool reuse verification
+        await client.execute_service(phase_services[6], {})
+        await wait_or_fail(phase_futures[6], 1.0, "Phase 6 did not complete", **pool)
+        await asyncio.sleep(0.1)  # Let Phase 6 timeouts complete
 
-            # Phase 7: Same-named defer optimization
-            await client.execute_service(phase_services[7], {})
-            await asyncio.wait_for(phase_futures[7], timeout=1.0)
-            await asyncio.sleep(0.05)  # Let the single defer execute
+        # Phase 7: Same-named defer optimization
+        await client.execute_service(phase_services[7], {})
+        await wait_or_fail(phase_futures[7], 1.0, "Phase 7 did not complete", **pool)
+        await asyncio.sleep(0.05)  # Let the single defer execute
 
-            # Complete test
-            await client.execute_service(complete_service, {})
-            await asyncio.wait_for(test_complete_future, timeout=0.5)
-
-        except TimeoutError as e:
-            # Print debug info if test times out
-            recent_logs = "\n".join(log_lines[-30:])
-            phases_completed = [num for num, fut in phase_futures.items() if fut.done()]
-            pytest.fail(
-                f"Test timed out waiting for phase/completion. Error: {e}\n"
-                f"  Phases completed: {phases_completed}\n"
-                f"  Pool stats:\n"
-                f"    Reuse count: {pool_reuse_count}\n"
-                f"    Recycle count: {pool_recycle_count}\n"
-                f"    Pool full count: {pool_full_count}\n"
-                f"    New alloc count: {new_alloc_count}\n"
-                f"Recent logs:\n{recent_logs}"
-            )
+        # Complete test
+        await client.execute_service(complete_service, {})
+        await wait_or_fail(test_complete_future, 0.5, "Test did not complete", **pool)
 
     # Verify all test phases ran
     for phase_num in range(1, 8):

tests/integration/test_script_wait_on_boot.py

@@ -26,6 +26,7 @@ import pytest
 
 from .state_utils import assert_device_name
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -91,33 +92,26 @@ async def test_script_wait_on_boot(
         # Wait for on_boot sequence to complete
         # The boot sequence should complete automatically
         # Timeout is generous to allow for delays in the scripts
-        try:
-            await asyncio.wait_for(test_complete.wait(), timeout=5.0)
-        except TimeoutError:
-            # Build a detailed error message showing where the boot sequence got stuck
-            progress = []
-            if boot_started:
-                progress.append("boot started")
-            if first_script_started:
-                progress.append("show_start_page started")
-            if first_script_completed:
-                progress.append("show_start_page completed")
-            if first_wait_returned:
-                progress.append("first script.wait returned")
-            if second_script_started:
-                progress.append("flip_thru_pages started")
-            if second_script_completed:
-                progress.append("flip_thru_pages completed")
+        def progress() -> list[str]:
+            steps = (
+                (boot_started, "boot started"),
+                (first_script_started, "show_start_page started"),
+                (first_script_completed, "show_start_page completed"),
+                (first_wait_returned, "first script.wait returned"),
+                (second_script_started, "flip_thru_pages started"),
+                (second_script_completed, "flip_thru_pages completed"),
+            )
+            return [name for done, name in steps if done]
 
-            if not first_wait_returned and first_script_completed:
-                pytest.fail(
-                    f"Test timed out - script.wait hung after show_start_page completed! "
-                    f"This is the issue #12043 bug. Progress: {', '.join(progress)}"
-                )
-            else:
-                pytest.fail(
-                    f"Test timed out. Progress: {', '.join(progress) if progress else 'none'}"
-                )
+        await wait_or_fail(
+            test_complete,
+            5.0,
+            "Boot sequence did not complete",
+            progress=progress,
+            issue_12043_script_wait_hung=lambda: (
+                first_script_completed and not first_wait_returned
+            ),
+        )
 
         # Verify the complete boot sequence executed in order
         assert boot_started, "on_boot did not start"

tests/integration/test_select_stringref_trigger.py

@@ -9,6 +9,7 @@ import pytest
 
 from .state_utils import assert_device_name, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 
 @pytest.mark.asyncio
@@ -109,27 +110,25 @@ async def test_select_stringref_trigger(
         client.select_command(action_entity.key, "Action B")
 
         # Wait for all log messages confirming StringRef operations work
-        try:
-            await asyncio.wait_for(
-                asyncio.gather(
-                    value_logged_future,
-                    concatenated_future,
-                    comparison_future,
-                    index_logged_future,
-                    length_future,
-                    find_substr_future,
-                    find_char_future,
-                    substr_future,
-                    set_action_future,
-                    stoi_future,
-                    stol_future,
-                    stof_future,
-                    stod_future,
-                ),
-                timeout=5.0,
-            )
-        except TimeoutError:
-            results = {
+        await wait_or_fail(
+            asyncio.gather(
+                value_logged_future,
+                concatenated_future,
+                comparison_future,
+                index_logged_future,
+                length_future,
+                find_substr_future,
+                find_char_future,
+                substr_future,
+                set_action_future,
+                stoi_future,
+                stol_future,
+                stof_future,
+                stod_future,
+            ),
+            5.0,
+            "StringRef operations failed",
+            received=lambda: {
                 "value_logged": value_logged_future.done(),
                 "concatenated": concatenated_future.done(),
                 "comparison": comparison_future.done(),
@@ -143,5 +142,5 @@ async def test_select_stringref_trigger(
                 "stol": stol_future.done(),
                 "stof": stof_future.done(),
                 "stod": stod_future.done(),
-            }
-            pytest.fail(f"StringRef operations failed - received: {results}")
+            },
+        )