9 Commits

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

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

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

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

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

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

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

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

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

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 16:27:31 +02:00
47 changed files with 1356 additions and 741 deletions

View File

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

View File

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

View File

@@ -87,6 +87,10 @@ if not provided is given in the table as well.
| | | see :ref:`Expected result<sec_expected_result>` | | | | see :ref:`Expected result<sec_expected_result>` |
| | | for details. | | | | for details. |
+-----------------------+-------------------+-------------------------------------------------------+ +-----------------------+-------------------+-------------------------------------------------------+
| ``store_result`` | / | Store the test result in a global variable. |
| | | see :ref:`Store result<sec_store_result>` |
| | | for details. |
+-----------------------+-------------------+-------------------------------------------------------+
last test result 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. 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 Export attribute
----------------------------------------------- -----------------------------------------------

View File

@@ -123,17 +123,12 @@ Each ``py_func`` item without a ``context_id`` runs in a dedicated subprocess th
is started and stopped around the call. State cannot be shared between two such is started and stopped around the call. State cannot be shared between two such
items using module-level variables. items using module-level variables.
Two mechanisms are available to share data across calls: 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
**Using the testium global dictionary** item, including other ``py_func`` items, without requiring a shared subprocess.
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.
.. code-block:: python .. 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 import py_func.tm as tm
@@ -144,36 +139,22 @@ items) without requiring a shared subprocess.
def consume(): def consume():
return tm.gd("my_shared_value", None) 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 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. reuse the same persistent subprocess. This allows sharing any Python object across
calls — including objects that cannot be transmitted to other processes.
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.
.. code-block:: python .. 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 import py_func.tm as tm
class _Connection: # not JSON-serializable
def __init__(self):
self.value = "open"
def open_connection(): def open_connection():
tm.setgd("conn", _Connection()) # stored locally in the subprocess tm.setgd("conn", MyConnection())
return "ok" return "ok"
def use_connection(): def use_connection():
conn = tm.gd("conn") # retrieved from the subprocess local dict conn = tm.gd("conn")
return conn.value return conn.status()
.. code-block:: yaml .. code-block:: yaml
:caption: ``py_func`` items sharing a persistent subprocess :caption: ``py_func`` items sharing a persistent subprocess

Binary file not shown.

View File

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

View File

@@ -1,5 +1,6 @@
import traceback import traceback
import textwrap import textwrap
from contextlib import contextmanager
class ETUMError(Exception): class ETUMError(Exception):
@@ -67,6 +68,28 @@ class ETUMParamError(ETUMError):
return lines 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): def print_exception(exc: ETUMError):
if not isinstance(exc, ETUMError): if not isinstance(exc, ETUMError):
print(traceback.format_exc(4)) print(traceback.format_exc(4))

View File

