v-and-v/testium
2
0
Fork 0
Code Issues 5 Pull Requests 1 Actions Packages Projects Releases 8 Wiki Activity

lua and python bin detection rationalized: bins.py module created.

Browse Source 
Added some api accessible from python and lua sub_processes. Now the tests only access to py_func.tm instead of direct api.testium module access.

Corrected some f"xxx" to allow working with old python (bookworm).

Changed param.yaml of the test to allow lua to work in all situations.

Various other small fixes for frozen app, wheel.

Tested in all situations, and OK. Ready for tag !

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Foué
2026-05-03 10:16:56 +02:00
parent 077e1a97c1
commit d3c5bd01e5
30 changed files with 585 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/testium/api/runtime_plot.py
View File

@@ -270,7 +270,7 @@ class RuntimePlotPeriodic(PeriodicTimer):
self.func_name = func_name
self.args = args
self.post_eval = post_eval
self.proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
self.proc = PyFuncExecEngine(api_request, 10)
self.proc.start()
if not self.proc.wait_ready(10):
raise ETUMRuntimeError(

2
src/testium/interpreter/process.py
View File

@@ -211,7 +211,7 @@ class TestProcess(Process):
env_init()
# Creation of the python evaluation process for loading of the complete test
eval_proc = eval_process_init("", api_request, 10, test_dir)
eval_proc = eval_process_init(api_request, 10, test_dir)
eval_proc.start()
tm.print_debug(f"python bin is: '{eval_proc.python_bin}'.")
if not eval_proc.wait_ready(10):

2
src/testium/interpreter/test_items/test_item_cycle.py
View File

@@ -207,7 +207,7 @@ then considered as 'False'""")
else:
pl = [self._currentLoop]
proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
proc = PyFuncExecEngine(api_request, 10)
proc.start()
if not proc.wait_ready(10):
raise ETUMRuntimeError(

4
src/testium/interpreter/test_items/test_item_lua_func.py
View File

@@ -31,7 +31,7 @@ class TestItemLuaFunc(TestItem):
self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param")
self._context_id = self._prms.getParam("context_id", default=None, processed=False)
self._lua_func_proc = LuaFuncExecEngine(tm.gd("lua_bin", ""), api_request, 10)
self._lua_func_proc = LuaFuncExecEngine(api_request, 10)
def _get_engine(self):
"""Return (engine, persistent). If context_id is set, use a shared persistent engine."""
@@ -41,7 +41,7 @@ class TestItemLuaFunc(TestItem):
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)
contexts[ctx_id] = LuaFuncExecEngine(api_request, 10)
tm.setgd(_LUA_FUNC_CONTEXTS_KEY, contexts)
return contexts[ctx_id], True

4
src/testium/interpreter/test_items/test_item_py_func.py
View File

@@ -31,7 +31,7 @@ class TestItemPyFunc(TestItem):
self.func_name = self._prms.getParam("func_name", required=True)
self.params = self._prms.getParamAll("param")
self._context_id = self._prms.getParam("context_id", default=None, processed=False)
self._py_func_proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
self._py_func_proc = PyFuncExecEngine(api_request, 10)
def _get_engine(self):
"""Return (engine, persistent). If context_id is set, use a shared persistent engine."""
@@ -41,7 +41,7 @@ class TestItemPyFunc(TestItem):
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)
contexts[ctx_id] = PyFuncExecEngine(api_request, 10)
tm.setgd(_PY_FUNC_CONTEXTS_KEY, contexts)
return contexts[ctx_id], True

4
src/testium/interpreter/test_items/test_item_runtime_plot.py
View File

@@ -40,6 +40,7 @@ class TestItemPlotActionOpen(TestItemPlotAction):
try:
gname = self._prms.expanse(self.token)
lpath = self._prms.expanse(self._log_path)
runtime_plot = importlib.import_module("api.runtime_plot")
gr = runtime_plot.RuntimePlot(gname, lpath)
tm.add_plot(gr)
@@ -233,6 +234,3 @@ class TestItemPlot(TestItemActions):
)
self.actions_token = self._prms.getParam("plot_name", required=True)
global runtime_plot
runtime_plot = importlib.import_module("api.runtime_plot")

1
src/testium/interpreter/test_report/test_report.py
View File

@@ -58,7 +58,6 @@ def _discover_plugins():
try:
cls = ep.load()
_EXPORTER_REGISTRY[ep.name] = lambda c=cls: c
print(f'[testium] Loaded report exporter plugin: "{ep.name}"')
except Exception as e:
print(f'[testium] Failed to load report exporter plugin "{ep.name}": {e}')
except Exception:

29
src/testium/interpreter/test_set.py
View File

@@ -8,6 +8,7 @@ import interpreter.utils.settings as prefs
from interpreter.test_report.test_report import TestReport
from interpreter.utils.py_func_exec import PyFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils import bins
from runtime.tum_except import ETUMRuntimeError
from interpreter.utils.constants import TestItemType as cst_type
import interpreter.utils.constants as cst
@@ -49,6 +50,28 @@ class TestSet:
self._tree = self.__loadTestTree(tum_fime)
self.dict_report = self._testdict.get("report", None)
self.set_post_exec()
self._validate_runtime_deps()
def _validate_runtime_deps(self):
"""Resolve external interpreters needed by this test tree and fail
early with a clear message if any is missing.
Python is always required (the eval engine always runs). Lua is
only required when at least one ``lua_func`` item is present.
"""
needed = ["python"]
if self.__has_item_type(self._rootItem, cst_type.TYPE_LUA_FUNCTION):
needed.append("lua")
bins.ensure(*needed)
def __has_item_type(self, parent, item_type):
for i in range(parent.childCount()):
child = parent.child(i)
if child.type() == item_type.item_name:
return True
if self.__has_item_type(child, item_type):
return True
return False
def execute(self):
self._report = TestReport(self.dict_report)
@@ -352,7 +375,7 @@ class TestSet:
tm.print_debug(f' No file: "{post_exec_file}".')
return
proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
proc = PyFuncExecEngine(api_request, 10)
# start the process for executing external python
proc.start()
try:
@@ -367,13 +390,13 @@ class TestSet:
# tests backup is done here
succ, res = proc.func_call(post_exec_file, "post_exec", [])
if not succ == TestValue.SUCCESS:
tm.print_debug(
tm.print_warn(
f"Test success but the \"post_exec\" function failed: {res}"
)
else:
succ, res = proc.func_call(post_exec_file, "post_exec_fail", [])
if not succ == TestValue.SUCCESS:
tm.print_debug(
tm.print_warn(
f"Test failed but the \"post_exec_fail\" function failed: {res}"
)
finally:

151
src/testium/interpreter/utils/bins.py Normal file
View File

@@ -0,0 +1,151 @@
"""Centralised resolution of external interpreter paths (Python, Lua).
The user can override the path through the global dict via the keys
``python_bin`` and ``lua_bin`` (typically populated from a YAML config).
When unset, the system PATH is searched for known candidates.
Resolution is cached in memory: each interpreter is resolved at most
once per testium process. Subsequent calls return the cached value.
Public API
----------
``python_bin()`` : resolved python3 path (or "" if missing)
``lua_bin()`` : resolved lua >= 5.1 path (or "" if missing)
``ensure(*names)`` : resolve every name and raise a clear error if
any is missing — meant for early validation at
test load time
``reset()`` : clear the cache (mostly useful for tests)
"""
import shutil
import subprocess
import api.testium as tm
from interpreter.utils.paths import sys_app_path_lin, sys_app_path_win
from runtime.tum_except import ETUMRuntimeError
# ---------- Discovery primitives ---------------------------------------------
_PYTHON_CANDIDATES = ["python3", "python"]
_LUA_CANDIDATES = ["lua", "lua5.5", "lua5.4", "lua5.3", "lua5.2", "lua5.1"]
def _which(name):
func = sys_app_path_win if tm.OS() == "Windows" else sys_app_path_lin
return func(name)
def _python_version(path):
cmd = [path, "-c", "import sys; print(sys.version_info[:3])"]
try:
r = subprocess.run(
cmd, capture_output=True, text=True,
encoding=tm.sys_encoding(), timeout=10,
)
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
return None
try:
return eval(r.stdout)
except Exception:
return None
def _is_python3(path):
v = _python_version(path)
return v is not None and v[0] == 3
def _lua_version(path):
try:
r = subprocess.run(
[path, "-v"], capture_output=True, text=True, timeout=10,
)
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
return None
# On Windows the version banner goes to stderr.
line = r.stdout or r.stderr
try:
major, minor, _patch = line.split(" ")[1].split(".")
return (int(major), int(minor))
except (IndexError, ValueError):
return None
def _is_lua51(path):
v = _lua_version(path)
return v is not None and v >= (5, 1)
# ---------- Resolver ---------------------------------------------------------
# (display name, globdict override key, candidate list, validator)
_SPECS = {
"python": ("Python 3", "python_bin", _PYTHON_CANDIDATES, _is_python3),
"lua": ("Lua 5.1+", "lua_bin", _LUA_CANDIDATES, _is_lua51),
}
_resolved = {}
def _resolve(name):
if name in _resolved:
return _resolved[name]
display, gd_key, candidates, validator = _SPECS[name]
override = tm.gd(gd_key, "") or ""
path = ""
if override:
if shutil.which(override) and validator(override):
path = override
else:
tm.print_warn(
f"Configured {display} interpreter '{override}' is not usable; "
f"falling back to discovery."
)
if not path:
for c in candidates:
p = _which(c)
if not p:
continue
if validator(p):
path = p
break
_resolved[name] = path
return path
def python_bin():
return _resolve("python")
def lua_bin():
return _resolve("lua")
def ensure(*names):
"""Resolve each of the given names; raise if any is missing.
Meant to be called at test load with the set of interpreters the
test tree actually needs, so the user gets a clear error before
execution starts instead of deep inside an engine spawn.
"""
missing = []
for n in names:
if not _resolve(n):
display, gd_key, candidates, _ = _SPECS[n]
missing.append(
f" - {display}: tried {candidates} on PATH, none usable. "
f"Set '{gd_key}' in the YAML config to override."
)
if missing:
raise ETUMRuntimeError(
"Required external interpreter(s) not found:\n" + "\n".join(missing)
)
def reset():
_resolved.clear()

2
src/testium/interpreter/utils/lua_func_exec.py
View File

@@ -29,7 +29,7 @@ class LuaFuncExecEngine(LuaProcessBase):
# In case an error was encountered in the called function
elif "error" in answer:
msg = f"{answer["error"]}"
msg = f"{answer['error']}"
return TestValue.FAILURE, msg
else:

113
src/testium/interpreter/utils/lua_process.py
View File

@@ -1,6 +1,5 @@
import os
import sys
import shutil
import subprocess
import socket
@@ -8,85 +7,7 @@ import api.testium as tm
from runtime.jrpc import JsonRpcClient
from interpreter.utils.paths import subproc_path
from runtime.tum_except import ETUMRuntimeError
from interpreter.utils.paths import sys_app_path_lin, sys_app_path_win
def _lua_version(path: str):
cmd = f'"{path}" -v'
try:
result = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
encoding=tm.sys_encoding(),
timeout=10,
)
# Under windows, the output is on stderr
data = result.stdout or result.stderr
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired) as e:
data = ""
try:
vers = ((data.split(" "))[1]).split(".")
if len(vers) != 3:
vers = (0, 0, 0)
except:
vers = (0, 0, 0)
return tuple(vers)
def _is_lua51(lua_bin):
res = False
v = _lua_version(lua_bin)
if (v[0] == "5") and (v[1] >= "1"):
res = True
return res
def _sys_lua_bin():
sys_lua_bin = tm.gd("_sys_lua_bin", "")
if sys_lua_bin != "":
return sys_lua_bin
cur_os = tm.OS()
if cur_os == "Windows":
func = sys_app_path_win
else:
func = sys_app_path_lin
sys_lua_bin = func("lua")
if (sys_lua_bin != "") and not _is_lua51(sys_lua_bin):
tm.print_debug(f"'{sys_lua_bin}' not a lua 5.1 min.")
sys_lua_bin = ""
tm.print_debug(f"lua bin is: '{sys_lua_bin}'.")
tm.setgd("_sys_lua_bin", sys_lua_bin)
return sys_lua_bin
def _is_lua_interpreter(path: str, timeout=2) -> bool:
"""
Checks if the given path points to a valid Lua interpreter.
Args:
path (str): Path to the executable to check.
timeout (int, optional): Timeout for the subprocess in seconds. Defaults to 2.
Returns:
bool: True if the path is a Lua interpreter, False otherwise.
"""
try:
result = subprocess.run(
[path, "-v"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
timeout=timeout,
)
return (result.returncode == 0) and (
(result.stdout.startswith("Lua") or result.stderr.startswith("Lua"))
)
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
return False
from interpreter.utils import bins
class LuaProcessBase:
@@ -96,35 +17,15 @@ class LuaProcessBase:
"LUA_CPATH": {"replace": True},
}
def __init__(self, lua_bin="", request_handler=None, timeout=10):
"""
Initializes the Lua function execution engine.
Args:
lua_bin (str, optional): Path to the Lua interpreter. Defaults to system path.
request_handler: Handler for JSON-RPC requests.
timeout (int, optional): Timeout for operations in seconds. Defaults to 10.
def __init__(self, request_handler=None, timeout=10):
"""Initializes the Lua function execution engine.
Raises:
ETUMRuntimeError: If the Lua path is invalid or no interpreter is found.
ETUMRuntimeError: If no Lua >= 5.1 interpreter is found.
"""
if lua_bin != "":
if shutil.which(lua_bin) is None:
raise ETUMRuntimeError(
f"The passed lua path is not pointing to an executable: '{lua_bin}'"
)
if not _is_lua_interpreter(lua_bin):
raise ETUMRuntimeError(
f"The passed executable is not a lua interpreter: '{lua_bin}'"
)
else:
lua_bin = _sys_lua_bin()
if lua_bin == "":
raise ETUMRuntimeError(f"No valid lua interpreter found")
tm.setgd("lua_bin", lua_bin)
self._lbin = lua_bin
self._lbin = bins.lua_bin()
if not self._lbin:
raise ETUMRuntimeError("No valid Lua 5.1+ interpreter found")
self._req_handler = request_handler
self._process = None
self._port = 0

4
src/testium/interpreter/utils/py_eval.py
View File

@@ -6,9 +6,9 @@ import api.testium as tm
eval_process = None
def eval_process_init(python_bin, request_handler, timeout, python_path):
def eval_process_init(request_handler, timeout, python_path):
global eval_process
eval_process = EvalExecEngine(python_bin, request_handler, timeout, python_path)
eval_process = EvalExecEngine(request_handler, timeout, python_path)
return eval_process

2
src/testium/interpreter/utils/py_func_exec.py
View File

@@ -29,7 +29,7 @@ class PyFuncExecEngine(PyProcessBase):
# In case an error was encountered in the called function
elif "error" in answer:
msg = f"{answer["error"]}"
msg = f"{answer['error']}"
return TestValue.FAILURE, msg
else:

94
src/testium/interpreter/utils/py_process.py
View File

@@ -1,77 +1,12 @@
import os
import shutil
import sys
import subprocess
import socket
from runtime.jrpc import JsonRpcClient
import api.testium as tm
from interpreter.utils.paths import sys_app_path_lin, sys_app_path_win
from runtime.tum_except import ETUMRuntimeError
from interpreter.utils.paths import testium_path, subproc_path
def _python_version(path: str):
cmd = f'"{path}" -c "import sys; print(sys.version_info[:3])"'
try:
result = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
encoding=tm.sys_encoding(),
timeout=10,
)
data = result.stdout
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired) as e:
tm.print_debug(str(e))
data = ""
return eval(data)
def _is_python3(python_bin):
try:
v = _python_version(python_bin)
if v[0] == 3:
res = True
except:
res = False
return res
def _is_python_interpreter(path: str, timeout=2) -> bool:
try:
result = subprocess.run(
[path, "-c", "import sys; print(sys.executable)"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
timeout=timeout,
)
return result.returncode == 0
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
return False
def _sys_python_bin():
sys_python_bin = ""
cur_os = tm.OS()
if cur_os == "Windows":
func = sys_app_path_win
else:
func = sys_app_path_lin
exe = ["python3", "python"]
for e in exe:
sys_python_bin = func(e)
if sys_python_bin == "":
continue
if _is_python3(sys_python_bin):
break
sys_python_bin = ""
return sys_python_bin
from interpreter.utils import bins
class PyProcessBase:
@@ -80,29 +15,10 @@ class PyProcessBase:
"PYTHONPATH": {"replace": True},
}
def __init__(self, python_bin="", request_handler=None, timeout=10, python_path=""):
self._pbin = python_bin
if (self._pbin is not None) and (self._pbin != ""):
if shutil.which(self._pbin) is None:
raise ETUMRuntimeError(
f"The passed python path is not pointing to an executable: '{self._pbin}'"
)
if not _is_python_interpreter(self._pbin):
raise ETUMRuntimeError(
f"The passed executable is not a python interpreter: '{self._pbin}'"
)
else:
self._pbin = tm.gd("_cached_python_bin", "")
if self._pbin == "":
self._pbin = _sys_python_bin()
tm.setgd("_cached_python_bin", self._pbin)
if self._pbin == "":
raise ETUMRuntimeError(f"No valid python interpreter found")
def __init__(self, request_handler=None, timeout=10, python_path=""):
self._pbin = bins.python_bin()
if not self._pbin:
raise ETUMRuntimeError("No valid Python 3 interpreter found")
self._ppath = python_path
self._req_handler = request_handler
self._process = None

