ControlLogs(fix): register control contexts in diagnostics

Author: tony

conftest.py

@@ -20,6 +20,7 @@
 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
@@ -123,7 +124,7 @@ def control_sandbox(
 def control_client_logs(
     control_sandbox: t.ContextManager[Server],
     tmp_path_factory: pytest.TempPathFactory,
-) -> t.Iterator[tuple[subprocess.Popen[str], pathlib.Path, pathlib.Path]]:
+) -> 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"
@@ -141,16 +142,22 @@ def control_client_logs(
         ]
         # 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=stdout_path.open("w+", buffering=1),
-            stderr=stderr_path.open("w+", buffering=1),
+            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, stdout_path, stderr_path
+            yield proc, proto
         finally:
             with contextlib.suppress(Exception):
                 if proc.stdin:

tests/test_control_client_logs.py

@@ -10,46 +10,63 @@
 
 
 @pytest.mark.engines(["control"])
-@pytest.mark.xfail(
-    reason="log file may be empty due to control client buffering",
-    strict=True,
-)
 def test_control_client_lists_clients(
-    control_client_logs: tuple[t.Any, t.Any, t.Any],
+    control_client_logs: tuple[t.Any, ControlProtocol],
 ) -> None:
     """Raw control client should report itself with control-mode flag."""
-    proc, stdout_path, _stderr_path = control_client_logs
+    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()
 
-    lines = stdout_path.read_text().splitlines()
-    assert any(len(line.split()) >= 2 and "C" in line.split()[1] for line in lines)
+    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)
+    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, t.Any, t.Any],
+    control_client_logs: tuple[t.Any, ControlProtocol],
 ) -> None:
     """Ensure ControlProtocol can parse raw stream from the logged control client."""
-    proc, stdout_path, _stderr_path = control_client_logs
+    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()
 
-    lines = stdout_path.read_text().splitlines()
-
-    proto = ControlProtocol()
-    ctx = CommandContext(argv=["tmux", "display-message", "-p", "hello"])
-    proto.register_command(ctx)
+    stdout_data, _ = proc.communicate(timeout=5)
 
-    # Feed lines; expect begin/payload/end
-    for line in lines:
-        proto.feed_line(line)
+    for line in stdout_data.splitlines():
+        proto.feed_line(line.rstrip("\n"))
 
-    # If tmux emitted begin/end, ctx should be done
-    if ctx.done.is_set():
-        result = proto.build_result(ctx)
-        assert "hello" in result.stdout
+    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)