feat(test): Add PaneWaiter utility for waiting on pane content

why: Tests need a reliable way to wait for pane content, especially with different shells

what:
- Add PaneWaiter class with wait_for_content, wait_for_prompt, wait_for_text methods
- Add WaitResult class to handle success/failure/error states
- Add comprehensive test suite for waiter functionality

Author: tony

src/libtmux/test/waiter.py

@@ -0,0 +1,154 @@
+"""Test utilities for waiting on tmux pane content.
+
+This module provides utilities for waiting on tmux pane content in tests.
+Inspired by Playwright's sync API for waiting on page content.
+"""
+
+from __future__ import annotations
+
+import typing as t
+from dataclasses import dataclass
+from typing import Callable, TypeVar
+
+from libtmux.test.retry import retry_until
+
+if t.TYPE_CHECKING:
+    from libtmux.pane import Pane
+
+T = TypeVar("T")
+
+
+@dataclass
+class WaitResult(t.Generic[T]):
+    """Result of a wait operation."""
+
+    success: bool
+    value: T | None = None
+    error: Exception | None = None
+
+
+class PaneWaiter:
+    """Utility class for waiting on tmux pane content."""
+
+    def __init__(self, pane: Pane, timeout: float = 2.0) -> None:
+        """Initialize PaneWaiter.
+
+        Parameters
+        ----------
+        pane : Pane
+            The tmux pane to wait on
+        timeout : float, optional
+            Default timeout in seconds, by default 2.0
+        """
+        self.pane = pane
+        self.timeout = timeout
+
+    def wait_for_content(
+        self,
+        predicate: Callable[[str], bool],
+        *,
+        timeout: float | None = None,
+        error_message: str | None = None,
+    ) -> WaitResult[str]:
+        """Wait for pane content to match predicate.
+
+        Parameters
+        ----------
+        predicate : Callable[[str], bool]
+            Function that takes pane content as string and returns bool
+        timeout : float | None, optional
+            Timeout in seconds, by default None (uses instance timeout)
+        error_message : str | None, optional
+            Custom error message if timeout occurs, by default None
+
+        Returns
+        -------
+        WaitResult[str]
+            Result containing success status and pane content if successful
+        """
+        timeout = timeout or self.timeout
+        result = WaitResult[str](success=False)
+
+        def check_content() -> bool:
+            try:
+                content = "\n".join(self.pane.capture_pane())
+                if predicate(content):
+                    result.success = True
+                    result.value = content
+                    return True
+                else:
+                    return False
+            except Exception as e:
+                result.error = e
+                return False
+
+        try:
+            success = retry_until(check_content, timeout, raises=False)
+            if not success:
+                result.error = Exception(
+                    error_message or "Timed out waiting for content",
+                )
+        except Exception as e:
+            result.error = e
+            if error_message:
+                result.error = Exception(error_message)
+
+        return result
+
+    def wait_for_prompt(
+        self,
+        prompt: str,
+        *,
+        timeout: float | None = None,
+        error_message: str | None = None,
+    ) -> WaitResult[str]:
+        """Wait for specific prompt to appear in pane.
+
+        Parameters
+        ----------
+        prompt : str
+            The prompt text to wait for
+        timeout : float | None, optional
+            Timeout in seconds, by default None (uses instance timeout)
+        error_message : str | None, optional
+            Custom error message if timeout occurs, by default None
+
+        Returns
+        -------
+        WaitResult[str]
+            Result containing success status and pane content if successful
+        """
+        return self.wait_for_content(
+            lambda content: prompt in content and len(content.strip()) > 0,
+            timeout=timeout,
+            error_message=error_message or f"Prompt '{prompt}' not found in pane",
+        )
+
+    def wait_for_text(
+        self,
+        text: str,
+        *,
+        timeout: float | None = None,
+        error_message: str | None = None,
+    ) -> WaitResult[str]:
+        """Wait for specific text to appear in pane.
+
+        Parameters
+        ----------
+        text : str
+            The text to wait for
+        timeout : float | None, optional
+            Timeout in seconds, by default None (uses instance timeout)
+        error_message : str | None, optional
+            Custom error message if timeout occurs, by default None
+
+        Returns
+        -------
+        WaitResult[str]
+            Result containing success status and pane content if successful
+        """
+        return self.wait_for_content(
+            lambda content: text in content,
+            timeout=timeout,
+            error_message=error_message or f"Text '{text}' not found in pane",
+        )

