From f36c1f5ffed6a3a9a248d3654189472a246f9873 Mon Sep 17 00:00:00 2001 From: Roland Walker Date: Sat, 8 Aug 2026 16:45:58 -0400 Subject: [PATCH] avoid blocking read on boundary CLI tool output --- mycli/boundary_tunnel.py | 33 ++++++++++------ test/pytests/test_boundary_tunnel.py | 58 ++++++++++++++++++++++++---- 2 files changed, 72 insertions(+), 19 deletions(-) diff --git a/mycli/boundary_tunnel.py b/mycli/boundary_tunnel.py index 73ce701a..da439810 100644 --- a/mycli/boundary_tunnel.py +++ b/mycli/boundary_tunnel.py @@ -44,6 +44,7 @@ def __init__( self.stdout = '' self._startup_error: OSError | ValueError | None = None self._started = threading.Event() + self._output_ready = threading.Event() self._ready = threading.Event() self._failed = threading.Event() self._thread: threading.Thread | None = None @@ -71,19 +72,23 @@ def start(self, *, show_expiration_warning: bool = True) -> None: self._thread.start() deadline = time.monotonic() + self.ready_timeout while time.monotonic() < deadline: - if self._failed.is_set(): - self.close() - if self._startup_error is not None: - raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error - raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.') + self._raise_if_failed() if self._started.is_set(): - self._read_stdout() break time.sleep(0.05) else: self.close() raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process to start.') + while time.monotonic() < deadline: + self._raise_if_failed() + if self._output_ready.is_set(): + break + time.sleep(0.05) + else: + self.close() + raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process output.') + connection_details = json.loads(self.stdout) self.username = connection_details['credentials'][0]['secret']['decoded']['username'] self.password = connection_details['credentials'][0]['secret']['decoded']['password'] @@ -93,11 +98,7 @@ def start(self, *, show_expiration_warning: bool = True) -> None: self.expiry = datetime.datetime.strftime(expiry_local, '%H:%M:%S %a %d %b %Y') while time.monotonic() < deadline: - if self._failed.is_set(): - self.close() - if self._startup_error is not None: - raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error - raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.') + self._raise_if_failed() if self._is_listening(): self._ready.set() return @@ -105,6 +106,14 @@ def start(self, *, show_expiration_warning: bool = True) -> None: self.close() raise BoundaryTunnelError('Timed out waiting for Boundary tunnel to become ready.') + def _raise_if_failed(self) -> None: + if not self._failed.is_set(): + return + self.close() + if self._startup_error is not None: + raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error + raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.') + def _read_stdout(self) -> None: process = self.process if process is None or process.stdout is None: @@ -141,6 +150,8 @@ def _run(self) -> None: env=self._environment(), ) self._started.set() + self._read_stdout() + self._output_ready.set() except (OSError, ValueError) as exc: self._startup_error = exc self._failed.set() diff --git a/test/pytests/test_boundary_tunnel.py b/test/pytests/test_boundary_tunnel.py index 4ca0661a..278752b4 100644 --- a/test/pytests/test_boundary_tunnel.py +++ b/test/pytests/test_boundary_tunnel.py @@ -3,6 +3,7 @@ import os import socket import subprocess +import threading from typing import Any, cast import pytest @@ -100,8 +101,8 @@ def test_find_free_local_port_returns_available_port() -> None: sock.bind(('127.0.0.1', port)) -def test_boundary_tunnel_start_reads_stdout_in_main_thread(monkeypatch: pytest.MonkeyPatch) -> None: - calls: list[str] = [] +def test_boundary_tunnel_start_reads_stdout_in_worker_thread(monkeypatch: pytest.MonkeyPatch) -> None: + read_threads: list[str] = [] class FakeProcess: returncode = 0 @@ -118,9 +119,11 @@ def wait(self, timeout: float | None = None) -> int: def fake_run() -> None: tunnel.process = cast(Any, FakeProcess()) tunnel._started.set() + tunnel._read_stdout() + tunnel._output_ready.set() def fake_read_stdout() -> None: - calls.append('read_stdout') + read_threads.append(threading.current_thread().name) tunnel.stdout = CONNECTION_DETAILS tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406) @@ -131,7 +134,7 @@ def fake_read_stdout() -> None: tunnel.start() - assert calls == ['read_stdout'] + assert read_threads == ['mycli-boundary-tunnel'] assert tunnel.stdout == CONNECTION_DETAILS assert tunnel.username == '1234' assert tunnel.password == '5678' @@ -145,12 +148,10 @@ def test_boundary_tunnel_start_waits_for_process_to_start(monkeypatch: pytest.Mo def fake_sleep(seconds: float) -> None: sleeps.append(seconds) tunnel._started.set() - - def fake_read_stdout() -> None: tunnel.stdout = CONNECTION_DETAILS + tunnel._output_ready.set() monkeypatch.setattr(tunnel, '_run', lambda: None) - monkeypatch.setattr(tunnel, '_read_stdout', fake_read_stdout) monkeypatch.setattr(tunnel, '_is_listening', lambda: True) monkeypatch.setattr(boundary_tunnel.time, 'sleep', fake_sleep) @@ -188,6 +189,8 @@ def test_boundary_tunnel_start_reports_process_exit_after_stdout(monkeypatch: py def fake_run() -> None: tunnel._started.set() + tunnel._read_stdout() + tunnel._output_ready.set() def fake_read_stdout() -> None: tunnel.stdout = CONNECTION_DETAILS @@ -205,6 +208,8 @@ def test_boundary_tunnel_start_reports_startup_error_after_stdout(monkeypatch: p def fake_run() -> None: tunnel._started.set() + tunnel._read_stdout() + tunnel._output_ready.set() def fake_read_stdout() -> None: tunnel.stdout = CONNECTION_DETAILS @@ -251,16 +256,51 @@ def test_boundary_tunnel_start_reports_process_start_timeout(monkeypatch: pytest tunnel.start() +def test_boundary_tunnel_start_reports_process_output_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[str] = [] + output_released = threading.Event() + + class BlockingStdout: + def readline(self) -> bytes: + calls.append('read') + output_released.wait() + return b'' + + class FakeProcess: + stdout = BlockingStdout() + + def poll(self) -> None: + return None + + def terminate(self) -> None: + calls.append('terminate') + output_released.set() + + def wait(self, timeout: float | None = None) -> int: + calls.append(f'wait:{timeout}') + return 0 + + monkeypatch.setattr(boundary_tunnel.subprocess, 'Popen', lambda *_args, **_kwargs: FakeProcess()) + tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406, ready_timeout=0.1) + + with pytest.raises(BoundaryTunnelError, match='Timed out waiting for Boundary tunnel process output'): + tunnel.start() + + assert calls[:2] == ['read', 'terminate'] + assert sorted(calls[2:]) == ['wait:5', 'wait:None'] + + def test_boundary_tunnel_start_reports_timeout(monkeypatch: pytest.MonkeyPatch) -> None: tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406) tunnel.stdout = CONNECTION_DETAILS def fake_run() -> None: tunnel._started.set() + tunnel._output_ready.set() monkeypatch.setattr(tunnel, '_run', fake_run) monkeypatch.setattr(tunnel, '_is_listening', lambda: False) - monotonic_values = iter([0.0, 0.0, 31.0]) + monotonic_values = iter([0.0, 0.0, 0.0, 31.0]) monkeypatch.setattr(boundary_tunnel.time, 'monotonic', lambda: next(monotonic_values)) with pytest.raises(BoundaryTunnelError, match='Timed out waiting for Boundary tunnel'): @@ -310,6 +350,8 @@ def test_boundary_tunnel_run_tracks_process_status( popen_calls: list[tuple[list[str], dict[str, Any]]] = [] class FakeProcess: + stdout = None + def wait(self) -> int: return return_code