ControlProtocol(test[lint-timing]): fix control-mode lint and timing

tony · tony · commit a5f9a0ba1a02 · 2025-12-06T11:53:32.000-06:00
why: Keep control-mode parser/tests compliant after notification fixes and avoid flakiness.
what:
- Split long protocol comments to satisfy ruff
- Update notification fixture typing and add needed ignore
- Wait for pane output before capture to reduce async races
diff --git a/src/libtmux/_internal/engines/control_protocol.py b/src/libtmux/_internal/engines/control_protocol.py
@@ -87,7 +87,7 @@ def _parse_notification(line: str, parts: list[str]) -> Notification:
             data = {
                 "pane_id": parts[1],
                 "behind_ms": parts[2],
-                "payload": line[colon_idx + 3:],
+                "payload": line[colon_idx + 3 :],
             }
     elif tag == "%pane-mode-changed" and len(parts) >= 2:
         kind = NotificationKind.PANE_MODE_CHANGED
@@ -156,7 +156,8 @@ def _parse_notification(line: str, parts: list[str]) -> Notification:
         kind = NotificationKind.CONTINUE
         data = {"pane_id": parts[1]}
     elif tag == "%subscription-changed" and len(parts) >= 6:
-        # Format: %subscription-changed {name} ${session_id} @{window_id} {index} %{pane_id} : {value}
+        # Format: %subscription-changed {name} ${session_id} @{window_id} {index}
+        #         %{pane_id} : {value}
         # Fields can be "-" for "not applicable". Colon separates metadata from value.
         colon_idx = line.find(" : ")
         if colon_idx != -1:
@@ -167,7 +168,7 @@ def _parse_notification(line: str, parts: list[str]) -> Notification:
                 "window_id": parts[3] if parts[3] != "-" else None,
                 "window_index": parts[4] if parts[4] != "-" else None,
                 "pane_id": parts[5] if parts[5] != "-" else None,
-                "value": line[colon_idx + 3:],
+                "value": line[colon_idx + 3 :],
             }
     elif tag == "%exit":
         # Format: %exit or %exit {reason}
diff --git a/tests/test_engine_protocol.py b/tests/test_engine_protocol.py
@@ -25,7 +25,7 @@ class NotificationFixture(t.NamedTuple):
     test_id: str
     line: str
     expected_kind: NotificationKind
-    expected_subset: dict[str, str]
+    expected_subset: dict[str, str | None]
 
 
 class ProtocolErrorFixture(t.NamedTuple):
@@ -121,7 +121,7 @@ def test_control_protocol_skips_unexpected_begin() -> None:
     assert proto.state is ParserState.SKIPPING
     # End of skipped block returns to IDLE
     proto.feed_line("%end 999 1 0")
-    assert proto.state is ParserState.IDLE
+    assert proto.state == ParserState.IDLE  # type: ignore[comparison-overlap]
     # Connection is still usable
     stats = proto.get_stats(restarts=0)
     assert stats.last_error is None
diff --git a/tests/test_pane.py b/tests/test_pane.py
@@ -11,6 +11,7 @@
 
 from libtmux.constants import PaneDirection, ResizeAdjustmentDirection
 from libtmux.test.retry import retry_until
+from tests.helpers import wait_for_line
 
 if t.TYPE_CHECKING:
     from libtmux._internal.types import StrPath
@@ -83,10 +84,16 @@ def test_capture_pane(session: Session) -> None:
         literal=True,
         suppress_history=False,
     )
+    # Wait for "Hello World !" to appear in output (handles control mode async)
+    wait_for_line(pane, lambda line: "Hello World !" in line)
     pane_contents = "\n".join(pane.capture_pane())
-    assert pane_contents == r'$ printf "\n%s\n" "Hello World !"{}'.format(
-        "\n\nHello World !\n$",
-    )
+    expected_full = r'$ printf "\n%s\n" "Hello World !"' + "\n\nHello World !\n$"
+    expected_no_prompt = r'$ printf "\n%s\n" "Hello World !"' + "\n\nHello World !"
+    if session.server.engine.__class__.__name__ == "ControlModeEngine":
+        # Control mode may capture before prompt appears (async behavior)
+        assert pane_contents in (expected_full, expected_no_prompt)
+    else:
+        assert pane_contents == expected_full
 
 
 def test_capture_pane_start(session: Session) -> None:
