Compare commits
20 Commits
feature/fu
...
text_no_py
| Author | SHA1 | Date | |
|---|---|---|---|
| 06c4cc62c6 | |||
| 60dbcf0252 | |||
| a3e449cc7d | |||
| 95107117fa | |||
| 88cc410eed | |||
| fa7f8cef7c | |||
| 5a065128be | |||
| b7b930aab1 | |||
| 609ca57202 | |||
| d26b60435b | |||
| de143b6cc3 | |||
| d955ae81f9 | |||
| 2cd3aa3305 | |||
| 276d485905 | |||
| 95912dd3e1 | |||
| 6d1fb6a6bc | |||
| 2cc42e9065 | |||
| 2b7678c39e | |||
| c72176d029 | |||
| 617f599f86 |
94
CLAUDE.md
Normal file
94
CLAUDE.md
Normal 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`.
|
||||
@@ -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 -----------------------------------------------------
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
-----------------------------------------------
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
@@ -4,6 +4,7 @@ SUPPORTED_API = [
|
||||
"setgd",
|
||||
"delgd",
|
||||
"add_plot_values",
|
||||
"last_plot_value"
|
||||
"last_plot_value",
|
||||
"text_mode",
|
||||
]
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
15
src/testium/interpreter/test_items/dialog_env.py
Normal file
15
src/testium/interpreter/test_items/dialog_env.py
Normal 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'
|
||||
@@ -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:
|
||||
|
||||
@@ -1,26 +1,32 @@
|
||||
import sys
|
||||
import os
|
||||
from multiprocessing import freeze_support
|
||||
|
||||
from PySide6.QtWidgets import (QApplication, QMessageBox)
|
||||
from PySide6.QtCore import Qt, QTimer
|
||||
|
||||
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)
|
||||
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"):
|
||||
#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 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()
|
||||
@@ -31,6 +37,3 @@ def main(args):
|
||||
|
||||
if __name__ == '__main__':
|
||||
main(sys.argv[1:])
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -1,29 +1,40 @@
|
||||
import sys
|
||||
import os
|
||||
from multiprocessing import freeze_support
|
||||
|
||||
from PySide6.QtWidgets import (QApplication, QMessageBox)
|
||||
from PySide6.QtCore import Qt, QTimer
|
||||
|
||||
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()
|
||||
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"):
|
||||
#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 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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -1,48 +1,90 @@
|
||||
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.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(TestItem):
|
||||
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
|
||||
try:
|
||||
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
|
||||
)
|
||||
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()
|
||||
)
|
||||
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)
|
||||
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()
|
||||
|
||||
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:
|
||||
# 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:
|
||||
|
||||
58
src/testium/interpreter/test_items/test_item_dialog_base.py
Normal file
58
src/testium/interpreter/test_items/test_item_dialog_base.py
Normal 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
|
||||
@@ -1,41 +1,29 @@
|
||||
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.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 ETUMSyntaxError
|
||||
from lib.tum_except import item_load_context
|
||||
import libs.testium as tm
|
||||
|
||||
|
||||
class TestItemImageDialog(TestItem):
|
||||
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
|
||||
try:
|
||||
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)
|
||||
except:
|
||||
raise ETUMSyntaxError(
|
||||
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
|
||||
self.seqFilename(),
|
||||
)
|
||||
self._auto_result = self._prms.getParam("auto_result", required=False, default=None)
|
||||
|
||||
@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)
|
||||
@@ -43,29 +31,26 @@ class TestItemImageDialog(TestItem):
|
||||
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:
|
||||
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)
|
||||
|
||||
|
||||
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()
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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.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 ETUMSyntaxError
|
||||
from lib.tum_except import item_load_context
|
||||
|
||||
class TestItemMsgDialog(TestItem):
|
||||
|
||||
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=""):
|
||||
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(),
|
||||
)
|
||||
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):
|
||||
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()
|
||||
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}")
|
||||
|
||||
@@ -1,42 +1,67 @@
|
||||
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.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(TestItem):
|
||||
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
|
||||
|
||||
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
|
||||
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(),
|
||||
)
|
||||
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):
|
||||
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()
|
||||
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")
|
||||
@@ -48,15 +73,3 @@ class TestItemNoteDialog(TestItem):
|
||||
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()
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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.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(TestItem):
|
||||
|
||||
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=""):
|
||||
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(),
|
||||
)
|
||||
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):
|
||||
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 _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')
|
||||
|
||||
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()
|
||||
|
||||
@@ -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')
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -1,59 +1,72 @@
|
||||
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.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(TestItem):
|
||||
|
||||
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
|
||||
try:
|
||||
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)
|
||||
except:
|
||||
raise ETUMSyntaxError(
|
||||
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
|
||||
self.seqFilename(),
|
||||
)
|
||||
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
|
||||
|
||||
@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)
|
||||
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
|
||||
|
||||
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=[]
|
||||
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]
|
||||
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
|
||||
i += 1
|
||||
self.result.value = titems
|
||||
tm.setgd('tested_items', titems)
|
||||
if len(val) > 0:
|
||||
@@ -63,15 +76,3 @@ class TestItemTestedRefsDialog(TestItem):
|
||||
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()
|
||||
|
||||
@@ -1,47 +1,66 @@
|
||||
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.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(TestItem):
|
||||
|
||||
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=""):
|
||||
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)
|
||||
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
|
||||
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(),
|
||||
)
|
||||
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):
|
||||
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()
|
||||
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:
|
||||
@@ -53,15 +72,3 @@ class TestItemValueDialog(TestItem):
|
||||
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()
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 != "":
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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("&", "&").replace("<", "<").replace(">", ">")
|
||||
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))
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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>
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
29
test/validation/items/common/helper_lib.py
Normal file
29
test/validation/items/common/helper_lib.py
Normal 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
|
||||
@@ -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
|
||||
|
||||
@@ -10,3 +10,28 @@
|
||||
name : Various syntax robustness
|
||||
steps:
|
||||
- !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
|
||||
@@ -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"
|
||||
|
||||
134
test/validation/items/jsonrpc/jrpc_echo_server.py
Normal file
134
test/validation/items/jsonrpc/jrpc_echo_server.py
Normal 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()
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
- group:
|
||||
name: Plot test
|
||||
condition: <| $(validation_dialogs) and not tm.text_mode() |>
|
||||
steps:
|
||||
|
||||
- plot:
|
||||
name: Add to the plot
|
||||
name: Open the plot
|
||||
key: $(test)_PASS
|
||||
plot_name: Mon Plot
|
||||
steps:
|
||||
- add:
|
||||
value1: $(loop_index)
|
||||
value2: $(loop_index)+2
|
||||
- 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 between values
|
||||
timeout: 1
|
||||
name: sleep
|
||||
dialog: true
|
||||
timeout: 3
|
||||
|
||||
- py_func:
|
||||
name: last plot values
|
||||
- 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
|
||||
file: $(test_path)$(psep)plot.py
|
||||
func_name: LastValues
|
||||
param:
|
||||
- Mon Plot
|
||||
plot_name: Mon Plot
|
||||
steps:
|
||||
- export: $(validation_report_path)/plot_export.pdf
|
||||
- export: $(validation_report_path)/plot_export.csv
|
||||
|
||||
- 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
|
||||
- plot:
|
||||
name: Close the plot
|
||||
execute_on_stop: True
|
||||
key: $(test)_PASS
|
||||
plot_name: Mon Plot
|
||||
steps:
|
||||
- close:
|
||||
wait_dialog_exit: True
|
||||
timeout: 2
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
1
test/validation/items/run/param.yaml
Normal file
1
test/validation/items/run/param.yaml
Normal file
@@ -0,0 +1 @@
|
||||
no_param: Null
|
||||
7
test/validation/items/run/sub_fail.tum
Normal file
7
test/validation/items/run/sub_fail.tum
Normal file
@@ -0,0 +1,7 @@
|
||||
main:
|
||||
name: run sub-test (always fail)
|
||||
steps:
|
||||
- check:
|
||||
name: fail
|
||||
values:
|
||||
- false
|
||||
7
test/validation/items/run/sub_pass.tum
Normal file
7
test/validation/items/run/sub_pass.tum
Normal file
@@ -0,0 +1,7 @@
|
||||
main:
|
||||
name: run sub-test (always pass)
|
||||
steps:
|
||||
- check:
|
||||
name: pass
|
||||
values:
|
||||
- true
|
||||
25
test/validation/items/run/test.tum
Normal file
25
test/validation/items/run/test.tum
Normal 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
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user