From b9968bd9449fa8152f795d9ea2a7216767b808fe Mon Sep 17 00:00:00 2001 From: Lahfir Date: Mon, 13 Jul 2026 23:25:47 -0700 Subject: [PATCH] test: split bounded e2e process support --- tests/e2e/bounded_process.py | 208 ++++++++++++++++++++++++++++++++++ tests/e2e/json_tool.py | 209 +---------------------------------- 2 files changed, 210 insertions(+), 207 deletions(-) create mode 100644 tests/e2e/bounded_process.py diff --git a/tests/e2e/bounded_process.py b/tests/e2e/bounded_process.py new file mode 100644 index 0000000..2633b07 --- /dev/null +++ b/tests/e2e/bounded_process.py @@ -0,0 +1,208 @@ +import dataclasses +import os +import resource +import selectors +import signal +import subprocess +import time + + +DEFAULT_TIMEOUT_SECONDS = 20.0 +DEFAULT_MAX_CAPTURE_BYTES = 2 * 1024 * 1024 +INTERACTION_LEASE_FD_ENV = "AGENT_DESKTOP_INTERACTION_LEASE_FD" +INHERIT_INTERACTION_LEASE_ENV = "AGENT_DESKTOP_E2E_INHERIT_LEASE" + + +@dataclasses.dataclass +class BoundedResult: + args: list[str] + returncode: int + stdout: str + stderr: str + timed_out: bool + output_limited: bool + termination_error: str | None + wall_ms: float + cpu_ms: float + + +def _terminate_group(process): + if process.poll() is not None: + return None + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + return None + except OSError as error: + try: + process.terminate() + except OSError: + pass + return f"process-group SIGTERM failed: {error}" + try: + process.wait(timeout=0.25) + return None + except subprocess.TimeoutExpired: + pass + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + return None + except OSError as error: + try: + process.kill() + except OSError: + pass + return f"process-group SIGKILL failed: {error}" + return None + + +def _merge_error(current, additional): + if not additional: + return current + return f"{current}; {additional}" if current else additional + + +def _interaction_lease_fds(env): + environment = os.environ if env is None else env + raw_fd = environment.get(INTERACTION_LEASE_FD_ENV) + if raw_fd is None: + return () + try: + fd = int(raw_fd) + except ValueError as error: + raise ValueError("interaction lease FD must be a decimal integer") from error + if fd < 0: + raise ValueError("interaction lease FD must be nonnegative") + os.fstat(fd) + return (fd,) + + +def _close_streams(selector, process): + for stream in (process.stdout, process.stderr): + if stream is None: + continue + try: + selector.unregister(stream) + except (KeyError, ValueError): + pass + stream.close() + + +def run_bounded( + command, + timeout_seconds=None, + max_capture_bytes=None, + env=None, + inherit_interaction_lease=False, +): + timeout_seconds = float( + timeout_seconds + if timeout_seconds is not None + else os.environ.get("AGENT_DESKTOP_E2E_TIMEOUT_SECONDS", DEFAULT_TIMEOUT_SECONDS) + ) + max_capture_bytes = int( + max_capture_bytes + if max_capture_bytes is not None + else os.environ.get("AGENT_DESKTOP_E2E_MAX_CAPTURE_BYTES", DEFAULT_MAX_CAPTURE_BYTES) + ) + if timeout_seconds <= 0 or max_capture_bytes <= 0: + raise ValueError("timeout and capture limits must be positive") + + usage_before = resource.getrusage(resource.RUSAGE_CHILDREN) + started = time.perf_counter() + child_env = dict(os.environ if env is None else env) + pass_fds = _interaction_lease_fds(child_env) if inherit_interaction_lease else () + if not inherit_interaction_lease: + child_env.pop(INTERACTION_LEASE_FD_ENV, None) + child_env.pop(INHERIT_INTERACTION_LEASE_ENV, None) + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + env=child_env, + pass_fds=pass_fds, + ) + selector = selectors.DefaultSelector() + selector.register(process.stdout, selectors.EVENT_READ, "stdout") + selector.register(process.stderr, selectors.EVENT_READ, "stderr") + chunks = {"stdout": [], "stderr": []} + captured = 0 + deadline = started + timeout_seconds + timed_out = False + output_limited = False + termination_error = None + + while selector.get_map(): + remaining = deadline - time.perf_counter() + if remaining <= 0: + timed_out = True + termination_error = _merge_error(termination_error, _terminate_group(process)) + _close_streams(selector, process) + break + events = selector.select(min(remaining, 0.1)) + if not events: + process.poll() + continue + for key, _ in events: + data = os.read(key.fileobj.fileno(), 65536) + if not data: + selector.unregister(key.fileobj) + key.fileobj.close() + continue + available = max_capture_bytes - captured + if len(data) > available: + if available > 0: + chunks[key.data].append(data[:available]) + captured += available + output_limited = True + termination_error = _merge_error(termination_error, _terminate_group(process)) + _close_streams(selector, process) + break + chunks[key.data].append(data) + captured += len(data) + if output_limited: + break + + selector.close() + if not timed_out and not output_limited: + remaining = max(0.0, deadline - time.perf_counter()) + try: + process.wait(timeout=remaining) + except subprocess.TimeoutExpired: + timed_out = True + termination_error = _merge_error(termination_error, _terminate_group(process)) + try: + process.wait(timeout=1) + except subprocess.TimeoutExpired: + termination_error = _merge_error(termination_error, _terminate_group(process)) + try: + process.wait(timeout=1) + except subprocess.TimeoutExpired: + termination_error = _merge_error( + termination_error, "child could not be reaped after termination" + ) + + usage_after = resource.getrusage(resource.RUSAGE_CHILDREN) + cpu_before = usage_before.ru_utime + usage_before.ru_stime + cpu_after = usage_after.ru_utime + usage_after.ru_stime + if timed_out: + returncode = 124 + elif output_limited: + returncode = 125 + elif process.returncode is None: + returncode = 126 + else: + returncode = int(process.returncode) + return BoundedResult( + args=list(command), + returncode=returncode, + stdout=b"".join(chunks["stdout"]).decode("utf-8", errors="replace"), + stderr=b"".join(chunks["stderr"]).decode("utf-8", errors="replace"), + timed_out=timed_out, + output_limited=output_limited, + termination_error=termination_error, + wall_ms=(time.perf_counter() - started) * 1000, + cpu_ms=max(0.0, (cpu_after - cpu_before) * 1000), + ) diff --git a/tests/e2e/json_tool.py b/tests/e2e/json_tool.py index ebb0648..1179f4d 100755 --- a/tests/e2e/json_tool.py +++ b/tests/e2e/json_tool.py @@ -1,217 +1,12 @@ #!/usr/bin/env python3 -import dataclasses import json import os -import resource -import selectors -import signal -import subprocess import sys -import time + +from bounded_process import INHERIT_INTERACTION_LEASE_ENV, run_bounded -DEFAULT_TIMEOUT_SECONDS = 20.0 -DEFAULT_MAX_CAPTURE_BYTES = 2 * 1024 * 1024 MAX_JSON_INPUT_BYTES = 4 * 1024 * 1024 -INTERACTION_LEASE_FD_ENV = "AGENT_DESKTOP_INTERACTION_LEASE_FD" -INHERIT_INTERACTION_LEASE_ENV = "AGENT_DESKTOP_E2E_INHERIT_LEASE" - - -@dataclasses.dataclass -class BoundedResult: - args: list[str] - returncode: int - stdout: str - stderr: str - timed_out: bool - output_limited: bool - termination_error: str | None - wall_ms: float - cpu_ms: float - - -def _terminate_group(process): - if process.poll() is not None: - return None - try: - os.killpg(process.pid, signal.SIGTERM) - except ProcessLookupError: - return None - except OSError as error: - try: - process.terminate() - except OSError: - pass - return f"process-group SIGTERM failed: {error}" - try: - process.wait(timeout=0.25) - return None - except subprocess.TimeoutExpired: - pass - try: - os.killpg(process.pid, signal.SIGKILL) - except ProcessLookupError: - return None - except OSError as error: - try: - process.kill() - except OSError: - pass - return f"process-group SIGKILL failed: {error}" - return None - - -def _merge_error(current, additional): - if not additional: - return current - return f"{current}; {additional}" if current else additional - - -def _interaction_lease_fds(env): - environment = os.environ if env is None else env - raw_fd = environment.get(INTERACTION_LEASE_FD_ENV) - if raw_fd is None: - return () - try: - fd = int(raw_fd) - except ValueError as error: - raise ValueError("interaction lease FD must be a decimal integer") from error - if fd < 0: - raise ValueError("interaction lease FD must be nonnegative") - os.fstat(fd) - return (fd,) - - -def _close_streams(selector, process): - for stream in (process.stdout, process.stderr): - if stream is None: - continue - try: - selector.unregister(stream) - except (KeyError, ValueError): - pass - stream.close() - - -def run_bounded( - command, - timeout_seconds=None, - max_capture_bytes=None, - env=None, - inherit_interaction_lease=False, -): - timeout_seconds = float( - timeout_seconds - if timeout_seconds is not None - else os.environ.get("AGENT_DESKTOP_E2E_TIMEOUT_SECONDS", DEFAULT_TIMEOUT_SECONDS) - ) - max_capture_bytes = int( - max_capture_bytes - if max_capture_bytes is not None - else os.environ.get("AGENT_DESKTOP_E2E_MAX_CAPTURE_BYTES", DEFAULT_MAX_CAPTURE_BYTES) - ) - if timeout_seconds <= 0 or max_capture_bytes <= 0: - raise ValueError("timeout and capture limits must be positive") - - usage_before = resource.getrusage(resource.RUSAGE_CHILDREN) - started = time.perf_counter() - child_env = dict(os.environ if env is None else env) - pass_fds = _interaction_lease_fds(child_env) if inherit_interaction_lease else () - if not inherit_interaction_lease: - child_env.pop(INTERACTION_LEASE_FD_ENV, None) - child_env.pop(INHERIT_INTERACTION_LEASE_ENV, None) - process = subprocess.Popen( - command, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - start_new_session=True, - env=child_env, - pass_fds=pass_fds, - ) - selector = selectors.DefaultSelector() - selector.register(process.stdout, selectors.EVENT_READ, "stdout") - selector.register(process.stderr, selectors.EVENT_READ, "stderr") - chunks = {"stdout": [], "stderr": []} - captured = 0 - deadline = started + timeout_seconds - timed_out = False - output_limited = False - termination_error = None - - while selector.get_map(): - remaining = deadline - time.perf_counter() - if remaining <= 0: - timed_out = True - termination_error = _merge_error(termination_error, _terminate_group(process)) - _close_streams(selector, process) - break - events = selector.select(min(remaining, 0.1)) - if not events: - process.poll() - continue - for key, _ in events: - data = os.read(key.fileobj.fileno(), 65536) - if not data: - selector.unregister(key.fileobj) - key.fileobj.close() - continue - available = max_capture_bytes - captured - if len(data) > available: - if available > 0: - chunks[key.data].append(data[:available]) - captured += available - output_limited = True - termination_error = _merge_error(termination_error, _terminate_group(process)) - _close_streams(selector, process) - break - chunks[key.data].append(data) - captured += len(data) - if output_limited: - break - - selector.close() - if not timed_out and not output_limited: - remaining = max(0.0, deadline - time.perf_counter()) - try: - process.wait(timeout=remaining) - except subprocess.TimeoutExpired: - timed_out = True - termination_error = _merge_error(termination_error, _terminate_group(process)) - try: - process.wait(timeout=1) - except subprocess.TimeoutExpired: - termination_error = _merge_error(termination_error, _terminate_group(process)) - try: - process.wait(timeout=1) - except subprocess.TimeoutExpired: - termination_error = _merge_error( - termination_error, "child could not be reaped after termination" - ) - - usage_after = resource.getrusage(resource.RUSAGE_CHILDREN) - cpu_before = usage_before.ru_utime + usage_before.ru_stime - cpu_after = usage_after.ru_utime + usage_after.ru_stime - if timed_out: - returncode = 124 - elif output_limited: - returncode = 125 - elif process.returncode is None: - returncode = 126 - else: - returncode = int(process.returncode) - return BoundedResult( - args=list(command), - returncode=returncode, - stdout=b"".join(chunks["stdout"]).decode("utf-8", errors="replace"), - stderr=b"".join(chunks["stderr"]).decode("utf-8", errors="replace"), - timed_out=timed_out, - output_limited=output_limited, - termination_error=termination_error, - wall_ms=(time.perf_counter() - started) * 1000, - cpu_ms=max(0.0, (cpu_after - cpu_before) * 1000), - ) - - def read_json(): payload = sys.stdin.buffer.read(MAX_JSON_INPUT_BYTES + 1) if len(payload) > MAX_JSON_INPUT_BYTES: