Skip to content

Commit 22c456a

Browse files
committed
ControlModeEngine(types): Introduce process protocol for test hooks
why: Remove mypy ignores and support injectable control-mode process factory. what: - Define _ControlProcess and _ProcessFactory protocols for control-mode transport - Type engine process and helper threads against protocol instead of Popen - Make close/restart logic rely on protocol methods without ignores
1 parent b4736f7 commit 22c456a

File tree

1 file changed

+43
-15
lines changed

1 file changed

+43
-15
lines changed

src/libtmux/_internal/engines/control_mode.py

Lines changed: 43 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,36 @@
2828
logger = logging.getLogger(__name__)
2929

3030

31+
class _ControlProcess(t.Protocol):
32+
"""Protocol for control-mode process handle (real or test fake)."""
33+
34+
stdin: t.TextIO | None
35+
stdout: t.Iterable[str] | None
36+
stderr: t.Iterable[str] | None
37+
38+
def terminate(self) -> None: ...
39+
40+
def kill(self) -> None: ...
41+
42+
def wait(self, timeout: float | None = None) -> t.Any: ...
43+
44+
45+
class _ProcessFactory(t.Protocol):
46+
"""Protocol for constructing a control-mode process."""
47+
48+
def __call__(
49+
self,
50+
cmd: list[str],
51+
*,
52+
stdin: t.Any,
53+
stdout: t.Any,
54+
stderr: t.Any,
55+
text: bool,
56+
bufsize: int,
57+
errors: str,
58+
) -> _ControlProcess: ...
59+
60+
3161
class ControlModeEngine(Engine):
3262
"""Engine that runs tmux commands via a persistent Control Mode process.
3363
@@ -46,7 +76,7 @@ def __init__(
4676
notification_queue_size: int = 4096,
4777
internal_session_name: str | None = None,
4878
attach_to: str | None = None,
49-
process_factory: t.Callable[[list[str]], subprocess.Popen[str]] | None = None,
79+
process_factory: _ProcessFactory | None = None,
5080
) -> None:
5181
"""Initialize control mode engine.
5282
@@ -69,12 +99,12 @@ def __init__(
6999
.. warning::
70100
Attaching to user sessions can cause notification spam from
71101
pane output. Use for advanced scenarios only.
72-
process_factory : Callable[[list[str]], subprocess.Popen], optional
102+
process_factory : _ProcessFactory, optional
73103
Test hook to override how the tmux control-mode process is created.
74104
When provided, it receives the argv list and must return an object
75105
compatible with ``subprocess.Popen`` (stdin/stdout/stderr streams).
76106
"""
77-
self.process: subprocess.Popen[str] | None = None
107+
self.process: _ControlProcess | None = None
78108
self._lock = threading.Lock()
79109
self._server_args: tuple[str | int, ...] | None = None
80110
self.command_timeout = command_timeout
@@ -98,15 +128,11 @@ def close(self) -> None:
98128
return
99129

100130
try:
101-
if hasattr(proc, "terminate"):
102-
proc.terminate() # type: ignore[call-arg]
103-
if hasattr(proc, "wait"):
104-
proc.wait(timeout=1) # type: ignore[call-arg]
131+
proc.terminate()
132+
proc.wait(timeout=1)
105133
except subprocess.TimeoutExpired:
106-
if hasattr(proc, "kill"):
107-
proc.kill() # type: ignore[call-arg]
108-
if hasattr(proc, "wait"):
109-
proc.wait() # type: ignore[call-arg]
134+
proc.kill()
135+
proc.wait()
110136
finally:
111137
self.process = None
112138
self._server_args = None
@@ -361,8 +387,10 @@ def _start_process(self, server_args: tuple[str | int, ...]) -> None:
361387
]
362388

363389
logger.debug("Starting Control Mode process: %s", cmd)
364-
popen_factory = self._process_factory or subprocess.Popen
365-
self.process = popen_factory( # type: ignore[arg-type]
390+
popen_factory: _ProcessFactory = (
391+
self._process_factory or subprocess.Popen # type: ignore[assignment]
392+
)
393+
self.process = popen_factory(
366394
cmd,
367395
stdin=subprocess.PIPE,
368396
stdout=subprocess.PIPE,
@@ -416,7 +444,7 @@ def _write_line(
416444
msg = "control mode process unavailable"
417445
raise exc.ControlModeConnectionError(msg) from None
418446

419-
def _reader(self, process: subprocess.Popen[str]) -> None:
447+
def _reader(self, process: _ControlProcess) -> None:
420448
assert process.stdout is not None
421449
try:
422450
for raw in process.stdout:
@@ -426,7 +454,7 @@ def _reader(self, process: subprocess.Popen[str]) -> None:
426454
finally:
427455
self._protocol.mark_dead("EOF from tmux")
428456

429-
def _drain_stderr(self, process: subprocess.Popen[str]) -> None:
457+
def _drain_stderr(self, process: _ControlProcess) -> None:
430458
if process.stderr is None:
431459
return
432460
for err_line in process.stderr:

0 commit comments

Comments
 (0)