File size: 2,957 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 | # 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
from __future__ import annotations
import sys
from pathlib import Path
import pytest
from speech_text_data_aligners.core.errors import (
BackendExecutionError,
BackendUnavailableError,
)
from speech_text_data_aligners.runtime.process import Command, SubprocessRunner
def test_runner_captures_stdout_stderr_stdin_environment_and_cwd(
tmp_path: Path,
) -> None:
command = Command(
(
sys.executable,
"-c",
"import os,sys; print(os.getcwd()); print(os.environ['ALIGN_TEST']); "
"print(sys.stdin.read()); print('err', file=sys.stderr)",
),
cwd=tmp_path,
environment=(("ALIGN_TEST", "yes"),),
stdin_text="input",
)
result = SubprocessRunner().run(command)
assert result.returncode == 0
assert str(tmp_path) in result.stdout
assert "yes\ninput" in result.stdout
assert result.stderr == "err\n"
assert result.duration_s >= 0
def test_runner_allows_explicitly_unchecked_nonzero_status() -> None:
result = SubprocessRunner().run(
Command((sys.executable, "-c", "raise SystemExit(7)"), check=False)
)
assert result.returncode == 7
def test_runner_translates_checked_nonzero_and_captures_context() -> None:
with pytest.raises(BackendExecutionError) as caught:
SubprocessRunner().run(
Command(
(
sys.executable,
"-c",
"import sys; print('out'); print('bad', file=sys.stderr); raise SystemExit(3)",
)
)
)
assert dict(caught.value.context)["returncode"] == 3
assert "bad" in str(dict(caught.value.context)["stderr"])
@pytest.mark.parametrize(
"command",
[Command(()), Command((sys.executable,), timeout_s=0.0)],
)
def test_runner_rejects_invalid_command(command: Command) -> None:
with pytest.raises(BackendExecutionError):
SubprocessRunner().run(command)
def test_runner_translates_timeout() -> None:
with pytest.raises(BackendExecutionError) as caught:
SubprocessRunner().run(
Command(
(sys.executable, "-c", "import time; time.sleep(2)"), timeout_s=0.01
)
)
assert caught.value.retryable
def test_runner_translates_missing_and_unstartable_executables(tmp_path: Path) -> None:
with pytest.raises(BackendUnavailableError, match="unavailable"):
SubprocessRunner().run(Command((str(tmp_path / "missing"),)))
directory = tmp_path / "directory"
directory.mkdir()
with pytest.raises(BackendUnavailableError, match="started"):
SubprocessRunner().run(Command((str(directory),)))
|