14 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
d92f518e1e Add context_id to py_func and lua_func for shared persistent subprocess
- py_func and lua_func items accept a context_id parameter; items sharing
  the same id reuse the same subprocess for the duration of the test run
- Subprocess-side tm.setgd/tm.gd use a local fallback dict for non-JSON-
  serializable values (py_func only); serializable values reach the main
  process global dict and are accessible from any test item or subprocess
- Shared subprocess engines are cleaned up in process.py finally block
- LuaProcessBase gains is_alive() (was missing, broke all lua_func items)
- Validation tests cover serializable sharing across different context ids,
  non-serializable sharing within the same context_id, and cross-item access
- RST documentation updated for both py_func and lua_func items

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 16:02:36 +02:00
49354b8664 Merge branch 'feature/reload-without-restart'
Reload test without restarting testium (closes #18)
2026-04-18 15:24:11 +02:00
7383820aba Reload test without restarting testium (closes #18)
Replace os.execv restart in actionRefresh with file_manager.reload(),
leveraging the subprocess architecture so py_func modules are freshly
imported on each reload. Add a modal progress dialog with step labels
during loading. Fix checkbox reappearing on breakpoint with
show_checkboxes OFF.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 15:23:53 +02:00
67c879ab10 Fix checkboxes reappearing when setting a breakpoint with show_checkboxes OFF
itemChanged fires for any data change (including icon updates in the pause
column), causing on_testChecked to inadvertently restore CheckStateRole via
synchronizeEnabledState. Guard against non-checkbox column changes.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 14:55:18 +02:00
aa72e349c6 Merge branch 'refactor/ui-modularization'
- Extract TestControllerService service layer over TestSetController
- Split MainWindow into TestRunner and TestFileManager coordinators
- Merge 21 QTestTreeItem subclasses into a single factory
- Replace _test_started/_test_paused booleans with TestState enum
2026-04-18 14:46:06 +02:00
53 changed files with 1737 additions and 766 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

@@ -39,11 +39,14 @@ The ``lua_func`` test item is of the form:
Beside common test items attributes, lua_func item has specific attribute, some of which being mandatory.
* ``file``: the script file name that contains the function to be executed.
Only python script format is supported.
Only Lua script format is supported.
* ``func_name``: The function name to be executed.
* ``param``: This is a list of parameters that are passed to the function
in the order they are presented in the script. These parameters are not
mandatory and are highly dependent of the function prototype.
* ``context_id``: Optional. When set, all ``lua_func`` items sharing the same
``context_id`` value run inside the same persistent Lua subprocess for the
duration of the test. See :ref:`lua_func context<sec_lua_func_context>` for details.
.. code-block:: yaml
:caption: ``lua_func`` test item example of usage
@@ -56,16 +59,71 @@ Beside common test items attributes, lua_func item has specific attribute, some
- $(my_param)
The result of the function (after eventual post treatment) is stored in the global
variable named ``pfn_<func_name>``
variable named ``lfn_<item_name>``
(See :ref:`global variables<sec_global_variables>` for more detail
on how to access to global variables from test items and scripts).
In the example above, the global variable ``$(lfn_activity)``
would be created at the end of the item execution. It would contain the resulting
value of the funcToBeExecuted python function.
value of the methodName Lua function.
The ``lua_func`` will always result ``PASS``, except if the called function raises
and exception or if the ``expected_result`` attribute is used.
an exception or if the ``expected_result`` attribute is used.
.. _sec_lua_func_context:
Sharing state between ``lua_func`` calls
------------------------------------------
Each ``lua_func`` item without a ``context_id`` runs in a dedicated subprocess that
is started and stopped around the call. Module-level variables are not preserved
between two such items.
Inside a ``lua_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 without requiring a shared
subprocess.
.. code-block:: lua
:caption: sharing a value via the global dictionary
local tm = require("tm")
local module = {}
function module.produce(val)
tm.setgd("my_shared_value", val)
return val
end
function module.consume()
return tm.gd("my_shared_value")
end
return module
When ``context_id`` is set, all ``lua_func`` items that share the same identifier
reuse the same persistent subprocess. This allows Lua-side state (upvalues, module
cache) to be retained across calls beyond what ``tm.setgd`` persists.
.. code-block:: yaml
:caption: ``lua_func`` items sharing a persistent subprocess
- lua_func:
name: produce value
file: my_script.lua
func_name: produce
context_id: my_context
param:
- hello
- lua_func:
name: consume value
file: my_script.lua
func_name: consume
context_id: my_context
expected_result: hello
The shared subprocess is automatically stopped at the end of the test run.
**Lua Interpreter environment setup**

View File

@@ -89,6 +89,9 @@ some of which being mandatory.
* ``param``: This is a list of parameters that are passed to the function
in the order they are presented in the script. These parameters are not
mandatory and are highly dependent of the function prototype.
* ``context_id``: Optional. When set, all ``py_func`` items sharing the same
``context_id`` value run inside the same persistent Python subprocess for the
duration of the test. See :ref:`py_func context<sec_py_func_context>` for details.
.. code-block:: yaml
:caption: ``py_func`` test item example of usage
@@ -111,6 +114,67 @@ value of the funcToBeExecuted python function.
The ``py_func`` will always result ``PASS``, except if the called function raises
and exception or if the ``expected_result`` attribute is used.
.. _sec_py_func_context:
Sharing state between ``py_func`` calls
------------------------------------------
Each ``py_func`` item without a ``context_id`` runs in a dedicated subprocess that
is started and stopped around the call. State cannot be shared between two such
items using module-level variables.
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 value via the global dictionary
import py_func.tm as tm
def produce(val):
tm.setgd("my_shared_value", val)
return val
def consume():
return tm.gd("my_shared_value", None)
When ``context_id`` is set, all ``py_func`` items that share the same identifier
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 an object via ``context_id``
import py_func.tm as tm
def open_connection():
tm.setgd("conn", MyConnection())
return "ok"
def use_connection():
conn = tm.gd("conn")
return conn.status()
.. code-block:: yaml
:caption: ``py_func`` items sharing a persistent subprocess
- py_func:
name: open connection
file: my_script.py
func_name: open_connection
context_id: my_context
expected_result: ok
- py_func:
name: use connection
file: my_script.py
func_name: use_connection
context_id: my_context
expected_result: open
The shared subprocess is automatically stopped at the end of the test run.
**Python Interpreter environment setup**
Some global variables have an impact on the ``py_func`` test item behavior:

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

@@ -1,19 +1,26 @@
import json
import sys
from py_func.handle import FuncHandler
from lib.tum_except import ETUMRuntimeError
from lib.api import SUPPORTED_API
thismodule = sys.modules[__name__]
# Shared FuncHandler instance used to forward API calls. Remains None
# until `_init_api` is invoked.
_func_call_thread = None
# Local storage for non-JSON-serializable values
_local_dict = {}
def _is_json_serializable(value):
try:
json.dumps(value)
return True
except (TypeError, ValueError):
return False
###############################################################################
# Dynamically create module-level functions for each supported API name.
# Each generated function shares the implementation of `api_call` but
# has a distinct name used as the remote action identifier.
def _make_api(name):
def _wrapper(*params):
if _func_call_thread is not None:
@@ -31,21 +38,86 @@ def _make_api(name):
return _wrapper
for k in SUPPORTED_API:
if k not in ('gd', 'setgd', 'delgd'):
setattr(thismodule, k, _make_api(k))
def _init_api(host, port, timeout):
"""Start and initialize the remote function handler.
Starts a ``FuncHandler`` bound to ``port``, runs it and blocks until
it signals readiness.
###############################################################################
# gd/setgd/delgd with local-dict fallback for non-serializable values
Args:
port: port number or identifier passed to ``FuncHandler``.
def gd(name, default=None):
"""Return a value from the testium global dictionary.
Returns:
The initialized ``FuncHandler`` instance assigned to
``_func_call_thread``.
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", (name, default))
if "result" in res:
return res["result"]
elif "error" in res:
raise ETUMRuntimeError(f"api call to 'tm.gd' failed with error '{res['error']}'")
else:
raise ETUMRuntimeError("api call failure in jrpc client to be reported to testium support team.")
raise ETUMRuntimeError("api not initialized")
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", (name, value))
if "result" in res:
return res["result"]
elif "error" in res:
raise ETUMRuntimeError(f"api call to 'tm.setgd' failed with error '{res['error']}'")
else:
raise ETUMRuntimeError("api call failure in jrpc client to be reported to testium support team.")
raise ETUMRuntimeError("api not initialized")
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", (name,))
if "result" in res:
return res["result"]
elif "error" in res:
raise ETUMRuntimeError(f"api call to 'tm.delgd' failed with error '{res['error']}'")
else:
raise ETUMRuntimeError("api call failure in jrpc client to be reported to testium support team.")
raise ETUMRuntimeError("api not initialized")
def _init_api(host, port, timeout):
global _func_call_thread
_func_call_thread = FuncHandler(host, port, timeout=timeout)
return _func_call_thread
@@ -53,17 +125,10 @@ def _init_api(host, port, timeout):
###############################################################################
def _remote_print(*values):
"""Forward print-like output to the remote handler.
If a ``_func_call_thread`` is available, this function calls the
handler with action name ``"print"`` and the provided values. Errors
during forwarding are ignored because printing is best-effort.
"""
if _func_call_thread is not None:
try:
_func_call_thread.call("print", values)
except:
# Best-effort: ignore forwarding failures
pass

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)
@@ -275,6 +285,13 @@ Is the python exec path correct ?"""
# Stop python eval execution process
eval_proc.stop()
eval_proc.join()
# Stop shared func context engines (keep_context_id)
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()
except Exception as e:
print_exception(e)
@@ -297,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,
@@ -330,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 PySide6.QtWidgets import (QApplication, QDialog)
from PySide6.QtCore import (Qt)
from PySide6.QtWidgets import QMessageBox
from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt
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 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)
from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import Qt
def main(args, conn):
try:
from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
msg = QMessageBox()
msg.setWindowFlags(Qt.WindowStaysOnTopHint)
msg.setWindowTitle(args[0])
msg.setText(args[1])
msg.setIcon(QMessageBox.Question)
msg.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
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,16 +19,11 @@ 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",
self.seqFilename(),
)
except:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
f"Missing required 'values' parameter",
self.seqFilename(),
)

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
@@ -12,10 +12,13 @@ from interpreter.utils.lua_func_exec import LuaFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils.constants import TestItemType as cst
_LUA_FUNC_CONTEXTS_KEY = "_lua_func_contexts"
class TestItemLuaFunc(TestItem):
"""lua_func item usage.
func file: func_file.lua, func_name: func, param: [$(variable1), [1, 2, 3], true]
Optional: context_id: <id> — share a persistent process with other lua_func items using the same id.
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
@@ -23,18 +26,25 @@ 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")
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(),
)
# Lua functions call subprocess initialization
self._context_id = self._prms.getParam("context_id", default=None, processed=False)
self._lua_func_proc = LuaFuncExecEngine(tm.gd("lua_bin", ""), api_request, 10)
def _get_engine(self):
"""Return (engine, persistent). If context_id is set, use a shared persistent engine."""
if self._context_id is None:
return self._lua_func_proc, False
ctx_id = self._prms.expanse(self._context_id)
contexts = tm.gd(_LUA_FUNC_CONTEXTS_KEY, {})
if ctx_id not in contexts:
contexts[ctx_id] = LuaFuncExecEngine(tm.gd("lua_bin", ""), api_request, 10)
tm.setgd(_LUA_FUNC_CONTEXTS_KEY, contexts)
return contexts[ctx_id], True
@test_run
def execute(self):
self.result.set(
@@ -48,8 +58,11 @@ class TestItemLuaFunc(TestItem):
print("Parameters list:")
print(textwrap.indent(pprint.pformat(pl), " |"))
self._lua_func_proc.start()
if not self._lua_func_proc.wait_ready(10):
engine, persistent = self._get_engine()
if not engine.is_alive():
engine.start()
if not engine.wait_ready(10):
raise ETUMRuntimeError(
f"""Impossible to start the external lua execution process.
Is the lua path correct ?
@@ -59,11 +72,11 @@ Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables
)
try:
success, ret = self._lua_func_proc.func_call(self.file_name, self.func_name, pl)
success, ret = engine.func_call(self.file_name, self.func_name, pl)
finally:
# Stops lua function execution process
self._lua_func_proc.stop()
self._lua_func_proc.join()
if not persistent:
engine.stop()
engine.join()
if success == TestValue.SUCCESS:
self.result.set(TestValue.SUCCESS)
@@ -73,7 +86,6 @@ Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables
print("Returned value:")
print(textwrap.indent(pprint.pformat(res), " |"))
# The result of the func test item is put in global dir and result
tm.setgd("lfn_" + self._name, res)
self.result.value = res
@@ -88,5 +100,5 @@ Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables
traceback.print_exception(*sys.exc_info())
self.result.set(
TestValue.FAILURE,
'Unrecoverable "py_func" item error from {}'.format(self.func_name),
'Unrecoverable "lua_func" item error from {}'.format(self.func_name),
)

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()
exitcode = self._run_dialog(msg_dialog.main, [self.name(), q])
if exitcode == 0:
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()
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
@@ -12,10 +12,13 @@ from interpreter.utils.py_func_exec import PyFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils.constants import TestItemType as cst
_PY_FUNC_CONTEXTS_KEY = "_py_func_contexts"
class TestItemPyFunc(TestItem):
"""py_func item usage.
func file: func_file.py, func_name: func, param: [$(variable1), [1, 2, 3], true]
Optional: context_id: <id> — share a persistent process with other py_func items using the same id.
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
@@ -23,17 +26,25 @@ 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")
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._context_id = self._prms.getParam("context_id", default=None, processed=False)
self._py_func_proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
def _get_engine(self):
"""Return (engine, persistent). If context_id is set, use a shared persistent engine."""
if self._context_id is None:
return self._py_func_proc, False
ctx_id = self._prms.expanse(self._context_id)
contexts = tm.gd(_PY_FUNC_CONTEXTS_KEY, {})
if ctx_id not in contexts:
contexts[ctx_id] = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
tm.setgd(_PY_FUNC_CONTEXTS_KEY, contexts)
return contexts[ctx_id], True
@test_run
def execute(self):
self.result.set(
@@ -47,20 +58,23 @@ class TestItemPyFunc(TestItem):
print("Parameters list:")
print(textwrap.indent(pprint.pformat(pl), " |"))
# start the process for executing external python
self._py_func_proc.start()
if not self._py_func_proc.wait_ready():
engine, persistent = self._get_engine()
if not engine.is_alive():
engine.start()
if not engine.wait_ready():
raise ETUMRuntimeError(
f"""Impossible to start the external python execution process.
Is the python path correct ?
python_bin = {tm.gd("python_bin", "no python path defined")}"""
)
try:
success, ret = self._py_func_proc.func_call(self.file_name, self.func_name, pl)
success, ret = engine.func_call(self.file_name, self.func_name, pl)
finally:
# Stops python function execution process
self._py_func_proc.stop()
self._py_func_proc.join()
if not persistent:
engine.stop()
engine.join()
if success == TestValue.SUCCESS:
self.result.set(TestValue.SUCCESS)
@@ -70,7 +84,6 @@ python_bin = {tm.gd("python_bin", "no python path defined")}"""
print("Returned value:")
print(textwrap.indent(pprint.pformat(res), " |"))
# The result of the func test item is put in global dir and result
tm.setgd("pfn_" + self._name, res)
self.result.value = res

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"]
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

@@ -221,6 +221,11 @@ class LuaProcessBase:
return self._rpc.wait_ready(timeout)
return False
def is_alive(self):
if self._rpc is not None:
return self._rpc.is_alive()
return False
def stop(self):
"""
Stops the RPC client.

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,8 +22,16 @@
</property>
<layout class="QVBoxLayout" name="verticalLayout_2">
<item>
<layout class="QHBoxLayout" name="horizontalLayout_2"/>
</item>
<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>
<layout class="QFormLayout" name="formLayout">
<item row="0" column="0">
@@ -87,6 +95,68 @@
</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="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>
</widget>
</widget>
</item>
<item>
<layout class="QHBoxLayout" name="horizontalLayout">
<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

@@ -3,7 +3,8 @@ import sys
import traceback
from queue import Empty
from PySide6.QtWidgets import QApplication, QFileDialog
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QApplication, QFileDialog, QProgressDialog
from interpreter.process import TestProcess
from interpreter.utils.test_ctrl import TestSetController
@@ -29,11 +30,18 @@ class TestFileManager:
):
w.test_service.stop()
w.test_service.close()
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
@@ -44,9 +52,25 @@ class TestFileManager:
self.load(file_name)
w.reconnect_signals()
def _make_progress(self, w):
progress = QProgressDialog("Starting test process…", None, 0, 0, w)
progress.setWindowTitle("Loading")
progress.setWindowFlags(Qt.Dialog | Qt.CustomizeWindowHint | Qt.WindowTitleHint)
progress.setWindowModality(Qt.WindowModal)
progress.setMinimumDuration(0)
progress.setMinimumWidth(320)
progress._force_close = False
progress.closeEvent = lambda e: e.accept() if progress._force_close else e.ignore()
return progress
def _close_progress(self, progress):
progress._force_close = True
progress.close()
def load(self, file_name: str) -> bool:
"""Load a test file. Returns True on success, False otherwise."""
w = self._win
progress = None
try:
if not file_name:
raise ETUMFileError("No file to load")
@@ -59,9 +83,14 @@ class TestFileManager:
if not os.path.isfile(file_name):
raise ETUMFileError("Could not find %s file" % file_name)
progress = self._make_progress(w)
progress.show()
QApplication.processEvents()
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,
@@ -71,29 +100,38 @@ class TestFileManager:
self._defaults_for_process(),
)
w.test_proc.start()
progress.setLabelText("Loading test file…")
while w.test_proc.is_alive():
try:
if w.test_service.loaded(timeout=1.0):
if w.test_service.loaded(timeout=0.05):
break
except Empty:
w.test_service.clear()
QApplication.processEvents()
if not w.test_proc.is_alive():
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
raise ETUMRuntimeError(
"Test could not be loaded (test process crashed for any reason)"
)
progress.setLabelText("Building test tree…")
QApplication.processEvents()
test_data = w.test_service.tree()
w.treeTests.clear()
QApplication.processEvents()
w.treeTests.loadTestRecursively(w.treeTests.invisibleRootItem(), test_data)
self._close_progress(progress)
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
@@ -109,6 +147,8 @@ class TestFileManager:
w.show_checkboxes()
return True
except:
if progress is not None:
self._close_progress(progress)
w.statusBar().showMessage("No test file could be loaded", 10000)
w.treeTests.clear()
print(traceback.format_exc())

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

@@ -8,7 +8,6 @@ from multiprocessing import Queue
from queue import Empty
from threading import Thread
import shutil
import ast
# Qt
from PySide6 import QtGui, QtWidgets
@@ -248,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:
@@ -471,21 +472,8 @@ class MainWindow(QMainWindow, Ui_MainWindow):
@Slot()
def on_actionRefresh_test_triggered(self):
self.on_exiting()
args = []
if not hasattr(sys, "frozen"):
args += [sys.executable]
args += [sys.argv[0]]
if len(self.defines) > 0:
for k, v in self.defines.items():
try:
val = ast.literal_eval(v)
except:
val = v
args += ["-d", f"{k}={val}"]
if (self.testFile is not None) and (isinstance(self.testFile, str)):
args += [self.testFile]
os.execv(sys.executable, args)
if self.testFile:
self.file_manager.reload(self.testFile)
@Slot()
def on_actionSave_report_triggered(self):
@@ -553,6 +541,8 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.reconnect_signals()
def on_testChecked(self, item, index):
if index != self.treeTests.cols['name']['index']:
return
self.checkSelect.setCheckState(Qt.PartiallyChecked)
self.disconnect_signals()
try:

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

@@ -32,5 +32,13 @@ function module.tuple_return(first, second)
return first, second
end
function module.set_context_value(val)
tm.setgd("_lua_ctx_test_value", val)
return val
end
function module.get_context_value()
return tm.gd("_lua_ctx_test_value")
end
return module

View File

@@ -179,3 +179,36 @@
file: $(test_path)$(psep)lua_func.lua
func_name: tuple_return
param: [ 0, "OK" ]
- group:
name: context_id tests
steps:
- lua_func:
name: set context value
key: $(test)_PASS
file: $(test_path)$(psep)lua_func.lua
func_name: set_context_value
context_id: lua_ctx_test
param:
- hello lua
expected_result: hello lua
- lua_func:
name: get context value (same context_id)
key: $(test)_PASS
file: $(test_path)$(psep)lua_func.lua
func_name: get_context_value
context_id: lua_ctx_test
expected_result: hello lua
- lua_func:
name: get context value (no context_id, from main gd)
key: $(test)_PASS
file: $(test_path)$(psep)lua_func.lua
func_name: get_context_value
expected_result: hello lua
- lua_func:
name: get context value (different context_id)
key: $(test)_PASS
file: $(test_path)$(psep)lua_func.lua
func_name: get_context_value
context_id: lua_ctx_other
expected_result: hello lua

View File

@@ -27,3 +27,23 @@ def echo(param):
def tuple_return(first, second):
return first, second
def set_context_value(val):
tm.setgd("_py_ctx_test_value", val)
return val
def get_context_value():
return tm.gd("_py_ctx_test_value", None)
class _NotSerializable:
def __init__(self, val):
self.val = val
def set_ns_value(val):
tm.setgd("_py_ctx_ns_value", _NotSerializable(val))
return val
def get_ns_value():
obj = tm.gd("_py_ctx_ns_value", None)
return obj.val if obj is not None else None

View File

@@ -189,3 +189,51 @@
func_name: tuple_return
param: [ 0, "OK" ]
expected_result: [0, "OK"]
- group:
name: context_id tests
steps:
- py_func:
name: set serializable value
key: $(test)_PASS
file: $(test_path)$(psep)py_func.py
func_name: set_context_value
param:
- hello context
expected_result: hello context
- py_func:
name: get serializable value (same context_id)
key: $(test)_PASS
file: $(test_path)$(psep)py_func.py
func_name: get_context_value
context_id: ctx_test
expected_result: hello context
- py_func:
name: get serializable value (no context_id, from main gd)
key: $(test)_PASS
file: $(test_path)$(psep)py_func.py
func_name: get_context_value
expected_result: hello context
- py_func:
name: get serializable value (different context_id)
key: $(test)_PASS
file: $(test_path)$(psep)py_func.py
func_name: get_context_value
context_id: ctx_other
expected_result: hello context
- py_func:
name: set non-serializable value
key: $(test)_PASS
file: $(test_path)$(psep)py_func.py
func_name: set_ns_value
context_id: ctx_ns_test
param:
- hello ns
expected_result: hello ns
- py_func:
name: get non-serializable value (same context_id)
key: $(test)_PASS
file: $(test_path)$(psep)py_func.py
func_name: get_ns_value
context_id: ctx_ns_test
expected_result: hello ns

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