v-and-v/testium
2
0
Fork 0
Code Issues 5 Pull Requests 1 Actions Packages Projects Releases 4 Wiki Activity

feat(run): capture sub-instance output, add batch param

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Foué
2026-06-15 22:53:30 +02:00
parent 5b5792a296
commit 87e62a7f2e
2 changed files with 66 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
src/testium/interpreter/test_items/test_item_run.py
View File

@@ -11,6 +11,7 @@ from interpreter.test_items.test_result import (TestValue)
import api.testium as tm
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from interpreter.utils.proc_drain import drain_to_log
from runtime.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
@@ -75,6 +76,9 @@ class TestItemRun(TestItem):
Param("wait_for_exec",
doc="If true, block until the time window opens. Requires both "
"start_time and end_time."),
Param("batch", default=False,
doc="Run the sub-instance headless (-b) with its output captured "
"into this test's log/report and result value, even in the GUI."),
)
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
@@ -90,6 +94,38 @@ class TestItemRun(TestItem):
self.start_time = self._prms.getParam('start_time')
self.end_time = self._prms.getParam('end_time')
self.wait_for_exec = self._prms.getParam('wait_for_exec')
self.batch = self._prms.getParam('batch', default=False)
def _launch(self, cmd, capture):
"""Run the sub-instance once. When *capture*, stream its output to the
log/report, keep it as the result value, and let Stop kill the child."""
if not capture:
subprocess.run(cmd)
return
sink = []
prefix = f"[{os.path.basename(self.tum_file)}] "
proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
threads = drain_to_log(proc, prefix=prefix, sink=sink)
try:
while True:
try:
proc.wait(timeout=0.2)
break
except subprocess.TimeoutExpired:
if self.isStopped():
proc.terminate()
try:
proc.wait(timeout=2)
except subprocess.TimeoutExpired:
proc.kill()
break
finally:
for t in threads:
t.join(timeout=2)
# Captured log -> result value (store_result / expected_result).
self.result.value = "\n".join(sink)
@test_run
def execute(self):
@@ -104,25 +140,26 @@ class TestItemRun(TestItem):
pf = self._prms.expanse(self.param_file)
lp = self._prms.expanse(self.log_path)
rp = self._prms.expanse(self.report_path)
# Capture (headless -b) in batch or when `batch: true`; else open
# the child's own GUI window (-r).
capture = bool(self.batch) or tm.text_mode()
cmd = _testium_launch_cmd()
if tm.text_mode():
cmd.append("-b")
if capture:
cmd += ["-b", "-o"] # -o: no colour codes in the captured log
else:
cmd.append("-r")
if lp == '':
lp = os.path.splitext(self.tum_file)[0] + "_" + \
datetime.utcnow().isoformat(timespec='seconds') + '.log'
cmd.append("-l")
cmd.append('"' + lp + '"')
cmd += ["-l", '"' + lp + '"']
if pf != '':
cmd.append("-c")
cmd.append('"' + pf + '"')
cmd += ["-c", '"' + pf + '"']
if rp != '':
cmd.append("-p")
cmd.append('"' + rp + '"')
cmd += ["-p", '"' + rp + '"']
cmd.append(self.tum_file)
for c in cmd:
print(c, end = ' ')
print(" ".join(cmd))
if self.start_time is not None:
self.start_time = datetime.strptime(
@@ -135,20 +172,24 @@ class TestItemRun(TestItem):
raise ETUMRuntimeError(
'"wait_for_exec" set but not start_time or end_time')
r = None
ran = False
if self.wait_for_exec:
while not nowInBetween(self.start_time, self.end_time):
sleep(60)
r = subprocess.run(cmd)
self._launch(cmd, capture)
ran = True
elif self.start_time is not None and self.end_time is not None:
if nowInBetween(self.start_time, self.end_time):
r = subprocess.run(cmd)
self._launch(cmd, capture)
ran = True
elif self.start_time is not None:
if self.start_time < datetime.now().time():
r = subprocess.run(cmd)
self._launch(cmd, capture)
ran = True
else:
r = subprocess.run(cmd)
if isinstance(r, subprocess.CompletedProcess):
self._launch(cmd, capture)
ran = True
if ran:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE, 'Sub-test did not execute')

20
src/testium/interpreter/utils/proc_drain.py
View File

@@ -13,7 +13,7 @@ from time import monotonic
from runtime.jrpc import RPC_PORT_SENTINEL
def _drain_pipe(pipe, prefix):
def _drain_pipe(pipe, prefix, sink=None):
try:
for raw in iter(pipe.readline, b""):
line = raw.decode("utf-8", errors="replace").rstrip("\r\n")
@@ -23,6 +23,9 @@ def _drain_pipe(pipe, prefix):
print(f"{prefix}{line}")
else:
print(line)
# sink keeps the clean (unprefixed) line for reuse as a result value
if sink is not None:
sink.append(line)
finally:
try:
pipe.close()
@@ -30,21 +33,16 @@ def _drain_pipe(pipe, prefix):
pass
def drain_to_log(process, prefix=""):
"""Spawn daemon threads that read ``process.stdout`` and
``process.stderr`` line by line and print each line through the
parent's stdout (so it reaches the log + live output).
Each thread exits cleanly when the subprocess closes the
corresponding pipe (i.e. when it exits). Daemon flag ensures they
do not block testium exit.
"""
def drain_to_log(process, prefix="", sink=None):
"""Stream the subprocess stdout/stderr line by line through the parent's
print pipeline (log + live output). If ``sink`` is a list, each clean line
is also appended to it (GIL-atomic, shared by both threads). Daemon threads."""
threads = []
for pipe in (process.stdout, process.stderr):
if pipe is None:
continue
t = threading.Thread(
target=_drain_pipe, args=(pipe, prefix), daemon=True,
target=_drain_pipe, args=(pipe, prefix, sink), daemon=True,
)
t.start()
threads.append(t)