refactor(tests[waiter]): Add waiter test examples into individual files

- Each test file focuses on a single feature or concept of the waiter module - Added descriptive docstrings to all test functions for better documentation - Created conftest.py with session fixture for waiter examples - Added helpers.py with utility functions for the test examples - Test files now follow a consistent naming convention for easier reference - Each test file is self-contained and demonstrates a single concept - All tests are marked with @pytest.mark.example for filtering This restructuring supports the documentation update to use literalinclude directives, making the documentation more maintainable and ensuring it stays in sync with actual code.
tmux-python · tony · Feb 27, 2025 · Feb 27, 2025 · Feb 27, 2025 · Feb 27, 2025
commit 17d2967a15c792ec1bd4891f4d14a900ab0bfffd
diff --git a/tests/examples/_internal/waiter/conftest.py b/tests/examples/_internal/waiter/conftest.py
@@ -0,0 +1,40 @@
+"""Pytest configuration for waiter examples."""
+
+from __future__ import annotations
+
+import contextlib
+from typing import TYPE_CHECKING
+
+import pytest
+
+from libtmux import Server
+
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
+    from libtmux.session import Session
+
+
+@pytest.fixture
+def session() -> Generator[Session, None, None]:
+    """Provide a tmux session for tests.
+
+    This fixture creates a new session specifically for the waiter examples,
+    and ensures it's properly cleaned up after the test.
+    """
+    server = Server()
+    session_name = "waiter_example_tests"
+
+    # Clean up any existing session with this name
+    with contextlib.suppress(Exception):
+        # Instead of using deprecated methods, use more direct approach
+        server.cmd("kill-session", "-t", session_name)
+
+    # Create a new session
+    session = server.new_session(session_name=session_name)
+
+    yield session
+
+    # Clean up
+    with contextlib.suppress(Exception):
+        session.kill()
diff --git a/tests/examples/_internal/waiter/helpers.py b/tests/examples/_internal/waiter/helpers.py
@@ -0,0 +1,55 @@
+"""Helper utilities for waiter tests."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from libtmux.pane import Pane
+    from libtmux.window import Window
+
+
+def ensure_pane(pane: Pane | None) -> Pane:
+    """Ensure that a pane is not None.
+
+    This helper is needed for type safety in the examples.
+
+    Args:
+        pane: The pane to check
+
+    Returns
+    -------
+        The pane if it's not None
+
+    Raises
+    ------
+        ValueError: If the pane is None
+    """
+    if pane is None:
+        msg = "Pane cannot be None"
+        raise ValueError(msg)
+    return pane
+
+
+def send_keys(pane: Pane | None, keys: str) -> None:
+    """Send keys to a pane after ensuring it's not None.
+
+    Args:
+        pane: The pane to send keys to
+        keys: The keys to send
+
+    Raises
+    ------
+        ValueError: If the pane is None
+    """
+    ensure_pane(pane).send_keys(keys)
+
+
+def kill_window_safely(window: Window | None) -> None:
+    """Kill a window if it's not None.
+
+    Args:
+        window: The window to kill
+    """
+    if window is not None:
+        window.kill()
diff --git a/tests/examples/_internal/waiter/test_custom_predicate.py b/tests/examples/_internal/waiter/test_custom_predicate.py
@@ -0,0 +1,40 @@
+"""Example of using a custom predicate function for matching."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from libtmux._internal.waiter import ContentMatchType, wait_for_pane_content
+
+if TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+@pytest.mark.example
+def test_custom_predicate(session: Session) -> None:
+    """Demonstrate using a custom predicate function for matching."""
+    window = session.new_window(window_name="test_custom_predicate")
+    pane = window.active_pane
+    assert pane is not None
+
+    # Send multiple lines of output
+    pane.send_keys("echo 'line 1'")
+    pane.send_keys("echo 'line 2'")
+    pane.send_keys("echo 'line 3'")
+
+    # Define a custom predicate function
+    def check_content(lines):
+        return len(lines) >= 3 and "error" not in "".join(lines).lower()
+
+    # Use the custom predicate
+    result = wait_for_pane_content(
+        pane,
+        check_content,
+        match_type=ContentMatchType.PREDICATE,
+    )
+    assert result.success
+
+    # Cleanup
+    window.kill()
diff --git a/tests/examples/_internal/waiter/test_fluent_basic.py b/tests/examples/_internal/waiter/test_fluent_basic.py
@@ -0,0 +1,30 @@
+"""Example of using the fluent API in libtmux waiters."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from libtmux._internal.waiter import expect
+
+if TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+@pytest.mark.example
+def test_fluent_basic(session: Session) -> None:
+    """Demonstrate basic usage of the fluent API."""
+    window = session.new_window(window_name="test_fluent_basic")
+    pane = window.active_pane
+    assert pane is not None
+
+    # Send a command
+    pane.send_keys("echo 'hello world'")
+
+    # Basic usage of the fluent API
+    result = expect(pane).wait_for_text("hello world")
+    assert result.success
+
+    # Cleanup
+    window.kill()
diff --git a/tests/examples/_internal/waiter/test_fluent_chaining.py b/tests/examples/_internal/waiter/test_fluent_chaining.py
@@ -0,0 +1,36 @@
+"""Example of method chaining with the fluent API in libtmux waiters."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from libtmux._internal.waiter import expect
+
+if TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+@pytest.mark.example
+def test_fluent_chaining(session: Session) -> None:
+    """Demonstrate method chaining wi
A3D4
th the fluent API."""
+    window = session.new_window(window_name="test_fluent_chaining")
+    pane = window.active_pane
+    assert pane is not None
+
+    # Send a command
+    pane.send_keys("echo 'completed successfully'")
+
+    # With method chaining
+    result = (
+        expect(pane)
+        .with_timeout(5.0)
+        .with_interval(0.1)
+        .without_raising()
+        .wait_for_text("completed successfully")
+    )
+    assert result.success
+
+    # Cleanup
+    window.kill()
diff --git a/tests/examples/_internal/waiter/test_mixed_pattern_types.py b/tests/examples/_internal/waiter/test_mixed_pattern_types.py
@@ -0,0 +1,44 @@
+"""Example of using different pattern types and match types."""
+
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING
+
+import pytest
+
+from libtmux._internal.waiter import ContentMatchType, wait_for_any_content
+
+if TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+@pytest.mark.example
+def test_mixed_pattern_types(session: Session) -> None:
+    """Demonstrate using different pattern types and match types."""
+    window = session.new_window(window_name="test_mixed_patterns")
+    pane = window.active_pane
+    assert pane is not None
+
+    # Send commands that will match different patterns
+    pane.send_keys("echo 'exact match'")
+    pane.send_keys("echo '10 items found'")
+
+    # Create a predicate function
+    def has_enough_lines(lines):
+        return len(lines) >= 2
+
+    # Wait for any of these patterns with different match types
+    result = wait_for_any_content(
+        pane,
+        [
+            "exact match",  # String for exact match
+            re.compile(r"\d+ items found"),  # Regex pattern
+            has_enough_lines,  # Predicate function
+        ],
+        [ContentMatchType.EXACT, ContentMatchType.REGEX, ContentMatchType.PREDICATE],
+    )
+    assert result.success
+
+    # Cleanup
+    window.kill()
diff --git a/tests/examples/_internal/waiter/test_timeout_handling.py b/tests/examples/_internal/waiter/test_timeout_handling.py
@@ -0,0 +1,40 @@
+"""Example of timeout handling with libtmux waiters."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from libtmux._internal.waiter import wait_for_pane_content
+
+if TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+@pytest.mark.example
+def test_timeout_handling(session: Session) -> None:
+    """Demonstrate handling timeouts gracefully without exceptions."""
+    window = session.new_window(window_name="test_timeout")
+    pane = window.active_pane
+    assert pane is not None
+
+    # Clear the pane
+    pane.send_keys("clear")
+
+    # Handle timeouts gracefully without exceptions
+    # Looking for content that won't appear (with a short timeout)
+    result = wait_for_pane_content(
+        pane,
+        "this text will not appear",
+        timeout=0.5,
+        raises=False,
+    )
+
+    # Should not raise an exception
+    assert not result.success
+    assert result.error is not None
+    assert "Timed out" in result.error
+
+    # Cleanup
+    window.kill()
diff --git a/tests/examples/_internal/waiter/test_wait_for_all_content.py b/tests/examples/_internal/waiter/test_wait_for_all_content.py
@@ -0,0 +1,41 @@
+"""Example of waiting for all conditions to be met."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+import pytest
+
+from libtmux._internal.waiter import ContentMatchType, wait_for_all_content
+
+if TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+@pytest.mark.example
+def test_wait_for_all_content(session: Session) -> None:
+    """Demonstrate waiting for all conditions to be met."""
+    window = session.new_window(window_name="test_all_content")
+    pane = window.active_pane
+    assert pane is not None
+
+    # Send commands with both required phrases
+    pane.send_keys("echo 'Database connected'")
+    pane.send_keys("echo 'Server started'")
+
+    # Wait for all conditions to be true
+    result = wait_for_all_content(
+        pane,
+        ["Database connected", "Server started"],
+        ContentMatchType.CONTAINS,
+    )
+    assert result.success
+    # For wait_for_all_content, the matched_content will be a list of matched patterns
+    assert result.matched_content is not None
+    matched_content = cast("list[str]", result.matched_content)
+    assert len(matched_content) == 2
+    assert "Database connected" in matched_content
+    assert "Server started" in matched_content
+
+    # Cleanup
+    window.kill()
diff --git a/tests/examples/_internal/waiter/test_wait_for_any_content.py b/tests/examples/_internal/waiter/test_wait_for_any_content.py
@@ -0,0 +1,36 @@
+"""Example of waiting for any of multiple conditions."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from libtmux._internal.waiter import ContentMatchType, wait_for_any_content
+
+if TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+@pytest.mark.example
+def test_wait_for_any_content(session: Session) -> None:
+    """Demonstrate waiting for any of multiple conditions."""
+    window = session.new_window(window_name="test_any_content")
+    pane = window.active_pane
+    assert pane is not None
+
+    # Send a command
+    pane.send_keys("echo 'Success'")
+
+    # Wait for any of these patterns
+    result = wait_for_any_content(
+        pane,
+        ["Success", "Error:", "timeout"],
+        ContentMatchType.CONTAINS,
+    )
+    assert result.success
+    assert result.matched_content == "Success"
+    assert result.matched_pattern_index == 0
+
+    # Cleanup
+    window.kill()