11 Commits

Author SHA1 Message Date
06c4cc62c6 doc: update run item documentation
Clarify result semantics (SUCCESS on launch, not on sub-test result),
batch vs GUI mode behaviour, and clean up attribute descriptions.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-27 07:53:26 +02:00
60dbcf0252 Fix run item and batch mode robustness
- 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>
2026-04-27 07:49:16 +02:00
a3e449cc7d in batch mode, the dialog allways return FAIL, except if auto_result is defined (validation only).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-26 11:52:52 +02:00
95107117fa Color is automatically adapted to the theme of the console.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-26 11:50:28 +02:00
88cc410eed fix of blocking of the text output in batch mode.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-26 11:49:52 +02:00
fa7f8cef7c Text mode upgrade (to be fixed). 2026-04-26 09:20:39 +02:00
5a065128be Fix lua delgd test: use sentinel default instead of nil comparison
cjson decodes JSON null as cjson.null, not Lua nil, so == nil always fails.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:25:30 +02:00
b7b930aab1 Add validation tests for OS, get_main_dir, timestamp, timestamp_as_sec helper functions
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:21:56 +02:00
609ca57202 Add delgd validation test for py_func and lua_func
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:18:31 +02:00
d26b60435b Add auto_result param to dialog items for automated validation
Each dialog test item now accepts an optional auto_result parameter
(ok/cancel/yes/no) and auto_value for text dialogs. When set, the dialog
window opens, stays visible 2 seconds, then closes automatically with the
specified result — allowing the validation suite to run without manual
interaction.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:07:07 +02:00
de143b6cc3 Fix EOFError crash when dialog subprocess exits without sending a result
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 10:06:55 +02:00
43 changed files with 891 additions and 226 deletions

94
CLAUDE.md Normal file
View File

@@ -0,0 +1,94 @@
# Testium — Claude Context
## What is testium
Testium is a test sequencer/runner written in Python. It executes YAML-based test scripts ("`.tum`" files) and supports three execution modes:
- **GUI mode** (default, no flag): PySide6 Qt application (`src/testium/main_win/`)
- **Batch mode** (`-b` / `--batch-execution`): headless, non-interactive, runs tests and exits
- **Terminal mode** (`-m` / `--terminal`): interactive REPL in the terminal (partially implemented / work-in-progress)
Run from repo root: `./run.sh` (Linux) or `run.bat` / `run.ps1` (Windows).
Direct invocation: `python3 -m src/testium [-b|-m] <test_file.tum>`
## Architecture
### Entry point
`src/testium/__init__.py` — parses CLI args, dispatches to the three modes.
`multiprocessing.set_start_method('spawn')` is called early (required for Linux dialog subprocesses).
### Core execution
- `src/testium/interpreter/process.py``TestProcess(multiprocessing.Process)`: runs the test in a child process. Stdout is redirected via a `StringQueue` → pipe → parent thread (`capture_stdout`) that writes to real stdout.
- `src/testium/interpreter/batch.py``Batch`: parent-side orchestrator for `-b` mode. Creates the `msg_queue`, starts `TestProcess`, waits for the "finished" signal.
- `src/testium/interpreter/test_set.py``TestSet`: builds and executes the tree of test items.
- `src/testium/interpreter/test_items/test_item*.py` — one file per test item type (check, cycle, group, let, unittest, py_func, lua_func, console, git, dialogs, report, …).
### Communication channels (parent ↔ child process)
- `msg_queue` (`multiprocessing.Queue`): carries status messages from child to parent.
- Item status: `{"id": <non-None>, "name": ..., "status": "started"|"finished", ...}`
- Global dict updates: `{"type": "gd_update"|"gd_delete", "key": ..., "value": ...}`**no "id" key**
- Process finished: `{"id": None, "name": "test_process", "status": "finished"}` — id key present but `None`
- `tst_ctrl` (`TestSetController`): sends control commands (execute, stop, pause, close, …) from parent to child.
- stdout pipe (`multiprocessing.Pipe`): streams test output from child back to parent's `capture_stdout` thread.
### Stdout pipeline (batch mode)
```
test item print()
→ sys.stdout (StringQueue, in child)
→ send_stdout thread (child) → pipe → capture_stdout thread (parent)
→ print() → sys.stdout (TermLog wrapping real stdout, in parent)
→ terminal
```
### Global dictionary
`src/testium/interpreter/utils/globdict.py` — shared state accessible from test scripts via `tm.gd()` / `tm.setgd()`. When `set_update_queue()` is active (during test execution), every `setgd`/`delgd` on a non-`_`-prefixed key pushes a message to `msg_queue`.
### Coloring (`-o` disables it)
`src/testium/interpreter/utils/termlog.py``TermLog` wraps stdout with colorama-based line coloring (PASS=green, FAIL=red, WARN=yellow, …). Applied in parent process for batch/terminal modes. Auto-detects light/dark terminal background via (in order): `COLORFGBG` env var, OSC 11 query, default dark.
### Dialog items in batch mode
All dialog items (`dialog_image`, `dialog_question`, `dialog_references`, `dialog_value`, `dialog_message`, `dialog_choices`, `dialog_note`) follow this rule in non-interactive text mode (`-b`):
- `auto_result` defined in the `.tum` → result controlled by it (`ok`/`yes` → SUCCESS, `cancel`/`no` → FAIL)
- `auto_result` absent → FAIL with `"Dialog not supported in batch mode"`
- `sleep dialog: true` → exception: just sleeps normally, no GUI, no failure
`auto_result` (and `auto_value` for value/note dialogs) is intended for the validation test suite (`test/validation/`) only.
## Key files
| Path | Role |
|------|------|
| `src/testium/__init__.py` | CLI entry, mode dispatch |
| `src/testium/interpreter/batch.py` | `-b` mode orchestrator |
| `src/testium/interpreter/process.py` | Child test process |
| `src/testium/interpreter/terminal.py` | `-m` mode (WIP) |
| `src/testium/interpreter/test_set.py` | Test tree builder/executor |
| `src/testium/interpreter/utils/globdict.py` | Global variable dict |
| `src/testium/interpreter/utils/termlog.py` | Terminal color output |
| `src/lib/stdout_redirect.py` | `StdioRedirect` singleton (`stdio_redir`) |
| `src/lib/string_queue.py` | Thread-safe string buffer used for stdout redirection |
| `src/testium/libs/testium.py` | Public API for test scripts (`tm.*`) |
## Known issues / WIP
- `-m` (terminal mode) is not functional yet.
- `text_no_pyside` branch: work in progress on text mode improvements.
### `run` item
`src/testium/interpreter/test_items/test_item_run.py` — launches a `.tum` file in a new testium instance (`-b` in batch mode, `-r` in GUI mode). Result:
- **SUCCESS** if the sub-instance launched and ran to completion (exit code is ignored)
- **FAILURE** if the file is not found, `wait_for_exec` is set without `start_time`/`end_time`, the time window was not reached, or any other launch error
The sub-test's own pass/fail result is intentionally not propagated.
## Recent fixes (branch `text_no_pyside`)
- `batch.py`: premature loop exit when `gd_update` messages (no `"id"` key) were mistaken for the "finished" signal — fix: `"id" in m and m["id"] is None`
- `batch.py`: `control("loaded")` deadlock if `TestProcess` crashed before `cmd_th` started — fix: daemon thread + `threading.Event` + `is_alive()` polling
- `termlog.py`: `COLOR_DEFAULT = Fore.WHITE` invisible on light terminals; added auto-detection + light palette. Also fixed `write()` residue accumulation bug (`s[pos:]``s[pos+1:]`).
- Dialog items: `auto_result`/`auto_value` now used in non-interactive text mode; dialogs without `auto_result` FAIL immediately in batch mode.
- `run` item: removed `stdout=PIPE` (caused deadlock with `multiprocessing` spawn); simplified result to SUCCESS on any completed subprocess.
## Validation tests
Located in `test/validation/`. Run with `-b` flag against each `.tum` file.
## Dependencies
See `src/requirements.txt`. Key ones: `pyside6`, `pyyaml`, `jinja2`, `colorama`, `gitpython`, `pexpect`, `matplotlib`.

