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

lua process implemented

Browse Source
This commit is contained in:
Foué
2025-12-30 12:59:20 +01:00
parent 59d19cb48c
commit 0ded2d4be9
19 changed files with 643 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
doc/examples/param.yaml
View File

@@ -8,3 +8,10 @@ global_loop_param_num: [1, 2, 3]
# Plot parameters
plot_log_path: /tmp/testium_plot/$(testrun_date)/$(testrun_time)/
python_path: $(home)/tmp/tum_venv/bin/python3
lua_path: /usr/bin/lua
lua_env:
LUA_PATH: /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: /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

4
doc/manual/sphinx/source/test_items/func_test_item.rst → doc/manual/sphinx/source/test_items/py_func_test_item.rst
View File

@@ -26,7 +26,7 @@ From this class it is possible to define some custom reported values with the fo
import py_func.tm as tm
class TestItemFunc(tm.FunctionItem)
class TestItemPyFunc(tm.FunctionItem)
def exec(param1, param2, param4, param4):
...
@@ -42,7 +42,7 @@ The ``exec`` method of the ``FunctionItem`` derived class is executed while runn
- py_func:
name: function test item
file: scriptTestFile.py
func_name: TestItemFunc
func_name: TestItemPyFunc
param:
- 123
- 0.123

2
doc/manual/sphinx/source/tum_syntax.rst
View File

@@ -231,7 +231,7 @@ step list attributes.
test_items/dialog_question_test_item.rst
test_items/dialog_reference_test_item.rst
test_items/dialog_value_test_item.rst
test_items/func_test_item.rst
test_items/py_func_test_item.rst
test_items/git_test_item.rst
test_items/group_test_item.rst
test_items/json-rpc_test_item.rst

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

@@ -22,7 +22,8 @@ from interpreter.utils.test_init import (
from interpreter.test_set import TestSet
from interpreter.utils.stdout_redirect import stdio_redir
from interpreter.utils.tum_except import print_exception
from interpreter.utils.func_exec import func_call_init
from interpreter.utils.py_func_exec import py_func_call_init
from interpreter.utils.lua_func_exec import lua_func_call_init
from interpreter.utils.api_srv import api_request
@@ -81,8 +82,9 @@ class TestProcess(Process):
test_set.report_path = locate_report_file(test_set.report_path)
# Python functions call subprocess initialization
fproc = func_call_init(tm.gd("python_path", ""), api_request)
# Python & lua functions call subprocess initialization
py_fproc = py_func_call_init(tm.gd("python_path", ""), api_request)
lua_fproc = lua_func_call_init(tm.gd("lua_path", "/usr/bin/lua"), api_request)
self.__loaded = True
@@ -99,8 +101,10 @@ class TestProcess(Process):
try:
test_run_init()
print(test_run_header())
fproc.start()
fproc.wait_ready()
py_fproc.start()
lua_fproc.start()
lua_fproc.wait_ready()
py_fproc.wait_ready()
test_set.execute()
finally:
if test_set.success():
@@ -111,8 +115,10 @@ class TestProcess(Process):
test_set.run_post_exec()
finally:
# Stop function execution process
fproc.stop()
fproc.join()
py_fproc.stop()
lua_fproc.stop()
lua_fproc.join()
py_fproc.join()
self.__exec = False
# Sends signal to the GUI
self.send_finished()

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

@@ -1,7 +1,7 @@
import traceback
from interpreter.utils.tum_except import ETUMSyntaxError, ETUMRuntimeError
from interpreter.utils.func_exec import func_exec
from interpreter.utils.py_func_exec import py_func_exec
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
import libs.testium as tm
@@ -204,7 +204,7 @@ class TestItemCycle(TestItem):
pl = self._prms.expanse(param_list)
else:
pl = [self._currentLoop]
fsucc, res = func_exec(file, func, pl)
fsucc, res = py_func_exec(file, func, pl)
if fsucc == TestValue.SUCCESS:
fres, _ = res
if fres:

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

@@ -7,12 +7,12 @@ import textwrap
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.func_exec import func_exec
from interpreter.utils.lua_func_exec import lua_func_exec
from interpreter.utils.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemFunc(TestItem):
class TestItemLuaFunc(TestItem):
"""py_func item usage.
func file: func_file.py, func_name: func, param: [$(variable1), [1, 2, 3], true]
"""

77
src/testium/interpreter/test_items/test_item_py_func.py Normal file
View File

@@ -0,0 +1,77 @@
import sys
import traceback
import pprint
import textwrap
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.py_func_exec import py_func_exec
from interpreter.utils.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
class TestItemPyFunc(TestItem):
"""py_func item usage.
func file: func_file.py, func_name: func, param: [$(variable1), [1, 2, 3], true]
"""
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_FUNCTION.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_FUNCTION
self.is_container = False
try:
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(),
)
@test_run
def execute(self):
self.result.set(
TestValue.FAILURE, "an exception occured during function execution."
)
try:
self.file_name = self._prms.expanse(self.file_name)
self.func_name = self._prms.expanse(self.func_name)
param_list = self._prms.getParamFromList(self.params)
pl = self._prms.expanse(param_list)
if tm.debug_enabled():
tm.print_debug("Parameters list:")
tm.print_debug(textwrap.indent(pprint.pformat(pl), " |"))
success, ret = py_func_exec(self.file_name, self.func_name, pl)
if success == TestValue.SUCCESS:
self.result.set(TestValue.SUCCESS)
res, reported_values = ret
reported_values = {**reported_values, "returned": res}
self.result.reported = ret[1]
if tm.debug_enabled():
tm.print_debug("Returned value:")
tm.print_debug(textwrap.indent(pprint.pformat(res), " |"))
# The result of the func test item is put in global dir and result
tm.setgd("fn_" + self._name, res)
self.result.value = res
else:
self.result.set(TestValue.FAILURE, ret)
if tm.debug_enabled():
tm.print_debug("Failed:")
tm.print_debug(textwrap.indent(pprint.pformat(ret), " |"))
return
except:
traceback.print_exception(*sys.exc_info())
self.result.set(
TestValue.FAILURE,
'Unrecoverable "py_func" item error from {}'.format(self.func_name),
)

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

