ControlMode(test): Add restart, overflow, attach failure, capture range, notifications

Author: tony

tests/test_control_client_logs.py

@@ -74,3 +74,26 @@ def test_control_client_capture_stream_parses(
     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)
+
+
+def test_control_client_notification_parsing(
+    control_client_logs: tuple[t.Any, ControlProtocol],
+) -> None:
+    """Control client log stream should produce notifications."""
+    proc, proto = control_client_logs
+    assert proc.stdin is not None
+
+    ctx = CommandContext(argv=["tmux", "display-message", "-p", "ping"])
+    proto.register_command(ctx)
+    # send a trivial command and rely on session-changed notification from attach
+    proc.stdin.write("display-message -p ping\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"))
+
+    notif = proto.get_notification(timeout=0.1)
+    assert notif is not None
+    assert notif.kind.name in {"SESSION_CHANGED", "CLIENT_SESSION_CHANGED", "RAW"}

tests/test_control_mode_engine.py

@@ -13,6 +13,7 @@
 from libtmux import exc
 from libtmux._internal.engines.base import ExitStatus
 from libtmux._internal.engines.control_mode import ControlModeEngine
+from libtmux._internal.engines.control_protocol import ControlProtocol
 from libtmux.server import Server
 
 
@@ -112,6 +113,40 @@ def fake_start(server_args: t.Sequence[str | int] | None) -> None:
     assert engine.process is None
 
 