tests/test/test_waiter.py

@@ -0,0 +1,122 @@
+"""Tests for libtmux test waiter utilities."""
+
+from __future__ import annotations
+
+import shutil
+import typing as t
+
+from libtmux.test.waiter import PaneWaiter
+
+if t.TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+def test_wait_for_prompt(session: Session) -> None:
+    """Test waiting for prompt."""
+    env = shutil.which("env")
+    assert env is not None, "Cannot find usable `env` in PATH."
+
+    session.new_window(
+        attach=True,
+        window_name="test_waiter",
+        window_shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh",
+    )
+    pane = session.active_window.active_pane
+    assert pane is not None
+
+    waiter = PaneWaiter(pane)
+    result = waiter.wait_for_prompt("READY>")
+    assert result.success
+    assert result.value is not None
+    assert "READY>" in result.value
+
+
+def test_wait_for_text(session: Session) -> None:
+    """Test waiting for text."""
+    env = shutil.which("env")
+    assert env is not None, "Cannot find usable `env` in PATH."
+
+    session.new_window(
+        attach=True,
+        window_name="test_waiter",
+        window_shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh",
+    )
+    pane = session.active_window.active_pane
+    assert pane is not None
+
+    waiter = PaneWaiter(pane)
+    waiter.wait_for_prompt("READY>")  # Wait for shell to be ready
+
+    pane.send_keys("echo 'Hello World'", literal=True)
+    result = waiter.wait_for_text("Hello World")
+    assert result.success
+    assert result.value is not None
+    assert "Hello World" in result.value
+
+
+def test_wait_timeout(session: Session) -> None:
+    """Test timeout behavior."""
+    env = shutil.which("env")
+    assert env is not None, "Cannot find usable `env` in PATH."
+
+    session.new_window(
+        attach=True,
+        window_name="test_waiter",
+        window_shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh",
+    )
+    pane = session.active_window.active_pane
+    assert pane is not None
+
+    waiter = PaneWaiter(pane, timeout=0.1)  # Short timeout
+    result = waiter.wait_for_text("this text will never appear")
+    assert not result.success
+    assert result.value is None
+    assert result.error is not None
+
+
+def test_custom_error_message(session: Session) -> None:
+    """Test custom error message."""
+    env = shutil.which("env")
+    assert env is not None, "Cannot find usable `env` in PATH."
+
+    session.new_window(
+        attach=True,
+        window_name="test_waiter",
+        window_shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh",
+    )
+    pane = session.active_window.active_pane
+    assert pane is not None
+
+    waiter = PaneWaiter(pane, timeout=0.1)  # Short timeout
+    custom_message = "Custom error message"
+    result = waiter.wait_for_text(
+        "this text will never appear",
+        error_message=custom_message,
+    )
+    assert not result.success
+    assert result.value is None
+    assert result.error is not None
+    assert str(result.error) == custom_message
+
+
+def test_wait_for_content_predicate(session: Session) -> None:
+    """Test waiting with custom predicate."""
+    env = shutil.which("env")
+    assert env is not None, "Cannot find usable `env` in PATH."
+
+    session.new_window(
+        attach=True,
+        window_name="test_waiter",
+        window_shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh",
+    )
+    pane = session.active_window.active_pane
+    assert pane is not None
+
+    waiter = PaneWaiter(pane)
+    waiter.wait_for_prompt("READY>")  # Wait for shell to be ready
+
+    pane.send_keys("echo '123'", literal=True)
+    result = waiter.wait_for_content(lambda content: "123" in content)
+    assert result.success
+    assert result.value is not None
+    assert "123" in result.value