View File

@@ -1,16 +1,23 @@
**run** test item
============================================================
This test item executes a new instance of testium.
This test item executes a new instance of testium with the specified ``.tum`` file.
* In **batch mode** (``-b``): the sub-instance is started with ``-b``.
* In **GUI mode**: the sub-instance is started with ``-r`` (run and close).
The item result is **SUCCESS** if the sub-instance launched and ran to completion,
regardless of whether the sub-tests passed or failed.
It is **FAILURE** if the file could not be found, the sub-instance could not be
launched, or the time window was not reached (see ``start_time`` / ``end_time``).
.. code-block:: yaml
:caption: ``run`` test item usage example
- run:
name: Execute TUM
tum_fime: example_cycle.tum
tum: example_cycle.tum
python_bin: python3
testium_path: /home/francois/projets/testium-new-report/testium.pyw
log_file: $(home)/reports/test.log
report_file: $(home)/reports/test.rep
@@ -19,12 +26,12 @@ Attributes
run test item has the following specific attributes:
* ``tum_fime``: mandatory the path of the file to execute, it can be relative to current execution folder,
* ``param_file`` (optional) the path of the parameter file to use, otherwise default parameter file is used.
* ``python_bin`` (optional) the path of a specific python to run your scripts,
* ``testium_path`` (optional) the path of a specific testium to run your scripts,
* ``log_file`` (optional) the path of log file to register, if not provided a file is created with timestamp at the location of TUM file.
* ``report_file`` (optional), the path of report file to create
* ``start_time`` (optional), start time for the script execution, in HH:MM format.
* ``end_time`` (optional), end time for an execution within a time frame, in HH:MM format.
* ``wait_for_exec`` (optional). True or False, wait to be in the execution window defined by start_time and end_time to run the script.
* ``tum``: mandatory, the path of the file to execute. Can be relative to the current execution folder.
* ``param_file`` (optional): the path of the parameter file to use; otherwise the default parameter file is used.
* ``python_bin`` (optional): the path of a specific Python interpreter to use.
* ``testium_path`` (optional): the path of a specific testium executable to use.
* ``log_file`` (optional): the path of the log file. In GUI mode, if not provided, a file is created with a timestamp next to the ``.tum`` file. Not used in batch mode.
* ``report_file`` (optional): the path of the report file to create.
* ``start_time`` (optional): earliest time to execute the sub-instance, in ``HH:MM`` format.
* ``end_time`` (optional): latest time for execution within a time frame, in ``HH:MM`` format.
* ``wait_for_exec`` (optional): ``true`` to wait until the time window defined by ``start_time`` and ``end_time`` is reached before running. Requires both ``start_time`` and ``end_time``.

View File

@@ -4,6 +4,7 @@ SUPPORTED_API = [
"setgd",
"delgd",
"add_plot_values",
"last_plot_value"
"last_plot_value",
"text_mode",
]

View File

@@ -102,7 +102,7 @@ def main():
if (lf != '') or (rf != '') or (tf != '') or (pn != []):
print('"-l", "-p", "-t", "-n" options are not supported in this mode.')
t = Terminal(os.getcwd(), cf, defines, args.no_color)
t = Terminal(os.getcwd(), cf, defines, args.no_color, text_mode=True)
loop = 1
while loop:
@@ -124,7 +124,8 @@ def main():
print('"-l" option is not supported in this mode.')
from interpreter.batch import Batch
b = Batch(tf, cf, defines, rf, args.report_type, pn, args.no_color)
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

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
@@ -22,6 +23,7 @@ class Batch:
report_type,
report_pattern,
no_color,
text_mode=False,
):
try:
try:
@@ -51,6 +53,7 @@ class Batch:
signal(SIGINT, self.sigint_handler)
self._success = False
msg_queue = Queue()
self.tst_ctrl = TestSetController()
tst_proc = TestProcess(
@@ -59,11 +62,21 @@ class Batch:
self.tst_ctrl,
config_files,
defines,
text_mode=text_mode,
)
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",
@@ -78,14 +91,18 @@ class Batch:
while True:
try:
m = msg_queue.get(timeout=0.2)
if m.get("id", None) is None:
# No id -> finished
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():
break
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:
@@ -94,5 +111,12 @@ class Batch:
finally:
stdio_redir.restore()
@property
def success(self):
return self._success
def sigint_handler(self, signal_received, frame):
self.tst_ctrl.control("stop")
try:
self.tst_ctrl.control("stop", timeout=5)
except Exception:
pass

