diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index e7b509efb..ee96e1f1c 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -2,10 +2,9 @@ import logging import os -import select +import signal import stat import tempfile -import time from collections.abc import Awaitable from dataclasses import dataclass from typing import TYPE_CHECKING, Callable, Literal @@ -23,26 +22,11 @@ logger = logging.getLogger(__name__) -MAX_DRAIN_BYTES = 256 * 1024 -DRAIN_TIMEOUT_SECONDS = 2.0 -DRAIN_MAX_EMPTY_POLLS = 10 - -# Upper bound on hook-contributed motd content read from $JMP_MOTD_FILE. MAX_MOTD_BYTES = 64 * 1024 -# Module-level reference to time.monotonic so tests can patch it without -# affecting the asyncio event loop (which also uses time.monotonic). -_monotonic = time.monotonic - def _flush_lines(buffer: bytes, output_lines: list[str]) -> bytes: - """Extract and log complete lines from a byte buffer. - - Splits the buffer on newline boundaries, decodes each complete line, - and appends non-empty lines to output_lines while logging them. - - Returns the remaining bytes after the last newline (incomplete line). - """ + """Extract complete lines from buffer, log them, return the remainder.""" while b"\n" in buffer: line, buffer = buffer.split(b"\n", 1) line_decoded = line.decode(errors="replace").rstrip() @@ -54,13 +38,7 @@ def _flush_lines(buffer: bytes, output_lines: list[str]) -> bytes: @dataclass class HookExecutionError(Exception): - """Raised when a hook fails and on_failure is set to 'endLease' or 'exit'. - - Attributes: - message: Error message describing the failure - on_failure: The on_failure mode that triggered this error ('endLease' or 'exit') - hook_type: The type of hook that failed ('before_lease' or 'after_lease') - """ + """Raised when a hook fails and on_failure is set to 'endLease' or 'exit'.""" message: str on_failure: Literal["endLease", "exit"] @@ -78,19 +56,6 @@ def should_end_lease(self) -> bool: return self.on_failure in ("endLease", "exit") -@dataclass -class PtyState: - """Mutable state for PTY file descriptors and reader coordination. - - Tracks which fds are still open (for cleanup) and provides a separate - stop flag to signal the reader task without affecting fd lifecycle. - """ - - parent_fd_open: bool = True - child_fd_open: bool = True - reader_stop: bool = False - - @dataclass(kw_only=True) class HookExecutor: """Executes lifecycle hooks with access to the j CLI.""" @@ -98,22 +63,13 @@ class HookExecutor: config: HookConfigV1Alpha1 def _create_hook_env(self, lease_scope: "LeaseContext") -> dict[str, str]: - """Create standardized hook environment variables. + """Create environment variables for hook execution. - Args: - lease_scope: LeaseScope containing lease metadata and socket paths - - Returns: - Dictionary of environment variables for hook execution - - Note: - Uses the hook_socket_path (if available) instead of the main socket_path - to prevent SSL frame corruption when hook j commands access the session - concurrently with client LogStream connections. + Uses hook_socket_path (if available) instead of the main socket_path + to prevent SSL frame corruption when hook j commands access the session + concurrently with client LogStream connections. """ hook_env = os.environ.copy() - # Use dedicated hook socket to prevent SSL corruption - # Falls back to main socket if hook socket not available (backward compatibility) socket_path = lease_scope.hook_socket_path or lease_scope.socket_path if lease_scope.hook_socket_path: logger.info( @@ -134,9 +90,8 @@ def _create_hook_env(self, lease_scope: "LeaseContext") -> dict[str, str]: "LEASE_NAME": lease_scope.lease_name, "CLIENT_NAME": lease_scope.client_name, # Signal noninteractive mode to the child process. - # Even though hooks run in a PTY (for line-buffered output), they - # are not interactive sessions. These variables prevent programs - # from displaying prompts or interactive UI. + # Hooks are not interactive sessions. These variables prevent + # programs from displaying prompts or interactive UI. "TERM": "dumb", "DEBIAN_FRONTEND": "noninteractive", "GIT_TERMINAL_PROMPT": "0", @@ -152,16 +107,7 @@ async def _execute_hook( lease_scope: "LeaseContext", log_source: LogSource, ) -> str | None: - """Execute a single hook command. - - Args: - hook_config: Hook configuration including script, timeout, and on_failure - lease_scope: LeaseScope containing lease metadata and session - log_source: Log source for hook output - - Returns: - Warning message string if hook failed with on_failure='warn', None otherwise - """ + """Execute a single hook command.""" command = hook_config.script if not command or not command.strip(): logger.debug("Hook command is empty, skipping") @@ -169,14 +115,11 @@ async def _execute_hook( logger.debug("Executing hook: %s", command.strip().split("\n")[0][:100]) - # Determine hook type from log source hook_type = "before_lease" if log_source == LogSource.BEFORE_LEASE_HOOK else "after_lease" - # Validate session is available for logging if lease_scope.session is None: raise RuntimeError("Cannot execute hook: lease_scope.session is None") - # Use existing session from lease_scope hook_env = self._create_hook_env(lease_scope) logger.debug( "Hook environment: JUMPSTARTER_HOST=%s, LEASE_NAME=%s, CLIENT_NAME=%s", @@ -239,20 +182,7 @@ def _handle_hook_failure( hook_type: Literal["before_lease", "after_lease"], cause: Exception | None = None, ) -> str | None: - """Handle hook failure according to on_failure setting. - - Args: - error_msg: Error message describing the failure - on_failure: The on_failure mode ('warn', 'endLease', or 'exit') - hook_type: The type of hook that failed - cause: Optional exception that caused the failure - - Returns: - Warning message string if on_failure is 'warn', None otherwise - - Raises: - HookExecutionError: If on_failure is 'endLease' or 'exit' - """ + """Handle hook failure according to on_failure setting.""" if on_failure == "warn": logger.warning("%s (on_failure=warn, continuing)", error_msg) return error_msg @@ -265,7 +195,6 @@ def _handle_hook_failure( hook_type=hook_type, ) - # Properly handle exception chaining if cause is not None: raise error from cause else: @@ -280,52 +209,28 @@ async def _execute_hook_process( # noqa: C901 logging_session: Session, hook_type: Literal["before_lease", "after_lease"], ) -> str | None: - """Execute the hook process with the given environment and logging session. - - Uses subprocess with a PTY to force line buffering in the subprocess, - ensuring logs stream in real-time rather than being block-buffered. - - Returns: - Warning message string if hook failed with on_failure='warn', None otherwise - """ - import pty + """Execute the hook process and capture its output via pipes.""" import subprocess command = hook_config.script timeout = hook_config.timeout on_failure = hook_config.on_failure - # Exception handling error_msg: str | None = None cause: Exception | None = None timed_out = False - # Route hook output logs to the client via the session's log stream logger.debug("Entering log source context for %s", log_source) with logging_session.context_log_source(__name__, log_source): - # Create a PTY pair - this forces line buffering in the subprocess logger.debug("Starting hook subprocess...") - logger.debug("Creating PTY pair...") - try: - parent_fd, child_fd = pty.openpty() - except Exception as e: - logger.error("Failed to create PTY: %s", e, exc_info=True) - raise - logger.debug("PTY created: parent_fd=%d, child_fd=%d", parent_fd, child_fd) - - pty_state = PtyState() process: subprocess.Popen | None = None try: - # Use subprocess.Popen with the PTY child as stdin/stdout/stderr - # This avoids the issues with os.fork() in async contexts - # Determine interpreter and invocation mode script_stripped = command.strip() is_file = "\n" not in script_stripped and os.path.isfile(script_stripped) interpreter = hook_config.exec_ if is_file and interpreter is None: - # Auto-detect interpreter from file extension import sys ext = os.path.splitext(script_stripped)[1].lower() @@ -349,137 +254,43 @@ async def _execute_hook_process( # noqa: C901 try: process = subprocess.Popen( cmd, - stdin=child_fd, - stdout=child_fd, - stderr=child_fd, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, env=hook_env, - start_new_session=True, # Equivalent to os.setsid() - close_fds=True, # Close inherited fds to prevent interference with gRPC connections + process_group=0, + close_fds=True, ) except Exception as e: logger.error("Failed to spawn subprocess: %s", e, exc_info=True) raise logger.debug("Subprocess spawned with PID %d", process.pid) - # Close child fd in parent process - subprocess has it now - os.close(child_fd) - pty_state.child_fd_open = False - logger.debug("Closed child_fd in parent process") output_lines: list[str] = [] - # Set parent fd to non-blocking mode - import fcntl - - flags = fcntl.fcntl(parent_fd, fcntl.F_GETFL) - fcntl.fcntl(parent_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - logger.debug("Parent fd set to non-blocking") + assert process.stdout is not None + pipe_fd = process.stdout.fileno() + os.set_blocking(pipe_fd, False) - async def read_pty_output() -> None: # noqa: C901 - """Read from PTY parent fd line by line using non-blocking I/O.""" - logger.debug("read_pty_output task started") + async def read_output() -> None: + """Read subprocess output via pipe using async non-blocking I/O.""" buffer = b"" - read_count = 0 - last_heartbeat = 0 - - start_time = _monotonic() try: - while not pty_state.reader_stop: + while True: try: - # Wait for fd to be readable with timeout with anyio.move_on_after(0.1): - await anyio.wait_readable(parent_fd) - - # Check stop flag immediately after timeout - # (main task may have signaled us to stop) - if pty_state.reader_stop: - logger.debug("read_pty_output: stop flag set, exiting") + await anyio.wait_readable(pipe_fd) + chunk = os.read(pipe_fd, 4096) + if not chunk: break - - read_count += 1 - # Log heartbeat every 2 seconds - elapsed = _monotonic() - start_time - if elapsed - last_heartbeat >= 2.0: - logger.debug( - "read_pty_output: heartbeat at %.1fs, iterations=%d", elapsed, read_count - ) - last_heartbeat = elapsed - - # Read available data (non-blocking) - try: - chunk = os.read(parent_fd, 4096) - if not chunk: - # EOF - logger.debug("read_pty_output: EOF received") - break - buffer += chunk - except BlockingIOError: - # No data available right now, continue loop - continue - except OSError as e: - # PTY closed or error - logger.debug("read_pty_output: OSError on read: %s", e) - break - - # Process complete lines - buffer = _flush_lines(buffer, output_lines) - + buffer += chunk + except BlockingIOError: + continue except OSError as e: - # PTY closed or read error - logger.debug("read_pty_output: OSError in loop: %s", e) + logger.debug("read_output: OSError: %s", e) break - finally: - # Drain any remaining data from the PTY buffer. - # On macOS, PTY output may still be in the kernel buffer - # after the subprocess exits and the stop flag is set. - # Use select() with a timeout to poll for readability - # instead of immediately breaking on BlockingIOError, - # giving the macOS PTY kernel buffer time to deliver - # remaining data. - # Bound the drain to prevent spinning indefinitely if a - # grandchild process holds the PTY slave fd open. - try: - drain_deadline = _monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - consecutive_empty = 0 - while drained < MAX_DRAIN_BYTES and _monotonic() < drain_deadline: - # Poll for readability with a short timeout. - # This avoids the race where a non-blocking read - # raises BlockingIOError because the macOS PTY - # kernel buffer hasn't delivered the data yet. - remaining = drain_deadline - _monotonic() - if remaining <= 0: - break - timeout_s = min(remaining, 0.1) - try: - readable, _, _ = select.select([parent_fd], [], [], timeout_s) - except (ValueError, OSError): - # fd closed or invalid - break - if not readable: - # On macOS, data may not be available on the - # first select() call even though the subprocess - # has already written and exited. Keep retrying - # until we see several consecutive empty polls, - # which indicates the buffer is truly drained. - consecutive_empty += 1 - if consecutive_empty >= DRAIN_MAX_EMPTY_POLLS: - break - continue - consecutive_empty = 0 - try: - chunk = os.read(parent_fd, 4096) - if not chunk: - break - buffer += chunk - drained += len(chunk) - except (BlockingIOError, OSError): - break - buffer = _flush_lines(buffer, output_lines) - except Exception: - logger.debug("read_pty_output: error during drain", exc_info=True) - - logger.debug("read_pty_output: exiting, processed %d iterations", read_count) + finally: if buffer: line_decoded = buffer.decode(errors="replace").rstrip() if line_decoded: @@ -487,80 +298,43 @@ async def read_pty_output() -> None: # noqa: C901 logger.info("%s", line_decoded) async def wait_for_process() -> int: - """Wait for the subprocess to complete. - - Ensures the subprocess is properly reaped even if cancelled, - preventing zombie processes. - """ + """Wait for the subprocess to complete.""" logger.debug("wait_for_process: waiting for PID %d", process.pid) - try: - result = await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) - logger.debug("wait_for_process: PID %d exited with code %d", process.pid, result) - return result - finally: - # Ensure subprocess is reaped on cancellation to prevent zombies - if process.poll() is None: - logger.debug("wait_for_process: cleaning up still-running PID %d", process.pid) - try: - process.terminate() - # Give it a moment to terminate gracefully - for _ in range(10): - if process.poll() is not None: - break - await anyio.sleep(0.1) - # Force kill if still running - if process.poll() is None: - logger.debug("wait_for_process: force killing PID %d", process.pid) - process.kill() - # Final reap with non-abandoning wait - await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=False) - except Exception as e: - logger.debug("wait_for_process: error during cleanup: %s", e) - - # Use move_on_after for timeout - returncode: int | None = None - logger.debug("Starting PTY output reader and process waiter (timeout=%d)", timeout) + result = await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) + logger.debug("wait_for_process: PID %d exited with code %d", process.pid, result) + return result - # Yield to event loop to ensure other tasks can progress - # This helps prevent race conditions in task scheduling - await anyio.sleep(0) + returncode: int | None = None + logger.debug("Starting output reader and process waiter (timeout=%d)", timeout) with anyio.move_on_after(timeout) as cancel_scope: - # Run output reading and process waiting concurrently async with anyio.create_task_group() as tg: - logger.debug("Task group created, starting tasks...") - tg.start_soon(read_pty_output) - logger.debug("Waiting for subprocess to complete...") + tg.start_soon(read_output) returncode = await wait_for_process() logger.debug("Subprocess completed with code: %s", returncode) - # Give a brief moment for any final output to be read - await anyio.sleep(0.2) - # Signal the read task to stop via the dedicated stop flag. - # The read task checks this flag after each 0.1s timeout - # and also receives EOF when the subprocess exits. - # Note: pty_state.parent_fd_open stays True so the finally block - # properly closes parent_fd. - pty_state.reader_stop = True - logger.debug("Stop flag set, waiting for read task to exit") - # Don't cancel - let the task exit naturally via EOF or flag check - # Cancellation can cause unexpected side effects on gRPC connections + # Yield to let the LogStream deliver any pending + # messages before reporting the hook result. + await anyio.sleep(0) if cancel_scope.cancelled_caught: timed_out = True error_msg = f"Hook timed out after {timeout} seconds" logger.error(error_msg) - # Terminate the process if process and process.poll() is None: - process.terminate() - # Give it a moment to terminate gracefully + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass try: with anyio.move_on_after(5): await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) except Exception: pass - # Force kill if still running if process.poll() is None: - process.kill() + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass try: await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) except Exception: @@ -577,40 +351,37 @@ async def wait_for_process() -> int: cause = e logger.error(error_msg, exc_info=True) finally: - # Clean up file descriptors - only close those still open to avoid - # closing an unrelated fd that reused the same number. - if pty_state.parent_fd_open: - try: - os.close(parent_fd) - except OSError: - pass - if pty_state.child_fd_open: - try: - os.close(child_fd) - except OSError: - pass + if process: + if process.poll() is None: + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + process.wait(timeout=5) + except Exception: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + try: + process.wait(timeout=5) + except Exception: + pass + if process.stdout: + try: + process.stdout.close() + except OSError: + pass - # Handle failure inside context_log_source so the WARNING log is - # routed to the client as a hook log (visible without --exporter-logs). if error_msg is not None: - # For timeout, create a TimeoutError as the cause if timed_out and cause is None: cause = TimeoutError(error_msg) return self._handle_hook_failure(error_msg, on_failure, hook_type, cause) return None async def execute_before_lease_hook(self, lease_scope: "LeaseContext") -> str | None: - """Execute the before-lease hook. - - Args: - lease_scope: LeaseScope with lease metadata and session - - Returns: - Warning message string if hook failed with on_failure='warn', None otherwise - - Raises: - HookExecutionError: If hook fails and on_failure is set to 'endLease' or 'exit' - """ + """Execute the before-lease hook.""" if not self.config.before_lease: logger.debug("No before-lease hook configured") return None @@ -623,17 +394,7 @@ async def execute_before_lease_hook(self, lease_scope: "LeaseContext") -> str | ) async def execute_after_lease_hook(self, lease_scope: "LeaseContext") -> str | None: - """Execute the after-lease hook. - - Args: - lease_scope: LeaseScope with lease metadata and session - - Returns: - Warning message string if hook failed with on_failure='warn', None otherwise - - Raises: - HookExecutionError: If hook fails and on_failure is set to 'endLease' or 'exit' - """ + """Execute the after-lease hook.""" if not self.config.after_lease: logger.debug("No after-lease hook configured") return None @@ -696,25 +457,14 @@ async def run_before_lease_hook( ) -> None: """Execute before-lease hook with full orchestration. - This method handles the complete lifecycle of running a before-lease hook: - - Waits for the lease scope to be ready (session/socket populated) - - Reports status changes via the provided callback - - Sets up the hook executor with the session for logging - - Executes the hook and handles errors - - Always signals the before_lease_hook event to unblock connections - - Args: - lease_scope: LeaseScope containing session, socket_path, and sync event - report_status: Async callback to report status changes to controller - shutdown: Callback to trigger exporter shutdown (accepts optional exit_code kwarg) - request_lease_release: Async callback to request lease release from controller + Always signals the before_lease_hook event to unblock connections, + even on failure. """ should_release = False try: if not await self._wait_for_lease_ready(lease_scope, report_status): return - # Check if hook is configured if not self.config.before_lease: logger.debug("No before-lease hook configured") await report_status(ExporterStatus.LEASE_READY, "Ready for commands") @@ -732,7 +482,6 @@ async def run_before_lease_hook( await report_status(ExporterStatus.BEFORE_LEASE_HOOK, "Running beforeLease hook") - # Execute hook with lease scope logger.info("Executing before-lease hook for lease %s", lease_scope.lease_name) warning = await self._execute_hook( self.config.before_lease, @@ -756,7 +505,6 @@ async def run_before_lease_hook( except HookExecutionError as e: if e.should_shutdown_exporter(): - # on_failure='exit' - defer shutdown until client handles the failure logger.error("beforeLease hook failed with on_failure='exit': %s", e) lease_scope.skip_after_lease_hook = True await report_status( @@ -767,10 +515,8 @@ async def run_before_lease_hook( ExporterStatus.OFFLINE, "Exporter shutting down due to beforeLease hook failure", ) - # Defer shutdown: sets _stop_requested=True, actual stop after lease cleanup shutdown(exit_code=1, wait_for_lease_exit=True, should_unregister=True) else: - # on_failure='endLease' - report failure, release in finally block logger.error("beforeLease hook failed with on_failure='endLease': %s", e) lease_scope.skip_after_lease_hook = True should_release = True @@ -785,13 +531,9 @@ async def run_before_lease_hook( ExporterStatus.BEFORE_LEASE_HOOK_FAILED, f"beforeLease hook failed: {e}", ) - # Unexpected errors don't trigger shutdown - just block the lease - finally: - # Always set the event to unblock connections lease_scope.before_lease_hook.set() - # Release lease for endLease failure mode. # Shielded from cancellation to ensure the release completes # even if the task group is being torn down. if should_release: @@ -806,32 +548,14 @@ async def run_after_lease_hook( shutdown: Callable[..., None], request_lease_release: Callable[[], Awaitable[None]] | None = None, ) -> None: - """Execute after-lease hook with full orchestration. - - This method handles the complete lifecycle of running an after-lease hook: - - Validates that the lease scope is ready - - Reports status changes via the provided callback - - Sets up the hook executor with the session for logging - - Executes the hook and handles errors - - Triggers shutdown on critical failures (HookExecutionError) - - Requests lease release from controller after hook completes - - Args: - lease_scope: LeaseScope containing session, socket_path, and client info - report_status: Async callback to report status changes to controller - shutdown: Callback to trigger exporter shutdown (accepts optional exit_code kwarg) - request_lease_release: Async callback to request lease release from controller - """ + """Execute after-lease hook with full orchestration.""" shutdown_called = False try: - # Verify lease scope is ready - for after-lease this should always be true - # since we've already processed the lease, but check defensively if not lease_scope.is_ready(): - logger.warning("LeaseScope not ready for after-lease hook, skipping") + logger.warning("LeaseContext not ready for after-lease hook, skipping") await report_status(ExporterStatus.AVAILABLE, "Available for new lease") return - # Check if hook is configured if not self.config.after_lease: logger.debug("No after-lease hook configured") await report_status(ExporterStatus.AVAILABLE, "Available for new lease") @@ -839,7 +563,6 @@ async def run_after_lease_hook( await report_status(ExporterStatus.AFTER_LEASE_HOOK, "Running afterLease hooks") - # Execute hook with lease scope logger.info("Executing after-lease hook for lease %s", lease_scope.lease_name) warning = await self._execute_hook( self.config.after_lease, @@ -856,7 +579,6 @@ async def run_after_lease_hook( except HookExecutionError as e: if e.should_shutdown_exporter(): - # on_failure='exit' - shut down the entire exporter logger.error("afterLease hook failed with on_failure='exit': %s", e) await report_status( ExporterStatus.AFTER_LEASE_HOOK_FAILED, @@ -866,16 +588,10 @@ async def run_after_lease_hook( ExporterStatus.OFFLINE, "Exporter shutting down due to afterLease hook failure", ) - # No delay needed - client is already polling and will see the failure logger.error("Shutting down exporter due to afterLease hook failure with on_failure='exit'") - # Exit code 1 tells the CLI not to restart the exporter shutdown(exit_code=1, should_unregister=True, wait_for_lease_exit=True) shutdown_called = True else: - # on_failure='endLease' - report failure to the client, then release the lease. - # AFTER_LEASE_HOOK_FAILED is a transient status: the client sees the failure, - # the lease is released in the finally block, and the exporter's main loop - # clears the lease context and accepts new leases. logger.error("afterLease hook failed with on_failure='endLease': %s", e) await report_status( ExporterStatus.AFTER_LEASE_HOOK_FAILED, @@ -883,9 +599,6 @@ async def run_after_lease_hook( ) except Exception as e: - # Unexpected errors: report failure but do not shut down. - # Same transient status - the lease is released and the exporter - # accepts new leases after the finally block completes. logger.error("afterLease hook failed with unexpected error: %s", e, exc_info=True) await report_status( ExporterStatus.AFTER_LEASE_HOOK_FAILED, @@ -893,11 +606,10 @@ async def run_after_lease_hook( ) finally: - # Always delay to give client time to poll the final status await anyio.sleep(1.0) - # Don't release lease when exporter is shutting down - unregistration handles cleanup. - # Releasing here would report AVAILABLE to the controller right before shutdown. + # Don't release lease when exporter is shutting down -- + # releasing here would report AVAILABLE right before shutdown. if request_lease_release and not shutdown_called: try: await request_lease_release() diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 01acee354..35662f7d5 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -1,5 +1,5 @@ import os -import sys +import subprocess from contextlib import nullcontext from unittest.mock import AsyncMock, MagicMock, patch @@ -8,98 +8,14 @@ from jumpstarter.common import HOOK_WARNING_PREFIX, ExporterStatus from jumpstarter.config.exporter import HookConfigV1Alpha1, HookInstanceConfigV1Alpha1 from jumpstarter.exporter.hooks import ( - DRAIN_MAX_EMPTY_POLLS, - DRAIN_TIMEOUT_SECONDS, - MAX_DRAIN_BYTES, MAX_MOTD_BYTES, HookExecutionError, HookExecutor, _flush_lines, - _monotonic, ) pytestmark = pytest.mark.anyio -# Tests that spawn real subprocesses via PTY and assert on captured logger -# output are flaky on macOS due to a PTY kernel buffer timing race condition. -# See https://github.com/jumpstarter-dev/jumpstarter/issues/821 -# Targeted for proper fix in 0.10.0. -macos_pty_xfail = pytest.mark.xfail( - condition=sys.platform == "darwin", - reason="PTY output race condition on macOS (#821)", - strict=False, -) - - -class _PtyTracker: - """Tracks PTY fd and EOF state for drain tests that need to intercept - os.read and pty.openpty calls. - - When ``return_drain_data`` is True (default), the first os.read after EOF - returns ``b"SHOULD_NOT_APPEAR\\n"``; otherwise it returns ``b""``. - """ - - def __init__(self, *, return_drain_data: bool = True) -> None: - import pty - - self.parent_fd: int | None = None - self.eof_seen: bool = False - self._drain_data_returned: bool = False - self._return_drain_data = return_drain_data - self._original_openpty = pty.openpty - self._original_os_read = os.read - - def tracking_openpty(self): - parent, child = self._original_openpty() - self.parent_fd = parent - return parent, child - - def os_read_with_drain_data(self, fd, size): - if fd != self.parent_fd: - return self._original_os_read(fd, size) - if not self.eof_seen: - try: - data = self._original_os_read(fd, size) - except (BlockingIOError, OSError): - self.eof_seen = True - raise - if not data: - self.eof_seen = True - return b"" - return data - if self._return_drain_data and not self._drain_data_returned: - self._drain_data_returned = True - return b"SHOULD_NOT_APPEAR\n" - return b"" - - -class _DrainDeadlineClock: - """A callable that replaces ``_monotonic`` to simulate the drain - deadline being exceeded between the ``while`` condition check and the - ``remaining`` calculation. - - Only patches the hooks module's ``_monotonic`` reference, leaving - ``time.monotonic`` (used by the asyncio event loop) unaffected. - """ - - def __init__(self, real_monotonic, state: _PtyTracker) -> None: - self._real = real_monotonic - self._state = state - self._call_count = 0 - self._deadline: float | None = None - - def __call__(self) -> float: - real_time = self._real() - if not self._state.eof_seen: - return real_time - self._call_count += 1 - if self._call_count == 1: - self._deadline = real_time + DRAIN_TIMEOUT_SECONDS - return real_time - if self._call_count == 2: - return self._deadline - 0.001 # type: ignore[operator] - return self._deadline + 1.0 # type: ignore[operator] - class TestFlushLines: def test_extracts_complete_lines(self) -> None: @@ -152,9 +68,7 @@ def lease_scope(): before_lease_hook=Event(), client_name="test-client", ) - # Add mock session to lease_scope mock_session = MagicMock() - # Return a no-op context manager for context_log_source mock_session.context_log_source.return_value = nullcontext() # Session.motd is str | None; model it so the beforeLease motd append works mock_session.motd = None @@ -173,7 +87,6 @@ async def test_empty_hook_execution(self, lease_scope) -> None: empty_config = HookConfigV1Alpha1() executor = HookExecutor(config=empty_config) - # Both hooks should return None for empty/None commands assert await executor.execute_before_lease_hook(lease_scope) is None assert await executor.execute_after_lease_hook(lease_scope) is None @@ -210,7 +123,7 @@ async def test_hook_timeout(self, lease_scope) -> None: assert "timed out after 1 seconds" in str(exc_info.value) assert exc_info.value.on_failure == "exit" - @macos_pty_xfail + async def test_hook_environment_variables(self, lease_scope) -> None: hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( @@ -310,7 +223,7 @@ def test_append_hook_motd_caps_size(self, tmp_path) -> None: HookExecutor._append_hook_motd(session, str(big)) assert len(session.motd) <= MAX_MOTD_BYTES - @macos_pty_xfail + async def test_real_time_output_logging(self, lease_scope) -> None: """Test that hook output is logged in real-time at INFO level.""" hook_config = HookConfigV1Alpha1( @@ -328,7 +241,7 @@ async def test_real_time_output_logging(self, lease_scope) -> None: assert any("Line 2" in call for call in info_calls) assert any("Line 3" in call for call in info_calls) - @macos_pty_xfail + async def test_post_lease_hook_execution_on_completion(self, lease_scope) -> None: """Test that post-lease hook executes when called directly.""" hook_config = HookConfigV1Alpha1( @@ -355,7 +268,6 @@ async def test_hook_timeout_with_warn(self, lease_scope) -> None: result = await executor.execute_before_lease_hook(lease_scope) assert result is not None assert "timed out" in result.lower() - # Verify WARNING log was created warning_calls = [str(call) for call in mock_logger.warning.call_args_list] assert any("on_failure=warn, continuing" in call for call in warning_calls) @@ -371,12 +283,8 @@ async def test_failed_hook_with_warn_returns_warning(self, lease_scope) -> None: assert "exit code 1" in result.lower() async def test_failed_hook_with_warn_logs_warning_inside_log_source_context(self) -> None: - """Test that the WARNING log for on_failure='warn' is emitted inside context_log_source. - - Issue #246: The WARNING log from _handle_hook_failure must be emitted while - the context_log_source context manager is active. This ensures the warning - is tagged with the hook source (BEFORE_LEASE_HOOK / AFTER_LEASE_HOOK) and - is visible to the client even without --exporter-logs. + """The WARNING log for on_failure='warn' must be emitted inside context_log_source + so the warning is tagged with the hook source and visible to the client. """ from contextlib import contextmanager @@ -389,7 +297,6 @@ async def test_failed_hook_with_warn_logs_warning_inside_log_source_context(self ) executor = HookExecutor(config=hook_config) - # Track whether context_log_source is active when warning is logged context_active = False warning_logged_in_context = False @@ -439,7 +346,7 @@ async def test_successful_hook_returns_none(self, lease_scope) -> None: result = await executor.execute_before_lease_hook(lease_scope) assert result is None - @macos_pty_xfail + async def test_exec_bash(self, lease_scope) -> None: """Test that exec=/bin/bash allows bash-specific syntax. @@ -461,7 +368,7 @@ async def test_exec_bash(self, lease_scope) -> None: info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("BASH_OK: world" in call for call in info_calls) - @macos_pty_xfail + async def test_exec_python3(self, lease_scope) -> None: """Test that exec=python3 runs inline Python. @@ -484,7 +391,7 @@ async def test_exec_python3(self, lease_scope) -> None: # Expected total: 0 + 1 + 4 + 9 == 14 assert any("PYTHON_OK: 14" in call for call in info_calls) - @macos_pty_xfail + async def test_script_file_sh(self, lease_scope, tmp_path) -> None: """Test that a .sh file auto-detects /bin/sh as interpreter.""" script_file = tmp_path / "hook_script.sh" @@ -507,7 +414,7 @@ async def test_script_file_sh(self, lease_scope, tmp_path) -> None: debug_calls = [str(call) for call in mock_logger.debug.call_args_list] assert any("Executing script file" in call for call in debug_calls) - @macos_pty_xfail + async def test_script_file_py_autodetects_python(self, lease_scope, tmp_path) -> None: """Test that a .py file auto-detects the exporter's Python as interpreter.""" import sys @@ -528,13 +435,11 @@ async def test_script_file_py_autodetects_python(self, lease_scope, tmp_path) -> assert result is None info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("PYFILE_OK" in call for call in info_calls) - # Verify it auto-detected Python (now logged at DEBUG level) debug_calls = [str(call) for call in mock_logger.debug.call_args_list] assert any("Auto-detected Python script" in call for call in debug_calls) - # Verify it used the exporter's own Python interpreter assert any(sys.executable in call for call in debug_calls) - @macos_pty_xfail + async def test_script_file_py_exec_override(self, lease_scope, tmp_path) -> None: """Test that explicit exec overrides .py auto-detection.""" script_file = tmp_path / "hook_script.py" @@ -554,11 +459,10 @@ async def test_script_file_py_exec_override(self, lease_scope, tmp_path) -> None assert result is None info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("OVERRIDE_OK" in call for call in info_calls) - # Should NOT say "Auto-detected" since exec was explicitly set debug_calls = [str(call) for call in mock_logger.debug.call_args_list] assert not any("Auto-detected" in call for call in debug_calls) - @macos_pty_xfail + async def test_noninteractive_environment(self, lease_scope) -> None: """Test that hooks receive noninteractive environment variables. @@ -566,8 +470,8 @@ async def test_noninteractive_environment(self, lease_scope) -> None: and that PS1 is not set in the env dict passed to the subprocess. Note: PS1 is verified via _create_hook_env directly because shells - started in a PTY may re-set PS1 from init files despite it being - removed from the environment. + may re-set PS1 from init files despite it being removed from the + environment. """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( @@ -582,7 +486,7 @@ async def test_noninteractive_environment(self, lease_scope) -> None: executor = HookExecutor(config=hook_config) # Verify PS1 is removed from the env dict (not via subprocess, since - # shells in a PTY may re-set PS1 from profile/init files) + # shells may re-set PS1 from profile/init files) hook_env = executor._create_hook_env(lease_scope) assert "PS1" not in hook_env @@ -658,168 +562,8 @@ async def test_before_lease_hook_endlease_handles_release_error(self, lease_scop assert lease_scope.skip_after_lease_hook is True mock_request_lease_release.assert_called_once() - async def test_pty_output_drained_after_stop_flag_set(self) -> None: - """Test that PTY drain captures data remaining after the stop flag is set. - - Simulates the macOS scenario where PTY output is still in the kernel - buffer after the subprocess exits and reader_stop is set. Uses a pipe - to inject data, sets reader_stop=True to skip the main loop, and - verifies the finally-block drain captures all lines. - """ - import fcntl - import time - - read_fd, write_fd = os.pipe() - try: - flags = fcntl.fcntl(read_fd, fcntl.F_GETFL) - fcntl.fcntl(read_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - - os.write(write_fd, b"DRAIN_LINE_1\nDRAIN_LINE_2\nDRAIN_LINE_3\n") - os.close(write_fd) - write_fd = -1 - - output_lines: list[str] = [] - buffer = b"" - - drain_deadline = time.monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and time.monotonic() < drain_deadline: - try: - chunk = os.read(read_fd, 4096) - if not chunk: - break - buffer += chunk - drained += len(chunk) - except (BlockingIOError, OSError): - break - - buffer = _flush_lines(buffer, output_lines) - - assert "DRAIN_LINE_1" in output_lines - assert "DRAIN_LINE_2" in output_lines - assert "DRAIN_LINE_3" in output_lines - finally: - os.close(read_fd) - if write_fd != -1: - os.close(write_fd) - - async def test_drain_respects_byte_limit(self) -> None: - """Verify the drain loop stops after MAX_DRAIN_BYTES to prevent - indefinite blocking when a grandchild process holds the PTY open. - - Directly tests the drain logic using a pipe with data exceeding the - byte limit. Uses non-blocking writes to fill the pipe without blocking. - """ - import fcntl - import time - - read_fd, write_fd = os.pipe() - try: - flags = fcntl.fcntl(read_fd, fcntl.F_GETFL) - fcntl.fcntl(read_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - wflags = fcntl.fcntl(write_fd, fcntl.F_GETFL) - fcntl.fcntl(write_fd, fcntl.F_SETFL, wflags | os.O_NONBLOCK) - - total_written = 0 - chunk = b"X" * 4000 + b"\n" - try: - while True: - os.write(write_fd, chunk) - total_written += len(chunk) - except BlockingIOError: - pass - - assert total_written > 0 - - output_lines: list[str] = [] - buffer = b"" - drain_deadline = time.monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and time.monotonic() < drain_deadline: - try: - data = os.read(read_fd, 4096) - if not data: - break - buffer += data - drained += len(data) - except (BlockingIOError, OSError): - break - - buffer = _flush_lines(buffer, output_lines) - - assert drained <= MAX_DRAIN_BYTES - assert len(output_lines) > 0 - finally: - os.close(read_fd) - os.close(write_fd) - - async def test_drain_completes_immediately_on_empty_buffer(self) -> None: - """Verify drain exits quickly when the PTY buffer is empty (EOF).""" - import time - - read_fd, write_fd = os.pipe() - os.close(write_fd) - try: - import fcntl - - flags = fcntl.fcntl(read_fd, fcntl.F_GETFL) - fcntl.fcntl(read_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - - output_lines: list[str] = [] - buffer = b"" - start = time.monotonic() - - drain_deadline = time.monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and time.monotonic() < drain_deadline: - try: - chunk = os.read(read_fd, 4096) - if not chunk: - break - buffer += chunk - drained += len(chunk) - except (BlockingIOError, OSError): - break - - buffer = _flush_lines(buffer, output_lines) - elapsed = time.monotonic() - start - - assert output_lines == [] - assert drained == 0 - assert elapsed < 0.5 - finally: - os.close(read_fd) - - async def test_drain_handles_oserror_gracefully(self) -> None: - """Verify drain exits gracefully when os.read raises OSError (e.g. EIO).""" - import time - - read_fd, write_fd = os.pipe() - os.close(write_fd) - os.close(read_fd) - - output_lines: list[str] = [] - buffer = b"" - - drain_deadline = time.monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and time.monotonic() < drain_deadline: - try: - chunk = os.read(read_fd, 4096) - if not chunk: - break - buffer += chunk - drained += len(chunk) - except (BlockingIOError, OSError): - break - - buffer = _flush_lines(buffer, output_lines) - - assert output_lines == [] - assert drained == 0 - @macos_pty_xfail - async def test_drain_captures_output_without_trailing_newline(self, lease_scope) -> None: + async def test_output_captured_without_trailing_newline(self, lease_scope) -> None: """Verify output without a trailing newline is still captured.""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( @@ -835,343 +579,6 @@ async def test_drain_captures_output_without_trailing_newline(self, lease_scope) info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("NO_NEWLINE_OUTPUT" in call for call in info_calls) - @macos_pty_xfail - async def test_drain_reads_data_remaining_in_pty_buffer(self, lease_scope) -> None: - """Verify the drain loop inside read_pty_output reads data left in the - PTY kernel buffer after the main read loop exits. - - Patches os.read so that, once the main loop has consumed the initial - subprocess output via EOF from the specific PTY fd, a subsequent read - returns additional data - simulating the macOS scenario where the - kernel buffers output that arrives after the reader stop flag is set. - """ - import pty - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo MAIN_OUTPUT", - timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - original_os_read = os.read - original_openpty = pty.openpty - pty_parent_fd = None - eof_seen_on_pty = False - - def tracking_openpty(): - nonlocal pty_parent_fd - parent, child = original_openpty() - pty_parent_fd = parent - return parent, child - - drain_data_returned = False - - def os_read_with_drain_data(fd, size): - nonlocal eof_seen_on_pty, drain_data_returned - if fd != pty_parent_fd: - return original_os_read(fd, size) - if not eof_seen_on_pty: - try: - data = original_os_read(fd, size) - except (BlockingIOError, OSError): - if not eof_seen_on_pty: - eof_seen_on_pty = True - raise - if not data: - eof_seen_on_pty = True - return b"" - return data - if not drain_data_returned: - drain_data_returned = True - return b"DRAIN_CAPTURED\n" - return b"" - - with ( - patch("pty.openpty", side_effect=tracking_openpty), - patch("os.read", side_effect=os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - assert pty_parent_fd is not None - assert eof_seen_on_pty - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("DRAIN_CAPTURED" in call for call in info_calls) - - @macos_pty_xfail - async def test_drain_select_oserror_exits_gracefully(self, lease_scope) -> None: - """Verify the drain loop exits gracefully when select.select() raises - OSError (e.g. fd closed during drain). - - Patches select.select inside the drain to raise OSError, simulating a - closed or invalid fd. The hook should still complete successfully. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker() - - def select_with_oserror(rlist, wlist, xlist, timeout=None): - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - raise OSError("simulated fd closed during drain") - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo SELECT_ERROR_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_with_oserror), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - assert state.eof_seen - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("SELECT_ERROR_TEST" in call for call in info_calls) - - @macos_pty_xfail - async def test_drain_select_valueerror_exits_gracefully(self, lease_scope) -> None: - """Verify the drain loop exits gracefully when select.select() raises - ValueError (e.g. negative fd). - - This covers the except (ValueError, OSError) handler in the drain loop. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker(return_drain_data=False) - - def select_with_valueerror(rlist, wlist, xlist, timeout=None): - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - raise ValueError("file descriptor cannot be a negative integer (-1)") - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo VALUEERROR_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_with_valueerror), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("VALUEERROR_TEST" in call for call in info_calls) - - @macos_pty_xfail - async def test_drain_exits_when_deadline_exceeded_before_select(self, lease_scope) -> None: - """Verify the drain loop exits when the deadline is exceeded between the - while condition and the remaining-time check (line: if remaining <= 0). - - Patches ``jumpstarter.exporter.hooks._monotonic`` (not ``time.monotonic`` - globally) to simulate a jump past the deadline after the while condition - passes but before the remaining check. Using the module-level - ``_monotonic`` reference avoids breaking the asyncio event loop, which - also relies on ``time.monotonic``. - """ - state = _PtyTracker() - clock = _DrainDeadlineClock(_monotonic, state) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo DEADLINE_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks._monotonic", side_effect=clock), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("DEADLINE_TEST" in call for call in info_calls) - # SHOULD_NOT_APPEAR should not be in output because the drain - # exited early due to remaining <= 0 before select could run - assert not any("SHOULD_NOT_APPEAR" in call for call in info_calls) - - @macos_pty_xfail - async def test_drain_exception_is_suppressed(self, lease_scope) -> None: - """Verify that an unexpected exception raised during the drain is caught - by the except-Exception handler and does not propagate to the caller. - - Patches _flush_lines so that the second call (inside the drain) raises - a RuntimeError. The hook should still complete successfully because the - drain's except-Exception block suppresses it. - """ - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo BEFORE_DRAIN_ERROR", - timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - original_flush = _flush_lines - call_count = 0 - - def flush_lines_with_drain_error(buffer, output_lines): - nonlocal call_count - call_count += 1 - result = original_flush(buffer, output_lines) - if call_count > 1: - raise RuntimeError("simulated drain error") - return result - - with ( - patch("jumpstarter.exporter.hooks._flush_lines", side_effect=flush_lines_with_drain_error), - patch("jumpstarter.exporter.hooks.logger"), - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - - @macos_pty_xfail - async def test_drain_retries_empty_select_then_captures_data(self, lease_scope) -> None: - """Verify that the drain retries after empty select() calls and still - captures data that arrives later. - - Patches select.select to return empty for the first N calls (where - N < DRAIN_MAX_EMPTY_POLLS), then reports the fd as readable. The - hook output should still be captured despite the initial empty polls. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker() - empty_count = 0 - empties_before_data = DRAIN_MAX_EMPTY_POLLS - 2 # e.g. 8 empties then data - - def select_with_delayed_ready(rlist, wlist, xlist, timeout=None): - nonlocal empty_count - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - empty_count += 1 - if empty_count <= empties_before_data: - return ([], [], []) # simulate delayed data - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo DELAYED_DRAIN_OK", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_with_delayed_ready), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("DELAYED_DRAIN_OK" in call for call in info_calls) - - @macos_pty_xfail - async def test_drain_terminates_after_max_empty_polls(self, lease_scope) -> None: - """Verify the drain loop terminates after DRAIN_MAX_EMPTY_POLLS - consecutive empty select() results. - - Patches select.select to always return empty during the drain phase. - The hook should still complete (no hang) and the drain data should - not appear since it's never read. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker(return_drain_data=False) - - def select_always_empty(rlist, wlist, xlist, timeout=None): - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - return ([], [], []) # always empty - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo MAX_EMPTY_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_always_empty), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - # Main loop should have captured the output before drain - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("MAX_EMPTY_TEST" in call for call in info_calls) - - @macos_pty_xfail - async def test_drain_empty_counter_resets_on_data(self, lease_scope) -> None: - """Verify the consecutive empty poll counter resets when data arrives. - - Simulates an empty-data-empty pattern during drain: a few empty polls, - then data becomes readable, then more empty polls. The counter should - reset after data is read, so the drain should tolerate more than - DRAIN_MAX_EMPTY_POLLS total empties as long as they are not consecutive. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker() - drain_select_call = 0 - # Pattern: 5 empties, then ready, then 5 more empties, then ready - # Total empties (10) >= DRAIN_MAX_EMPTY_POLLS but never consecutive - pattern = [False] * 5 + [True] + [False] * 5 + [True] - - def select_with_interleaved_empties(rlist, wlist, xlist, timeout=None): - nonlocal drain_select_call - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - idx = drain_select_call - drain_select_call += 1 - if idx < len(pattern) and not pattern[idx]: - return ([], [], []) - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo INTERLEAVE_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_with_interleaved_empties), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("INTERLEAVE_TEST" in call for call in info_calls) - - async def test_drain_constants_are_reasonable(self) -> None: - assert MAX_DRAIN_BYTES == 256 * 1024 - assert DRAIN_TIMEOUT_SECONDS == 2.0 - assert DRAIN_MAX_EMPTY_POLLS == 10 - async def test_exec_default_is_none(self) -> None: """Test that the default exec is None (auto-detect).""" hook = HookInstanceConfigV1Alpha1(script="echo hello") @@ -1181,15 +588,10 @@ async def test_exec_default_is_none(self) -> None: class TestHookExecutorPRRegressions: """Regression tests for issues reported during PR review of hooks feature.""" - @macos_pty_xfail - async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> None: - """Issue A1: Hook infrastructure messages should be at DEBUG, not INFO. - Infrastructure messages like 'Starting hook subprocess', 'Creating PTY', - 'Spawning subprocess', 'Subprocess spawned', 'Subprocess completed', and - 'Hook executed successfully' must be logged at DEBUG level so they don't - appear in the client LogStream at the default INFO level. Only user output - from the hook script should be at INFO. + async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> None: + """Infrastructure messages must be at DEBUG, not INFO, so they + don't appear in the client LogStream. """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="echo 'user output'", timeout=10), @@ -1202,10 +604,8 @@ async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> N debug_calls = [str(call) for call in mock_logger.debug.call_args_list] info_calls = [str(call) for call in mock_logger.info.call_args_list] - # Infrastructure messages should be at DEBUG level infra_messages = [ "Starting hook subprocess", - "Creating PTY", "Spawning subprocess", "Subprocess spawned", "Hook executed successfully", @@ -1218,15 +618,11 @@ async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> N f"Infrastructure message '{msg}' should NOT be at INFO level" ) - # User output should be at INFO level assert any("user output" in call for call in info_calls) async def test_before_lease_hook_always_sets_event_on_failure(self, lease_scope) -> None: - """Issue C3: before_lease_hook event must be set even when hook fails. - - When the beforeLease hook fails with on_failure=endLease, the event must - still be set to unblock process_connections in handle_lease. Otherwise - the lease hangs indefinitely. + """before_lease_hook event must be set even when hook fails, to + unblock process_connections in handle_lease. """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="endLease"), @@ -1244,15 +640,10 @@ async def test_before_lease_hook_always_sets_event_on_failure(self, lease_scope) mock_shutdown, ) - # Event must always be set to unblock connections assert lease_scope.before_lease_hook.is_set() async def test_before_lease_hook_always_sets_event_on_exit(self, lease_scope) -> None: - """Issue C3b: before_lease_hook event must be set when hook fails with exit. - - Same as C3 but for on_failure=exit. The event must be set, shutdown called, - and skip_after_lease_hook set to True. - """ + """before_lease_hook event must be set when hook fails with on_failure=exit.""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), ) @@ -1272,12 +663,7 @@ async def test_before_lease_hook_always_sets_event_on_exit(self, lease_scope) -> mock_shutdown.assert_called_once() async def test_no_hooks_transitions_to_lease_ready(self, lease_scope) -> None: - """Issue D1: No hooks configured should transition directly to LEASE_READY. - - When no hooks are configured, run_before_lease_hook should report - LEASE_READY immediately, preventing the 'create lease, never use → stuck' - scenario. - """ + """No hooks configured should transition directly to LEASE_READY.""" empty_config = HookConfigV1Alpha1() executor = HookExecutor(config=empty_config) @@ -1294,19 +680,13 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # Should have reported LEASE_READY assert any( status == ExporterStatus.LEASE_READY and msg == "Ready for commands" for status, msg in status_calls ), f"Expected LEASE_READY status, got: {status_calls}" async def test_skip_after_lease_prevents_after_hook_execution(self, lease_scope) -> None: - """Issue E1: beforeLease fail+exit should prevent afterLease hook execution. - - When beforeLease fails with on_failure=exit, skip_after_lease_hook is set - to True. The handle_lease finally block checks this flag and skips the - afterLease hook. This test verifies the orchestration sequence. - """ + """beforeLease fail+exit should prevent afterLease hook execution.""" # Config with both hooks hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), @@ -1321,7 +701,6 @@ async def mock_report_status(status, msg): mock_shutdown = MagicMock() - # Run before hook (which fails and sets skip flag) await executor.run_before_lease_hook( lease_scope, mock_report_status, @@ -1330,8 +709,6 @@ async def mock_report_status(status, msg): assert lease_scope.skip_after_lease_hook is True - # Now simulate what handle_lease does: check the flag before running after hook - # This mirrors the actual code: `if not lease_scope.skip_after_lease_hook:` if not lease_scope.skip_after_lease_hook: await executor.run_after_lease_hook( lease_scope, @@ -1339,19 +716,13 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # AFTER_LEASE_HOOK status should never have been reported after_hook_statuses = [s for s, _ in status_calls if s == ExporterStatus.AFTER_LEASE_HOOK] assert len(after_hook_statuses) == 0, ( f"afterLease hook should have been skipped, but AFTER_LEASE_HOOK was reported: {status_calls}" ) async def test_before_hook_exit_reports_failed_not_available(self, lease_scope) -> None: - """Issue E2: beforeLease fail+exit should report FAILED, not AVAILABLE. - - When beforeLease hook fails with on_failure=exit, the last status must be - BEFORE_LEASE_HOOK_FAILED. It should NOT report AVAILABLE, which would - incorrectly tell the controller the exporter is ready for new leases. - """ + """beforeLease fail+exit should report FAILED, not AVAILABLE.""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), ) @@ -1370,35 +741,25 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # Last status should be OFFLINE (reported before shutdown to prevent new leases) last_status, _ = status_calls[-1] assert last_status == ExporterStatus.OFFLINE, ( f"Expected last status to be OFFLINE, got {last_status}" ) - # BEFORE_LEASE_HOOK_FAILED should also be present (reported before OFFLINE) failed_statuses = [s for s, _ in status_calls if s == ExporterStatus.BEFORE_LEASE_HOOK_FAILED] assert len(failed_statuses) > 0, ( f"Expected BEFORE_LEASE_HOOK_FAILED status, got: {status_calls}" ) - # AVAILABLE should never have been reported available_statuses = [s for s, _ in status_calls if s == ExporterStatus.AVAILABLE] assert len(available_statuses) == 0, ( f"AVAILABLE should NOT be reported when beforeLease exits, got: {status_calls}" ) - # Shutdown should have been called with correct args mock_shutdown.assert_called_once_with(exit_code=1, wait_for_lease_exit=True, should_unregister=True) async def test_after_hook_exit_reports_failed_calls_shutdown(self, lease_scope) -> None: - """Issue E3: afterLease fail+exit should report FAILED and call shutdown. - - When afterLease hook fails with on_failure=exit: - - AFTER_LEASE_HOOK_FAILED status must be reported - - AVAILABLE must NOT be reported - - shutdown must be called (not request_lease_release) - """ + """afterLease fail+exit should report FAILED and call shutdown.""" hook_config = HookConfigV1Alpha1( after_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), ) @@ -1419,27 +780,22 @@ async def mock_report_status(status, msg): mock_request_release, ) - # AFTER_LEASE_HOOK_FAILED should be in statuses failed_statuses = [s for s, _ in status_calls if s == ExporterStatus.AFTER_LEASE_HOOK_FAILED] assert len(failed_statuses) > 0, ( f"Expected AFTER_LEASE_HOOK_FAILED status, got: {status_calls}" ) - # AVAILABLE should NOT be in statuses available_statuses = [s for s, _ in status_calls if s == ExporterStatus.AVAILABLE] assert len(available_statuses) == 0, ( f"AVAILABLE should NOT be reported when afterLease exits, got: {status_calls}" ) - # Shutdown called (not request_lease_release) mock_shutdown.assert_called_once_with(exit_code=1, should_unregister=True, wait_for_lease_exit=True) mock_request_release.assert_not_called() async def test_before_hook_warn_includes_warning_prefix(self, lease_scope) -> None: - """Issue E5: beforeLease hook fail with warn should include HOOK_WARNING_PREFIX. - - The status message for LEASE_READY must start with '[HOOK_WARNING] ' so that - shell.py can detect it and display a user-visible warning. + """beforeLease hook fail with warn should include HOOK_WARNING_PREFIX + so shell.py can detect and display a user-visible warning. """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="warn"), @@ -1459,7 +815,6 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # Find the LEASE_READY status call ready_calls = [(s, m) for s, m in status_calls if s == ExporterStatus.LEASE_READY] assert len(ready_calls) == 1, f"Expected exactly one LEASE_READY, got: {status_calls}" _, msg = ready_calls[0] @@ -1468,11 +823,7 @@ async def mock_report_status(status, msg): ) async def test_before_hook_exit_reports_offline_before_shutdown(self, lease_scope) -> None: - """When beforeLease hook fails with on_failure=exit, the exporter must - report OFFLINE status to the controller before initiating shutdown. - This prevents the controller from assigning new leases to a dying - exporter during the shutdown window. - """ + """OFFLINE must be reported before shutdown to prevent new lease assignment.""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), ) @@ -1572,11 +923,9 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # beforeLease with warn should still transition to LEASE_READY ready_calls = [s for s, _ in status_calls if s == ExporterStatus.LEASE_READY] assert len(ready_calls) == 1 - # Now run afterLease (simulating premature lease-end cleanup) await executor.run_after_lease_hook( lease_scope, mock_report_status, @@ -1584,18 +933,45 @@ async def mock_report_status(status, msg): mock_request_release, ) - # afterLease hook should run and transition to AVAILABLE available_calls = [s for s, _ in status_calls if s == ExporterStatus.AVAILABLE] assert len(available_calls) > 0, ( f"Expected AVAILABLE status after warn+afterLease, got: {status_calls}" ) - async def test_after_hook_warn_includes_warning_prefix(self, lease_scope) -> None: - """Issue E5b: afterLease hook fail with warn should include HOOK_WARNING_PREFIX. + async def test_after_lease_hook_skips_when_lease_context_not_ready(self) -> None: + from anyio import Event - The status message for AVAILABLE must start with '[HOOK_WARNING] ' so that - shell.py can detect it and display a user-visible warning after session ends. - """ + from jumpstarter.exporter.lease_context import LeaseContext + + hook_config = HookConfigV1Alpha1( + after_lease=HookInstanceConfigV1Alpha1(script="echo should-not-run", timeout=10), + ) + executor = HookExecutor(config=hook_config) + + lease_scope = LeaseContext( + lease_name="test-lease", + before_lease_hook=Event(), + client_name="test-client", + ) + + status_calls: list[tuple] = [] + + async def mock_report_status(status, msg): + status_calls.append((status, msg)) + + mock_shutdown = MagicMock() + + await executor.run_after_lease_hook( + lease_scope, + mock_report_status, + mock_shutdown, + ) + + assert any(s == ExporterStatus.AVAILABLE for s, _ in status_calls) + assert not any(s == ExporterStatus.AFTER_LEASE_HOOK for s, _ in status_calls) + + async def test_after_hook_warn_includes_warning_prefix(self, lease_scope) -> None: + """afterLease hook fail with warn should include HOOK_WARNING_PREFIX.""" hook_config = HookConfigV1Alpha1( after_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="warn"), ) @@ -1614,7 +990,6 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # Find the AVAILABLE status call available_calls = [(s, m) for s, m in status_calls if s == ExporterStatus.AVAILABLE] assert len(available_calls) == 1, f"Expected exactly one AVAILABLE, got: {status_calls}" _, msg = available_calls[0] @@ -1795,3 +1170,333 @@ async def mock_report_status(status, msg): assert lease_scope.before_lease_hook.is_set(), ( "before_lease_hook event must be set to unblock downstream waiters" ) + + +class TestPipeOutputEdgeCases: + """Edge cases for pipe-based output capture (PR #837).""" + + async def test_stderr_captured_via_pipe_merge(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo STDOUT_LINE; echo STDERR_LINE >&2", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("STDOUT_LINE" in call for call in info_calls) + assert any("STDERR_LINE" in call for call in info_calls) + + async def test_large_output_spanning_multiple_reads(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script=( + "seq 1 200 | while read n; do " + "echo \"LINE_${n}_PADDING_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX\"; " + "done" + ), + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("LINE_1_" in call for call in info_calls) + assert any("LINE_100_" in call for call in info_calls) + assert any("LINE_200_" in call for call in info_calls) + + async def test_non_utf8_output_decoded_with_replacement(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="python3", + script=( + "import sys; " + "sys.stdout.buffer.write(b'VALID_PREFIX\\x80VALID_SUFFIX\\n')" + ), + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + matching = [ + call for call in info_calls + if "VALID_PREFIX" in call and "VALID_SUFFIX" in call + ] + assert len(matching) > 0 + + async def test_rapid_exit_with_buffered_output(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo FAST_1; echo FAST_2; echo FAST_3; echo FAST_4; echo FAST_5", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + for i in range(1, 6): + assert any(f"FAST_{i}" in call for call in info_calls), ( + f"FAST_{i} was not captured" + ) + + async def test_spawn_failure_cleans_up_without_crash(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="/nonexistent/interpreter", + script="echo should not run", + timeout=10, + on_failure="warn", + ), + ) + executor = HookExecutor(config=hook_config) + + result = await executor.execute_before_lease_hook(lease_scope) + assert result is not None + assert "error" in result.lower() + + async def test_interleaved_stdout_and_stderr_captured(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script=( + "echo OUT_1; echo ERR_1 >&2; " + "echo OUT_2; echo ERR_2 >&2; " + "echo OUT_3" + ), + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + for label in ("OUT_1", "OUT_2", "OUT_3", "ERR_1", "ERR_2"): + assert any(label in call for call in info_calls), ( + f"{label} was not captured" + ) + + async def test_timeout_with_grandchild_holding_pipe(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo GRANDCHILD_TEST; sleep 10 &", + timeout=2, + on_failure="warn", + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is not None + assert "timed out" in result.lower() + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("GRANDCHILD_TEST" in call for call in info_calls) + + + async def test_timeout_cleanup_handles_process_lookup_errors(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="python3", + script=( + "import signal, time\n" + "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" + "time.sleep(300)\n" + ), + timeout=1, + on_failure="warn", + ), + ) + executor = HookExecutor(config=hook_config) + + original_killpg = os.killpg + + def killpg_raises_after_real_signal(pgid, sig): + try: + original_killpg(pgid, sig) + except ProcessLookupError: + pass + raise ProcessLookupError + + with patch("os.killpg", side_effect=killpg_raises_after_real_signal): + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is not None + assert "timed out" in result.lower() + + async def test_exception_during_hook_triggers_finally_cleanup(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="python3", + script=( + "import signal, time\n" + "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" + "time.sleep(300)\n" + ), + timeout=30, + on_failure="warn", + ), + ) + executor = HookExecutor(config=hook_config) + + original_wait = subprocess.Popen.wait + original_killpg = os.killpg + wait_calls = [0] + + def failing_then_real_wait(self_popen, timeout=None): + wait_calls[0] += 1 + if wait_calls[0] == 1: + import time as _time + + _time.sleep(0.3) + raise RuntimeError("simulated wait failure") + return original_wait(self_popen, timeout=timeout) + + def killpg_raises_after_real_signal(pgid, sig): + try: + original_killpg(pgid, sig) + except ProcessLookupError: + pass + raise ProcessLookupError + + with ( + patch.object(subprocess.Popen, "wait", failing_then_real_wait), + patch("os.killpg", side_effect=killpg_raises_after_real_signal), + ): + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is not None + assert "error" in result.lower() + + +class TestReadOutputErrorPaths: + """Tests for BlockingIOError and OSError handling in read_output. + + These exercise the read_output error paths via real subprocesses + that produce controlled output patterns, and via _flush_lines + for the partial buffer flush path. + """ + + async def test_blocking_io_error_path_via_nonblocking_pipe(self, lease_scope) -> None: + """On a non-blocking pipe, reading before data arrives raises + BlockingIOError. read_output handles this by continuing the loop. + Verified via a script that delays output. + """ + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="sleep 0.1; echo DELAYED_OUTPUT", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("DELAYED_OUTPUT" in call for call in info_calls) + + async def test_reader_exits_on_eof(self, lease_scope) -> None: + """read_output exits cleanly when os.read returns empty bytes (EOF).""" + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo BEFORE_CLOSE", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("BEFORE_CLOSE" in call for call in info_calls) + + def test_flush_lines_partial_buffer_preserved(self) -> None: + """Partial buffer (no trailing newline) is returned for later flush.""" + output: list[str] = [] + remainder = _flush_lines(b"complete\npartial_data", output) + assert output == ["complete"] + assert remainder == b"partial_data" + + async def test_partial_buffer_flushed_on_exit(self, lease_scope) -> None: + """When the subprocess exits with output lacking a trailing newline, + the finally block in read_output flushes the partial buffer. + """ + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="printf 'NO_TRAILING_NEWLINE'", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("NO_TRAILING_NEWLINE" in call for call in info_calls) + + async def test_oserror_during_pipe_read_exits_gracefully(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo done", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + async def wait_readable_oserror(fd): + raise OSError("simulated fd error") + + with patch("anyio.wait_readable", side_effect=wait_readable_oserror): + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + debug_calls = [str(call) for call in mock_logger.debug.call_args_list] + assert any("OSError" in call for call in debug_calls) + + async def test_mixed_complete_and_partial_lines(self, lease_scope) -> None: + """Complete lines are flushed immediately; the trailing partial + is flushed when the subprocess exits. + """ + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="printf 'LINE_A\\nLINE_B\\nPARTIAL_C'", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("LINE_A" in call for call in info_calls) + assert any("LINE_B" in call for call in info_calls) + assert any("PARTIAL_C" in call for call in info_calls)