!squash pytest examples

Author: tony

tests/pytest_examples/test_complex_layouts.py

@@ -11,7 +11,8 @@ def test_complex_layouts(session) -> None:
     import pytest
 
     pytest.skip(
-        "Skipping test_complex_layouts due to platform-specific and timing-sensitive issues",
+        "Skipping test_complex_layouts due to platform-specific "
+        "and timing-sensitive issues",
     )
 
     # Create a window with multiple panes in a specific layout

tests/pytest_examples/test_process_control.py

@@ -2,10 +2,10 @@
 
 from __future__ import annotations
 
-import time
-
 import pytest
 
+from libtmux.test.retry import retry_until
+
 
 @pytest.fixture
 def window(session):
@@ -19,26 +19,31 @@ def pane(window):
     pane = window.active_pane
     # Clear the pane at the start
     pane.send_keys("clear", enter=True)
-    time.sleep(0.2)
+
+    def pane_is_cleared() -> bool:
+        pane_contents = "\n".join(pane.capture_pane())
+        return "clear" not in pane_contents
+
+    retry_until(pane_is_cleared, 1)
     return pane
 
 
 def test_process_detection(pane) -> None:
     """Test detecting running processes in a pane."""
     # Start a long-running process
-    pane.send_keys("sleep 10 &", enter=True)  # Run in background
-    time.sleep(0.5)
+    pane.send_keys("sleep 10 &", enter=True)
 
     # Get the pane's TTY
     pane_tty = pane.cmd("display-message", "-p", "#{pane_tty}").stdout[0]
 
-    # Clear the pane
-    pane.send_keys("clear", enter=True)
-    time.sleep(0.5)
-
-    # Get information about the running process using system ps command
+    # Run ps command to list processes
     pane.send_keys(f"ps -t {pane_tty} -o pid,command | grep sleep", enter=True)
-    time.sleep(1)
+
+    def ps_output_contains_sleep() -> bool:
+        ps_output = pane.capture_pane()
+        return any("sleep 10" in line for line in ps_output)
+
+    retry_until(ps_output_contains_sleep, 3)
 
     # Capture the output
     ps_output = pane.capture_pane()
@@ -48,28 +53,68 @@ def test_process_detection(pane) -> None:
         f"Expected 'sleep 10' in: {ps_output}"
     )
 
+    # Clear the pane
+    pane.send_keys("clear", enter=True)
+
+    def pane_is_cleared() -> bool:
+        pane_contents = "\n".join(pane.capture_pane())
+        return "clear" not in pane_contents
+
+    retry_until(pane_is_cleared, 1)
+
     # Kill the process (find PID and kill it)
     pane.send_keys("pkill -f 'sleep 10'", enter=True)
-    time.sleep(1)
 
-    # Clear the pane again
-    pane.send_keys("clear", enter=True)
-    time.sleep(0.5)
+    # Run ps command again to verify process is gone
+    pane.send_keys(
+        f"ps -t {pane_tty} -o pid,command | grep sleep || echo 'Process not found'",
+        enter=True,
+    )
+
+    def process_is_killed() -> bool:
+        ps_output = pane.capture_pane()
+        return any("Process not found" in line for line in ps_output)
+
+    retry_until(process_is_killed, 3)
 
     # Verify the process has stopped
-    pane.send_keys(f"ps -t {pane_tty} -o pid,command | grep sleep", enter=True)
-    time.sleep(1)
     ps_output = pane.capture_pane()
-    assert not any("sleep 10" in line for line in ps_output), (
-        f"Found 'sleep 10' in: {ps_output}"
-    )
+
+    # Check if there's an actual running sleep process
+    # We need to filter out:
+    # 1. The command line (pkill -f 'sleep 10')
+    # 2. The grep command itself (grep sleep)
+    # 3. The termination notice ([1]  + terminated  sleep 10)
+    is_running = False
+    for line in ps_output:
+        # Look for a line with PID and 'sleep 10' without being a command or
+        # termination message
+        is_command_line = line.startswith("d%") or "grep sleep" in line
+        is_termination = "terminated" in line
+
+        if (
+            "sleep 10" in line
+            and not is_command_line
+            and not is_termination
+            and "Process not found" not in line
+        ):
+            is_running = True
+            break
+
+    assert not is_running, f"Found running 'sleep 10' in: {ps_output}"
 
 
 def test_command_output_scrollback(pane) -> None:
     """Test handling command output that exceeds visible pane height."""
     # Generate a lot of output
     pane.send_keys('for i in $(seq 1 100); do echo "Line $i"; done', enter=True)
-    time.sleep(1.0)
+
+    def output_generated() -> bool:
+        output = pane.capture_pane(start="-100")
+        lines_with_numbers = [line for line in output if "Line " in line]
+        return len(lines_with_numbers) > 50
+
+    retry_until(output_generated, 3)
 
     # Capture all the scrollback buffer
     output = pane.capture_pane(start="-100")
@@ -90,7 +135,12 @@ def test_command_output_scrollback(pane) -> None:
 
     # Clear the scrollback buffer
     pane.clear()
-    time.sleep(0.5)
+
+    def buffer_cleared() -> bool:
+        cleared_output = pane.capture_pane()
+        return len([line for line in cleared_output if line.strip()]) <= 1
+
+    retry_until(buffer_cleared, 1)
 
     # Verify the buffer is now empty or only has the prompt
     cleared_output = pane.capture_pane()
@@ -102,29 +152,34 @@ def test_running_background_process(pane) -> None:
     # Start a background process that writes to a file
     pane.send_keys("touch /tmp/background_test.txt", enter=True)
     pane.send_keys(
-        '(for i in $(seq 1 5); do echo "Update $i" >> /tmp/background_test.txt; sleep 1; done) &',
+        "(for i in $(seq 1 5); do "
+        "echo 'Update $i' >> /tmp/background_test.txt; "
+        "sleep 0.1; "
+        "done) &",
         enter=True,
     )
-    time.sleep(0.5)
 
     # Send a simple command with a unique string to verify responsiveness
     pane.send_keys("echo 'UNIQUE_BACKGROUND_STRING_789'", enter=True)
-    time.sleep(1.0)
 
-    # Check the output contains our test string
-    for _ in range(3):  # Try a few times if needed
+    def unique_string_in_output() -> bool:
         output = pane.capture_pane()
-        if any("UNIQUE_BACKGROUND_STRING_789" in line for line in output):
-            break
-        time.sleep(0.5)
+        return any("UNIQUE_BACKGROUND_STRING_789" in line for line in output)
+
+    retry_until(unique_string_in_output, 2)
 
     # Verify the command was executed
+    output = pane.capture_pane()
     assert any("UNIQUE_BACKGROUND_STRING_789" in line for line in output)
 
     # Check the file periodically
-    for _attempt in range(5):
-        pane.send_keys("cat /tmp/background_test.txt", enter=True)
-        time.sleep(1.0)
+    pane.send_keys("cat /tmp/background_test.txt", enter=True)
+
+    def updates_in_file() -> bool:
+        output = pane.capture_pane()
+        return any("Update" in line for line in output)
+
+    retry_until(updates_in_file, 3)
 
     # Verify we got at least some updates
     output = pane.capture_pane()