mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
4ee2b259c9
- Add build-context helper to copy Docker context safely (ignore runtime data, keep symlinks) - Update UI and core actions to use ~/.prole/build and shared copy helper - Add/adjust tests and scripts; introduce knoe ops helpers and update manifests Co-authored-by: Junie <junie@jetbrains.com>
95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
import sys
|
|
import time
|
|
|
|
from knoe.core.stream_exec import run_streaming_cmd
|
|
|
|
|
|
def test_run_streaming_cmd_streams_stdout_incrementally():
|
|
chunks: list[str] = []
|
|
first_chunk_at: list[float] = []
|
|
|
|
def _on_stdout(text: str):
|
|
chunks.append(text)
|
|
if not first_chunk_at:
|
|
first_chunk_at.append(time.time())
|
|
|
|
started_at = time.time()
|
|
rc = run_streaming_cmd(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"import sys,time;"
|
|
"sys.stdout.write('first\\n');sys.stdout.flush();"
|
|
"time.sleep(0.35);"
|
|
"sys.stdout.write('second\\n');sys.stdout.flush()",
|
|
],
|
|
on_stdout=_on_stdout,
|
|
)
|
|
finished_at = time.time()
|
|
|
|
assert rc == 0
|
|
assert first_chunk_at
|
|
# The first callback should fire while the process is still running.
|
|
assert first_chunk_at[0] < finished_at - 0.15
|
|
output = "".join(chunks)
|
|
assert "first" in output
|
|
assert "second" in output
|
|
assert started_at < first_chunk_at[0]
|
|
|
|
|
|
def test_run_streaming_cmd_nonzero_exit():
|
|
rc = run_streaming_cmd([sys.executable, "-c", "import sys; sys.exit(42)"])
|
|
assert rc == 42
|
|
|
|
|
|
def test_run_streaming_cmd_string_form():
|
|
"""A plain string command is wrapped in bash -lc."""
|
|
chunks: list[str] = []
|
|
rc = run_streaming_cmd(
|
|
f"{sys.executable} -c \"print('strtest')\"",
|
|
on_stdout=lambda t: chunks.append(t),
|
|
)
|
|
assert rc == 0
|
|
assert "strtest" in "".join(chunks)
|
|
|
|
|
|
def test_run_streaming_cmd_stdin_text():
|
|
"""stdin_text is written to the subprocess stdin."""
|
|
chunks: list[str] = []
|
|
rc = run_streaming_cmd(
|
|
[sys.executable, "-c", "import sys; print(sys.stdin.read().strip())"],
|
|
stdin_text="hello_from_stdin",
|
|
on_stdout=lambda t: chunks.append(t),
|
|
)
|
|
assert rc == 0
|
|
assert "hello_from_stdin" in "".join(chunks)
|
|
|
|
|
|
def test_run_streaming_cmd_no_callbacks():
|
|
"""Running without callbacks should not crash."""
|
|
rc = run_streaming_cmd([sys.executable, "-c", "print('silent')"])
|
|
assert rc == 0
|
|
|
|
|
|
def test_run_streaming_cmd_keeps_stderr_and_ansi_sequences():
|
|
stdout_chunks: list[str] = []
|
|
stderr_chunks: list[str] = []
|
|
|
|
rc = run_streaming_cmd(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"import sys;"
|
|
"sys.stdout.write('OUT\\n');sys.stdout.flush();"
|
|
"sys.stderr.write('\\x1b[31mERR\\x1b[0m\\n');sys.stderr.flush()",
|
|
],
|
|
on_stdout=lambda text: stdout_chunks.append(text),
|
|
on_stderr=lambda text: stderr_chunks.append(text),
|
|
)
|
|
|
|
assert rc == 0
|
|
assert "OUT" in "".join(stdout_chunks)
|
|
stderr_out = "".join(stderr_chunks)
|
|
assert "ERR" in stderr_out
|
|
assert "\x1b[31m" in stderr_out
|