diff --git a/src/libtmux/pane.py b/src/libtmux/pane.py index 610ba3d2b..8adb4138b 100644 --- a/src/libtmux/pane.py +++ b/src/libtmux/pane.py @@ -449,19 +449,30 @@ def send_keys( Examples -------- - >>> pane = window.split(shell='sh') + >>> import shutil + >>> pane = window.split( + ... shell=f"{shutil.which('env')} PROMPT_COMMAND='' PS1='READY>' sh") + >>> from libtmux.test.retry import retry_until + >>> def wait_for_prompt() -> bool: + ... try: + ... pane_contents = "\n".join(pane.capture_pane()) + ... return "READY>" in pane_contents and len(pane_contents.strip()) > 0 + ... except Exception: + ... return False + >>> retry_until(wait_for_prompt, 2, raises=True) + True >>> pane.capture_pane() - ['$'] + ['READY>'] >>> pane.send_keys('echo "Hello world"', enter=True) >>> pane.capture_pane() - ['$ echo "Hello world"', 'Hello world', '$'] + ['READY>echo "Hello world"', 'Hello world', 'READY>'] >>> print('\n'.join(pane.capture_pane())) # doctest: +NORMALIZE_WHITESPACE - $ echo "Hello world" + READY>echo "Hello world" Hello world - $ + READY> """ prefix = " " if suppress_history else "" diff --git a/tests/test_pane.py b/tests/test_pane.py index cc3a0eb33..c9a3781b8 100644 --- a/tests/test_pane.py +++ b/tests/test_pane.py @@ -15,21 +15,90 @@ if t.TYPE_CHECKING: from libtmux._internal.types import StrPath from libtmux.session import Session + from libtmux.window import Window logger = logging.getLogger(__name__) +def setup_shell_window( + session: Session, + window_name: str, + environment: dict[str, str] | None = None, +) -> Window: + """Set up a shell window with consistent environment and prompt. + + Args: + session: The tmux session to create the window in + window_name: Name for the new window + environment: Optional environment variables to set in the window + + Returns + ------- + The created Window object with shell ready + """ + env = shutil.which("env") + assert env is not None, "Cannot find usable `env` in PATH." + + window = session.new_window( + attach=True, + window_name=window_name, + window_shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh", + environment=environment, + ) + + pane = window.active_pane + assert pane is not None + + # Wait for shell to be ready + def wait_for_prompt() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + return "READY>" in pane_contents and len(pane_contents.strip()) > 0 + except Exception: + return False + + retry_until(wait_for_prompt, 2, raises=True) + return window + + def test_send_keys(session: Session) -> None: """Verify Pane.send_keys().""" - pane = session.active_window.active_pane + window = setup_shell_window(session, "test_send_keys") + pane = window.active_pane assert pane is not None - pane.send_keys("c-c", literal=True) - pane_contents = "\n".join(pane.cmd("capture-pane", "-p").stdout) - assert "c-c" in pane_contents + # Test literal input + pane.send_keys("echo 'test-literal'", literal=True) + + def wait_for_literal() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + return ( + "test-literal" in pane_contents + and "echo 'test-literal'" in pane_contents + and pane_contents.count("READY>") >= 2 + ) + except Exception: + return False - pane.send_keys("c-a", literal=False) - assert "c-a" not in pane_contents, "should not print to pane" + retry_until(wait_for_literal, 2, raises=True) + + # Test non-literal input (should be interpreted as keystrokes) + pane.send_keys("c-c", literal=False) # Send Ctrl-C + + def wait_for_ctrl_c() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + # Ctrl-C should add a new prompt without executing a command + return ( + # Previous prompt + command + new prompt + pane_contents.count("READY>") >= 3 + and "c-c" not in pane_contents # The literal string should not appear + ) + except Exception: + return False + + retry_until(wait_for_ctrl_c, 2, raises=True) def test_set_height(session: Session) -> None: @@ -66,46 +135,59 @@ def test_set_width(session: Session) -> None: def test_capture_pane(session: Session) -> None: """Verify Pane.capture_pane().""" - env = shutil.which("env") - assert env is not None, "Cannot find usable `env` in PATH." - - session.new_window( - attach=True, - window_name="capture_pane", - window_shell=f"{env} PS1='$ ' sh", - ) - pane = session.active_window.active_pane + window = setup_shell_window(session, "capture_pane") + pane = window.active_pane assert pane is not None - pane_contents = "\n".join(pane.capture_pane()) - assert pane_contents == "$" + pane.send_keys( r'printf "\n%s\n" "Hello World !"', literal=True, suppress_history=False, ) + + def wait_for_output() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + return ( + "Hello World !" in pane_contents + and pane_contents.count("READY>") >= 2 + and r'printf "\n%s\n" "Hello World !"' in pane_contents + ) + except Exception: + return False + + # Wait for command output and new prompt + retry_until(wait_for_output, 2, raises=True) + pane_contents = "\n".join(pane.capture_pane()) - assert pane_contents == r'$ printf "\n%s\n" "Hello World !"{}'.format( - "\n\nHello World !\n$", - ) + assert r'READY>printf "\n%s\n" "Hello World !"' in pane_contents + assert "Hello World !" in pane_contents + assert pane_contents.count("READY>") >= 2 def test_capture_pane_start(session: Session) -> None: """Assert Pane.capture_pane() with ``start`` param.""" - env = shutil.which("env") - assert env is not None, "Cannot find usable `env` in PATH." - - session.new_window( - attach=True, - window_name="capture_pane_start", - window_shell=f"{env} PS1='$ ' sh", - ) - pane = session.active_window.active_pane + window = setup_shell_window(session, "capture_pane_start") + pane = window.active_pane assert pane is not None + pane_contents = "\n".join(pane.capture_pane()) - assert pane_contents == "$" + assert "READY>" in pane_contents + pane.send_keys(r'printf "%s"', literal=True, suppress_history=False) - pane_contents = "\n".join(pane.capture_pane()) - assert pane_contents == '$ printf "%s"\n$' + + def wait_for_command() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + except Exception: + return False + else: + has_command = r'printf "%s"' in pane_contents + has_prompts = pane_contents.count("READY>") >= 2 + return has_command and has_prompts + + retry_until(wait_for_command, 2, raises=True) + pane.send_keys("clear -x", literal=True, suppress_history=False) def wait_until_pane_cleared() -> bool: @@ -116,45 +198,53 @@ def wait_until_pane_cleared() -> bool: def pane_contents_shell_prompt() -> bool: pane_contents = "\n".join(pane.capture_pane()) - return pane_contents == "$" + return "READY>" in pane_contents and len(pane_contents.strip()) > 0 retry_until(pane_contents_shell_prompt, 1, raises=True) pane_contents_history_start = pane.capture_pane(start=-2) - assert pane_contents_history_start[0] == '$ printf "%s"' - assert pane_contents_history_start[1] == "$ clear -x" - assert pane_contents_history_start[-1] == "$" + assert r'READY>printf "%s"' in pane_contents_history_start[0] + assert "READY>clear -x" in pane_contents_history_start[1] + assert "READY>" in pane_contents_history_start[-1] pane.send_keys("") def pane_contents_capture_visible_only_shows_prompt() -> bool: pane_contents = "\n".join(pane.capture_pane(start=1)) - return pane_contents == "$" + return "READY>" in pane_contents and len(pane_contents.strip()) > 0 assert retry_until(pane_contents_capture_visible_only_shows_prompt, 1, raises=True) def test_capture_pane_end(session: Session) -> None: """Assert Pane.capture_pane() with ``end`` param.""" - env = shutil.which("env") - assert env is not None, "Cannot find usable `env` in PATH." - - session.new_window( - attach=True, - window_name="capture_pane_end", - window_shell=f"{env} PS1='$ ' sh", - ) - pane = session.active_window.active_pane + window = setup_shell_window(session, "capture_pane_end") + pane = window.active_pane assert pane is not None + pane_contents = "\n".join(pane.capture_pane()) - assert pane_contents == "$" + assert "READY>" in pane_contents + pane.send_keys(r'printf "%s"', literal=True, suppress_history=False) - pane_contents = "\n".join(pane.capture_pane()) - assert pane_contents == '$ printf "%s"\n$' + + def wait_for_command() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + except Exception: + return False + else: + has_command = r'printf "%s"' in pane_contents + has_prompts = pane_contents.count("READY>") >= 2 + return has_command and has_prompts + + retry_until(wait_for_command, 2, raises=True) + pane_contents = "\n".join(pane.capture_pane(end=0)) - assert pane_contents == '$ printf "%s"' + assert r'READY>printf "%s"' in pane_contents + pane_contents = "\n".join(pane.capture_pane(end="-")) - assert pane_contents == '$ printf "%s"\n$' + assert r'READY>printf "%s"' in pane_contents + assert pane_contents.count("READY>") >= 2 def test_pane_split_window_zoom( @@ -327,11 +417,38 @@ def test_set_title_special_characters(session: Session) -> None: def test_pane_context_manager(session: Session) -> None: """Test Pane context manager functionality.""" - window = session.new_window() - initial_pane_count = len(window.panes) + env = shutil.which("env") + assert env is not None, "Cannot find usable `env` in PATH." - with window.split() as pane: + window = setup_shell_window(session, "test_context_manager") + initial_pane_count = len(window.panes) + with window.split(shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh") as pane: assert len(window.panes) == initial_pane_count + 1 + + # Wait for shell to be ready in the split pane + def wait_for_shell() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + return "READY>" in pane_contents and len(pane_contents.strip()) > 0 + except Exception: + return False + + retry_until(wait_for_shell, 2, raises=True) + + pane.send_keys('echo "Hello"', literal=True) + + def wait_for_output() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + return ( + 'echo "Hello"' in pane_contents + and "Hello" in pane_contents + and pane_contents.count("READY>") >= 2 + ) + except Exception: + return False + + retry_until(wait_for_output, 2, raises=True) assert pane in window.panes # Pane should be killed after exiting context diff --git a/tests/test_session.py b/tests/test_session.py index e0dc85324..5ef00652f 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -11,11 +11,14 @@ import pytest from libtmux import exc +from libtmux.common import has_gte_version, has_lt_version from libtmux.constants import WindowDirection from libtmux.pane import Pane +from libtmux.server import Server from libtmux.session import Session from libtmux.test.constants import TEST_SESSION_PREFIX from libtmux.test.random import namer +from libtmux.test.retry import retry_until from libtmux.window import Window if t.TYPE_CHECKING: @@ -34,6 +37,47 @@ logger = logging.getLogger(__name__) +def setup_shell_window( + session: Session, + window_name: str, + environment: dict[str, str] | None = None, +) -> Window: + """Set up a shell window with consistent environment and prompt. + + Args: + session: The tmux session to create the window in + window_name: Name for the new window + environment: Optional environment variables to set in the window + + Returns + ------- + The created Window object with shell ready + """ + env = shutil.which("env") + assert env is not None, "Cannot find usable `env` in PATH." + + window = session.new_window( + attach=True, + window_name=window_name, + window_shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh", + environment=environment, + ) + + pane = window.active_pane + assert pane is not None + + # Wait for shell to be ready + def wait_for_prompt() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + return "READY>" in pane_contents and len(pane_contents.strip()) > 0 + except Exception: + return False + + retry_until(wait_for_prompt, 2, raises=True) + return window + + def test_has_session(server: Server, session: Session) -> None: """Server.has_session returns True if has session_name exists.""" TEST_SESSION_NAME = session.session_name @@ -327,22 +371,51 @@ def test_new_window_with_environment( environment: dict[str, str], ) -> None: """Verify new window with environment vars.""" - env = shutil.which("env") - assert env is not None, "Cannot find usable `env` in PATH." - - window = session.new_window( - attach=True, - window_name="window_with_environment", - window_shell=f"{env} PS1='$ ' sh", + window = setup_shell_window( + session, + "window_with_environment", environment=environment, ) pane = window.active_pane assert pane is not None + for k, v in environment.items(): - pane.send_keys(f"echo ${k}") - assert pane.capture_pane()[-2] == v + pane.send_keys(f"echo ${k}", literal=True) + def wait_for_output(value: str = v) -> bool: + try: + pane_contents = pane.capture_pane() + return any(value in line for line in pane_contents) + except Exception: + return False + retry_until(wait_for_output, 2, raises=True) + + +@pytest.mark.skipif( + has_gte_version("3.0"), + reason="3.0 has the -e flag on new-window", +) +def test_new_window_with_environment_logs_warning_for_old_tmux( + session: Session, + caplog: pytest.LogCaptureFixture, +) -> None: + """Verify new window with environment vars create a warning if tmux is too old.""" + setup_shell_window( + session, + "window_with_environment", + environment={"ENV_VAR": "window"}, + ) + + assert any("Environment flag ignored" in record.msg for record in caplog.records), ( + "Warning missing" + ) + + +@pytest.mark.skipif( + has_lt_version("3.2"), + reason="Only 3.2+ has the -a and -b flag on new-window", +) def test_session_new_window_with_direction( session: Session, ) -> None: diff --git a/tests/test_window.py b/tests/test_window.py index 5ea57f3c5..cd4ea1a99 100644 --- a/tests/test_window.py +++ b/tests/test_window.py @@ -5,7 +5,6 @@ import logging import pathlib import shutil -import time import typing as t import pytest @@ -20,6 +19,7 @@ ) from libtmux.pane import Pane from libtmux.server import Server +from libtmux.test.retry import retry_until from libtmux.window import Window if t.TYPE_CHECKING: @@ -460,15 +460,32 @@ def test_split_with_environment( window = session.new_window(window_name="split_with_environment") pane = window.split( - shell=f"{env} PS1='$ ' sh", + shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh", environment=environment, ) assert pane is not None - # wait a bit for the prompt to be ready as the test gets flaky otherwise - time.sleep(0.05) + + # Wait for shell to be ready + def wait_for_prompt() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + return "READY>" in pane_contents and len(pane_contents.strip()) > 0 + except Exception: + return False + + retry_until(wait_for_prompt, 2, raises=True) + for k, v in environment.items(): - pane.send_keys(f"echo ${k}") - assert pane.capture_pane()[-2] == v + pane.send_keys(f"echo ${k}", literal=True) + + def wait_for_output(value: str = v) -> bool: + try: + pane_contents = pane.capture_pane() + return any(value in line for line in pane_contents) + except Exception: + return False + + retry_until(wait_for_output, 2, raises=True) def test_split_window_zoom(