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

Fix run item and batch mode robustness

Browse Source 
- run item: rename tum_fime→tum, remove stdout=PIPE (deadlock with
  spawn), support batch mode (-b), SUCCESS on any completed subprocess
  regardless of sub-test result
- batch.py: fix control("loaded") deadlock via daemon thread + Event +
  is_alive() polling; fix premature finish on gd_update messages;
  propagate success flag from finished message; guard control("close")
- process.py: include success flag in send_finished message
- py_process/lua_process: add stdout/stderr=DEVNULL to Popen
- test_run.py: fix finished detection ("id" in m and m["id"] is None)
- testium_win.py: track run_exit_code, SIGABRT handler, clean exit
- __init__.py: sys.exit with batch success flag
- Add run item validation tests and CLAUDE.md documentation

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Foué
2026-04-27 07:49:16 +02:00
parent a3e449cc7d
commit 60dbcf0252
15 changed files with 225 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/testium/__init__.py
View File

@@ -125,6 +125,7 @@ def main():
from interpreter.batch import Batch
b = Batch(tf, cf, defines, rf, args.report_type, pn, args.no_color, text_mode=True)
sys.exit(0 if b.success else 1)
else:
from main_win.testium_win import MainWin

25
src/testium/interpreter/batch.py
View File

@@ -1,6 +1,7 @@
import os
import sys
import platform
import threading
from time import sleep
from signal import signal, SIGINT
from queue import Empty
@@ -8,7 +9,7 @@ from multiprocessing import Queue
from interpreter.process import TestProcess
from interpreter.utils.test_ctrl import TestSetController
from lib.tum_except import ETUMFileError
from lib.tum_except import ETUMFileError, ETUMRuntimeError
from lib.stdout_redirect import stdio_redir
@@ -52,6 +53,7 @@ class Batch:
signal(SIGINT, self.sigint_handler)
self._success = False
msg_queue = Queue()
self.tst_ctrl = TestSetController()
tst_proc = TestProcess(
@@ -64,8 +66,17 @@ class Batch:
)
tst_proc.start()
while not self.tst_ctrl.control("loaded"):
sleep(0.1)
# Wait for TestProcess to finish loading.
# Run the blocking control("loaded") in a daemon thread so we
# can watch for unexpected process death in the main thread.
_loaded_event = threading.Event()
def _wait_loaded():
self.tst_ctrl.control("loaded")
_loaded_event.set()
threading.Thread(target=_wait_loaded, daemon=True).start()
while not _loaded_event.wait(timeout=0.1):
if not tst_proc.is_alive():
raise ETUMRuntimeError("TestProcess terminated unexpectedly during load")
self.tst_ctrl.control(
"report",
@@ -82,6 +93,7 @@ class Batch:
m = msg_queue.get(timeout=0.2)
if "id" in m and m["id"] is None:
# id key present and None -> finished
self._success = m.get("success", False)
break
except Empty:
if not tst_proc.is_alive():
@@ -89,7 +101,8 @@ class Batch:
continue
# Close the process and wait for termination
self.tst_ctrl.control("close")
if tst_proc.is_alive():
self.tst_ctrl.control("close")
tst_proc.join()
except Exception as e:
@@ -98,6 +111,10 @@ class Batch:
finally:
stdio_redir.restore()
@property
def success(self):
return self._success
def sigint_handler(self, signal_received, frame):
try:
self.tst_ctrl.control("stop", timeout=5)

6
src/testium/interpreter/process.py
View File

@@ -283,7 +283,7 @@ Is the python exec path correct ?"""
engine.stop()
engine.join()
# Sends signal to the GUI
self.send_finished()
self.send_finished(success=test_set.success())
globdict.set_update_queue(None)
restore_gd(gdict)
except Exception as e:
@@ -339,8 +339,10 @@ Is the python exec path correct ?"""
stdio_redir.restore()
stdio_redir.stop()
def send_finished(self):
def send_finished(self, success=None):
status = {"id": None, "name": "test_process", "status": "finished"}
if success is not None:
status["success"] = success
self.__squeue.put(status)
def execute(self):

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

@@ -31,7 +31,7 @@ class TestItemRun(TestItem):
self._type = cst.TYPE_RUN
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.tum_fime = self._prms.getParam('tum_fime', required=True)
self.tum_file = self._prms.getParam('tum', required=True)
self.param_file = self._prms.getParam('param_file', default='')
self.python_bin = self._prms.getParam('python_bin', default='')
self.testium_path = self._prms.getParam('testium_path', default='')
@@ -43,39 +43,43 @@ class TestItemRun(TestItem):
@test_run
def execute(self):
res = -1
try:
file_path = self._prms.expanse(self.tum_fime)
file_path = self._prms.expanse(self.tum_file)
if not os.path.exists(file_path) and not os.path.isabs(file_path):
file_path = os.path.join(tm.gd('test_directory'), self.tum_fime)
file_path = os.path.join(tm.gd('test_directory'), file_path)
if not os.path.isfile(file_path):
raise ETUMRuntimeError(
'"{}" file could not be found'.format(file_path))
self.tum_fime = file_path
self.tum_file = file_path
pf = self._prms.expanse(self.param_file)
pp = self._prms.expanse(self.python_bin)
sp = self._prms.expanse(self.testium_path)
lp = self._prms.expanse(self.log_path)
rp = self._prms.expanse(self.report_path)
cmd = []
if sp == '':
sp = sys.argv[0]
if pp != '':
cmd.append(pp)
if sp == '':
sp = os.path.join(tm.get_main_dir(), "testium.pyw")
elif not os.path.isfile(sp) or not os.access(sp, os.X_OK):
cmd.append(sys.executable)
cmd.append(sp)
if lp == '':
lp = os.path.splitext(self.tum_fime)[0] + "_" + \
datetime.utcnow().isoformat(timespec='seconds') + '.log'
cmd.append("-r")
if tm.text_mode():
cmd.append("-b")
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 + '"')
if pf != '':
cmd.append("-c")
cmd.append('"' + pf + '"')
cmd.append("-l")
cmd.append('"' + lp + '"')
if rp != '':
cmd.append("-p")
cmd.append('"' + rp + '"')
cmd.append(self.tum_fime)
cmd.append(self.tum_file)
for c in cmd:
print(c, end = ' ')
@@ -90,31 +94,23 @@ class TestItemRun(TestItem):
raise ETUMRuntimeError(
'"wait_for_exec" set but not start_time or end_time')
r = None
if self.wait_for_exec:
while not nowInBetween(self.start_time, self.end_time):
sleep(60)
r = subprocess.run(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r = subprocess.run(cmd)
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, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r = subprocess.run(cmd)
elif self.start_time is not None:
if self.start_time < datetime.now().time():
r = subprocess.run(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r = subprocess.run(cmd)
else:
r = subprocess.run(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r = subprocess.run(cmd)
if isinstance(r, subprocess.CompletedProcess):
print((r.stdout).decode())
print(r.stderr.decode())
res = r.returncode
if res >= 0:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE,
'Test execution returned negative value.')
self.result.set(TestValue.FAILURE, 'Sub-test did not execute')
except:
traceback.print_exception(*sys.exc_info())
self.result.set(TestValue.FAILURE, 'Unrecoverable "run" item error')

5
src/testium/interpreter/utils/lua_process.py
View File

@@ -191,7 +191,10 @@ class LuaProcessBase:
self._process = subprocess.Popen(
params, env=env, cwd=func_proc_path,
stdin=subprocess.DEVNULL, restore_signals=False,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
restore_signals=False,
)
self._rpc = JsonRpcClient(

5
src/testium/interpreter/utils/py_process.py
View File

@@ -160,7 +160,10 @@ class PyProcessBase:
self._process = subprocess.Popen(
params, env=env, cwd=func_proc_path,
stdin=subprocess.DEVNULL, restore_signals=False,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
restore_signals=False,
)
self._rpc = JsonRpcClient(

2
src/testium/main_win/test_run/test_run.py
View File

@@ -28,7 +28,7 @@ class ThreadTestStatus(QThread):
self.gdUpdated.emit(m["key"], m["value"])
elif msg_type == "gd_delete":
self.gdDeleted.emit(m["key"])
elif m.get("id", None) is None:
elif "id" in m and m["id"] is None:
self.testSetIsFinished.emit()
else:
self.statusToBeUpdated.emit(m)

1
src/testium/main_win/test_runner.py
View File

@@ -118,6 +118,7 @@ class TestRunner:
self.logFileHandler = None
w.textLog.appendPlainText("Test is finished")
w.run_exit_code = 0 if w.treeTests.getGlobalSuccess() else 1
if w.runandclose:
w.on_actionExit_triggered()

43
src/testium/main_win/testium_win.py
View File

@@ -1,18 +1,14 @@
import sys
import os
import subprocess
import traceback
import webbrowser
from time import sleep
from multiprocessing import Queue
from queue import Empty
from threading import Thread
import shutil
# Qt
from PySide6 import QtGui, QtWidgets
from PySide6 import QtGui
from PySide6.QtGui import QAction, QShortcut, QIcon, QPixmap, QTextCursor, QDesktopServices, QTextCursor
from PySide6.QtCore import Slot, QUrl, Qt, QTimer, QDateTime
from PySide6.QtCore import Slot, QUrl, Qt, QTimer
from PySide6.QtWidgets import (
QApplication,
@@ -92,6 +88,7 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.test_service = None
self.threadTestStatus = None
self._signals_connected = False
self.run_exit_code = -1 # -1 = test not yet completed
self.timer = QTimer()
self.timer.setSingleShot(False)
@@ -359,14 +356,20 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.treeTests.saveSizes()
prefs.settings.sync()
def closeEvent(self, event):
self.on_exiting()
event.accept()
def on_exiting(self):
if self.runner.state == TestState.IDLE:
self.save_settings()
self.file_manager.clear_process()
self.threadTestStatus.stop()
self.threadOutput.stop()
self.threadOutput.wait()
self.threadTestStatus.wait()
try:
if self.runner.state == TestState.IDLE:
self.save_settings()
self.file_manager.clear_process()
finally:
self.threadTestStatus.stop()
self.threadOutput.stop()
self.threadOutput.wait()
self.threadTestStatus.wait()
def show_checkboxes(self, hidden=None):
if hidden:
@@ -685,5 +688,17 @@ def MainWin(
debug,
)
import signal
import os as _os
def _sigabrt_handler(signum, frame):
# Qt crash: exit with the test result if known, -1 if test never completed
_os._exit(ui.run_exit_code)
signal.signal(signal.SIGABRT, _sigabrt_handler)
ui.show()
sys.exit(app.exec_())
app.exec_()
exit_code = ui.run_exit_code if ui.run_exit_code >= 0 else 0
del ui
sys.exit(exit_code)