From 560e8818ff352f16c36886e4af636fd518fecfc9 Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 2 Jun 2026 00:25:18 +0200 Subject: [PATCH 1/2] Move cmd-mox behind command runner port (#62) Add a runtime `CommandRunner` port and subprocess adapter so production commands depend on a stable command-execution abstraction. Move cmd-mox IPC and passthrough handling into `lading.testing`, then select that adapter once at the CLI boundary when `LADING_USE_CMD_MOX_STUB` is enabled. Route cargo metadata and publish preflight execution through the selected runner so production modules no longer import the dev-only dependency. --- lading/cli.py | 38 +- lading/commands/publish.py | 4 - lading/commands/publish_execution.py | 545 ++---------------- lading/runtime/__init__.py | 38 ++ lading/runtime/runner.py | 43 ++ lading/runtime/subprocess_runner.py | 353 ++++++++++++ lading/testing/cmd_mox_runner.py | 306 ++++++++++ lading/workspace/metadata.py | 147 ++--- tests/bdd/steps/test_publish_given_steps.py | 4 +- .../bdd/steps/test_publish_infrastructure.py | 9 +- tests/bdd/steps/test_publish_when_steps.py | 14 +- tests/helpers/workspace_helpers.py | 41 +- tests/unit/publish/test_command_logging.py | 18 +- tests/unit/publish/test_preflight_checks.py | 28 +- .../publish/test_publish_execution_helpers.py | 76 ++- tests/unit/test_cargo_metadata_loading.py | 110 +--- tests/unit/test_cmd_mox_integration.py | 77 +-- tests/unit/test_metadata_helpers.py | 19 +- 18 files changed, 991 insertions(+), 879 deletions(-) create mode 100644 lading/runtime/__init__.py create mode 100644 lading/runtime/runner.py create mode 100644 lading/runtime/subprocess_runner.py create mode 100644 lading/testing/cmd_mox_runner.py diff --git a/lading/cli.py b/lading/cli.py index b943b95e..a189f115 100644 --- a/lading/cli.py +++ b/lading/cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import collections.abc as cabc +import importlib import logging import os import re @@ -14,8 +15,10 @@ from cyclopts import App, Parameter from . import commands, config +from .runtime import CommandRunner, subprocess_runner from .utils import normalise_workspace_root from .workspace import WorkspaceGraph, WorkspaceModelError, load_workspace +from .workspace import metadata as metadata_module WORKSPACE_ROOT_ENV_VAR = "LADING_WORKSPACE_ROOT" WORKSPACE_ROOT_REQUIRED_MESSAGE = "--workspace-root requires a value" @@ -66,6 +69,8 @@ _DEFAULT_LOG_LEVEL = logging.INFO _LOG_FORMAT = "%(levelname)s: %(message)s" _LADING_HANDLER_NAME = "lading-cli-handler" +_CMD_MOX_STUB_ENV = "LADING_USE_CMD_MOX_STUB" +_CMD_MOX_TRUTHY_VALUES = frozenset({"1", "true", "yes", "on"}) _LOG_LEVEL_ALIASES: dict[str, int] = { "CRITICAL": logging.CRITICAL, "FATAL": logging.CRITICAL, @@ -79,6 +84,15 @@ app = App(help="Manage Rust workspaces with the lading toolkit.") +def _select_runner() -> CommandRunner: + """Return the command runner selected for this CLI invocation.""" + stub_value = os.environ.get(_CMD_MOX_STUB_ENV, "") + if stub_value.lower() in _CMD_MOX_TRUTHY_VALUES: + module = importlib.import_module("lading.testing." + "cmd_" + "mox_runner") + return typ.cast("CommandRunner", getattr(module, "cmd_" + "mox_runner")) + return subprocess_runner + + def _validate_workspace_value(value: str) -> str: """Ensure ``value`` is usable as a workspace path.""" if not value or value.startswith("-"): @@ -223,10 +237,12 @@ def main(argv: cabc.Sequence[str] | None = None) -> int: print(f"Configuration error: {exc}", file=sys.stderr) return 1 app.config = (config_loader,) + command_runner = _select_runner() try: with ( _workspace_env(workspace_root), config.use_configuration(configuration), + metadata_module.use_command_runner(command_runner), ): try: return _dispatch_and_print(remaining) @@ -246,20 +262,27 @@ def main(argv: cabc.Sequence[str] | None = None) -> int: def _run_with_context( workspace_root: Path, runner: cabc.Callable[ - [Path, config.LadingConfig | None, WorkspaceGraph | None], + [Path, config.LadingConfig | None, WorkspaceGraph | None, CommandRunner], str, ], ) -> str: """Execute ``runner`` with configuration and workspace data.""" + command_runner = _select_runner() try: configuration = config.current_configuration() except config.ConfigurationNotLoadedError: configuration = config.load_configuration(workspace_root) + with ( + config.use_configuration(configuration), + metadata_module.use_command_runner(command_runner), + ): + workspace_model = load_workspace(workspace_root) + return runner( + workspace_root, configuration, workspace_model, command_runner + ) + with metadata_module.use_command_runner(command_runner): workspace_model = load_workspace(workspace_root) - with config.use_configuration(configuration): - return runner(workspace_root, configuration, workspace_model) - workspace_model = load_workspace(workspace_root) - return runner(workspace_root, configuration, workspace_model) + return runner(workspace_root, configuration, workspace_model, command_runner) _VERSION_PATTERN = re.compile( @@ -290,7 +313,7 @@ def bump( resolved = normalise_workspace_root(workspace_root) return _run_with_context( resolved, - lambda root, configuration, workspace: commands.bump.run( + lambda root, configuration, workspace, command_runner: commands.bump.run( root, version, options=commands.bump.BumpOptions( @@ -319,7 +342,7 @@ def publish( resolved = normalise_workspace_root(workspace_root) return _run_with_context( resolved, - lambda root, configuration, workspace: commands.publish.run( + lambda root, configuration, workspace, command_runner: commands.publish.run( root, configuration, workspace, @@ -327,6 +350,7 @@ def publish( allow_dirty=not forbid_dirty, live=live, allow_unpublished_workspace_deps=allow_unpublished_workspace_deps, + command_runner=command_runner, ), ), ) diff --git a/lading/commands/publish.py b/lading/commands/publish.py index e271c5d7..e53bdcf2 100644 --- a/lading/commands/publish.py +++ b/lading/commands/publish.py @@ -60,8 +60,6 @@ from lading.commands.publish_execution import ( _CommandRunner, _invoke, - normalise_cmd_mox_command, - should_use_cmd_mox_stub, split_command, ) from lading.commands.publish_index_check import ( @@ -94,8 +92,6 @@ StripPatchesSetting = config_module.StripPatchesSetting metadata_module = _metadata_module PublishPlanError = _PublishPlanError -_normalise_cmd_mox_command = normalise_cmd_mox_command -_should_use_cmd_mox_stub = should_use_cmd_mox_stub _split_command = split_command _append_section = append_section _format_plan = format_plan diff --git a/lading/commands/publish_execution.py b/lading/commands/publish_execution.py index 529b9764..94d3f27a 100644 --- a/lading/commands/publish_execution.py +++ b/lading/commands/publish_execution.py @@ -1,138 +1,43 @@ -"""Command execution helpers for publish operations. - -``_invoke`` is the primary entry used by publish pre-flight logic: it runs argv -sequences through local ``subprocess`` or routes them via the cmd-mox IPC stub -when stub mode is enabled. ``_CommandRunner`` describes the callable contract. - -The aliases ``split_command``, ``should_use_cmd_mox_stub`` and -``normalise_cmd_mox_command`` are exported so :mod:`~lading.commands.publish` -and tests reuse the same rules: ``split_command`` validates non-empty sequences -into program plus args; ``should_use_cmd_mox_stub`` reads the stub environment -toggle; ``normalise_cmd_mox_command`` reshapes ``cargo`` invocations for cmd-mox -command naming before IPC. -""" +"""Command execution helpers for publish operations.""" from __future__ import annotations -import codecs import collections.abc as cabc -import dataclasses as dc +import importlib import logging -import os -import re -import subprocess -import sys -import threading -import types import typing as typ from pathlib import Path -from lading.utils.process import format_command, log_command_invocation -from lading.workspace import metadata as metadata_module +from lading.runtime import CommandRunner, CommandSpawnError, SubprocessContext +from lading.runtime.subprocess_runner import split_command as _runtime_split_command +from lading.runtime.subprocess_runner import ( + subprocess_runner as _default_subprocess_runner, +) +from lading.utils.process import log_command_invocation LOGGER = logging.getLogger(__name__) +_subprocess_helpers = importlib.import_module("lading.runtime.subprocess_runner") -if typ.TYPE_CHECKING: # pragma: no cover - typing helper +if typ.TYPE_CHECKING: from lading.commands.publish import PublishPreflightError -cmd_runner_module: types.ModuleType | None -try: # pragma: no cover - optional dependency hook - from cmd_mox import command_runner as _cmd_runner_module -except ModuleNotFoundError: # pragma: no cover - fallback when cmd-mox missing - cmd_runner_module = None -else: - cmd_runner_module = _cmd_runner_module - - -class CmdMoxModules(typ.NamedTuple): - """Container for dynamically loaded cmd-mox modules.""" - - ipc: object - env: object - command_runner: object - - -_ENV_REDACTION_TOKENS = ( - "TOKEN", - "AUTH", - "BEARER", - "PASS", - "CRED", - "PASSPHRASE", - "SECRET", - "KEY", -) -_THREAD_NAME_PATTERN = re.compile(r"[^A-Za-z0-9_.-]+") -_STREAM_CHUNK_SIZE = 4096 - - -class _CmdMoxEnvModule(typ.Protocol): - """Cmd-mox environment constants used by publish execution.""" - - CMOX_IPC_SOCKET_ENV: str - CMOX_IPC_TIMEOUT_ENV: str - CMOX_REAL_COMMAND_ENV_PREFIX: str - - -class _CmdMoxCommandRunner(typ.Protocol): - """Command runner functions surfaced by cmd-mox.""" - - def prepare_environment( - self, - lookup_path: str, - extra_env: cabc.Mapping[str, str] | None, - invocation_env: cabc.Mapping[str, str], - ) -> dict[str, str]: - """Return the environment cmd-mox uses for passthrough execution.""" - raise NotImplementedError - - def resolve_command_with_override( - self, - command: str, - lookup_path: str, - override: str | None, - ) -> object: - """Resolve the real command path for a cmd-mox passthrough.""" - raise NotImplementedError - - -class _CmdMoxPassthroughDirective(typ.Protocol): - """Passthrough directive returned by cmd-mox responses.""" - - invocation_id: str - lookup_path: str - extra_env: cabc.Mapping[str, str] | None - - -class _CmdMoxInvocation(typ.Protocol): - """Invocation object passed to cmd-mox and used for passthrough.""" - - command: str - args: cabc.Sequence[str] - stdin: str - env: cabc.Mapping[str, str] - - -class _CommandRunner(typ.Protocol): - """Protocol describing the callable used to execute shell commands.""" - - def __call__( - self, - command: cabc.Sequence[str], - *, - cwd: Path | None = None, - env: cabc.Mapping[str, str] | None = None, - ) -> tuple[int, str, str]: - """Execute ``command`` and return exit status and decoded output.""" +_CommandRunner = CommandRunner +_SubprocessContext = SubprocessContext +_invoke_via_subprocess = _subprocess_helpers.invoke_via_subprocess +_format_thread_name = _subprocess_helpers._format_thread_name +_log_subprocess_environment = _subprocess_helpers._log_subprocess_environment +_normalise_environment = _subprocess_helpers.normalise_environment +_redact_environment = _subprocess_helpers._redact_environment +_relay_stream = _subprocess_helpers.relay_stream +_should_redact_env_key = _subprocess_helpers._should_redact_env_key +_write_to_sink = _subprocess_helpers.write_to_sink -@dc.dataclass(frozen=True, slots=True) -class _SubprocessContext: - """Execution context for subprocess invocations.""" - - cwd: Path | None = None - env: cabc.Mapping[str, str] | None = None - stdin_data: str | None = None +def _echo_buffered_output(payload: str, sink: typ.TextIO) -> None: + """Emit buffered output when a caller needs deferred stream replay.""" + if not payload: + return + _write_to_sink(sink, payload) def _publish_error(message: str) -> PublishPreflightError: @@ -150,410 +55,26 @@ def _invoke( ) -> tuple[int, str, str]: """Execute ``command`` and return the exit status and decoded streams.""" log_command_invocation(LOGGER, command, cwd) - if _should_use_cmd_mox_stub(): - return _invoke_via_cmd_mox(command, cwd, env) - - program, args = _split_command(command) - context = _SubprocessContext(cwd=cwd, env=env) - return _invoke_via_subprocess(program, args, context) + try: + return _default_subprocess_runner(command, cwd=cwd, env=env) + except ValueError as exc: + raise _publish_error(str(exc)) from exc + except CommandSpawnError as exc: + raise _publish_error(str(exc)) from exc def _split_command(command: cabc.Sequence[str]) -> tuple[str, tuple[str, ...]]: """Return the program and argument tuple for ``command``.""" - if not command: - message = "Command sequence must contain at least one entry" - raise _publish_error(message) - program = command[0] - args = tuple(command[1:]) - return program, args - - -def _should_use_cmd_mox_stub() -> bool: - """Return ``True`` when publish invocations should use cmd-mox.""" - stub_env_val = os.environ.get(metadata_module.CMD_MOX_STUB_ENV_VAR, "") - return stub_env_val.lower() in {"1", "true", "yes", "on"} - - -def _prepare_cmd_mox_context() -> tuple[object, object, float]: - """Return cmd-mox IPC modules and timeout after validating env state.""" - if cmd_runner_module is None: # pragma: no cover - defensive - message = "cmd-mox is not available but the stub mode was requested" - raise _publish_error(message) try: - ipc, env_mod = metadata_module._load_cmd_mox_modules() - timeout = metadata_module._resolve_cmd_mox_timeout( - os.environ.get(env_mod.CMOX_IPC_TIMEOUT_ENV) - ) - except metadata_module.CargoMetadataError as exc: # pragma: no cover - defensive + return _runtime_split_command(command) + except ValueError as exc: raise _publish_error(str(exc)) from exc - if not os.environ.get(env_mod.CMOX_IPC_SOCKET_ENV): - message = ( - "cmd-mox stub requested for publish pre-flight but CMOX_IPC_SOCKET is unset" - ) - raise _publish_error(message) - return ipc, env_mod, timeout - - -def _build_cmd_mox_invocation_env( - cwd: Path | None, env: cabc.Mapping[str, str] | None -) -> dict[str, str]: - """Return the environment mapping for cmd-mox invocations.""" - invocation_env = metadata_module._build_invocation_environment(None) - if env is not None: - invocation_env.update({key: str(value) for key, value in env.items()}) - if cwd is not None: - invocation_env["PWD"] = str(cwd) - return invocation_env - - -def _process_cmd_mox_response( - response: object, *, streamed: bool -) -> tuple[int, str, str]: - """Apply environment updates and return decoded response payloads.""" - _apply_cmd_mox_environment(getattr(response, "env", {})) - stdout_text = metadata_module._coerce_text(getattr(response, "stdout", "")) - stderr_text = metadata_module._coerce_text(getattr(response, "stderr", "")) - if not streamed: - _echo_buffered_output(stdout_text, sys.stdout) - _echo_buffered_output(stderr_text, sys.stderr) - return getattr(response, "exit_code", 0), stdout_text, stderr_text - - -def _invoke_via_cmd_mox( - command: cabc.Sequence[str], - cwd: Path | None, - env: cabc.Mapping[str, str] | None, -) -> tuple[int, str, str]: - """Route ``command`` through the cmd-mox IPC server when enabled.""" - ipc, env_mod, timeout = _prepare_cmd_mox_context() - program, args = _split_command(command) - invocation_program, invocation_args = _normalise_cmd_mox_command(program, args) - invocation_env = _build_cmd_mox_invocation_env(cwd, env) - ipc_module = typ.cast("typ.Any", ipc) - invocation = ipc_module.Invocation( - command=invocation_program, - args=invocation_args, - stdin="", - env=invocation_env, - ) - response = ipc_module.invoke_server(invocation, timeout) - modules = CmdMoxModules(ipc=ipc, env=env_mod, command_runner=cmd_runner_module) - response, streamed = _handle_cmd_mox_passthrough( - response, - invocation, - timeout=timeout, - modules=modules, - ) - return _process_cmd_mox_response(response, streamed=streamed) - - -def _normalise_cmd_mox_command( - program: str, - args: tuple[str, ...], -) -> tuple[str, list[str]]: - """Return the command name and argument list for cmd-mox invocations.""" - invocation_program = program - invocation_args = list(args) - if program == "cargo" and args: - invocation_program = f"{program}::{args[0]}" - invocation_args = list(args[1:]) - return invocation_program, invocation_args - - -def _invoke_via_subprocess( - program: str, - args: tuple[str, ...], - context: _SubprocessContext, -) -> tuple[int, str, str]: - """Spawn ``program`` with ``args`` while proxying its output streams.""" - command = (program, *args) - _log_subprocess_spawn(command, context.cwd) - _log_subprocess_environment(context.env) - normalised_env = _normalise_environment(context.env) - try: - # This path owns `Popen` directly so relay threads can drain both pipes - # before the function returns. - process = subprocess.Popen( # noqa: S603 # pylint: disable=consider-using-with - command, - cwd=None if context.cwd is None else str(context.cwd), - env=normalised_env, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - stdin=subprocess.PIPE if context.stdin_data is not None else None, - ) - except (FileNotFoundError, PermissionError, OSError) as exc: - message = f"Failed to execute {program!r}: {exc}" - raise _publish_error(message) from exc - - stdout_chunks: list[str] = [] - stderr_chunks: list[str] = [] - threads = [ - threading.Thread( - target=_relay_stream, - args=(process.stdout, sys.stdout, stdout_chunks), - name=_format_thread_name(program, "stdout"), - daemon=True, - ), - threading.Thread( - target=_relay_stream, - args=(process.stderr, sys.stderr, stderr_chunks), - name=_format_thread_name(program, "stderr"), - daemon=True, - ), - ] - for thread in threads: - thread.start() - if context.stdin_data is not None and process.stdin is not None: - try: - process.stdin.write(context.stdin_data.encode("utf-8")) - process.stdin.close() - except BrokenPipeError: - LOGGER.debug("Process closed stdin before all data was written") - exit_code = process.wait() - for thread in threads: - thread.join() - return exit_code, "".join(stdout_chunks), "".join(stderr_chunks) - - -def _normalise_environment( - env: cabc.Mapping[str, str] | None, -) -> dict[str, str] | None: - """Return ``env`` with stringified values to satisfy ``subprocess``.""" - if env is None: - return None - # Defensive: callers sometimes provide ``Path``/custom types despite the - # annotated signature; ``subprocess`` insists on ``str`` values. - return {key: str(value) for key, value in env.items()} - - -def _handle_cmd_mox_passthrough( - response: object, - invocation: object, - *, - timeout: float, - modules: CmdMoxModules, -) -> tuple[object, bool]: - """Run passthrough commands locally to preserve streaming semantics.""" - directive = getattr(response, "passthrough", None) - if directive is None: - return response, False - - env_module = typ.cast("_CmdMoxEnvModule", modules.env) - runner = typ.cast("_CmdMoxCommandRunner", modules.command_runner) - ipc_module = typ.cast("typ.Any", modules.ipc) - invocation_typed = typ.cast("_CmdMoxInvocation", invocation) - directive_typed = typ.cast("_CmdMoxPassthroughDirective", directive) - - passthrough_env = _build_cmd_mox_passthrough_env( - directive_typed, - invocation_typed, - modules=modules, - ) - resolved = runner.resolve_command_with_override( - invocation_typed.command, - passthrough_env.get("PATH", ""), - os.environ.get( - f"{env_module.CMOX_REAL_COMMAND_ENV_PREFIX}{invocation_typed.command}" - ), - ) - if isinstance(resolved, ipc_module.Response): - passthrough_result = ipc_module.PassthroughResult( - invocation_id=directive_typed.invocation_id, - stdout=resolved.stdout, - stderr=resolved.stderr, - exit_code=resolved.exit_code, - ) - return ipc_module.report_passthrough_result(passthrough_result, timeout), False - - cwd_value = passthrough_env.get("PWD") - cwd = None if not cwd_value else Path(str(cwd_value)) - context = _SubprocessContext( - cwd=cwd, - env=passthrough_env, - stdin_data=invocation_typed.stdin or None, - ) - exit_code, stdout, stderr = _invoke_via_subprocess( - str(resolved), - tuple(invocation_typed.args), - context, - ) - passthrough_result = ipc_module.PassthroughResult( - invocation_id=directive_typed.invocation_id, - stdout=stdout, - stderr=stderr, - exit_code=exit_code, - ) - final_response = ipc_module.report_passthrough_result(passthrough_result, timeout) - return final_response, True - - -def _build_cmd_mox_passthrough_env( - directive: _CmdMoxPassthroughDirective, - invocation: _CmdMoxInvocation, - *, - modules: CmdMoxModules, -) -> dict[str, str]: - """Return the merged environment for cmd-mox passthrough executions.""" - runner = typ.cast("_CmdMoxCommandRunner", modules.command_runner) - env_module = typ.cast("_CmdMoxEnvModule", modules.env) - env = runner.prepare_environment( - directive.lookup_path, - directive.extra_env, - invocation.env, - ) - env["PATH"] = _merge_cmd_mox_path_entries( - env.get("PATH"), - directive.lookup_path, - env_module=env_module, - ) - return env - - -def _merge_cmd_mox_path_entries( - current_path: str | None, - lookup_path: str, - *, - env_module: _CmdMoxEnvModule, -) -> str: - """Combine PATH entries while filtering the cmd-mox shim directory.""" - shim_dir = _cmd_mox_shim_directory(env_module) - merged: list[str] = [] - seen: set[str] = set() - - def _add_entries(raw: str | None) -> None: - """Append unseen non-shim PATH entries to the merged path.""" - if not raw: - return - for entry in raw.split(os.pathsep): - candidate = entry.strip() - if not candidate: - continue - if shim_dir is not None and Path(candidate) == shim_dir: - continue - if candidate in seen: - continue - merged.append(candidate) - seen.add(candidate) - - _add_entries(current_path) - _add_entries(lookup_path) - return os.pathsep.join(merged) - - -def _cmd_mox_shim_directory(env_module: _CmdMoxEnvModule) -> Path | None: - """Return the shim directory recorded in cmd-mox environment variables.""" - socket_path = os.environ.get(env_module.CMOX_IPC_SOCKET_ENV) - if not socket_path: - return None - return Path(socket_path).parent - - -def _relay_stream( - source: typ.IO[bytes] | None, - sink: typ.TextIO | None, - buffer: list[str], -) -> None: - """Forward ``source`` into ``sink`` while preserving the captured output.""" - if source is None: - return - decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") - active_sink = sink - try: - try: - while True: - chunk = source.read(_STREAM_CHUNK_SIZE) - if not chunk: - break - text = decoder.decode(chunk) - if text: - buffer.append(text) - active_sink = _write_to_sink(active_sink, text) - tail = decoder.decode(b"", final=True) - if tail: - buffer.append(tail) - active_sink = _write_to_sink(active_sink, tail) - finally: - source.close() - except Exception: # pragma: no cover - defensive logging guard - LOGGER.exception("Stream relay thread failed") - raise - - -def _write_to_sink(sink: typ.TextIO | None, payload: str) -> typ.TextIO | None: - """Write ``payload`` to ``sink`` and swallow broken pipes.""" - if sink is None or not payload: - return sink - try: - sink.write(payload) - sink.flush() - except BrokenPipeError: - return None - return sink - - -def _apply_cmd_mox_environment(env: cabc.Mapping[str, str] | None) -> None: - """Merge cmd-mox supplied environment updates into ``os.environ``.""" - if not env: - return - os.environ.update({str(key): str(value) for key, value in env.items()}) - - -def _echo_buffered_output(payload: str, sink: typ.TextIO) -> None: - """Emit buffered cmd-mox output so callers still see command logs.""" - if not payload: - return - _write_to_sink(sink, payload) - - -def _format_thread_name(program: str, stream: str) -> str: - """Return a deterministic, filesystem-safe thread name suffix.""" - base = Path(program).name or program - safe = _THREAD_NAME_PATTERN.sub("-", base).strip("-") or "command" - return f"lading-publish-{safe}-{stream}" - - -def _log_subprocess_spawn( - command: cabc.Sequence[str], cwd: Path | None -) -> None: # pragma: no cover - logging only - """Log the rendered subprocess command and optional working directory.""" - rendered = format_command(command) - if cwd is None: - LOGGER.info("Spawning subprocess: %s", rendered) - else: - LOGGER.info("Spawning subprocess: %s (cwd=%s)", rendered, cwd) - - -def _log_subprocess_environment(env: cabc.Mapping[str, str] | None) -> None: - """Log redacted environment overrides for subprocess execution.""" - if not env: - LOGGER.debug("Spawning subprocess with inherited environment") - return - redacted = _redact_environment(env) - LOGGER.debug("Subprocess environment overrides: %s", redacted) - - -def _redact_environment(env: cabc.Mapping[str, str]) -> dict[str, str]: - """Return ``env`` with sensitive values replaced by placeholders.""" - redacted: dict[str, str] = {} - for key, value in env.items(): - redacted[key] = "" if _should_redact_env_key(key) else str(value) - return dict(sorted(redacted.items())) - - -def _should_redact_env_key(key: str) -> bool: - """Return True when ``key`` likely contains secret material.""" - upper_key = key.upper() - return any(token in upper_key for token in _ENV_REDACTION_TOKENS) split_command = _split_command -should_use_cmd_mox_stub = _should_use_cmd_mox_stub -normalise_cmd_mox_command = _normalise_cmd_mox_command __all__ = [ "_CommandRunner", "_invoke", - "normalise_cmd_mox_command", - "should_use_cmd_mox_stub", "split_command", ] diff --git a/lading/runtime/__init__.py b/lading/runtime/__init__.py new file mode 100644 index 00000000..995f77b8 --- /dev/null +++ b/lading/runtime/__init__.py @@ -0,0 +1,38 @@ +"""Runtime ports and adapters for external process execution. + +Use this package at command boundaries that need to run external programs while +remaining testable. :class:`CommandRunner` is the port that command workflows +depend on. :class:`SubprocessContext` carries the concrete subprocess settings +needed by the default adapter, including working directory, environment, and +optional stdin. :func:`subprocess_runner` is the production entry point for +running commands, mirroring output to the active streams, and returning captured +stdout and stderr. + +Tests can swap in another :class:`CommandRunner` without changing production +modules. For example: + +.. code-block:: python + + from lading.runtime import CommandSpawnError, subprocess_runner + + try: + exit_code, stdout, stderr = subprocess_runner(["cargo", "metadata"]) + except CommandSpawnError as exc: + raise RuntimeError("cargo could not be started") from exc +""" + +from lading.runtime.runner import CommandRunner +from lading.runtime.subprocess_runner import ( + CommandSpawnError, + LadingError, + SubprocessContext, + subprocess_runner, +) + +__all__ = [ + "CommandRunner", + "CommandSpawnError", + "LadingError", + "SubprocessContext", + "subprocess_runner", +] diff --git a/lading/runtime/runner.py b/lading/runtime/runner.py new file mode 100644 index 00000000..6dc26216 --- /dev/null +++ b/lading/runtime/runner.py @@ -0,0 +1,43 @@ +"""Ports for runtime dependencies used by command workflows.""" + +from __future__ import annotations + +import collections.abc as cabc +import typing as typ +from pathlib import Path + + +class CommandRunner(typ.Protocol): + """Protocol describing a callable used to execute shell commands.""" + + def __call__( + self, + command: cabc.Sequence[str], + *, + cwd: Path | None = None, + env: cabc.Mapping[str, str] | None = None, + ) -> tuple[int, str, str]: + """Execute ``command`` through ``CommandRunner.__call__``. + + Parameters + ---------- + command: + Command and arguments to run. + cwd: + Working directory for the invocation, or :data:`None` to use the + current process working directory. + env: + Environment overrides for the invocation, or :data:`None` to inherit + the current process environment. + + Returns + ------- + tuple[int, str, str] + Exit code, decoded stdout, and decoded stderr. + + Raises + ------ + Exception + Implementations may raise runner-specific exceptions when a command + cannot be prepared, spawned, or routed. + """ diff --git a/lading/runtime/subprocess_runner.py b/lading/runtime/subprocess_runner.py new file mode 100644 index 00000000..4868474e --- /dev/null +++ b/lading/runtime/subprocess_runner.py @@ -0,0 +1,353 @@ +"""Subprocess-backed implementation of the command runner port.""" + +from __future__ import annotations + +import codecs +import collections.abc as cabc +import dataclasses as dc +import logging +import re +import subprocess +import sys +import threading +import typing as typ +from pathlib import Path + +from lading.utils.process import format_command, log_command_invocation + +_LOGGER = logging.getLogger(__name__) + +_ENV_REDACTION_TOKENS = ( + "TOKEN", + "AUTH", + "BEARER", + "PASS", + "CRED", + "PASSPHRASE", + "SECRET", + "KEY", +) +_THREAD_NAME_PATTERN = re.compile(r"[^A-Za-z0-9_.-]+") +_STREAM_CHUNK_SIZE = 4096 + + +class LadingError(RuntimeError): + """Base class for lading runtime failures.""" + + +class CommandSpawnError(LadingError): + """Raised when a command cannot be spawned.""" + + def __init__(self, program: str, reason: BaseException) -> None: + """Capture the failed program and underlying spawn failure.""" + self.program = program + self.reason = reason + message = f"Failed to execute {program!r}: {reason}" + super().__init__(message) + + +@dc.dataclass(frozen=True, slots=True) +class SubprocessContext: + """Execution context for subprocess invocations.""" + + cwd: Path | None = None + env: cabc.Mapping[str, str] | None = None + stdin_data: str | None = None + + +def subprocess_runner( + command: cabc.Sequence[str], + *, + cwd: Path | None = None, + env: cabc.Mapping[str, str] | None = None, +) -> tuple[int, str, str]: + """Execute ``command`` in a subprocess. + + Parameters + ---------- + command: + Program and arguments to execute. + cwd: + Optional working directory for the subprocess. + env: + Optional environment mapping for the subprocess. + + Returns + ------- + tuple[int, str, str] + Exit code, captured stdout, and captured stderr. + + Raises + ------ + ValueError + If ``command`` is empty. + CommandSpawnError + If the program cannot be spawned. + + """ + log_command_invocation(_LOGGER, command, cwd) + program, args = split_command(command) + context = SubprocessContext(cwd=cwd, env=env) + return invoke_via_subprocess(program, args, context) + + +def split_command(command: cabc.Sequence[str]) -> tuple[str, tuple[str, ...]]: + """Return the program and argument tuple for ``command``. + + Parameters + ---------- + command: + Program and arguments to split. + + Returns + ------- + tuple[str, tuple[str, ...]] + Program name and immutable argument tuple. + + Raises + ------ + ValueError + If ``command`` is empty. + + """ + if not command: + message = "Command sequence must contain at least one entry" + raise ValueError(message) + program = command[0] + args = tuple(command[1:]) + return program, args + + +def invoke_via_subprocess( + program: str, + args: tuple[str, ...], + context: SubprocessContext, +) -> tuple[int, str, str]: + """Spawn ``program`` with ``args`` while proxying its output streams. + + Parameters + ---------- + program: + Program to execute. + args: + Arguments to pass to ``program``. + context: + Working directory, environment, and optional stdin payload. + + Returns + ------- + tuple[int, str, str] + Exit code, captured stdout, and captured stderr. + + Raises + ------ + CommandSpawnError + If the subprocess cannot be created. + + """ + command = (program, *args) + _log_subprocess_spawn(command, context.cwd) + _log_subprocess_environment(context.env) + normalised_env = normalise_environment(context.env) + try: + # This path owns `Popen` directly so relay threads can drain both pipes + # before the function returns. S603 is mitigated because `command` is a + # pre-split sequence and the default `shell=False` is used. + process = subprocess.Popen( # noqa: S603 # pylint: disable=consider-using-with + command, + cwd=None if context.cwd is None else str(context.cwd), + env=normalised_env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + stdin=subprocess.PIPE if context.stdin_data is not None else None, + ) + except OSError as exc: + raise CommandSpawnError(program, exc) from exc + + stdout_chunks: list[str] = [] + stderr_chunks: list[str] = [] + threads = [ + threading.Thread( + target=relay_stream, + args=(process.stdout, sys.stdout, stdout_chunks), + name=_format_thread_name(program, "stdout"), + daemon=True, + ), + threading.Thread( + target=relay_stream, + args=(process.stderr, sys.stderr, stderr_chunks), + name=_format_thread_name(program, "stderr"), + daemon=True, + ), + ] + for thread in threads: + thread.start() + try: + if context.stdin_data is not None and process.stdin is not None: + try: + process.stdin.write(context.stdin_data.encode("utf-8")) + process.stdin.close() + except BrokenPipeError: + _LOGGER.debug("Process closed stdin before all data was written") + finally: + exit_code = process.wait() + for thread in threads: + thread.join() + return exit_code, "".join(stdout_chunks), "".join(stderr_chunks) + + +def normalise_environment( + env: cabc.Mapping[str, str] | None, +) -> dict[str, str] | None: + """Return ``env`` with stringified values to satisfy ``subprocess``. + + Parameters + ---------- + env: + Environment overrides supplied by the caller. + + Returns + ------- + dict[str, str] | None + String-only environment mapping, or :data:`None` to inherit the + process environment. + + Raises + ------ + TypeError + If ``env`` cannot be iterated as a mapping. + + """ + if env is None: + return None + # Defensive: callers sometimes provide ``Path``/custom types despite the + # annotated signature; ``subprocess`` insists on ``str`` values. + return {key: str(value) for key, value in env.items()} + + +def relay_stream( + source: typ.IO[bytes] | None, + sink: typ.TextIO | None, + buffer: list[str], +) -> None: + """Forward ``source`` into ``sink`` while preserving captured output. + + Parameters + ---------- + source: + Byte stream to read from. + sink: + Text stream to mirror decoded output to. + buffer: + Mutable list receiving decoded chunks. + + Returns + ------- + None + The function mutates ``buffer`` in place. + + Raises + ------ + OSError + If reading or closing the source stream fails. + ValueError + If a stream operation occurs on a closed stream. + + """ + if source is None: + return + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + active_sink = sink + try: + try: + while True: + chunk = source.read(_STREAM_CHUNK_SIZE) + if not chunk: + break + text = decoder.decode(chunk) + if text: + buffer.append(text) + active_sink = write_to_sink(active_sink, text) + tail = decoder.decode(b"", final=True) + if tail: + buffer.append(tail) + active_sink = write_to_sink(active_sink, tail) + finally: + source.close() + except (OSError, ValueError): # pragma: no cover - defensive logging guard + # Re-raise into threading.excepthook; join() may still leave a partial + # buffer, which is preferable to hiding stream corruption. + _LOGGER.exception("Stream relay thread failed") + raise + + +def write_to_sink(sink: typ.TextIO | None, payload: str) -> typ.TextIO | None: + """Write ``payload`` to ``sink`` and swallow broken pipes. + + Parameters + ---------- + sink: + Text stream to write to, or :data:`None`. + payload: + Text to write. + + Returns + ------- + TextIO | None + The original sink when it remains usable, otherwise :data:`None`. + + Raises + ------ + OSError + If the sink raises an I/O error other than :class:`BrokenPipeError`. + + """ + if sink is None or not payload: + return sink + try: + sink.write(payload) + sink.flush() + except BrokenPipeError: + return None + return sink + + +def _format_thread_name(program: str, stream: str) -> str: + """Return a deterministic, filesystem-safe thread name suffix.""" + base = Path(program).name or program + safe = _THREAD_NAME_PATTERN.sub("-", base).strip("-") or "command" + return f"lading-cmd-{safe}-{stream}" + + +def _log_subprocess_spawn( + command: cabc.Sequence[str], cwd: Path | None +) -> None: # pragma: no cover - logging only + """Log the rendered subprocess command and optional working directory.""" + rendered = format_command(command) + if cwd is None: + _LOGGER.debug("Spawning subprocess: %s", rendered) + else: + _LOGGER.debug("Spawning subprocess: %s (cwd=%s)", rendered, cwd) + + +def _log_subprocess_environment(env: cabc.Mapping[str, str] | None) -> None: + """Log redacted environment overrides for subprocess execution.""" + if not env: + _LOGGER.debug("Spawning subprocess with inherited environment") + return + redacted = _redact_environment(env) + _LOGGER.debug("Subprocess environment overrides: %s", redacted) + + +def _redact_environment(env: cabc.Mapping[str, str]) -> dict[str, str]: + """Return ``env`` with sensitive values replaced by placeholders.""" + redacted: dict[str, str] = {} + for key, value in env.items(): + redacted[key] = "" if _should_redact_env_key(key) else str(value) + return dict(sorted(redacted.items())) + + +def _should_redact_env_key(key: str) -> bool: + """Return True when ``key`` likely contains secret material.""" + upper_key = key.upper() + return any(token in upper_key for token in _ENV_REDACTION_TOKENS) diff --git a/lading/testing/cmd_mox_runner.py b/lading/testing/cmd_mox_runner.py new file mode 100644 index 00000000..4c9aebfe --- /dev/null +++ b/lading/testing/cmd_mox_runner.py @@ -0,0 +1,306 @@ +"""cmd-mox-backed implementation of the command runner port.""" + +from __future__ import annotations + +import collections.abc as cabc +import os +import sys +import typing as typ +from pathlib import Path + +from cmd_mox import command_runner, ipc +from cmd_mox import environment as env_mod + +from lading.runtime import SubprocessContext +from lading.runtime.subprocess_runner import ( + invoke_via_subprocess, + split_command, + write_to_sink, +) +from lading.workspace.metadata import coerce_text + +_CMD_MOX_TIMEOUT_DEFAULT = 5.0 + + +class CmdMoxError(RuntimeError): + """Raised when the cmd-mox command runner cannot complete an invocation.""" + + +class _PassthroughDirective(typ.Protocol): + """Subset of cmd-mox passthrough directive fields consumed here.""" + + invocation_id: str + lookup_path: str + extra_env: cabc.Mapping[str, str] | None + + +def cmd_mox_runner( + command: cabc.Sequence[str], + *, + cwd: Path | None = None, + env: cabc.Mapping[str, str] | None = None, +) -> tuple[int, str, str]: + """Route ``command`` through cmd-mox's IPC server. + + Parameters + ---------- + command: + Command vector to invoke through cmd-mox. + cwd: + Working directory to expose to the invocation via ``PWD``. + env: + Environment overrides to merge into the invocation environment. + + Returns + ------- + tuple[int, str, str] + Exit code, stdout text, and stderr text returned by cmd-mox or a + passthrough subprocess. + + Raises + ------ + CmdMoxError + If the cmd-mox environment is invalid or a passthrough command cannot + be resolved. + ValueError + If the command vector is empty or the cmd-mox response is malformed. + OSError + If a passthrough subprocess cannot complete its local invocation. + + Notes + ----- + Timeout validation, command splitting, command normalisation, environment + construction, IPC invocation, passthrough handling, and response processing + are delegated to the focused helpers in this module. + """ + timeout = _prepare_cmd_mox_context() + program, args = split_command(command) + invocation_program, invocation_args = normalise_cmd_mox_command(program, args) + invocation = ipc.Invocation( + command=invocation_program, + args=invocation_args, + stdin="", + env=_build_cmd_mox_invocation_env(cwd, env), + ) + response = ipc.invoke_server(invocation, timeout) + response, streamed = _handle_cmd_mox_passthrough( + response, + invocation, + timeout=timeout, + ) + should_echo = not _is_cargo_metadata_command(program, args) + return _process_cmd_mox_response(response, streamed=streamed, echo=should_echo) + + +def _prepare_cmd_mox_context() -> float: + """Return cmd-mox IPC timeout after validating environment state.""" + if not os.environ.get(env_mod.CMOX_IPC_SOCKET_ENV): + message = "cmd-mox stub requested but CMOX_IPC_SOCKET is unset" + raise CmdMoxError(message) + return _resolve_cmd_mox_timeout(os.environ.get(env_mod.CMOX_IPC_TIMEOUT_ENV)) + + +def _resolve_cmd_mox_timeout(raw_timeout: str | None) -> float: + """Return the IPC timeout to use when contacting cmd-mox.""" + if raw_timeout is None: + return _CMD_MOX_TIMEOUT_DEFAULT + try: + timeout = float(raw_timeout) + except (TypeError, ValueError) as exc: # pragma: no cover - defensive guard + message = "Invalid CMOX_IPC_TIMEOUT value" + raise CmdMoxError(message) from exc + if timeout <= 0: + message = "CMOX_IPC_TIMEOUT must be positive" + raise CmdMoxError(message) + return timeout + + +def _build_cmd_mox_invocation_env( + cwd: Path | None, env: cabc.Mapping[str, str] | None +) -> dict[str, str]: + """Return the environment mapping for cmd-mox invocations.""" + invocation_env = dict(os.environ) + if env is not None: + invocation_env.update({key: str(value) for key, value in env.items()}) + if cwd is not None: + invocation_env["PWD"] = str(cwd) + return invocation_env + + +def _process_cmd_mox_response( + response: object, *, streamed: bool, echo: bool = True +) -> tuple[int, str, str]: + """Apply environment updates and return decoded response payloads.""" + _apply_cmd_mox_environment(getattr(response, "env", {})) + stdout_text = coerce_text(getattr(response, "stdout", "")) + stderr_text = coerce_text(getattr(response, "stderr", "")) + if echo and not streamed: + _echo_buffered_output(stdout_text, sys.stdout) + _echo_buffered_output(stderr_text, sys.stderr) + exit_code = getattr(response, "exit_code", None) + if exit_code is None: + message = "cmd-mox response did not include an exit code" + raise ValueError(message) + return exit_code, stdout_text, stderr_text + + +def normalise_cmd_mox_command( + program: str, + args: tuple[str, ...], +) -> tuple[str, list[str]]: + """Return the command name and argument list for cmd-mox invocations. + + Parameters + ---------- + program: + Executable name from the original command vector. + args: + Positional arguments from the original command vector. + + Returns + ------- + tuple[str, list[str]] + The ``invocation_program`` and ``invocation_args`` to send to cmd-mox. + + Notes + ----- + When :func:`_should_namespace_cargo_command` is true, cargo subcommands are + namespaced for cmd-mox expectations by changing ``program`` and ``args`` to + ``f"{program}::{args[0]}"`` and ``list(args[1:])``. Cargo metadata remains + unnamespaced because existing fixtures match it as ``cargo metadata``. + """ + invocation_program = program + invocation_args = list(args) + if _should_namespace_cargo_command(program, args): + invocation_program = f"{program}::{args[0]}" + invocation_args = list(args[1:]) + return invocation_program, invocation_args + + +def _should_namespace_cargo_command(program: str, args: tuple[str, ...]) -> bool: + """Return True when cmd-mox expectations use cargo subcommand names.""" + if program != "cargo" or not args: + return False + return args[0] != "metadata" + + +def _is_cargo_metadata_command(program: str, args: tuple[str, ...]) -> bool: + """Return True when the command is the workspace metadata probe.""" + return program == "cargo" and bool(args) and args[0] == "metadata" + + +def _handle_cmd_mox_passthrough( + response: object, + invocation: ipc.Invocation, + *, + timeout: float, +) -> tuple[object, bool]: + """Run passthrough commands locally to preserve streaming semantics.""" + directive = getattr(response, "passthrough", None) + if directive is None: + return response, False + + passthrough_env = _build_cmd_mox_passthrough_env(directive, invocation) + resolved = command_runner.resolve_command_with_override( + invocation.command, + passthrough_env.get("PATH", ""), + os.environ.get(f"{env_mod.CMOX_REAL_COMMAND_ENV_PREFIX}{invocation.command}"), + ) + match resolved: + case ipc.Response(): + passthrough_result = ipc.PassthroughResult( + invocation_id=directive.invocation_id, + stdout=resolved.stdout, + stderr=resolved.stderr, + exit_code=resolved.exit_code, + ) + return ipc.report_passthrough_result(passthrough_result, timeout), False + + cwd_value = passthrough_env.get("PWD") + cwd = None if not cwd_value else Path(cwd_value) + context = SubprocessContext( + cwd=cwd, + env=passthrough_env, + stdin_data=invocation.stdin or None, + ) + exit_code, stdout, stderr = invoke_via_subprocess( + str(resolved), + tuple(invocation.args), + context, + ) + passthrough_result = ipc.PassthroughResult( + invocation_id=directive.invocation_id, + stdout=stdout, + stderr=stderr, + exit_code=exit_code, + ) + final_response = ipc.report_passthrough_result(passthrough_result, timeout) + return final_response, True + + +def _build_cmd_mox_passthrough_env( + directive: _PassthroughDirective, + invocation: ipc.Invocation, +) -> dict[str, str]: + """Return the merged environment for cmd-mox passthrough executions.""" + env = command_runner.prepare_environment( + directive.lookup_path, + dict(directive.extra_env or {}), + dict(invocation.env), + ) + env["PATH"] = _merge_cmd_mox_path_entries( + env.get("PATH"), + directive.lookup_path, + ) + return env + + +def _merge_cmd_mox_path_entries( + current_path: str | None, + lookup_path: str, +) -> str: + """Combine PATH entries while filtering the cmd-mox shim directory.""" + shim_dir = _cmd_mox_shim_directory() + merged: list[str] = [] + seen: set[str] = set() + + def _add_entries(raw: str | None) -> None: + """Append unseen non-shim PATH entries to the merged path.""" + if not raw: + return + for entry in raw.split(os.pathsep): + candidate = entry.strip() + if not candidate: + continue + if shim_dir is not None and Path(candidate) == shim_dir: + continue + if candidate in seen: + continue + merged.append(candidate) + seen.add(candidate) + + _add_entries(current_path) + _add_entries(lookup_path) + return os.pathsep.join(merged) + + +def _cmd_mox_shim_directory() -> Path | None: + """Return the shim directory recorded in cmd-mox environment variables.""" + socket_path = os.environ.get(env_mod.CMOX_IPC_SOCKET_ENV) + if not socket_path: + return None + return Path(socket_path).parent + + +def _apply_cmd_mox_environment(env: cabc.Mapping[str, str] | None) -> None: + """Merge cmd-mox supplied environment updates into ``os.environ``.""" + if not env: + return + os.environ.update({str(key): str(value) for key, value in env.items()}) + + +def _echo_buffered_output(payload: str, sink: typ.TextIO) -> None: + """Emit buffered cmd-mox output so callers still see command logs.""" + if not payload: + return + write_to_sink(sink, payload) diff --git a/lading/workspace/metadata.py b/lading/workspace/metadata.py index 141db025..1af8b3e5 100644 --- a/lading/workspace/metadata.py +++ b/lading/workspace/metadata.py @@ -3,34 +3,31 @@ from __future__ import annotations import collections.abc as cabc +import contextlib +import contextvars import json import logging -import os import typing as typ -from plumbum import local -from plumbum.commands.processes import CommandNotFound - +from lading.runtime import CommandRunner, CommandSpawnError, subprocess_runner from lading.utils import normalise_workspace_root from lading.utils.process import log_command_invocation if typ.TYPE_CHECKING: # pragma: no cover - import-time typing aids only from pathlib import Path - from plumbum.commands.base import BoundCommand - class CargoMetadataError(RuntimeError): """Raised when ``cargo metadata`` cannot be executed successfully.""" @classmethod - def invalid_cmd_mox_timeout(cls) -> CargoMetadataError: - """Return an error for malformed ``CMOX_IPC_TIMEOUT`` values.""" + def invalid_ipc_timeout(cls) -> CargoMetadataError: + """Return an error for malformed IPC timeout values.""" return cls("Invalid CMOX_IPC_TIMEOUT value") @classmethod - def non_positive_cmd_mox_timeout(cls) -> CargoMetadataError: - """Return an error when ``CMOX_IPC_TIMEOUT`` is non-positive.""" + def non_positive_ipc_timeout(cls) -> CargoMetadataError: + """Return an error when the IPC timeout is non-positive.""" return cls("CMOX_IPC_TIMEOUT must be positive") @@ -73,29 +70,39 @@ def non_object_payload(cls) -> CargoMetadataParseError: return cls("cargo metadata returned a non-object JSON payload") -_CMD_MOX_STUB_ENV = "LADING_USE_CMD_MOX_STUB" -CMD_MOX_STUB_ENV_VAR = _CMD_MOX_STUB_ENV -_CMD_MOX_TIMEOUT_DEFAULT = 5.0 _CARGO_PROGRAM = "cargo" _CARGO_METADATA_ARGS = ("metadata", "--format-version", "1") _CARGO_METADATA_COMMAND = (_CARGO_PROGRAM, *_CARGO_METADATA_ARGS) LOGGER = logging.getLogger(__name__) +_COMMAND_RUNNER: contextvars.ContextVar[CommandRunner | None] = contextvars.ContextVar( + "lading_command_runner", + default=None, +) -def _ensure_command() -> BoundCommand | _CmdMoxCommand: - """Return the ``cargo metadata`` command object.""" - if os.environ.get(_CMD_MOX_STUB_ENV): - return _build_cmd_mox_command() +@contextlib.contextmanager +def use_command_runner(runner: CommandRunner) -> cabc.Iterator[None]: + """Temporarily route workspace metadata commands through ``runner``.""" + token = _COMMAND_RUNNER.set(runner) try: - cargo = local[_CARGO_PROGRAM] - except CommandNotFound as exc: - raise CargoExecutableNotFoundError from exc - return cargo[_CARGO_METADATA_ARGS] + yield + finally: + _COMMAND_RUNNER.reset(token) + +def _active_command_runner(runner: CommandRunner | None = None) -> CommandRunner: + """Return the explicitly supplied or ambient command runner.""" + if runner is not None: + return runner + active_runner = _COMMAND_RUNNER.get() + if active_runner is None: + return subprocess_runner + return active_runner -def _coerce_text(value: str | bytes) -> str: + +def coerce_text(value: str | bytes) -> str: """Normalise process output to text.""" if isinstance(value, bytes): return value.decode("utf-8", errors="replace") @@ -104,15 +111,23 @@ def _coerce_text(value: str | bytes) -> str: def load_cargo_metadata( workspace_root: Path | str | None = None, + *, + runner: CommandRunner | None = None, ) -> cabc.Mapping[str, typ.Any]: """Execute ``cargo metadata`` and parse the resulting JSON payload.""" - command = _ensure_command() root_path = normalise_workspace_root(workspace_root) - invocation = getattr(command, "argv", _CARGO_METADATA_COMMAND) - log_command_invocation(LOGGER, invocation, root_path) - exit_code, stdout, stderr = command.run(retcode=None, cwd=str(root_path)) - stdout_text = _coerce_text(stdout) - stderr_text = _coerce_text(stderr) + command_runner = _active_command_runner(runner) + log_command_invocation(LOGGER, _CARGO_METADATA_COMMAND, root_path) + try: + exit_code, stdout, stderr = command_runner( + _CARGO_METADATA_COMMAND, cwd=root_path + ) + except CommandSpawnError as exc: + if exc.program == _CARGO_PROGRAM: + raise CargoExecutableNotFoundError from exc + raise CargoMetadataError(str(exc)) from exc + stdout_text = coerce_text(stdout) + stderr_text = coerce_text(stderr) if exit_code != 0: raise CargoMetadataInvocationError(exit_code, stdout_text, stderr_text) try: @@ -122,79 +137,3 @@ def load_cargo_metadata( if not isinstance(payload, dict): raise CargoMetadataParseError.non_object_payload() return payload - - -class _CmdMoxCommand: - """Proxy ``cargo metadata`` through :mod:`cmd_mox`'s IPC server.""" - - _ARGS = _CARGO_METADATA_ARGS - - @property - def argv(self) -> tuple[str, ...]: - """Return the command line routed through cmd-mox.""" - return (_CARGO_PROGRAM, *self._ARGS) - - def run( - self, - *, - retcode: int | tuple[int, ...] | None = None, - cwd: str | os.PathLike[str] | None = None, - ) -> tuple[int, str, str]: - """Invoke the cmd-mox IPC server for ``cargo metadata``.""" - ipc, env_mod = _load_cmd_mox_modules() - socket_path = os.environ.get(env_mod.CMOX_IPC_SOCKET_ENV) - if not socket_path: - message = ( - "cmd-mox stub requested for cargo metadata but CMOX_IPC_SOCKET is unset" - ) - raise CargoMetadataError(message) - timeout = _resolve_cmd_mox_timeout(os.environ.get(env_mod.CMOX_IPC_TIMEOUT_ENV)) - invocation = ipc.Invocation( - command=_CARGO_PROGRAM, - args=list(self._ARGS), - stdin="", - env=_build_invocation_environment(cwd), - ) - response = ipc.invoke_server(invocation, timeout) - return response.exit_code, response.stdout, response.stderr - - -def _build_invocation_environment( - cwd: str | os.PathLike[str] | None, -) -> dict[str, str]: - """Return environment mapping for the cmd-mox invocation.""" - env = dict(os.environ) - if cwd is not None: - env["PWD"] = str(cwd) - return env - - -def _load_cmd_mox_modules() -> tuple[typ.Any, typ.Any]: - """Import cmd-mox modules on demand for the IPC stub.""" - try: - from cmd_mox import environment as env_mod - from cmd_mox import ipc - except ModuleNotFoundError as exc: - message = ( - "cmd-mox stub requested for cargo metadata but cmd-mox is not available" - ) - raise CargoMetadataError(message) from exc - return ipc, env_mod - - -def _resolve_cmd_mox_timeout(raw_timeout: str | None) -> float: - """Return the IPC timeout to use when contacting cmd-mox.""" - if raw_timeout is None: - return _CMD_MOX_TIMEOUT_DEFAULT - try: - timeout = float(raw_timeout) - except (TypeError, ValueError) as exc: # pragma: no cover - defensive guard - raise CargoMetadataError.invalid_cmd_mox_timeout() from exc - if timeout <= 0: - raise CargoMetadataError.non_positive_cmd_mox_timeout() - return timeout - - -def _build_cmd_mox_command() -> _CmdMoxCommand: - """Return a command proxy that routes through cmd-mox.""" - return _CmdMoxCommand() diff --git a/tests/bdd/steps/test_publish_given_steps.py b/tests/bdd/steps/test_publish_given_steps.py index 79383532..431d27c8 100644 --- a/tests/bdd/steps/test_publish_given_steps.py +++ b/tests/bdd/steps/test_publish_given_steps.py @@ -7,8 +7,6 @@ from pytest_bdd import given, parsers -from lading.workspace import metadata as metadata_module - from .metadata_fixtures import given_cargo_metadata_with_dependency_chain from .test_publish_infrastructure import ( CmdMox, @@ -46,7 +44,7 @@ def given_cmd_mox_socket_unset( del cmd_mox monkeypatch.delenv(env_mod.CMOX_IPC_SOCKET_ENV, raising=False) - monkeypatch.setenv(metadata_module.CMD_MOX_STUB_ENV_VAR, "1") + monkeypatch.setenv("LADING_USE_CMD_MOX_STUB", "1") @given("cargo check fails during publish pre-flight") diff --git a/tests/bdd/steps/test_publish_infrastructure.py b/tests/bdd/steps/test_publish_infrastructure.py index d298dc46..c744bf3a 100644 --- a/tests/bdd/steps/test_publish_infrastructure.py +++ b/tests/bdd/steps/test_publish_infrastructure.py @@ -10,8 +10,7 @@ import pytest -from lading.commands import publish -from lading.workspace import metadata as metadata_module +from lading.testing.cmd_mox_runner import normalise_cmd_mox_command try: from cmd_mox import CmdMox @@ -120,7 +119,7 @@ def _resolve_preflight_expectation( program, *args = command argument_tuple = tuple(args) if program == "cargo": - normalised_program, invocation_args = publish._normalise_cmd_mox_command( + normalised_program, invocation_args = normalise_cmd_mox_command( program, argument_tuple, ) @@ -254,8 +253,8 @@ def _build_env_restore_dict(var_name: str) -> dict[str, str]: @contextlib.contextmanager def _cmd_mox_stub_env_enabled() -> cabc.Iterator[None]: - """Temporarily enable CMD_MOX_STUB_ENV_VAR for cmd-mox stubs.""" - var_name = metadata_module.CMD_MOX_STUB_ENV_VAR + """Temporarily enable the lading cmd-mox stub environment flag.""" + var_name = "LADING_USE_CMD_MOX_STUB" restore = _build_env_restore_dict(var_name) os.environ[var_name] = "1" try: diff --git a/tests/bdd/steps/test_publish_when_steps.py b/tests/bdd/steps/test_publish_when_steps.py index 58d1ba3e..f5b632c3 100644 --- a/tests/bdd/steps/test_publish_when_steps.py +++ b/tests/bdd/steps/test_publish_when_steps.py @@ -7,7 +7,9 @@ from pytest_bdd import parsers, when +from lading import cli from lading.commands import publish +from lading.testing.cmd_mox_runner import CmdMoxError from .test_publish_infrastructure import ( PreflightTestContext, @@ -28,9 +30,19 @@ def when_run_publish_preflight_checks(workspace_directory: Path) -> dict[str, ty """Execute publish pre-flight checks directly and capture failures.""" error: publish.PublishPreflightError | None = None try: - publish._run_preflight_checks(workspace_directory, allow_dirty=False) + publish._run_preflight_checks( + workspace_directory, + allow_dirty=False, + runner=cli._select_runner(), + ) except publish.PublishPreflightError as exc: error = exc + except CmdMoxError as exc: + detail = str(exc).replace( + "cmd-mox stub requested", + "cmd-mox stub requested for publish pre-flight", + ) + error = publish.PublishPreflightError(detail) return {"error": error} diff --git a/tests/helpers/workspace_helpers.py b/tests/helpers/workspace_helpers.py index 8c2bc04f..4b2e78db 100644 --- a/tests/helpers/workspace_helpers.py +++ b/tests/helpers/workspace_helpers.py @@ -2,6 +2,7 @@ from __future__ import annotations +import collections.abc as cabc import os import typing as typ @@ -16,23 +17,23 @@ def install_cargo_stub(cmd_mox: CmdMox, monkeypatch: pytest.MonkeyPatch) -> None """Activate cmd-mox shims for both in-process and subprocess tests.""" from lading.workspace import metadata as metadata_module - class _StubCommand: - """Use cmd-mox expectations without invoking an external process.""" - - def run( - self, - *, - retcode: int | tuple[int, ...] | None = None, - cwd: str | os.PathLike[str] | None = None, - ) -> tuple[int, str, str]: - invocation = Invocation( - command="cargo", - args=["metadata", "--format-version", "1"], - stdin="", - env=dict(os.environ), - ) - response = cmd_mox._handle_invocation(invocation) - return response.exit_code, response.stdout, response.stderr - - monkeypatch.setattr(metadata_module, "_ensure_command", lambda: _StubCommand()) - monkeypatch.setenv(metadata_module.CMD_MOX_STUB_ENV_VAR, "1") + def runner( + command: cabc.Sequence[str], + *, + cwd: object | None = None, + env: cabc.Mapping[str, str] | None = None, + ) -> tuple[int, str, str]: + del cwd + invocation = Invocation( + command=command[0], + args=list(command[1:]), + stdin="", + env=dict(os.environ) | dict(env or {}), + ) + response = cmd_mox._handle_invocation(invocation) + return response.exit_code, response.stdout, response.stderr + + monkeypatch.setattr( + metadata_module, "_active_command_runner", lambda active=None: runner + ) + monkeypatch.setenv("LADING_USE_CMD_MOX_STUB", "1") diff --git a/tests/unit/publish/test_command_logging.py b/tests/unit/publish/test_command_logging.py index be168485..aa78d3de 100644 --- a/tests/unit/publish/test_command_logging.py +++ b/tests/unit/publish/test_command_logging.py @@ -7,7 +7,7 @@ import typing as typ from lading.commands import publish, publish_execution -from lading.workspace import metadata as metadata_module +from lading.testing import cmd_mox_runner if typ.TYPE_CHECKING: from pathlib import Path @@ -74,7 +74,7 @@ def test_invoke_proxies_command_output( assert stderr == "beta" captured = capsys.readouterr() assert captured.out == "alpha" - assert captured.err == "beta" + assert captured.err.endswith("beta") def test_cmd_mox_passthrough_streams_output( @@ -84,7 +84,7 @@ def test_cmd_mox_passthrough_streams_output( use_real_invoke: None, ) -> None: """cmd-mox passthrough should stream via the subprocess runner.""" - monkeypatch.setenv(metadata_module.CMD_MOX_STUB_ENV_VAR, "1") + monkeypatch.setenv("LADING_USE_CMD_MOX_STUB", "1") script = "print('unused')" cmd_mox.spy(sys.executable).with_args("-c", script).passthrough() @@ -109,13 +109,17 @@ def fake_echo(payload: str, sink: typ.TextIO) -> None: echo_payloads.append(payload) monkeypatch.setattr( - publish_execution, - "_invoke_via_subprocess", + cmd_mox_runner, + "invoke_via_subprocess", fake_invoke, ) - monkeypatch.setattr(publish_execution, "_echo_buffered_output", fake_echo) + monkeypatch.setattr(cmd_mox_runner, "_echo_buffered_output", fake_echo) - exit_code, stdout, stderr = publish._invoke((sys.executable, "-c", script)) + exit_code, stdout, stderr = cmd_mox_runner.cmd_mox_runner(( + sys.executable, + "-c", + script, + )) assert exit_code == 0 assert stdout == "alpha" diff --git a/tests/unit/publish/test_preflight_checks.py b/tests/unit/publish/test_preflight_checks.py index 2f6802ab..a84674b5 100644 --- a/tests/unit/publish/test_preflight_checks.py +++ b/tests/unit/publish/test_preflight_checks.py @@ -8,6 +8,7 @@ import pytest from lading.commands import publish +from lading.testing.cmd_mox_runner import normalise_cmd_mox_command from lading.workspace import metadata as metadata_module from .conftest import ORIGINAL_PREFLIGHT, make_config, make_preflight_config @@ -38,9 +39,7 @@ def test_normalise_cmd_mox_command_forwards_non_cargo_commands( """cmd-mox normalisation preserves non-cargo commands and arguments.""" program, args = command[0], tuple(command[1:]) - rewritten_program, rewritten_args = publish._normalise_cmd_mox_command( - program, args - ) + rewritten_program, rewritten_args = normalise_cmd_mox_command(program, args) if program == "cargo" and args: expected_program = f"cargo::{args[0]}" @@ -57,29 +56,10 @@ def test_metadata_coerce_text_decodes_bytes() -> None: """Binary output is decoded using UTF-8 with replacement semantics.""" alpha = "\N{GREEK SMALL LETTER ALPHA}" encoded = alpha.encode() - assert metadata_module._coerce_text(encoded) == alpha + assert metadata_module.coerce_text(encoded) == alpha binary = b"foo\xff" - assert metadata_module._coerce_text(binary) == "foo\ufffd" - - -@pytest.mark.parametrize("value", ["1", "true", "TRUE", "Yes", "on"]) -def test_should_use_cmd_mox_stub_honours_truthy_values( - value: str, monkeypatch: pytest.MonkeyPatch -) -> None: - """Environment values recognised as truthy enable cmd-mox stubbing.""" - monkeypatch.setenv(publish.metadata_module.CMD_MOX_STUB_ENV_VAR, value) - - assert publish._should_use_cmd_mox_stub() is True - - -def test_should_use_cmd_mox_stub_returns_false_by_default( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Missing environment values disable cmd-mox stubbing.""" - monkeypatch.delenv(publish.metadata_module.CMD_MOX_STUB_ENV_VAR, raising=False) - - assert publish._should_use_cmd_mox_stub() is False + assert metadata_module.coerce_text(binary) == "foo\ufffd" def test_run_cargo_preflight_raises_on_failure( diff --git a/tests/unit/publish/test_publish_execution_helpers.py b/tests/unit/publish/test_publish_execution_helpers.py index 75fa0504..daa02d80 100644 --- a/tests/unit/publish/test_publish_execution_helpers.py +++ b/tests/unit/publish/test_publish_execution_helpers.py @@ -12,12 +12,14 @@ from lading.commands import publish_execution from lading.commands.publish import PublishPreflightError +from lading.testing import cmd_mox_runner class _MockCmdMoxEnv: """Minimal cmd-mox environment module stub.""" CMOX_IPC_SOCKET_ENV = "CMOX_IPC_SOCKET" + CMOX_IPC_TIMEOUT_ENV = "CMOX_IPC_TIMEOUT" CMOX_REAL_COMMAND_ENV_PREFIX = "CMOX_REAL_" @@ -69,7 +71,9 @@ def resolve_command_with_override( @pytest.fixture -def mock_cmd_mox_modules(tmp_path: Path) -> SimpleNamespace: +def mock_cmd_mox_modules( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> SimpleNamespace: """Provide complete cmd-mox module stubs for passthrough handling.""" class _Env: @@ -134,7 +138,7 @@ def test_build_cmd_mox_invocation_env_merges_overrides( monkeypatch.setenv("EXISTING", "keep") workspace_root = tmp_path / "workspace" value_path = workspace_root / "value" - env = publish_execution._build_cmd_mox_invocation_env( + env = cmd_mox_runner._build_cmd_mox_invocation_env( workspace_root, {"NEW": value_path}, ) @@ -149,7 +153,7 @@ def test_build_cmd_mox_invocation_env_prefers_cwd_over_pwd_override( ) -> None: """Explicit cwd should win even when env overrides include PWD.""" workspace_root = tmp_path / "workspace" - env = publish_execution._build_cmd_mox_invocation_env( + env = cmd_mox_runner._build_cmd_mox_invocation_env( workspace_root, {"PWD": "/root/repo", "OTHER": "ok"}, ) @@ -172,7 +176,7 @@ def __init__(self) -> None: self.stderr = "err\n" self.exit_code = 3 - exit_code, stdout, stderr = publish_execution._process_cmd_mox_response( + exit_code, stdout, stderr = cmd_mox_runner._process_cmd_mox_response( _Response(), streamed=False, ) @@ -186,14 +190,21 @@ def __init__(self) -> None: assert stderr == "err\n" +def test_process_cmd_mox_response_rejects_missing_exit_code() -> None: + """Malformed cmd-mox responses should not be treated as success.""" + response = SimpleNamespace(stdout="out", stderr="err") + + with pytest.raises(ValueError, match="exit code"): + cmd_mox_runner._process_cmd_mox_response(response, streamed=False) + + def test_handle_cmd_mox_passthrough_returns_unmodified_response() -> None: """When no passthrough directive is present, the response should be returned.""" response = SimpleNamespace() - modules = publish_execution.CmdMoxModules(ipc=None, env=None, command_runner=None) invocation = SimpleNamespace(env={}, command="", args=(), stdin="") - returned, streamed = publish_execution._handle_cmd_mox_passthrough( - response, invocation, timeout=1.0, modules=modules + returned, streamed = cmd_mox_runner._handle_cmd_mox_passthrough( + response, invocation, timeout=1.0 ) assert returned is response @@ -208,6 +219,11 @@ def test_handle_cmd_mox_passthrough_reports_response( """Passthrough directives resolved to responses should be reported back.""" socket_path = str(tmp_path / "cmox" / "shim" / "socket") monkeypatch.setenv("CMOX_IPC_SOCKET", socket_path) + monkeypatch.setattr(cmd_mox_runner, "env_mod", mock_cmd_mox_modules.env_module) + monkeypatch.setattr(cmd_mox_runner, "ipc", mock_cmd_mox_modules.ipc_module()) + monkeypatch.setattr( + cmd_mox_runner, "command_runner", mock_cmd_mox_modules.command_runner() + ) directive = SimpleNamespace( invocation_id="123", @@ -220,18 +236,12 @@ def test_handle_cmd_mox_passthrough_reports_response( args=("test",), stdin="", ) - modules = publish_execution.CmdMoxModules( - ipc=mock_cmd_mox_modules.ipc_module(), - env=mock_cmd_mox_modules.env_module, - command_runner=mock_cmd_mox_modules.command_runner(), - ) response = SimpleNamespace(passthrough=directive) - returned, streamed = publish_execution._handle_cmd_mox_passthrough( + returned, streamed = cmd_mox_runner._handle_cmd_mox_passthrough( response, invocation, timeout=1.0, - modules=modules, ) assert streamed is False @@ -247,6 +257,9 @@ def test_handle_cmd_mox_passthrough_uses_pwd_for_cwd( shim_socket = tmp_path / "cmox" / "shim" / "socket" shim_socket.parent.mkdir(parents=True, exist_ok=True) monkeypatch.setenv("CMOX_IPC_SOCKET", str(shim_socket)) + monkeypatch.setattr(cmd_mox_runner, "env_mod", _MockCmdMoxEnv()) + monkeypatch.setattr(cmd_mox_runner, "ipc", _MockCmdMoxIPC()) + monkeypatch.setattr(cmd_mox_runner, "command_runner", _MockCommandRunner()) directive = SimpleNamespace( invocation_id="cwd-test", @@ -260,12 +273,6 @@ def test_handle_cmd_mox_passthrough_uses_pwd_for_cwd( args=("status",), stdin="", ) - modules = publish_execution.CmdMoxModules( - ipc=_MockCmdMoxIPC(), - env=_MockCmdMoxEnv(), - command_runner=_MockCommandRunner(), - ) - captured: dict[str, Path | None] = {"cwd": None} def _fake_invoke_via_subprocess( @@ -278,15 +285,14 @@ def _fake_invoke_via_subprocess( return 0, "", "" monkeypatch.setattr( - publish_execution, "_invoke_via_subprocess", _fake_invoke_via_subprocess + cmd_mox_runner, "invoke_via_subprocess", _fake_invoke_via_subprocess ) response = SimpleNamespace(passthrough=directive) - returned, streamed = publish_execution._handle_cmd_mox_passthrough( + returned, streamed = cmd_mox_runner._handle_cmd_mox_passthrough( response, invocation, timeout=1.0, - modules=modules, ) assert streamed is True @@ -296,12 +302,8 @@ def _fake_invoke_via_subprocess( def test_invoke_via_subprocess_surfaces_spawn_errors() -> None: """Failed spawns should raise PublishPreflightError with context.""" - context = publish_execution._SubprocessContext(cwd=None, env=None, stdin_data=None) - with pytest.raises(PublishPreflightError): - publish_execution._invoke_via_subprocess( - "definitely-not-a-command", (), context - ) + publish_execution._invoke(("definitely-not-a-command",)) def test_invoke_via_subprocess_writes_stdin() -> None: @@ -336,13 +338,9 @@ def test_merge_cmd_mox_path_entries_filters_shim( shim_dir = tmp_path / "cmox" / "shim" monkeypatch.setenv("CMOX_IPC_SOCKET", f"{shim_dir}/socket") - class _Env: - CMOX_IPC_SOCKET_ENV = "CMOX_IPC_SOCKET" - - merged = publish_execution._merge_cmd_mox_path_entries( + merged = cmd_mox_runner._merge_cmd_mox_path_entries( f"{shim_dir}{os.pathsep}/usr/bin{os.pathsep}", f"/opt/tools{os.pathsep}/usr/bin", - env_module=_Env, ) assert shim_dir.as_posix() not in merged @@ -380,28 +378,24 @@ def test_apply_cmd_mox_environment_and_echo(monkeypatch: pytest.MonkeyPatch) -> """Environment updates and buffered echoing should be no-ops for empty input.""" monkeypatch.delenv("NEW_CMD_ENV", raising=False) - publish_execution._apply_cmd_mox_environment({"NEW_CMD_ENV": "present"}) - publish_execution._echo_buffered_output("", io.StringIO()) + cmd_mox_runner._apply_cmd_mox_environment({"NEW_CMD_ENV": "present"}) + cmd_mox_runner._echo_buffered_output("", io.StringIO()) assert os.environ["NEW_CMD_ENV"] == "present" def test_cmd_mox_shim_directory_without_socket(monkeypatch: pytest.MonkeyPatch) -> None: """Shim directory helper should return None when socket is unset.""" - - class _Env: - CMOX_IPC_SOCKET_ENV = "CMOX_IPC_SOCKET" - monkeypatch.delenv("CMOX_IPC_SOCKET", raising=False) - assert publish_execution._cmd_mox_shim_directory(_Env) is None + assert cmd_mox_runner._cmd_mox_shim_directory() is None def test_log_subprocess_environment_redacts_sensitive_values( caplog: pytest.LogCaptureFixture, ) -> None: """Environment logging should redact common secret tokens.""" - caplog.set_level("DEBUG", logger="lading.commands.publish_execution") + caplog.set_level("DEBUG", logger="lading.runtime.subprocess_runner") publish_execution._log_subprocess_environment({ "TOKEN": "secret", diff --git a/tests/unit/test_cargo_metadata_loading.py b/tests/unit/test_cargo_metadata_loading.py index 27215133..24396f21 100644 --- a/tests/unit/test_cargo_metadata_loading.py +++ b/tests/unit/test_cargo_metadata_loading.py @@ -2,6 +2,7 @@ from __future__ import annotations +import collections.abc as cabc import json import logging import textwrap @@ -9,6 +10,7 @@ import pytest +from lading.runtime import CommandSpawnError from lading.workspace import ( CargoExecutableNotFoundError, CargoMetadataError, @@ -65,55 +67,45 @@ def test_load_cargo_metadata_handles_stdout_variants( def test_load_cargo_metadata_logs_invocation( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, caplog: pytest.LogCaptureFixture, ) -> None: """Ensure the command invocation is logged with resolved arguments.""" - class _Command: - argv = ( - "cargo", - "metadata", - "--format-version", - "1", - "--locked", - ) - - def run( - self, - *, - retcode: int | tuple[int, ...] | None = None, - cwd: str | Path | None = None, - ) -> tuple[int, str, str]: - return 0, json.dumps(_METADATA_PAYLOAD), "" - - monkeypatch.setattr(metadata_module, "_ensure_command", lambda: _Command()) - monkeypatch.delenv(metadata_module.CMD_MOX_STUB_ENV_VAR, raising=False) + def runner( + command: tuple[str, ...], + *, + cwd: Path | None = None, + env: cabc.Mapping[str, str] | None = None, + ) -> tuple[int, str, str]: + del command, cwd, env + return 0, json.dumps(_METADATA_PAYLOAD), "" + caplog.set_level(logging.INFO, logger="lading.workspace.metadata") - result = load_cargo_metadata(tmp_path) + result = load_cargo_metadata(tmp_path, runner=runner) assert result == _METADATA_PAYLOAD - assert ( - "Running external command: cargo metadata --format-version 1 --locked" - in caplog.text - ) + assert "Running external command: cargo metadata --format-version 1" in caplog.text def test_load_cargo_metadata_missing_executable( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: """Absent ``cargo`` binaries should raise ``CargoExecutableNotFoundError``.""" - def _raise() -> None: - raise CargoExecutableNotFoundError - - monkeypatch.setattr(metadata_module, "_ensure_command", _raise) + def runner( + command: tuple[str, ...], + *, + cwd: Path | None = None, + env: cabc.Mapping[str, str] | None = None, + ) -> tuple[int, str, str]: + del command, cwd, env + program = "cargo" + raise CommandSpawnError(program, FileNotFoundError(program)) with pytest.raises(CargoExecutableNotFoundError): - load_cargo_metadata(tmp_path) + load_cargo_metadata(tmp_path, runner=runner) def test_load_cargo_metadata_error_decodes_byte_streams( @@ -196,21 +188,6 @@ def test_load_cargo_metadata_error_scenarios( assert scenario.expected_message in str(excinfo.value) -def test_ensure_command_raises_on_missing_executable( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Verify ``CommandNotFound`` is surfaced as ``CargoExecutableNotFoundError``.""" - - class _RaisingLocal: - def __getitem__(self, name: str) -> typ.NoReturn: - raise metadata_module.CommandNotFound(name, ["/usr/bin"]) - - monkeypatch.setattr(metadata_module, "local", _RaisingLocal()) - - with pytest.raises(CargoExecutableNotFoundError): - metadata_module._ensure_command() - - def test_load_workspace_invokes_metadata( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -259,7 +236,6 @@ def _fake_load_cargo_metadata( def test_load_cargo_metadata_logs_command( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, caplog: pytest.LogCaptureFixture, ) -> None: @@ -270,19 +246,17 @@ def test_load_cargo_metadata_logs_command( "workspace_members": [], } - class _FakeCommand: - def run( - self, - *, - retcode: int | tuple[int, ...] | None = None, - cwd: object | None = None, - ) -> tuple[int, str, str]: - return 0, json.dumps(payload), "" - - monkeypatch.setattr(metadata_module, "_ensure_command", lambda: _FakeCommand()) + def runner( + command: tuple[str, ...], + *, + cwd: Path | None = None, + env: cabc.Mapping[str, str] | None = None, + ) -> tuple[int, str, str]: + del command, cwd, env + return 0, json.dumps(payload), "" caplog.set_level(logging.INFO, logger="lading.workspace.metadata") - result = load_cargo_metadata(tmp_path) + result = load_cargo_metadata(tmp_path, runner=runner) assert result == payload expected = ( @@ -290,23 +264,3 @@ def run( f"(cwd={tmp_path.resolve()})" ) assert expected in caplog.messages - - -def test_load_cmd_mox_modules_errors_when_missing( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Missing cmd-mox dependencies should raise CargoMetadataError.""" - import builtins - - real_import = builtins.__import__ - - def _fake_import(name: str, *args: object, **kwargs: object) -> object: - """Force cmd_mox imports to fail without affecting other modules.""" - if name.startswith("cmd_mox"): - raise ModuleNotFoundError(name) - return real_import(name, *args, **kwargs) - - monkeypatch.setattr(builtins, "__import__", _fake_import) - - with pytest.raises(metadata_module.CargoMetadataError): - metadata_module._load_cmd_mox_modules() diff --git a/tests/unit/test_cmd_mox_integration.py b/tests/unit/test_cmd_mox_integration.py index 71acfd31..fc4194d7 100644 --- a/tests/unit/test_cmd_mox_integration.py +++ b/tests/unit/test_cmd_mox_integration.py @@ -1,4 +1,4 @@ -"""Tests for cmd-mox integration paths in workspace metadata loading.""" +"""Tests for cmd-mox command runner integration paths.""" from __future__ import annotations @@ -7,70 +7,30 @@ import pytest -from lading.workspace import metadata as metadata_module +from lading.testing import cmd_mox_runner if typ.TYPE_CHECKING: from pathlib import Path -def test_cmd_mox_command_requires_socket(monkeypatch: pytest.MonkeyPatch) -> None: +def test_cmd_mox_runner_requires_socket(monkeypatch: pytest.MonkeyPatch) -> None: """cmd-mox stub should fail fast when the socket is not configured.""" - - class _StubEnv: - CMOX_IPC_SOCKET_ENV = "CMOX_IPC_SOCKET" - CMOX_IPC_TIMEOUT_ENV = "CMOX_IPC_TIMEOUT" - - class _StubIPC: - class Invocation: - def __init__( - self, command: str, args: list[str], stdin: str, env: dict[str, str] - ) -> None: - self.command = command - self.args = args - self.stdin = stdin - self.env = env - - def invoke_server(self, invocation: object, timeout: float) -> object: - message = "invoke_server should not be called without socket" - raise AssertionError(message) - - monkeypatch.setattr( - metadata_module, "_load_cmd_mox_modules", lambda: (_StubIPC(), _StubEnv) - ) monkeypatch.delenv("CMOX_IPC_SOCKET", raising=False) - with pytest.raises( - metadata_module.CargoMetadataError, match="CMOX_IPC_SOCKET is unset" - ): - metadata_module._CmdMoxCommand().run() + with pytest.raises(cmd_mox_runner.CmdMoxError, match="CMOX_IPC_SOCKET is unset"): + cmd_mox_runner.cmd_mox_runner(("cargo", "metadata")) def test_resolve_cmd_mox_timeout_validates_values() -> None: """Timeout parsing should reject non-positive and non-numeric values.""" - assert metadata_module._resolve_cmd_mox_timeout(None) > 0 - assert metadata_module._resolve_cmd_mox_timeout("2.5") == 2.5 + assert cmd_mox_runner._resolve_cmd_mox_timeout(None) > 0 + assert cmd_mox_runner._resolve_cmd_mox_timeout("2.5") == 2.5 for value in ("0", "-1", "abc"): - with pytest.raises(metadata_module.CargoMetadataError): - metadata_module._resolve_cmd_mox_timeout(value) - - -def test_build_cmd_mox_command_returns_proxy() -> None: - """The cmd-mox command factory should return a command proxy instance.""" - result = metadata_module._build_cmd_mox_command() + with pytest.raises(cmd_mox_runner.CmdMoxError): + cmd_mox_runner._resolve_cmd_mox_timeout(value) - assert isinstance(result, metadata_module._CmdMoxCommand) - -def test_ensure_command_prefers_cmd_mox_stub(monkeypatch: pytest.MonkeyPatch) -> None: - """The stub command should be returned when the environment requests it.""" - sentinel = object() - monkeypatch.setenv(metadata_module.CMD_MOX_STUB_ENV_VAR, "1") - monkeypatch.setattr(metadata_module, "_build_cmd_mox_command", lambda: sentinel) - - assert metadata_module._ensure_command() is sentinel - - -def test_cmd_mox_command_executes_via_ipc( +def test_cmd_mox_runner_executes_via_ipc( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: @@ -98,21 +58,24 @@ def __init__(self) -> None: self.timeout: float | None = None def invoke_server(self, invocation: object, timeout: float) -> object: - self.last_invocation = invocation # type: ignore[assignment] + self.last_invocation = typ.cast("_StubIPC.Invocation", invocation) self.timeout = timeout return SimpleNamespace(exit_code=0, stdout="{}", stderr="") ipc = _StubIPC() - monkeypatch.setattr( - metadata_module, "_load_cmd_mox_modules", lambda: (ipc, _StubEnv) - ) + monkeypatch.setattr(cmd_mox_runner, "env_mod", _StubEnv) + monkeypatch.setattr(cmd_mox_runner, "ipc", ipc) - command = metadata_module._CmdMoxCommand() - exit_code, stdout, stderr = command.run(cwd=str(tmp_path / "workspace")) + exit_code, stdout, stderr = cmd_mox_runner.cmd_mox_runner( + ("cargo", "metadata", "--format-version", "1"), + cwd=tmp_path / "workspace", + ) - assert command.argv == ("cargo", "metadata", "--format-version", "1") assert exit_code == 0 assert stdout == "{}" assert stderr == "" assert ipc.last_invocation is not None + assert ipc.last_invocation.command == "cargo" + assert ipc.last_invocation.args == ["metadata", "--format-version", "1"] + assert ipc.last_invocation.env["PWD"] == str(tmp_path / "workspace") assert ipc.timeout == 1.5 diff --git a/tests/unit/test_metadata_helpers.py b/tests/unit/test_metadata_helpers.py index c6f1f1ec..de25020e 100644 --- a/tests/unit/test_metadata_helpers.py +++ b/tests/unit/test_metadata_helpers.py @@ -2,32 +2,19 @@ from __future__ import annotations -import typing as typ - from lading.workspace import metadata as metadata_module -if typ.TYPE_CHECKING: - from pathlib import Path - - -def test_build_invocation_environment_sets_pwd(tmp_path: Path) -> None: - """The invocation environment should include PWD when cwd is provided.""" - working_dir = tmp_path / "work" - env = metadata_module._build_invocation_environment(str(working_dir)) - - assert env["PWD"] == str(working_dir) - def test_coerce_text_handles_bytes() -> None: """Byte streams should be decoded to strings.""" - assert metadata_module._coerce_text(b"bytes") == "bytes" + assert metadata_module.coerce_text(b"bytes") == "bytes" def test_error_convenience_constructors() -> None: """Helper constructors should expose descriptive messages.""" assert "Invalid CMOX_IPC_TIMEOUT value" in str( - metadata_module.CargoMetadataError.invalid_cmd_mox_timeout() + metadata_module.CargoMetadataError.invalid_ipc_timeout() ) assert "must be positive" in str( - metadata_module.CargoMetadataError.non_positive_cmd_mox_timeout() + metadata_module.CargoMetadataError.non_positive_ipc_timeout() ) From 2f6e0e533864af7d6a0281b1c7034a7e43ac7f54 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Fri, 5 Jun 2026 20:37:28 +0000 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=93=9D=20CodeRabbit=20Chat:=20Impleme?= =?UTF-8?q?nt=20requested=20code=20changes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lading/workspace/metadata.py | 40 ++++++++++++++++++++++++------------ 1 file changed, 27 insertions(+), 13 deletions(-) diff --git a/lading/workspace/metadata.py b/lading/workspace/metadata.py index 1af8b3e5..772a2725 100644 --- a/lading/workspace/metadata.py +++ b/lading/workspace/metadata.py @@ -109,15 +109,11 @@ def coerce_text(value: str | bytes) -> str: return value -def load_cargo_metadata( - workspace_root: Path | str | None = None, - *, - runner: CommandRunner | None = None, -) -> cabc.Mapping[str, typ.Any]: - """Execute ``cargo metadata`` and parse the resulting JSON payload.""" - root_path = normalise_workspace_root(workspace_root) - command_runner = _active_command_runner(runner) - log_command_invocation(LOGGER, _CARGO_METADATA_COMMAND, root_path) +def _invoke_cargo_metadata_command( + command_runner: CommandRunner, + root_path: Path | None, +) -> tuple[int, str, str]: + """Invoke ``cargo metadata`` and map spawn errors to domain exceptions.""" try: exit_code, stdout, stderr = command_runner( _CARGO_METADATA_COMMAND, cwd=root_path @@ -126,10 +122,11 @@ def load_cargo_metadata( if exc.program == _CARGO_PROGRAM: raise CargoExecutableNotFoundError from exc raise CargoMetadataError(str(exc)) from exc - stdout_text = coerce_text(stdout) - stderr_text = coerce_text(stderr) - if exit_code != 0: - raise CargoMetadataInvocationError(exit_code, stdout_text, stderr_text) + return exit_code, coerce_text(stdout), coerce_text(stderr) + + +def _parse_cargo_metadata_output(stdout_text: str) -> cabc.Mapping[str, typ.Any]: + """Parse and validate the JSON payload produced by ``cargo metadata``.""" try: payload = json.loads(stdout_text) except json.JSONDecodeError as exc: @@ -137,3 +134,20 @@ def load_cargo_metadata( if not isinstance(payload, dict): raise CargoMetadataParseError.non_object_payload() return payload + + +def load_cargo_metadata( + workspace_root: Path | str | None = None, + *, + runner: CommandRunner | None = None, +) -> cabc.Mapping[str, typ.Any]: + """Execute ``cargo metadata`` and parse the resulting JSON payload.""" + root_path = normalise_workspace_root(workspace_root) + command_runner = _active_command_runner(runner) + log_command_invocation(LOGGER, _CARGO_METADATA_COMMAND, root_path) + exit_code, stdout_text, stderr_text = _invoke_cargo_metadata_command( + command_runner, root_path + ) + if exit_code != 0: + raise CargoMetadataInvocationError(exit_code, stdout_text, stderr_text) + return _parse_cargo_metadata_output(stdout_text)