ControlMode(test): Add control-mode regression coverage and helpers

why: exercise control-mode engine behaviour, env propagation, logging, and sandbox usage.
what:
- add control sandbox fixture and wait_for_line helper
- integrate control engine into pytest markers/fixtures and adjust legacy env tests
- add protocol, regression, logging, and sandbox test suites plus tweaks to existing pane/server/session/window tests

Author: tony

conftest.py

@@ -10,12 +10,17 @@
 
 from __future__ import annotations
 
+import contextlib
+import pathlib
 import shutil
+import subprocess
 import typing as t
+import uuid
 
 import pytest
 from _pytest.doctest import DoctestItem
 
+from libtmux._internal.engines.control_protocol import CommandContext, ControlProtocol
 from libtmux.pane import Pane
 from libtmux.pytest_plugin import USING_ZSH
 from libtmux.server import Server
@@ -73,3 +78,123 @@ def setup_session(
     """Session-level test configuration for pytest."""
     if USING_ZSH:
         request.getfixturevalue("zshrc")
+
+
+# ---------------------------------------------------------------------------
+# Control-mode sandbox helper
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+@contextlib.contextmanager
+def control_sandbox(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> t.Iterator[Server]:
+    """Provide an isolated control-mode server for a test.
+
+    - Creates a unique tmux socket name per invocation
+    - Isolates HOME and TMUX_TMPDIR under a per-test temp directory
+    - Clears TMUX env var to avoid inheriting user sessions
+    - Uses ControlModeEngine; on exit, kills the server best-effort
+    """
+    socket_name = f"libtmux_test_{uuid.uuid4().hex[:8]}"
+    base = tmp_path_factory.mktemp("ctrl_sandbox")
+    home = base / "home"
+    tmux_tmpdir = base / "tmux"
+    home.mkdir()
+    tmux_tmpdir.mkdir()
+
+    monkeypatch.setenv("HOME", str(home))
+    monkeypatch.setenv("TMUX_TMPDIR", str(tmux_tmpdir))
+    monkeypatch.delenv("TMUX", raising=False)
+
+    from libtmux._internal.engines.control_mode import ControlModeEngine
+
+    server = Server(socket_name=socket_name, engine=ControlModeEngine())
+
+    try:
+        yield server
+    finally:
+        with contextlib.suppress(Exception):
+            server.kill()
+
+
+@pytest.fixture
+def control_client_logs(
+    control_sandbox: t.ContextManager[Server],
+    tmp_path_factory: pytest.TempPathFactory,
+) -> t.Iterator[tuple[subprocess.Popen[str], ControlProtocol]]:
+    """Spawn a raw tmux -C client against the sandbox and log stdout/stderr."""
+    base = tmp_path_factory.mktemp("ctrl_logs")
+    stdout_path = base / "control_stdout.log"
+    stderr_path = base / "control_stderr.log"
+
+    with control_sandbox as server:
+        cmd = [
+            "tmux",
+            "-L",
+            server.socket_name or "",
+            "-C",
+            "attach-session",
+            "-t",
+            "ctrltest",
+        ]
+        # Ensure ctrltest exists
+        server.cmd("new-session", "-d", "-s", "ctrltest")
+        stdout_path.open("w+", buffering=1)
+        stderr_f = stderr_path.open("w+", buffering=1)
+        proc = subprocess.Popen(
+            cmd,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=stderr_f,
+            text=True,
+            bufsize=1,
+        )
+        proto = ControlProtocol()
+        # tmux -C will emit a %begin/%end pair for this initial attach-session;
+        # queue a matching context so the parser has a pending command.
+        proto.register_command(CommandContext(argv=list(cmd)))
+        try:
+            yield proc, proto
+        finally:
+            with contextlib.suppress(Exception):
+                if proc.stdin:
+                    proc.stdin.write("kill-session -t ctrltest\n")
+                    proc.stdin.flush()
+            proc.terminate()
+            proc.wait(timeout=2)
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    """Add CLI options for selecting tmux engine."""
+    parser.addoption(
+        "--engine",
+        action="store",
+        default="subprocess",
+        choices=["subprocess", "control"],
+        help="Select tmux engine for fixtures (default: subprocess).",
+    )
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    """Register custom markers."""
+    config.addinivalue_line(
+        "markers",
+        (
+            "engines(names): run the test once for each engine in 'names' "
+            "(e.g. ['control', 'subprocess'])."
+        ),
+    )
+
+
+def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
+    """Parametrize engine_name when requested by tests."""
+    if "engine_name" in metafunc.fixturenames:
+        marker = metafunc.definition.get_closest_marker("engines")
+        if marker:
+            params = list(marker.args[0])
+        else:
+            params = [metafunc.config.getoption("--engine")]
+        metafunc.parametrize("engine_name", params, indirect=True)

tests/helpers.py

@@ -0,0 +1,30 @@
+"""Test helpers for control-mode flakiness handling."""
+
+from __future__ import annotations
+
+import time
+import typing as t
+
+from libtmux.pane import Pane
+
+
+def wait_for_line(
+    pane: Pane,
+    predicate: t.Callable[[str], bool],
+    *,
+    timeout: float = 1.0,
+    interval: float = 0.05,
+) -> list[str]:
+    """Poll capture_pane until a line satisfies ``predicate``.
+
+    Returns the final capture buffer (may be empty if timeout elapses).
+    """
+    deadline = time.monotonic() + timeout
+    last: list[str] = []
+    while time.monotonic() < deadline:
+        captured = pane.capture_pane()
+        last = [captured] if isinstance(captured, str) else list(captured)
+        if any(predicate(line) for line in last):
+            break
+        time.sleep(interval)
+    return last

tests/legacy_api/test_window.py

@@ -4,7 +4,6 @@
 
 import logging
 import shutil
-import time
 import typing as t
 
 import pytest
@@ -382,8 +381,6 @@ def test_split_window_with_environment(
         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)
     for k, v in environment.items():
         pane.send_keys(f"echo ${k}")
         assert pane.capture_pane()[-2] == v

tests/test/test_retry.py

@@ -29,7 +29,7 @@ def call_me_three_times() -> bool:
 
     end = time()
 
-    assert 0.9 <= (end - ini) <= 1.1  # Allow for small timing variations
+    assert 0.9 <= (end - ini) <= 1.3  # Allow for small timing variations
 
 
 def test_function_times_out() -> None:
@@ -47,7 +47,7 @@ def never_true() -> bool:
 
     end = time()
 
-    assert 0.9 <= (end - ini) <= 1.1  # Allow for small timing variations
+    assert 0.9 <= (end - ini) <= 1.3  # Allow for small timing variations
 
 
 def test_function_times_out_no_raise() -> None:

tests/test_control_client_logs.py

@@ -0,0 +1,76 @@
+"""Diagnostic tests using raw control client logs."""
+
+from __future__ import annotations
+
+import typing as t
+
+import pytest
+
+from libtmux._internal.engines.control_protocol import CommandContext, ControlProtocol
+from libtmux.common import has_lt_version
+
+
+@pytest.mark.engines(["control"])
+def test_control_client_lists_clients(
+    control_client_logs: tuple[t.Any, ControlProtocol],
+) -> None:
+    """Raw control client should report itself with control-mode flag."""
+    proc, proto = control_client_logs
+
+    assert proc.stdin is not None
+    list_ctx = CommandContext(
+        argv=[
+            "tmux",
+            "list-clients",
+            "-F",
+            "#{client_pid} #{client_flags} #{session_name}",
+        ],
+    )
+    proto.register_command(list_ctx)
+    detach_ctx = CommandContext(argv=["tmux", "detach-client"])
+    proto.register_command(detach_ctx)
+    proc.stdin.write('list-clients -F"#{client_pid} #{client_flags} #{session_name}"\n')
+    proc.stdin.write("detach-client\n")
+    proc.stdin.flush()
+
+    stdout_data, _ = proc.communicate(timeout=5)
+    for line in stdout_data.splitlines():
+        proto.feed_line(line.rstrip("\n"))
+
+    assert list_ctx.done.wait(timeout=0.5)
+    result = proto.build_result(list_ctx)
+    if has_lt_version("3.2"):
+        pytest.xfail("tmux < 3.2 omits client_flags field in list-clients")
+
+    saw_control_flag = any(
+        len(parts := line.split()) >= 2
+        and ("C" in parts[1] or "control-mode" in parts[1])
+        for line in result.stdout
+    )
+    assert saw_control_flag
+
+
+@pytest.mark.engines(["control"])
+def test_control_client_capture_stream_parses(
+    control_client_logs: tuple[t.Any, ControlProtocol],
+) -> None:
+    """Ensure ControlProtocol can parse raw stream from the logged control client."""
+    proc, proto = control_client_logs
+    assert proc.stdin is not None
+
+    display_ctx = CommandContext(argv=["tmux", "display-message", "-p", "hello"])
+    proto.register_command(display_ctx)
+    detach_ctx = CommandContext(argv=["tmux", "detach-client"])
+    proto.register_command(detach_ctx)
+    proc.stdin.write("display-message -p hello\n")
+    proc.stdin.write("detach-client\n")
+    proc.stdin.flush()
+
+    stdout_data, _ = proc.communicate(timeout=5)
+
+    for line in stdout_data.splitlines():
+        proto.feed_line(line.rstrip("\n"))
+
+    assert display_ctx.done.wait(timeout=0.5)
+    result = proto.build_result(display_ctx)
+    assert "hello" in result.stdout or "hello" in "".join(result.stdout)