@@ -8,7 +8,7 @@ from interpreter.utils.tum_except import (
)
import interpreter.utils.settings as prefs
from interpreter.test_report.test_report import TestReport
from interpreter.utils.func_exec import func_exec
from interpreter.utils.py_func_exec import py_func_exec
from interpreter.utils.constants import TestItemType as cst_type
import interpreter.utils.constants as cst
from interpreter.utils.constants import TEST_TYPE_LIST
@@ -331,13 +331,13 @@ class TestSet:
tm.print_debug(f'Post-execution from: "{post_exec_file}"')
if self.rootItem().result.success:
# tests backup is done here
succ, res = func_exec(post_exec_file, "post_exec", [])
succ, res = py_func_exec(post_exec_file, "post_exec", [])
if not succ == TestValue.SUCCESS:
tm.print_debug(
f"Test success but the \"post_exec\" function failed: {res}"
)
else:
succ, res = func_exec(post_exec_file, "post_exec_fail", [])
succ, res = py_func_exec(post_exec_file, "post_exec_fail", [])
if not succ == TestValue.SUCCESS:
tm.print_debug(
f"Test failed but the \"post_exec_fail\" function failed: {res}"

153
src/testium/interpreter/utils/lua_func_exec.py Normal file
View File

@@ -0,0 +1,153 @@
import os
import shutil
import subprocess
import socket
import libs.testium as tm
from interpreter.utils.tum_except import ETUMRuntimeError
from interpreter.utils.jrpc import JsonRpcClient
from interpreter.test_items.test_result import TestValue
function_call_process = None
def lua_func_call_init(lua_path, request_handler):
global function_call_process
function_call_process = LuaFuncExecEngine(lua_path, request_handler)
return function_call_process
def lua_version(path: str):
result = subprocess.run(
[path, "-v"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
vers = ((result.stdout.split(" "))[1]).split(".")
return (vers[0], vers[1], vers[2])
def is_lua_interpreter(path: str, timeout=2) -> bool:
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"))
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
return False
class LuaFuncExecEngine:
def __init__(self, lua_path="", request_handler=None):
if shutil.which(lua_path) is None:
raise ETUMRuntimeError(
f"The passed lua path is not pointing to an executable: '{lua_path}'"
)
if not is_lua_interpreter(lua_path):
raise ETUMRuntimeError(
f"The passed executable is not a lua interpreter: '{lua_path}'"
)
self._lpath = lua_path
self._req_handler = request_handler
self._process = None
self._port = 0
self._rpc = None
def start(self):
"""
run the subprocess to execute the python functions of the test.
"""
# This thread is not closed until new test is loaded
if self._process is not None:
raise ETUMRuntimeError("The function subprocess has already been started.")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("localhost", 0))
self._port = sock.getsockname()[1]
func_proc_path = os.path.join(tm.gd("testium_path"),"lua_func")
lua_env = tm.gd("lua_env", {})
self._process = subprocess.Popen(
[self._lpath, "main.lua", "--host", "localhost", "--port", f"{self._port}"], env=lua_env, cwd=func_proc_path
)
# Port was reserved until the sub-process is started. Now released.
if sock is not None:
sock.close()
self._rpc = JsonRpcClient(self._port, req_handler=self._req_handler)
self._rpc.start()
def join(self):
if self._rpc is not None:
self._rpc.join()
self._rpc = None
self._process = None
def wait_ready(self, timeout=None):
if self._rpc is not None and self._rpc.is_alive():
return self._rpc.wait_ready(timeout)
return False
def stop(self):
if self._rpc is not None:
self._rpc.stop()
def func_call(self, file: str, func_name: str, params: list, verbose: bool = True):
if (self._rpc is not None) and self._rpc.is_alive():
answer = self._rpc.call(
"func_call",
{
"file": file,
"fname": func_name,
"params": params,
"verbose": verbose,
},
)
if "result" in answer:
reported_values = answer["result"].get("reported_values", {})
if "returned_value" in answer["result"]:
res = answer["result"]["returned_value"]
return TestValue.SUCCESS, (res, reported_values)
else:
raise ETUMRuntimeError(
"Unexepected py_func jrpc result. To be reported to testium support team."
)
# In case an error was encountered in the called function
elif "error" in answer:
msg = f"{answer["error"]}"
return TestValue.FAILURE, msg
else:
raise ETUMRuntimeError(
"Unexepected py_func call failure to be reported to testium support team."
)
else:
raise ETUMRuntimeError(
"No function execution process active. To be reported to testium support team."
)
def lua_func_exec(file: str, func_name: str, params: list, verbose: bool = True):
"""Executes a python function and returns its result and reported values"""
global function_call_process
if function_call_process is not None:
success, result = function_call_process.func_call(
file, func_name, params, verbose
)
else:
raise ETUMRuntimeError(
"No function execution process active. To be reported to testium support team."
)
return success, result

8
src/testium/interpreter/utils/func_exec.py → src/testium/interpreter/utils/py_func_exec.py
View File

@@ -10,9 +10,9 @@ from interpreter.test_items.test_result import TestValue
function_call_process = None
def func_call_init(python_path, request_handler):
def py_func_call_init(python_path, request_handler):
global function_call_process
function_call_process = FuncExecEngine(python_path, request_handler)
function_call_process = PyFuncExecEngine(python_path, request_handler)
return function_call_process
@@ -40,7 +40,7 @@ def is_python_interpreter(path: str, timeout=2) -> bool:
return False
class FuncExecEngine:
class PyFuncExecEngine:
def __init__(self, python_path="", request_handler=None):
if python_path != "":
@@ -140,7 +140,7 @@ class FuncExecEngine:
)
def func_exec(file: str, func_name: str, params: list, verbose: bool = True):
def py_func_exec(file: str, func_name: str, params: list, verbose: bool = True):
"""Executes a python function and returns its result and reported values"""
global function_call_process

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

@@ -32,7 +32,7 @@ from interpreter.test_items.test_item_cycle import TestItemCycle
from interpreter.test_items.test_item_runtime_plot import TestItemPlot
from interpreter.test_items.test_item_group import TestItemGroup
from interpreter.test_items.test_item_git import TestItemGit
from interpreter.test_items.test_item_func import TestItemFunc
from interpreter.test_items.test_item_py_func import TestItemPyFunc
from interpreter.test_items.test_item_let import TestItemLet
from interpreter.test_items.test_item_check import TestItemCheckValue
from interpreter.test_items.test_item_json_rpc import TestItemJSON_RPC
@@ -51,7 +51,7 @@ from interpreter.test_items.test_item_report import TestItemReport
def _constants_init():
cst.TYPE_CONSOLE.item_class = TestItemConsole
cst.TYPE_CYCLE.item_class = TestItemCycle
cst.TYPE_FUNCTION.item_class = TestItemFunc
cst.TYPE_FUNCTION.item_class = TestItemPyFunc
cst.TYPE_GIT.item_class = TestItemGit
cst.TYPE_GRAPH.item_class = TestItemPlot
cst.TYPE_GROUP.item_class = TestItemGroup

4
src/testium/libs/runtime_plot.py
View File

@@ -18,7 +18,7 @@ from datetime import datetime, timedelta, timezone
from interpreter.test_items.test_result import TestValue
from interpreter.utils.tum_except import ETUMRuntimeError
from interpreter.utils.func_exec import func_exec
from interpreter.utils.py_func_exec import py_func_exec
from interpreter.utils.eval import post_evaluate
from interpreter.utils.periodic_timer import PeriodicTimer
from interpreter.utils.paths import abs_path_from_file, prepare_file_to_save
@@ -272,7 +272,7 @@ class RuntimePlotPeriodic(PeriodicTimer):
self.on_timer_event()
def on_timer_event(self):
succ, ret = func_exec(self.file, self.func_name, self.args)
succ, ret = py_func_exec(self.file, self.func_name, self.args)
if succ == TestValue.SUCCESS:
res, _ = ret
res = post_evaluate(self.post_eval, res)

56
src/testium/lua_func/handle.lua Normal file
View File

@@ -0,0 +1,56 @@
local utils = require("utils")
local tm = require("tm")
local handle = {}
local function _get_func_by_path(file_path, func_name)
-- 1. Load the file from the path
-- loadfile returns a 'chunk' (a function that runs the file's code)
local chunk, load_err = loadfile(file_path)
if not chunk then
return nil, "Failed to load file: " .. tostring(load_err)
end
-- 2. Execute the chunk to get the module's return value
-- Most Lua modules end with 'return { ... }'
local ok, module = pcall(chunk)
if not ok then
return nil, "Error executing file: " .. tostring(module)
end
-- 3. Validate the module is a table and contains the function
if type(module) ~= "table" then
return nil, "Module did not return a table (returned " .. type(module) .. ")"
end
local target_func = module[func_name]
if type(target_func) ~= "function" then
return nil, "Function '" .. func_name .. "' not found in " .. file_path
end
return target_func
end
function handle.func_call(file, fname, params)
local pfile = file
-- 1. modify the file path if it is relative
if utils.is_relative_path(file) then
local td = tm.gd("test_directory")
pfile = utils.join_paths(td, file)
end
-- 2. retrieve the function "fname"
local func, err = _get_func_by_path(pfile, fname)
-- 3. Execute the function
local res = nil
if err == nil then
succ, res = pcall(func, table.unpack(params))
end
-- 4. Returns result
return res, err
end
return handle

108
src/testium/lua_func/json-rpc.lua Normal file
View File

@@ -0,0 +1,108 @@
local json = require("cjson")
local JSONRPC = {}
JSONRPC.__index = JSONRPC
function JSONRPC.new(send_fn)
local self = setmetatable({}, JSONRPC)
self.send_raw = send_fn -- Function to transmit string data to transport (TCP/Websocket)
self.methods = {} -- Methods the server provides to the client
self.pending = {} -- Requests sent to client waiting for response
self.next_id = 1
return self
end
--- Register a method the client can call
function JSONRPC:register(name, callback)
self.methods[name] = callback
end
--- Handle incoming raw data from the transport layer
function JSONRPC:handle_message(raw_data)
local ok, msg = pcall(json.decode, raw_data)
if not ok then return self:_send_error(nil, -32700, "Parse error") end
-- 1. Check if it's a Response (has 'result' or 'error' and 'id')
if msg.result ~= nil or msg.error ~= nil then
return self:_handle_response(msg)
end
-- 2. Check if it's a Request
if msg.method then
return self:_handle_request(msg)
end
end
--- INTERNAL: Handle requests from the client
function JSONRPC:_handle_request(req)
local method = self.methods[req.method]
if not method then
if req.id then self:_send_error(req.id, -32601, "Method not found") end
return
end
local ok, result = pcall(method, req.params)
-- Only send response if it's not a Notification (notifications have no ID)
if req.id then
if ok then
self:_send({ jsonrpc = "2.0", result = result, id = req.id })
else
self:_send_error(req.id, -32603, "Internal error: " .. tostring(result))
end
end
end
--- INTERNAL: Handle responses to requests WE sent
function JSONRPC:_handle_response(res)
local callback = self.pending[res.id]
if callback then
callback(res.error, res.result)
self.pending[res.id] = nil
end
end
--- Call a method on the client
function JSONRPC:call(method, params, callback)
local id = self.next_id
self.next_id = self.next_id + 1
if callback then
self.pending[id] = callback
end
self:_send({
jsonrpc = "2.0",
method = method,
params = params,
id = id
})
end
function JSONRPC:call_sync(method, params)
local callco = coroutine.create(function(m, p)
local co = coroutine.running()
-- Call the async version, but use the callback to resume this coroutine
self:call(m, p, function(err, res)
coroutine.resume(co, err, res)
end)
-- Pause execution here until 'resume' is called
return coroutine.yield()
end)
return coroutine.resume(callco, method, params)
end
function JSONRPC:_send(data)
self.send_raw(json.encode(data))
end
function JSONRPC:_send_error(id, code, message)
self:_send({
jsonrpc = "2.0",
error = { code = code, message = message },
id = id
})
end
return JSONRPC

111
src/testium/lua_func/main.lua Normal file
View File

@@ -0,0 +1,111 @@
-- =========================
-- Options par défaut
-- =========================
local config = {
host = "0.0.0.0",
port = 9000,
timeout = 60,
verbose = true,
}
local function usage()
print([[
Usage: lua lua_func [options]
Options:
--host <ip> Adresse d'écoute (default: 0.0.0.0)
--port <port> Port TCP (default: 9000)
--timeout <sec> Timeout client en secondes (default: 60)
--verbose Logs détaillés
--help Affiche cette aide
]])
os.exit(0)
end
-- =========================
-- Parsing des arguments
-- =========================
local i = 1
while i <= #arg do
local a = arg[i]
if a == "--host" then
i = i + 1
config.host = arg[i]
elseif a == "--port" then
i = i + 1
config.port = tonumber(arg[i])
elseif a == "--timeout" then
i = i + 1
config.timeout = tonumber(arg[i])
elseif a == "--verbose" then
config.verbose = true
elseif a == "--help" then
usage()
else
print("Unknown option:", a)
usage()
end
i = i + 1
end
local socket = require("socket")
local JSONRPC = require("json-rpc") -- The module from the previous response
local utils = require("utils")
utils.verbose = config.verbose
-- Create the master socket
local server_sock = assert(socket.bind(config.host, config.port))
utils.log("listening on %s:%d", config.host, config.port)
server_sock:settimeout(config.timeout) -- Prevents hanging on dead connections
-- Main Server Loop
local client_sock, err = server_sock:accept()
if err then
utils.log("connection failed: %s", err)
os.exit(0)
end
client_sock:settimeout(10) -- Prevents hanging on dead connections
utils.log("Client connected!")
-- Initialize the RPC instance for this specific connection
local rpc = JSONRPC.new(function(data)
client_sock:send(data .. "\n") -- Standard JSON-RPC uses newline delimiters over TCP
end)
utils.setup_remote_print(rpc)
-- Define Server Methods
rpc:register("echo", function(params)
return params
end)
-- Example: Send a request TO the client immediately upon connection
rpc:call("greet", { msg = "Welcome to the server" }, function(err, res)
if not err then print("Client replied to greeting:", res) end
end)
-- Communication Loop for this client
while true do
local line, err = client_sock:receive() -- Read until newline
if err == "closed" then
utils.log("Connection closed:", err)
break
elseif err then
socket.sleep(0.01)
else
rpc:handle_message(line)
end
end
client_sock:close()

29
src/testium/lua_func/tm.lua Normal file
View File

@@ -0,0 +1,29 @@
local tm = {}
local SUPPORTED_API = {
"gd",
"setgd",
"delgd",
}
-- underlying function
function tm._init_api(rpc)
tm._rpc = rpc
local function _api_request(fname, ...)
local args = {...}
return tm._rpc:call_sync(fname, args)
end
for _, fname in ipairs(SUPPORTED_API) do
-- create a closure that calls common_handler with fname
tm[fname] = function(...)
return _api_request(fname, ...)
end
end
end
return tm

67
src/testium/lua_func/utils.lua Normal file
View File

@@ -0,0 +1,67 @@
local utils = {}
utils.verbose = false
function utils.log(fmt, ...)
if utils.verbose then
-- print("[lua_func server]", ...)
io.stdout:write(string.format("[lua_func server] - " .. fmt .. "\n", ...))
end
end
utils.sep = package.config:sub(1,1)
function utils.join_paths(p1, p2)
return p1 .. utils.sep .. p2
end
function utils.is_absolute_path(path)
if not path or path == "" then return false end
-- 1. Check for POSIX absolute path (starts with /)
if path:sub(1, 1) == "/" then
return true
end
-- 2. Check for Windows drive letter (e.g., C:\ or D:/)
-- Pattern: %a (letter) followed by : (colon)
if path:match("^%a:[/\\]") or path:match("^%a:$") then
return true
end
-- 3. Check for Windows UNC/Network paths (starts with \\ or //)
if path:match("^[/\\][/\\]") then
return true
end
return false
end
function utils.is_relative_path(path)
return not utils.is_absolute_path(path)
end
function utils.setup_remote_print(rpc)
-- Store the original print if you still need to log to the server console
_G.native_print = _G.native_print or _G.print
-- Define the new local print
_G.print = function (...)
local args = table.pack(...)
local output = {}
for i = 1, args.n do
table.insert(output, tostring(args[i]))
end
local message = table.concat(output, "\t")
pcall(function()
rpc:call_sync("print", message )
end)
-- Optional: Still print to the server's local console
-- utils.log("[Remote Log Sent]: " .. message)
end
end
return utils

6
src/testium/py_func/__init__.py
View File

@@ -1,6 +1,6 @@
#!/usr/bin/env python
import multiprocessing
from py_func.tm import _init_api, remote_print
from py_func.tm import _init_api, _remote_print
from interpreter.utils.stdout_redirect import stdio_redir
@@ -9,7 +9,7 @@ class TcpStdOut:
pass
def write(self, s: str) -> None:
remote_print(s)
_remote_print(s)
def flush(self):
pass
@@ -30,7 +30,7 @@ def main():
outstream = TcpStdOut()
stdio_redir.redirect(outstream)
# debug the server
# thrd_api.dbg_out = stdio_redir.ini_stdout
thrd_api.dbg_out = stdio_redir.ini_stdout
try:
while thrd_api.is_alive():
thrd_api.join(1)

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

@@ -70,7 +70,7 @@ def _init_api(port):
###############################################################################
def remote_print(*values):
def _remote_print(*values):
"""Forward print-like output to the remote handler.
If a ``_func_call_thread`` is available, this function calls the