View File

@@ -1,4 +1,5 @@
import os
import signal
from multiprocessing import Process, Queue, Pipe
from queue import Empty
from threading import Thread
@@ -41,6 +42,7 @@ class TestProcess(Process):
config_files,
defines,
gui_defaults={},
text_mode=False,
) -> None:
super().__init__()
self.__fname = file_name
@@ -49,6 +51,7 @@ class TestProcess(Process):
self.__cfgf = config_files
self.__defs = defines
self.__gui_defaults = gui_defaults # default values coming from GUI prefs
self.__text_mode = text_mode
self.__exec = False
self.__loaded = False
self.__closed = False
@@ -194,6 +197,7 @@ class TestProcess(Process):
def run(self):
signal.signal(signal.SIGINT, signal.SIG_IGN)
try:
try:
# Thread for stdout redirection
@@ -224,6 +228,10 @@ Is the python exec path correct ?"""
# Load the test file
test_dict, param_files = self._load_test(init_param_files, glob_variables)
if self.__text_mode:
tm.setgd("_text_mode", True)
tm.setgd("_interactive", False)
# Backup the global dict in case of restart of the test
gdict = backup_gd()
@@ -275,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:
@@ -331,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):
@@ -421,7 +431,7 @@ Is the python exec path correct ?"""
try:
# read the pipe data
data = cconn.recv()
print(data, end="")
print(data, end="", flush=True)
except EOFError:
# exit the loop is the pipe is closed
break

View File

@@ -56,7 +56,7 @@ class Terminal(Cmd):
cst.TYPE_CYCLE
]
def __init__(self, working_dir, config_files, defines, no_color):
def __init__(self, working_dir, config_files, defines, no_color, text_mode=False):
super().__init__()
self.working_dir = working_dir
self.config_files = config_files
@@ -69,6 +69,8 @@ class Terminal(Cmd):
# Define the builtin variables
set_standard_gd_keys("Unnamed", self.working_dir, '', config_files)
update_global([], defines)
if text_mode:
tm.setgd("_text_mode", True)
# creation of the functions
for tst in self.SUPPORTED_TESTS:

View File

@@ -5,7 +5,7 @@ from itertools import chain
from PySide6.QtGui import QIcon, QPixmap
from PySide6.QtWidgets import QApplication, QDialog, QDialogButtonBox
from PySide6.QtCore import Qt, QSettings, QSize
from PySide6.QtCore import Qt, QSettings, QTimer, QSize
from PySide6.QtGui import QFont, QFontInfo
from PySide6.QtWidgets import QTreeWidgetItem
@@ -207,6 +207,9 @@ def main(args, conn=None):
d.connect_checked()
d.choicesView.setFocus()
auto_result = args[4] if len(args) > 4 else None
if auto_result is not None:
QTimer.singleShot(2000, lambda: d.accept() if auto_result.lower() == 'ok' else d.reject())
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -1,6 +1,6 @@
import sys
from PySide6.QtCore import (Qt)
from PySide6.QtCore import Qt, QTimer
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6 import (QtGui)
@@ -38,6 +38,10 @@ def main(args, conn):
d.labelImage.setPixmap(QtGui.QPixmap.fromImage(image2))
auto_result = args[3] if len(args) > 3 else None
if auto_result is not None:
QTimer.singleShot(2000, lambda: d.accept() if auto_result.lower() == 'ok' else d.reject())
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -2,7 +2,7 @@ import sys
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt
from PySide6.QtCore import Qt, QTimer
def main(args):
@@ -15,6 +15,8 @@ def main(args):
msg.setText(args[1])
msg.setIcon(QMessageBox.Information)
msg.setStandardButtons(QMessageBox.Ok)
if len(args) > 2:
QTimer.singleShot(2000, lambda: msg.button(QMessageBox.Ok).click())
msg.exec()
if hasattr(sys, "frozen"):

View File

@@ -2,7 +2,7 @@ import sys
import os
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtCore import Qt, QTimer
from interpreter.test_items.dialog_note_files import dialog_note_win
from multiprocessing import freeze_support
@@ -23,6 +23,14 @@ def main(args, conn=None):
d.setWindowTitle(args[0])
d.labelDialog.setText(args[1])
d.textEdit.setFocus()
auto_result = args[2] if len(args) > 2 else None
if auto_result is not None:
auto_value = args[3] if len(args) > 3 else None
def _auto_close():
if auto_value is not None:
d.textEdit.setPlainText(auto_value)
d.accept() if auto_result.lower() == 'ok' else d.reject()
QTimer.singleShot(2000, _auto_close)
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -2,7 +2,7 @@ import sys
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt
from PySide6.QtCore import Qt, QTimer
def main(args, conn):
@@ -16,6 +16,10 @@ def main(args, conn):
msg.setText(args[1])
msg.setIcon(QMessageBox.Question)
msg.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
auto_result = args[2] if len(args) > 2 else None
if auto_result is not None:
btn = QMessageBox.Yes if auto_result.lower() == 'yes' else QMessageBox.No
QTimer.singleShot(2000, lambda: msg.button(btn).click())
reply = msg.exec()
conn.send(reply)
except Exception as e:

View File

