File size: 5,356 Bytes
51e84d2 fc1dcbc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | # License: Apache-2.0 License
# Created by: Patrick Lumbantobing, VertoX-AI
# Copyright (c) 2026 VertoX-AI. All rights reserved.
#
# This work is licensed under the Apache-2.0 License.
# To view a copy of this license, visit
# https://www.apache.org/licenses/LICENSE-2.0
"""Run native commands through an injectable, timeout-aware, shell-free boundary."""
from __future__ import annotations
import os
import subprocess
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol
from speech_text_data_aligners.core.errors import (
BackendExecutionError,
BackendUnavailableError,
)
@dataclass(frozen=True, slots=True)
class Command:
"""Describe a subprocess invocation without shell interpretation.
Parameters
----------
argv: Non-empty executable and argument tuple.
cwd: Optional working directory.
environment: Environment entries layered over the current process.
timeout_s: Optional positive timeout in seconds.
stdin_text: Optional UTF-8 text supplied on standard input.
check: Raise a typed error when the process exits non-zero.
"""
argv: tuple[str, ...]
cwd: Path | None = None
environment: tuple[tuple[str, str], ...] = ()
timeout_s: float | None = None
stdin_text: str | None = None
check: bool = True
@dataclass(frozen=True, slots=True)
class CommandResult:
"""Capture one completed process's text streams, status, and elapsed time.
Parameters
----------
argv: Exact argument vector invoked.
returncode: Native process exit status.
stdout: Captured decoded standard output.
stderr: Captured decoded standard error.
duration_s: Non-negative monotonic elapsed duration.
"""
argv: tuple[str, ...]
returncode: int
stdout: str
stderr: str
duration_s: float
class ProcessRunner(Protocol):
"""Abstract native process execution for injectable backend testing."""
def run(self, command: Command) -> CommandResult:
"""Execute one command and return captured results.
Parameters
----------
command: Shell-free command description.
Returns
-------
Captured successful or explicitly unchecked process result.
Raises
------
BackendUnavailableError: If the executable cannot be started.
BackendExecutionError: If execution times out or a checked command fails.
"""
...
class SubprocessRunner:
"""Execute commands via :mod:`subprocess` with no shell or inherited stdin."""
def run(self, command: Command) -> CommandResult:
"""Execute one validated command and translate expected OS/process failures.
Parameters
----------
command: Shell-free command description.
Returns
-------
Captured result, including non-zero status when ``check`` is false.
Raises
------
BackendUnavailableError: If the executable is missing or inaccessible.
BackendExecutionError: If the command is invalid, times out, or exits
non-zero while ``check`` is true.
"""
if not command.argv or not command.argv[0]:
raise BackendExecutionError("command argument vector cannot be empty")
if command.timeout_s is not None and command.timeout_s <= 0:
raise BackendExecutionError("command timeout must be positive")
environment = os.environ.copy()
environment.update(command.environment)
started = time.monotonic()
try:
completed = subprocess.run(
command.argv,
cwd=command.cwd,
env=environment,
input=command.stdin_text,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=command.timeout_s,
check=False,
shell=False,
)
except FileNotFoundError as error:
raise BackendUnavailableError(
f"backend executable is unavailable: {command.argv[0]}"
) from error
except OSError as error:
raise BackendUnavailableError(
f"backend executable could not be started: {command.argv[0]}"
) from error
except subprocess.TimeoutExpired as error:
raise BackendExecutionError(
f"backend command timed out after {command.timeout_s} seconds",
retryable=True,
) from error
duration = time.monotonic() - started
result = CommandResult(
argv=command.argv,
returncode=completed.returncode,
stdout=completed.stdout,
stderr=completed.stderr,
duration_s=duration,
)
if command.check and result.returncode != 0:
raise BackendExecutionError(
f"backend command exited with status {result.returncode}",
context=(
("returncode", result.returncode),
("stdout", result.stdout[-4000:]),
("stderr", result.stderr[-4000:]),
),
)
return result
|