diff --git a/src/openjd/sessions/_action_filter.py b/src/openjd/sessions/_action_filter.py index 25d43aa..4c49e37 100644 --- a/src/openjd/sessions/_action_filter.py +++ b/src/openjd/sessions/_action_filter.py @@ -95,7 +95,14 @@ class ActionMessageKind(Enum): ) filter_matcher = re.compile(filter_regex) -openjd_env_actions_filter_regex = "^(openjd_env|openjd_redacted_env|openjd_unset_env)" +# The near-miss detector for env macros. The trailing `(?::|\s|\Z)` is +# load-bearing: without it any line merely *starting with* one of these tokens +# is reported as a malformed macro and FAILS the action, so a script logging +# `openjd_environment: ready` or `openjd_env_setup: done` kills its own task. +# openjd-rs anchors the same three tokens on `:`, a space, or end-of-string +# (action_filter.rs is_malformed_env_command, whose comment names exactly this +# false positive), so this is parity as well as a bug fix. +openjd_env_actions_filter_regex = r"^(openjd_env|openjd_redacted_env|openjd_unset_env)(?::|\s|\Z)" openjd_env_actions_filter_matcher = re.compile(openjd_env_actions_filter_regex) # A regex for matching the assignment of a value to an environment variable diff --git a/src/openjd/sessions/_runner_base.py b/src/openjd/sessions/_runner_base.py index 1d8e917..c98c13c 100644 --- a/src/openjd/sessions/_runner_base.py +++ b/src/openjd/sessions/_runner_base.py @@ -5,11 +5,12 @@ import re import stat import shlex +import sys from abc import ABC, abstractmethod from concurrent.futures import Future, ThreadPoolExecutor from dataclasses import dataclass from datetime import datetime, timedelta, timezone -from enum import Enum +from enum import Enum, auto from pathlib import Path from threading import Lock, Timer from typing import Any, Callable, Literal, Optional, Sequence, Type, cast @@ -20,12 +21,6 @@ from openjd.model import FormatStringError from openjd.model import evaluate_let_bindings -# The EXPR engine's typed value. Imported concretely (rather than duck-typed -# with getattr) so that a model API change fails loudly at import time instead -# of silently mis-classifying every optional integer field as "omitted". -# openjd.expr ships in the same distribution as openjd.model, which this module -# already hard-imports unreleased API from. -from openjd.expr import ExprValue, TypeCode from openjd.model.v2023_09 import Action as Action_2023_09 from openjd.model.v2023_09 import CancelationMethodDeferred as CancelationMethodDeferred_2023_09 from openjd.model.v2023_09 import CancelationMode as CancelationMode_2023_09 @@ -132,6 +127,79 @@ def _over_range_message(description: str, value: int) -> str: return f"{description} must be at most {MAX_INT_FIELD_VALUE}, got '{value}'" +# Why the EXPR engine's types are reached lazily, from here down: +# +# ``openjd.expr`` is a thin facade over the native ``openjd._openjd_rs`` +# extension, so a module-level import of it makes ``import openjd.sessions`` +# load the extension unconditionally -- including for a worker that only ever +# runs non-EXPR templates. openjd.model deliberately avoids that (see +# ``openjd.model._format_strings._parser``, which duplicates a constant rather +# than import the Rust surface), and this module matches that discipline. +# +# The guard is on ``sys.modules`` rather than on the value's type, because +# ``FormatString.resolve_value`` is not limited to ``str`` and ``ExprValue``: +# a legacy (non-EXPR) whole-field interpolation returns the symbol's own value, +# so an ``int`` reaches here on a purely non-EXPR path. Testing "not a str" +# would therefore still load the extension for e.g. an integer-valued +# ``timeout``. +_EXTENSION_MODULE = "openjd._openjd_rs" + + +class _ExprKind(Enum): + """How a resolved format-string value relates to the EXPR type system.""" + + NOT_EXPR = auto() + """Not an ``ExprValue``: a plain string, or a legacy interpolation's own + value. Resolves to its string form; typed semantics do not apply.""" + + NULL = auto() + """The engine's typed null -- "field omitted" / "argument skipped".""" + + LIST = auto() + """A list, whose elements flatten to one argument each (RFC 0005 §1.3.2).""" + + SCALAR = auto() + """Any other typed value; resolves to its string form.""" + + +def _classify_expr_value(value: Any) -> _ExprKind: + """Classify ``value`` against the EXPR type system. + + This is the *only* place in this module that imports ``openjd.expr``, so + that every crossing into the native extension sits behind the one + ``sys.modules`` guard below. Doing the whole classification here rather + than exposing a separate "is it a list" predicate keeps that property + structural instead of merely documented: there is no second, unguarded + entry point for a future caller to reach with an arbitrary value. + + ``ExprValue`` instances are created only by the native extension, so if + that extension has not been loaded then ``value`` cannot be one and the + answer is ``NOT_EXPR`` without importing anything. Once it *has* been + loaded -- i.e. an EXPR expression has been evaluated in this process -- + the import is a ``sys.modules`` hit. + + ``ExprValue`` is imported concretely (rather than duck-typed with + ``getattr``) so that a model API change fails loudly here instead of + silently mis-classifying every optional integer field as "omitted". + """ + if _EXTENSION_MODULE not in sys.modules: + return _ExprKind.NOT_EXPR + from openjd.expr import ExprValue, TypeCode + + if not isinstance(value, ExprValue): + return _ExprKind.NOT_EXPR + if value.is_null: + return _ExprKind.NULL + if value.type.type_code == TypeCode.LIST: + return _ExprKind.LIST + return _ExprKind.SCALAR + + +def _is_expr_null(value: Any) -> bool: + """True if ``value`` is the EXPR engine's typed null.""" + return _classify_expr_value(value) is _ExprKind.NULL + + def _timeout_from_seconds(seconds: int, logger: LoggerAdapter) -> Optional[timedelta]: """Convert a resolved timeout in seconds into an enforceable time limit. @@ -236,13 +304,20 @@ def resolve_action_arg_values(args: Optional[Sequence], symtab: SymbolTable) -> # argument is genuinely unresolvable. resolved.append(arg.resolve(symtab=symtab)) continue - if isinstance(value, str): - resolved.append(value) - elif value.is_null: + kind = _classify_expr_value(value) + if kind is _ExprKind.NOT_EXPR: + # A plain string, or -- for a legacy (non-EXPR) whole-field + # interpolation -- the symbol's own value, which may be any + # Python type a caller put in the symbol table (an ``int`` for + # an INT job parameter, a ``bool`` for a BOOL one). Typed + # null/list semantics exist only under EXPR whole-field + # resolution, so everything here resolves to its string form. + resolved.append(value if isinstance(value, str) else str(value)) + elif kind is _ExprKind.NULL: continue - elif value.type.type_code == TypeCode.LIST: + elif kind is _ExprKind.LIST: resolved.extend(str(element) for element in value) - else: + else: # _ExprKind.SCALAR resolved.append(str(value)) return resolved @@ -302,7 +377,7 @@ def resolve_optional_int_field( # to plain string resolution — correct, since typed nulls only exist # under EXPR whole-field semantics (Template Schemas 5.3). resolved_value = value.resolve_value(symtab=symtab) - if isinstance(resolved_value, ExprValue) and resolved_value.is_null: + if _is_expr_null(resolved_value): return None resolved = str(resolved_value) # Strict ASCII integer grammar, matching the Rust runtime's str::parse @@ -381,7 +456,7 @@ def resolve_period(period: Any) -> Optional[int]: # openjd-rs runtime, which errors on any non-null, non-mode-name # result). mode_value = cancelation.mode.resolve_value(symtab=symtab) - if isinstance(mode_value, ExprValue) and mode_value.is_null: + if _is_expr_null(mode_value): # Null mode drops the ENTIRE cancelation object: mode is the # object's required discriminator, so an "omitted" mode cannot # leave a partial object behind. The action behaves exactly as @@ -1251,7 +1326,14 @@ def _cancel( write_file_for_user( self._session_working_directory / "cancel_info.json", notify_end, self._user ) - except OSError as err: + except Exception as err: + # `Exception`, not `OSError`: write_file_for_user reaches + # WindowsPermissionHelper, which raises RuntimeError (not an + # OSError) when it cannot set an ACL for an impersonated user. + # With the narrower handler that escaped here and unwound the + # runtime-limit Timer thread -- no notify, no grace timer, a + # live child, and the timeout silently unenforced. + # # F6 fix: If we cannot write the cancel_info.json (disk full, permission # denied, etc.), log and fall back to immediate termination. A script # waiting on that file would hang forever otherwise. diff --git a/src/openjd/sessions/_session.py b/src/openjd/sessions/_session.py index 73063e1..e950393 100644 --- a/src/openjd/sessions/_session.py +++ b/src/openjd/sessions/_session.py @@ -279,6 +279,23 @@ class Session(object): """OS environment variables defined by Open Job Description Environments """ + _session_env_vars: dict[str, str] + """Session-defined environment variables, for the lifetime of the Session. + + Deliberately separate from :attr:`_created_env_vars`, which is keyed by + environment and read through ``_environments_entered`` to build the child + *process* environment -- that view must shrink when an environment exits. + This one must not: RFC 0008 (``rfcs/0008-environment-wrap-actions.md``, + "MUST include in ``WrappedAction.Environment`` every ``openjd_env``-defined + variable emitted by any earlier action in the same session") makes + session-lifetime inclusion a requirement, so a wrap script can forward a + variable exported by an environment that has since exited. + + The only remover is an explicit ``openjd_unset_env``. openjd-rs holds the + same split -- its session-lifetime ``env_vars`` beside its per-environment + ``created_env_vars`` -- and this mirrors it. + """ + _wrap_env_file_records: dict[EnvironmentIdentifier, list["_FileRecord"]] """RFC 0008: per wrap environment, the embedded-file records whose on-disk paths were allocated on the environment's first wrap-hook invocation and @@ -437,6 +454,7 @@ def __init__( self._running_environment_identifier = None self._process_env = dict(os_env_vars) if os_env_vars else dict() self._created_env_vars = dict() + self._session_env_vars = dict() self._wrap_env_file_records = dict() self._retain_working_dir = retain_working_dir self._user = user @@ -872,6 +890,10 @@ def enter_environment( ) env_var_changes = SimplifiedEnvironmentVariableChanges(resolved_variables) self._created_env_vars[identifier] = env_var_changes + # Session-lifetime copy for WrappedAction.Environment. openjd-rs + # writes declarative `variables:` into its session-lifetime + # `env_vars` alongside `created_env_vars` at the same point. + self._session_env_vars.update(resolved_variables) else: # Running the environment may define environment variable # mutations via its stdout. We create an empty env changes @@ -883,7 +905,14 @@ def enter_environment( # Must be called _after_ we append to _environments_entered action_env_vars = self._evaluate_current_session_env_vars(os_env_vars) - self._materialize_path_mapping(environment.revision, action_env_vars, symtab) + try: + self._materialize_path_mapping(environment.revision, action_env_vars, symtab) + except RuntimeError as e: + self._fail_action_before_start(str(e)) + # The identifier is still returned on a pre-start failure: the + # environment is on the entered stack, so the caller must be able to + # exit it. Same contract as the resolve-variables failure above. + return identifier # Note: RUNNING is set below, immediately before the runner is asked to # start, and never before `self._runner` exists. `cancel_action()` is a @@ -1045,7 +1074,11 @@ def exit_environment( self._running_environment_identifier = identifier symtab = self._symbol_table(environment.revision) - self._materialize_path_mapping(environment.revision, action_env_vars, symtab) + try: + self._materialize_path_mapping(environment.revision, action_env_vars, symtab) + except RuntimeError as e: + self._fail_action_before_start(str(e)) + return # Re-seed the owning step's name (Step.Name, RFC 0005 EXPR) and # re-apply the extra `let` bindings this environment was entered with @@ -1222,7 +1255,11 @@ def run_task( if step_name is not None: symtab["Step.Name"] = step_name action_env_vars = self._evaluate_current_session_env_vars(os_env_vars) - self._materialize_path_mapping(step_script.revision, action_env_vars, symtab) + try: + self._materialize_path_mapping(step_script.revision, action_env_vars, symtab) + except RuntimeError as e: + self._fail_action_before_start(str(e)) + return # Note: the step script's EXPR `let` bindings (RFC 0005) are evaluated # by the script runner, after embedded-file paths are allocated (so @@ -1356,7 +1393,11 @@ def _run_task_without_session_env( if os_env_vars: action_env_vars.update(**os_env_vars) - self._materialize_path_mapping(step_script.revision, action_env_vars, symtab) + try: + self._materialize_path_mapping(step_script.revision, action_env_vars, symtab) + except RuntimeError as e: + self._fail_action_before_start(str(e)) + return # Note: the step script's EXPR `let` bindings (RFC 0005) are evaluated # by the script runner, after embedded-file paths are allocated. @@ -1883,31 +1924,26 @@ def _try_inject_wrapped_symbols( return True def _collect_session_env_list(self) -> list[str]: - """Collect all session-defined variables across the active - environment stack as ``["KEY=value", ...]`` for + """The session-defined variables as ``["KEY=value", ...]`` for ``WrappedAction.Environment``. - Session-defined variables are ``openjd_env`` stdout definitions - *and* entered environments' declarative ``variables:`` maps - (openjd-rs #277); host-inherited variables remain intentionally - excluded per RFC 0008. - - The per-environment changes are flattened cumulatively, in - environment-entry order, so the list reflects the *effective* - state exactly like the real subprocess environment does: a later - set overrides an earlier value for the same name (one entry, not - two), and a later unset removes the name entirely. This mirrors - the Rust runtime's single cumulative ``env_vars`` map.""" - effective: dict[str, Optional[str]] = {} - for env_id in self._environments_entered: - if env_id in self._created_env_vars: - changes = self._created_env_vars[env_id] - # effective_items() is insertion-ordered, holding both the - # environment's declarative `variables:` seed and any - # openjd_env sets/unsets, so the list order is deterministic. - for key, value in changes.effective_items().items(): - effective[key] = value - return [f"{key}={value}" for key, value in effective.items() if value is not None] + Session-defined variables are ``openjd_env`` stdout definitions *and* + entered environments' declarative ``variables:`` maps (openjd-rs #277); + host-inherited variables remain intentionally excluded per RFC 0008. + + Read from :attr:`_session_env_vars`, which is **session-lifetime**, not + from ``_environments_entered``. This used to walk the entered stack and + flatten each environment's changes, which meant an export vanished from + this symbol the moment its environment exited -- a violation of RFC + 0008's MUST that every ``openjd_env`` variable from any earlier action + in the session be included. The child *process* environment is the view + that correctly shrinks on exit; see ``_evaluate_current_session_env_vars``. + + ``_session_env_vars`` is insertion-ordered and already holds the + effective value per name -- a later set overwrites in place, and an + explicit ``openjd_unset_env`` removes the name -- so this is a + formatting step only.""" + return [f"{name}={value}" for name, value in self._session_env_vars.items()] def _resolve_action_timeout(self, action: Any, symtab: SymbolTable) -> Optional[int]: """Return the wrapped action's timeout as an int (seconds), or @@ -2135,9 +2171,20 @@ def _materialize_path_mapping( ParameterValueType.BOOL.value ) rules_json = json.dumps(rules_dict) - file_handle, filename = mkstemp(dir=self.working_directory, suffix=".json", text=True) - os.close(file_handle) - write_file_for_user(Path(filename), rules_json, self._user) + # An OSError here used to escape run_task/enter_environment/exit_environment + # with no callback and no ActionStatus, because all three call this before + # their runner exists. openjd-rs propagates the same failure as + # SessionError::WorkingDirectory and every call site maps it to + # fail_action_setup(), which publishes FAILED and notifies -- so translate + # to a RuntimeError the callers already turn into that terminal status. + try: + file_handle, filename = mkstemp(dir=self.working_directory, suffix=".json", text=True) + os.close(file_handle) + write_file_for_user(Path(filename), rules_json, self._user) + except Exception as err: + raise RuntimeError( + f"Could not write the path mapping rules file in {self.working_directory}: {err}" + ) from err symtab[ValueReferenceConstants_2023_09.PATH_MAPPING_RULES_FILE.value] = str(filename) symtab.expr_types[ValueReferenceConstants_2023_09.PATH_MAPPING_RULES_FILE.value] = ( ParameterValueType.PATH.value @@ -2220,6 +2267,10 @@ def _action_log_filter_callback( env_vars.simplify_ordered_changes( changes=[EnvironmentVariableSetChange(name=value["name"], value=value["value"])] ) + # Session-lifetime copy for WrappedAction.Environment; see + # _session_env_vars. Runs on the LoggingSubprocess IO thread, like + # the line above -- a plain dict assignment, so no new hazard. + self._session_env_vars[value["name"]] = value["value"] return elif kind == ActionMessageKind.UNSET_ENV: if self._running_environment_identifier is None: @@ -2242,6 +2293,9 @@ def _action_log_filter_callback( assert isinstance(value, str) env_vars = self._created_env_vars[self._running_environment_identifier] env_vars.simplify_ordered_changes(changes=[EnvironmentVariableUnsetChange(name=value)]) + # An explicit unset is the one remover from the session-lifetime + # map, matching openjd-rs. Environment exit is not. + self._session_env_vars.pop(value, None) return else: # ActionMessageKind.SESSION_RUNTIME_LOGLEVEL assert isinstance(value, int) diff --git a/src/openjd/sessions/_subprocess.py b/src/openjd/sessions/_subprocess.py index a48bd02..d1c7335 100644 --- a/src/openjd/sessions/_subprocess.py +++ b/src/openjd/sessions/_subprocess.py @@ -175,11 +175,21 @@ def run(self) -> None: if self._has_started.is_set(): raise RuntimeError("The process has already been run") - self._process = self._start_subprocess() - # Set _has_started regardless of whether we started the process successfully or - # not. That will wake up anyone waiting on wait_until_started() to know whether - # we've gotten this far. - self._has_started.set() + try: + self._process = self._start_subprocess() + finally: + # Set _has_started regardless of whether we started the process successfully or + # not. That will wake up anyone waiting on wait_until_started() to know whether + # we've gotten this far. + # + # In a `finally` because `_start_subprocess` can also *raise*: its + # `except Exception` handler logs "Process failed to start", and that + # log can itself fail (a raising `logging.Filter`), which propagates + # out past this point. `ScriptRunnerBase._run` waits on + # `wait_until_started()` with no timeout, so leaving the event unset + # hangs the thread that called the public `Session` API -- forever, + # on a launch that has already definitively failed. + self._has_started.set() if self._process is None: # We failed to start the subprocess self._start_failed = True @@ -187,57 +197,72 @@ def run(self) -> None: self._callback() return - self._pid = self._process.pid - - # Would use is_posix(), but it doesn't short-circuit mypy which then complains - # about os.getpgid not being a valid attribute. - if not sys.platform == "win32": - # A trivial command can exit before we get here. Looking up the - # process group of an already-reaped child raises ProcessLookupError - # (ESRCH), which must not fail the action: the child ran, and its - # exit code is still collected below. - try: - if not self._user or self._user.is_process_user(): - self._sudo_child_process_group_id = os.getpgid(self._process.pid) - else: - self._sudo_child_process_group_id = find_sudo_child_process_group_id( - logger=self._logger, - sudo_process=self._process, + # Bind the Popen once, and enter the ownership `try` immediately: from here + # on this method owns a live child, so every exit path -- including an + # exceptional one -- has to go through the `finally` that reaps it. + # + # The process-group lookup and the two startup log lines used to sit + # *before* the `try`, outside that guarantee. A `logging.Filter` that + # raises (filters propagate out of `Logger.handle`, unlike handlers) then + # escaped `run()` having never waited on the child: a live process with no + # exit code recorded, while the runner published a terminal state for it. + # Installing such a filter is a supported use of this library -- `Session` + # installs its own. Reproduced before this change; see + # TestStartupLoggingDoesNotAbandonTheChild. + proc = self._process + try: + self._pid = proc.pid + + # Would use is_posix(), but it doesn't short-circuit mypy which then complains + # about os.getpgid not being a valid attribute. + if not sys.platform == "win32": + # A trivial command can exit before we get here. Looking up the + # process group of an already-reaped child raises ProcessLookupError + # (ESRCH), which must not fail the action: the child ran, and its + # exit code is still collected below. + try: + if not self._user or self._user.is_process_user(): + self._sudo_child_process_group_id = os.getpgid(proc.pid) + else: + self._sudo_child_process_group_id = find_sudo_child_process_group_id( + logger=self._logger, + sudo_process=proc, + ) + except ProcessLookupError: + # R5-9 fix: leave this as None -- "no process group known" -- + # rather than substituting the pid. The lookup failed *because* + # the process is gone, so its pid is dead: signalling it is at + # best a no-op, and after pid recycling on a busy host `killpg` + # would target an unrelated group. In the sudo branch the pid is + # not even the right identifier, since it belongs to `sudo` and + # not to the workload's process group. + # + # None is the established "unknown" value on this field -- + # find_sudo_child_process_group_id returns it for the same reason + # (_linux/_sudo.py), and _posix_signal_subprocess already retries + # the lookup and then declines to signal when it is still unset. + self._sudo_child_process_group_id = None + self._logger.debug( + f"Process {proc.pid} exited before its process group could be " + "determined; no process group will be signalled for it.", + extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), ) - except ProcessLookupError: - # R5-9 fix: leave this as None -- "no process group known" -- - # rather than substituting the pid. The lookup failed *because* - # the process is gone, so its pid is dead: signalling it is at - # best a no-op, and after pid recycling on a busy host `killpg` - # would target an unrelated group. In the sudo branch the pid is - # not even the right identifier, since it belongs to `sudo` and - # not to the workload's process group. - # - # None is the established "unknown" value on this field -- - # find_sudo_child_process_group_id returns it for the same reason - # (_linux/_sudo.py), and _posix_signal_subprocess already retries - # the lookup and then declines to signal when it is still unset. - self._sudo_child_process_group_id = None - self._logger.debug( - f"Process {self._process.pid} exited before its process group could be " - "determined; no process group will be signalled for it.", - extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), - ) - self._logger.info( - f"Command started as pid: {self._process.pid}", - extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), - ) - self._logger.info( - "Output:", - extra=LogExtraInfo(openjd_log_content=LogContent.BANNER | LogContent.COMMAND_OUTPUT), - ) + self._logger.info( + f"Command started as pid: {proc.pid}", + extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), + ) + self._logger.info( + "Output:", + extra=LogExtraInfo( + openjd_log_content=LogContent.BANNER | LogContent.COMMAND_OUTPUT + ), + ) - try: self._log_subproc_stdout() # Blocking - self._process.wait() - self._returncode = self._process.returncode - self._log_returncode() + proc.wait() + self._returncode = proc.returncode + self._log_returncode(proc) if self._callback: self._callback() finally: @@ -245,7 +270,6 @@ def run(self) -> None: # PopenWindowsAsUser and there's stuff to deallocate. `proc` itself is # a function local and dies with the frame; a `del proc` here would be # a no-op, which is what CodeQL's "unnecessary delete" reports. - proc = self._process self._process = None # This method owns the subprocess for its whole life, and clearing # `_process` is the point after which nothing else can reach it: @@ -276,21 +300,20 @@ def _reap(self, proc: Optional[Popen]) -> None: # evil -- the alternative is an orphan holding the session directory # and, for a cross-user action, running as the job user with no # owner. - self._logger.error( + # + # Every log call in this method goes through + # `_log_error_best_effort`: the failure that routes us here is a + # raising `logging.Filter`, so the log call in front of the kill was + # itself the thing that skipped the kill. + self._log_error_best_effort( f"Abandoning the subprocess {proc.pid} before it exited; terminating it so it " - "is not orphaned.", - extra=LogExtraInfo( - openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO - ), + "is not orphaned." ) try: self._terminate_process(proc) except Exception as e: # noqa: BLE001 - self._logger.error( - f"Could not terminate the abandoned subprocess {proc.pid}: {e}", - extra=LogExtraInfo( - openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO - ), + self._log_error_best_effort( + f"Could not terminate the abandoned subprocess {proc.pid}: {e}" ) try: # Bounded: a SIGKILLed child is reaped promptly, and this runs on @@ -298,13 +321,10 @@ def _reap(self, proc: Optional[Popen]) -> None: # runner's future rather than surface anything useful. proc.wait(timeout=ABANDONED_PROCESS_REAP_TIMEOUT_SECONDS) except TimeoutExpired: - self._logger.error( + self._log_error_best_effort( f"Subprocess {proc.pid} did not exit within " f"{ABANDONED_PROCESS_REAP_TIMEOUT_SECONDS}s of being terminated; it is left " - "unreaped.", - extra=LogExtraInfo( - openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO - ), + "unreaped." ) # Record the exit status. Unconditional on purpose: `Popen.returncode` is # populated by the poll()/wait() above and is the authority either way -- @@ -317,6 +337,41 @@ def _reap(self, proc: Optional[Popen]) -> None: # it left every test passing. An unfalsifiable check is worse than none. self._returncode = proc.returncode + def _log_error_best_effort(self, message: str) -> None: + """Log ``message`` at ERROR, swallowing any failure of the logging stack + itself. + + :meth:`_reap` runs inside :meth:`run`'s ``finally``, so anything raised + from it replaces the exception already propagating and hides the real + cause. That containment was already in place for the terminate and the + wait, but not for the log calls between them -- and the failure mode that + routes us into ``_reap`` in the first place is a ``logging.Filter`` that + raises, which makes those calls the *likeliest* thing to raise here, not + the least. + + Silently swallowing is normally the wrong answer. It is right here + because the alternative is strictly worse on both counts: the caller + loses the genuine exception, and the child is left unreaped. + + Catches ``Exception``, deliberately not ``BaseException``. A + ``KeyboardInterrupt`` or ``SystemExit`` must still propagate, and neither + is reachable here anyway: ``run()`` executes on a ``ThreadPoolExecutor`` + worker, and CPython delivers those to the main thread. The kill is + therefore not ordered around this call -- an ordering that would only + matter for a ``BaseException`` was tried, proved inert by mutation + testing, and dropped rather than shipped unfalsifiable. + """ + try: + self._logger.error( + message, + extra=LogExtraInfo( + openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO + ), + ) + except Exception: # noqa: BLE001 + # Nowhere left to report this: the logger is what failed. + pass + def notify(self) -> None: """The 'Notify' part of Open Job Description's subprocess cancelation method. On Linux/macOS: @@ -351,7 +406,18 @@ def terminate(self) -> None: """ # Review22-F4 fix: Bind _process once. See notify() for rationale. proc = self._process - if proc is not None and proc.poll() is None: + # Deliberately NOT gated on `proc.poll() is None`. The signal target is a + # process *group*, and killpg still reaches surviving members after the + # group leader has exited and been reaped -- which is precisely the shape + # a wrap environment produces (a wrapper that execs and returns while its + # workload lives on). Gating on leader liveness made this a no-op exactly + # when the recorded _sudo_child_process_group_id was the only handle left + # on the survivors, orphaning them past terminal publication. + # + # openjd-rs never gates: every cancel/timeout/reap-failure path calls + # send_terminate(pid) -> killpg unconditionally (subprocess.rs), and its + # is_process_alive() probe is #[allow(dead_code)] for this reason. + if proc is not None: self._terminate_process(proc) def _terminate_process(self, process: Popen) -> None: @@ -568,13 +634,20 @@ def _enqueue_stdout(): line, extra=LogExtraInfo(openjd_log_content=LogContent.COMMAND_OUTPUT) ) - def _log_returncode(self): - """Logs the return code of the exited subprocess""" + def _log_returncode(self, proc: Popen) -> None: + """Logs the return code of the exited subprocess. + + Takes the ``Popen`` rather than reading ``self._process``, for the same + reason ``notify()``/``terminate()``/``exit_code`` bind it once: this runs + on the pool worker while ``run()``'s ``finally`` is the thing that clears + the attribute, and an ``AttributeError`` on ``None`` here would unwind + ``run()`` before the child was waited on. + """ if self._returncode is not None: # Print out the signed representation of returncodes that would be negative as a 32-bit signed integer if self._returncode < 0x7FFFFFFF: self._logger.info( - f"Process pid {self._process.pid} exited with code: {self._returncode} (unsigned) / {hex(self._returncode)} (hex)", + f"Process pid {proc.pid} exited with code: {self._returncode} (unsigned) / {hex(self._returncode)} (hex)", extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), ) else: @@ -584,7 +657,7 @@ def _tosigned(n: int) -> int: return int.from_bytes(b, "big", signed=True) self._logger.info( - f"Process pid {self._process.pid} exited with code: {self._returncode} (unsigned) / {hex(self._returncode)} (hex) / {_tosigned(self._returncode)} (signed)", + f"Process pid {proc.pid} exited with code: {self._returncode} (unsigned) / {hex(self._returncode)} (hex) / {_tosigned(self._returncode)} (signed)", extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), ) @@ -672,11 +745,21 @@ def _posix_signal_subprocess( with ctx_mgr as has_cap_kill: if has_cap_kill or not self._user or self._user.is_process_user(): try: + # Signal first, announce second. The announcement used to come + # first, which put the kill behind a log call: a raising + # `logging.Filter` skipped `killpg` entirely and the exception + # left this method without a signal ever being sent. That is + # reachable from LoggingSubprocess.run()'s reap, whose whole + # purpose is to not abandon a live child. + # + # Only OSError is caught here, so a raising filter propagates + # either way; the difference is whether the child was killed + # before it did. + os.killpg(self._sudo_child_process_group_id, numeric_signal) self._logger.info( f'INTERRUPT: Sending signal "{signal_name}" to process group {self._sudo_child_process_group_id}', extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), ) - os.killpg(self._sudo_child_process_group_id, numeric_signal) except OSError: self._logger.info( "Could not directly send signal {signal_name} to {self._posix_signal_target.pid}, trying sudo.", diff --git a/test/openjd/sessions_v0/test_action_filter_hardening.py b/test/openjd/sessions_v0/test_action_filter_hardening.py index 6446dda..7a1b002 100644 --- a/test/openjd/sessions_v0/test_action_filter_hardening.py +++ b/test/openjd/sessions_v0/test_action_filter_hardening.py @@ -363,3 +363,62 @@ def callback(kind: ActionMessageKind, value: Any, fail: bool) -> None: # THEN assert "a perfectly ordinary boom" in record.getMessage() + + +class TestNearMissEnvMacroIsAnchored: + """The near-miss detector must not fire on a *longer* token. + + Its regex had no trailing boundary, so any line merely starting with + `openjd_env`, `openjd_redacted_env` or `openjd_unset_env` was reported as a + malformed macro -- and that path fails the running action. A script logging + `openjd_environment: ready` therefore killed its own task, with no adversary + involved. + + openjd-rs anchors the same three tokens on ':', a space, or end-of-string + (`action_filter.rs` `is_malformed_env_command`), and its comment names this + exact false positive. + """ + + @pytest.mark.parametrize( + "line", + [ + "openjd_environment: ready", + "openjd_env_setup: done", + "openjd_envvar=1", + "openjd_unset_environment_now", + "openjd_redacted_envelope: opened", + ], + ) + def test_a_longer_token_is_ordinary_output(self, line: str) -> None: + # GIVEN a line whose prefix only *looks* like a macro + callback = MagicMock() + f = ActionMonitoringFilter(session_id="foo", callback=callback) + + # WHEN + f.filter(_make_record(line)) + + # THEN: nothing is reported -- it is just a log line + assert callback.call_count == 0 + + @pytest.mark.parametrize( + "line", + [ + "openjd_env FOO=bar", + "openjd_env:FOO=bar", + "OpenJD_Env: FOO=bar", + "openjd_unset_env", + " openjd_env: FOO", + ], + ) + def test_a_genuine_near_miss_still_fails_the_action(self, line: str) -> None: + """Negative control: anchoring must not disarm the detector.""" + # GIVEN a genuinely malformed macro + callback = MagicMock() + f = ActionMonitoringFilter(session_id="foo", callback=callback) + + # WHEN + f.filter(_make_record(line)) + + # THEN: it is still reported as a failure + fails = [c for c in callback.call_args_list if c[0][0] == ActionMessageKind.FAIL] + assert len(fails) == 1, callback.call_args_list diff --git a/test/openjd/sessions_v0/test_runner_base.py b/test/openjd/sessions_v0/test_runner_base.py index 4a993b4..3f8e612 100644 --- a/test/openjd/sessions_v0/test_runner_base.py +++ b/test/openjd/sessions_v0/test_runner_base.py @@ -1233,3 +1233,42 @@ def test_materialize_files_fails( assert not os.path.exists(dest_dir / "test_materialize_files.txt") messages = collect_queue_messages(message_queue) assert any(m.startswith("openjd_fail") for m in messages) + + +class TestCancelInfoWriteFailureIsContained: + """A failed `cancel_info.json` write must not unwind the cancel. + + The guard named `OSError`, but `write_file_for_user` reaches + `WindowsPermissionHelper`, which raises `RuntimeError` when it cannot set an + ACL for an impersonated user -- and `RuntimeError` is not an `OSError`. The + exception therefore escaped and unwound the runtime-limit `Timer` thread: no + notify, no grace timer, a live child, and the timeout silently unenforced. + + openjd-rs cannot hit this: `write_cancel_info` discards the write result and + schedules notify plus the delayed terminate regardless. + """ + + @pytest.mark.parametrize( + "error", + [ + OSError(28, "No space left on device"), + RuntimeError("Could not set the ACL for the session user"), + ], + ids=["oserror", "runtimeerror"], + ) + def test_the_action_still_gets_a_terminate(self, error: Exception, tmp_path: Path) -> None: + # GIVEN a notify-then-terminate cancel whose cancel_info write fails + logger = build_logger(QueueHandler(SimpleQueue())) + with NotifyingRunner(logger=logger, session_working_directory=tmp_path) as runner: + live_child = MagicMock() + live_child.is_running = True + runner._process = live_child + + # WHEN + with patch("openjd.sessions._runner_base.write_file_for_user", side_effect=error): + runner.cancel() + + # THEN: no exception escaped, and we fell back to terminating rather + # than leaving the child running with no scheduled kill. + assert live_child.terminate.called + assert not live_child.notify.called diff --git a/test/openjd/sessions_v0/test_session_let_bindings.py b/test/openjd/sessions_v0/test_session_let_bindings.py index 47ba8ad..2b3c5dd 100644 --- a/test/openjd/sessions_v0/test_session_let_bindings.py +++ b/test/openjd/sessions_v0/test_session_let_bindings.py @@ -11,7 +11,10 @@ enter and exit sides); - binding-RHS parsing is memoized across applications; - the unified optional int-or-format-string field resolver - (``resolve_optional_int_field``) enforces consistent bounds. + (``resolve_optional_int_field``) enforces consistent bounds; +- and, as the mirror of the EXPR cases above, a LEGACY (non-EXPR) + whole-field interpolation of a non-string symbol value resolves to its + string form instead of raising ``AttributeError``. """ from __future__ import annotations @@ -19,11 +22,12 @@ import sys import time import uuid +from unittest.mock import patch as mock_patch from typing import Any import pytest -from openjd.model import SymbolTable +from openjd.model import ParameterValue, ParameterValueType, SymbolTable from openjd.model.v2023_09 import ( Action as Action_2023_09, ArgString as ArgString_2023_09, @@ -77,6 +81,37 @@ def _format_string_field(raw: str) -> Any: return action.timeout +def _legacy_args_action(*args: str) -> Action_2023_09: + """An action with FormatString-typed ``args`` and EXPR *not* enabled. + + Parsed through ``model_validate`` with an explicit context rather than + built like :func:`_action`, because only the FEATURE_BUNDLE_1 parse gives + the args the FormatString behaviour whose whole-field resolution is under + test here. + """ + context = ModelParsingContext_2023_09(supported_extensions=["FEATURE_BUNDLE_1"]) + return Action_2023_09.model_validate({"command": "echo", "args": list(args)}, context=context) + + +class _StrDiffersFromRepr(int): + """An int-valued symbol whose ``str`` and ``repr`` differ. + + Needed to pin *which* coercion the string-form branch applies: for a plain + int, bool or float ``str(v) == repr(v)``, so none of those can tell ``str`` + from ``repr``. + + Overrides ``__str__`` rather than ``__repr__`` on purpose. ``int`` has no + separate ``__str__`` slot, so overriding ``__repr__`` would change ``str()`` + too and discriminate nothing. An ``IntEnum`` would not work either: + ``str(SomeIntEnum.MEMBER)`` is ``'E.MEMBER'`` on Python 3.9-3.10 and + ``'3'`` from 3.11 on, which would fail on 6 of the 18 CI jobs. This class + behaves identically on 3.9 through 3.14 (verified). + """ + + def __str__(self) -> str: + return "str-form" + + # --------------------------------------------------------------------------- # A failing extra `let` binding must FAIL the action via the callback path, # never raise out of enter_environment()/exit_environment(). @@ -248,6 +283,160 @@ def test_let_bound_list_flattens_into_args(self) -> None: # whitespace preserved — not rendered as a single stringified list. assert resolved == ["front", "alpha beta", "gamma", "back"] + def test_let_bound_scalar_becomes_exactly_one_argument(self) -> None: + """The counterpart to list flattening: a non-null, non-list typed value + is ONE argument and must not be flattened. + + Pins the scalar arm of the typed-argument dispatch, which list and null + coverage alone leaves open. A scalar ``ExprValue`` is not iterable, so + misclassifying one as a list raises TypeError out of the runner -- and + with only list/null cases covered, nothing noticed. + """ + # GIVEN: a step script binding an int and a string with embedded + # whitespace, both consumed as whole-field argument expressions. + context = ModelParsingContext_2023_09(supported_extensions=["EXPR"]) + script = StepScript_2023_09.model_validate( + { + "let": ["count = 7", "label = 'solo value'"], + "actions": { + "onRun": { + "command": "echo", + "args": ["front", "{{ count }}", "{{ label }}", "back"], + } + }, + }, + context=context, + ) + + # WHEN + symtab = SymbolTable() + apply_let_bindings(symtab=symtab, let_bindings=script.let or []) + resolved = resolve_action_arg_values(script.actions.onRun.args, symtab) + + # THEN: one argument each -- the int rendered, and the string kept whole + # rather than split on its space or flattened character-wise. + assert resolved == ["front", "7", "solo value", "back"] + + +# --------------------------------------------------------------------------- +# The mirror of the case above (RFC 0005 §1.3.2, the LEGACY path): a legacy +# (non-EXPR) whole-field interpolation returns the symbol's OWN value, not an +# ExprValue. Typed null/list handling must therefore be reached only for a real +# ExprValue -- reading `.is_null` off an int or a bool raised AttributeError +# straight out of the public Session API. `ParameterValue` documents +# INT/FLOAT/PATH values as strings but does not enforce it, and it carries BOOL +# values natively *by design* (RFC 0007), so both types genuinely arrive here. +# +# Which values reach the branch is decided by openjd-model, not by the symbol +# table (`SymbolTable` validates nothing -- `symtab["X"] = object()` succeeds): +# `FullNameNode.allows_nonscalar` is False, so a legacy whole-field result that +# is not a `numbers.Real` or a `str` raises FormatStringError and takes the +# pre-existing plain-resolution fallback instead. That is why a `list` never +# reaches the branch, and why None does not either. +# --------------------------------------------------------------------------- + + +class TestLegacyNonStringSymbolValuesInArgs: + @pytest.mark.parametrize( + "value, expected", + [ + pytest.param(30, "30", id="int-INT-parameter"), + pytest.param(True, "True", id="bool-BOOL-parameter"), + pytest.param(1.5, "1.5", id="float-FLOAT-parameter"), + # str() and repr() differ only here, so this is the one case that + # pins which coercion the branch applies. + pytest.param(_StrDiffersFromRepr(3), "str-form", id="str-not-repr"), + pytest.param("30", "30", id="str-control"), + # None does NOT reach the branch: the whole-field arg raises + # FormatStringError and takes the pre-existing fallback (the + # multi-segment arg still enters the branch, as an ordinary str). + # Kept as a control that the fallback agrees, not as cover for the + # branch. + pytest.param(None, "", id="none-via-fallback"), + ], + ) + def test_legacy_whole_field_value_resolves_to_its_string_form( + self, value: Any, expected: str + ) -> None: + """A non-string symbol value resolves to its string form, and agrees + with what a multi-segment arg over the same symbol produces. + + The agreement is the invariant that makes string-form resolution the + right answer: ``pre-{{ Param.V }}`` has always gone through plain + string resolution, so a whole-field ``{{ Param.V }}`` must render the + same way or the two disagree within one legacy template. + + Both are asserted against a literal rather than against each other -- + ``multi_segment == f"pre-{whole_field}"`` would hold if both were wrong + in the same way, or if both were empty. + + Holds for the scalar types that reach the branch. A ``str`` *subclass* + is the known exception: the branch passes it through untouched while + the multi-segment path applies ``str()`` to it. That divergence + predates this fix and is not addressed here. + """ + # GIVEN: a non-EXPR action with a whole-field arg and a multi-segment + # arg over the same symbol, whose value is not a string. + action = _legacy_args_action("{{ Param.V }}", "pre-{{ Param.V }}") + symtab = SymbolTable() + symtab["Param.V"] = value + + # WHEN + whole_field, multi_segment = resolve_action_arg_values(action.args, symtab) + + # THEN: the string form. Before the fix an int/bool/float raised + # AttributeError ('is_null') here. + assert whole_field == expected + assert multi_segment == f"pre-{expected}" + + def test_int_job_parameter_value_reaches_the_subprocess_as_a_string( + self, caplog: pytest.LogCaptureFixture, python_exe: str + ) -> None: + """End-to-end through the public API, asserting the argv the process + actually received -- not merely that nothing crashed. + + ``ParameterValue`` documents INT values as strings but types the field + ``Any`` and does not validate it, so a caller passing the int itself + reached ``resolve_action_arg_values``. The failure was an + ``AttributeError`` escaping ``run_task`` -- not a failed action, an + unhandled crash in the caller's thread. + """ + # GIVEN: an INT job parameter carrying an int, consumed as a + # whole-field argument by a non-EXPR step script. + context = ModelParsingContext_2023_09(supported_extensions=["FEATURE_BUNDLE_1"]) + step = StepScript_2023_09.model_validate( + { + "actions": { + "onRun": { + "command": python_exe, + "args": [ + "-c", + "import sys; print('ARGV=' + repr(sys.argv[1:]))", + "{{ Param.N }}", + ], + } + } + }, + context=context, + ) + job_parameters = {"N": ParameterValue(type=ParameterValueType.INT, value=30)} + + with Session(session_id=uuid.uuid4().hex, job_parameter_values=job_parameters) as session: + # WHEN: this raised AttributeError before the fix. + session.run_task(step_script=step, task_parameter_values={}) + _run_until_ready(session) + + # THEN: the action succeeded AND passed the single argument "30". An + # empty or dropped argument would also avoid the crash, so the content + # is what pins the behaviour. + status = session.action_status + assert status is not None + assert status.state == ActionState.SUCCESS + expected = "ARGV=" + repr(["30"]) + assert any(expected in m for m in caplog.messages), [ + m for m in caplog.messages if "ARGV=" in m + ] + # --------------------------------------------------------------------------- # resolve_optional_int_field: one implementation for the three @@ -521,3 +710,55 @@ def test_task_file_supports_path_property_access( assert status is not None assert status.state == ActionState.SUCCESS assert any("PARENT=" in m and "embedded_files" in m for m in caplog.messages) + + +# --------------------------------------------------------------------------- +# A failure writing the path-mapping rules file must FAIL the action via the +# callback path, never raise out of the public Session API. +# --------------------------------------------------------------------------- +class TestPathMappingRulesFileWriteFailure: + """`_materialize_path_mapping` had no failure path. + + All four call sites run before their runner exists, so an OSError from + `mkstemp` or `write_file_for_user` escaped `enter_environment` / `run_task` / + `exit_environment` with no callback and no ActionStatus -- the caller was + left with no terminal state to observe. + + openjd-rs supplies the contract: `std::fs::write` maps to + `SessionError::WorkingDirectory` and every call site maps that to + `fail_action_setup()`, which publishes Failed and notifies. + """ + + def test_a_write_failure_fails_the_action_instead_of_raising(self) -> None: + # GIVEN: the rules-file write cannot succeed + callback_events: list[ActionStatus] = [] + + def callback(session_id: str, status: ActionStatus) -> None: + callback_events.append(status) + + context = ModelParsingContext_2023_09(supported_extensions=["EXPR"]) + environment = Environment_2023_09.model_validate( + { + "name": "AnyEnv", + "script": {"actions": {"onEnter": {"command": "echo", "args": ["entered"]}}}, + }, + context=context, + ) + with Session( + session_id=uuid.uuid4().hex, job_parameter_values={}, callback=callback + ) as session: + with mock_patch( + "openjd.sessions._session.write_file_for_user", + side_effect=OSError(13, "Permission denied"), + ): + # WHEN: entering must not raise + identifier = session.enter_environment(environment=environment) + + # THEN: a terminal FAILED status naming the working directory, and a + # session the caller can still unwind. + assert identifier is not None + status = session.action_status + assert status is not None + assert status.state == ActionState.FAILED + assert status.fail_message is not None + assert "path mapping rules" in status.fail_message diff --git a/test/openjd/sessions_v0/test_subprocess_reaping.py b/test/openjd/sessions_v0/test_subprocess_reaping.py index 703fb38..a5afe62 100644 --- a/test/openjd/sessions_v0/test_subprocess_reaping.py +++ b/test/openjd/sessions_v0/test_subprocess_reaping.py @@ -26,10 +26,13 @@ import logging import os import signal +import threading import time +from contextlib import contextmanager +from pathlib import Path from logging.handlers import QueueHandler from queue import SimpleQueue -from typing import Any, Optional +from typing import Any, Generator, Optional from unittest.mock import MagicMock, patch import pytest @@ -37,7 +40,7 @@ from openjd.sessions._os_checker import is_posix from openjd.sessions._subprocess import LoggingSubprocess -from .conftest import build_logger +from .conftest import build_logger, create_unique_logger_name, serial_process # A child that emits one line (to drive the pump), then outlives the pump. _LONG_CHILD = "import time\nprint('MARKER', flush=True)\ntime.sleep(60)\n" @@ -99,6 +102,59 @@ def exploding_logger() -> Any: logger.removeFilter(exploder) +class _FilterRaisingOnAny(logging.Filter): + """Raises on any record whose message contains one of ``markers``. + + Separate from :class:`_ExplodingFilter` so the tests using that keep their + exact single-marker behaviour. + + The raised message names the marker that triggered it, so a test can tell + *which* log call failed -- which is what distinguishes "the startup line + raised and was contained" from "something inside the reap raised and replaced + it". Both are ``RuntimeError`` from the same filter, so nothing else does. + """ + + def __init__(self, *markers: str) -> None: + super().__init__() + self._markers = markers + self.fired = False + self.fired_markers: list[str] = [] + + def filter(self, record: logging.LogRecord) -> bool: + message = str(record.msg) + for marker in self._markers: + if marker in message: + self.fired = True + self.fired_markers.append(marker) + raise RuntimeError(f"filter blew up on: {marker}") + return True + + +@contextmanager +def logger_carrying(filt: logging.Filter) -> Generator[Any, None, None]: + """A uniquely-named logger carrying `filt`, removed again on exit. + + The removal is not tidiness: ``logging.getLogger`` interns loggers in a + process-global registry, so a *raising* filter left installed would outlive + its test and fire inside an unrelated one. + + Yields ``Any`` rather than ``logging.Logger`` because ``LoggingSubprocess`` + declares its ``logger`` parameter as a ``LoggerAdapter``, while these tests + pass a plain ``Logger`` -- as the ``exploding_logger`` fixture above already + does, for the same reason. + """ + logger = logging.getLogger(create_unique_logger_name(prefix="openjd.test.reaping.startup")) + logger.setLevel(logging.INFO) + handler = logging.NullHandler() + logger.addHandler(handler) + logger.addFilter(filt) + try: + yield logger + finally: + logger.removeFilter(filt) + logger.removeHandler(handler) + + @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") class TestPumpExceptionDoesNotAbandonTheChild: def test_live_child_is_terminated_and_reaped( @@ -314,6 +370,44 @@ def test_a_terminate_failure_does_not_escape(self, python_exe: str) -> None: popen.kill() popen.wait() + def test_a_logging_failure_during_the_reap_does_not_escape(self, python_exe: str) -> None: + """`_reap`'s own logging must not raise either. + + The containment here already covered the terminate and the wait, but not + the three `logger.error` calls between them -- and a raising + `logging.Filter` is exactly what routes execution into `_reap`, which + makes those calls the *likeliest* thing to fail, not the least. Anything + escaping replaces the exception already propagating out of `run()`. + + This drives all three call sites at once: the abandoning notice, the + terminate-failure notice, and the unreaped-after-timeout notice. + """ + # GIVEN a live child, a termination that fails, a 0s wait budget so the + # timeout branch is reached too, and a logger that raises on every error + logger = MagicMock() + logger.error.side_effect = RuntimeError("the logging stack blew up") + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + popen = proc._start_subprocess() + assert popen is not None + try: + with ( + patch.object( + LoggingSubprocess, + "_terminate_process", + side_effect=OSError("cannot signal"), + ), + patch("openjd.sessions._subprocess.ABANDONED_PROCESS_REAP_TIMEOUT_SECONDS", 0), + ): + # WHEN / THEN: nothing escapes, despite every log call failing + proc._reap(popen) + + # All three messages were attempted, so this really did exercise + # every call site rather than returning early. + assert logger.error.call_count == 3 + finally: + popen.kill() + popen.wait() + def test_a_child_that_survives_termination_is_reported_not_hung(self, python_exe: str) -> None: """The wait is bounded: a child that somehow outlives SIGKILL must not hang the pool worker.""" @@ -338,3 +432,467 @@ def test_a_child_that_survives_termination_is_reported_not_hung(self, python_exe finally: popen.kill() popen.wait() + + +class TestTerminateDoesNotDependOnLeaderLiveness: + """`terminate()` must still signal after the group leader has exited. + + The signal target is a process *group*, and killpg still reaches surviving + members once the leader has exited and been reaped. Gating the call on + `proc.poll() is None` made it a no-op in exactly the case where the recorded + group was the only remaining handle on those survivors -- which is the shape + a wrap environment produces, a wrapper that execs and returns while its + workload lives on. The children then outlived terminal publication. + + openjd-rs never gates: every cancel/timeout/reap-failure path calls + `send_terminate(pid)` -> killpg unconditionally, and its `is_process_alive()` + probe is `#[allow(dead_code)]` for this reason. + """ + + def test_signal_is_sent_even_when_poll_reports_the_leader_gone(self) -> None: + # GIVEN a subprocess whose leader has already exited + subproc = LoggingSubprocess(logger=MagicMock(), args=["unused"]) + departed = MagicMock() + departed.poll.return_value = 0 # reaped; a live workload may remain + subproc._process = departed + + # WHEN + with patch.object(subproc, "_terminate_process") as terminate_process: + subproc.terminate() + + # THEN: the group is still signalled + terminate_process.assert_called_once_with(departed) + + def test_nothing_is_signalled_once_the_process_is_released(self) -> None: + """Negative control: `run()`'s finally clears `_process` after reaping, + and terminate() must stay a no-op from then on -- otherwise the pid + could be recycled and an unrelated group signalled.""" + # GIVEN a subprocess that has released its handle + subproc = LoggingSubprocess(logger=MagicMock(), args=["unused"]) + subproc._process = None + + # WHEN + with patch.object(subproc, "_terminate_process") as terminate_process: + subproc.terminate() + + # THEN + terminate_process.assert_not_called() + + +@serial_process +class TestStartupLoggingDoesNotAbandonTheChild: + """The ownership `try/finally` must begin at process creation, not after the + startup logging. + + `run()` logged "Command started as pid" and "Output:" *before* entering the + `try` whose `finally` reaps, and did the POSIX process-group lookup there + too. Anything raising in that window escaped `run()` having never waited on + the child, leaving a live process with no exit code recorded while the runner + published a terminal state for it. + + A `logging.Filter` is the reachable trigger because filters propagate out of + `Logger.handle`, unlike handlers, whose exceptions go to `handleError` -- + and `Session` installs a filter of its own. The process-group lookup is + covered by its own test below, which is why the boundary moved to + immediately after creation rather than merely above the two log calls. + """ + + @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") + def test_a_filter_raising_on_the_pid_line_does_not_leak_the_child( + self, python_exe: str + ) -> None: + # GIVEN: a long-lived child, and a filter that raises on the first + # startup log line -- while the child is still alive + exploder = _FilterRaisingOnAny("Command started as pid") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: the filter really did fire on that line... + assert exploder.fired is True + pid = proc.pid + assert pid is not None + # ...the child is gone, not merely signalled... + assert _wait_gone(pid), f"child {pid} was left running" + # ...and its status was recorded rather than lost. + assert proc.exit_code == -signal.SIGKILL # type: ignore + assert proc._process is None + + @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") + def test_a_filter_raising_on_the_output_banner_does_not_leak_the_child( + self, python_exe: str + ) -> None: + """A second, separate unprotected line -- not the same one as above.""" + # GIVEN + exploder = _FilterRaisingOnAny("Output:") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN + assert exploder.fired is True + pid = proc.pid + # Not decoration: `_wait_gone(None)` returns True, so without this the + # liveness assertion below degrades to a tautology. + assert pid is not None + assert _wait_gone(pid), f"child {pid} was left running" + assert proc.exit_code == -signal.SIGKILL # type: ignore + assert proc._process is None + + @pytest.mark.skipif(not is_posix(), reason="the process-group lookup is POSIX-only") + def test_a_failing_process_group_lookup_does_not_leak_the_child( + self, message_queue: SimpleQueue, queue_handler: QueueHandler, python_exe: str + ) -> None: + """The POSIX process-group lookup was in the unguarded window too. + + `ProcessLookupError` from it is caught deliberately (a trivial command + can exit first), but nothing else was, and it sits between the two log + lines -- so covering only the logging would leave the middle of the same + window unprotected. No filter here: the failure is injected into the + lookup itself, so this is independent of the logging stack. + + What is asserted is only that the reap is *reached*, and deliberately + NOT that the child dies. When the group lookup is itself what failed, + `_sudo_child_process_group_id` stays unset, and + `_posix_signal_subprocess` then declines to signal at all rather than + target a possibly-recycled pid (the R5-9 "no process group known" + convention). So on this one path the boundary buys the attempt and the + handle release, not the kill. + + That is a real, pre-existing limitation of the signalling path -- not + something this change introduces or claims to fix -- and it is recorded + as a follow-up. Asserting a dead child here would require standing in for + the signal, which would make this test assert a kill the shipped code + does not perform. + """ + # GIVEN a live child and a process-group lookup that fails in a way the + # production code does not catch + logger = build_logger(queue_handler) + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with ( + patch( + "openjd.sessions._subprocess.os.getpgid", + side_effect=RuntimeError("getpgid blew up"), + ), + # Stubbed rather than stood-in-for: the real call would spend a second + # scanning for a sudo child it will not find, and this test is about + # whether the reap is reached at all. + patch.object( + LoggingSubprocess, "_terminate_process", autospec=True + ) as terminate_process, + # No wait budget: the child cannot have been killed by the stub, so + # the bounded wait would otherwise be spent in full. + patch("openjd.sessions._subprocess.ABANDONED_PROCESS_REAP_TIMEOUT_SECONDS", 0), + ): + with pytest.raises(RuntimeError, match="getpgid blew up"): + proc.run() + + pid = proc.pid + try: + # THEN: ownership was discharged -- the reap ran, instead of the + # failure escaping in front of it + terminate_process.assert_called_once() + assert proc._process is None + finally: + # The stubbed terminate killed nothing, so this test owns the child. + if pid is not None and _pid_alive(pid): + os.kill(pid, signal.SIGKILL) # type: ignore[attr-defined] + try: + os.waitpid(pid, 0) + except ChildProcessError: + # Already reaped between the liveness check and here -- the + # desired end state, so there is nothing left to wait for. + pass + + @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") + def test_the_reap_kills_the_child_even_when_its_own_logging_raises( + self, python_exe: str + ) -> None: + """`_reap` must terminate before it logs. + + With the boundary moved, a raising filter now routes *into* `_reap` -- so + if that same filter also matches `_reap`'s own "Abandoning the subprocess" + message, a log-then-terminate order meant the `logger.error` raised and the + kill never ran. The leak would have survived the boundary fix. + + Two properties, both load-bearing and each failing for a different + mutation: the child dies (log-then-terminate breaks this), and the caller + still sees the *startup* failure rather than one raised from inside the + `finally` (a reap log that propagates breaks this). + """ + # GIVEN: a filter that raises on the startup line AND on the message + # `_reap` emits while cleaning up. "Running command" is deliberately not + # matched, so the child is really created. + exploder = _FilterRaisingOnAny("Command started as pid", "Abandoning the subprocess") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with pytest.raises(RuntimeError) as excinfo: + proc.run() + + # THEN: the child is reaped despite the logging stack being broken... + pid = proc.pid + assert pid is not None + assert _wait_gone(pid), f"child {pid} was left running" + assert proc.exit_code == -signal.SIGKILL # type: ignore + assert proc._process is None + # ...and the exception is the startup failure. Naming the marker matters: + # both failures are RuntimeError from the same filter, so only the marker + # distinguishes "contained" from "replaced the propagating exception". + assert "filter blew up on: Command started as pid" in str(excinfo.value) + assert "Abandoning the subprocess" not in str(excinfo.value) + + def test_the_startup_lines_are_still_logged_on_a_healthy_run( + self, message_queue: SimpleQueue, queue_handler: QueueHandler, python_exe: str + ) -> None: + """Negative control. Deleting the startup logging would satisfy every + assertion above, so pin that it still happens -- and that a healthy run + keeps its own exit code.""" + # GIVEN / WHEN + logger = build_logger(queue_handler) + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _SHORT_CHILD]) + proc.run() + + # THEN + lines = [] + while not message_queue.empty(): + lines.append(message_queue.get().getMessage()) + assert any(f"Command started as pid: {proc.pid}" in line for line in lines), lines + assert any(line == "Output:" for line in lines), lines + assert proc.exit_code == 7 + assert proc.failed_to_start is False + + +class TestStartFailureAlwaysReleasesWaiters: + """`_has_started` must be set however `_start_subprocess` returns. + + `_start_subprocess` catches every `Exception` from the launch and logs + "Process failed to start" -- but that log can itself raise, and then the + exception propagates out of `run()` before `_has_started.set()` was reached. + `ScriptRunnerBase._run` waits on `wait_until_started()` with **no timeout**, + so the thread that called the public `Session` API blocked forever on a + launch that had already definitively failed. + + Platform-agnostic: no child is ever created (the failing log precedes + `Popen`), so there is nothing to signal or reap and no POSIX guard is needed. + """ + + def test_a_raising_log_in_the_failure_handler_still_releases_waiters( + self, python_exe: str + ) -> None: + # GIVEN a filter that raises on the pre-launch "Running command" line. + # The failure handler then logs "Process failed to start: ", + # whose text contains the marker too, so the handler's own log raises as + # well -- which is the case that used to escape. + exploder = _FilterRaisingOnAny("Running command") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", "pass"]) + + # WHEN + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: waiters are released rather than blocked forever + assert proc._has_started.is_set() is True + # No child was created, so nothing leaked. + assert proc.pid is None + + def test_wait_until_started_returns_after_a_raising_failure_handler( + self, python_exe: str + ) -> None: + """The observable form of the above, through the public API that hung. + + Uses a bounded wait on a daemon thread deliberately: if this regresses, + the test fails on the flag rather than hanging the suite. + """ + # GIVEN + exploder = _FilterRaisingOnAny("Running command") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", "pass"]) + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + returned = threading.Event() + + def waiter() -> None: + proc.wait_until_started() # as ScriptRunnerBase._run calls it: no timeout + returned.set() + + # WHEN + threading.Thread(target=waiter, daemon=True).start() + + # THEN + assert returned.wait(timeout=10.0) is True, "wait_until_started() did not return" + + def test_a_normal_start_failure_is_still_reported( + self, message_queue: SimpleQueue, queue_handler: QueueHandler + ) -> None: + """Negative control: the ordinary failure-to-start path is unchanged -- + `_start_subprocess` returns None, `run()` does not raise, and the object + reports `failed_to_start`.""" + # GIVEN a command that cannot be launched, and a working logger + logger = build_logger(queue_handler) + proc = LoggingSubprocess( + logger=logger, args=[str(Path("this-command-does-not-exist-openjd-test"))] + ) + + # WHEN + proc.run() + + # THEN + assert proc.failed_to_start is True + assert proc._has_started.is_set() is True + assert proc.is_running is False + + +@serial_process +@pytest.mark.skipif(not is_posix(), reason="POSIX process-group signalling") +class TestTheKillIsNotGatedBehindItsOwnLog: + """`_posix_signal_subprocess` must send the signal before announcing it. + + The announcement came first, inside a `try` that catches only `OSError`, so a + raising `logging.Filter` skipped `os.killpg` and left the method without ever + signalling. That is reachable from `run()`'s reap -- which exists precisely so + a live child is not abandoned -- so the reap could be entered, attempt the + kill, and still leave the child running. + """ + + def test_the_child_dies_even_when_the_interrupt_log_raises(self, python_exe: str) -> None: + # GIVEN a live child, and a filter that raises both on the startup line + # (to reach the reap with the child alive) and on the signal announcement + exploder = _FilterRaisingOnAny("Command started as pid", 'INTERRUPT: Sending signal "kill"') + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: the announcement really did raise -- without this the test could + # pass on a tree where the signal path was never reached at all... + assert 'INTERRUPT: Sending signal "kill"' in exploder.fired_markers + pid = proc.pid + assert pid is not None + # ...and the child was killed regardless + assert _wait_gone(pid), f"child {pid} was left running" + + def test_the_signal_is_still_announced_on_a_healthy_cancel( + self, message_queue: SimpleQueue, queue_handler: QueueHandler, python_exe: str + ) -> None: + """Negative control: deleting the announcement would satisfy the test + above, so pin that a working logger still gets it -- and that the child + still dies.""" + # GIVEN a running child + logger = build_logger(queue_handler) + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + runner = threading.Thread(target=proc.run, daemon=True) + runner.start() + proc.wait_until_started() + deadline = time.time() + 10.0 + while proc._sudo_child_process_group_id is None and time.time() < deadline: + time.sleep(0.02) + + # WHEN + proc.terminate() + runner.join(timeout=15.0) + + # THEN + lines = [] + while not message_queue.empty(): + lines.append(message_queue.get().getMessage()) + assert any( + 'INTERRUPT: Sending signal "kill" to process group' in line for line in lines + ), lines + assert proc.exit_code == -signal.SIGKILL # type: ignore + + +@serial_process +@pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") +class TestObservableStateOnTheAbandonedPath: + """Two things the ownership fix changed, or deliberately left alone, that no + test asserted either way. + + Neither is a defect. They are pinned because they are *observable* -- one is a + public property, the other a consumer callback -- and an unasserted observable + is one a later change can flip without anyone noticing. + """ + + def test_is_running_reports_false_once_the_child_is_abandoned(self, python_exe: str) -> None: + """`is_running` is `_has_started and _process is not None`. + + Before the ownership fix, an abort in the startup window left `_process` + set forever, so this property reported **True** for a subprocess nobody + would ever wait on -- indefinitely, since only `run()` clears it. Now the + `finally` clears it, so the answer is False and it is accurate. + """ + # GIVEN a child abandoned by a raising filter in the startup window + exploder = _FilterRaisingOnAny("Command started as pid") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: it does not claim to be running + assert proc.is_running is False + # ...and the two inputs to that answer are what we think they are, so a + # future change to either is visible here. + assert proc.has_started is True + assert proc._process is None + # Not a failure to *start*: the child did start, and then was abandoned. + assert proc.failed_to_start is False + assert _wait_gone(proc.pid), f"child {proc.pid} was left running" + + def test_the_subprocess_callback_is_not_invoked_on_the_abandoned_path( + self, python_exe: str + ) -> None: + """Deliberate, and pinned so it stays deliberate. + + `run()` invokes its callback on the two paths that end normally: a failure + to start, and a completed `wait()`. The abandoned path does neither, even + though `_reap` now records a real exit code. That is correct rather than an + omission: `ScriptRunnerBase` registers `_on_process_exit` as the future's + done-callback, so the Session is notified through the future's exception + regardless, and invoking this callback too would notify twice for one + action. + """ + # GIVEN a child abandoned by a raising filter, and a callback + callback = MagicMock() + exploder = _FilterRaisingOnAny("Command started as pid") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess( + logger=logger, args=[python_exe, "-c", _LONG_CHILD], callback=callback + ) + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: not invoked from this path... + callback.assert_not_called() + # ...even though a terminal exit code was recorded. + assert proc.exit_code == -signal.SIGKILL # type: ignore + assert _wait_gone(proc.pid), f"child {proc.pid} was left running" + + def test_the_subprocess_callback_is_still_invoked_on_a_healthy_run( + self, message_queue: SimpleQueue, queue_handler: QueueHandler, python_exe: str + ) -> None: + """Negative control: the assertion above must not be satisfiable by a + callback that is simply never invoked at all.""" + # GIVEN / WHEN + callback = MagicMock() + logger = build_logger(queue_handler) + proc = LoggingSubprocess( + logger=logger, args=[python_exe, "-c", _SHORT_CHILD], callback=callback + ) + proc.run() + + # THEN + callback.assert_called_once_with() + assert proc.exit_code == 7 diff --git a/test/openjd/sessions_v0/test_wrap_actions.py b/test/openjd/sessions_v0/test_wrap_actions.py index f3fd5b5..121948e 100644 --- a/test/openjd/sessions_v0/test_wrap_actions.py +++ b/test/openjd/sessions_v0/test_wrap_actions.py @@ -600,43 +600,31 @@ def test_variables_map_included_in_wrapped_environment(self) -> None: assert any(entry == "DECLARED_VAR=from-variables-map" for entry in env_list), env_list def test_later_set_overrides_earlier_value_without_duplicate(self) -> None: - # Cumulative flattening: when a later environment's openjd_env sets - # a name an earlier environment already set, the list carries only - # the effective (later) value — matching the real subprocess env - # and the Rust runtime's single cumulative map. - from openjd.sessions._session import ( - EnvironmentVariableSetChange, - EnvironmentVariableUnsetChange, - SimplifiedEnvironmentVariableChanges, - ) - + """A later set of the same name yields one entry, the effective value. + + Rewritten to drive real environments. The previous version appended + fabricated identifiers to ``_environments_entered`` and assigned + ``_created_env_vars`` directly, which bypassed the production write path + entirely -- so it kept passing while ``WrappedAction.Environment`` + violated RFC 0008's session-lifetime MUST. A test that never exercises + the writers cannot notice the writers being wrong. + """ + # GIVEN: two environments, the second overriding the first's export with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: - scenarios: list[ - tuple[str, list[EnvironmentVariableSetChange | EnvironmentVariableUnsetChange]] - ] = [ - ("env-a", [EnvironmentVariableSetChange(name="FOO", value="1")]), - ("env-b", [EnvironmentVariableSetChange(name="FOO", value="2")]), - ] - for env_id, changes in scenarios: - session._environments_entered.append(env_id) - tracked = SimplifiedEnvironmentVariableChanges({}) - tracked.simplify_ordered_changes(changes) - session._created_env_vars[env_id] = tracked - - assert session._collect_session_env_list() == ["FOO=2"] - - # And a later unset removes the name entirely. - session._environments_entered.append("env-c") - tracked = SimplifiedEnvironmentVariableChanges({}) - tracked.simplify_ordered_changes([EnvironmentVariableUnsetChange(name="FOO")]) - session._created_env_vars["env-c"] = tracked - - assert session._collect_session_env_list() == [] + for value in ("1", "2"): + session.enter_environment( + environment=_env( + f"Setter{value}", + onEnter=_action("sh", "-c", f"echo 'openjd_env: FOO={value}'"), + onExit=_NOOP, + ) + ) + _run_until_ready(session) - # Clear the fake stack so Session.cleanup() doesn't try to - # unwind environments that were never really entered. - session._environments_entered.clear() - session._created_env_vars.clear() + # THEN: one entry, carrying the later value + assert [e for e in session._collect_session_env_list() if e.startswith("FOO=")] == [ + "FOO=2" + ] def test_exit_path_includes_exiting_envs_openjd_env_vars(self) -> None: # On the onWrapEnvExit path, WrappedAction.Environment must include @@ -689,6 +677,75 @@ def test_openjd_env_vars_included_in_wrapped_environment(self) -> None: env_list = session._collect_session_env_list() assert "EMITTED_VAR=from-openjd-env" in env_list + def test_an_exited_environments_export_is_still_listed(self) -> None: + """RFC 0008 makes session-lifetime inclusion a MUST. + + `rfcs/0008-environment-wrap-actions.md:404-412`: runtimes "MUST include + in `WrappedAction.Environment` every `openjd_env`-defined variable + emitted by any earlier action in the same session -- regardless of + whether that action ran normally or via a wrap hook". + + `_collect_session_env_list()` used to iterate `_environments_entered`, + which `exit_environment` pops, so an earlier environment's export + disappeared the moment that environment exited. openjd-rs keeps a + session-lifetime `env_vars` map for exactly this symbol, held separate + from the per-entered view that feeds the child process environment. + """ + # GIVEN: an environment that exports a variable and is then exited + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + early = _env( + "Early", + onEnter=_action("sh", "-c", "echo 'openjd_env: EARLY_VAR=early-value'"), + onExit=_NOOP, + ) + early_id = session.enter_environment(environment=early) + _run_until_ready(session) + assert "EARLY_VAR=early-value" in session._collect_session_env_list() + + # WHEN + session.exit_environment(identifier=early_id, keep_session_running=True) + _run_until_ready(session) + + # THEN: the export outlives the environment that emitted it + assert "EARLY_VAR=early-value" in session._collect_session_env_list() + + def test_an_explicit_unset_still_removes_the_name(self) -> None: + """Negative control for the test above. + + Session-lifetime retention must not become "nothing is ever removed". + An explicit `openjd_unset_env` is the one remover -- the same contract + openjd-rs has, where `env_vars` is only ever erased by an unset. + """ + # GIVEN: a variable exported and then explicitly unset by a later env + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + setter = _env( + "Setter", + onEnter=_action("sh", "-c", "echo 'openjd_env: DOOMED=value'"), + onExit=_NOOP, + ) + setter_id = session.enter_environment(environment=setter) + _run_until_ready(session) + assert "DOOMED=value" in session._collect_session_env_list() + + unsetter = _env( + "Unsetter", + onEnter=_action("sh", "-c", "echo 'openjd_unset_env: DOOMED'"), + onExit=_NOOP, + ) + unsetter_id = session.enter_environment(environment=unsetter) + _run_until_ready(session) + + # WHEN: both environments exit + session.exit_environment(identifier=unsetter_id, keep_session_running=True) + _run_until_ready(session) + session.exit_environment(identifier=setter_id, keep_session_running=True) + _run_until_ready(session) + + # THEN: the unset wins, and it too outlives its environment + assert not any( + entry.startswith("DOOMED=") for entry in session._collect_session_env_list() + ) + class TestRunWrapHookGuard: def test_unknown_hook_name_raises(self) -> None: diff --git a/test/openjd/sessions_v0/test_wrap_task_run.py b/test/openjd/sessions_v0/test_wrap_task_run.py index 53446eb..1e7803f 100644 --- a/test/openjd/sessions_v0/test_wrap_task_run.py +++ b/test/openjd/sessions_v0/test_wrap_task_run.py @@ -150,22 +150,16 @@ def test_injects_wrapped_environment_as_key_value_list(self) -> None: # simplify_ordered_changes) AND the declarative variables:-map seed. session = Session(session_id=uuid.uuid4().hex, job_parameter_values={}) try: - from openjd.sessions._session import ( - EnvironmentVariableSetChange, - SimplifiedEnvironmentVariableChanges, + # Seeded through the session-lifetime map that now backs this symbol. + # Previously this fabricated an entry in `_environments_entered` plus + # a `_created_env_vars` record, which is the per-entered view that + # feeds the child *process* environment -- not this symbol. That + # bypassed the production writers, which is why it kept passing while + # WrappedAction.Environment violated RFC 0008's session-lifetime MUST. + session._session_env_vars.update( + {"DECLARED": "from-variables-map", "FOO": "bar", "BAZ": "qux"} ) - fake_id = "env-1" - session._environments_entered.append(fake_id) - changes = SimplifiedEnvironmentVariableChanges({"DECLARED": "from-variables-map"}) - changes.simplify_ordered_changes( - [ - EnvironmentVariableSetChange(name="FOO", value="bar"), - EnvironmentVariableSetChange(name="BAZ", value="qux"), - ] - ) - session._created_env_vars[fake_id] = changes - symtab = SymbolTable() script = _step_script("echo", ["hi"]) session._inject_wrapped_task_symbols(symtab, script, "Step1", inner_symtab=symtab) diff --git a/test/openjd/test_import_purity.py b/test/openjd/test_import_purity.py new file mode 100644 index 0000000..e057c66 --- /dev/null +++ b/test/openjd/test_import_purity.py @@ -0,0 +1,283 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +"""``openjd.sessions`` must not load the native EXPR extension unless used. + +``openjd.expr`` is a thin facade over the native ``openjd._openjd_rs`` +extension. A module-level import of it anywhere on the pure-Python session +path makes the extension a *load-time* requirement of ``openjd.sessions``, so +a consumer (e.g. the Deadline Cloud worker agent) that only runs non-EXPR +templates can no longer start without a working native extension for its +platform. openjd.model keeps the Rust surface off its import path deliberately +(see ``openjd.model._format_strings._parser``, which duplicates a constant +rather than import it); these tests hold openjd.sessions to the same contract, +at import time *and* while resolving a non-EXPR template. + +Each check runs in a fresh interpreter, because by the time any given test runs +the extension may already have been loaded by another test in the same worker. + +Note for coverage readers: these probes run in child processes with no +``COVERAGE_PROCESS_START``, so the guard in ``_runner_base._is_expr_null`` +that they exercise still reports as uncovered. Do not "fix" that by deleting +the guard -- it is what these tests exist to protect. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +import textwrap +from pathlib import Path + +import pytest + +# Kept below the suite-wide ``--timeout=30`` (see pyproject addopts) so that a +# hung probe is reported as a probe timeout with its output, rather than +# pytest-timeout killing the test first and losing it. +_PROBE_TIMEOUT_SECONDS = 20 + +# Injected ahead of every probe body, so a body never needs interpolation -- +# probe bodies contain OpenJD ``{{ ... }}`` format strings, which an f-string +# or ``str.format`` would silently mangle into ``{ ... }``. +_PROBE_PREAMBLE = """\ +import json +import sys + +sys.path[:] = json.loads(sys.argv[1]) + +RS = "openjd._openjd_rs" +""" + + +def _probe_python() -> str: + """The interpreter to run a probe with. + + Mirrors the ``python_exe`` fixture in ``sessions_v0/conftest.py``: under + Windows Session 0 ``sys.executable`` is ``pythonservice.exe``, which cannot + run a script. + """ + if sys.platform == "win32" and "pythonservice.exe" in sys.executable.lower(): + return sys.executable.lower().replace("pythonservice.exe", "python.exe") + return sys.executable + + +def _run_probe(tmp_path: Path, body: str) -> str: + """Run ``body`` in a fresh interpreter and return its stdout, stripped. + + ``body`` may use the name ``RS`` (the native extension's module name) and + ``sys``. It is written to a script file rather than passed with ``-c`` so + that a traceback is inspectable, and ``sys.path`` is handed over as a JSON + argument -- not embedded in the source -- so that a path which is not + encodable as UTF-8 source cannot break the probe. + """ + script = tmp_path / "probe.py" + script.write_text(_PROBE_PREAMBLE + textwrap.dedent(body), encoding="utf-8") + completed = subprocess.run( + [_probe_python(), str(script), json.dumps(sys.path)], + capture_output=True, + encoding="utf-8", + errors="replace", + timeout=_PROBE_TIMEOUT_SECONDS, + ) + assert completed.returncode == 0, ( + f"probe exited {completed.returncode}\n" + f"--- script ---\n{script.read_text(encoding='utf-8')}\n" + f"--- stdout ---\n{completed.stdout}\n--- stderr ---\n{completed.stderr}" + ) + return completed.stdout.strip() + + +def test_importing_sessions_does_not_load_native_extension(tmp_path: Path) -> None: + """The 0.10.11 regression: a module-level ``from openjd.expr import ...`` + on the pure-Python path makes ``import openjd.sessions`` load the native + extension.""" + # WHEN + loaded = _run_probe( + tmp_path, + """ + import openjd.sessions + + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "False", ( + "`import openjd.sessions` loaded the native extension. Something on the " + "pure-Python session path imports openjd.expr (or openjd._openjd_rs) at " + "module level; it must be imported lazily instead." + ) + + +def test_importing_sessions_does_not_resolve_openjd_expr(tmp_path: Path) -> None: + """``openjd.expr`` is the facade the extension gets pulled in through, so + it must not appear either -- this still fails on a module-level import if + some future build made the extension itself lazy.""" + # WHEN + resolved = _run_probe( + tmp_path, + """ + import openjd.sessions + + print("openjd.expr" in sys.modules) + """, + ) + + # THEN + assert resolved == "False", ( + "`import openjd.sessions` resolved openjd.expr at import time; it must " + "only be imported from inside a function on the EXPR path." + ) + + +def test_native_extension_is_available(tmp_path: Path) -> None: + """Negative control: the extension really is installed here. Without this, + every "must not be loaded" test above would pass trivially in an + environment where it simply is not present.""" + # WHEN + loaded = _run_probe( + tmp_path, + """ + import openjd.expr + + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "True", ( + "openjd.expr did not load the native extension, so the purity tests in " + "this file prove nothing. Check the openjd-model install." + ) + + +def test_probe_detects_a_loaded_extension(tmp_path: Path) -> None: + """Negative control for the probe harness: an extension loaded *before* + ``openjd.sessions`` must be observed as loaded. Guards against the checks + above passing because ``sys.modules`` is being read wrongly.""" + # WHEN + loaded = _run_probe( + tmp_path, + """ + import openjd.expr + import openjd.sessions + + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "True", "the probe cannot observe a loaded extension at all" + + +def test_resolving_a_non_expr_field_does_not_load_native_extension( + tmp_path: Path, +) -> None: + """Import-time purity is not enough: *running* a non-EXPR template must + not load the extension either. + + ``_is_expr_null`` runs for every optional int field (an action's + ``timeout``, a cancelation's ``notifyPeriodInSeconds``). Its + ``sys.modules`` guard is what keeps that path extension-free. A type-based + guard is not sufficient and this test is why: a legacy whole-field + interpolation returns the symbol's own value, so ``resolve_value`` here + returns an ``int`` -- not a ``str`` and not an ``ExprValue``. + """ + # WHEN: a legacy (non-EXPR) interpolation resolves against an int symbol. + loaded = _run_probe( + tmp_path, + """ + from openjd.model import SymbolTable + from openjd.model.v2023_09 import Action, ModelParsingContext + from openjd.sessions._runner_base import resolve_optional_int_field + + context = ModelParsingContext(supported_extensions=["FEATURE_BUNDLE_1"]) + action = Action.model_validate( + {"command": "echo", "timeout": "{{ Param.Seconds }}"}, context=context + ) + symtab = SymbolTable() + symtab["Param.Seconds"] = 30 + + resolved = resolve_optional_int_field( + action.timeout, symtab, ge=1, description="timeout" + ) + assert resolved == 30, resolved + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "False", ( + "resolving a non-EXPR timeout loaded the native extension. The " + "`sys.modules` guard in _runner_base._is_expr_null is the thing that " + "prevents this; check it has not been replaced by a type-based test." + ) + + +def test_resolving_an_expr_field_still_resolves_null(tmp_path: Path) -> None: + """Negative control for over-correction: the EXPR path must still work. + + Deliberately asserts the resolved *value*, not ``sys.modules``: parsing an + EXPR-extension template already loads the extension inside openjd.model, + so a ``sys.modules`` assertion here would hold no matter what + ``_is_expr_null`` did. A whole-field EXPR expression resolving to null must + yield ``None`` ("field omitted"), which is exactly the branch the lazy + ``ExprValue`` import serves -- if it stopped working, the strict integer + parse would raise instead. + """ + # WHEN + resolved = _run_probe( + tmp_path, + """ + from openjd.model import SymbolTable + from openjd.model.v2023_09 import Action, ModelParsingContext + from openjd.sessions._runner_base import resolve_optional_int_field + + context = ModelParsingContext(supported_extensions=["FEATURE_BUNDLE_1", "EXPR"]) + action = Action.model_validate( + {"command": "echo", "timeout": "{{ Seconds }}"}, context=context + ) + symtab = SymbolTable() + symtab["Seconds"] = None + + print(repr(resolve_optional_int_field( + action.timeout, symtab, ge=1, description="timeout" + ))) + """, + ) + + # THEN + assert resolved == "None", ( + "a whole-field EXPR expression resolving to null must be treated as " + f"'field omitted' (None), got {resolved}" + ) + + +@pytest.mark.parametrize( + "module", + [ + "openjd.sessions._runner_base", + "openjd.sessions._session", + "openjd.sessions._runner_step_script", + "openjd.sessions._runner_env_script", + ], +) +def test_pure_path_module_does_not_load_native_extension(tmp_path: Path, module: str) -> None: + """Importing any individual pure-Python-path module must stay extension + free, so the contract does not silently depend on ``__init__`` ordering. + + ``openjd.sessions._v1`` is deliberately excluded: it is the opt-in + Rust-backed API surface, is not reachable from ``openjd.sessions``, and is + expected to load the extension. + """ + # WHEN + loaded = _run_probe( + tmp_path, + f""" + import {module} + + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "False", f"importing {module} loaded the native extension"