tests/unit/runtime/test_process.py

# 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),)))