+def test_control_mode_per_command_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Per-call timeout should close process and raise ControlModeTimeout."""
+
+    class FakeProcess:
+        def __init__(self) -> None:
+            self.stdin = io.StringIO()
+            self.stdout: t.Iterator[str] = iter([])  # no output
+            self.stderr = None
+            self._terminated = False
+
+        def terminate(self) -> None:
+            self._terminated = True
+
+        def kill(self) -> None:
+            self._terminated = True
+
+        def wait(self, timeout: float | None = None) -> None:
+            return None
+
+    engine = ControlModeEngine(command_timeout=5.0)
+
+    def fake_start(server_args: t.Sequence[str | int] | None) -> None:
+        engine.tmux_bin = "tmux"
+        engine._server_args = tuple(server_args or ())
+        engine.process = t.cast(subprocess.Popen[str], FakeProcess())
+
+    monkeypatch.setattr(engine, "_start_process", fake_start)
+
+    with pytest.raises(exc.ControlModeTimeout):
+        engine.run("list-sessions", timeout=0.01)
+
+    assert engine.process is None
+
+
 def test_control_mode_custom_session_name(tmp_path: pathlib.Path) -> None:
     """Control mode engine can use custom internal session name."""
     socket_path = tmp_path / "tmux-custom-session-test"
@@ -173,3 +208,83 @@ def test_control_mode_attach_to_existing(tmp_path: pathlib.Path) -> None:
     server2.kill()
     assert control_engine.process is not None
     control_engine.process.wait(timeout=2)
+
+
+class RestartFixture(t.NamedTuple):
+    """Fixture for restart/broken-pipe handling."""
+
+    test_id: str
+    should_raise: type[BaseException]
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        RestartFixture(
+            test_id="broken_pipe_increments_restart",
+            should_raise=exc.ControlModeConnectionError,
+        ),
+    ],
+    ids=lambda c: c.test_id,
+)
+def test_write_line_broken_pipe_increments_restart(
+    case: RestartFixture,
+) -> None:
+    """Broken pipe should raise ControlModeConnectionError and bump restarts."""
+
+    class FakeStdin:
+        def write(self, _: str) -> None:
+            raise BrokenPipeError
+
+        def flush(self) -> None:  # pragma: no cover - not reached
+            return None
+
+    class FakeProcess:
+        def __init__(self) -> None:
+            self.stdin = FakeStdin()
+
+    engine = ControlModeEngine()
+    engine.process = FakeProcess()  # type: ignore[assignment]
+
+    with pytest.raises(case.should_raise):
+        engine._write_line("list-sessions", server_args=())
+    assert engine._restarts == 1
+    assert engine.process is None
+
+
+class NotificationOverflowFixture(t.NamedTuple):
+    """Fixture for notification overflow handling."""
+
+    test_id: str
+    queue_size: int
+    overflow: int
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        NotificationOverflowFixture(
+            test_id="iter_notifications_after_drop",
+            queue_size=1,
+            overflow=3,
+        ),
+    ],
+    ids=lambda c: c.test_id,
+)
+def test_iter_notifications_survives_overflow(
+    case: NotificationOverflowFixture,
+) -> None:
+    """iter_notifications should continue yielding after queue drops."""
+    engine = ControlModeEngine()
+    engine._protocol = ControlProtocol(notification_queue_size=case.queue_size)
+
+    for _ in range(case.overflow):
+        engine._protocol.feed_line("%sessions-changed")
+
+    stats = engine.get_stats()
+    assert stats.dropped_notifications >= case.overflow - case.queue_size
+
+    notif_iter = engine.iter_notifications(timeout=0.01)
+    first = next(notif_iter, None)
+    assert first is not None
+    assert first.kind.name == "SESSIONS_CHANGED"

tests/test_control_mode_regressions.py

@@ -36,6 +36,7 @@ class AttachFixture(t.NamedTuple):
 
     test_id: str
     attach_to: str
+    expect_attached: bool
 
 
 TRAILING_OUTPUT_CASES = [
@@ -573,11 +574,116 @@ def test_session_kill_handles_control_eof() -> None:
             server.kill()
 
 
+class CaptureRangeFixture(t.NamedTuple):
+    """Fixture for capture-pane range/flag behavior."""
+
+    test_id: str
+    start: int | None
+    end: int | None
+    expected_tail: str
+
+
+class InternalNameCollisionFixture(t.NamedTuple):
+    """Fixture for internal session name collisions."""
+
+    test_id: str
+    internal_name: str
+
+
+@pytest.mark.engines(["control"])
+@pytest.mark.parametrize(
+    "case",
+    [
+        CaptureRangeFixture(
+            test_id="capture_with_range_untrimmed",
+            start=-1,
+            end=-1,
+            expected_tail="line2",
+        ),
+    ],
+    ids=lambda c: c.test_id,
+)
+def test_capture_pane_respects_range(case: CaptureRangeFixture) -> None:
+    """capture-pane with explicit range should return requested lines."""
+    socket_name = f"libtmux_test_{uuid.uuid4().hex[:8]}"
+    engine = ControlModeEngine()
+    server = Server(socket_name=socket_name, engine=engine)
+    try:
+        session = server.new_session(
+            session_name="capture_range",
+            attach=False,
+            kill_session=True,
+        )
+        pane = session.active_pane
+        assert pane is not None
+        pane.send_keys(
+            'printf "line1\\nline2\\n"',
+            literal=True,
+            suppress_history=False,
+        )
+        lines = pane.capture_pane(start=case.start, end=case.end)
+        assert lines
+        assert lines[-1].strip() == case.expected_tail
+    finally:
+        with contextlib.suppress(Exception):
+            server.kill()
+
+
 @pytest.mark.engines(["control"])
 @pytest.mark.parametrize(
     "case",
     [
-        AttachFixture(test_id="attach_existing", attach_to="shared_session"),
+        pytest.param(
+            InternalNameCollisionFixture(
+                test_id="collision_same_name",
+                internal_name="libtmux_control_mode",
+            ),
+            marks=pytest.mark.xfail(
+                reason="Engine does not yet guard internal session name collisions",
+            ),
+        ),
+    ],
+    ids=lambda c: c.test_id,
+)
+def test_internal_session_name_collision(case: InternalNameCollisionFixture) -> None:
+    """Two control engines with same internal name should not mask user sessions."""
+    socket_one = f"libtmux_test_{uuid.uuid4().hex[:8]}"
+    socket_two = f"libtmux_test_{uuid.uuid4().hex[:8]}"
+    engine1 = ControlModeEngine(internal_session_name=case.internal_name)
+    engine2 = ControlModeEngine(internal_session_name=case.internal_name)
+    server1 = Server(socket_name=socket_one, engine=engine1)
+    server2 = Server(socket_name=socket_two, engine=engine2)
+
+    try:
+        server1.new_session(session_name="user_one", attach=False, kill_session=True)
+        server2.new_session(session_name="user_two", attach=False, kill_session=True)
+
+        assert any(s.session_name == "user_one" for s in server1.sessions)
+        assert any(s.session_name == "user_two" for s in server2.sessions)
+    finally:
+        with contextlib.suppress(Exception):
+            server1.kill()
+        with contextlib.suppress(Exception):
+            server2.kill()
+
+
+@pytest.mark.engines(["control"])
+@pytest.mark.parametrize(
+    "case",
+    [
+        AttachFixture(
+            test_id="attach_existing",
+            attach_to="shared_session",
+            expect_attached=True,
+        ),
+        pytest.param(
+            AttachFixture(
+                test_id="attach_missing",
+                attach_to="missing_session",
+                expect_attached=False,
+            ),
+            id="attach_missing_session",
+        ),
     ],
     ids=lambda c: c.test_id,
 )
@@ -594,6 +700,11 @@ def test_attach_to_existing_session(case: AttachFixture) -> None:
         )
         engine = ControlModeEngine(attach_to=case.attach_to)
         server = Server(socket_name=socket_name, engine=engine)
+        if not case.expect_attached:
+            with pytest.raises(exc.ControlModeConnectionError):
+                _ = server.sessions
+            return
+
         sessions = server.sessions
         assert len(sessions) == 1
         assert sessions[0].session_name == case.attach_to