ControlMode(test): Cover protocol errors, attach_to, sandbox isolation

Author: tony

tests/test_control_mode_regressions.py

@@ -31,6 +31,13 @@ class TrailingOutputFixture(t.NamedTuple):
     expected_stdout: list[str]
 
 
+class AttachFixture(t.NamedTuple):
+    """Fixture for attach_to behaviours."""
+
+    test_id: str
+    attach_to: str
+
+
 TRAILING_OUTPUT_CASES = [
     pytest.param(
         TrailingOutputFixture(
@@ -564,3 +571,36 @@ def test_session_kill_handles_control_eof() -> None:
     finally:
         with contextlib.suppress(Exception):
             server.kill()
+
+
+@pytest.mark.engines(["control"])
+@pytest.mark.parametrize(
+    "case",
+    [
+        AttachFixture(test_id="attach_existing", attach_to="shared_session"),
+    ],
+    ids=lambda c: c.test_id,
+)
+def test_attach_to_existing_session(case: AttachFixture) -> None:
+    """Control mode attach_to should not create/hide a management session."""
+    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,
+        )
+        engine = ControlModeEngine(attach_to=case.attach_to)
+        server = Server(socket_name=socket_name, engine=engine)
+        sessions = server.sessions
+        assert len(sessions) == 1
+        assert sessions[0].session_name == case.attach_to
+
+        attached = server.attached_sessions
+        assert len(attached) == 1
+        assert attached[0].session_name == case.attach_to
+    finally:
+        with contextlib.suppress(Exception):
+            bootstrap.kill()

tests/test_control_sandbox.py

@@ -24,3 +24,14 @@ def test_control_sandbox_smoke(control_sandbox: t.ContextManager[Server]) -> Non
         # Run a simple command to ensure control mode path works.
         out = server.cmd("display-message", "-p", "hi")
         assert out.stdout == ["hi"]
+
+
+def test_control_sandbox_isolation(control_sandbox: t.ContextManager[Server]) -> None:
+    """Sandbox should isolate HOME/TMUX_TMPDIR and use a unique socket."""
+    with control_sandbox as server:
+        assert server.socket_name is not None
+        assert server.socket_name.startswith("libtmux_test")
+        # Ensure TMUX is unset so the sandbox never reuses a user server
+        import os
+
+        assert "TMUX" not in os.environ

tests/test_engine_protocol.py

@@ -12,7 +12,11 @@
     NotificationKind,
     command_result_to_tmux_cmd,
 )
-from libtmux._internal.engines.control_protocol import CommandContext, ControlProtocol
+from libtmux._internal.engines.control_protocol import (
+    CommandContext,
+    ControlProtocol,
+    ParserState,
+)
 
 
 class NotificationFixture(t.NamedTuple):
@@ -24,6 +28,14 @@ class NotificationFixture(t.NamedTuple):
     expected_subset: dict[str, str]
 
 
+class ProtocolErrorFixture(t.NamedTuple):
+    """Fixture for protocol error handling."""
+
+    test_id: str
+    line: str
+    expected_reason: str
+
+
 def test_command_result_wraps_tmux_cmd() -> None:
     """CommandResult should adapt cleanly into tmux_cmd wrapper."""
     result = CommandResult(
@@ -74,6 +86,31 @@ def test_control_protocol_notifications() -> None:
     assert proto.get_stats(restarts=0).dropped_notifications >= 1
 
 
+PROTOCOL_ERROR_CASES: list[ProtocolErrorFixture] = [
+    ProtocolErrorFixture(
+        test_id="unexpected_end",
+        line="%end 123 1 0",
+        expected_reason="unexpected %end",
+    ),
+    ProtocolErrorFixture(
+        test_id="no_pending_begin",
+        line="%begin 999 1 0",
+        expected_reason="no pending command for %begin",
+    ),
+]
+
+
+@pytest.mark.parametrize("case", PROTOCOL_ERROR_CASES, ids=lambda c: c.test_id)
+def test_control_protocol_errors(case: ProtocolErrorFixture) -> None:
+    """Protocol errors should mark the parser DEAD and record last_error."""
+    proto = ControlProtocol()
+    proto.feed_line(case.line)
+    stats = proto.get_stats(restarts=0)
+    assert proto.state is ParserState.DEAD
+    assert stats.last_error is not None
+    assert case.expected_reason in stats.last_error
+
+
 NOTIFICATION_FIXTURES: list[NotificationFixture] = [
     NotificationFixture(
         test_id="layout_change",