Type expect by pattern kind and note that one line resolves every matching expectation

tests/integration/README.md

@@ -370,38 +370,34 @@ async with run_compiled(yaml_config, line_callback=on_log_line):
     # Test implementation
 ```
 
-Example using futures for specific log patterns:
+To wait for log lines, use `LineWaiter` (`log_utils.py`) instead of a hand written
+callback with one future per pattern:
 ```python
 import re
 
-loop = asyncio.get_running_loop()
-connected_future = loop.create_future()
-service_future = loop.create_future()
+from .log_utils import LineWaiter
 
-# Patterns to match
-connected_pattern = re.compile(r"Client .* connected from")
-service_pattern = re.compile(r"Service called")
+log = LineWaiter()
+# Arm expectations before the action that logs them; any number may be pending.
+connected = log.expect(re.compile(r"Client .* connected from"))
+value = log.expect(re.compile(r"value=(\d+)"))
 
-
-def check_output(line: str) -> None:
-    """Check log output for expected messages."""
-    if not connected_future.done() and connected_pattern.search(line):
-        connected_future.set_result(True)
-    elif not service_future.done() and service_pattern.search(line):
-        service_future.set_result(True)
-
-
-async with run_compiled(yaml_config, line_callback=check_output):
+async with run_compiled(yaml_config, line_callback=log.callback):
     async with api_client_connected() as client:
-        # Wait for specific log message
-        await asyncio.wait_for(connected_future, timeout=5.0)
-
-        # Do test actions...
-
-        # Wait for service log
-        await asyncio.wait_for(service_future, timeout=5.0)
+        await log.wait(connected)
+        match = await log.wait(value, timeout=5.0, msg="no value logged")
+        assert match.group(1) == "42"
+        assert log.count("Service called") == 1
 ```
 
+- `expect(pattern)` only matches lines that arrive after the call. A `str` matches as a
+  substring and resolves to the line; a compiled regex is searched and resolves to its match.
+  Each expectation resolves on the next matching line, so two armed for the same text both
+  resolve on its first occurrence: arm the second after the first resolves, or use `count`.
+- `wait(future)` fails the test on timeout with the pattern and the last 20 log lines.
+- `wait_for(*needles)` also matches lines seen before the call, but only one may be pending.
+- `count(pattern)` counts the lines seen so far.
+
 **Note**: Tests that monitor log messages typically have fewer race conditions compared to state-based testing, making them more reliable. However, be aware that the host platform currently does not have a thread-safe logger, so logging from threads will not work correctly.
 
 ##### Timeout Handling

tests/integration/log_utils.py

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

tests/unit_tests/test_integration_log_utils.py

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