@@ -2,7 +2,7 @@ import sys
import os
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtCore import Qt, QTimer
from interpreter.test_items.dialog_value_files import dialog_value_win
from multiprocessing import freeze_support
@@ -25,6 +25,14 @@ def main(args, conn=None):
d.labelDialog.setText(args[1])
d.lineEdit.setText(args[2])
d.lineEdit.setFocus()
auto_result = args[3] if len(args) > 3 else None
if auto_result is not None:
auto_value = args[4] if len(args) > 4 else None
def _auto_close():
if auto_value is not None:
d.lineEdit.setText(auto_value)
d.accept() if auto_result.lower() == 'ok' else d.reject()
QTimer.singleShot(2000, _auto_close)
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -1,7 +1,6 @@
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.dialog_choices_files import choices_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
@@ -17,13 +16,69 @@ class TestItemChoicesDialog(TestItemDialogBase):
self._question = self._prms.getParam("question", required=True)
self._choices = self._prms.getParam("choices", required=True)
self._default_icon = self._prms.getParam("icon", required=False, default=None)
self._auto_result = self._prms.getParam("auto_result", required=False, default=None)
def _print_choices(self, choices, indent=0):
if not isinstance(choices, list):
return
for choice in choices:
name = choice.get("name", "")
desc = choice.get("description", "")
line = " " * indent + f"- {name}"
if desc:
line += f": {desc}"
print(line)
sub = choice.get("choices", None)
if sub:
self._print_choices(sub, indent + 1)
def _all_checked(self, choices):
result = []
if not isinstance(choices, list):
return result
for choice in choices:
item = {"name": choice.get("name", ""), "checked": True}
sub = choice.get("choices", None)
if sub is not None:
item["choices"] = self._all_checked(sub)
result.append(item)
return result
@test_run
def execute(self):
q = self._prms.expanse(self._question)
choices = self._prms.expanse(self._choices)
icon = self._prms.expanse(self._default_icon)
result = self._run_dialog_with_result(choices_dialog.main, [self.name(), q, choices, icon])
if _is_text_mode():
print(f"Choices: {q}")
self._print_choices(choices)
if _is_interactive():
ans = input("Accept all? (y/n) [default: y]: ").strip().lower()
if ans in ('n', 'no'):
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, "Cancelled")
else:
val = self._all_checked(choices)
self.result.value = val
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
elif ar == 'cancel':
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, "Cancelled")
else:
val = self._all_checked(choices)
self.result.value = val
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
return
from interpreter.test_items.dialog_choices_files import choices_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q, choices, icon] + ([ar] if ar is not None else [])
result = self._run_dialog_with_result(choices_dialog.main, args)
if result is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return

View File

@@ -1,7 +1,16 @@
import multiprocessing
import libs.testium as tm
from interpreter.test_items.test_item import TestItem
def _is_text_mode():
return tm.text_mode()
def _is_interactive():
return bool(tm.gd("_interactive", True))
_spawn_ctx = multiprocessing.get_context('spawn')
@@ -40,7 +49,10 @@ class TestItemDialogBase(TestItem):
result = None
while p.is_alive() and not self._is_stopped:
if parent_conn.poll(0.5):
result = parent_conn.recv()
try:
result = parent_conn.recv()
except EOFError:
pass
break
self._cleanup_process(p)
return result

View File

@@ -2,8 +2,7 @@ import os
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.dialog_image_files import dialog_image
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
@@ -21,6 +20,7 @@ class TestItemImageDialog(TestItemDialogBase):
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam("question", required=True)
self._filename = self._prms.getParam("filename", required=True)
self._auto_result = self._prms.getParam("auto_result", required=False, default=None)
@test_run
def execute(self):
@@ -31,7 +31,23 @@ class TestItemImageDialog(TestItemDialogBase):
image_path = os.path.normpath(
os.path.join(tm.gd("test_directory"), image_path)
)
succ = self._run_dialog_with_result(dialog_image.main, [self.name(), q, image_path])
if _is_text_mode():
if _is_interactive():
ans = input("Accept? (y/n) [default: y]: ").strip().lower()
self.result.set(TestValue.FAILURE if ans in ('n', 'no') else TestValue.SUCCESS)
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
elif ar == 'cancel':
self.result.set(TestValue.FAILURE)
else:
self.result.set(TestValue.SUCCESS)
return
from interpreter.test_items.dialog_image_files import dialog_image
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q, image_path] + ([ar] if ar is not None else [])
succ = self._run_dialog_with_result(dialog_image.main, args)
if succ is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
elif succ:

View File

@@ -3,8 +3,7 @@ import sys
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.dialog_msg_files import msg_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
@@ -20,12 +19,27 @@ class TestItemMsgDialog(TestItemDialogBase):
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
print("Message Displayed:\n" + q)
exitcode = self._run_dialog(msg_dialog.main, [self.name(), q])
if _is_text_mode():
if _is_interactive():
input("Press Enter to continue...")
self.result.set(TestValue.SUCCESS)
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is not None:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
return
from interpreter.test_items.dialog_msg_files import msg_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q] + ([ar] if ar is not None else [])
exitcode = self._run_dialog(msg_dialog.main, args)
if exitcode == 0:
self.result.set(TestValue.SUCCESS)
else:

View File

@@ -1,7 +1,6 @@
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.dialog_note_files import test_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
@@ -15,12 +14,50 @@ class TestItemNoteDialog(TestItemDialogBase):
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
self._auto_value = self._prms.getParam('auto_value', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
print("Question:\n" + q)
result = self._run_dialog_with_result(test_dialog.main, [self.name(), q])
if _is_text_mode():
if _is_interactive():
print("Enter your note (type '.' on a new line to finish, empty line to cancel):")
lines = []
while True:
line = input()
if line == '.':
break
lines.append(line)
val = '\n'.join(lines)
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
av = self._prms.expanse(self._auto_value) if self._auto_value is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
return
if ar == 'cancel':
self.result.set(TestValue.FAILURE, 'Dialog cancelled')
return
val = av if av is not None else ''
tm.setgd(self.name(), val)
print("\n" + ("-" * 80) + "\n")
print("- Test note\n")
print("-" * 80 + "\n")
print(val)
print("-" * 80 + "\n")
self.result.reported = {'note': val}
if val:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
return
from interpreter.test_items.dialog_note_files import test_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
av = self._prms.expanse(self._auto_value) if self._auto_value is not None else None
args = [self.name(), q] + ([ar, av] if ar is not None else [])
result = self._run_dialog_with_result(test_dialog.main, args)
if result is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return

View File

@@ -1,9 +1,6 @@
from PySide6.QtWidgets import QMessageBox
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.dialog_question_files import question_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
@@ -19,15 +16,40 @@ class TestItemQuestionDialog(TestItemDialogBase):
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
print('Question asked:\n' + q + '\n')
succ = self._run_dialog_with_result(question_dialog.main, [self.name(), q])
if _is_text_mode():
if _is_interactive():
ans = input("Answer yes (y) or no (n) [default: y]: ").strip().lower()
if ans in ('n', 'no'):
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')
else:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
elif ar in ('no', 'cancel'):
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')
else:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')
return
from interpreter.test_items.dialog_question_files import question_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q] + ([ar] if ar is not None else [])
succ = self._run_dialog_with_result(question_dialog.main, args)
if succ is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return
from PySide6.QtWidgets import QMessageBox
if succ == QMessageBox.Yes:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')

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')

View File

@@ -3,9 +3,9 @@ from time import sleep
from datetime import timedelta
from multiprocessing import Process, Pipe
import libs.testium as tm
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
from interpreter.test_items.dialog_sleep_files import dialog_sleep
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
@@ -43,6 +43,20 @@ class TestItemSleep(TestItem):
#test core function
if has_dialog:
if tm.text_mode():
import time as _time
print(f"Sleep {timeout}s (press Ctrl+C to abort)...")
end_time = _time.time() + float(timeout)
while _time.time() < end_time and not self._is_stopped:
sleep(0.2)
if self._is_stopped:
print("Aborted")
self.result.set(TestValue.FAILURE, 'Sleep aborted')
else:
self.result.set(TestValue.SUCCESS, f'Sleep {timeout} sec')
return
from interpreter.test_items.dialog_sleep_files import dialog_sleep
parent_conn, child_conn = Pipe()
p=Process(target=dialog_sleep.main, args=([self.name(), timeout],child_conn))
p.start()

View File

@@ -1,7 +1,6 @@
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.tested_references_files import tested_refs_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
@@ -16,12 +15,40 @@ class TestItemTestedRefsDialog(TestItemDialogBase):
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._init_values = self._prms.getParamAll('reference', required=False, processed=True)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
init_values = ','.join(self._init_values)
result = self._run_dialog_with_result(tested_refs_dialog.main, [self.name(), q, init_values])
if _is_text_mode():
print(f"References: {q}")
rows = init_values.split(',') if init_values else ['']
result_rows = []
for i, row in enumerate(rows):
parts = (row.split('/') + ['', '', ''])[:3]
if _is_interactive():
ref = input(f"Row {i+1} - Reference [{parts[0]}]: ").strip() or parts[0]
rev = input(f"Row {i+1} - Revision [{parts[1]}]: ").strip() or parts[1]
serial = input(f"Row {i+1} - Serial [{parts[2]}]: ").strip() or parts[2]
else:
ref, rev, serial = parts[0], parts[1], parts[2]
result_rows.append(f"{ref}/{rev}/{serial}")
val = ','.join(result_rows)
if _is_interactive():
succ = True
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
if ar is None:
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
return
succ = ar != 'cancel'
result = [val, succ]
else:
from interpreter.test_items.tested_references_files import tested_refs_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
args = [self.name(), q, init_values] + ([ar] if ar is not None else [])
result = self._run_dialog_with_result(tested_refs_dialog.main, args)
if result is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return

View File

@@ -1,7 +1,6 @@
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.dialog_value_files import test_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
import libs.testium as tm
@@ -19,13 +18,45 @@ class TestItemValueDialog(TestItemDialogBase):
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._default = self._prms.getParam('default', '')
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
self._auto_value = self._prms.getParam('auto_value', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
d = self._prms.expanse(self._default)
print("Question:\n" + q)
result = self._run_dialog_with_result(test_dialog.main, [self.name(), q, d])
if _is_text_mode():
if _is_interactive():
prompt = f"Enter value [{d}]: " if d else "Enter value: "
ans = input(prompt).strip()
else:
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
av = self._prms.expanse(self._auto_value) if self._auto_value is not None else None
if ar is None:
print("Answer: \nDialog not supported in batch mode")
self.result.set(TestValue.FAILURE, 'Dialog not supported in batch mode')
return
if ar == 'cancel':
print("Answer: \nDialog cancelled")
self.result.set(TestValue.FAILURE, 'Dialog cancelled')
return
ans = av if av is not None else ''
val = ans if ans else d
tm.setgd(self.name(), val)
print("Answer: " + str(val))
if val:
self.result.reported = {'question': q, 'answer': val}
self.result.value = val
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, 'No value entered')
return
from interpreter.test_items.dialog_value_files import test_dialog
ar = self._prms.expanse(self._auto_result) if self._auto_result is not None else None
av = self._prms.expanse(self._auto_value) if self._auto_value is not None else None
args = [self.name(), q, d] + ([ar, av] if ar is not None else [])
result = self._run_dialog_with_result(test_dialog.main, args)
if result is None:
self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return

View File

@@ -2,7 +2,7 @@ import sys
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QDialog, QTableWidgetItem)
from PySide6.QtCore import (Qt, QSettings)
from PySide6.QtCore import Qt, QSettings, QTimer
try:
from interpreter.test_items.tested_references_files import tested_refs_win
@@ -52,6 +52,9 @@ def main(args, conn=None):
i += 1
d.tableReferences.setFocus()
auto_result = args[3] if len(args) > 3 else None
if auto_result is not None:
QTimer.singleShot(2000, lambda: d.accept() if auto_result.lower() == 'ok' else d.reject())
dres = d.exec()
if dres == QDialog.Rejected:

View File

