[tests] Add wait_or_fail for integration test timeouts (#21152)

tests/integration/README.md

@@ -405,12 +405,15 @@ async with run_compiled(yaml_config, line_callback=check_output):
 **Note**: Tests that monitor log messages typically have fewer race conditions compared to state-based testing, making them more reliable. However, be aware that the host platform currently does not have a thread-safe logger, so logging from threads will not work correctly.
 
 ##### Timeout Handling
+Always use timeouts for async operations. `wait_or_fail` (`wait_utils.py`) awaits a future,
+coroutine or `asyncio.Event` and fails the test with a message on timeout. Keyword arguments
+are added to the message only on failure, so pass live collections; lists, tuples, deques and
+mappings show their last 20 items.
 ```python
-# Always use timeouts for async operations
-try:
-    result = await asyncio.wait_for(some_future, timeout=5.0)
-except asyncio.TimeoutError:
-    pytest.fail("Operation timed out - check test expectations")
+from .wait_utils import wait_or_fail
+
+result = await wait_or_fail(some_future, 5.0, "no reply to the service call")
+await wait_or_fail(done_event, 5.0, "sensor never updated", log_lines=log_lines)
 ```
 
 #### 6. Common Assertions

tests/integration/wait_utils.py

@@ -0,0 +1,61 @@
+"""Helpers for waiting on futures and events in integration tests."""
+
+from __future__ import annotations
+
+import asyncio
+from collections import deque
+from collections.abc import Awaitable, Mapping
+
+import pytest
+
+# How many trailing items of a collection are shown in a failure message.
+TAIL_ITEMS = 20
+
+
+def _format_context(context: dict[str, object]) -> str:
+    """Render context values for a failure message, keeping only the tail of collections."""
+    parts: list[str] = []
+    for name, value in context.items():
+        if isinstance(value, Mapping):
+            items = [f"{key}: {item!r}" for key, item in value.items()]
+        elif isinstance(value, (list, tuple, deque)):
+            items = [str(item) for item in value]
+        else:
+            parts.append(f"{name}: {value!r}")
+            continue
+        tail = items[-TAIL_ITEMS:]
+        header = f"{name} (last {len(tail)} of {len(items)}):"
+        parts.append("\n  ".join([header, *tail]))
+    return "".join(f"\n{part}" for part in parts)
+
+
+async def wait_or_fail[T](
+    aw: Awaitable[T] | asyncio.Event,
+    timeout: float = 5.0,
+    msg: str = "timed out",
+    **context: object,
+) -> T:
+    """Await ``aw``, or wait for an ``asyncio.Event`` to be set, and fail the test on timeout.
+
+    Behaves like ``asyncio.wait_for``, so a timed out future is cancelled; a TimeoutError raised
+    by ``aw`` itself propagates unchanged, and ``timeout`` must be positive. ``context`` values
+    are formatted only on failure, so pass live collections; lists, tuples, deques and mappings
+    show their last 20 items.
+    """
+    if timeout <= 0:
+        # asyncio.wait_for treats this as already timed out; a test should never ask for it
+        if asyncio.iscoroutine(aw):
+            aw.close()
+        raise ValueError(f"wait_or_fail timeout must be positive, got {timeout}")
+    if isinstance(aw, asyncio.Event):
+        aw = aw.wait()
+    deadline = asyncio.timeout(timeout)
+    try:
+        async with deadline:
+            return await aw
+    except TimeoutError:
+        # A TimeoutError raised by aw itself is not this wait running out
+        if not deadline.expired():
+            raise
+    # Failing outside the except block keeps the TimeoutError traceback out of the report.
+    pytest.fail(f"{msg} (after {timeout}s){_format_context(context)}")

tests/unit_tests/test_integration_wait_utils.py

@@ -0,0 +1,96 @@
+"""Tests for the integration test wait helpers."""
+
+from __future__ import annotations
+
+import asyncio
+from collections import deque
+
+import pytest
+
+from tests.integration.wait_utils import wait_or_fail
+
+
+@pytest.mark.asyncio
+async def test_wait_or_fail_returns_result() -> None:
+    future: asyncio.Future[int] = asyncio.get_running_loop().create_future()
+    future.set_result(42)
+    assert await wait_or_fail(future) == 42
+
+
+@pytest.mark.asyncio
+async def test_wait_or_fail_accepts_coroutine_and_event() -> None:
+    async def produce() -> str:
+        return "done"
+
+    assert await wait_or_fail(produce()) == "done"
+    event = asyncio.Event()
+    asyncio.get_running_loop().call_soon(event.set)
+    assert await wait_or_fail(event) is True
+
+
+@pytest.mark.asyncio
+async def test_wait_or_fail_timeout_cancels_and_reports_context() -> None:
+    future: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+    lines = [f"line {i}" for i in range(25)]
+    seen: dict[str, int] = {}
+    waiting = asyncio.ensure_future(
+        wait_or_fail(future, 0.01, "no reply", lines=lines, seen=seen)
+    )
+    # Context is formatted on failure, so later changes to live objects show up.
+    lines.append("late line")
+    seen["a"] = 1
+    with pytest.raises(pytest.fail.Exception) as err:
+        await waiting
+    message = str(err.value)
+    assert message.startswith("no reply (after 0.01s)")
+    assert "lines (last 20 of 26):" in message
+    assert "line 5\n" not in message
+    assert "line 6\n" in message
+    assert message.endswith("late line\nseen (last 1 of 1):\n  a: 1")
+    assert future.cancelled()
+
+
+@pytest.mark.asyncio
+async def test_wait_or_fail_trims_deques_and_mappings() -> None:
+    lines = deque(f"line {i}" for i in range(30))
+    states = {f"key{i}": i for i in range(25)}
+    pairs = tuple(range(25))
+    with pytest.raises(pytest.fail.Exception) as err:
+        await wait_or_fail(
+            asyncio.Event(),
+            0.01,
+            "no state",
+            lines=lines,
+            states=states,
+            pairs=pairs,
+        )
+    message = str(err.value)
+    assert "lines (last 20 of 30):\n  line 10\n" in message
+    assert "states (last 20 of 25):\n  key5: 5\n" in message
+    assert "key4: 4" not in message
+    assert "pairs (last 20 of 25):\n  5\n" in message
+
+
+@pytest.mark.asyncio
+async def test_wait_or_fail_reraises_a_timeout_from_the_awaitable() -> None:
+    async def inner() -> None:
+        raise TimeoutError("api call timed out")
+
+    with pytest.raises(TimeoutError, match="api call timed out"):
+        await wait_or_fail(inner(), 5.0, "no reply")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("timeout", [0, -1.0])
+async def test_wait_or_fail_rejects_a_non_positive_timeout(timeout: float) -> None:
+    event = asyncio.Event()
+    event.set()
+    done: asyncio.Future[int] = asyncio.get_running_loop().create_future()
+    done.set_result(1)
+
+    async def finished() -> int:
+        return 1
+
+    for aw in (finished(), event, done):
+        with pytest.raises(ValueError, match="must be positive"):
+            await wait_or_fail(aw, timeout)