test(control-mode): Add xfail coverage for backlog, notifications, collisions

Author: tony

tests/test_control_mode_engine.py

@@ -334,3 +334,39 @@ def test_run_result_retries_after_broken_pipe(
     # TODO: Implement retry simulation when engine supports injectable I/O.
     with pytest.raises(exc.ControlModeConnectionError):
         engine.run("list-sessions")
+
+
+class BackpressureFixture(t.NamedTuple):
+    """Fixture for notification backpressure integration."""
+
+    test_id: str
+    queue_size: int
+    overflow: int
+    expect_iter: bool
+
+
+@pytest.mark.xfail(
+    reason="control-mode notification backpressure integration not stable yet",
+    strict=False,
+)
+@pytest.mark.parametrize(
+    "case",
+    [
+        BackpressureFixture(
+            test_id="notif_overflow_iter",
+            queue_size=1,
+            overflow=5,
+            expect_iter=True,
+        ),
+    ],
+    ids=lambda c: c.test_id,
+)
+def test_notifications_overflow_then_iter(case: BackpressureFixture) -> None:
+    """Flood notif queue then ensure iter_notifications still yields."""
+    engine = ControlModeEngine()
+    engine._protocol = ControlProtocol(notification_queue_size=case.queue_size)
+    for _ in range(case.overflow):
+        engine._protocol.feed_line("%sessions-changed")
+    if case.expect_iter:
+        notif = next(engine.iter_notifications(timeout=0.05), None)
+        assert notif is not None

tests/test_control_mode_regressions.py

@@ -767,3 +767,29 @@ def test_attach_to_existing_session(case: AttachFixture) -> None:
     finally:
         with contextlib.suppress(Exception):
             bootstrap.kill()
+
+
+@pytest.mark.engines(["control"])
+@pytest.mark.xfail(
+    reason="list-clients filtering vs attached_sessions needs stable control flag",
+    strict=False,
+)
+def test_list_clients_control_flag_filters_attached() -> None:
+    """Control client row should have C flag and be filtered from attached_sessions."""
+    if has_lt_version("3.2"):
+        pytest.skip("tmux < 3.2 omits client_flags")
+    socket_name = f"libtmux_test_{uuid.uuid4().hex[:8]}"
+    engine = ControlModeEngine()
+    server = Server(socket_name=socket_name, engine=engine)
+    try:
+        _ = server.sessions  # start control mode
+        res = server.cmd(
+            "list-clients",
+            "-F",
+            "#{client_pid} #{client_flags} #{session_name}",
+        )
+        assert any("C" in line.split()[1] for line in res.stdout)
+        assert server.attached_sessions == []
+    finally:
+        with contextlib.suppress(Exception):
+            server.kill()