17
src/testium/interpreter/utils/test_ctrl.py
View File

@@ -25,12 +25,17 @@ class TestSetController:
if "timeout" in args:
timeout = args.pop("timeout")
self._test_ctrl.put({cmd: args})
res = self._test_resp.get(block, timeout)
if isinstance(res, tuple):
raise ETUMRuntimeError(f"Test set command '{cmd}' failed: '{res[1]}'")
if isinstance(res, dict) and not cmd in res.keys():
raise ETUMRuntimeError(f"Unexpected return error in test set controller")
return res[cmd]
# Drain stale responses (left over from earlier polled commands that
# we had given up on waiting). They can land in the queue after our
# clear() because the TestProcess may have pulled their request
# before the clear, processed them, and pushed the response after.
while True:
res = self._test_resp.get(block, timeout)
if isinstance(res, tuple):
raise ETUMRuntimeError(f"Test set command '{cmd}' failed: '{res[1]}'")
if isinstance(res, dict) and cmd in res.keys():
return res[cmd]
# Anything else is a stale response — discard and keep waiting.
def clear(self):
while True:

4
src/testium/py_func/handle.py
View File

@@ -41,7 +41,7 @@ class FuncHandler(JsonRpcSrv):
except Exception as e:
tb = traceback.format_exc()
return {
"error": f"bad jrpc req handler 'func_call' arguments ({"\n".join(tb.splitlines())}). To be reported to testium support team."
"error": "bad jrpc req handler 'func_call' arguments (" + "\n".join(tb.splitlines()) + "). To be reported to testium support team."
}
if method == "eval":
try:
@@ -57,7 +57,7 @@ class FuncHandler(JsonRpcSrv):
except Exception as e:
tb = traceback.format_exc()
return {
"error": f"bad jrpc req handler 'eval' arguments ({"\n".join(tb.splitlines())}). To be reported to testium support team."
"error": "bad jrpc req handler 'eval' arguments (" + "\n".join(tb.splitlines()) + "). To be reported to testium support team."
}
else:
return {

2
src/testium/py_func/tm.py
View File

@@ -28,7 +28,7 @@ def _make_api(name):
if "result" in res:
ret_val = res["result"]
elif "error" in res:
raise ETUMRuntimeError(f"api call to 'tm.{name}' failed with error '{res["error"]}'")
raise ETUMRuntimeError(f"api call to 'tm.{name}' failed with error '{res['error']}'")
else:
raise ETUMRuntimeError("api call failure in jrpc client to be reported to testium support team.")
return ret_val

5
src/testium/runtime/api.py
View File

@@ -6,5 +6,10 @@ SUPPORTED_API = [
"add_plot_values",
"last_plot_value",
"text_mode",
"OS",
"get_main_dir",
"init_timestamp",
"timestamp",
"timestamp_as_sec",
]

2
src/testium/runtime/jrpc.py
View File

@@ -145,7 +145,7 @@ class JsonRpcConnection:
self.pending[msg["id"]]["response"] = msg
self.pending[msg["id"]]["event"].set()
else:
self.print_info(f"msg id '{msg["id"]}' inconsistency")
self.print_info(f"msg id '{msg['id']}' inconsistency")
# ---------- Handler ----------
def _handle_request(self, meth, params, rid=None):