ControlMode(test): Add attach failure, flags, and retry placeholder

Author: tony

tests/test_control_client_logs.py

@@ -97,3 +97,37 @@ def test_control_client_notification_parsing(
     notif = proto.get_notification(timeout=0.1)
     assert notif is not None
     assert notif.kind.name in {"SESSION_CHANGED", "CLIENT_SESSION_CHANGED", "RAW"}
+
+
+@pytest.mark.engines(["control"])
+def test_control_client_lists_control_flag(
+    control_client_logs: tuple[t.Any, ControlProtocol],
+) -> None:
+    """list-clients should show control client with C flag on tmux >= 3.2."""
+    proc, proto = control_client_logs
+    if has_lt_version("3.2"):
+        pytest.skip("tmux < 3.2 omits client_flags")
+
+    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)
+    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)
+    assert any(
+        len(parts := line.split()) >= 2 and "C" in parts[1] for line in result.stdout
+    )

tests/test_control_mode_engine.py

@@ -298,3 +298,32 @@ def test_iter_notifications_survives_overflow(
     first = next(notif_iter, None)
     assert first is not None
     assert first.kind.name == "SESSIONS_CHANGED"
+
+
+class RestartRetryFixture(t.NamedTuple):
+    """Fixture for restart + retry behavior."""
+
+    test_id: str
+    raise_once: bool
+
+
+@pytest.mark.xfail(reason="Engine retry path not covered yet", strict=False)
+@pytest.mark.parametrize(
+    "case",
+    [
+        RestartRetryFixture(
+            test_id="retry_after_broken_pipe",
+            raise_once=True,
+        ),
+    ],
+    ids=lambda c: c.test_id,
+)
+def test_run_result_retries_after_broken_pipe(
+    case: RestartRetryFixture,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Placeholder: run_result should retry after broken pipe and succeed."""
+    engine = ControlModeEngine()
+    # TODO: Implement retry simulation when engine supports injectable I/O.
+    with pytest.raises(exc.ControlModeConnectionError):
+        engine.run("list-sessions")

tests/test_control_mode_regressions.py

@@ -692,12 +692,13 @@ def test_attach_to_existing_session(case: AttachFixture) -> None:
     socket_name = f"libtmux_test_{uuid.uuid4().hex[:8]}"
     bootstrap = Server(socket_name=socket_name)
     try:
-        # Create the target session via subprocess engine
-        bootstrap.new_session(
-            session_name=case.attach_to,
-            attach=False,
-            kill_session=True,
-        )
+        if case.expect_attached:
+            # Create the target session via subprocess engine
+            bootstrap.new_session(
+                session_name=case.attach_to,
+                attach=False,
+                kill_session=True,
+            )
         engine = ControlModeEngine(attach_to=case.attach_to)
         server = Server(socket_name=socket_name, engine=engine)
         if not case.expect_attached: