20 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
d955ae81f9 Added variables list in "F1" dialog. They are modifiable. To be tested. 2026-04-20 23:27:39 +02:00
2cd3aa3305 updated doc + param 2026-04-20 22:55:09 +02:00
276d485905 Add store_result common attribute to test items
Allows any test item to store its result (or PASS/FAIL status when result
is None) into a named global variable, available to subsequent items via
$(variable_name). store_result runs after expected_result but before
no_fail so the real outcome is always captured.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-20 22:26:47 +02:00
95912dd3e1 Fix 'process_result must fail' test missing expected_result
Without expected_result, a False process_result value does not fail the
test. Adding expected_result: True makes the comparison fail as intended.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-20 21:43:33 +02:00
6d1fb6a6bc Add JSON-RPC echo server for validation suite
Replaces the external jrpces binary dependency with a self-contained
Python script. The server supports TCP (newline-delimited JSON, port 4321)
and UDP (port 4323), handles JSON-RPC 1.0 and 2.0, and implements:
  - echo(*args) -> [args, {}]
  - unknown methods -> error {code: -32000, message: "function not found"}

test.tum is updated to launch jrpc_echo_server.py via python3 and wait
for the "ready" readiness message before running tests.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-20 21:34:43 +02:00
2cc42e9065 Fix 15-second close delay after dialog tests
Dialog subprocesses were forked from TestProcess, inheriting its
multiprocessing Queue objects and their process-shared POSIX semaphores
(_wlock). If a fork happened while the feeder thread held _wlock, the
child exited without releasing it, permanently blocking the feeder
thread on the next wacquire() and stalling Python's atexit _finalize_join
— causing test_proc.join() (no timeout) to hang the app for ~15 seconds.

Fix: use multiprocessing.get_context('spawn') for dialog subprocesses so
they start with a clean interpreter and inherit no semaphores or Queue
state. Also add a terminate/kill fallback timeout to test_proc.join() as
a safety net, and fix the missing return in JsonRpcConnection.is_alive().

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-20 21:14:33 +02:00
2b7678c39e Fix dialog subprocesses: centralize Qt env setup and extract base class
- Add dialog_env.py service: forces QT_QPA_PLATFORM=xcb on Linux so Qt
  doesn't crash under Wayland in spawned subprocesses
- Use QMessageBox instance (instead of static methods) for msg/question
  dialogs so WindowStaysOnTopHint can be set, making them visible
- Add TestItemDialogBase with _run_dialog/_run_dialog_with_result/_cleanup_process,
  removing duplicated subprocess launch/poll/terminate logic from all 7 dialog items
- Reduce terminate() join timeout from 2s to 0.2s across all dialog items

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-19 20:01:56 +02:00
c72176d029 Improve loading error messages with item context and hierarchy path
Add item_load_context() context manager to tum_except.py that enriches
ETUMSyntaxError with the item type, name, and parent path instead of
replacing the original message with a vague 'missing or wrong parameter'.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-19 10:05:40 +02:00
617f599f86 Fix context subprocess leak and document py_func.tm helpers
- process.py: stop context_id engines in the inner finally block, before
  restore_gd() wipes _py_func_contexts/_lua_func_contexts from the global
  dict — engines were previously orphaned after every test run
- py_func/tm.py: add user-facing docstrings to gd/setgd/delgd; remove
  internal JSON-serialization details from the docs
- helper_lib.rst: auto-generate global variable helpers from py_func.tm
  (the actual subprocess API) instead of globdict
- conf.py: add src/ to Sphinx sys.path so py_func.tm is importable
- py_func_test_item.rst: simplify context sharing section, remove
  JSON-serializable/non-serializable distinction for end users
- Regenerated PDF manual

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 16:27:31 +02:00
70 changed files with 2220 additions and 940 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

@@ -14,6 +14,7 @@ import os
import sys
sys.path.insert(0, os.path.abspath('../../../../src/testium/'))
sys.path.insert(0, os.path.abspath('../../../../src/'))
# -- Project information -----------------------------------------------------

View File

@@ -21,7 +21,7 @@ Global variables helper functions
To manage values in the global variables dataset, the following testium library API
must be used:
.. automodule:: interpreter.utils.globdict
.. automodule:: py_func.tm
:members: gd, setgd, delgd
:undoc-members:
:no-index:

View File

@@ -87,6 +87,10 @@ if not provided is given in the table as well.
| | | see :ref:`Expected result<sec_expected_result>` |
| | | for details. |
+-----------------------+-------------------+-------------------------------------------------------+
| ``store_result`` | / | Store the test result in a global variable. |
| | | see :ref:`Store result<sec_store_result>` |
| | | for details. |
+-----------------------+-------------------+-------------------------------------------------------+
last test result
@@ -183,6 +187,61 @@ If the result and the expected_result is equal, the test will be *PASSED* if ``T
The special ``$(result)`` variable is replaced in the ``expected_result`` attribute content with the test result value.
.. _sec_store_result:
Store result
-----------------------------------------------
The ``store_result`` attribute stores the test result into a named global variable,
making it available to subsequent test items via ``$(variable_name)``.
If the test item returns a value (e.g. ``py_func``, ``json_rpc``), that value is stored.
If ``process_result`` is also specified, the stored value is the post-processed result.
If the test item produces no value (result is ``None``), the stored value is the
test status string: ``"PASS"`` or ``"FAIL"``, evaluated after ``expected_result``
but **before** ``no_fail``. This ensures the real outcome is captured even when
``no_fail: True`` would otherwise mask a failure.
.. code-block:: yaml
:caption: Store a function return value
- py_func:
name: Read sensor
func_name: read_temperature
store_result: temperature
- py_func:
name: Check temperature in range
func_name: check_range
param: [$(temperature), 20, 30]
.. code-block:: yaml
:caption: Store a post-processed value
- py_func:
name: Get firmware version string
func_name: get_version
process_result: "'$(result)'.split('.')[0]"
store_result: fw_major
.. code-block:: yaml
:caption: Store the pass/fail status of a test with no return value
- console:
name: Send command
console_name: device
steps:
- writeln: reboot
- read_until: {expected: "ready", timeout: 10}
store_result: reboot_status
- py_func:
name: Use reboot status
func_name: log_status
param: [$(reboot_status)]
Export attribute
-----------------------------------------------

View File

@@ -123,17 +123,12 @@ Each ``py_func`` item without a ``context_id`` runs in a dedicated subprocess th
is started and stopped around the call. State cannot be shared between two such
items using module-level variables.
Two mechanisms are available to share data across calls:
**Using the testium global dictionary**
Inside a ``py_func`` script, the ``tm`` module exposes ``tm.setgd`` and ``tm.gd``
to read and write the testium global dictionary of the test process. Values stored
this way are accessible from any subsequent test item (including other ``py_func``
items) without requiring a shared subprocess.
Inside a ``py_func`` script, ``tm.setgd`` and ``tm.gd`` read and write the testium
global dictionary. Values stored this way are accessible from any subsequent test
item, including other ``py_func`` items, without requiring a shared subprocess.
.. code-block:: python
:caption: sharing a serializable value via the global dictionary
:caption: sharing a value via the global dictionary
import py_func.tm as tm
@@ -144,36 +139,22 @@ items) without requiring a shared subprocess.
def consume():
return tm.gd("my_shared_value", None)
Values stored with ``tm.setgd`` must be JSON-serializable (str, int, float, list,
dict, bool, None). Non-serializable values (objects, connections, file handles…)
are handled transparently by the local fallback described below.
**Using a shared persistent subprocess (``context_id``)**
When ``context_id`` is set, all ``py_func`` items that share the same identifier
reuse the same subprocess. The subprocess is kept alive until the end of the test.
This is required for non-JSON-serializable objects (e.g. a socket connection, a
device handle). Calling ``tm.setgd`` with such a value stores it inside the
subprocess local dictionary instead of sending it to the main process. It can then
be retrieved with ``tm.gd`` from any subsequent call that runs in the same subprocess.
reuse the same persistent subprocess. This allows sharing any Python object across
calls — including objects that cannot be transmitted to other processes.
.. code-block:: python
:caption: sharing a non-serializable object via ``context_id``
:caption: sharing an object via ``context_id``
import py_func.tm as tm
class _Connection: # not JSON-serializable
def __init__(self):
self.value = "open"
def open_connection():
tm.setgd("conn", _Connection()) # stored locally in the subprocess
tm.setgd("conn", MyConnection())
return "ok"
def use_connection():
conn = tm.gd("conn") # retrieved from the subprocess local dict
return conn.value
conn = tm.gd("conn")
return conn.status()
.. code-block:: yaml
:caption: ``py_func`` items sharing a persistent subprocess

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``.

Binary file not shown.

View File

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

View File

@@ -228,7 +228,7 @@ class JsonRpcConnection:
self.recv_thread.join()
def is_alive(self):
self.recv_thread.is_alive()
return self.recv_thread.is_alive()
def wait_ready(self, timeout=None):
return self._event_ready.wait(timeout)

View File

@@ -1,5 +1,6 @@
import traceback
import textwrap
from contextlib import contextmanager
class ETUMError(Exception):
@@ -67,6 +68,28 @@ class ETUMParamError(ETUMError):
return lines
@contextmanager
def item_load_context(item_type: str, item_name: str, filename: str = ""):
"""Context manager that enriches ETUMSyntaxError with item context during loading.
Usage in test item __init__:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.param = self._prms.getParam("param", required=True)
"""
try:
yield
except ETUMSyntaxError as e:
raise ETUMSyntaxError(
f"In '{item_type}' item named '{item_name}':\n{e._message}",
filename or e._file,
) from e
except Exception as e:
raise ETUMSyntaxError(
f"In '{item_type}' item named '{item_name}':\nUnexpected error: {e}",
filename,
) from e
def print_exception(exc: ETUMError):
if not isinstance(exc, ETUMError):
print(traceback.format_exc(4))

View File

@@ -45,12 +45,21 @@ for k in SUPPORTED_API:
###############################################################################
# gd/setgd/delgd with local-dict fallback for non-serializable values
def gd(*params):
key = params[0] if params else None
if key is not None and key in _local_dict:
return _local_dict[key]
def gd(name, default=None):
"""Return a value from the testium global dictionary.
The value is accessible from any test item and from any ``py_func``
subprocess, regardless of the ``context_id`` used.
:param name: Name of the entry to retrieve.
:type name: str
:param default: Value returned when the key is absent. Defaults to ``None``.
:return: The stored value, or *default* if not found.
"""
if name is not None and name in _local_dict:
return _local_dict[name]
if _func_call_thread is not None:
res = _func_call_thread.call("gd", params)
res = _func_call_thread.call("gd", (name, default))
if "result" in res:
return res["result"]
elif "error" in res:
@@ -60,14 +69,25 @@ def gd(*params):
raise ETUMRuntimeError("api not initialized")
def setgd(*params):
key = params[0] if params else None
value = params[1] if len(params) > 1 else None
if key is not None and not _is_json_serializable(value):
_local_dict[key] = value
def setgd(name, value):
"""Store a value in the testium global dictionary.
The stored value is accessible from any subsequent test item and from any
``py_func`` subprocess via :func:`gd`.
When ``context_id`` is used on the ``py_func`` item, any Python object
(including those that cannot be transmitted to other processes) can be
stored and shared between calls running in the same subprocess.
:param name: Name of the entry to set.
:type name: str
:param value: Value to store.
"""
if name is not None and not _is_json_serializable(value):
_local_dict[name] = value
return None
if _func_call_thread is not None:
res = _func_call_thread.call("setgd", params)
res = _func_call_thread.call("setgd", (name, value))
if "result" in res:
return res["result"]
elif "error" in res:
@@ -77,13 +97,17 @@ def setgd(*params):
raise ETUMRuntimeError("api not initialized")
def delgd(*params):
key = params[0] if params else None
if key is not None and key in _local_dict:
del _local_dict[key]
def delgd(name):
"""Remove an entry from the testium global dictionary.
:param name: Name of the entry to remove.
:type name: str
"""
if name is not None and name in _local_dict:
del _local_dict[name]
return None
if _func_call_thread is not None:
res = _func_call_thread.call("delgd", params)
res = _func_call_thread.call("delgd", (name,))
if "result" in res:
return res["result"]
elif "error" in res:

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
@@ -8,6 +9,7 @@ import copy
from lib.string_queue import StringQueue
from lib.tum_except import print_exception, ETUMRuntimeError, ETUMSyntaxError
import libs.testium as tm
import interpreter.utils.globdict as globdict
from interpreter.utils.params import expanse
from interpreter.utils.test_ctrl import TestSetController
from interpreter.utils.test_init import (
@@ -40,6 +42,7 @@ class TestProcess(Process):
config_files,
defines,
gui_defaults={},
text_mode=False,
) -> None:
super().__init__()
self.__fname = file_name
@@ -48,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
@@ -193,6 +197,7 @@ class TestProcess(Process):
def run(self):
signal.signal(signal.SIGINT, signal.SIG_IGN)
try:
try:
# Thread for stdout redirection
@@ -223,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()
@@ -255,6 +264,7 @@ Is the python exec path correct ?"""
try:
test_run_init()
print(test_run_header())
globdict.set_update_queue(self.__squeue)
test_set.execute()
finally:
if test_set.success():
@@ -265,8 +275,16 @@ Is the python exec path correct ?"""
test_set.run_post_exec()
finally:
self.__exec = False
# Stop shared context engines before restore_gd wipes them
for engine in tm.gd("_py_func_contexts", {}).values():
engine.stop()
engine.join()
for engine in tm.gd("_lua_func_contexts", {}).values():
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:
print_exception(e)
@@ -304,6 +322,9 @@ Is the python exec path correct ?"""
"enabled_state": test_set.getEnabledState,
"process_param": self.process_param,
"set_test_outputs": self.set_test_outputs,
"get_gd_vars": self.get_gd_vars,
"set_gd_var": self.set_gd_var,
"del_gd_var": self.del_gd_var,
"set_enabled_state": test_set.setEnabledState,
"check_uncheck_all": test_set.checkUncheckAll,
"get_folded": test_set.getFolded,
@@ -318,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):
@@ -337,6 +360,25 @@ Is the python exec path correct ?"""
def set_test_outputs(self, outputs: list):
tm.setgd("test_outputs", outputs)
def get_gd_vars(self):
import json
result = {}
for k, v in globdict.global_dict.items():
if k.startswith("_"):
continue
try:
json.dumps(v)
result[k] = v
except (TypeError, ValueError):
pass
return result
def set_gd_var(self, name: str, value):
tm.setgd(name, value)
def del_gd_var(self, name: str):
tm.delgd(name)
def process_control_commands(self, tctrl):
term = False
while (not term) and (not self.__closed):
@@ -389,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
@@ -185,7 +185,9 @@ def main(args, conn=None):
SettingsApplication = "testium_choices_dlg_" + args[0]
SettingsLastChoices = "last_choice"
success = True
app = QApplication()
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = ChoicesDialog()
d.setFixedSize(800, 600)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
@@ -205,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

@@ -0,0 +1,15 @@
"""Qt platform environment setup for dialog subprocesses.
Call setup() at the start of every dialog subprocess main() function
to ensure the correct Qt platform plugin is selected.
"""
import sys
import os
def setup():
"""Configure the Qt environment for dialog subprocess usage."""
if sys.platform.startswith('linux'):
# On Linux/Wayland, force X11 (via XWayland) to avoid crashes
# when Qt is initialized inside a multiprocessing subprocess.
os.environ['QT_QPA_PLATFORM'] = 'xcb'

View File

@@ -1,7 +1,6 @@
import sys
import os
from PySide6.QtCore import (Qt)
from PySide6.QtCore import Qt, QTimer
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6 import (QtGui)
@@ -18,7 +17,9 @@ class TestDialogWindow(QDialog, dialog_image_win.Ui_Dialog):
def main(args, conn):
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow()
d.setFixedSize(700,600)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
@@ -37,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

@@ -1,36 +1,39 @@
import sys
import os
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtWidgets import QMessageBox
from multiprocessing import freeze_support
def main(args):
app = QApplication(sys.argv)
reply = QMessageBox.information(None, args[0], args[1], QMessageBox.Ok)
if hasattr(sys, "frozen"):
#all standard streams are replaced by dummy one to avoid cx_freeze flushing bug.
class dummyStream:
''' dummyStream behaves like a stream but does nothing. '''
def __init__(self): pass
def write(self,data): pass
def read(self,data): pass
def flush(self): pass
def close(self): pass
# and now redirect all default streams to this dummyStream:
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()
if __name__ == '__main__':
main(sys.argv[1:])
import sys
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt, QTimer
def main(args):
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
msg = QMessageBox()
msg.setWindowFlags(Qt.WindowStaysOnTopHint)
msg.setWindowTitle(args[0])
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"):
class dummyStream:
def __init__(self): pass
def write(self, data): pass
def read(self, data): pass
def flush(self): pass
def close(self): pass
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()
if __name__ == '__main__':
main(sys.argv[1:])

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
@@ -14,13 +14,23 @@ class TestDialogWindow(QDialog, dialog_note_win.Ui_Dialog):
def main(args, conn=None):
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow()
d.setFixedSize(387,224)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
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

@@ -1,32 +1,43 @@
import sys
import os
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtWidgets import QMessageBox
from multiprocessing import freeze_support
def main(args, conn):
app = QApplication(sys.argv)
reply = QMessageBox.question(None, args[0], args[1], QMessageBox.Yes|QMessageBox.No)
conn.send(reply)
conn.close()
if hasattr(sys, "frozen"):
#all standard streams are replaced by dummy one to avoid cx_freeze flushing bug.
class dummyStream:
''' dummyStream behaves like a stream but does nothing. '''
def __init__(self): pass
def write(self,data): pass
def read(self,data): pass
def flush(self): pass
def close(self): pass
# and now redirect all default streams to this dummyStream:
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()
import sys
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt, QTimer
def main(args, conn):
try:
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
msg = QMessageBox()
msg.setWindowFlags(Qt.WindowStaysOnTopHint)
msg.setWindowTitle(args[0])
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:
print(f"dialog_question error: {e}", file=sys.stderr)
finally:
conn.close()
if hasattr(sys, "frozen"):
class dummyStream:
def __init__(self): pass
def write(self, data): pass
def read(self, data): pass
def flush(self): pass
def close(self): pass
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()

View File

@@ -39,7 +39,9 @@ class DialogSleepWindow(QDialog, dialog_sleep_win.Ui_SleepDialogWindow):
def main(args, conn=None):
success = True
app = QApplication(sys.argv)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = DialogSleepWindow()
d.setFixedSize(379,129)
d.setWindowFlags(Qt.WindowStaysOnTopHint)

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
@@ -15,7 +15,9 @@ class TestDialogWindow(QDialog, dialog_value_win.Ui_Dialog):
def main(args, conn=None):
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow()
d.setFixedSize(387,224)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
@@ -23,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

@@ -7,7 +7,7 @@ import libs.testium as tm
from interpreter.utils.params import TestItemParams
from interpreter.utils.constants import TestItemType as cst_type
from interpreter.utils.eval import eval_to_boolean, evaluate, post_evaluate
from lib.tum_except import ETUMSyntaxError
from lib.tum_except import ETUMSyntaxError, item_load_context
LOG_TEST_STOP = '<----- step "{}" finished'
LOG_TEST_START = '-----> step "{}" started'
@@ -101,6 +101,7 @@ class TestItem:
self.status_queue = status_queue
self._execute_on_stop = False
self._post_eval = None
self._store_result = None
self._expected_result = None
self._no_fail = None
self._is_stopped = False
@@ -131,11 +132,11 @@ class TestItem:
if s:
try:
self.skipped = eval_to_boolean(s)
except:
except Exception as e:
raise ETUMSyntaxError(
f"'{self.cmd()}' test item named '{self.name()}':\nskipped expresion can only be a static expression as it is evaluated during loading of TUM : {s}",
self.seqFilename(),
)
) from e
# This allow disabling test item directly by using its name inside param.yaml file
elif self._name in tm.gd("skipped_test_item", []):
self.skipped = True
@@ -155,6 +156,9 @@ class TestItem:
if "process_result" in dict_item:
self._post_eval = dict_item["process_result"]
if "store_result" in dict_item:
self._store_result = dict_item["store_result"]
if "expected_result" in dict_item:
self._expected_result = dict_item["expected_result"]
@@ -164,11 +168,13 @@ class TestItem:
self.banner = LOG_TEST_START.format(self._name)
self.footer = LOG_TEST_STOP.format(self._name)
except:
except ETUMSyntaxError:
raise
except Exception as e:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
f"The '{self.cmd()}' test item named '{self.name()}' has an unexpected loading error: {e}",
self.seqFilename(),
)
) from e
self.result = TestResult(self, TestValue.FAILURE, "Failure by default")
@@ -275,6 +281,9 @@ class TestItem:
self.process_result()
# expected_result treatment
self.result_expected()
# Store result in a global variable if requested (before no_fail so
# the real outcome is captured when result.value is None)
self.store_result()
# Case of the no_fail true parameter
self.process_no_fail()
@@ -317,6 +326,17 @@ class TestItem:
print(e)
self.result.set(TestValue.FAILURE, "Result processing failed")
def store_result(self):
if self._store_result is None:
return
var_name = self._prms.expanse(self._store_result)
if self.result.value is None:
value = str(self.result.test_result)
else:
value = self.result.value
tm.setgd(var_name, value)
print(f"Stored result in '$({var_name})': {value}")
def process_report(self, report_eval):
tm.print_debug(f"Export reported values:")
rep_eval = self._prms.expanse(report_eval)

View File

@@ -1,7 +1,7 @@
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import TestValue
from lib.tum_except import ETUMSyntaxError
from lib.tum_except import ETUMSyntaxError, item_load_context
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import evaluate
@@ -15,21 +15,16 @@ class TestItemCheckValue(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_CHECK
self.is_container = False
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._action_list = self._prms.getParamAll('steps', default=[], required=False)
if len(self._action_list) > 0:
tm.print_warn("'steps' argument of check test item is deprecated and is replaced by 'values'")
self._action_list += self._prms.getParamAll('values', default=[], required=False)
if len(self._action_list) <= 0:
raise ETUMSyntaxError(
f" The '{self.cmd()}' test item named '{self.name()}' must have a 'values' parameter",
f"Missing required 'values' parameter",
self.seqFilename()
)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' (a child of: '{self.parent().name()}') has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):

View File

@@ -1,50 +1,92 @@
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.test_items.dialog_choices_files import choices_dialog
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemChoicesDialog(TestItem):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_CHOICES_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_CHOICES_DLG
self.is_container = False
try:
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
)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' (a child of: '{self.parent().name()}') has a missing or wrong parameter",
self.seqFilename()
)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
choices = self._prms.expanse(self._choices)
icon = self._prms.expanse(self._default_icon)
parent_conn, child_conn = Pipe()
p = Process(
target=choices_dialog.main, args=([self.name(), q, choices, icon], child_conn)
)
p.start()
val, succ = parent_conn.recv()
p.join()
self.result.value = val
if succ:
# The result of the test item is put into the global dict
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
else:
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, str(val))
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
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
class TestItemChoicesDialog(TestItemDialogBase):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_CHOICES_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_CHOICES_DLG
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
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)
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
val, succ = result
self.result.value = val
if succ:
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
else:
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, str(val))

View File

@@ -0,0 +1,58 @@
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')
class TestItemDialogBase(TestItem):
"""Base class for test items that launch a Qt dialog in a subprocess."""
def _cleanup_process(self, p):
if p.is_alive():
p.terminate()
p.join(timeout=0.2)
if p.is_alive():
p.kill()
p.join()
def _run_dialog(self, target, args):
"""Launch target(args) in a subprocess with no return value.
Returns the subprocess exit code.
"""
p = _spawn_ctx.Process(target=target, args=(args,))
p.start()
while p.is_alive() and not self._is_stopped:
p.join(timeout=0.5)
self._cleanup_process(p)
return p.exitcode
def _run_dialog_with_result(self, target, args):
"""Launch target(args, child_conn) in a subprocess and return what it sends.
Returns the received value, or None if stopped or if the subprocess crashed.
"""
parent_conn, child_conn = _spawn_ctx.Pipe()
p = _spawn_ctx.Process(target=target, args=(args, child_conn))
p.start()
child_conn.close()
result = None
while p.is_alive() and not self._is_stopped:
if parent_conn.poll(0.5):
try:
result = parent_conn.recv()
except EOFError:
pass
break
self._cleanup_process(p)
return result

View File

@@ -1,71 +1,56 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.test_items.dialog_image_files import dialog_image
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError
class TestItemImageDialog(TestItem):
"""dialog_image item usage.
dialog_image name: Nice image, question: could you press the red button, filename: img.jpg
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_IMAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_IMAGE_DLG
self.is_container = False
try:
self._question = self._prms.getParam("question", required=True)
self._filename = self._prms.getParam("filename", required=True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(
os.path.dirname(ourpath), "dialog_image_files", "dialog_image.py"
)
q = self._prms.expanse(self._question)
image_path = self._prms.expanse(self._filename)
print("Image Displayed:\n" + q + "\n" + image_path)
if not os.path.isfile(image_path):
image_path = os.path.normpath(
os.path.join(tm.gd("test_directory"), image_path)
)
parent_conn, child_conn = Pipe()
p = Process(
target=dialog_image.main, args=([self.name(), q, image_path], child_conn)
)
p.start()
succ = parent_conn.recv()
p.join()
if succ:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__ == "__main__":
p = Process(target=test_dialog.main, args=(["bob", "bab"],))
p.start()
p.join()
import os
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
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
class TestItemImageDialog(TestItemDialogBase):
"""dialog_image item usage.
dialog_image name: Nice image, question: could you press the red button, filename: img.jpg
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_IMAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_IMAGE_DLG
self.is_container = False
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):
q = self._prms.expanse(self._question)
image_path = self._prms.expanse(self._filename)
print("Image Displayed:\n" + q + "\n" + image_path)
if not os.path.isfile(image_path):
image_path = os.path.normpath(
os.path.join(tm.gd("test_directory"), 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:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE)

View File

@@ -5,7 +5,7 @@ import time
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from lib.tum_except import ETUMSyntaxError
from lib.tum_except import ETUMSyntaxError, item_load_context
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
@@ -19,18 +19,13 @@ class TestItemLet(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_LET
self.is_container = False
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._values_list = self._prms.getParamAll('values', default=[], required=False)
if len(self._values_list) <= 0:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' must have a 'values' parameter",
f"Missing required 'values' parameter",
self.seqFilename(),
)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):

View File

@@ -4,7 +4,7 @@ import traceback
import pprint
import textwrap
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestValue
import libs.testium as tm
@@ -26,16 +26,11 @@ class TestItemLuaFunc(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_LUA_FUNCTION
self.is_container = False
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param")
self._context_id = self._prms.getParam("context_id", default=None, processed=False)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' (child of '{self.parent.name()}') has a missing or wrong parameter",
self.seqFilename(),
)
self._lua_func_proc = LuaFuncExecEngine(tm.gd("lua_bin", ""), api_request, 10)
def _get_engine(self):

View File

@@ -1,54 +1,46 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
from interpreter.test_items.dialog_msg_files import msg_dialog
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError
class TestItemMsgDialog(TestItem):
"""dialog_message item usage.
dialog_message name: Nice message, question: Open the door and press OK
"""
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_MESSAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_MESSAGE_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_msg_files',
'msg_dialog.py')
q = self._prms.expanse(self._question)
print("Message Displayed:\n" + q)
parent_conn, child_conn = Pipe()
p=Process(target=msg_dialog.main,
args=([self.name(), q],))
p.start()
p.join()
self.result.set(TestValue.SUCCESS)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=msg_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
import os
import sys
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
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
class TestItemMsgDialog(TestItemDialogBase):
"""dialog_message item usage.
dialog_message name: Nice message, question: Open the door and press OK
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_MESSAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_MESSAGE_DLG
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)
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:
self.result.set(TestValue.FAILURE, f"Dialog subprocess exited with code {exitcode}")

View File

@@ -1,62 +1,75 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.dialog_note_files import test_dialog
from lib.tum_except import ETUMSyntaxError
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
class TestItemNoteDialog(TestItem):
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_NOTE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_NOTE_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_note_files',
'test_dialog.py')
q = self._prms.expanse(self._question)
print("Question:\n" + q)
parent_conn, child_conn = Pipe()
p=Process(target=test_dialog.main, args=([self.name(), q],child_conn))
p.start()
val, succ = parent_conn.recv()
p.join()
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 succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
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
class TestItemNoteDialog(TestItemDialogBase):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_NOTE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_NOTE_DLG
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)
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
val, succ = result
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 succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)

View File

@@ -4,7 +4,7 @@ import time
import pprint
import textwrap
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestValue
import libs.testium as tm
@@ -26,16 +26,11 @@ class TestItemPyFunc(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_PY_FUNCTION
self.is_container = False
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param")
self._context_id = self._prms.getParam("context_id", default=None, processed=False)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' (child of '{self.parent.name()}') has a missing or wrong parameter",
self.seqFilename(),
)
self._py_func_proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
def _get_engine(self):

View File

@@ -1,62 +1,58 @@
import os
import sys
from multiprocessing import Process, Pipe
from PySide6.QtWidgets import QMessageBox
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.dialog_question_files import question_dialog
from lib.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemQuestionDialog(TestItem):
"""dialog_question item usage.
dialog_question name: Nice question, question: "If OK, press OK, If not, press cancel"
"""
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_QUESTION_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_QUESTION_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_question_files',
'question_dialog.py')
q = self._prms.expanse(self._question)
print('Question asked:\n' + q + '\n')
parent_conn, child_conn = Pipe()
p=Process(target=question_dialog.main,
args=([self.name(), q],child_conn))
p.start()
succ = parent_conn.recv()
p.join()
if succ == QMessageBox.Yes:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')
else:
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
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
class TestItemQuestionDialog(TestItemDialogBase):
"""dialog_question item usage.
dialog_question name: Nice question, question: "If OK, press OK, If not, press cancel"
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_QUESTION_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_QUESTION_DLG
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')
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')
else:
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')

View File

@@ -10,7 +10,7 @@ from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
def nowInBetween(start, end):
@@ -30,8 +30,8 @@ class TestItemRun(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_RUN
self.is_container = False
try:
self.tum_fime = self._prms.getParam('tum_fime', required=True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
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='')
@@ -40,47 +40,46 @@ 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')
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@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 = ' ')
@@ -95,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

@@ -4,7 +4,7 @@ import traceback
from functools import wraps
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError
from lib.tum_except import ETUMSyntaxError, item_load_context
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.test_items.item_actions import TestItemActions
@@ -108,17 +108,12 @@ class TestItemPlotActionPeriodic(TestItemPlotAction):
)
# Periodic function call
try:
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.period = self._prms.getParam("period", required=True)
self.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param")
self.post_eval = self._prms.getParam("eval", default="")
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' 'periodic' action settings syntax error",
self.seqFilename(),
)
@test_run
def execute(self):

View File

@@ -3,11 +3,11 @@ 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
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
class TestItemSleep(TestItem):
"""sleep item usage.
@@ -19,14 +19,9 @@ class TestItemSleep(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_SLEEP
self.is_container = False
try:
self._timeout = self._prms.getParam('timeout', required = True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._timeout = self._prms.getParam('timeout', required=True)
self._has_dialog = self._prms.getParam('dialog', default=False)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
@@ -48,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,77 +1,78 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.tested_references_files import tested_refs_dialog
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemTestedRefsDialog(TestItem):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_REFERENCE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_REFERENCE_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required=True)
self._init_values = self._prms.getParamAll('reference', required=False, processed=True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath=__file__
test_file=os.path.join(os.path.dirname(ourpath),
'tested_references_files',
'tested_refs_dialog.py')
q=self._prms.expanse(self._question)
parent_conn, child_conn=Pipe()
init_values=','.join(self._init_values)
p=Process(target=tested_refs_dialog.main,
args=([self.name(), q, init_values],
child_conn))
p.start()
val, succ=parent_conn.recv()
p.join()
titems=[]
if len(val) > 0:
i = 0
for sitem in val.split(','):
titem={}
telems=sitem.split('/')
titem['reference']=telems[0]
titem['revision']=telems[1]
titem['serial']=telems[2]
print("Identification:\n" + str(titem))
titems.append(titem)
self.result.reported = {'reference_{}'.format(i): titem}
i = i + 1
self.result.value = titems
tm.setgd('tested_items', titems)
if len(val) > 0:
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__ == '__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
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
class TestItemTestedRefsDialog(TestItemDialogBase):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_REFERENCE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_REFERENCE_DLG
self.is_container = False
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)
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
val, succ = result
titems = []
if len(val) > 0:
i = 0
for sitem in val.split(','):
titem = {}
telems = sitem.split('/')
titem['reference'] = telems[0]
titem['revision'] = telems[1]
titem['serial'] = telems[2]
print("Identification:\n" + str(titem))
titems.append(titem)
self.result.reported = {'reference_{}'.format(i): titem}
i += 1
self.result.value = titems
tm.setgd('tested_items', titems)
if len(val) > 0:
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')

View File

@@ -1,67 +1,74 @@
import os
import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.dialog_value_files import test_dialog
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemValueDialog(TestItem):
"""dialog_value item usage.
dialog_value name: Enter value, question: "Which value did you measure?"
"""
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_VALUE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_VALUE_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
self._default = self._prms.getParam('default', '')
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_value_files',
'test_dialog.py')
q = self._prms.expanse(self._question)
d = self._prms.expanse(self._default)
print("Question:\n" + q)
parent_conn, child_conn = Pipe()
p=Process(target=test_dialog.main, args=([self.name(), q, d],child_conn))
p.start()
val, succ = parent_conn.recv()
p.join()
tm.setgd(self.name(), val)
print("Answer: " + val)
if len(val) > 0:
self.result.reported = {'question': q, 'answer': val}
self.result.value = val
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
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
class TestItemValueDialog(TestItemDialogBase):
"""dialog_value item usage.
dialog_value name: Enter value, question: "Which value did you measure?"
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_VALUE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_VALUE_DLG
self.is_container = False
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)
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
val, succ = result
tm.setgd(self.name(), val)
print("Answer: " + val)
if len(val) > 0:
self.result.reported = {'question': q, 'answer': val}
self.result.value = val
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')

View File

@@ -1,9 +1,8 @@
import sys
import os
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
@@ -20,7 +19,9 @@ def main(args, conn=None):
SettingsApplication = 'testium_ref_item'
SettingsLastReference = 'lastReference'
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestedRefsWindow()
d.setFixedSize(481,386)
d.setWindowFlags(Qt.WindowStaysOnTopHint)
@@ -51,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

@@ -3,9 +3,7 @@ import datetime
from queue import Queue
from interpreter.utils.params import expanse
import libs.testium as tm
from lib.tum_except import (
ETUMSyntaxError,
)
from lib.tum_except import ETUMSyntaxError
import interpreter.utils.settings as prefs
from interpreter.test_report.test_report import TestReport
from interpreter.utils.py_func_exec import PyFuncExecEngine
@@ -19,6 +17,17 @@ from interpreter.test_items.item_actions import TestItemActions
from interpreter.test_items.test_result import TestValue
def _build_item_path(item) -> str:
"""Build a breadcrumb path like 'main > Group > sub-group' from an item to root."""
parts = []
current = item
while current is not None:
name = current.name()
parts.append(name if name else f"[{current.type()}]")
current = current.parent()
return " > ".join(reversed(parts))
class TestSet:
def __init__(
self,
@@ -479,12 +488,19 @@ class TestSet:
action_name = cst.FOLDED_CHAR + it.item_cmd
seq_filename = action[action_name]["seq_filename"]
item = (it.item_class)(
action[action_name],
tree_parent,
self.status_queue,
filename=seq_filename
)
try:
item = (it.item_class)(
action[action_name],
tree_parent,
self.status_queue,
filename=seq_filename
)
except ETUMSyntaxError as e:
path = _build_item_path(tree_parent)
raise ETUMSyntaxError(
f"In: {path}\n{e._message}",
e._file or seq_filename,
) from e
item.is_folded = is_folded
child = {}
# case where the test item loads itself its descendants

View File

@@ -1,3 +1,4 @@
import json
from threading import Lock
@@ -5,6 +6,30 @@ global_dict = {}
global_dict_lock = Lock()
_update_queue = None
def set_update_queue(q):
global _update_queue
_update_queue = q
def _push_update(key, value):
if _update_queue is None or key.startswith("_"):
return
try:
json.dumps(value)
_update_queue.put({"type": "gd_update", "key": key, "value": value})
except (TypeError, ValueError):
pass
def _push_delete(key):
if _update_queue is None or key.startswith("_"):
return
_update_queue.put({"type": "gd_delete", "key": key})
# Global dictionnary helper functions
def gd(name, default=None):
''' Function which returns a variable from the global dictionary of testium
@@ -31,6 +56,7 @@ def setgd(name, value):
'''
with global_dict_lock:
global_dict.update({name: value})
_push_update(name, value)
def delgd(name):
''' Function which removes a variable from the global dictionary of testium
@@ -44,6 +70,7 @@ def delgd(name):
del global_dict[name]
except:
pass
_push_delete(name)
def cleargd():
with global_dict_lock:

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

@@ -1,11 +1,16 @@
import ast
import json
import os
import sys
import subprocess
import re
import subprocess
import sys
from PySide6.QtWidgets import QDialog
from PySide6.QtWidgets import (
QDialog, QDialogButtonBox, QHeaderView, QMenu, QMessageBox,
QPushButton, QTextEdit, QVBoxLayout,
)
from PySide6.QtGui import QSyntaxHighlighter, QTextCharFormat, QColor, QFont, QDesktopServices
from PySide6.QtCore import Qt, QUrl
from PySide6.QtCore import Qt, QUrl, Slot
from main_win.f1_win.f1_win_core import Ui_F1Dialog
@@ -16,58 +21,253 @@ class YamlHighlighter(QSyntaxHighlighter):
self.highlightingRules = []
# --- KEY formatting (before colon) ---
key_format = QTextCharFormat()
key_format.setForeground(QColor("#268bd2")) # Solarized blue
key_format.setForeground(QColor("#268bd2"))
key_format.setFontWeight(QFont.Bold)
self.highlightingRules.append((r"^\s*[^:]+(?=:)", key_format))
# --- VALUE formatting (strings) ---
value_format = QTextCharFormat()
value_format.setForeground(QColor("#2aa198")) # teal
value_format.setForeground(QColor("#2aa198"))
self.highlightingRules.append((r":\s*[^#\n]+", value_format))
# --- Booleans (true/false) ---
bool_format = QTextCharFormat()
bool_format.setForeground(QColor("#b58900")) # yellow
bool_format.setForeground(QColor("#b58900"))
bool_format.setFontWeight(QFont.Bold)
self.highlightingRules.append((r"\b(true|false)\b", bool_format))
# --- Numbers ---
num_format = QTextCharFormat()
num_format.setForeground(QColor("#d33682")) # magenta
num_format.setForeground(QColor("#d33682"))
self.highlightingRules.append((r"\b[0-9]+\b", num_format))
# --- Comments (# ...) ---
comment_format = QTextCharFormat()
comment_format.setForeground(QColor("#586e75")) # gray
comment_format.setForeground(QColor("#586e75"))
self.highlightingRules.append((r"#.*", comment_format))
def highlightBlock(self, text):
for pattern, fmt in self.highlightingRules:
for match in re.finditer(pattern, text):
start, end = match.span()
self.setFormat(start, end-start, fmt)
self.setFormat(start, end - start, fmt)
class GdVarEditDialog(QDialog):
"""JSON editor dialog for dict/list values."""
def __init__(self, key, value, parent=None):
super().__init__(parent)
self.setWindowTitle(f"Edit: {key}")
self.result_value = None
layout = QVBoxLayout(self)
self._edit = QTextEdit()
self._edit.setPlainText(json.dumps(value, indent=2))
font = QFont("Monospace")
font.setStyleHint(QFont.StyleHint.TypeWriter)
font.setPointSize(9)
self._edit.setFont(font)
layout.addWidget(self._edit)
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel)
buttons.accepted.connect(self._on_ok)
buttons.rejected.connect(self.reject)
layout.addWidget(buttons)
self.resize(400, 300)
def _on_ok(self):
try:
self.result_value = json.loads(self._edit.toPlainText())
self.accept()
except json.JSONDecodeError as e:
QMessageBox.warning(self, "Invalid JSON", str(e))
class DialogF1(QDialog):
def __init__(self, parent = None):
def __init__(self, parent=None):
super().__init__(parent)
self.ui = Ui_F1Dialog()
self.ui.setupUi(self)
self.highlighter = YamlHighlighter(self.ui.TestContentEdit.document())
self.setWindowFlags(
Qt.Window | Qt.WindowStaysOnTopHint | Qt.Tool
)
self.setWindowFlags(Qt.Window | Qt.WindowStaysOnTopHint | Qt.Tool)
self.ui.ButtLocOpen.clicked.connect(self.on_butlocopen_click)
self.ui.ButtClose.clicked.connect(self.close)
self._service = None
self._key_rows = {}
self._updating = False
self._mono_font = QFont("Monospace")
self._mono_font.setStyleHint(QFont.StyleHint.TypeWriter)
self._mono_bold_font = QFont("Monospace")
self._mono_bold_font.setStyleHint(QFont.StyleHint.TypeWriter)
self._mono_bold_font.setBold(True)
self._setup_vars_tab()
def _setup_vars_tab(self):
table = self.ui.varsTable
table.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.ResizeToContents)
table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch)
table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.Fixed)
table.setColumnWidth(2, 36)
table.verticalHeader().setVisible(False)
table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
table.customContextMenuRequested.connect(self._on_context_menu)
table.cellChanged.connect(self._on_cell_changed)
table.setEnabled(False)
self.ui.addVarButton.setEnabled(False)
self.ui.addVarButton.clicked.connect(self._on_add_var)
def load_initial_vars(self, vars_dict: dict):
for key, value in vars_dict.items():
self.gd_var_updated(key, value)
def set_service(self, service):
self._service = service
enabled = service is not None
self.ui.varsTable.setEnabled(enabled)
self.ui.addVarButton.setEnabled(enabled)
if not enabled:
self._updating = True
try:
self.ui.varsTable.setRowCount(0)
finally:
self._updating = False
self._key_rows.clear()
@Slot(str, object)
def gd_var_updated(self, key, value):
if key in self._key_rows:
self._refresh_row(self._key_rows[key], key, value)
else:
self._updating = True
try:
row = self.ui.varsTable.rowCount()
self.ui.varsTable.insertRow(row)
finally:
self._updating = False
self._key_rows[key] = row
self._refresh_row(row, key, value)
@Slot(str)
def gd_var_deleted(self, key):
if key not in self._key_rows:
return
row = self._key_rows.pop(key)
self._updating = True
try:
self.ui.varsTable.removeRow(row)
finally:
self._updating = False
self._key_rows = {k: (r - 1 if r > row else r) for k, r in self._key_rows.items()}
def _refresh_row(self, row, key, value):
from PySide6.QtWidgets import QTableWidgetItem
self._updating = True
try:
table = self.ui.varsTable
key_item = QTableWidgetItem(key)
key_item.setFlags(key_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
key_item.setFont(self._mono_bold_font)
table.setItem(row, 0, key_item)
display = self._display_value(value)
val_item = QTableWidgetItem(display)
val_item.setData(Qt.ItemDataRole.UserRole, value)
val_item.setToolTip(self._full_tooltip(value))
val_item.setFont(self._mono_font)
if self._is_complex(value):
val_item.setFlags(val_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
table.setItem(row, 1, val_item)
if self._is_complex(value):
btn = QPushButton("[…]")
captured_key = key
btn.clicked.connect(lambda: self._on_edit_complex(captured_key))
table.setCellWidget(row, 2, btn)
else:
table.setCellWidget(row, 2, None)
table.setItem(row, 2, QTableWidgetItem())
finally:
self._updating = False
def _is_complex(self, value):
return isinstance(value, (dict, list))
def _display_value(self, value):
if self._is_complex(value):
text = repr(value)
return (text[:60] + "") if len(text) > 60 else text
return repr(value)
def _full_tooltip(self, value):
try:
text = json.dumps(value, indent=2)
except (TypeError, ValueError):
text = repr(value)
escaped = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
return f"<pre>{escaped}</pre>"
def _on_cell_changed(self, row, col):
if self._updating or col != 1 or self._service is None:
return
from PySide6.QtWidgets import QTableWidgetItem
key_item = self.ui.varsTable.item(row, 0)
val_item = self.ui.varsTable.item(row, 1)
if key_item is None or val_item is None:
return
key = key_item.text()
text = val_item.text()
try:
value = ast.literal_eval(text)
except (ValueError, SyntaxError):
value = text
self._service.set_gd_var(key, value)
def _on_edit_complex(self, key):
if key not in self._key_rows:
return
val_item = self.ui.varsTable.item(self._key_rows[key], 1)
if val_item is None:
return
value = val_item.data(Qt.ItemDataRole.UserRole)
dlg = GdVarEditDialog(key, value, self)
if dlg.exec() == QDialog.DialogCode.Accepted and self._service is not None:
self._service.set_gd_var(key, dlg.result_value)
def _on_add_var(self):
key = self.ui.newKeyEdit.text().strip()
value_text = self.ui.newValueEdit.text().strip()
if not key or self._service is None:
return
try:
value = ast.literal_eval(value_text)
except (ValueError, SyntaxError):
value = value_text
self._service.set_gd_var(key, value)
self.ui.newKeyEdit.clear()
self.ui.newValueEdit.clear()
def _on_context_menu(self, pos):
row = self.ui.varsTable.rowAt(pos.y())
if row < 0:
return
key_item = self.ui.varsTable.item(row, 0)
if key_item is None or self._service is None:
return
key = key_item.text()
menu = QMenu(self)
delete_action = menu.addAction("Delete")
if menu.exec(self.ui.varsTable.mapToGlobal(pos)) == delete_action:
self._service.del_gd_var(key)
def on_butlocopen_click(self):
file = self.ui.sequenceFileNameLineEdit.text()
if os.path.exists(file):
if sys.platform.startswith("win"): # Windows
if sys.platform.startswith("win"):
subprocess.Popen(f'explorer "{file}"')
else: # Linux / autres
else:
subprocess.Popen(["xdg-open", file])
QDesktopServices.openUrl(QUrl.fromLocalFile(file))
QDesktopServices.openUrl(QUrl.fromLocalFile(file))

View File

@@ -3,7 +3,7 @@
################################################################################
## Form generated from reading UI file 'f1_win_core.ui'
##
## Created by: Qt User Interface Compiler version 6.10.1
## Created by: Qt User Interface Compiler version 6.11.0
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################
@@ -16,8 +16,9 @@ from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor,
QImage, QKeySequence, QLinearGradient, QPainter,
QPalette, QPixmap, QRadialGradient, QTransform)
from PySide6.QtWidgets import (QApplication, QDialog, QFormLayout, QHBoxLayout,
QLabel, QLineEdit, QPushButton, QSizePolicy,
QSpacerItem, QTextEdit, QToolButton, QVBoxLayout,
QHeaderView, QLabel, QLineEdit, QPushButton,
QSizePolicy, QSpacerItem, QTabWidget, QTableWidget,
QTableWidgetItem, QTextEdit, QToolButton, QVBoxLayout,
QWidget)
import f1_win_rc
@@ -25,7 +26,7 @@ class Ui_F1Dialog(object):
def setupUi(self, F1Dialog):
if not F1Dialog.objectName():
F1Dialog.setObjectName(u"F1Dialog")
F1Dialog.resize(400, 300)
F1Dialog.resize(550, 450)
icon = QIcon()
if QIcon.hasThemeIcon(QIcon.ThemeIcon.HelpAbout):
icon = QIcon.fromTheme(QIcon.ThemeIcon.HelpAbout)
@@ -36,19 +37,20 @@ class Ui_F1Dialog(object):
F1Dialog.setLayoutDirection(Qt.LayoutDirection.LeftToRight)
self.verticalLayout_2 = QVBoxLayout(F1Dialog)
self.verticalLayout_2.setObjectName(u"verticalLayout_2")
self.horizontalLayout_2 = QHBoxLayout()
self.horizontalLayout_2.setObjectName(u"horizontalLayout_2")
self.verticalLayout_2.addLayout(self.horizontalLayout_2)
self.tabWidget = QTabWidget(F1Dialog)
self.tabWidget.setObjectName(u"tabWidget")
self.tabTestItem = QWidget()
self.tabTestItem.setObjectName(u"tabTestItem")
self.verticalLayout_tab0 = QVBoxLayout(self.tabTestItem)
self.verticalLayout_tab0.setObjectName(u"verticalLayout_tab0")
self.formLayout = QFormLayout()
self.formLayout.setObjectName(u"formLayout")
self.typeLabel = QLabel(F1Dialog)
self.typeLabel = QLabel(self.tabTestItem)
self.typeLabel.setObjectName(u"typeLabel")
self.formLayout.setWidget(0, QFormLayout.ItemRole.LabelRole, self.typeLabel)
self.typeLineEdit = QLineEdit(F1Dialog)
self.typeLineEdit = QLineEdit(self.tabTestItem)
self.typeLineEdit.setObjectName(u"typeLineEdit")
sizePolicy = QSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
sizePolicy.setHorizontalStretch(0)
@@ -59,20 +61,20 @@ class Ui_F1Dialog(object):
self.formLayout.setWidget(0, QFormLayout.ItemRole.FieldRole, self.typeLineEdit)
self.sequenceFileNameLabel = QLabel(F1Dialog)
self.sequenceFileNameLabel = QLabel(self.tabTestItem)
self.sequenceFileNameLabel.setObjectName(u"sequenceFileNameLabel")
self.formLayout.setWidget(1, QFormLayout.ItemRole.LabelRole, self.sequenceFileNameLabel)
self.horizontalLayout_3 = QHBoxLayout()
self.horizontalLayout_3.setObjectName(u"horizontalLayout_3")
self.sequenceFileNameLineEdit = QLineEdit(F1Dialog)
self.sequenceFileNameLineEdit = QLineEdit(self.tabTestItem)
self.sequenceFileNameLineEdit.setObjectName(u"sequenceFileNameLineEdit")
self.sequenceFileNameLineEdit.setReadOnly(True)
self.horizontalLayout_3.addWidget(self.sequenceFileNameLineEdit)
self.ButtLocOpen = QToolButton(F1Dialog)
self.ButtLocOpen = QToolButton(self.tabTestItem)
self.ButtLocOpen.setObjectName(u"ButtLocOpen")
self.horizontalLayout_3.addWidget(self.ButtLocOpen)
@@ -81,18 +83,61 @@ class Ui_F1Dialog(object):
self.formLayout.setLayout(1, QFormLayout.ItemRole.FieldRole, self.horizontalLayout_3)
self.verticalLayout_2.addLayout(self.formLayout)
self.verticalLayout_tab0.addLayout(self.formLayout)
self.label = QLabel(F1Dialog)
self.label = QLabel(self.tabTestItem)
self.label.setObjectName(u"label")
self.verticalLayout_2.addWidget(self.label)
self.verticalLayout_tab0.addWidget(self.label)
self.TestContentEdit = QTextEdit(F1Dialog)
self.TestContentEdit = QTextEdit(self.tabTestItem)
self.TestContentEdit.setObjectName(u"TestContentEdit")
self.TestContentEdit.setReadOnly(True)
self.verticalLayout_2.addWidget(self.TestContentEdit)
self.verticalLayout_tab0.addWidget(self.TestContentEdit)
self.tabWidget.addTab(self.tabTestItem, "")
self.tabVariables = QWidget()
self.tabVariables.setObjectName(u"tabVariables")
self.verticalLayout_tab1 = QVBoxLayout(self.tabVariables)
self.verticalLayout_tab1.setObjectName(u"verticalLayout_tab1")
self.varsTable = QTableWidget(self.tabVariables)
if (self.varsTable.columnCount() < 3):
self.varsTable.setColumnCount(3)
__qtablewidgetitem = QTableWidgetItem()
self.varsTable.setHorizontalHeaderItem(0, __qtablewidgetitem)
__qtablewidgetitem1 = QTableWidgetItem()
self.varsTable.setHorizontalHeaderItem(1, __qtablewidgetitem1)
__qtablewidgetitem2 = QTableWidgetItem()
self.varsTable.setHorizontalHeaderItem(2, __qtablewidgetitem2)
self.varsTable.setObjectName(u"varsTable")
self.verticalLayout_tab1.addWidget(self.varsTable)
self.addVarLayout = QHBoxLayout()
self.addVarLayout.setObjectName(u"addVarLayout")
self.newKeyEdit = QLineEdit(self.tabVariables)
self.newKeyEdit.setObjectName(u"newKeyEdit")
self.addVarLayout.addWidget(self.newKeyEdit)
self.newValueEdit = QLineEdit(self.tabVariables)
self.newValueEdit.setObjectName(u"newValueEdit")
self.addVarLayout.addWidget(self.newValueEdit)
self.addVarButton = QPushButton(self.tabVariables)
self.addVarButton.setObjectName(u"addVarButton")
self.addVarButton.setMaximumSize(QSize(30, 16777215))
self.addVarLayout.addWidget(self.addVarButton)
self.verticalLayout_tab1.addLayout(self.addVarLayout)
self.tabWidget.addTab(self.tabVariables, "")
self.verticalLayout_2.addWidget(self.tabWidget)
self.horizontalLayout = QHBoxLayout()
self.horizontalLayout.setObjectName(u"horizontalLayout")
@@ -113,6 +158,9 @@ class Ui_F1Dialog(object):
self.retranslateUi(F1Dialog)
self.tabWidget.setCurrentIndex(0)
QMetaObject.connectSlotsByName(F1Dialog)
# setupUi
@@ -122,6 +170,15 @@ class Ui_F1Dialog(object):
self.sequenceFileNameLabel.setText(QCoreApplication.translate("F1Dialog", u"Test file name", None))
self.ButtLocOpen.setText(QCoreApplication.translate("F1Dialog", u"...", None))
self.label.setText(QCoreApplication.translate("F1Dialog", u"Test content:", None))
self.tabWidget.setTabText(self.tabWidget.indexOf(self.tabTestItem), QCoreApplication.translate("F1Dialog", u"Test item", None))
___qtablewidgetitem = self.varsTable.horizontalHeaderItem(0)
___qtablewidgetitem.setText(QCoreApplication.translate("F1Dialog", u"Key", None))
___qtablewidgetitem1 = self.varsTable.horizontalHeaderItem(1)
___qtablewidgetitem1.setText(QCoreApplication.translate("F1Dialog", u"Value", None))
self.newKeyEdit.setPlaceholderText(QCoreApplication.translate("F1Dialog", u"New key", None))
self.newValueEdit.setPlaceholderText(QCoreApplication.translate("F1Dialog", u"Value", None))
self.addVarButton.setText(QCoreApplication.translate("F1Dialog", u"+", None))
self.tabWidget.setTabText(self.tabWidget.indexOf(self.tabVariables), QCoreApplication.translate("F1Dialog", u"Variables", None))
self.ButtClose.setText(QCoreApplication.translate("F1Dialog", u"Close", None))
# retranslateUi

View File

@@ -6,8 +6,8 @@
<rect>
<x>0</x>
<y>0</y>
<width>400</width>
<height>300</height>
<width>550</width>
<height>450</height>
</rect>
</property>
<property name="windowTitle">
@@ -22,69 +22,139 @@
</property>
<layout class="QVBoxLayout" name="verticalLayout_2">
<item>
<layout class="QHBoxLayout" name="horizontalLayout_2"/>
</item>
<item>
<layout class="QFormLayout" name="formLayout">
<item row="0" column="0">
<widget class="QLabel" name="typeLabel">
<property name="text">
<string>Test step type</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QLineEdit" name="typeLineEdit">
<property name="sizePolicy">
<sizepolicy hsizetype="Fixed" vsizetype="Fixed">
<horstretch>0</horstretch>
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
<item row="1" column="0">
<widget class="QLabel" name="sequenceFileNameLabel">
<property name="text">
<string>Test file name</string>
</property>
</widget>
</item>
<item row="1" column="1">
<layout class="QHBoxLayout" name="horizontalLayout_3">
<widget class="QTabWidget" name="tabWidget">
<property name="currentIndex">
<number>0</number>
</property>
<!-- Tab 0: Test item -->
<widget class="QWidget" name="tabTestItem">
<attribute name="title">
<string>Test item</string>
</attribute>
<layout class="QVBoxLayout" name="verticalLayout_tab0">
<item>
<widget class="QLineEdit" name="sequenceFileNameLineEdit">
<layout class="QFormLayout" name="formLayout">
<item row="0" column="0">
<widget class="QLabel" name="typeLabel">
<property name="text">
<string>Test step type</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QLineEdit" name="typeLineEdit">
<property name="sizePolicy">
<sizepolicy hsizetype="Fixed" vsizetype="Fixed">
<horstretch>0</horstretch>
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
<item row="1" column="0">
<widget class="QLabel" name="sequenceFileNameLabel">
<property name="text">
<string>Test file name</string>
</property>
</widget>
</item>
<item row="1" column="1">
<layout class="QHBoxLayout" name="horizontalLayout_3">
<item>
<widget class="QLineEdit" name="sequenceFileNameLineEdit">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
<item>
<widget class="QToolButton" name="ButtLocOpen">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
</layout>
</item>
</layout>
</item>
<item>
<widget class="QLabel" name="label">
<property name="text">
<string>Test content:</string>
</property>
</widget>
</item>
<item>
<widget class="QTextEdit" name="TestContentEdit">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
</layout>
</widget>
<!-- Tab 1: Variables -->
<widget class="QWidget" name="tabVariables">
<attribute name="title">
<string>Variables</string>
</attribute>
<layout class="QVBoxLayout" name="verticalLayout_tab1">
<item>
<widget class="QToolButton" name="ButtLocOpen">
<property name="text">
<string>...</string>
</property>
<widget class="QTableWidget" name="varsTable">
<column>
<property name="text">
<string>Key</string>
</property>
</column>
<column>
<property name="text">
<string>Value</string>
</property>
</column>
<column>
<property name="text">
<string/>
</property>
</column>
</widget>
</item>
<item>
<layout class="QHBoxLayout" name="addVarLayout">
<item>
<widget class="QLineEdit" name="newKeyEdit">
<property name="placeholderText">
<string>New key</string>
</property>
</widget>
</item>
<item>
<widget class="QLineEdit" name="newValueEdit">
<property name="placeholderText">
<string>Value</string>
</property>
</widget>
</item>
<item>
<widget class="QPushButton" name="addVarButton">
<property name="text">
<string>+</string>
</property>
<property name="maximumSize">
<size>
<width>30</width>
<height>16777215</height>
</size>
</property>
</widget>
</item>
</layout>
</item>
</layout>
</item>
</layout>
</item>
<item>
<widget class="QLabel" name="label">
<property name="text">
<string>Test content:</string>
</property>
</widget>
</item>
<item>
<widget class="QTextEdit" name="TestContentEdit">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</widget>
</item>
<item>

View File

@@ -69,3 +69,12 @@ class TestControllerService:
def set_test_outputs(self, outputs: list) -> None:
self._ctrl.control("set_test_outputs", outputs=outputs)
def get_gd_vars(self) -> dict:
return self._ctrl.control("get_gd_vars")
def set_gd_var(self, name: str, value) -> None:
self._ctrl.control("set_gd_var", name=name, value=value)
def del_gd_var(self, name: str) -> None:
self._ctrl.control("del_gd_var", name=name)

View File

@@ -30,11 +30,18 @@ class TestFileManager:
):
w.test_service.stop()
w.test_service.close()
w.test_proc.join()
w.test_proc.join(timeout=5)
if w.test_proc.is_alive():
w.test_proc.terminate()
w.test_proc.join(timeout=2)
if w.test_proc.is_alive():
w.test_proc.kill()
w.test_proc.join()
del w.test_proc
w.test_proc = None
del w.test_service
w.test_service = None
w.d_f1_win.set_service(None)
del w.ts_controller
w.ts_controller = None
@@ -83,6 +90,7 @@ class TestFileManager:
w.testFile = None
w.ts_controller = TestSetController()
w.test_service = TestControllerService(w.ts_controller)
w.d_f1_win.set_service(w.test_service)
w.test_proc = TestProcess(
file_name,
w.status_queue,
@@ -106,6 +114,7 @@ class TestFileManager:
w.test_proc = None
del w.test_service
w.test_service = None
w.d_f1_win.set_service(None)
del w.ts_controller
w.ts_controller = None
raise ETUMRuntimeError(
@@ -122,6 +131,7 @@ class TestFileManager:
progress = None
w.treeTests.setFoldDefault()
w.treeTests.updateTreeSkipState(w.test_service)
w.d_f1_win.load_initial_vars(w.test_service.get_gd_vars())
w.checkSelect.setChecked(True)
w.testFile = file_name

View File

@@ -6,6 +6,8 @@ from PySide6.QtCore import (Signal, QThread)
class ThreadTestStatus(QThread):
statusToBeUpdated = Signal(dict)
testSetIsFinished = Signal()
gdUpdated = Signal(str, object)
gdDeleted = Signal(str)
def __init__(self, status_queue, parent=None, debug=False):
super().__init__(parent)
@@ -21,7 +23,12 @@ class ThreadTestStatus(QThread):
while True:
while not self._status_queue.empty():
m = self._status_queue.get()
if m.get("id", None) is None:
msg_type = m.get("type")
if msg_type == "gd_update":
self.gdUpdated.emit(m["key"], m["value"])
elif msg_type == "gd_delete":
self.gdDeleted.emit(m["key"])
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)
@@ -247,6 +244,8 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.threadTestStatus.testSetIsFinished.connect(self.runner.on_run_finished)
self.threadTestStatus.statusToBeUpdated.connect(self.treeTests.updateStatus)
self.threadTestStatus.gdUpdated.connect(self.d_f1_win.gd_var_updated)
self.threadTestStatus.gdDeleted.connect(self.d_f1_win.gd_var_deleted)
self.reconnect_signals()
if runandclose:
@@ -357,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:
@@ -683,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

@@ -92,13 +92,14 @@
func_name: echo
param: [ $(str_example) ]
process_result: "'44' in '$(result)'"
expected_result: True
- py_func:
name: Save the result in a global variable
key: $(test)_PASS
file: $(test_path)$(psep)results$(psep)results.py
func_name: echo
param: [ 44 ]
process_result: "tm.setgd('process_result_value', $(result))"
store_result: process_result_value
- py_func:
name: Check the saved global variable
key: $(test)_PASS
@@ -107,6 +108,68 @@
param: [ 44 ]
expected_result: $(process_result_value)
- py_func:
name: store_result with process_result
key: $(test)_PASS
file: $(test_path)$(psep)results$(psep)results.py
func_name: echo
param: [ $(str_example) ]
process_result: "'$(result)'.upper()"
store_result: upper_str_example
- py_func:
name: Check store_result with process_result
key: $(test)_PASS
file: $(test_path)$(psep)results$(psep)results.py
func_name: echo
param: [ $(str_example) ]
process_result: "'$(result)'.upper()"
expected_result: $(upper_str_example)
- let:
name: store_result on let item (None value → stores PASS)
key: $(test)_PASS
values:
- dummy: 0
store_result: let_store_result
- py_func:
name: Check store_result on let stores PASS
key: $(test)_PASS
file: $(test_path)$(psep)results$(psep)results.py
func_name: echo
param: [PASS]
expected_result: $(let_store_result)
- py_func:
name: store_result on failing test (None value → stores FAIL)
key: $(test)_FAIL
file: $(test_path)$(psep)results$(psep)results.py
func_name: return_none
expected_result: FAIL
store_result: none_fail_store_result
- py_func:
name: Check store_result on failing test stores FAIL
key: $(test)_PASS
file: $(test_path)$(psep)results$(psep)results.py
func_name: echo
param: [FAIL]
expected_result: $(none_fail_store_result)
- py_func:
name: store_result with no_fail (None value → stores real FAIL, not forced PASS)
key: $(test)_PASS
file: $(test_path)$(psep)results$(psep)results.py
func_name: return_none
expected_result: FAIL
no_fail: True
store_result: none_nofail_store_result
- py_func:
name: Check store_result with no_fail stores real FAIL
key: $(test)_PASS
file: $(test_path)$(psep)results$(psep)results.py
func_name: echo
param: [FAIL]
expected_result: $(none_nofail_store_result)
- py_func:
name: Process result when result is None (must fail)
key: $(test)_FAIL

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

@@ -0,0 +1,134 @@
#!/usr/bin/env python3
"""JSON-RPC echo server for the testium validation suite.
Listens on TCP (newline-delimited JSON) and UDP.
Supports JSON-RPC 1.0 and 2.0.
Handlers:
echo(*args) -> [args, {}]
<unknown> -> error {code: -32000, message: "function not found"}
Usage:
python3 jrpc_echo_server.py -c jrpces.ini
"""
import argparse
import configparser
import json
import socket
import sys
import threading
def _dispatch(method, params):
if method == "echo":
if not isinstance(params, list):
params = [params]
return True, [params, {}]
return False, {"code": -32000, "message": "function not found"}
def _build_response(req, success, data):
req_id = req.get("id", None)
if req.get("jsonrpc") == "2.0":
if success:
return {"jsonrpc": "2.0", "result": data, "id": req_id}
else:
return {"jsonrpc": "2.0", "error": data, "id": req_id}
else:
if success:
return {"result": data, "error": None, "id": req_id}
else:
return {"result": None, "error": data, "id": req_id}
def handle(raw: str) -> str:
try:
req = json.loads(raw)
method = req.get("method", "")
params = req.get("params", [])
success, data = _dispatch(method, params)
return json.dumps(_build_response(req, success, data))
except Exception as exc:
return json.dumps({"result": None, "error": {"code": -32700, "message": str(exc)}, "id": None})
# ── TCP ──────────────────────────────────────────────────────────────────────
def _handle_tcp_client(conn):
buf = b""
with conn:
conn.settimeout(5.0)
while True:
try:
chunk = conn.recv(4096)
except (socket.timeout, ConnectionResetError, OSError):
break
if not chunk:
break
buf += chunk
while b"\n" in buf:
line, buf = buf.split(b"\n", 1)
line = line.strip()
if line:
resp = handle(line.decode())
conn.sendall((resp + "\n").encode())
def _tcp_server(host, port):
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((host, port))
srv.listen(5)
srv.settimeout(1.0)
print(f"TCP listening on {host}:{port}", flush=True)
while True:
try:
conn, _ = srv.accept()
except socket.timeout:
continue
threading.Thread(target=_handle_tcp_client, args=(conn,), daemon=True).start()
# ── UDP ──────────────────────────────────────────────────────────────────────
def _udp_server(host, port):
srv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((host, port))
print(f"UDP listening on {host}:{port}", flush=True)
while True:
data, addr = srv.recvfrom(65535)
resp = handle(data.decode())
srv.sendto(resp.encode(), addr)
# ── Main ─────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(description="JSON-RPC echo server")
parser.add_argument("-c", "--config", required=True, help="Path to .ini config file")
args = parser.parse_args()
cfg = configparser.ConfigParser()
cfg.read(args.config)
tcp_host = cfg.get("jsonrpc_tcp", "host", fallback="0.0.0.0")
tcp_port = cfg.getint("jsonrpc_tcp", "port", fallback=4321)
udp_host = cfg.get("jsonrpc_udp", "host", fallback="0.0.0.0")
udp_port = cfg.getint("jsonrpc_udp", "port", fallback=4323)
tcp_thread = threading.Thread(target=_tcp_server, args=(tcp_host, tcp_port), daemon=True)
udp_thread = threading.Thread(target=_udp_server, args=(udp_host, udp_port), daemon=True)
tcp_thread.start()
udp_thread.start()
print("JSON-RPC echo server ready", flush=True)
try:
tcp_thread.join()
except KeyboardInterrupt:
sys.exit(0)
if __name__ == "__main__":
main()

View File

@@ -1,27 +1,27 @@
- console:
name: json rpc echo server
doc: check if the jsonrpc echo server is installed
doc: check if jrpc_echo_server.py is available
console_name: jrpces
key: $(test)_PASS
steps:
- open:
protocol: terminal
- read_until: {expected: $(terminal_prompt), timeout: 1, no_fail: True}
- writeln: which jrpces
- read_until: {expected: jrpces, timeout: 2}
- writeln: test -f {{include_directory}}/jrpc_echo_server.py && echo JRPC_OK
- read_until: {expected: JRPC_OK, timeout: 2, no_fail: True}
- group:
name: jsonrpc tests
condition: <| '/jrpces' in r'''$(cn_json rpc echo server)''' |>
condition: <| 'JRPC_OK' in r'''$(cn_json rpc echo server)''' |>
steps:
- console:
name: Start the json rpc echo server
console_name: jrpces
key: $(test)_PASS
steps:
- writeln: jrpces -c {{include_directory}}/jrpces.ini
- read_until: {expected: $(terminal_prompt), timeout: 1, no_fail: True}
- writeln: python3 {{include_directory}}/jrpc_echo_server.py -c {{include_directory}}/jrpces.ini
- read_until: {expected: ready, timeout: 5}
- console:
name: Open the raw tcp Console

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

View File

@@ -30,8 +30,10 @@ linux_prompt: "$ "
inc_no_template: "inc no template"
inc_with_template: "inc with template"
LUA_PATH_Linux: /usr/share/lua/5.4/?.lua;/usr/local/share/lua/5.4/?.lua;/usr/local/share/lua/5.4/?/init.lua;/usr/share/lua/5.4/?/init.lua;/usr/local/lib/lua/5.4/?.lua;/usr/local/lib/lua/5.4/?/init.lua;/usr/lib/lua/5.4/?.lua;/usr/lib/lua/5.4/?/init.lua;./?.lua;./?/init.lua;/home/francois/.luarocks/share/lua/5.4/?.lua;/home/francois/.luarocks/share/lua/5.4/?/init.lua
LUA_CPATH_Linux: /usr/local/lib/lua/5.4/?.so;/usr/lib/lua/5.4/?.so;/usr/local/lib/lua/5.4/loadall.so;/usr/lib/lua/5.4/loadall.so;./?.so;/home/francois/.luarocks/lib/lua/5.4/?.so
lua_rev: 5.5
LUA_PATH_Linux: /usr/share/lua/$(lua_rev)/?.lua;/usr/local/share/lua/$(lua_rev)/?.lua;/usr/local/share/lua/$(lua_rev)/?/init.lua;/usr/share/lua/$(lua_rev)/?/init.lua;/usr/local/lib/lua/$(lua_rev)/?.lua;/usr/local/lib/lua/$(lua_rev)/?/init.lua;/usr/lib/lua/$(lua_rev)/?.lua;/usr/lib/lua/$(lua_rev)/?/init.lua;./?.lua;./?/init.lua;/home/francois/.luarocks/share/lua/$(lua_rev)/?.lua;/home/francois/.luarocks/share/lua/$(lua_rev)/?/init.lua
LUA_CPATH_Linux: /usr/local/lib/lua/$(lua_rev)/?.so;/usr/lib/lua/$(lua_rev)/?.so;/usr/local/lib/lua/$(lua_rev)/loadall.so;/usr/lib/lua/$(lua_rev)/loadall.so;./?.so;/home/francois/.luarocks/lib/lua/$(lua_rev)/?.so
PATH_Linux:
LUA_PATH_Windows: ;.\?.lua;C:\Lua\5.1\lua\?.lua;C:\Lua\5.1\lua\?\init.lua;C:\Lua\5.1\?.lua;C:\Lua\5.1\?\init.lua;C:\Lua\5.1\lua\?.luac