Skip to content

Tests: Add Waiter #579

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 5 commits into from
Closed
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
!squash docs and doctest
  • Loading branch information
tony committed Feb 25, 2025
commit cb737f9361251a8179baf4ea012a3a5fe679a93c
295 changes: 287 additions & 8 deletions src/libtmux/test/waiter.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,39 @@

This module provides utilities for waiting on tmux pane content in tests.
Inspired by Playwright's sync API for waiting on page content.

The main class is :class:`PaneWaiter` which provides methods to wait for specific
content to appear in a tmux pane. This is particularly useful for testing shell
commands and their output.

Examples
--------
>>> from libtmux.test.waiter import PaneWaiter
>>> # Create a new window and get its pane
>>> window = session.new_window(window_name="test_waiter")
>>> pane = window.active_pane
>>> # Create a waiter for the pane
>>> waiter = PaneWaiter(pane)
>>> # Wait for a specific prompt
>>> result = waiter.wait_for_prompt("$ ")
>>> result.success
True
>>> # Send a command and wait for its output
>>> pane.send_keys("echo 'Hello World'")
>>> result = waiter.wait_for_text("Hello World")
>>> result.success
True
>>> "Hello World" in result.value
True

The waiter also handles timeouts and errors gracefully:

>>> # Wait for text that won't appear (times out)
>>> result = waiter.wait_for_text("this won't appear", timeout_seconds=0.1)
>>> result.success
False
>>> isinstance(result.error, WaiterTimeoutError)
True
"""

from __future__ import annotations
Expand All @@ -23,28 +56,140 @@


class WaiterError(LibTmuxException):
"""Base exception for waiter errors."""
"""Base exception for waiter errors.

This is the parent class for all waiter-specific exceptions.
"""


class WaiterTimeoutError(WaiterError):
"""Exception raised when waiting for content times out."""
"""Exception raised when waiting for content times out.

This exception is raised when the content being waited for does not appear
within the specified timeout period.

Examples
--------
>>> waiter = PaneWaiter(pane, timeout=0.1) # Short timeout
>>> result = waiter.wait_for_text("won't appear")
>>> isinstance(result.error, WaiterTimeoutError)
True
>>> str(result.error)
"Text 'won't appear' not found in pane"
"""


class WaiterContentError(WaiterError):
"""Exception raised when there's an error getting or checking content."""
r"""Exception raised when there's an error getting or checking content.

This exception is raised when there's an error accessing or reading the
pane content, for example if the pane is no longer available.

Examples
--------
>>> # Example of handling content errors
>>> try:
... content = "\\n".join(pane.capture_pane())
... except Exception as e:
... error = WaiterContentError("Error capturing pane content")
... error.__cause__ = e
... raise error from e
"""


@dataclass
class WaitResult(t.Generic[T]):
"""Result of a wait operation."""
"""Result of a wait operation.

This class encapsulates the result of a wait operation, including whether it
succeeded, the value found (if any), and any error that occurred.

Parameters
----------
success : bool
Whether the wait operation succeeded
value : T | None
The value found, if any
error : Exception | None
The error that occurred, if any

Examples
--------
>>> # Successful wait result
>>> result = WaitResult[str](success=True, value="found content")
>>> result.success
True
>>> result.value
'found content'
>>> result.error is None
True

>>> # Failed wait result with error
>>> error = WaiterTimeoutError("Timed out")
>>> result = WaitResult[str](success=False, error=error)
>>> result.success
False
>>> result.value is None
True
>>> isinstance(result.error, WaiterTimeoutError)
True
"""

success: bool
value: T | None = None
error: Exception | None = None


class PaneWaiter:
"""Utility class for waiting on tmux pane content."""
"""Utility class for waiting on tmux pane content.

This class provides methods to wait for specific content to appear in a tmux pane.
It supports waiting for exact text matches, prompts, and custom predicates.

Parameters
----------
pane : Pane
The tmux pane to wait on
timeout : float, optional
Default timeout in seconds, by default 2.0

Examples
--------
Basic usage with text:

>>> waiter = PaneWaiter(pane)
>>> pane.send_keys("echo 'test'")
>>> result = waiter.wait_for_text("test")
>>> result.success
True
>>> "test" in result.value
True

Waiting for a prompt:

>>> waiter = PaneWaiter(pane)
>>> result = waiter.wait_for_prompt("$ ")
>>> result.success
True
>>> "$ " in result.value
True

Custom predicate:

>>> waiter = PaneWaiter(pane)
>>> result = waiter.wait_for_content(lambda content: "error" not in content.lower())
>>> result.success
True

Handling timeouts:

>>> waiter = PaneWaiter(pane, timeout=0.1) # Short timeout
>>> result = waiter.wait_for_text("won't appear")
>>> result.success
False
>>> isinstance(result.error, WaiterTimeoutError)
True
"""

