ControlModeEngine(test[live]): exercise flags/flow/subscriptions

why: ensure control-mode helpers behave against real tmux clients.
what:
- add live control-mode tests for set_client_flags, set_pane_flow, and subscribe
- run control engine inside sandbox sessions to verify client flags and pane flow usability
- drop redundant stub-based cases from unit suite

Author: tony

tests/test_control_mode_engine.py

@@ -323,231 +323,20 @@ def test_iter_notifications_survives_overflow(
     assert first.kind.name == "SESSIONS_CHANGED"
 
 
-class SetClientFlagsCase(t.NamedTuple):
-    """Fixture for refresh-client flag construction."""
-
-    test_id: str
-    kwargs: dict[str, t.Any]
-    expected_flags: set[str]
-    expect_run: bool
-
-
-@pytest.mark.parametrize(
-    "case",
-    [
-        SetClientFlagsCase(
-            test_id="enable_no_output_with_pause",
-            kwargs={"no_output": True, "pause_after": 1},
-            expected_flags={"no-output", "pause-after=1"},
-            expect_run=True,
-        ),
-        SetClientFlagsCase(
-            test_id="disable_no_output_clear_pause",
-            kwargs={"no_output": False, "pause_after": 0},
-            expected_flags={"!no-output", "!pause-after"},
-            expect_run=True,
-        ),
-        SetClientFlagsCase(
-            test_id="wait_exit_and_read_only",
-            kwargs={"wait_exit": True, "read_only": True},
-            expected_flags={"wait-exit", "read-only"},
-            expect_run=True,
-        ),
-        SetClientFlagsCase(
-            test_id="clear_wait_exit",
-            kwargs={"wait_exit": False},
-            expected_flags={"!wait-exit"},
-            expect_run=True,
-        ),
-        SetClientFlagsCase(
-            test_id="toggle_misc_flags",
-            kwargs={"ignore_size": True, "active_pane": False},
-            expected_flags={"ignore-size", "!active-pane"},
-            expect_run=True,
-        ),
-        SetClientFlagsCase(
-            test_id="noop_when_no_flags",
-            kwargs={},
-            expected_flags=set(),
-            expect_run=False,
-        ),
-    ],
-    ids=lambda c: c.test_id,
-)
-def test_set_client_flags_builds_refresh_client(case: SetClientFlagsCase) -> None:
-    """set_client_flags should call refresh-client with correct flag string."""
-    engine = ControlModeEngine(start_threads=False)
-    calls: list[tuple[str, tuple[str, ...], tuple[str, ...]]] = []
-
-    class DummyCmd:
-        stdout: t.ClassVar[list[str]] = []
-        stderr: t.ClassVar[list[str]] = []
-        returncode: t.ClassVar[int] = 0
-
-    def fake_run(
-        cmd: str,
-        cmd_args: t.Sequence[str | int] | None = None,
-        server_args: t.Sequence[str | int] | None = None,
-        timeout: float | None = None,
-    ) -> DummyCmd:
-        cmd_args_tuple = tuple(str(a) for a in (cmd_args or ()))
-        server_args_tuple = tuple(str(a) for a in (server_args or ()))
-        calls.append((cmd, cmd_args_tuple, server_args_tuple))
-        return DummyCmd()
-
-    engine.run = fake_run  # type: ignore[assignment]
-
-    engine.set_client_flags(**case.kwargs)
-
-    if not case.expect_run:
-        assert calls == []
-        return
-
-    assert len(calls) == 1
-    cmd, cmd_args, server_args = calls[0]
-    assert cmd == "refresh-client"
-    assert cmd_args and cmd_args[0] == "-f"
-    flags_str = cmd_args[1] if len(cmd_args) > 1 else ""
-    for flag in case.expected_flags:
-        assert flag in flags_str
-    assert server_args == ()
-
-
 def test_set_client_flags_rejects_negative_pause() -> None:
     """pause_after must be non-negative."""
     engine = ControlModeEngine(start_threads=False)
     with pytest.raises(ValueError):
         engine.set_client_flags(pause_after=-1)
 
 
-class PaneFlowCase(t.NamedTuple):
-    """Fixture for refresh-client -A flow control."""
-
-    test_id: str
-    pane_id: str | int
-    state: str
-    expected_arg: str
-
-
-@pytest.mark.parametrize(
-    "case",
-    [
-        PaneFlowCase(
-            test_id="resume_default",
-            pane_id="%1",
-            state="continue",
-            expected_arg="%1:continue",
-        ),
-        PaneFlowCase(
-            test_id="pause_pane",
-            pane_id=3,
-            state="pause",
-            expected_arg="3:pause",
-        ),
-    ],
-    ids=lambda c: c.test_id,
-)
-def test_set_pane_flow_builds_refresh_client(case: PaneFlowCase) -> None:
-    """set_pane_flow should build refresh-client -A args."""
-    engine = ControlModeEngine(start_threads=False)
-    calls: list[tuple[str, tuple[str, ...], tuple[str, ...]]] = []
-
-    class DummyCmd:
-        stdout: t.ClassVar[list[str]] = []
-        stderr: t.ClassVar[list[str]] = []
-        returncode: t.ClassVar[int] = 0
-
-    def fake_run(
-        cmd: str,
-        cmd_args: t.Sequence[str | int] | None = None,
-        server_args: t.Sequence[str | int] | None = None,
-        timeout: float | None = None,
-    ) -> DummyCmd:
-        cmd_args_tuple = tuple(str(a) for a in (cmd_args or ()))
-        server_args_tuple = tuple(str(a) for a in (server_args or ()))
-        calls.append((cmd, cmd_args_tuple, server_args_tuple))
-        return DummyCmd()
-
-    engine.run = fake_run  # type: ignore[assignment]
-
-    engine.set_pane_flow(case.pane_id, state=case.state)
-
-    assert calls
-    cmd, cmd_args, server_args = calls[0]
-    assert cmd == "refresh-client"
-    assert cmd_args == ("-A", case.expected_arg)
-    assert server_args == ()
-
-
 def test_set_pane_flow_validates_state() -> None:
     """Invalid flow state should raise."""
     engine = ControlModeEngine(start_threads=False)
     with pytest.raises(ValueError):
         engine.set_pane_flow("%1", state="bad")
 
 
-class SubscribeCase(t.NamedTuple):
-    """Fixture for refresh-client -B subscription arguments."""
-
-    test_id: str
-    name: str
-    what: str | None
-    format: str | None
-    expected_args: tuple[str, ...]
-
-
-@pytest.mark.parametrize(
-    "case",
-    [
-        SubscribeCase(
-            test_id="add_subscription",
-            name="focus",
-            what="%1",
-            format="#{pane_active}",
-            expected_args=("-B", "focus:%1:#{pane_active}"),
-        ),
-        SubscribeCase(
-            test_id="remove_subscription",
-            name="focus",
-            what=None,
-            format=None,
-            expected_args=("-B", "focus"),
-        ),
-    ],
-    ids=lambda c: c.test_id,
-)
-def test_subscribe_builds_refresh_client(case: SubscribeCase) -> None:
-    """Subscribe should wrap refresh-client -B calls."""
-    engine = ControlModeEngine(start_threads=False)
-    calls: list[tuple[str, tuple[str, ...], tuple[str, ...]]] = []
-
-    class DummyCmd:
-        stdout: t.ClassVar[list[str]] = []
-        stderr: t.ClassVar[list[str]] = []
-        returncode: t.ClassVar[int] = 0
-
-    def fake_run(
-        cmd: str,
-        cmd_args: t.Sequence[str | int] | None = None,
-        server_args: t.Sequence[str | int] | None = None,
-        timeout: float | None = None,
-    ) -> DummyCmd:
-        cmd_args_tuple = tuple(str(a) for a in (cmd_args or ()))
-        server_args_tuple = tuple(str(a) for a in (server_args or ()))
-        calls.append((cmd, cmd_args_tuple, server_args_tuple))
-        return DummyCmd()
-
-    engine.run = fake_run  # type: ignore[assignment]
-
-    engine.subscribe(case.name, what=case.what, fmt=case.format)
-
-    assert calls
-    cmd, cmd_args, server_args = calls[0]
-    assert cmd == "refresh-client"
-    assert cmd_args == case.expected_args
-    assert server_args == ()
-
-
 class ScriptedStdin:
     """Fake stdin that can optionally raise BrokenPipeError on write."""
 

tests/test_control_mode_live.py

@@ -0,0 +1,148 @@
+"""Live control-mode functional tests (no fakes)."""
+
+from __future__ import annotations
+
+import typing as t
+from typing import NamedTuple
+
+import pytest
+
+from libtmux._internal.engines.control_mode import ControlModeEngine
+from libtmux.server import Server
+from tests.helpers import wait_for_line
+
+
+class ClientFlagCase(NamedTuple):
+    """Fixture for exercising set_client_flags against real tmux."""
+
+    test_id: str
+    kwargs: dict[str, t.Any]
+    present: set[str]
+    absent: frozenset[str] = frozenset()
+
+
+CLIENT_FLAG_CASES: list[ClientFlagCase] = [
+    ClientFlagCase(
+        test_id="enable_no_output_pause",
+        kwargs={"no_output": True, "pause_after": 1},
+        present={"no-output", "pause-after=1"},
+    ),
+    ClientFlagCase(
+        test_id="clear_no_output_pause",
+        kwargs={"no_output": False, "pause_after": 0, "wait_exit": False},
+        present=set(),
+        absent=frozenset({"no-output", "pause-after", "pause-after=1", "wait-exit"}),
+    ),
+    ClientFlagCase(
+        test_id="enable_wait_exit",
+        kwargs={"wait_exit": True},
+        present={"wait-exit"},
+    ),
+    ClientFlagCase(
+        test_id="enable_active_pane",
+        kwargs={"active_pane": True},
+        present={"active-pane"},
+    ),
+]
+CLIENT_FLAG_IDS = [case.test_id for case in CLIENT_FLAG_CASES]
+
+
+@pytest.mark.engines(["control"])
+@pytest.mark.parametrize("case", CLIENT_FLAG_CASES, ids=CLIENT_FLAG_IDS)
+def test_set_client_flags_live(
+    case: ClientFlagCase,
+    control_sandbox: t.ContextManager[Server],
+) -> None:
+    """set_client_flags should actually toggle tmux client flags."""
+    with control_sandbox as server:
+        engine = t.cast(ControlModeEngine, server.engine)
+        engine.set_client_flags(**case.kwargs)
+
+        flags_line = server.cmd("list-clients", "-F", "#{client_flags}").stdout
+        assert flags_line
+        flags = set(flags_line[0].split(","))
+
+        for flag in case.present:
+            assert flag in flags
+        for flag in case.absent:
+            assert flag not in flags
+
+
+class PaneFlowLiveCase(NamedTuple):
+    """Fixture for exercising set_pane_flow against real tmux."""
+
+    test_id: str
+    state: str
+
+
+PANE_FLOW_CASES = [
+    PaneFlowLiveCase(test_id="pause", state="pause"),
+    PaneFlowLiveCase(test_id="continue", state="continue"),
+]
+PANE_FLOW_IDS = [case.test_id for case in PANE_FLOW_CASES]
+
+
+@pytest.mark.engines(["control"])
+@pytest.mark.parametrize("case", PANE_FLOW_CASES, ids=PANE_FLOW_IDS)
+def test_set_pane_flow_live(
+    case: PaneFlowLiveCase,
+    control_sandbox: t.ContextManager[Server],
+) -> None:
+    """set_pane_flow should succeed and leave the client usable."""
+    with control_sandbox as server:
+        session = server.new_session(
+            session_name="flow_case",
+            attach=True,
+            kill_session=True,
+        )
+        pane = session.active_pane
+        assert pane is not None
+        pane_id = t.cast(str, pane.pane_id)
+
+        engine = t.cast(ControlModeEngine, server.engine)
+        engine.set_pane_flow(pane_id, state=case.state)
+
+        pane.send_keys('printf "flow-test"\\n', literal=True, suppress_history=False)
+        lines = wait_for_line(pane, lambda line: "flow-test" in line)
+        assert any("flow-test" in line for line in lines)
+
+
+class SubscribeLiveCase(NamedTuple):
+    """Fixture for exercising subscribe/unsubscribe against real tmux."""
+
+    test_id: str
+    what_fmt: tuple[str, str]
+
+
+SUBSCRIBE_CASES = [
+    SubscribeLiveCase(
+        test_id="active_pane_subscription",
+        what_fmt=("%1", "#{pane_active}"),
+    ),
+]
+SUBSCRIBE_IDS = [case.test_id for case in SUBSCRIBE_CASES]
+
+
+@pytest.mark.engines(["control"])
+@pytest.mark.parametrize("case", SUBSCRIBE_CASES, ids=SUBSCRIBE_IDS)
+def test_subscribe_roundtrip_live(
+    case: SubscribeLiveCase,
+    control_sandbox: t.ContextManager[Server],
+) -> None:
+    """subscribe/unsubscribe should succeed without breaking control client."""
+    with control_sandbox as server:
+        engine = t.cast(ControlModeEngine, server.engine)
+        session = server.new_session(
+            session_name="sub_case",
+            attach=True,
+            kill_session=True,
+        )
+        pane = session.active_pane
+        assert pane is not None
+
+        target = case.what_fmt[0] if case.what_fmt[0] != "%1" else pane.pane_id
+        engine.subscribe("focus_test", what=target, fmt=case.what_fmt[1])
+        assert server.cmd("display-message", "-p", "ok").stdout == ["ok"]
+
+        engine.subscribe("focus_test", fmt=None)
+        assert server.cmd("display-message", "-p", "ok").stdout == ["ok"]