From 10d0749f189bf690e12aebb1bc869fc2e1bc4606 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 23 Feb 2025 14:30:10 -0600 Subject: [PATCH 1/5] 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 --- src/libtmux/test/waiter.py | 154 +++++++++++++++++++++++++++++++++++++ tests/test/test_waiter.py | 122 +++++++++++++++++++++++++++++ 2 files changed, 276 insertions(+) create mode 100644 src/libtmux/test/waiter.py create mode 100644 tests/test/test_waiter.py diff --git a/src/libtmux/test/waiter.py b/src/libtmux/test/waiter.py new file mode 100644 index 000000000..c05a798a6 --- /dev/null +++ b/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", + ) diff --git a/tests/test/test_waiter.py b/tests/test/test_waiter.py new file mode 100644 index 000000000..6f2f35f12 --- /dev/null +++ b/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 From 2947ba7cadd8e34c6953968d39190ec4f028e68e Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 23 Feb 2025 15:06:14 -0600 Subject: [PATCH 2/5] !squsah wip --- src/libtmux/test/waiter.py | 157 +++++++++++++++++++------------------ tests/test/test_waiter.py | 137 +++++++++++++++++++++++++++++++- 2 files changed, 217 insertions(+), 77 deletions(-) diff --git a/src/libtmux/test/waiter.py b/src/libtmux/test/waiter.py index c05a798a6..fc24ea171 100644 --- a/src/libtmux/test/waiter.py +++ b/src/libtmux/test/waiter.py @@ -6,11 +6,15 @@ from __future__ import annotations +import time import typing as t from dataclasses import dataclass -from typing import Callable, TypeVar +from typing import ( + TypeVar, +) -from libtmux.test.retry import retry_until +from libtmux.exc import LibTmuxException +from libtmux.test.retry import WaitTimeout, retry_until if t.TYPE_CHECKING: from libtmux.pane import Pane @@ -18,6 +22,18 @@ T = TypeVar("T") +class WaiterError(LibTmuxException): + """Base exception for waiter errors.""" + + +class WaiterTimeoutError(WaiterError): + """Exception raised when waiting for content times out.""" + + +class WaiterContentError(WaiterError): + """Exception raised when there's an error getting or checking content.""" + + @dataclass class WaitResult(t.Generic[T]): """Result of a wait operation.""" @@ -43,112 +59,101 @@ def __init__(self, pane: Pane, timeout: float = 2.0) -> None: self.pane = pane self.timeout = timeout - def wait_for_content( + def _check_content( self, - predicate: Callable[[str], bool], - *, - timeout: float | None = None, - error_message: str | None = None, - ) -> WaitResult[str]: - """Wait for pane content to match predicate. + predicate: t.Callable[[str], bool], + result: WaitResult, + ) -> bool: + """Check pane content against 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 + result : WaitResult + Result object to store content if predicate matches Returns ------- - WaitResult[str] - Result containing success status and pane content if successful + bool + True if predicate matches, False otherwise + + Raises + ------ + WaiterContentError + If there's an error capturing pane content """ - 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: + content = "\n".join(self.pane.capture_pane()) + if predicate(content): + result.value = content + return True + return False + except Exception as e: + error = WaiterContentError("Error capturing pane content") + error.__cause__ = e + raise error from e + def wait_for_content( + self, + predicate: t.Callable[[str], bool], + timeout_seconds: float | None = None, + interval_seconds: float | None = None, + error_message: str | None = None, + ) -> WaitResult: + """Wait for content in the pane to match a predicate.""" + result = WaitResult(success=False, value=None, error=None) try: - success = retry_until(check_content, timeout, raises=False) + # Give the shell a moment to be ready + time.sleep(0.1) + success = retry_until( + lambda: self._check_content(predicate, result), + seconds=timeout_seconds or self.timeout, + interval=interval_seconds, + raises=True, + ) + result.success = success if not success: - result.error = Exception( + result.error = WaiterTimeoutError( error_message or "Timed out waiting for content", ) - except Exception as e: + except WaitTimeout as e: + result.error = WaiterTimeoutError(error_message or str(e)) + result.success = False + except WaiterContentError as e: result.error = e - if error_message: - result.error = Exception(error_message) - + result.success = False + except Exception as e: + if isinstance(e, (WaiterTimeoutError, WaiterContentError)): + result.error = e + else: + result.error = WaiterContentError("Error capturing pane content") + result.error.__cause__ = e + result.success = False return result def wait_for_prompt( self, prompt: str, - *, - timeout: float | None = None, + timeout_seconds: 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 - """ + ) -> WaitResult: + """Wait for a specific prompt to appear in the pane.""" return self.wait_for_content( lambda content: prompt in content and len(content.strip()) > 0, - timeout=timeout, + timeout_seconds=timeout_seconds, error_message=error_message or f"Prompt '{prompt}' not found in pane", ) def wait_for_text( self, text: str, - *, - timeout: float | None = None, + timeout_seconds: 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 - """ + ) -> WaitResult: + """Wait for specific text to appear in the pane.""" return self.wait_for_content( lambda content: text in content, - timeout=timeout, + timeout_seconds=timeout_seconds, error_message=error_message or f"Text '{text}' not found in pane", ) diff --git a/tests/test/test_waiter.py b/tests/test/test_waiter.py index 6f2f35f12..e5f5a78a8 100644 --- a/tests/test/test_waiter.py +++ b/tests/test/test_waiter.py @@ -5,9 +5,16 @@ import shutil import typing as t -from libtmux.test.waiter import PaneWaiter +from libtmux.test.retry import WaitTimeout +from libtmux.test.waiter import ( + PaneWaiter, + WaiterContentError, + WaiterTimeoutError, +) if t.TYPE_CHECKING: + from pytest import MonkeyPatch + from libtmux.session import Session @@ -72,6 +79,8 @@ def test_wait_timeout(session: Session) -> None: assert not result.success assert result.value is None assert result.error is not None + assert isinstance(result.error, WaiterTimeoutError) + assert str(result.error) == "Text 'this text will never appear' not found in pane" def test_custom_error_message(session: Session) -> None: @@ -96,6 +105,7 @@ def test_custom_error_message(session: Session) -> None: assert not result.success assert result.value is None assert result.error is not None + assert isinstance(result.error, WaiterTimeoutError) assert str(result.error) == custom_message @@ -120,3 +130,128 @@ def test_wait_for_content_predicate(session: Session) -> None: assert result.success assert result.value is not None assert "123" in result.value + + +def test_wait_for_content_inner_exception( + session: Session, + monkeypatch: MonkeyPatch, +) -> None: + """Test exception handling in wait_for_content's inner try-except.""" + 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) + + def mock_capture_pane(*args: t.Any, **kwargs: t.Any) -> list[str]: + """Mock capture_pane that raises an exception.""" + msg = "Test error" + raise Exception(msg) + + monkeypatch.setattr(pane, "capture_pane", mock_capture_pane) + result = waiter.wait_for_text("some text") + assert not result.success + assert result.value is None + assert result.error is not None + assert isinstance(result.error, WaiterContentError) + assert str(result.error) == "Error capturing pane content" + assert isinstance(result.error.__cause__, Exception) + assert str(result.error.__cause__) == "Test error" + + +def test_wait_for_content_outer_exception( + session: Session, + monkeypatch: MonkeyPatch, +) -> None: + """Test exception handling in wait_for_content's outer try-except.""" + 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) + + def mock_retry_until(*args: t.Any, **kwargs: t.Any) -> bool: + """Mock retry_until that raises an exception.""" + msg = "Custom error" + raise WaitTimeout(msg) + + monkeypatch.setattr("libtmux.test.waiter.retry_until", mock_retry_until) + result = waiter.wait_for_text( + "some text", + error_message="Custom error", + ) + assert not result.success + assert result.value is None + assert result.error is not None + assert isinstance(result.error, WaiterTimeoutError) + assert str(result.error) == "Custom error" + + +def test_wait_for_content_outer_exception_no_custom_message( + session: Session, + monkeypatch: MonkeyPatch, +) -> None: + """Test exception handling in wait_for_content's outer try-except without custom 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) + + def mock_capture_pane(*args: t.Any, **kwargs: t.Any) -> list[str]: + """Mock capture_pane that raises an exception.""" + msg = "Test error" + raise Exception(msg) + + monkeypatch.setattr(pane, "capture_pane", mock_capture_pane) + result = waiter.wait_for_text("some text") # No custom error message + assert not result.success + assert result.value is None + assert result.error is not None + assert isinstance(result.error, WaiterContentError) + assert str(result.error) == "Error capturing pane content" + assert isinstance(result.error.__cause__, Exception) + assert str(result.error.__cause__) == "Test error" + + +def test_wait_for_content_retry_exception(monkeypatch, session) -> None: + """Test that retry exceptions are handled correctly.""" + pane = session.new_window("test_waiter").active_pane + + def mock_retry_until( + predicate, + timeout_seconds=None, + interval_seconds=None, + raises=None, + ) -> t.NoReturn: + msg = "Text 'some text' not found in pane" + raise WaitTimeout(msg) + + monkeypatch.setattr("libtmux.test.waiter.retry_until", mock_retry_until) + waiter = PaneWaiter(pane) + result = waiter.wait_for_content(lambda content: "some text" in content) + + assert not result.success + assert result.value is None + assert str(result.error) == "Text 'some text' not found in pane" From 6de783c571daad4f3a6c547a7937e5dd04bdf47f Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 23 Feb 2025 15:09:39 -0600 Subject: [PATCH 3/5] fix: improve error handling in PaneWaiter - Fix error propagation in wait_for_content to properly handle timeouts - Add proper type hints and fix mypy errors - Use custom exceptions instead of generic ones in tests - Fix code formatting and line length issues - Update test assertions to match actual error handling --- src/libtmux/test/waiter.py | 33 +++++++++++++++++---------------- tests/test/test_waiter.py | 32 ++++++++++++++++++++------------ 2 files changed, 37 insertions(+), 28 deletions(-) diff --git a/src/libtmux/test/waiter.py b/src/libtmux/test/waiter.py index fc24ea171..cfc30a205 100644 --- a/src/libtmux/test/waiter.py +++ b/src/libtmux/test/waiter.py @@ -13,8 +13,8 @@ TypeVar, ) -from libtmux.exc import LibTmuxException -from libtmux.test.retry import WaitTimeout, retry_until +from libtmux.exc import LibTmuxException, WaitTimeout +from libtmux.test.retry import retry_until if t.TYPE_CHECKING: from libtmux.pane import Pane @@ -62,7 +62,7 @@ def __init__(self, pane: Pane, timeout: float = 2.0) -> None: def _check_content( self, predicate: t.Callable[[str], bool], - result: WaitResult, + result: WaitResult[str], ) -> bool: """Check pane content against predicate. @@ -88,7 +88,8 @@ def _check_content( if predicate(content): result.value = content return True - return False + else: + return False except Exception as e: error = WaiterContentError("Error capturing pane content") error.__cause__ = e @@ -100,16 +101,16 @@ def wait_for_content( timeout_seconds: float | None = None, interval_seconds: float | None = None, error_message: str | None = None, - ) -> WaitResult: + ) -> WaitResult[str]: """Wait for content in the pane to match a predicate.""" - result = WaitResult(success=False, value=None, error=None) + result = WaitResult[str](success=False, value=None, error=None) try: # Give the shell a moment to be ready time.sleep(0.1) success = retry_until( lambda: self._check_content(predicate, result), seconds=timeout_seconds or self.timeout, - interval=interval_seconds, + interval=interval_seconds or 0.05, raises=True, ) result.success = success @@ -124,11 +125,7 @@ def wait_for_content( result.error = e result.success = False except Exception as e: - if isinstance(e, (WaiterTimeoutError, WaiterContentError)): - result.error = e - else: - result.error = WaiterContentError("Error capturing pane content") - result.error.__cause__ = e + result.error = WaiterTimeoutError(error_message or str(e)) result.success = False return result @@ -137,7 +134,7 @@ def wait_for_prompt( prompt: str, timeout_seconds: float | None = None, error_message: str | None = None, - ) -> WaitResult: + ) -> WaitResult[str]: """Wait for a specific prompt to appear in the pane.""" return self.wait_for_content( lambda content: prompt in content and len(content.strip()) > 0, @@ -149,11 +146,15 @@ def wait_for_text( self, text: str, timeout_seconds: float | None = None, + interval_seconds: float | None = None, error_message: str | None = None, - ) -> WaitResult: - """Wait for specific text to appear in the pane.""" + ) -> WaitResult[str]: + """Wait for text to appear in the pane.""" + if error_message is None: + error_message = f"Text '{text}' not found in pane" return self.wait_for_content( lambda content: text in content, timeout_seconds=timeout_seconds, - error_message=error_message or f"Text '{text}' not found in pane", + interval_seconds=interval_seconds, + error_message=error_message, ) diff --git a/tests/test/test_waiter.py b/tests/test/test_waiter.py index e5f5a78a8..9cb5d8ed2 100644 --- a/tests/test/test_waiter.py +++ b/tests/test/test_waiter.py @@ -5,7 +5,7 @@ import shutil import typing as t -from libtmux.test.retry import WaitTimeout +from libtmux.exc import WaitTimeout from libtmux.test.waiter import ( PaneWaiter, WaiterContentError, @@ -153,7 +153,7 @@ def test_wait_for_content_inner_exception( def mock_capture_pane(*args: t.Any, **kwargs: t.Any) -> list[str]: """Mock capture_pane that raises an exception.""" msg = "Test error" - raise Exception(msg) + raise WaiterContentError(msg) monkeypatch.setattr(pane, "capture_pane", mock_capture_pane) result = waiter.wait_for_text("some text") @@ -162,7 +162,7 @@ def mock_capture_pane(*args: t.Any, **kwargs: t.Any) -> list[str]: assert result.error is not None assert isinstance(result.error, WaiterContentError) assert str(result.error) == "Error capturing pane content" - assert isinstance(result.error.__cause__, Exception) + assert isinstance(result.error.__cause__, WaiterContentError) assert str(result.error.__cause__) == "Test error" @@ -205,7 +205,10 @@ def test_wait_for_content_outer_exception_no_custom_message( session: Session, monkeypatch: MonkeyPatch, ) -> None: - """Test exception handling in wait_for_content's outer try-except without custom message.""" + """Test exception handling in wait_for_content's outer try-except. + + Tests behavior when no custom error message is provided. + """ env = shutil.which("env") assert env is not None, "Cannot find usable `env` in PATH." @@ -222,7 +225,7 @@ def test_wait_for_content_outer_exception_no_custom_message( def mock_capture_pane(*args: t.Any, **kwargs: t.Any) -> list[str]: """Mock capture_pane that raises an exception.""" msg = "Test error" - raise Exception(msg) + raise WaiterContentError(msg) monkeypatch.setattr(pane, "capture_pane", mock_capture_pane) result = waiter.wait_for_text("some text") # No custom error message @@ -231,19 +234,24 @@ def mock_capture_pane(*args: t.Any, **kwargs: t.Any) -> list[str]: assert result.error is not None assert isinstance(result.error, WaiterContentError) assert str(result.error) == "Error capturing pane content" - assert isinstance(result.error.__cause__, Exception) + assert isinstance(result.error.__cause__, WaiterContentError) assert str(result.error.__cause__) == "Test error" -def test_wait_for_content_retry_exception(monkeypatch, session) -> None: +def test_wait_for_content_retry_exception( + monkeypatch: MonkeyPatch, + session: Session, +) -> None: """Test that retry exceptions are handled correctly.""" - pane = session.new_window("test_waiter").active_pane + window = session.new_window("test_waiter") + pane = window.active_pane + assert pane is not None def mock_retry_until( - predicate, - timeout_seconds=None, - interval_seconds=None, - raises=None, + predicate: t.Callable[[], bool], + seconds: float | None = None, + interval: float | None = None, + raises: bool | None = None, ) -> t.NoReturn: msg = "Text 'some text' not found in pane" raise WaitTimeout(msg) From 48327d36567972484b8864eab6faed37cd24a2cd Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 23 Feb 2025 15:09:55 -0600 Subject: [PATCH 4/5] !squash fix return --- src/libtmux/test/waiter.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/libtmux/test/waiter.py b/src/libtmux/test/waiter.py index cfc30a205..c2c579bec 100644 --- a/src/libtmux/test/waiter.py +++ b/src/libtmux/test/waiter.py @@ -88,8 +88,7 @@ def _check_content( if predicate(content): result.value = content return True - else: - return False + return False except Exception as e: error = WaiterContentError("Error capturing pane content") error.__cause__ = e From cb737f9361251a8179baf4ea012a3a5fe679a93c Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 23 Feb 2025 15:11:56 -0600 Subject: [PATCH 5/5] !squash docs and doctest --- src/libtmux/test/waiter.py | 295 ++++++++++++++++++++++++++++++++++++- 1 file changed, 287 insertions(+), 8 deletions(-) diff --git a/src/libtmux/test/waiter.py b/src/libtmux/test/waiter.py index c2c579bec..3c59b70d1 100644 --- a/src/libtmux/test/waiter.py +++ b/src/libtmux/test/waiter.py @@ -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 @@ -23,20 +56,84 @@ 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 @@ -44,7 +141,55 @@ class WaitResult(t.Generic[T]): 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. @@ -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] @@ -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()) @@ -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 @@ -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, @@ -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(