@@ -45,12 +45,21 @@ for k in SUPPORTED_API:
############################################################################### ###############################################################################
# gd/setgd/delgd with local-dict fallback for non-serializable values # gd/setgd/delgd with local-dict fallback for non-serializable values
def gd(*params): def gd(name, default=None):
key = params[0] if params else None """Return a value from the testium global dictionary.
if key is not None and key in _local_dict:
return _local_dict[key] 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: 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: if "result" in res:
return res["result"] return res["result"]
elif "error" in res: elif "error" in res:
@@ -60,14 +69,25 @@ def gd(*params):
raise ETUMRuntimeError("api not initialized") raise ETUMRuntimeError("api not initialized")
def setgd(*params): def setgd(name, value):
key = params[0] if params else None """Store a value in the testium global dictionary.
value = params[1] if len(params) > 1 else None
if key is not None and not _is_json_serializable(value): The stored value is accessible from any subsequent test item and from any
_local_dict[key] = value ``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 return None
if _func_call_thread is not 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: if "result" in res:
return res["result"] return res["result"]
elif "error" in res: elif "error" in res:
@@ -77,13 +97,17 @@ def setgd(*params):
raise ETUMRuntimeError("api not initialized") raise ETUMRuntimeError("api not initialized")
def delgd(*params): def delgd(name):
key = params[0] if params else None """Remove an entry from the testium global dictionary.
if key is not None and key in _local_dict:
del _local_dict[key] :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 return None
if _func_call_thread is not 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: if "result" in res:
return res["result"] return res["result"]
elif "error" in res: elif "error" in res:

View File

@@ -8,6 +8,7 @@ import copy
from lib.string_queue import StringQueue from lib.string_queue import StringQueue
from lib.tum_except import print_exception, ETUMRuntimeError, ETUMSyntaxError from lib.tum_except import print_exception, ETUMRuntimeError, ETUMSyntaxError
import libs.testium as tm import libs.testium as tm
import interpreter.utils.globdict as globdict
from interpreter.utils.params import expanse from interpreter.utils.params import expanse
from interpreter.utils.test_ctrl import TestSetController from interpreter.utils.test_ctrl import TestSetController
from interpreter.utils.test_init import ( from interpreter.utils.test_init import (
@@ -255,6 +256,7 @@ Is the python exec path correct ?"""
try: try:
test_run_init() test_run_init()
print(test_run_header()) print(test_run_header())
globdict.set_update_queue(self.__squeue)
test_set.execute() test_set.execute()
finally: finally:
if test_set.success(): if test_set.success():
@@ -265,8 +267,16 @@ Is the python exec path correct ?"""
test_set.run_post_exec() test_set.run_post_exec()
finally: finally:
self.__exec = False 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 # Sends signal to the GUI
self.send_finished() self.send_finished()
globdict.set_update_queue(None)
restore_gd(gdict) restore_gd(gdict)
except Exception as e: except Exception as e:
print_exception(e) print_exception(e)
@@ -304,6 +314,9 @@ Is the python exec path correct ?"""
"enabled_state": test_set.getEnabledState, "enabled_state": test_set.getEnabledState,
"process_param": self.process_param, "process_param": self.process_param,
"set_test_outputs": self.set_test_outputs, "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, "set_enabled_state": test_set.setEnabledState,
"check_uncheck_all": test_set.checkUncheckAll, "check_uncheck_all": test_set.checkUncheckAll,
"get_folded": test_set.getFolded, "get_folded": test_set.getFolded,
@@ -337,6 +350,25 @@ Is the python exec path correct ?"""
def set_test_outputs(self, outputs: list): def set_test_outputs(self, outputs: list):
tm.setgd("test_outputs", outputs) 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): def process_control_commands(self, tctrl):
term = False term = False
while (not term) and (not self.__closed): while (not term) and (not self.__closed):

View File

@@ -185,7 +185,9 @@ def main(args, conn=None):
SettingsApplication = "testium_choices_dlg_" + args[0] SettingsApplication = "testium_choices_dlg_" + args[0]
SettingsLastChoices = "last_choice" SettingsLastChoices = "last_choice"
success = True success = True
app = QApplication() from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = ChoicesDialog() d = ChoicesDialog()
d.setFixedSize(800, 600) d.setFixedSize(800, 600)
d.setWindowFlags(Qt.WindowStaysOnTopHint) d.setWindowFlags(Qt.WindowStaysOnTopHint)

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -7,7 +7,7 @@ import libs.testium as tm
from interpreter.utils.params import TestItemParams from interpreter.utils.params import TestItemParams
from interpreter.utils.constants import TestItemType as cst_type from interpreter.utils.constants import TestItemType as cst_type
from interpreter.utils.eval import eval_to_boolean, evaluate, post_evaluate 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_STOP = '<----- step "{}" finished'
LOG_TEST_START = '-----> step "{}" started' LOG_TEST_START = '-----> step "{}" started'
@@ -101,6 +101,7 @@ class TestItem:
self.status_queue = status_queue self.status_queue = status_queue
self._execute_on_stop = False self._execute_on_stop = False
self._post_eval = None self._post_eval = None
self._store_result = None
self._expected_result = None self._expected_result = None
self._no_fail = None self._no_fail = None
self._is_stopped = False self._is_stopped = False
@@ -131,11 +132,11 @@ class TestItem:
if s: if s:
try: try:
self.skipped = eval_to_boolean(s) self.skipped = eval_to_boolean(s)
except: except Exception as e:
raise ETUMSyntaxError( 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}", 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(), self.seqFilename(),
) ) from e
# This allow disabling test item directly by using its name inside param.yaml file # This allow disabling test item directly by using its name inside param.yaml file
elif self._name in tm.gd("skipped_test_item", []): elif self._name in tm.gd("skipped_test_item", []):
self.skipped = True self.skipped = True
@@ -155,6 +156,9 @@ class TestItem:
if "process_result" in dict_item: if "process_result" in dict_item:
self._post_eval = dict_item["process_result"] 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: if "expected_result" in dict_item:
self._expected_result = dict_item["expected_result"] self._expected_result = dict_item["expected_result"]
@@ -164,11 +168,13 @@ class TestItem:
self.banner = LOG_TEST_START.format(self._name) self.banner = LOG_TEST_START.format(self._name)
self.footer = LOG_TEST_STOP.format(self._name) self.footer = LOG_TEST_STOP.format(self._name)
except: except ETUMSyntaxError:
raise
except Exception as e:
raise ETUMSyntaxError( 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(), self.seqFilename(),
) ) from e
self.result = TestResult(self, TestValue.FAILURE, "Failure by default") self.result = TestResult(self, TestValue.FAILURE, "Failure by default")
@@ -275,6 +281,9 @@ class TestItem:
self.process_result() self.process_result()
# expected_result treatment # expected_result treatment
self.result_expected() 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 # Case of the no_fail true parameter
self.process_no_fail() self.process_no_fail()
@@ -317,6 +326,17 @@ class TestItem:
print(e) print(e)
self.result.set(TestValue.FAILURE, "Result processing failed") 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): def process_report(self, report_eval):
tm.print_debug(f"Export reported values:") tm.print_debug(f"Export reported values:")
rep_eval = self._prms.expanse(report_eval) rep_eval = self._prms.expanse(report_eval)

View File

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

View File

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

View File

@@ -0,0 +1,46 @@
import multiprocessing
from interpreter.test_items.test_item import TestItem
_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):
result = parent_conn.recv()
break
self._cleanup_process(p)
return result

View File

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

View File

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

View File

@@ -4,7 +4,7 @@ import traceback
import pprint import pprint
import textwrap 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_item import TestItem, test_run
from interpreter.test_items.test_result import TestValue from interpreter.test_items.test_result import TestValue
import libs.testium as tm import libs.testium as tm
@@ -26,16 +26,11 @@ class TestItemLuaFunc(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename) super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_LUA_FUNCTION self._type = cst.TYPE_LUA_FUNCTION
self.is_container = False 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.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True) self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param") self.params = self._prms.getParamAll("param")
self._context_id = self._prms.getParam("context_id", default=None, processed=False) 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) self._lua_func_proc = LuaFuncExecEngine(tm.gd("lua_bin", ""), api_request, 10)
def _get_engine(self): def _get_engine(self):

View File

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

View File

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

View File

@@ -4,7 +4,7 @@ import time
import pprint import pprint
import textwrap 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_item import TestItem, test_run
from interpreter.test_items.test_result import TestValue from interpreter.test_items.test_result import TestValue
import libs.testium as tm import libs.testium as tm
@@ -26,16 +26,11 @@ class TestItemPyFunc(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename) super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_PY_FUNCTION self._type = cst.TYPE_PY_FUNCTION
self.is_container = False 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.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True) self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param") self.params = self._prms.getParamAll("param")
self._context_id = self._prms.getParam("context_id", default=None, processed=False) 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) self._py_func_proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
def _get_engine(self): def _get_engine(self):

View File

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

View File

@@ -10,7 +10,7 @@ from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue) from interpreter.test_items.test_result import (TestValue)
import libs.testium as tm import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst 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): def nowInBetween(start, end):
@@ -30,7 +30,7 @@ class TestItemRun(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename) super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_RUN self._type = cst.TYPE_RUN
self.is_container = False self.is_container = False
try: with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.tum_fime = self._prms.getParam('tum_fime', required=True) self.tum_fime = self._prms.getParam('tum_fime', required=True)
self.param_file = self._prms.getParam('param_file', default='') self.param_file = self._prms.getParam('param_file', default='')
self.python_bin = self._prms.getParam('python_bin', default='') self.python_bin = self._prms.getParam('python_bin', default='')
@@ -40,11 +40,6 @@ class TestItemRun(TestItem):
self.start_time = self._prms.getParam('start_time') self.start_time = self._prms.getParam('start_time')
self.end_time = self._prms.getParam('end_time') self.end_time = self._prms.getParam('end_time')
self.wait_for_exec = self._prms.getParam('wait_for_exec') 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 @test_run
def execute(self): def execute(self):

View File

@@ -4,7 +4,7 @@ import traceback
from functools import wraps from functools import wraps
import libs.testium as tm 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_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.test_items.item_actions import TestItemActions from interpreter.test_items.item_actions import TestItemActions
@@ -108,17 +108,12 @@ class TestItemPlotActionPeriodic(TestItemPlotAction):
) )
# Periodic function call # Periodic function call
try: with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.period = self._prms.getParam("period", required=True) self.period = self._prms.getParam("period", required=True)
self.file_name = self._prms.getParam("file", required=True) self.file_name = self._prms.getParam("file", required=True)
self.func_name = self._prms.getParam("func_name", required=True) self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param") self.params = self._prms.getParamAll("param")
self.post_eval = self._prms.getParam("eval", default="") 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 @test_run
def execute(self): def execute(self):

View File

@@ -7,7 +7,7 @@ from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue) from interpreter.test_items.test_result import (TestValue)
from interpreter.test_items.dialog_sleep_files import dialog_sleep from interpreter.test_items.dialog_sleep_files import dialog_sleep
from interpreter.utils.constants import TestItemType as cst 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): class TestItemSleep(TestItem):
"""sleep item usage. """sleep item usage.
@@ -19,14 +19,9 @@ class TestItemSleep(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename) super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_SLEEP self._type = cst.TYPE_SLEEP
self.is_container = False self.is_container = False
try: with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._timeout = self._prms.getParam('timeout', required = True) self._timeout = self._prms.getParam('timeout', required=True)
self._has_dialog = self._prms.getParam('dialog', default=False) 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 @test_run
def execute(self): def execute(self):

View File

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

View File

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

View File

@@ -1,5 +1,4 @@
import sys import sys
import os
from multiprocessing import freeze_support from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QDialog, QTableWidgetItem) from PySide6.QtWidgets import (QApplication, QDialog, QTableWidgetItem)
@@ -20,7 +19,9 @@ def main(args, conn=None):
SettingsApplication = 'testium_ref_item' SettingsApplication = 'testium_ref_item'
SettingsLastReference = 'lastReference' SettingsLastReference = 'lastReference'
success = True success = True
app = QApplication(args) from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestedRefsWindow() d = TestedRefsWindow()
d.setFixedSize(481,386) d.setFixedSize(481,386)
d.setWindowFlags(Qt.WindowStaysOnTopHint) d.setWindowFlags(Qt.WindowStaysOnTopHint)

View File

@@ -3,9 +3,7 @@ import datetime
from queue import Queue from queue import Queue
from interpreter.utils.params import expanse from interpreter.utils.params import expanse
import libs.testium as tm import libs.testium as tm
from lib.tum_except import ( from lib.tum_except import ETUMSyntaxError
ETUMSyntaxError,
)
import interpreter.utils.settings as prefs import interpreter.utils.settings as prefs
from interpreter.test_report.test_report import TestReport from interpreter.test_report.test_report import TestReport
from interpreter.utils.py_func_exec import PyFuncExecEngine 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 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: class TestSet:
def __init__( def __init__(
self, self,
@@ -479,12 +488,19 @@ class TestSet:
action_name = cst.FOLDED_CHAR + it.item_cmd action_name = cst.FOLDED_CHAR + it.item_cmd
seq_filename = action[action_name]["seq_filename"] seq_filename = action[action_name]["seq_filename"]
item = (it.item_class)( try:
action[action_name], item = (it.item_class)(
tree_parent, action[action_name],
self.status_queue, tree_parent,
filename=seq_filename 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 item.is_folded = is_folded
child = {} child = {}
# case where the test item loads itself its descendants # case where the test item loads itself its descendants

View File

@@ -1,3 +1,4 @@
import json
from threading import Lock from threading import Lock
@@ -5,6 +6,30 @@ global_dict = {}
global_dict_lock = Lock() 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 # Global dictionnary helper functions
def gd(name, default=None): def gd(name, default=None):
''' Function which returns a variable from the global dictionary of testium ''' Function which returns a variable from the global dictionary of testium
@@ -31,6 +56,7 @@ def setgd(name, value):
''' '''
with global_dict_lock: with global_dict_lock:
global_dict.update({name: value}) global_dict.update({name: value})
_push_update(name, value)
def delgd(name): def delgd(name):
''' Function which removes a variable from the global dictionary of testium ''' Function which removes a variable from the global dictionary of testium
@@ -44,6 +70,7 @@ def delgd(name):
del global_dict[name] del global_dict[name]
except: except:
pass pass
_push_delete(name)
def cleargd(): def cleargd():
with global_dict_lock: with global_dict_lock:

View File

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

View File

@@ -3,7 +3,7 @@
################################################################################ ################################################################################
## Form generated from reading UI file 'f1_win_core.ui' ## 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! ## 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, QImage, QKeySequence, QLinearGradient, QPainter,
QPalette, QPixmap, QRadialGradient, QTransform) QPalette, QPixmap, QRadialGradient, QTransform)
from PySide6.QtWidgets import (QApplication, QDialog, QFormLayout, QHBoxLayout, from PySide6.QtWidgets import (QApplication, QDialog, QFormLayout, QHBoxLayout,
QLabel, QLineEdit, QPushButton, QSizePolicy, QHeaderView, QLabel, QLineEdit, QPushButton,
QSpacerItem, QTextEdit, QToolButton, QVBoxLayout, QSizePolicy, QSpacerItem, QTabWidget, QTableWidget,
QTableWidgetItem, QTextEdit, QToolButton, QVBoxLayout,
QWidget) QWidget)
import f1_win_rc import f1_win_rc
@@ -25,7 +26,7 @@ class Ui_F1Dialog(object):
def setupUi(self, F1Dialog): def setupUi(self, F1Dialog):
if not F1Dialog.objectName(): if not F1Dialog.objectName():
F1Dialog.setObjectName(u"F1Dialog") F1Dialog.setObjectName(u"F1Dialog")
F1Dialog.resize(400, 300) F1Dialog.resize(550, 450)
icon = QIcon() icon = QIcon()
if QIcon.hasThemeIcon(QIcon.ThemeIcon.HelpAbout): if QIcon.hasThemeIcon(QIcon.ThemeIcon.HelpAbout):
icon = QIcon.fromTheme(QIcon.ThemeIcon.HelpAbout) icon = QIcon.fromTheme(QIcon.ThemeIcon.HelpAbout)
@@ -36,19 +37,20 @@ class Ui_F1Dialog(object):
F1Dialog.setLayoutDirection(Qt.LayoutDirection.LeftToRight) F1Dialog.setLayoutDirection(Qt.LayoutDirection.LeftToRight)
self.verticalLayout_2 = QVBoxLayout(F1Dialog) self.verticalLayout_2 = QVBoxLayout(F1Dialog)
self.verticalLayout_2.setObjectName(u"verticalLayout_2") self.verticalLayout_2.setObjectName(u"verticalLayout_2")
self.horizontalLayout_2 = QHBoxLayout() self.tabWidget = QTabWidget(F1Dialog)
self.horizontalLayout_2.setObjectName(u"horizontalLayout_2") self.tabWidget.setObjectName(u"tabWidget")
self.tabTestItem = QWidget()
self.verticalLayout_2.addLayout(self.horizontalLayout_2) self.tabTestItem.setObjectName(u"tabTestItem")
self.verticalLayout_tab0 = QVBoxLayout(self.tabTestItem)
self.verticalLayout_tab0.setObjectName(u"verticalLayout_tab0")
self.formLayout = QFormLayout() self.formLayout = QFormLayout()
self.formLayout.setObjectName(u"formLayout") self.formLayout.setObjectName(u"formLayout")
self.typeLabel = QLabel(F1Dialog) self.typeLabel = QLabel(self.tabTestItem)
self.typeLabel.setObjectName(u"typeLabel") self.typeLabel.setObjectName(u"typeLabel")
self.formLayout.setWidget(0, QFormLayout.ItemRole.LabelRole, self.typeLabel) self.formLayout.setWidget(0, QFormLayout.ItemRole.LabelRole, self.typeLabel)
self.typeLineEdit = QLineEdit(F1Dialog) self.typeLineEdit = QLineEdit(self.tabTestItem)
self.typeLineEdit.setObjectName(u"typeLineEdit") self.typeLineEdit.setObjectName(u"typeLineEdit")
sizePolicy = QSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed) sizePolicy = QSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
sizePolicy.setHorizontalStretch(0) sizePolicy.setHorizontalStretch(0)
@@ -59,20 +61,20 @@ class Ui_F1Dialog(object):
self.formLayout.setWidget(0, QFormLayout.ItemRole.FieldRole, self.typeLineEdit) self.formLayout.setWidget(0, QFormLayout.ItemRole.FieldRole, self.typeLineEdit)
self.sequenceFileNameLabel = QLabel(F1Dialog) self.sequenceFileNameLabel = QLabel(self.tabTestItem)
self.sequenceFileNameLabel.setObjectName(u"sequenceFileNameLabel") self.sequenceFileNameLabel.setObjectName(u"sequenceFileNameLabel")
self.formLayout.setWidget(1, QFormLayout.ItemRole.LabelRole, self.sequenceFileNameLabel) self.formLayout.setWidget(1, QFormLayout.ItemRole.LabelRole, self.sequenceFileNameLabel)
self.horizontalLayout_3 = QHBoxLayout() self.horizontalLayout_3 = QHBoxLayout()
self.horizontalLayout_3.setObjectName(u"horizontalLayout_3") self.horizontalLayout_3.setObjectName(u"horizontalLayout_3")
self.sequenceFileNameLineEdit = QLineEdit(F1Dialog) self.sequenceFileNameLineEdit = QLineEdit(self.tabTestItem)
self.sequenceFileNameLineEdit.setObjectName(u"sequenceFileNameLineEdit") self.sequenceFileNameLineEdit.setObjectName(u"sequenceFileNameLineEdit")
self.sequenceFileNameLineEdit.setReadOnly(True) self.sequenceFileNameLineEdit.setReadOnly(True)
self.horizontalLayout_3.addWidget(self.sequenceFileNameLineEdit) self.horizontalLayout_3.addWidget(self.sequenceFileNameLineEdit)
self.ButtLocOpen = QToolButton(F1Dialog) self.ButtLocOpen = QToolButton(self.tabTestItem)
self.ButtLocOpen.setObjectName(u"ButtLocOpen") self.ButtLocOpen.setObjectName(u"ButtLocOpen")
self.horizontalLayout_3.addWidget(self.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.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.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.setObjectName(u"TestContentEdit")
self.TestContentEdit.setReadOnly(True) 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 = QHBoxLayout()
self.horizontalLayout.setObjectName(u"horizontalLayout") self.horizontalLayout.setObjectName(u"horizontalLayout")
@@ -113,6 +158,9 @@ class Ui_F1Dialog(object):
self.retranslateUi(F1Dialog) self.retranslateUi(F1Dialog)
self.tabWidget.setCurrentIndex(0)
QMetaObject.connectSlotsByName(F1Dialog) QMetaObject.connectSlotsByName(F1Dialog)
# setupUi # setupUi
@@ -122,6 +170,15 @@ class Ui_F1Dialog(object):
self.sequenceFileNameLabel.setText(QCoreApplication.translate("F1Dialog", u"Test file name", None)) self.sequenceFileNameLabel.setText(QCoreApplication.translate("F1Dialog", u"Test file name", None))
self.ButtLocOpen.setText(QCoreApplication.translate("F1Dialog", u"...", None)) self.ButtLocOpen.setText(QCoreApplication.translate("F1Dialog", u"...", None))
self.label.setText(QCoreApplication.translate("F1Dialog", u"Test content:", 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)) self.ButtClose.setText(QCoreApplication.translate("F1Dialog", u"Close", None))
# retranslateUi # retranslateUi

View File

@@ -6,8 +6,8 @@
<rect> <rect>
<x>0</x> <x>0</x>
<y>0</y> <y>0</y>
<width>400</width> <width>550</width>
<height>300</height> <height>450</height>
</rect> </rect>
</property> </property>
<property name="windowTitle"> <property name="windowTitle">
@@ -22,69 +22,139 @@
</property> </property>
<layout class="QVBoxLayout" name="verticalLayout_2"> <layout class="QVBoxLayout" name="verticalLayout_2">
<item> <item>
<layout class="QHBoxLayout" name="horizontalLayout_2"/> <widget class="QTabWidget" name="tabWidget">
</item> <property name="currentIndex">
<item> <number>0</number>
<layout class="QFormLayout" name="formLayout"> </property>
<item row="0" column="0"> <!-- Tab 0: Test item -->
<widget class="QLabel" name="typeLabel"> <widget class="QWidget" name="tabTestItem">
<property name="text"> <attribute name="title">
<string>Test step type</string> <string>Test item</string>
</property> </attribute>
</widget> <layout class="QVBoxLayout" name="verticalLayout_tab0">
</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> <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"> <property name="readOnly">
<bool>true</bool> <bool>true</bool>
</property> </property>
</widget> </widget>
</item> </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> <item>
<widget class="QToolButton" name="ButtLocOpen"> <widget class="QTableWidget" name="varsTable">
<property name="text"> <column>
<string>...</string> <property name="text">
</property> <string>Key</string>
</property>
</column>
<column>
<property name="text">
<string>Value</string>
</property>
</column>
<column>
<property name="text">
<string/>
</property>
</column>
</widget> </widget>
</item> </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> </layout>
</item> </widget>
</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>
<item> <item>

View File

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

View File

@@ -30,11 +30,18 @@ class TestFileManager:
): ):
w.test_service.stop() w.test_service.stop()
w.test_service.close() 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 del w.test_proc
w.test_proc = None w.test_proc = None
del w.test_service del w.test_service
w.test_service = None w.test_service = None
w.d_f1_win.set_service(None)
del w.ts_controller del w.ts_controller
w.ts_controller = None w.ts_controller = None
@@ -83,6 +90,7 @@ class TestFileManager:
w.testFile = None w.testFile = None
w.ts_controller = TestSetController() w.ts_controller = TestSetController()
w.test_service = TestControllerService(w.ts_controller) w.test_service = TestControllerService(w.ts_controller)
w.d_f1_win.set_service(w.test_service)
w.test_proc = TestProcess( w.test_proc = TestProcess(
file_name, file_name,
w.status_queue, w.status_queue,
@@ -106,6 +114,7 @@ class TestFileManager:
w.test_proc = None w.test_proc = None
del w.test_service del w.test_service
w.test_service = None w.test_service = None
w.d_f1_win.set_service(None)
del w.ts_controller del w.ts_controller
w.ts_controller = None w.ts_controller = None
raise ETUMRuntimeError( raise ETUMRuntimeError(
@@ -122,6 +131,7 @@ class TestFileManager:
progress = None progress = None
w.treeTests.setFoldDefault() w.treeTests.setFoldDefault()
w.treeTests.updateTreeSkipState(w.test_service) w.treeTests.updateTreeSkipState(w.test_service)
w.d_f1_win.load_initial_vars(w.test_service.get_gd_vars())
w.checkSelect.setChecked(True) w.checkSelect.setChecked(True)
w.testFile = file_name w.testFile = file_name

View File

@@ -6,6 +6,8 @@ from PySide6.QtCore import (Signal, QThread)
class ThreadTestStatus(QThread): class ThreadTestStatus(QThread):
statusToBeUpdated = Signal(dict) statusToBeUpdated = Signal(dict)
testSetIsFinished = Signal() testSetIsFinished = Signal()
gdUpdated = Signal(str, object)
gdDeleted = Signal(str)
def __init__(self, status_queue, parent=None, debug=False): def __init__(self, status_queue, parent=None, debug=False):
super().__init__(parent) super().__init__(parent)
@@ -21,7 +23,12 @@ class ThreadTestStatus(QThread):
while True: while True:
while not self._status_queue.empty(): while not self._status_queue.empty():
m = self._status_queue.get() 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 m.get("id", None) is None:
self.testSetIsFinished.emit() self.testSetIsFinished.emit()
else: else:
self.statusToBeUpdated.emit(m) self.statusToBeUpdated.emit(m)

View File

@@ -247,6 +247,8 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.threadTestStatus.testSetIsFinished.connect(self.runner.on_run_finished) self.threadTestStatus.testSetIsFinished.connect(self.runner.on_run_finished)
self.threadTestStatus.statusToBeUpdated.connect(self.treeTests.updateStatus) 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() self.reconnect_signals()
if runandclose: if runandclose:

View File

@@ -92,13 +92,14 @@
func_name: echo func_name: echo
param: [ $(str_example) ] param: [ $(str_example) ]
process_result: "'44' in '$(result)'" process_result: "'44' in '$(result)'"
expected_result: True
- py_func: - py_func:
name: Save the result in a global variable name: Save the result in a global variable
key: $(test)_PASS key: $(test)_PASS
file: $(test_path)$(psep)results$(psep)results.py file: $(test_path)$(psep)results$(psep)results.py
func_name: echo func_name: echo
param: [ 44 ] param: [ 44 ]
process_result: "tm.setgd('process_result_value', $(result))" store_result: process_result_value
- py_func: - py_func:
name: Check the saved global variable name: Check the saved global variable
key: $(test)_PASS key: $(test)_PASS
@@ -107,6 +108,68 @@
param: [ 44 ] param: [ 44 ]
expected_result: $(process_result_value) 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: - py_func:
name: Process result when result is None (must fail) name: Process result when result is None (must fail)
key: $(test)_FAIL key: $(test)_FAIL

View File

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

View File

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

View File

@@ -30,8 +30,10 @@ linux_prompt: "$ "
inc_no_template: "inc no template" inc_no_template: "inc no template"
inc_with_template: "inc with 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_rev: 5.5
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_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: 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 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