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
sys.path.insert(0, os.path.abspath('../../../../src/testium/'))
sys.path.insert(0, os.path.abspath('../../../../src/'))
# -- Project information -----------------------------------------------------

View File

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

View File

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

View File

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

Binary file not shown.

View File

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

View File

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

View File

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

View File

@@ -8,6 +8,7 @@ import copy
from lib.string_queue import StringQueue
from lib.tum_except import print_exception, ETUMRuntimeError, ETUMSyntaxError
import libs.testium as tm
import interpreter.utils.globdict as globdict
from interpreter.utils.params import expanse
from interpreter.utils.test_ctrl import TestSetController
from interpreter.utils.test_init import (
@@ -255,6 +256,7 @@ Is the python exec path correct ?"""
try:
test_run_init()
print(test_run_header())
globdict.set_update_queue(self.__squeue)
test_set.execute()
finally:
if test_set.success():
@@ -265,8 +267,16 @@ Is the python exec path correct ?"""
test_set.run_post_exec()
finally:
self.__exec = False
# Stop shared context engines before restore_gd wipes them
for engine in tm.gd("_py_func_contexts", {}).values():
engine.stop()
engine.join()
for engine in tm.gd("_lua_func_contexts", {}).values():
engine.stop()
engine.join()
# Sends signal to the GUI
self.send_finished()
globdict.set_update_queue(None)
restore_gd(gdict)
except Exception as e:
print_exception(e)
@@ -304,6 +314,9 @@ Is the python exec path correct ?"""
"enabled_state": test_set.getEnabledState,
"process_param": self.process_param,
"set_test_outputs": self.set_test_outputs,
"get_gd_vars": self.get_gd_vars,
"set_gd_var": self.set_gd_var,
"del_gd_var": self.del_gd_var,
"set_enabled_state": test_set.setEnabledState,
"check_uncheck_all": test_set.checkUncheckAll,
"get_folded": test_set.getFolded,
@@ -337,6 +350,25 @@ Is the python exec path correct ?"""
def set_test_outputs(self, outputs: list):
tm.setgd("test_outputs", outputs)
def get_gd_vars(self):
import json
result = {}
for k, v in globdict.global_dict.items():
if k.startswith("_"):
continue
try:
json.dumps(v)
result[k] = v
except (TypeError, ValueError):
pass
return result
def set_gd_var(self, name: str, value):
tm.setgd(name, value)
def del_gd_var(self, name: str):
tm.delgd(name)
def process_control_commands(self, tctrl):
term = False
while (not term) and (not self.__closed):

View File

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

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

View File

@@ -1,26 +1,30 @@
import sys
import os
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt
from PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtWidgets import QMessageBox
from multiprocessing import freeze_support
def main(args):
app = QApplication(sys.argv)
reply = QMessageBox.information(None, args[0], args[1], QMessageBox.Ok)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
msg = QMessageBox()
msg.setWindowFlags(Qt.WindowStaysOnTopHint)
msg.setWindowTitle(args[0])
msg.setText(args[1])
msg.setIcon(QMessageBox.Information)
msg.setStandardButtons(QMessageBox.Ok)
msg.exec()
if hasattr(sys, "frozen"):
#all standard streams are replaced by dummy one to avoid cx_freeze flushing bug.
class dummyStream:
''' dummyStream behaves like a stream but does nothing. '''
def __init__(self): pass
def write(self,data): pass
def read(self,data): pass
def write(self, data): pass
def read(self, data): pass
def flush(self): pass
def close(self): pass
# and now redirect all default streams to this dummyStream:
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
@@ -31,6 +35,3 @@ def main(args):
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):
success = True
app = QApplication(args)
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow()
d.setFixedSize(387,224)
d.setWindowFlags(Qt.WindowStaysOnTopHint)

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -1,62 +1,36 @@
import os
import sys
from multiprocessing import Process, Pipe
from PySide6.QtWidgets import QMessageBox
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.dialog_question_files import question_dialog
from lib.tum_except import ETUMSyntaxError
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import item_load_context
class TestItemQuestionDialog(TestItem):
class TestItemQuestionDialog(TestItemDialogBase):
"""dialog_question item usage.
dialog_question name: Nice question, question: "If OK, press OK, If not, press cancel"
"""
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_QUESTION_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_QUESTION_DLG
self.is_container = False
try:
self._question = self._prms.getParam('question', required = True)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
@test_run
def execute(self):
ourpath = __file__
test_file = os.path.join(os.path.dirname(ourpath),
'dialog_question_files',
'question_dialog.py')
q = self._prms.expanse(self._question)
print('Question asked:\n' + q + '\n')
parent_conn, child_conn = Pipe()
p=Process(target=question_dialog.main,
args=([self.name(), q],child_conn))
p.start()
succ = parent_conn.recv()
p.join()
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")
return
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)
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
def nowInBetween(start, end):
@@ -30,7 +30,7 @@ class TestItemRun(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_RUN
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.param_file = self._prms.getParam('param_file', 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.end_time = self._prms.getParam('end_time')
self.wait_for_exec = self._prms.getParam('wait_for_exec')
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):

View File

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

View File

@@ -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.dialog_sleep_files import dialog_sleep
from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError
from lib.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
class TestItemSleep(TestItem):
"""sleep item usage.
@@ -19,14 +19,9 @@ class TestItemSleep(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_SLEEP
self.is_container = False
try:
self._timeout = self._prms.getParam('timeout', required = True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._timeout = self._prms.getParam('timeout', required=True)
self._has_dialog = self._prms.getParam('dialog', default=False)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
self.seqFilename(),
)
@test_run
def execute(self):

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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