diff --git a/src/libtmux/pane.py b/src/libtmux/pane.py index a60bb36f6..c921919f5 100644 --- a/src/libtmux/pane.py +++ b/src/libtmux/pane.py @@ -371,19 +371,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/legacy_api/test_pane.py b/tests/legacy_api/test_pane.py index 31200a9b9..5735e41bd 100644 --- a/tests/legacy_api/test_pane.py +++ b/tests/legacy_api/test_pane.py @@ -6,16 +6,61 @@ import shutil import typing as t +from libtmux.test.retry import retry_until + if t.TYPE_CHECKING: 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_resize_pane(session: Session) -> None: - """Test Pane.resize_pane().""" - window = session.attached_window - window.rename_window("test_resize_pane") + """Verify Pane.resize_pane().""" + window = setup_shell_window(session, "test_resize_pane") + pane = window.active_pane + assert pane is not None pane1 = window.attached_pane assert pane1 is not None @@ -32,15 +77,24 @@ def test_resize_pane(session: Session) -> None: def test_send_keys(session: Session) -> None: """Verify Pane.send_keys().""" - pane = session.attached_window.attached_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 + pane.send_keys("echo 'test'", literal=True) + + def wait_for_echo() -> bool: + try: + pane_contents = "\n".join(pane.capture_pane()) + return ( + "test" in pane_contents + and "echo 'test'" 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_echo, 2, raises=True) def test_set_height(session: Session) -> None: @@ -75,24 +129,9 @@ 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.attached_window.attached_pane + window = setup_shell_window(session, "test_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, - ) - pane_contents = "\n".join(pane.capture_pane()) - assert pane_contents == r'$ printf "\n%s\n" "Hello World !"{}'.format( - "\n\nHello World !\n$", - ) + assert "READY>" in pane_contents diff --git a/tests/legacy_api/test_session.py b/tests/legacy_api/test_session.py index c756999ea..76c9a5059 100644 --- a/tests/legacy_api/test_session.py +++ b/tests/legacy_api/test_session.py @@ -14,6 +14,7 @@ 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: @@ -264,6 +265,47 @@ def test_cmd_inserts_session_id(session: Session) -> None: assert cmd.cmd[-1] == last_arg +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 + + @pytest.mark.skipif( has_lt_version("3.0"), reason="needs -e flag for new-window which was introduced in 3.0", @@ -280,20 +322,25 @@ 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.attached_pane + 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( diff --git a/tests/legacy_api/test_window.py b/tests/legacy_api/test_window.py index 23668f45c..28ae83137 100644 --- a/tests/legacy_api/test_window.py +++ b/tests/legacy_api/test_window.py @@ -4,7 +4,6 @@ import logging import shutil -import time import typing as t import pytest @@ -13,6 +12,7 @@ from libtmux.common import has_gte_version, has_lt_version, has_version 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: @@ -389,6 +389,47 @@ def test_empty_window_name(session: Session) -> None: assert "''" in cmd.stdout +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 + + @pytest.mark.skipif( has_lt_version("3.0"), reason="needs -e flag for split-window which was introduced in 3.0", @@ -406,19 +447,36 @@ def test_split_window_with_environment( ) -> None: """Verify splitting window with environment variables.""" env = shutil.which("env") - assert env is not None, "Cannot find usable `env` in Path." + assert env is not None, "Cannot find usable `env` in PATH." - window = session.new_window(window_name="split_window_with_environment") - pane = window.split_window( - shell=f"{env} PS1='$ ' sh", + window = setup_shell_window(session, "split_with_environment") + pane = window.split( + 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) @pytest.mark.skipif( diff --git a/tests/test_pane.py b/tests/test_pane.py index 746467851..19c1e64ac 100644 --- a/tests/test_pane.py +++ b/tests/test_pane.py @@ -14,21 +14,90 @@ if t.TYPE_CHECKING: 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: @@ -65,46 +134,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: @@ -115,45 +197,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 @pytest.mark.skipif( @@ -326,9 +416,35 @@ def test_split_pane_size(session: Session) -> None: def test_pane_context_manager(session: Session) -> None: """Test Pane context manager functionality.""" - window = session.new_window() - with window.split() as pane: - pane.send_keys('echo "Hello"') + env = shutil.which("env") + assert env is not None, "Cannot find usable `env` in PATH." + + window = setup_shell_window(session, "test_context_manager") + with window.split(shell=f"{env} PROMPT_COMMAND='' PS1='READY>' sh") as pane: + # 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 assert len(window.panes) == 2 # Initial pane + new pane diff --git a/tests/test_session.py b/tests/test_session.py index 88d5f79e6..dae5448f1 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -12,9 +12,11 @@ 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: @@ -23,6 +25,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 @@ -328,20 +371,25 @@ 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( @@ -353,13 +401,9 @@ def test_new_window_with_environment_logs_warning_for_old_tmux( caplog: pytest.LogCaptureFixture, ) -> None: """Verify new window with environment vars create a warning if tmux is too old.""" - env = shutil.which("env") - assert env is not None, "Cannot find usable `env` in PATH." - - session.new_window( - attach=True, - window_name="window_with_environment", - window_shell=f"{env} PS1='$ ' sh", + setup_shell_window( + session, + "window_with_environment", environment={"ENV_VAR": "window"}, ) diff --git a/tests/test_window.py b/tests/test_window.py index 0be62613e..d938d9791 100644 --- a/tests/test_window.py +++ b/tests/test_window.py @@ -4,7 +4,6 @@ import logging import shutil -import time import typing as t import pytest @@ -19,6 +18,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: @@ -444,15 +444,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) @pytest.mark.skipif(