def __init__(self, pane: Pane, timeout: float = 2.0) -> None:
"""Initialize PaneWaiter.
Expand All @@ -66,6 +211,9 @@ def _check_content(
) -> bool:
"""Check pane content against predicate.

This internal method captures the pane content and checks it against
the provided predicate function.

Parameters
----------
predicate : Callable[[str], bool]
Expand All @@ -82,6 +230,16 @@ def _check_content(
------
WaiterContentError
If there's an error capturing pane content

Examples
--------
>>> waiter = PaneWaiter(pane)
>>> result = WaitResult[str](success=False)
>>> success = waiter._check_content(lambda c: "test" in c, result)
>>> success # True if "test" is found in pane content
True
>>> result.value is not None
True
"""
try:
content = "\n".join(self.pane.capture_pane())
Expand All @@ -101,7 +259,56 @@ def wait_for_content(
interval_seconds: float | None = None,
error_message: str | None = None,
) -> WaitResult[str]:
"""Wait for content in the pane to match a predicate."""
"""Wait for content in the pane to match a predicate.

This is the core waiting method that other methods build upon. It repeatedly
checks the pane content against a predicate function until it returns True
or times out.

Parameters
----------
predicate : Callable[[str], bool]
Function that takes pane content as string and returns bool
timeout_seconds : float | None, optional
Maximum time to wait in seconds, by default None (uses instance timeout)
interval_seconds : float | None, optional
Time between checks in seconds, by default None (uses 0.05)
error_message : str | None, optional
Custom error message for timeout, by default None

Returns
-------
WaitResult[str]
Result of the wait operation

Examples
--------
>>> waiter = PaneWaiter(pane)
>>> # Wait for content containing "success" but not "error"
>>> result = waiter.wait_for_content(
... lambda content: "success" in content and "error" not in content
... )
>>> result.success
True

>>> # Wait with custom timeout and interval
>>> result = waiter.wait_for_content(
... lambda content: "test" in content,
... timeout_seconds=5.0,
... interval_seconds=0.1,
... )
>>> result.success
True

>>> # Wait with custom error message
>>> result = waiter.wait_for_content(
... lambda content: False, # Never succeeds
... timeout_seconds=0.1,
... error_message="Custom timeout message",
... )
>>> str(result.error)
'Custom timeout message'
"""
result = WaitResult[str](success=False, value=None, error=None)
try:
# Give the shell a moment to be ready
Expand Down Expand Up @@ -134,7 +341,40 @@ def wait_for_prompt(
timeout_seconds: float | None = None,
error_message: str | None = None,
) -> WaitResult[str]:
"""Wait for a specific prompt to appear in the pane."""
"""Wait for a specific prompt to appear in the pane.

This method waits for a specific shell prompt to appear in the pane.
It ensures the prompt is at the end of non-empty content.

Parameters
----------
prompt : str
The prompt text to wait for
timeout_seconds : float | None, optional
Maximum time to wait in seconds, by default None (uses instance timeout)
error_message : str | None, optional
Custom error message for timeout, by default None

Returns
-------
WaitResult[str]
Result of the wait operation

Examples
--------
>>> waiter = PaneWaiter(pane)
>>> # Wait for bash prompt
>>> result = waiter.wait_for_prompt("$ ")
>>> result.success
True
>>> "$ " in result.value
True

>>> # Wait for custom prompt
>>> result = waiter.wait_for_prompt("my_prompt> ")
>>> result.success
True
"""
return self.wait_for_content(
lambda content: prompt in content and len(content.strip()) > 0,
timeout_seconds=timeout_seconds,
Expand All @@ -148,7 +388,46 @@ def wait_for_text(
interval_seconds: float | None = None,
error_message: str | None = None,
) -> WaitResult[str]:
"""Wait for text to appear in the pane."""
"""Wait for text to appear in the pane.

This method waits for specific text to appear anywhere in the pane content.

Parameters
----------
text : str
The text to wait for
timeout_seconds : float | None, optional
Maximum time to wait in seconds, by default None (uses instance timeout)
interval_seconds : float | None, optional
Time between checks in seconds, by default None (uses 0.05)
error_message : str | None, optional
Custom error message for timeout, by default None

Returns
-------
WaitResult[str]
Result of the wait operation

Examples
--------
>>> waiter = PaneWaiter(pane)
>>> # Send a command and wait for its output
>>> pane.send_keys("echo 'Hello World'")
>>> result = waiter.wait_for_text("Hello World")
>>> result.success
True
>>> "Hello World" in result.value
True

>>> # Wait with custom timeout
>>> result = waiter.wait_for_text(
... "test output",
... timeout_seconds=5.0,
... error_message="Failed to find test output",
... )
>>> result.success
True
"""
if error_message is None:
error_message = f"Text '{text}' not found in pane"
return self.wait_for_content(
Expand Down
Loading