@@ -189,7 +189,13 @@ class LuaProcessBase:
if tm.debug_enabled() and tm.gd("debug_rpc", False):
params.append("--verbose")
self._process = subprocess.Popen(params, env=env, cwd=func_proc_path)
self._process = subprocess.Popen(
params, env=env, cwd=func_proc_path,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
restore_signals=False,
)
self._rpc = JsonRpcClient(
"localhost", self._port, req_handler=self._req_handler

View File

@@ -158,7 +158,13 @@ class PyProcessBase:
if tm.debug_enabled() and tm.gd("debug_rpc", False):
params.append("-v")
self._process = subprocess.Popen(params, env=env, cwd=func_proc_path)
self._process = subprocess.Popen(
params, env=env, cwd=func_proc_path,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
restore_signals=False,
)
self._rpc = JsonRpcClient(
"localhost", self._port, req_handler=self._req_handler

View File

@@ -1,32 +1,102 @@
import colorama
import os
import re
import sys
import colorama
from colorama import Fore, Style
COLOR_DEFAULT = Fore.WHITE
COLOR_RESET = Fore.RESET + Style.RESET_ALL + COLOR_DEFAULT
def _detect_dark_background() -> bool:
"""Detect whether the terminal has a dark background.
def colored_string(string: str, inputs: list) -> None:
"""Function which calculate the coloring of strings with many layers.
Overlap of layers and inner layers are managed.
Tries the following methods in order:
1. ``COLORFGBG`` environment variable (Konsole, rxvt, …)
2. OSC 11 terminal query — reads the actual background colour from the
terminal emulator (xterm, VTE, kitty, WezTerm, …)
3. ``darkdetect`` module — OS-level dark-mode preference (optional dep)
Returns ``True`` for a dark background (default assumption).
"""
cols = [COLOR_DEFAULT for i in range(len(string))]
for input in inputs:
for i in range(input[0][0], input[0][1]):
cols[i] = input[1]
# --- Method 1: COLORFGBG ---
colorfgbg = os.environ.get("COLORFGBG", "")
if colorfgbg:
try:
bg = int(colorfgbg.split(";")[-1])
# 0-6: dark palette entries, 7-15: light palette entries
return bg < 7
except (ValueError, IndexError):
pass
# --- Method 2: OSC 11 terminal query ---
if sys.stdin.isatty() and sys.stdout.isatty():
try:
import select
import termios
import tty
fd = sys.stdin.fileno()
old = termios.tcgetattr(fd)
try:
tty.setraw(fd)
# Query background colour
sys.stdout.write("\033]11;?\007")
sys.stdout.flush()
ready, _, _ = select.select([sys.stdin], [], [], 0.2)
if ready:
response = ""
while True:
r2, _, _ = select.select([sys.stdin], [], [], 0.05)
if not r2:
break
chunk = os.read(fd, 64).decode("latin-1", errors="replace")
response += chunk
# Terminal answers with ESC]11;rgb:RR../GG../BB..<BEL|ST>
if response.endswith("\007") or response.endswith("\033\\"):
break
m = re.search(
r"rgb:([0-9a-fA-F]+)/([0-9a-fA-F]+)/([0-9a-fA-F]+)",
response,
)
if m:
# Components are 8- or 16-bit hex; normalise to 0-255
def _norm(h: str) -> float:
return int(h[:2], 16)
r_v = _norm(m.group(1))
g_v = _norm(m.group(2))
b_v = _norm(m.group(3))
luminance = 0.299 * r_v + 0.587 * g_v + 0.114 * b_v
return luminance < 128
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old)
except Exception:
pass
# Default: assume dark terminal
return True
def _colored_string(string: str, inputs: list, color_default: str, color_reset: str) -> str:
"""Return *string* with ANSI colour codes applied according to *inputs*.
*inputs* is a list of ``[[start, end], color_code]`` pairs.
Overlapping layers are handled: the last listed colour wins.
"""
cols = [color_default for _ in range(len(string))]
for span, color in inputs:
for i in range(span[0], span[1]):
cols[i] = color
# construction of the string
s = ""
ilast = 0
last_col = COLOR_DEFAULT
last_col = color_default
for i in range(len(string)):
if last_col != cols[i]:
s = s + string[ilast:i] + COLOR_RESET + cols[i]
s = s + string[ilast:i] + color_reset + cols[i]
ilast = i
last_col = cols[i]
return s + string[ilast:] + COLOR_RESET
return s + string[ilast:] + color_reset
class TermLog:
@@ -37,46 +107,74 @@ class TermLog:
DEBUG = ["DEBUG"]
BOOL = ["False", "True", "false", "true", "FALSE", "TRUE"]
def __init__(self, out) -> None:
"""Class used to color the stdout in batch and terminal mode."""
def __init__(self, out, dark_bg: bool = None) -> None:
"""Class used to colour the stdout in batch and terminal mode.
:param out: Underlying output stream.
:param dark_bg: ``True`` for dark background, ``False`` for light.
``None`` (default) triggers auto-detection.
"""
colorama.init()
self.out = out
self.pats = []
self.pats = self.pats + [
[re.compile('(\\"[^\\"]+\\")'), Fore.LIGHTBLUE_EX + Style.BRIGHT],
[re.compile("(\\'[^\\']+\\')"), Fore.LIGHTBLUE_EX + Style.BRIGHT],
[re.compile("(<-----|----->) step"), Fore.BLUE],
[
re.compile(
r"([\d\.]+)",
),
Fore.MAGENTA,
],
[re.compile(r"(@@\d+@@)"), Fore.BLACK],
self.residue = ""
if dark_bg is None:
dark_bg = _detect_dark_background()
if dark_bg:
color_default = Fore.WHITE
color_string = Fore.LIGHTBLUE_EX + Style.BRIGHT
color_number = Fore.MAGENTA
color_bool = Fore.MAGENTA
color_step = Fore.BLUE
color_marker = Fore.BLACK
color_warn = Fore.YELLOW
color_info = Style.BRIGHT
color_debug = Fore.BLUE + Style.BRIGHT
color_pass = Fore.GREEN + Style.BRIGHT
color_fail = Fore.RED + Style.BRIGHT
else:
color_default = Fore.RESET
color_string = Fore.BLUE
color_number = Fore.MAGENTA
color_bool = Fore.MAGENTA
color_step = Fore.BLUE
color_marker = Fore.RESET
color_warn = Fore.YELLOW + Style.BRIGHT
color_info = Fore.CYAN
color_debug = Fore.BLUE
color_pass = Fore.GREEN
color_fail = Fore.RED + Style.BRIGHT
self._color_default = color_default
self._color_reset = Fore.RESET + Style.RESET_ALL + color_default
self.pats = [
[re.compile(r'("(?:[^"]+)")'), color_string],
[re.compile(r"('(?:[^']+)')"), color_string],
[re.compile(r"(<-----|----->) step"), color_step],
[re.compile(r"([\d\.]+)"), color_number],
[re.compile(r"(@@\d+@@)"), color_marker],
]
for word in self.BOOL:
self.pats.append([re.compile("({})".format(word)), Fore.MAGENTA])
self.pats.append([re.compile(r"({})".format(word)), color_bool])
for word in self.WARN:
self.pats.append([re.compile("({})".format(word)), Fore.YELLOW])
self.pats.append([re.compile(r"({})".format(word)), color_warn])
for word in self.INFO:
self.pats.append([re.compile("({})".format(word)), Style.BRIGHT])
self.pats.append([re.compile(r"({})".format(word)), color_info])
for word in self.DEBUG:
self.pats.append([re.compile("({})".format(word)), Fore.BLUE + Style.BRIGHT])
self.pats.append([re.compile(r"({})".format(word)), color_debug])
for word in self.PASS:
self.pats.append(
[re.compile("({})".format(word)), Fore.GREEN + Style.BRIGHT]
)
self.pats.append([re.compile(r"({})".format(word)), color_pass])
for word in self.FAIL:
self.pats.append([re.compile("({})".format(word)), Fore.RED + Style.BRIGHT])
self.residue = ""
self.pats.append([re.compile(r"({})".format(word)), color_fail])
def find_pats(self, line):
spans = []
for p in self.pats:
it = p[0].finditer(line)
for m in it:
for p, color in self.pats:
for m in p.finditer(line):
if m:
spans.append([m.span(), p[1]])
spans.append([m.span(), color])
return spans
def write(self, s: str) -> None:
@@ -87,15 +185,19 @@ class TermLog:
if s[-1:] != "\n":
pos = s.rfind("\n")
if pos >= 0:
self.residue = s[pos:]
s = s[:pos]
self.residue = s[pos + 1:]
s = s[:pos + 1]
else:
# only one line
self.out.write(colored_string(s, self.find_pats(s)))
# single incomplete line — output immediately
self.out.write(_colored_string(s, self.find_pats(s),
self._color_default, self._color_reset))
return
# multiline case
for l in s.splitlines():
self.out.write(colored_string(l, self.find_pats(l)) + "\n")
# one or more complete lines
for line in s.splitlines():
self.out.write(
_colored_string(line, self.find_pats(line),
self._color_default, self._color_reset) + "\n"
)
def flush(self):
if self.residue != "":

View File

@@ -209,6 +209,15 @@ def OS():
return platform.system()
def text_mode():
"""Whether testium is running in text mode (batch ``-b`` or terminal ``-m``).
:return: ``True`` if running in text mode, ``False`` otherwise.
:rtype: bool
"""
return bool(globdict.gd("_text_mode", False))
def sys_encoding():
if OS() == "Windows":
enc = "oem"

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)

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()

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)

View File

@@ -0,0 +1,29 @@
import libs.testium as libtm
def check_os(expected_os):
result = libtm.OS()
assert result == expected_os, f"Expected {expected_os!r}, got {result!r}"
return 0
def check_get_main_dir():
d = libtm.get_main_dir()
assert isinstance(d, str) and len(d) > 0
return 0
def check_timestamp_as_sec_conversion():
assert libtm.timestamp_as_sec(0) == 0.0
assert libtm.timestamp_as_sec(10000) == 1.0
assert libtm.timestamp_as_sec(5000) == 0.5
return 0
def check_timestamp():
libtm.init_timestamp()
t = libtm.timestamp()
assert isinstance(t, int) and t >= 0
ts = libtm.timestamp_as_sec()
assert isinstance(ts, float) and ts >= 0.0
return 0

View File

@@ -9,4 +9,29 @@
- group:
name : Various syntax robustness
steps:
- !include syntax_robustness/test.tum
- !include syntax_robustness/test.tum
- group:
name: Helper lib functions
steps:
- py_func:
name: OS
key: $(test)_PASS
file: $(test_path)$(psep)helper_lib.py
func_name: check_os
param:
- $(os)
- py_func:
name: get_main_dir
key: $(test)_PASS
file: $(test_path)$(psep)helper_lib.py
func_name: check_get_main_dir
- py_func:
name: timestamp_as_sec conversion
key: $(test)_PASS
file: $(test_path)$(psep)helper_lib.py
func_name: check_timestamp_as_sec_conversion
- py_func:
name: timestamp and timestamp_as_sec
key: $(test)_PASS
file: $(test_path)$(psep)helper_lib.py
func_name: check_timestamp

View File

@@ -2,6 +2,7 @@
name: dialog image PASS
condition: $(validation_dialogs)
question: click ok if you see the image
auto_result: "ok"
key: $(test)_PASS
filename: $(test_path)$(psep)IMG_20140213_171455.jpg
@@ -9,6 +10,7 @@
name: dialog image FAIL
condition: $(validation_dialogs)
question: click cancel
auto_result: "cancel"
key: $(test)_FAIL
filename: $(test_path)$(psep)IMG_20140213_171455.jpg
@@ -17,45 +19,54 @@
condition: $(validation_dialogs)
key: $(test)_PASS
question: click ok
auto_result: "ok"
- dialog_references:
name: dialog_reference FAIL
condition: $(validation_dialogs)
key: $(test)_FAIL
question: click cancel
auto_result: "cancel"
- dialog_value:
name: dialog_value PASS
condition: $(validation_dialogs)
key: $(test)_PASS
question: enter 123 and click ok
auto_result: "ok"
auto_value: "123"
- dialog_value:
name: dialog_value empty FAIL
condition: $(validation_dialogs)
key: $(test)_FAIL
question: enter nothing and click ok
auto_result: "ok"
- dialog_value:
name: dialog_value canceled FAIL
condition: $(validation_dialogs)
key: $(test)_FAIL
question: enter nothing and click cancel
auto_result: "cancel"
- dialog_message:
name: dialog_message PASS
condition: $(validation_dialogs)
key: $(test)_PASS
question: click ok
auto_result: "ok"
- dialog_question:
name: dialog_question PASS
condition: $(validation_dialogs)
key: $(test)_PASS
question: click yes
auto_result: "yes"
- dialog_question:
name: dialog_question FAIL
condition: $(validation_dialogs)
key: $(test)_FAIL
question: click no
auto_result: "no"

View File

@@ -41,4 +41,12 @@ function module.get_context_value()
return tm.gd("_lua_ctx_test_value")
end
function module.test_delgd()
tm.setgd("_lua_delgd_test", 42)
assert(tm.gd("_lua_delgd_test") == 42)
tm.delgd("_lua_delgd_test")
assert(tm.gd("_lua_delgd_test", "__deleted__") == "__deleted__")
return 0
end
return module

View File

@@ -180,6 +180,12 @@
func_name: tuple_return
param: [ 0, "OK" ]
- lua_func:
name: delgd test
key: $(test)_PASS
file: $(test_path)$(psep)lua_func.lua
func_name: test_delgd
- group:
name: context_id tests
steps:

View File

@@ -1,74 +1,73 @@
- plot:
name: Open the plot
condition: $(validation_dialogs)
key: $(test)_PASS
plot_name: Mon Plot
steps:
- open:
log_path: $(validation_report_path)
- plot:
name: Add periodic to the plot
condition: $(validation_dialogs)
key: $(test)_PASS
plot_name: Mon Plot
steps:
- periodic:
period: 1
file: $(test_path)$(psep)plot.py
func_name: random_value
eval: '{"periodic": $(result)}'
- sleep:
name: sleep
condition: $(validation_dialogs)
dialog: true
timeout: 3
- loop:
name: Add of other data in the plot
condition: $(validation_dialogs)
iterator: 10
steps:
- plot:
name: Add to the plot
key: $(test)_PASS
plot_name: Mon Plot
steps:
- add:
value1: $(loop_index)
value2: $(loop_index)+2
- sleep:
name: sleep between values
timeout: 1
- py_func:
name: last plot values
key: $(test)_PASS
file: $(test_path)$(psep)plot.py
func_name: LastValues
param:
- Mon Plot
- plot:
name: Export
execute_on_stop: True
condition: $(validation_dialogs)
key: $(test)_PASS
plot_name: Mon Plot
steps:
- export: $(validation_report_path)/plot_export.pdf
- export: $(validation_report_path)/plot_export.csv
- plot:
name: Close the plot
execute_on_stop: True
condition: $(validation_dialogs)
key: $(test)_PASS
plot_name: Mon Plot
steps:
- close:
wait_dialog_exit: True
timeout: 60
- group:
name: Plot test
condition: <| $(validation_dialogs) and not tm.text_mode() |>
steps:
- plot:
name: Open the plot
key: $(test)_PASS
plot_name: Mon Plot
steps:
- open:
log_path: $(validation_report_path)
- plot:
name: Add periodic to the plot
key: $(test)_PASS
plot_name: Mon Plot
steps:
- periodic:
period: 1
file: $(test_path)$(psep)plot.py
func_name: random_value
eval: '{"periodic": $(result)}'
- sleep:
name: sleep
dialog: true
timeout: 3
- loop:
name: Add of other data in the plot
iterator: 10
steps:
- plot:
name: Add to the plot
key: $(test)_PASS
plot_name: Mon Plot
steps:
- add:
value1: $(loop_index)
value2: $(loop_index)+2
- sleep:
name: sleep between values
timeout: 1
- py_func:
name: last plot values
key: $(test)_PASS
file: $(test_path)$(psep)plot.py
func_name: LastValues
param:
- Mon Plot
- plot:
name: Export
execute_on_stop: True
key: $(test)_PASS
plot_name: Mon Plot
steps:
- export: $(validation_report_path)/plot_export.pdf
- export: $(validation_report_path)/plot_export.csv
- plot:
name: Close the plot
execute_on_stop: True
key: $(test)_PASS
plot_name: Mon Plot
steps:
- close:
wait_dialog_exit: True
timeout: 2

View File

@@ -47,3 +47,10 @@ def set_ns_value(val):
def get_ns_value():
obj = tm.gd("_py_ctx_ns_value", None)
return obj.val if obj is not None else None
def test_delgd():
tm.setgd("_py_delgd_test", 42)
assert tm.gd("_py_delgd_test") == 42
tm.delgd("_py_delgd_test")
assert tm.gd("_py_delgd_test", None) is None
return 0

View File

@@ -190,6 +190,12 @@
param: [ 0, "OK" ]
expected_result: [0, "OK"]
- py_func:
name: delgd test
key: $(test)_PASS
file: $(test_path)$(psep)py_func.py
func_name: test_delgd
- group:
name: context_id tests
steps:

View File

@@ -0,0 +1 @@
no_param: Null

View File

@@ -0,0 +1,7 @@
main:
name: run sub-test (always fail)
steps:
- check:
name: fail
values:
- false

View File

@@ -0,0 +1,7 @@
main:
name: run sub-test (always pass)
steps:
- check:
name: pass
values:
- true

View File

@@ -0,0 +1,25 @@
# run item: launches a .tum file in a new testium instance.
# In batch mode the sub-instance runs with -b; in GUI mode with -r.
# The run item result is SUCCESS if the sub-instance launched successfully,
# regardless of its own test result.
- run:
name: run PASS (valid file, passing sub-test)
key: $(test)_PASS
tum: $(test_path)$(psep)sub_pass.tum
- run:
name: run PASS (valid file, failing sub-test)
key: $(test)_PASS
tum: $(test_path)$(psep)sub_fail.tum
- run:
name: run FAIL (file not found)
key: $(test)_FAIL
tum: $(test_path)$(psep)non_existent.tum
- run:
name: run FAIL (wait_for_exec without time window)
key: $(test)_FAIL
tum: $(test_path)$(psep)sub_pass.tum
wait_for_exec: true