67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
import inspect
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from l4d2host.process import CommandCancelledError, run_command
|
|
|
|
|
|
def test_callbacks_receive_lines() -> None:
|
|
out: list[str] = []
|
|
err: list[str] = []
|
|
run_command(
|
|
["python3", "-c", "import sys; print('ok'); print('warn', file=sys.stderr)"],
|
|
on_stdout=out.append,
|
|
on_stderr=err.append,
|
|
)
|
|
assert out == ["ok"]
|
|
assert err == ["warn"]
|
|
|
|
|
|
def test_nonzero_exit_raises() -> None:
|
|
with pytest.raises(subprocess.CalledProcessError):
|
|
run_command(["python3", "-c", "import sys; sys.exit(7)"])
|
|
|
|
|
|
def test_cancelled_command_raises_cancelled_error() -> None:
|
|
should_cancel = False
|
|
lines: list[str] = []
|
|
|
|
def on_stdout(line: str) -> None:
|
|
nonlocal should_cancel
|
|
lines.append(line)
|
|
should_cancel = True
|
|
|
|
with pytest.raises(CommandCancelledError):
|
|
run_command(
|
|
["python3", "-c", "import time; print('ready', flush=True); time.sleep(30)"],
|
|
on_stdout=on_stdout,
|
|
should_cancel=lambda: should_cancel,
|
|
cancel_poll_seconds=0.01,
|
|
cancel_terminate_timeout=0.2,
|
|
)
|
|
|
|
assert lines == ["ready"]
|
|
|
|
|
|
def test_run_command_avoids_runtime_unsafe_nested_annotations() -> None:
|
|
source = inspect.getsource(run_command)
|
|
assert "subprocess.Popen[str].stdout" not in source
|
|
|
|
|
|
def test_run_command_passthrough_writes_to_sys_streams(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import sys
|
|
from io import StringIO
|
|
|
|
mock_stdout = StringIO()
|
|
mock_stderr = StringIO()
|
|
monkeypatch.setattr(sys, "stdout", mock_stdout)
|
|
monkeypatch.setattr(sys, "stderr", mock_stderr)
|
|
|
|
run_command(
|
|
["python3", "-c", "import sys; print('passed out'); print('passed err', file=sys.stderr)"],
|
|
passthrough=True,
|
|
)
|
|
|
|
assert mock_stdout.getvalue() == "passed out\n"
|
|
assert mock_stderr.getvalue() == "passed err\n"
|