Compare commits
14 Commits
refactor/u
...
f1_variabl
| Author | SHA1 | Date | |
|---|---|---|---|
| d955ae81f9 | |||
| 2cd3aa3305 | |||
| 276d485905 | |||
| 95912dd3e1 | |||
| 6d1fb6a6bc | |||
| 2cc42e9065 | |||
| 2b7678c39e | |||
| c72176d029 | |||
| 617f599f86 | |||
| d92f518e1e | |||
| 49354b8664 | |||
| 7383820aba | |||
| 67c879ab10 | |||
| aa72e349c6 |
@@ -14,6 +14,7 @@ import os
|
||||
|
||||
import sys
|
||||
sys.path.insert(0, os.path.abspath('../../../../src/testium/'))
|
||||
sys.path.insert(0, os.path.abspath('../../../../src/'))
|
||||
|
||||
|
||||
# -- Project information -----------------------------------------------------
|
||||
|
||||
@@ -21,7 +21,7 @@ Global variables helper functions
|
||||
To manage values in the global variables dataset, the following testium library API
|
||||
must be used:
|
||||
|
||||
.. automodule:: interpreter.utils.globdict
|
||||
.. automodule:: py_func.tm
|
||||
:members: gd, setgd, delgd
|
||||
:undoc-members:
|
||||
:no-index:
|
||||
|
||||
@@ -87,6 +87,10 @@ if not provided is given in the table as well.
|
||||
| | | see :ref:`Expected result<sec_expected_result>` |
|
||||
| | | for details. |
|
||||
+-----------------------+-------------------+-------------------------------------------------------+
|
||||
| ``store_result`` | / | Store the test result in a global variable. |
|
||||
| | | see :ref:`Store result<sec_store_result>` |
|
||||
| | | for details. |
|
||||
+-----------------------+-------------------+-------------------------------------------------------+
|
||||
|
||||
|
||||
last test result
|
||||
@@ -183,6 +187,61 @@ If the result and the expected_result is equal, the test will be *PASSED* if ``T
|
||||
The special ``$(result)`` variable is replaced in the ``expected_result`` attribute content with the test result value.
|
||||
|
||||
|
||||
.. _sec_store_result:
|
||||
|
||||
Store result
|
||||
-----------------------------------------------
|
||||
|
||||
The ``store_result`` attribute stores the test result into a named global variable,
|
||||
making it available to subsequent test items via ``$(variable_name)``.
|
||||
|
||||
If the test item returns a value (e.g. ``py_func``, ``json_rpc``), that value is stored.
|
||||
If ``process_result`` is also specified, the stored value is the post-processed result.
|
||||
|
||||
If the test item produces no value (result is ``None``), the stored value is the
|
||||
test status string: ``"PASS"`` or ``"FAIL"``, evaluated after ``expected_result``
|
||||
but **before** ``no_fail``. This ensures the real outcome is captured even when
|
||||
``no_fail: True`` would otherwise mask a failure.
|
||||
|
||||
.. code-block:: yaml
|
||||
:caption: Store a function return value
|
||||
|
||||
- py_func:
|
||||
name: Read sensor
|
||||
func_name: read_temperature
|
||||
store_result: temperature
|
||||
|
||||
- py_func:
|
||||
name: Check temperature in range
|
||||
func_name: check_range
|
||||
param: [$(temperature), 20, 30]
|
||||
|
||||
.. code-block:: yaml
|
||||
:caption: Store a post-processed value
|
||||
|
||||
- py_func:
|
||||
name: Get firmware version string
|
||||
func_name: get_version
|
||||
process_result: "'$(result)'.split('.')[0]"
|
||||
store_result: fw_major
|
||||
|
||||
.. code-block:: yaml
|
||||
:caption: Store the pass/fail status of a test with no return value
|
||||
|
||||
- console:
|
||||
name: Send command
|
||||
console_name: device
|
||||
steps:
|
||||
- writeln: reboot
|
||||
- read_until: {expected: "ready", timeout: 10}
|
||||
store_result: reboot_status
|
||||
|
||||
- py_func:
|
||||
name: Use reboot status
|
||||
func_name: log_status
|
||||
param: [$(reboot_status)]
|
||||
|
||||
|
||||
Export attribute
|
||||
-----------------------------------------------
|
||||
|
||||
|
||||
@@ -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**
|
||||
|
||||
|
||||
@@ -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.
@@ -228,7 +228,7 @@ class JsonRpcConnection:
|
||||
self.recv_thread.join()
|
||||
|
||||
def is_alive(self):
|
||||
self.recv_thread.is_alive()
|
||||
return self.recv_thread.is_alive()
|
||||
|
||||
def wait_ready(self, timeout=None):
|
||||
return self._event_ready.wait(timeout)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import traceback
|
||||
import textwrap
|
||||
from contextlib import contextmanager
|
||||
|
||||
|
||||
class ETUMError(Exception):
|
||||
@@ -67,6 +68,28 @@ class ETUMParamError(ETUMError):
|
||||
return lines
|
||||
|
||||
|
||||
@contextmanager
|
||||
def item_load_context(item_type: str, item_name: str, filename: str = ""):
|
||||
"""Context manager that enriches ETUMSyntaxError with item context during loading.
|
||||
|
||||
Usage in test item __init__:
|
||||
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
|
||||
self.param = self._prms.getParam("param", required=True)
|
||||
"""
|
||||
try:
|
||||
yield
|
||||
except ETUMSyntaxError as e:
|
||||
raise ETUMSyntaxError(
|
||||
f"In '{item_type}' item named '{item_name}':\n{e._message}",
|
||||
filename or e._file,
|
||||
) from e
|
||||
except Exception as e:
|
||||
raise ETUMSyntaxError(
|
||||
f"In '{item_type}' item named '{item_name}':\nUnexpected error: {e}",
|
||||
filename,
|
||||
) from e
|
||||
|
||||
|
||||
def print_exception(exc: ETUMError):
|
||||
if not isinstance(exc, ETUMError):
|
||||
print(traceback.format_exc(4))
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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)
|
||||
|
||||
15
src/testium/interpreter/test_items/dialog_env.py
Normal file
15
src/testium/interpreter/test_items/dialog_env.py
Normal file
@@ -0,0 +1,15 @@
|
||||
"""Qt platform environment setup for dialog subprocesses.
|
||||
|
||||
Call setup() at the start of every dialog subprocess main() function
|
||||
to ensure the correct Qt platform plugin is selected.
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
|
||||
|
||||
def setup():
|
||||
"""Configure the Qt environment for dialog subprocess usage."""
|
||||
if sys.platform.startswith('linux'):
|
||||
# On Linux/Wayland, force X11 (via XWayland) to avoid crashes
|
||||
# when Qt is initialized inside a multiprocessing subprocess.
|
||||
os.environ['QT_QPA_PLATFORM'] = 'xcb'
|
||||
@@ -1,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)
|
||||
|
||||
@@ -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 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:])
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 flush(self): pass
|
||||
def close(self): pass
|
||||
|
||||
# and now redirect all default streams to this dummyStream:
|
||||
sys.stdout = dummyStream()
|
||||
sys.stderr = dummyStream()
|
||||
sys.stdin = dummyStream()
|
||||
|
||||
@@ -39,7 +39,9 @@ class DialogSleepWindow(QDialog, dialog_sleep_win.Ui_SleepDialogWindow):
|
||||
|
||||
def main(args, conn=None):
|
||||
success = True
|
||||
app = QApplication(sys.argv)
|
||||
from interpreter.test_items import dialog_env
|
||||
dialog_env.setup()
|
||||
app = QApplication(['testium'])
|
||||
d = DialogSleepWindow()
|
||||
d.setFixedSize(379,129)
|
||||
d.setWindowFlags(Qt.WindowStaysOnTopHint)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -7,7 +7,7 @@ import libs.testium as tm
|
||||
from interpreter.utils.params import TestItemParams
|
||||
from interpreter.utils.constants import TestItemType as cst_type
|
||||
from interpreter.utils.eval import eval_to_boolean, evaluate, post_evaluate
|
||||
from lib.tum_except import ETUMSyntaxError
|
||||
from lib.tum_except import ETUMSyntaxError, item_load_context
|
||||
|
||||
LOG_TEST_STOP = '<----- step "{}" finished'
|
||||
LOG_TEST_START = '-----> step "{}" started'
|
||||
@@ -101,6 +101,7 @@ class TestItem:
|
||||
self.status_queue = status_queue
|
||||
self._execute_on_stop = False
|
||||
self._post_eval = None
|
||||
self._store_result = None
|
||||
self._expected_result = None
|
||||
self._no_fail = None
|
||||
self._is_stopped = False
|
||||
@@ -131,11 +132,11 @@ class TestItem:
|
||||
if s:
|
||||
try:
|
||||
self.skipped = eval_to_boolean(s)
|
||||
except:
|
||||
except Exception as e:
|
||||
raise ETUMSyntaxError(
|
||||
f"'{self.cmd()}' test item named '{self.name()}':\nskipped expresion can only be a static expression as it is evaluated during loading of TUM : {s}",
|
||||
self.seqFilename(),
|
||||
)
|
||||
) from e
|
||||
# This allow disabling test item directly by using its name inside param.yaml file
|
||||
elif self._name in tm.gd("skipped_test_item", []):
|
||||
self.skipped = True
|
||||
@@ -155,6 +156,9 @@ class TestItem:
|
||||
if "process_result" in dict_item:
|
||||
self._post_eval = dict_item["process_result"]
|
||||
|
||||
if "store_result" in dict_item:
|
||||
self._store_result = dict_item["store_result"]
|
||||
|
||||
if "expected_result" in dict_item:
|
||||
self._expected_result = dict_item["expected_result"]
|
||||
|
||||
@@ -164,11 +168,13 @@ class TestItem:
|
||||
self.banner = LOG_TEST_START.format(self._name)
|
||||
self.footer = LOG_TEST_STOP.format(self._name)
|
||||
|
||||
except:
|
||||
except ETUMSyntaxError:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise ETUMSyntaxError(
|
||||
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
|
||||
f"The '{self.cmd()}' test item named '{self.name()}' has an unexpected loading error: {e}",
|
||||
self.seqFilename(),
|
||||
)
|
||||
) from e
|
||||
|
||||
self.result = TestResult(self, TestValue.FAILURE, "Failure by default")
|
||||
|
||||
@@ -275,6 +281,9 @@ class TestItem:
|
||||
self.process_result()
|
||||
# expected_result treatment
|
||||
self.result_expected()
|
||||
# Store result in a global variable if requested (before no_fail so
|
||||
# the real outcome is captured when result.value is None)
|
||||
self.store_result()
|
||||
# Case of the no_fail true parameter
|
||||
self.process_no_fail()
|
||||
|
||||
@@ -317,6 +326,17 @@ class TestItem:
|
||||
print(e)
|
||||
self.result.set(TestValue.FAILURE, "Result processing failed")
|
||||
|
||||
def store_result(self):
|
||||
if self._store_result is None:
|
||||
return
|
||||
var_name = self._prms.expanse(self._store_result)
|
||||
if self.result.value is None:
|
||||
value = str(self.result.test_result)
|
||||
else:
|
||||
value = self.result.value
|
||||
tm.setgd(var_name, value)
|
||||
print(f"Stored result in '$({var_name})': {value}")
|
||||
|
||||
def process_report(self, report_eval):
|
||||
tm.print_debug(f"Export reported values:")
|
||||
rep_eval = self._prms.expanse(report_eval)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
|
||||
from interpreter.test_items.test_item import (TestItem, test_run)
|
||||
from interpreter.test_items.test_result import TestValue
|
||||
from lib.tum_except import ETUMSyntaxError
|
||||
from lib.tum_except import ETUMSyntaxError, item_load_context
|
||||
import libs.testium as tm
|
||||
from interpreter.utils.constants import TestItemType as cst
|
||||
from interpreter.utils.eval import evaluate
|
||||
@@ -15,21 +15,16 @@ class TestItemCheckValue(TestItem):
|
||||
super().__init__(dict_item, parent, status_queue, filename=filename)
|
||||
self._type = cst.TYPE_CHECK
|
||||
self.is_container = False
|
||||
try:
|
||||
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
|
||||
self._action_list = self._prms.getParamAll('steps', default=[], required=False)
|
||||
if len(self._action_list) > 0:
|
||||
tm.print_warn("'steps' argument of check test item is deprecated and is replaced by 'values'")
|
||||
self._action_list += self._prms.getParamAll('values', default=[], required=False)
|
||||
if len(self._action_list) <= 0:
|
||||
raise ETUMSyntaxError(
|
||||
f" The '{self.cmd()}' test item named '{self.name()}' must have a 'values' parameter",
|
||||
f"Missing required 'values' parameter",
|
||||
self.seqFilename()
|
||||
)
|
||||
except:
|
||||
raise ETUMSyntaxError(
|
||||
f"The '{self.cmd()}' test item named '{self.name()}' (a child of: '{self.parent().name()}') has a missing or wrong parameter",
|
||||
self.seqFilename(),
|
||||
)
|
||||
|
||||
@test_run
|
||||
def execute(self):
|
||||
|
||||
@@ -1,48 +1,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:
|
||||
|
||||
46
src/testium/interpreter/test_items/test_item_dialog_base.py
Normal file
46
src/testium/interpreter/test_items/test_item_dialog_base.py
Normal 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
|
||||
@@ -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()
|
||||
|
||||
@@ -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(),
|
||||
)
|
||||
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
|
||||
@@ -1,14 +1,15 @@
|
||||
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
|
||||
"""
|
||||
@@ -17,38 +18,15 @@ class TestItemMsgDialog(TestItem):
|
||||
super().__init__(dict_item, parent, status_queue, filename=filename)
|
||||
self._type = cst.TYPE_MESSAGE_DLG
|
||||
self.is_container = False
|
||||
try:
|
||||
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
|
||||
self._question = self._prms.getParam('question', required=True)
|
||||
except:
|
||||
raise ETUMSyntaxError(
|
||||
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
|
||||
self.seqFilename(),
|
||||
)
|
||||
|
||||
@test_run
|
||||
def execute(self):
|
||||
ourpath = __file__
|
||||
test_file = os.path.join(os.path.dirname(ourpath),
|
||||
'dialog_msg_files',
|
||||
'msg_dialog.py')
|
||||
|
||||
q = self._prms.expanse(self._question)
|
||||
print("Message Displayed:\n" + q)
|
||||
parent_conn, child_conn = Pipe()
|
||||
p=Process(target=msg_dialog.main,
|
||||
args=([self.name(), q],))
|
||||
p.start()
|
||||
p.join()
|
||||
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}")
|
||||
|
||||
@@ -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):
|
||||
|
||||
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:
|
||||
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
|
||||
self._question = self._prms.getParam('question', required=True)
|
||||
except:
|
||||
raise ETUMSyntaxError(
|
||||
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
|
||||
self.seqFilename(),
|
||||
)
|
||||
|
||||
@test_run
|
||||
def execute(self):
|
||||
ourpath = __file__
|
||||
test_file = os.path.join(os.path.dirname(ourpath),
|
||||
'dialog_note_files',
|
||||
'test_dialog.py')
|
||||
|
||||
q = self._prms.expanse(self._question)
|
||||
print("Question:\n" + q)
|
||||
parent_conn, child_conn = Pipe()
|
||||
p=Process(target=test_dialog.main, args=([self.name(), q],child_conn))
|
||||
p.start()
|
||||
val, succ = parent_conn.recv()
|
||||
p.join()
|
||||
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()
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -1,16 +1,14 @@
|
||||
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"
|
||||
"""
|
||||
@@ -19,44 +17,20 @@ class TestItemQuestionDialog(TestItem):
|
||||
super().__init__(dict_item, parent, status_queue, filename=filename)
|
||||
self._type = cst.TYPE_QUESTION_DLG
|
||||
self.is_container = False
|
||||
try:
|
||||
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
|
||||
self._question = self._prms.getParam('question', required=True)
|
||||
except:
|
||||
raise ETUMSyntaxError(
|
||||
f"The '{self.cmd()}' test item named '{self.name()}' has a missing or wrong parameter",
|
||||
self.seqFilename(),
|
||||
)
|
||||
|
||||
@test_run
|
||||
def execute(self):
|
||||
ourpath = __file__
|
||||
test_file = os.path.join(os.path.dirname(ourpath),
|
||||
'dialog_question_files',
|
||||
'question_dialog.py')
|
||||
|
||||
q = self._prms.expanse(self._question)
|
||||
print('Question asked:\n' + q + '\n')
|
||||
parent_conn, child_conn = Pipe()
|
||||
p=Process(target=question_dialog.main,
|
||||
args=([self.name(), q],child_conn))
|
||||
p.start()
|
||||
succ = parent_conn.recv()
|
||||
p.join()
|
||||
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()
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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:
|
||||
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):
|
||||
|
||||
@@ -1,45 +1,31 @@
|
||||
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)
|
||||
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()
|
||||
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
|
||||
|
||||
titems = []
|
||||
if len(val) > 0:
|
||||
@@ -53,7 +39,7 @@ class TestItemTestedRefsDialog(TestItem):
|
||||
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()
|
||||
|
||||
@@ -1,15 +1,13 @@
|
||||
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?"
|
||||
"""
|
||||
@@ -18,30 +16,20 @@ class TestItemValueDialog(TestItem):
|
||||
super().__init__(dict_item, parent, status_queue, filename=filename)
|
||||
self._type = cst.TYPE_VALUE_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._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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
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):
|
||||
super().__init__(parent)
|
||||
self.ui = Ui_F1Dialog()
|
||||
self.ui.setupUi(self)
|
||||
self.highlighter = YamlHighlighter(self.ui.TestContentEdit.document())
|
||||
self.setWindowFlags(
|
||||
Qt.Window | Qt.WindowStaysOnTopHint | Qt.Tool
|
||||
)
|
||||
self.setWindowFlags(Qt.Window | Qt.WindowStaysOnTopHint | Qt.Tool)
|
||||
|
||||
self.ui.ButtLocOpen.clicked.connect(self.on_butlocopen_click)
|
||||
self.ui.ButtClose.clicked.connect(self.close)
|
||||
|
||||
self._service = None
|
||||
self._key_rows = {}
|
||||
self._updating = False
|
||||
self._mono_font = QFont("Monospace")
|
||||
self._mono_font.setStyleHint(QFont.StyleHint.TypeWriter)
|
||||
self._mono_bold_font = QFont("Monospace")
|
||||
self._mono_bold_font.setStyleHint(QFont.StyleHint.TypeWriter)
|
||||
self._mono_bold_font.setBold(True)
|
||||
|
||||
self._setup_vars_tab()
|
||||
|
||||
def _setup_vars_tab(self):
|
||||
table = self.ui.varsTable
|
||||
table.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.ResizeToContents)
|
||||
table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch)
|
||||
table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.Fixed)
|
||||
table.setColumnWidth(2, 36)
|
||||
table.verticalHeader().setVisible(False)
|
||||
table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
|
||||
table.customContextMenuRequested.connect(self._on_context_menu)
|
||||
table.cellChanged.connect(self._on_cell_changed)
|
||||
table.setEnabled(False)
|
||||
self.ui.addVarButton.setEnabled(False)
|
||||
self.ui.addVarButton.clicked.connect(self._on_add_var)
|
||||
|
||||
def load_initial_vars(self, vars_dict: dict):
|
||||
for key, value in vars_dict.items():
|
||||
self.gd_var_updated(key, value)
|
||||
|
||||
def set_service(self, service):
|
||||
self._service = service
|
||||
enabled = service is not None
|
||||
self.ui.varsTable.setEnabled(enabled)
|
||||
self.ui.addVarButton.setEnabled(enabled)
|
||||
if not enabled:
|
||||
self._updating = True
|
||||
try:
|
||||
self.ui.varsTable.setRowCount(0)
|
||||
finally:
|
||||
self._updating = False
|
||||
self._key_rows.clear()
|
||||
|
||||
@Slot(str, object)
|
||||
def gd_var_updated(self, key, value):
|
||||
if key in self._key_rows:
|
||||
self._refresh_row(self._key_rows[key], key, value)
|
||||
else:
|
||||
self._updating = True
|
||||
try:
|
||||
row = self.ui.varsTable.rowCount()
|
||||
self.ui.varsTable.insertRow(row)
|
||||
finally:
|
||||
self._updating = False
|
||||
self._key_rows[key] = row
|
||||
self._refresh_row(row, key, value)
|
||||
|
||||
@Slot(str)
|
||||
def gd_var_deleted(self, key):
|
||||
if key not in self._key_rows:
|
||||
return
|
||||
row = self._key_rows.pop(key)
|
||||
self._updating = True
|
||||
try:
|
||||
self.ui.varsTable.removeRow(row)
|
||||
finally:
|
||||
self._updating = False
|
||||
self._key_rows = {k: (r - 1 if r > row else r) for k, r in self._key_rows.items()}
|
||||
|
||||
def _refresh_row(self, row, key, value):
|
||||
from PySide6.QtWidgets import QTableWidgetItem
|
||||
self._updating = True
|
||||
try:
|
||||
table = self.ui.varsTable
|
||||
|
||||
key_item = QTableWidgetItem(key)
|
||||
key_item.setFlags(key_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
|
||||
key_item.setFont(self._mono_bold_font)
|
||||
table.setItem(row, 0, key_item)
|
||||
|
||||
display = self._display_value(value)
|
||||
val_item = QTableWidgetItem(display)
|
||||
val_item.setData(Qt.ItemDataRole.UserRole, value)
|
||||
val_item.setToolTip(self._full_tooltip(value))
|
||||
val_item.setFont(self._mono_font)
|
||||
if self._is_complex(value):
|
||||
val_item.setFlags(val_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
|
||||
table.setItem(row, 1, val_item)
|
||||
|
||||
if self._is_complex(value):
|
||||
btn = QPushButton("[…]")
|
||||
captured_key = key
|
||||
btn.clicked.connect(lambda: self._on_edit_complex(captured_key))
|
||||
table.setCellWidget(row, 2, btn)
|
||||
else:
|
||||
table.setCellWidget(row, 2, None)
|
||||
table.setItem(row, 2, QTableWidgetItem())
|
||||
finally:
|
||||
self._updating = False
|
||||
|
||||
def _is_complex(self, value):
|
||||
return isinstance(value, (dict, list))
|
||||
|
||||
def _display_value(self, value):
|
||||
if self._is_complex(value):
|
||||
text = repr(value)
|
||||
return (text[:60] + "…") if len(text) > 60 else text
|
||||
return repr(value)
|
||||
|
||||
def _full_tooltip(self, value):
|
||||
try:
|
||||
text = json.dumps(value, indent=2)
|
||||
except (TypeError, ValueError):
|
||||
text = repr(value)
|
||||
escaped = text.replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
return f"<pre>{escaped}</pre>"
|
||||
|
||||
def _on_cell_changed(self, row, col):
|
||||
if self._updating or col != 1 or self._service is None:
|
||||
return
|
||||
from PySide6.QtWidgets import QTableWidgetItem
|
||||
key_item = self.ui.varsTable.item(row, 0)
|
||||
val_item = self.ui.varsTable.item(row, 1)
|
||||
if key_item is None or val_item is None:
|
||||
return
|
||||
key = key_item.text()
|
||||
text = val_item.text()
|
||||
try:
|
||||
value = ast.literal_eval(text)
|
||||
except (ValueError, SyntaxError):
|
||||
value = text
|
||||
self._service.set_gd_var(key, value)
|
||||
|
||||
def _on_edit_complex(self, key):
|
||||
if key not in self._key_rows:
|
||||
return
|
||||
val_item = self.ui.varsTable.item(self._key_rows[key], 1)
|
||||
if val_item is None:
|
||||
return
|
||||
value = val_item.data(Qt.ItemDataRole.UserRole)
|
||||
dlg = GdVarEditDialog(key, value, self)
|
||||
if dlg.exec() == QDialog.DialogCode.Accepted and self._service is not None:
|
||||
self._service.set_gd_var(key, dlg.result_value)
|
||||
|
||||
def _on_add_var(self):
|
||||
key = self.ui.newKeyEdit.text().strip()
|
||||
value_text = self.ui.newValueEdit.text().strip()
|
||||
if not key or self._service is None:
|
||||
return
|
||||
try:
|
||||
value = ast.literal_eval(value_text)
|
||||
except (ValueError, SyntaxError):
|
||||
value = value_text
|
||||
self._service.set_gd_var(key, value)
|
||||
self.ui.newKeyEdit.clear()
|
||||
self.ui.newValueEdit.clear()
|
||||
|
||||
def _on_context_menu(self, pos):
|
||||
row = self.ui.varsTable.rowAt(pos.y())
|
||||
if row < 0:
|
||||
return
|
||||
key_item = self.ui.varsTable.item(row, 0)
|
||||
if key_item is None or self._service is None:
|
||||
return
|
||||
key = key_item.text()
|
||||
menu = QMenu(self)
|
||||
delete_action = menu.addAction("Delete")
|
||||
if menu.exec(self.ui.varsTable.mapToGlobal(pos)) == delete_action:
|
||||
self._service.del_gd_var(key)
|
||||
|
||||
def on_butlocopen_click(self):
|
||||
file = self.ui.sequenceFileNameLineEdit.text()
|
||||
if os.path.exists(file):
|
||||
if sys.platform.startswith("win"): # Windows
|
||||
if sys.platform.startswith("win"):
|
||||
subprocess.Popen(f'explorer "{file}"')
|
||||
else: # Linux / autres
|
||||
else:
|
||||
subprocess.Popen(["xdg-open", file])
|
||||
QDesktopServices.openUrl(QUrl.fromLocalFile(file))
|
||||
@@ -3,7 +3,7 @@
|
||||
################################################################################
|
||||
## Form generated from reading UI file 'f1_win_core.ui'
|
||||
##
|
||||
## Created by: Qt User Interface Compiler version 6.10.1
|
||||
## Created by: Qt User Interface Compiler version 6.11.0
|
||||
##
|
||||
## WARNING! All changes made in this file will be lost when recompiling UI file!
|
||||
################################################################################
|
||||
@@ -16,8 +16,9 @@ from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor,
|
||||
QImage, QKeySequence, QLinearGradient, QPainter,
|
||||
QPalette, QPixmap, QRadialGradient, QTransform)
|
||||
from PySide6.QtWidgets import (QApplication, QDialog, QFormLayout, QHBoxLayout,
|
||||
QLabel, QLineEdit, QPushButton, QSizePolicy,
|
||||
QSpacerItem, QTextEdit, QToolButton, QVBoxLayout,
|
||||
QHeaderView, QLabel, QLineEdit, QPushButton,
|
||||
QSizePolicy, QSpacerItem, QTabWidget, QTableWidget,
|
||||
QTableWidgetItem, QTextEdit, QToolButton, QVBoxLayout,
|
||||
QWidget)
|
||||
import f1_win_rc
|
||||
|
||||
@@ -25,7 +26,7 @@ class Ui_F1Dialog(object):
|
||||
def setupUi(self, F1Dialog):
|
||||
if not F1Dialog.objectName():
|
||||
F1Dialog.setObjectName(u"F1Dialog")
|
||||
F1Dialog.resize(400, 300)
|
||||
F1Dialog.resize(550, 450)
|
||||
icon = QIcon()
|
||||
if QIcon.hasThemeIcon(QIcon.ThemeIcon.HelpAbout):
|
||||
icon = QIcon.fromTheme(QIcon.ThemeIcon.HelpAbout)
|
||||
@@ -36,19 +37,20 @@ class Ui_F1Dialog(object):
|
||||
F1Dialog.setLayoutDirection(Qt.LayoutDirection.LeftToRight)
|
||||
self.verticalLayout_2 = QVBoxLayout(F1Dialog)
|
||||
self.verticalLayout_2.setObjectName(u"verticalLayout_2")
|
||||
self.horizontalLayout_2 = QHBoxLayout()
|
||||
self.horizontalLayout_2.setObjectName(u"horizontalLayout_2")
|
||||
|
||||
self.verticalLayout_2.addLayout(self.horizontalLayout_2)
|
||||
|
||||
self.tabWidget = QTabWidget(F1Dialog)
|
||||
self.tabWidget.setObjectName(u"tabWidget")
|
||||
self.tabTestItem = QWidget()
|
||||
self.tabTestItem.setObjectName(u"tabTestItem")
|
||||
self.verticalLayout_tab0 = QVBoxLayout(self.tabTestItem)
|
||||
self.verticalLayout_tab0.setObjectName(u"verticalLayout_tab0")
|
||||
self.formLayout = QFormLayout()
|
||||
self.formLayout.setObjectName(u"formLayout")
|
||||
self.typeLabel = QLabel(F1Dialog)
|
||||
self.typeLabel = QLabel(self.tabTestItem)
|
||||
self.typeLabel.setObjectName(u"typeLabel")
|
||||
|
||||
self.formLayout.setWidget(0, QFormLayout.ItemRole.LabelRole, self.typeLabel)
|
||||
|
||||
self.typeLineEdit = QLineEdit(F1Dialog)
|
||||
self.typeLineEdit = QLineEdit(self.tabTestItem)
|
||||
self.typeLineEdit.setObjectName(u"typeLineEdit")
|
||||
sizePolicy = QSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
|
||||
sizePolicy.setHorizontalStretch(0)
|
||||
@@ -59,20 +61,20 @@ class Ui_F1Dialog(object):
|
||||
|
||||
self.formLayout.setWidget(0, QFormLayout.ItemRole.FieldRole, self.typeLineEdit)
|
||||
|
||||
self.sequenceFileNameLabel = QLabel(F1Dialog)
|
||||
self.sequenceFileNameLabel = QLabel(self.tabTestItem)
|
||||
self.sequenceFileNameLabel.setObjectName(u"sequenceFileNameLabel")
|
||||
|
||||
self.formLayout.setWidget(1, QFormLayout.ItemRole.LabelRole, self.sequenceFileNameLabel)
|
||||
|
||||
self.horizontalLayout_3 = QHBoxLayout()
|
||||
self.horizontalLayout_3.setObjectName(u"horizontalLayout_3")
|
||||
self.sequenceFileNameLineEdit = QLineEdit(F1Dialog)
|
||||
self.sequenceFileNameLineEdit = QLineEdit(self.tabTestItem)
|
||||
self.sequenceFileNameLineEdit.setObjectName(u"sequenceFileNameLineEdit")
|
||||
self.sequenceFileNameLineEdit.setReadOnly(True)
|
||||
|
||||
self.horizontalLayout_3.addWidget(self.sequenceFileNameLineEdit)
|
||||
|
||||
self.ButtLocOpen = QToolButton(F1Dialog)
|
||||
self.ButtLocOpen = QToolButton(self.tabTestItem)
|
||||
self.ButtLocOpen.setObjectName(u"ButtLocOpen")
|
||||
|
||||
self.horizontalLayout_3.addWidget(self.ButtLocOpen)
|
||||
@@ -81,18 +83,61 @@ class Ui_F1Dialog(object):
|
||||
self.formLayout.setLayout(1, QFormLayout.ItemRole.FieldRole, self.horizontalLayout_3)
|
||||
|
||||
|
||||
self.verticalLayout_2.addLayout(self.formLayout)
|
||||
self.verticalLayout_tab0.addLayout(self.formLayout)
|
||||
|
||||
self.label = QLabel(F1Dialog)
|
||||
self.label = QLabel(self.tabTestItem)
|
||||
self.label.setObjectName(u"label")
|
||||
|
||||
self.verticalLayout_2.addWidget(self.label)
|
||||
self.verticalLayout_tab0.addWidget(self.label)
|
||||
|
||||
self.TestContentEdit = QTextEdit(F1Dialog)
|
||||
self.TestContentEdit = QTextEdit(self.tabTestItem)
|
||||
self.TestContentEdit.setObjectName(u"TestContentEdit")
|
||||
self.TestContentEdit.setReadOnly(True)
|
||||
|
||||
self.verticalLayout_2.addWidget(self.TestContentEdit)
|
||||
self.verticalLayout_tab0.addWidget(self.TestContentEdit)
|
||||
|
||||
self.tabWidget.addTab(self.tabTestItem, "")
|
||||
self.tabVariables = QWidget()
|
||||
self.tabVariables.setObjectName(u"tabVariables")
|
||||
self.verticalLayout_tab1 = QVBoxLayout(self.tabVariables)
|
||||
self.verticalLayout_tab1.setObjectName(u"verticalLayout_tab1")
|
||||
self.varsTable = QTableWidget(self.tabVariables)
|
||||
if (self.varsTable.columnCount() < 3):
|
||||
self.varsTable.setColumnCount(3)
|
||||
__qtablewidgetitem = QTableWidgetItem()
|
||||
self.varsTable.setHorizontalHeaderItem(0, __qtablewidgetitem)
|
||||
__qtablewidgetitem1 = QTableWidgetItem()
|
||||
self.varsTable.setHorizontalHeaderItem(1, __qtablewidgetitem1)
|
||||
__qtablewidgetitem2 = QTableWidgetItem()
|
||||
self.varsTable.setHorizontalHeaderItem(2, __qtablewidgetitem2)
|
||||
self.varsTable.setObjectName(u"varsTable")
|
||||
|
||||
self.verticalLayout_tab1.addWidget(self.varsTable)
|
||||
|
||||
self.addVarLayout = QHBoxLayout()
|
||||
self.addVarLayout.setObjectName(u"addVarLayout")
|
||||
self.newKeyEdit = QLineEdit(self.tabVariables)
|
||||
self.newKeyEdit.setObjectName(u"newKeyEdit")
|
||||
|
||||
self.addVarLayout.addWidget(self.newKeyEdit)
|
||||
|
||||
self.newValueEdit = QLineEdit(self.tabVariables)
|
||||
self.newValueEdit.setObjectName(u"newValueEdit")
|
||||
|
||||
self.addVarLayout.addWidget(self.newValueEdit)
|
||||
|
||||
self.addVarButton = QPushButton(self.tabVariables)
|
||||
self.addVarButton.setObjectName(u"addVarButton")
|
||||
self.addVarButton.setMaximumSize(QSize(30, 16777215))
|
||||
|
||||
self.addVarLayout.addWidget(self.addVarButton)
|
||||
|
||||
|
||||
self.verticalLayout_tab1.addLayout(self.addVarLayout)
|
||||
|
||||
self.tabWidget.addTab(self.tabVariables, "")
|
||||
|
||||
self.verticalLayout_2.addWidget(self.tabWidget)
|
||||
|
||||
self.horizontalLayout = QHBoxLayout()
|
||||
self.horizontalLayout.setObjectName(u"horizontalLayout")
|
||||
@@ -113,6 +158,9 @@ class Ui_F1Dialog(object):
|
||||
|
||||
self.retranslateUi(F1Dialog)
|
||||
|
||||
self.tabWidget.setCurrentIndex(0)
|
||||
|
||||
|
||||
QMetaObject.connectSlotsByName(F1Dialog)
|
||||
# setupUi
|
||||
|
||||
@@ -122,6 +170,15 @@ class Ui_F1Dialog(object):
|
||||
self.sequenceFileNameLabel.setText(QCoreApplication.translate("F1Dialog", u"Test file name", None))
|
||||
self.ButtLocOpen.setText(QCoreApplication.translate("F1Dialog", u"...", None))
|
||||
self.label.setText(QCoreApplication.translate("F1Dialog", u"Test content:", None))
|
||||
self.tabWidget.setTabText(self.tabWidget.indexOf(self.tabTestItem), QCoreApplication.translate("F1Dialog", u"Test item", None))
|
||||
___qtablewidgetitem = self.varsTable.horizontalHeaderItem(0)
|
||||
___qtablewidgetitem.setText(QCoreApplication.translate("F1Dialog", u"Key", None))
|
||||
___qtablewidgetitem1 = self.varsTable.horizontalHeaderItem(1)
|
||||
___qtablewidgetitem1.setText(QCoreApplication.translate("F1Dialog", u"Value", None))
|
||||
self.newKeyEdit.setPlaceholderText(QCoreApplication.translate("F1Dialog", u"New key", None))
|
||||
self.newValueEdit.setPlaceholderText(QCoreApplication.translate("F1Dialog", u"Value", None))
|
||||
self.addVarButton.setText(QCoreApplication.translate("F1Dialog", u"+", None))
|
||||
self.tabWidget.setTabText(self.tabWidget.indexOf(self.tabVariables), QCoreApplication.translate("F1Dialog", u"Variables", None))
|
||||
self.ButtClose.setText(QCoreApplication.translate("F1Dialog", u"Close", None))
|
||||
# retranslateUi
|
||||
|
||||
|
||||
@@ -6,8 +6,8 @@
|
||||
<rect>
|
||||
<x>0</x>
|
||||
<y>0</y>
|
||||
<width>400</width>
|
||||
<height>300</height>
|
||||
<width>550</width>
|
||||
<height>450</height>
|
||||
</rect>
|
||||
</property>
|
||||
<property name="windowTitle">
|
||||
@@ -22,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>
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
134
test/validation/items/jsonrpc/jrpc_echo_server.py
Normal file
134
test/validation/items/jsonrpc/jrpc_echo_server.py
Normal file
@@ -0,0 +1,134 @@
|
||||
#!/usr/bin/env python3
|
||||
"""JSON-RPC echo server for the testium validation suite.
|
||||
|
||||
Listens on TCP (newline-delimited JSON) and UDP.
|
||||
Supports JSON-RPC 1.0 and 2.0.
|
||||
|
||||
Handlers:
|
||||
echo(*args) -> [args, {}]
|
||||
<unknown> -> error {code: -32000, message: "function not found"}
|
||||
|
||||
Usage:
|
||||
python3 jrpc_echo_server.py -c jrpces.ini
|
||||
"""
|
||||
import argparse
|
||||
import configparser
|
||||
import json
|
||||
import socket
|
||||
import sys
|
||||
import threading
|
||||
|
||||
|
||||
def _dispatch(method, params):
|
||||
if method == "echo":
|
||||
if not isinstance(params, list):
|
||||
params = [params]
|
||||
return True, [params, {}]
|
||||
return False, {"code": -32000, "message": "function not found"}
|
||||
|
||||
|
||||
def _build_response(req, success, data):
|
||||
req_id = req.get("id", None)
|
||||
if req.get("jsonrpc") == "2.0":
|
||||
if success:
|
||||
return {"jsonrpc": "2.0", "result": data, "id": req_id}
|
||||
else:
|
||||
return {"jsonrpc": "2.0", "error": data, "id": req_id}
|
||||
else:
|
||||
if success:
|
||||
return {"result": data, "error": None, "id": req_id}
|
||||
else:
|
||||
return {"result": None, "error": data, "id": req_id}
|
||||
|
||||
|
||||
def handle(raw: str) -> str:
|
||||
try:
|
||||
req = json.loads(raw)
|
||||
method = req.get("method", "")
|
||||
params = req.get("params", [])
|
||||
success, data = _dispatch(method, params)
|
||||
return json.dumps(_build_response(req, success, data))
|
||||
except Exception as exc:
|
||||
return json.dumps({"result": None, "error": {"code": -32700, "message": str(exc)}, "id": None})
|
||||
|
||||
|
||||
# ── TCP ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def _handle_tcp_client(conn):
|
||||
buf = b""
|
||||
with conn:
|
||||
conn.settimeout(5.0)
|
||||
while True:
|
||||
try:
|
||||
chunk = conn.recv(4096)
|
||||
except (socket.timeout, ConnectionResetError, OSError):
|
||||
break
|
||||
if not chunk:
|
||||
break
|
||||
buf += chunk
|
||||
while b"\n" in buf:
|
||||
line, buf = buf.split(b"\n", 1)
|
||||
line = line.strip()
|
||||
if line:
|
||||
resp = handle(line.decode())
|
||||
conn.sendall((resp + "\n").encode())
|
||||
|
||||
|
||||
def _tcp_server(host, port):
|
||||
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
srv.bind((host, port))
|
||||
srv.listen(5)
|
||||
srv.settimeout(1.0)
|
||||
print(f"TCP listening on {host}:{port}", flush=True)
|
||||
while True:
|
||||
try:
|
||||
conn, _ = srv.accept()
|
||||
except socket.timeout:
|
||||
continue
|
||||
threading.Thread(target=_handle_tcp_client, args=(conn,), daemon=True).start()
|
||||
|
||||
|
||||
# ── UDP ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def _udp_server(host, port):
|
||||
srv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
srv.bind((host, port))
|
||||
print(f"UDP listening on {host}:{port}", flush=True)
|
||||
while True:
|
||||
data, addr = srv.recvfrom(65535)
|
||||
resp = handle(data.decode())
|
||||
srv.sendto(resp.encode(), addr)
|
||||
|
||||
|
||||
# ── Main ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="JSON-RPC echo server")
|
||||
parser.add_argument("-c", "--config", required=True, help="Path to .ini config file")
|
||||
args = parser.parse_args()
|
||||
|
||||
cfg = configparser.ConfigParser()
|
||||
cfg.read(args.config)
|
||||
|
||||
tcp_host = cfg.get("jsonrpc_tcp", "host", fallback="0.0.0.0")
|
||||
tcp_port = cfg.getint("jsonrpc_tcp", "port", fallback=4321)
|
||||
udp_host = cfg.get("jsonrpc_udp", "host", fallback="0.0.0.0")
|
||||
udp_port = cfg.getint("jsonrpc_udp", "port", fallback=4323)
|
||||
|
||||
tcp_thread = threading.Thread(target=_tcp_server, args=(tcp_host, tcp_port), daemon=True)
|
||||
udp_thread = threading.Thread(target=_udp_server, args=(udp_host, udp_port), daemon=True)
|
||||
tcp_thread.start()
|
||||
udp_thread.start()
|
||||
|
||||
print("JSON-RPC echo server ready", flush=True)
|
||||
|
||||
try:
|
||||
tcp_thread.join()
|
||||
except KeyboardInterrupt:
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,27 +1,27 @@
|
||||
|
||||
- console:
|
||||
name: json rpc echo server
|
||||
doc: check if the jsonrpc echo server is installed
|
||||
doc: check if jrpc_echo_server.py is available
|
||||
console_name: jrpces
|
||||
key: $(test)_PASS
|
||||
steps:
|
||||
- open:
|
||||
protocol: terminal
|
||||
- read_until: {expected: $(terminal_prompt), timeout: 1, no_fail: True}
|
||||
- writeln: which jrpces
|
||||
- read_until: {expected: jrpces, timeout: 2}
|
||||
- writeln: test -f {{include_directory}}/jrpc_echo_server.py && echo JRPC_OK
|
||||
- read_until: {expected: JRPC_OK, timeout: 2, no_fail: True}
|
||||
|
||||
- group:
|
||||
name: jsonrpc tests
|
||||
condition: <| '/jrpces' in r'''$(cn_json rpc echo server)''' |>
|
||||
condition: <| 'JRPC_OK' in r'''$(cn_json rpc echo server)''' |>
|
||||
steps:
|
||||
- console:
|
||||
name: Start the json rpc echo server
|
||||
console_name: jrpces
|
||||
key: $(test)_PASS
|
||||
steps:
|
||||
- writeln: jrpces -c {{include_directory}}/jrpces.ini
|
||||
- read_until: {expected: $(terminal_prompt), timeout: 1, no_fail: True}
|
||||
- writeln: python3 {{include_directory}}/jrpc_echo_server.py -c {{include_directory}}/jrpces.ini
|
||||
- read_until: {expected: ready, timeout: 5}
|
||||
|
||||
- console:
|
||||
name: Open the raw tcp Console
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -30,8 +30,10 @@ linux_prompt: "$ "
|
||||
inc_no_template: "inc no template"
|
||||
inc_with_template: "inc with template"
|
||||
|
||||
LUA_PATH_Linux: /usr/share/lua/5.4/?.lua;/usr/local/share/lua/5.4/?.lua;/usr/local/share/lua/5.4/?/init.lua;/usr/share/lua/5.4/?/init.lua;/usr/local/lib/lua/5.4/?.lua;/usr/local/lib/lua/5.4/?/init.lua;/usr/lib/lua/5.4/?.lua;/usr/lib/lua/5.4/?/init.lua;./?.lua;./?/init.lua;/home/francois/.luarocks/share/lua/5.4/?.lua;/home/francois/.luarocks/share/lua/5.4/?/init.lua
|
||||
LUA_CPATH_Linux: /usr/local/lib/lua/5.4/?.so;/usr/lib/lua/5.4/?.so;/usr/local/lib/lua/5.4/loadall.so;/usr/lib/lua/5.4/loadall.so;./?.so;/home/francois/.luarocks/lib/lua/5.4/?.so
|
||||
lua_rev: 5.5
|
||||
|
||||
LUA_PATH_Linux: /usr/share/lua/$(lua_rev)/?.lua;/usr/local/share/lua/$(lua_rev)/?.lua;/usr/local/share/lua/$(lua_rev)/?/init.lua;/usr/share/lua/$(lua_rev)/?/init.lua;/usr/local/lib/lua/$(lua_rev)/?.lua;/usr/local/lib/lua/$(lua_rev)/?/init.lua;/usr/lib/lua/$(lua_rev)/?.lua;/usr/lib/lua/$(lua_rev)/?/init.lua;./?.lua;./?/init.lua;/home/francois/.luarocks/share/lua/$(lua_rev)/?.lua;/home/francois/.luarocks/share/lua/$(lua_rev)/?/init.lua
|
||||
LUA_CPATH_Linux: /usr/local/lib/lua/$(lua_rev)/?.so;/usr/lib/lua/$(lua_rev)/?.so;/usr/local/lib/lua/$(lua_rev)/loadall.so;/usr/lib/lua/$(lua_rev)/loadall.so;./?.so;/home/francois/.luarocks/lib/lua/$(lua_rev)/?.so
|
||||
PATH_Linux:
|
||||
|
||||
LUA_PATH_Windows: ;.\?.lua;C:\Lua\5.1\lua\?.lua;C:\Lua\5.1\lua\?\init.lua;C:\Lua\5.1\?.lua;C:\Lua\5.1\?\init.lua;C:\Lua\5.1\lua\?.luac
|
||||
|
||||
Reference in New Issue
Block a user