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

processes management refundation.

Browse Source 
Evaluation removed from conditions, let and exit_conditional
This commit is contained in:
Foué
2026-01-25 15:03:52 +01:00
parent 929e75215b
commit 60636b4fd2
45 changed files with 863 additions and 752 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
doc/examples/dummy/plot.py
View File

@@ -1,6 +1,6 @@
import random
import libs.testium as tm
from libs.testium import FunctionItem
import py_func.tm as tm
from py_func.tm import FunctionItem
def random_value():
return random.random()

30
doc/examples/example_conditional.tum
View File

@@ -10,12 +10,12 @@ main:
- let:
name: Set test variables for Linux
condition: "'$(os)' == 'Linux'"
condition: $| "$(os)" == "Linux" |
values:
- terminal_prompt: $(linux_prompt)
- let:
name: Set test variables for Windows
condition: "'$(os)' == 'Windows'"
condition: $| "$(os)" == "Windows" |
values:
- terminal_prompt: $(windows_prompt)
@@ -35,17 +35,17 @@ main:
no_fail: True
exit_condition:
value: "'$(last_test_result)' == 'PASS'"
value: $| "$(last_test_result)" == "PASS" |
- let:
name: let
eval:
- conditional_exec: "random.randint(1, 4)"
values:
- conditional_exec: $| random.randint(1, 4) |
- console:
name: Console creation
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
doc: Opening the console
steps:
@@ -55,56 +55,56 @@ main:
- console:
name: Console read_until with timeout
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
steps:
- read_until: {expected: "$(terminal_prompt)", timeout: 10}
- console:
name: Console write
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
steps:
- writeln: echo 0
- sleep:
name: sleep item
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
timeout: 5
- console:
name: Console read_until immediate
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
steps:
- read_until: {expected: "0", timeout: 0}
- console:
name: Console read_until immediate (2)
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
steps:
- read_until: {expected: "$(terminal_prompt)", timeout: 0}
- console:
name: Console closure
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
steps:
- close: consname
- sleep:
name: sleep item
condition: "$(conditional_exec) == 2"
condition: $| $(conditional_exec) == 2 |
timeout: 5
- dialog_image:
name: dialog image item
condition: "$(conditional_exec) == 3"
condition: $| $(conditional_exec) == 3 |
question: click ok if you see the image
filename: image.jpg
- dialog_value:
name: dialog_value item
condition: "$(conditional_exec) == 4"
condition: $| $(conditional_exec) == 4 |
question: enter something and click ok

2
doc/examples/example_cycle.tum
View File

@@ -139,7 +139,7 @@ main:
timeout: 0.2
dialog: false
exit_condition:
value: "$(variable) >= 10"
value: $| $(variable) >= 10 |
# This loop must fail du to an exception in exit condition.
- loop:

4
doc/examples/example_items.tum
View File

@@ -16,7 +16,7 @@ main:
- group:
name: Set test variables for Linux
condition: "'$(os)' == 'Linux'"
condition: $| "$(os)" == "Linux" |
steps:
- let:
@@ -26,7 +26,7 @@ main:
- group:
name: Set test variables for Windows
condition: "'$(os)' == 'Windows'"
condition: $| "$(os)" == "Windows" |
steps:
- let:

2
doc/examples/example_sequence.tum
View File

@@ -22,6 +22,6 @@ main:
- *seq_sleep
exit_condition:
value: "$(variable) >= 3"
value: $| $(variable) >= 3 |
- !include {file: seq2.tum, is_dialog: True, sleep_timeout: 12, func_para: truc}

2
doc/examples/example_simple.tum
View File

@@ -23,5 +23,5 @@ main:
- sleep:
name: sleep item
dialog: true
timeout: 3600
timeout: $| 3600 + random.randint(1, 10) |
no_fail: true

6
doc/examples/example_text_manip.tum
View File

@@ -20,10 +20,10 @@ main:
- let:
name: Extract data
eval:
- text_extract: "[l for l in '''$(rand_text)'''.splitlines() if '$(text_searched)' in l][0]"
values:
- text_extract: $| [l for l in '''$(rand_text)'''.splitlines() if '$(text_searched)' in l][0] |
- dialog_message:
condition: len('$(text_extract)') > 0
condition: $| len('$(text_extract)') > 0 |
name: dialog value test item
question: Tataaaaa !

6
doc/manual/sphinx/source/test_items/group_test_item.rst
View File

@@ -8,10 +8,10 @@ This element is of the following form:
- group:
name: Group Item
condition: "'$(OS)' == 'Linux'"
condition: $| "$(OS)" == "Linux" |
steps:
- unittest_file:
test_file: test_prod_rio6_8093.py
test_file: test_prod_alpha_13.py
test_method:
...
- sleep:
@@ -23,4 +23,4 @@ Attributes
--------------------
* The ``steps`` list describes the sequence executed in the group.
It is a list of any of the testium test items,
It is a list of any of the `testium` test items,

9
doc/manual/sphinx/source/test_items/items_common_attributes.rst
View File

@@ -72,9 +72,9 @@ if not provided is given in the table as well.
|``report`` | / | This attribute defines values (a dictionary) which |
| | | will be added in the ``data`` field of the report. |
+-----------------------+-------------------+-------------------------------------------------------+
| ``condition`` | / | The test item is not executed if its |
| | | ``condition`` attribute content is |
| | | evaluated as ``False``. |
| ``condition`` | / | The test item is executed if its |
| | | ``condition`` attribute content is a boolean |
| | | ``True``. |
| | | see :ref:`Conditional |
| | | execution<sec_conditional_execution>`. |
+-----------------------+-------------------+-------------------------------------------------------+
@@ -133,7 +133,8 @@ or in configuration file (see :ref:`config files<sec_configuration_files>`) as a
Conditional execution
-----------------------------------------------
The ``condition`` attribute content is evaluated as a python string.
The ``condition`` attribute content must be a boolean value (if not ``True``, the condition is considered
not met).
.. _sec_process_result:

7
doc/manual/sphinx/source/test_items/let_test_item.rst
View File

@@ -11,15 +11,12 @@ This element is of the following form:
values:
key1: value1
key2: value2
eval:
key3: $(variable)[$(loop_index)]
key3: $| $(variable)[$(loop_index)] |
The ``let`` element is used to set values in the global directory.
Attributes
----------------
* The values list gives the {<key>, <value>} couples to set in the
* The ``values`` list gives the {<key>, <value>} couples to set in the
global directory,
* The eval list gives the strings to evaluate prior to its storage into
the <key> of global directory.

6
doc/manual/sphinx/source/test_items/loop_test_item.rst
View File

@@ -39,12 +39,12 @@ Below are described loop test item specific attributes.
* ``Iterator``: giving the number of loop iteration (see dedicated chapter below).
* ``steps``: describes the sequence executed at each cycle; it is
a list of any of the testium test items.
* ``exit_condition``: allows to exit the loop. If False is returned, loop continues
else, it breaks. exit_condition attributes are:
* ``exit_condition``: allows to exit the loop. If True is returned loop continues
otherwise it breaks. exit_condition attributes are:
* ``time``: the loop stops after the time (in minutes) is elapsed (optional)
* ``value``: the loop stops when the content of the value attribute is
evaluated as True (optional)
True (optional)
* ``file``: the loop the script file name that contains a function to be
executed on each loop. Only python script format is supported (optional
if another exit_condition attribute is defined)

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

@@ -111,12 +111,12 @@ value of the funcToBeExecuted python function.
Some global variables have an impact on the ``py_func`` test item behavior:
* ``python_path``: This optional global variable can be used to define
* ``python_bin``: This optional global variable can be used to define
the python executable path. If not defined, the python interpreter is
searched in at the default places in the system.
* ``python_env``: This global variable can be used to define
environment variables for the lua script execution environment.
Only `PATH`, `LUA_PATH`, and `LUA_CPATH` are supported.
Only `PATH` and `PYTHONPATH` are supported.
.. code-block:: yaml
:caption: example of configuration file: param.yaml

4
doc/manual/sphinx/source/test_items/run_test_item.rst
View File

@@ -9,7 +9,7 @@ This test item executes a new instance of testium.
- run:
name: Execute TUM
tum_fime: example_cycle.tum
python_path: python3
python_bin: python3
testium_path: /home/francois/projets/testium-new-report/testium.pyw
log_file: $(home)/reports/test.log
report_file: $(home)/reports/test.rep
@@ -21,7 +21,7 @@ run test item has the following specific attributes:
* ``tum_fime``: mandatory the path of the file to execute, it can be relative to current execution folder,
* ``param_file`` (optional) the path of the parameter file to use, otherwise default parameter file is used.
* ``python_path`` (optional) the path of a specific python to run your scripts,
* ``python_bin`` (optional) the path of a specific python to run your scripts,
* ``testium_path`` (optional) the path of a specific testium to run your scripts,
* ``log_file`` (optional) the path of log file to register, if not provided a file is created with timestamp at the location of TUM file.
* ``report_file`` (optional), the path of report file to create

4
src/testium/__init__.py
View File

@@ -7,7 +7,6 @@ from pathlib import Path
ourpath = Path(__file__)
ourpath = ourpath.resolve()
sys.path.append(os.path.abspath(ourpath.parent))
from interpreter.utils.eval import evaluate
import interpreter.utils.constants as cst
@@ -71,8 +70,7 @@ def main():
d = define.split('=', 1)
if d[0].strip() != '':
if len(d) > 1:
_, edef = evaluate(d[1])
defines.update({d[0].strip(): edef})
defines.update({d[0].strip(): d[1]})
else:
defines.update({d[0].strip(): True})

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

@@ -4,16 +4,18 @@ from multiprocessing import Process, Queue, Pipe
from queue import Empty
from threading import Thread
from time import sleep
import traceback
import copy
import libs.testium as tm
from interpreter.utils.params import expanse
from interpreter.utils.string_queue import StringQueue
from interpreter.utils.tum_except import ETUMRuntimeError
from interpreter.utils.tum_except import ETUMRuntimeError, ETUMSyntaxError
from interpreter.utils.test_ctrl import TestSetController
from interpreter.utils.test_init import (
env_init,
load_test,
prepare_global,
update_global,
set_standard_gd_keys,
test_run_init,
test_run_header,
locate_report_file,
@@ -22,10 +24,12 @@ from interpreter.utils.test_init import (
)
from interpreter.utils.constants import TestItemType as cst_type
from interpreter.test_set import TestSet
from interpreter.utils.include import TUMLoader, TUMLoaderNoIncludes, TUMLoaderRawIncludes
from interpreter.utils.stdout_redirect import stdio_redir
from interpreter.utils.template import template_to_test
from interpreter.utils.yaml_load import yaml_load
from interpreter.utils.tum_except import print_exception
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.py_eval import eval_process_init
from interpreter.utils.api_srv import api_request
@@ -51,6 +55,145 @@ class TestProcess(Process):
self.__closed = False
self.__pconn = self.redirect_stdout()
def _check_test_dict(self, test_dict):
if not isinstance(test_dict, dict):
raise ETUMSyntaxError(
"The tum file has a major problem. Please check the documentation for syntax.")
if not 'main' in test_dict.keys():
raise ETUMSyntaxError(
"The tum file has a major problem. The 'main' section could not be found.")
def _locate_config_files(self, test_dir, config_files, silent=False):
ret = []
pf = []
if len(config_files) == 0:
for p in ['param.yaml', 'param.yml']:
param_filename = os.path.join(test_dir, p)
if os.path.exists(param_filename):
pf.append(param_filename)
if not silent:
tm.print_info(f"Configuration file loaded: {p}.")
else:
if not silent:
tm.print_info(f"Default param file \"{p}\" does not exist.")
else:
pf = config_files
for p in pf:
ret.append(p)
return ret
def _config_files_from_test(self, test_dict, config_files=None):
test_dir = tm.gd('test_directory')
pf = []
if isinstance(config_files, list) and len(config_files) == 0:
param_filename = test_dict.get('config_file', None)
if param_filename is None:
param_node = test_dict.get('param_file', None)
if param_node is not None:
if isinstance(param_node, dict):
p = param_node.get('file_name', None)
if p is not None:
param_filename = p
else:
param_filename = param_node
else:
param_filename = param_node
if param_filename is None:
pf = self._locate_config_files(test_dir, [])
elif isinstance(param_filename, str):
pf.append(param_filename)
elif isinstance(param_filename, (list)):
pf = []
for p in param_filename:
if isinstance(p, list):
for pp in p:
pf.append(pp)
elif p is not None:
pf.append(p)
else:
raise ETUMSyntaxError(
'Unrecognized tum "param_file" : {}'.format(param_filename))
elif isinstance(config_files, list):
pf = config_files
elif isinstance(config_files, str):
pf = [config_files]
else:
raise ETUMSyntaxError(
'Unrecognized config_files parameter : {}'.format(config_files))
return pf
def _load_test_dict(self, test_file, variables: dict, no_include: bool = False, raw_include: bool = False):
loader = TUMLoader
loader = TUMLoaderRawIncludes if raw_include else loader
loader = TUMLoaderNoIncludes if no_include else loader
# Jinja template processing
tmpf = template_to_test(test_file, variables)
try:
d = yaml_load(tmpf, test_file, loader)
finally:
tmpf.close()
return d
def _load_initial_params(self, test_dir):
# First step: populate config files without includes considered
test_dict = self._load_test_dict(self.__fname, {}, no_include=True)
self._check_test_dict(test_dict)
prepare_global()
# Define the global builtin variables
set_standard_gd_keys(test_dict["main"].get(
"name", "Unnamed"), test_dir, self.__fname, self.__cfgf)
# Include the content of the first config files into glob dict
old_pfs = self._config_files_from_test(test_dict, self.__cfgf)
# Variables updated
gd = update_global(old_pfs, self.__defs, self.__gui_defaults, silent=True)
return old_pfs, gd
def _load_test(self, init_param_files, glob_variables):
old_pfs = init_param_files
gd = glob_variables
while True:
# Loop to check param files until all param files are identified
test_dict = self._load_test_dict(self.__fname, gd, raw_include=True)
new_pfs = self._config_files_from_test(test_dict, self.__cfgf)
# Check if things have changed since previous evaluation of
# config files
new_stuff = False
if len(old_pfs) != len(new_pfs):
new_stuff = True
if not new_stuff:
for i in range(len(old_pfs)):
if old_pfs[i] != new_pfs[i]:
new_stuff = True
break
# If the param files are identical, we continue in loading process
if not new_stuff:
break
# Variables updated
gd = update_global(new_pfs, self.__defs, self.__gui_defaults, silent=False)
old_pfs = copy.copy(new_pfs)
# Processing (with includes) for complete file loading
test_dict = self._load_test_dict(self.__fname, gd)
return test_dict, new_pfs
def run(self):
try:
try:
@@ -64,21 +207,26 @@ class TestProcess(Process):
env_init()
# Load the test file
test_dict, cfg_files = load_test(
self.__fname,
test_dir,
self.__cfgf,
self.__defs,
self.__gui_defaults,
# Creation of the python evaluation process for loading of the complete test
eval_proc = eval_process_init("", api_request, 10, test_dir)
eval_proc.start()
if not eval_proc.wait_ready(10):
raise ETUMRuntimeError(
f"""Impossible to start the external python execution process.
Is the python exec path correct ?"""
)
try:
# Loading of the param files without inclusions (first level)
init_param_files, glob_variables = self._load_initial_params(test_dir)
# Load the test file
test_dict, param_files = self._load_test(init_param_files, glob_variables)
# Backup the global dict in case of restart of the test
gdict = backup_gd()
# The path of the test file is included in PYTHONPATH
sys.path.append(os.path.dirname(self.__fname))
# Now create the test structure and objects
test_set = TestSet(self.__fname, test_dict, self.__squeue)
@@ -91,16 +239,8 @@ class TestProcess(Process):
)
self.cmd_th.start()
# Set the report path
test_set.report_path = locate_report_file(test_set.report_path)
# Python & lua functions call subprocess initialization
py_fproc = py_func_call_init(tm.gd("python_path", ""), api_request, 10)
# Lua functions call subprocess initialization
lua_fproc = None
if test_set.isTestTypePresent(cst_type.TYPE_LUA_FUNCTION):
lua_fproc = lua_func_call_init(tm.gd("lua_path", ""), api_request, 10)
self.__loaded = True
while True:
@@ -116,24 +256,6 @@ class TestProcess(Process):
try:
test_run_init()
print(test_run_header())
# start the process for executing external python
py_fproc.start()
if not py_fproc.wait_ready(10):
raise ETUMRuntimeError(
f"""Impossible to start the external python execution process.
Is the python path correct ?
python_path = {tm.gd("python_path", "no python path defined")}"""
)
if lua_fproc is not None:
lua_fproc.start()
if not lua_fproc.wait_ready(10):
raise ETUMRuntimeError(
f"""Impossible to start the external lua execution process.
Is the lua path correct ?
lua_path = {tm.gd("lua_path", "no lua path defined")}
Are "lua-sockets" and "lua-cjson" installed ?
Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables ?"""
)
test_set.execute()
finally:
if test_set.success():
@@ -143,12 +265,6 @@ Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables
test_set.run_post_exec()
finally:
# Stop function execution process
py_fproc.stop()
py_fproc.join()
if lua_fproc is not None:
lua_fproc.stop()
lua_fproc.join()
self.__exec = False
# Sends signal to the GUI
self.send_finished()
@@ -156,6 +272,11 @@ Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables
except Exception as e:
print_exception(e)
finally:
# Stop python eval execution process
eval_proc.stop()
eval_proc.join()
except Exception as e:
print_exception(e)
@@ -238,7 +359,7 @@ Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables
continue
def redirect_stdout(self):
pipe = pconn, cconn = Pipe()
pconn, cconn = Pipe()
redir = Thread(target=self.capture_stdout, args=(cconn,))
redir.daemon = True
redir.start()

18
src/testium/interpreter/test_items/test_item.py
View File

@@ -33,21 +33,13 @@ def test_run(f):
c = self._prms.expanse(raw_condition)
if isinstance(c, bool):
condition = c
elif isinstance(c, (str, bytes)):
is_evaluated, condition = evaluate(c)
if not is_evaluated:
print("eval with c: {}".format(c))
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a 'condition' impossible to evaluate",
self.seqFilename(),
)
else:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' has a 'condition' result ({c}) which is not string or bool",
self.seqFilename(),
)
c = False
msg = '"{}" --> "{}"'.format(raw_condition, c)
if raw_condition == c:
msg = f'"{c}"'
else:
msg = f'"{raw_condition}" --> "{c}"'
# Do we have to skip the test because of a true condition ?
if condition:

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

@@ -1,13 +1,13 @@
import traceback
from interpreter.utils.tum_except import ETUMSyntaxError, ETUMRuntimeError
from interpreter.utils.py_func_exec import py_func_exec
from interpreter.utils.py_func_exec import PyFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.test_items.test_item import TestItem, test_run
from interpreter.test_items.test_result import TestResult, TestValue
import libs.testium as tm
from interpreter.utils.params import TestItemParams
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import evaluate
class TestItemCycle(TestItem):
@@ -96,11 +96,10 @@ class TestItemCycle(TestItem):
if isinstance(iter, str):
iter = self._prms.expanse(iter)
if not isinstance(iter, (list, tuple, int)):
_, iter = evaluate(iter)
if not isinstance(iter, (list, tuple, int)):
self.result.set(TestValue.FAILURE, f"unrecognized type for iterator '{str(iter)}'")
return
if not isinstance(iter, int):
r = []
for i in iter:
@@ -174,8 +173,13 @@ class TestItemCycle(TestItem):
exit_val = self._prms.expanse(
self._exit_condition
)
_, exit_val = evaluate(exit_val)
if exit_val:
ev = False
if isinstance(exit_val, bool):
ev = exit_val
else:
tm.print_warn(f"""Loop 'exit_condition' is not a boolean value ({exit_val}),
then considered as 'False'""")
if ev:
# exit condition is True
self.result.reported = {
"exit": "condition",
@@ -190,9 +194,7 @@ class TestItemCycle(TestItem):
break
else:
print(
'Continuing. Condition "{}" not met.'.format(
self._exit_condition
)
f"Continuing. Condition '{self._exit_condition}' not a 'True' boolean."
)
if self._exit_func:
@@ -204,7 +206,21 @@ class TestItemCycle(TestItem):
pl = self._prms.expanse(param_list)
else:
pl = [self._currentLoop]
fsucc, res = py_func_exec(file, func, pl)
proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
proc.start()
if not proc.wait_ready(10):
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:
fsucc, res = proc.func_call(file, func, pl)
finally:
proc.stop()
proc.join()
if fsucc == TestValue.SUCCESS:
fres, _ = res
if fres:

23
src/testium/interpreter/test_items/test_item_let.py
View File

@@ -8,7 +8,6 @@ from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.utils.tum_except import ETUMSyntaxError
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import evaluate
class TestItemLet(TestItem):
"""let item usage.
@@ -22,10 +21,9 @@ class TestItemLet(TestItem):
self.is_container = False
try:
self._values_list = self._prms.getParamAll('values', default=[], required=False)
self._eval_list = self._prms.getParamAll('eval', default=[], required=False)
if (len(self._values_list) <= 0) and (len(self._eval_list) <= 0):
if len(self._values_list) <= 0:
raise ETUMSyntaxError(
f"The '{self.cmd()}' test item named '{self.name()}' must have a 'values' or 'eval' parameter",
f"The '{self.cmd()}' test item named '{self.name()}' must have a 'values' parameter",
self.seqFilename(),
)
except:
@@ -41,11 +39,6 @@ class TestItemLet(TestItem):
for k in self._values_list.keys():
l.append({k: self._values_list[k]})
self._values_list = l
if isinstance(self._eval_list, dict):
l = []
for k in self._eval_list.keys():
l.append({k: self._eval_list[k]})
self._eval_list = l
#test core function
for i in self._values_list:
for k, v in i.items():
@@ -55,16 +48,4 @@ class TestItemLet(TestItem):
self.result.reported = {key: ev}
print('global value "{}" set to "{}"'.format(key, ev))
for i in self._eval_list:
for k, v in i.items():
key = self._prms.expanse(k)
val = self._prms.expanse(v)
is_evaluated, ev = evaluate(val)
if not is_evaluated:
self.result.set(TestValue.FAILURE, "Error evaluating: '{}'".format(val))
return
tm.setgd(key, ev)
self.result.reported = {key: ev}
print('global value "{}" set to "{}"'.format(key, ev))
self.result.set(TestValue.SUCCESS, 'Variable set')

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

@@ -7,8 +7,9 @@ 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.lua_func_exec import lua_func_exec
from interpreter.utils.tum_except import ETUMSyntaxError
from interpreter.utils.lua_func_exec import lua_func_call_init, lua_func_exec
from interpreter.utils.api_srv import api_request
from interpreter.utils.tum_except import ETUMSyntaxError, ETUMRuntimeError
from interpreter.utils.constants import TestItemType as cst
@@ -31,6 +32,9 @@ class TestItemLuaFunc(TestItem):
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._proc = lua_func_call_init(tm.gd("lua_path", ""), api_request, 10)
@test_run
def execute(self):
@@ -45,7 +49,24 @@ class TestItemLuaFunc(TestItem):
if tm.debug_enabled():
tm.print_debug("Parameters list:")
tm.print_debug(textwrap.indent(pprint.pformat(pl), " |"))
if self._proc is not None:
self._proc.start()
if not self._proc.wait_ready(10):
raise ETUMRuntimeError(
f"""Impossible to start the external lua execution process.
Is the lua path correct ?
lua_path = {tm.gd("lua_path", "no lua path defined")}
Are "lua-sockets" and "lua-cjson" installed ?
Is the lua environnment well defined in the "LUA_PATH" and "LUA_CPATH" variables ?"""
)
try:
success, ret = lua_func_exec(self.file_name, self.func_name, pl)
finally:
# Stops lua function execution process
self._proc.stop()
self._proc.join()
if success == TestValue.SUCCESS:
self.result.set(TestValue.SUCCESS)

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

@@ -7,8 +7,9 @@ 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.py_func_exec import PyFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils.tum_except import ETUMSyntaxError, ETUMRuntimeError
from interpreter.utils.constants import TestItemType as cst
@@ -31,6 +32,7 @@ class TestItemPyFunc(TestItem):
f"The '{self.cmd()}' test item named '{self.name()}' (child of '{self.parent.name()}') has a missing or wrong parameter",
self.seqFilename(),
)
self._proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
@test_run
def execute(self):
@@ -45,7 +47,22 @@ class TestItemPyFunc(TestItem):
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)
# start the process for executing external python
self._proc.start()
if not self._proc.wait_ready(10):
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._proc.func_call(self.file_name, self.func_name, pl)
finally:
# Stops python function execution process
self._proc.stop()
self._proc.join()
if success == TestValue.SUCCESS:
self.result.set(TestValue.SUCCESS)

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

@@ -33,7 +33,7 @@ class TestItemRun(TestItem):
try:
self.tum_fime = self._prms.getParam('tum_fime', required=True)
self.param_file = self._prms.getParam('param_file', default='')
self.python_path = self._prms.getParam('python_path', default='')
self.python_bin = self._prms.getParam('python_bin', default='')
self.testium_path = self._prms.getParam('testium_path', default='')
self.log_path = self._prms.getParam('log_file', default='')
self.report_path = self._prms.getParam('report_file', default='')
@@ -58,7 +58,7 @@ class TestItemRun(TestItem):
'"{}" file could not be found'.format(file_path))
self.tum_fime = file_path
pf = self._prms.expanse(self.param_file)
pp = self._prms.expanse(self.python_path)
pp = self._prms.expanse(self.python_bin)
sp = self._prms.expanse(self.testium_path)
lp = self._prms.expanse(self.log_path)
rp = self._prms.expanse(self.report_path)

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

@@ -8,7 +8,9 @@ from interpreter.utils.tum_except import (
)
import interpreter.utils.settings as prefs
from interpreter.test_report.test_report import TestReport
from interpreter.utils.py_func_exec import py_func_exec
from interpreter.utils.py_func_exec import PyFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils.tum_except import ETUMRuntimeError
from interpreter.utils.constants import TestItemType as cst_type
import interpreter.utils.constants as cst
from interpreter.utils.constants import TEST_TYPE_LIST
@@ -342,21 +344,33 @@ class TestSet:
tm.print_debug(f' No file: "{post_exec_file}".')
return
proc = PyFuncExecEngine(tm.gd("python_bin", ""), api_request, 10)
# start the process for executing external python
proc.start()
try:
if not proc.wait_ready(10):
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")}"""
)
tm.print_debug(f'Post-execution from: "{post_exec_file}"')
if self.rootItem().result.success:
# tests backup is done here
succ, res = py_func_exec(post_exec_file, "post_exec", [])
succ, res = proc.func_call(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 = py_func_exec(post_exec_file, "post_exec_fail", [])
succ, res = proc.func_call(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}"
)
finally:
proc.stop()
proc.join()
def rootItem(self):
return self._rootItem

3
src/testium/interpreter/utils/api_srv.py
View File

@@ -5,8 +5,9 @@ import libs.testium as tm
# Fill the api_dict with the function of tm
api_dict = {k: getattr(tm, k) for k in SUPPORTED_API if hasattr(tm, k)}
def api_request(method, params):
global api_dict
if method in api_dict.keys():
if params is None:
params = []

40
src/testium/interpreter/utils/eval.py
View File

@@ -1,12 +1,7 @@
import random
import os
import sys
import time
import platform
import math
import json
import libs.testium as tm
from interpreter.utils.tum_except import (ETUMSyntaxError, ETUMRuntimeError)
from interpreter.utils.py_eval import eval_exec
from interpreter.utils.tum_except import ETUMSyntaxError, ETUMRuntimeError
def evaluate(val, **replacement_dict):
v2 = val
@@ -16,37 +11,52 @@ def evaluate(val, **replacement_dict):
for key, replacement in replacement_dict.items():
val = val.replace(f"$({key})", str(replacement))
try:
v2 = eval(val)
v2 = eval_exec(val)
except Exception as e:
# eval can crash
if tm.debug_enabled():
s = f"Evaluation of '{val}' failed with message:\n " + str(e)
tm.print_debug(s)
v2 = val
evaluated = (val != v2)
evaluated = val != v2
return evaluated, v2
def eval_to_boolean(c):
if isinstance(c, bool):
condition = c
elif isinstance(c, (str, bytes)):
if c.lower() in ['true', 't', 'y', 'yes', 'ok', ]:
if c.lower() in [
"true",
"t",
"y",
"yes",
"ok",
]:
condition = True
elif c.lower() in ['f', 'n', 'nok', 'ko', 'false', 'no',]:
elif c.lower() in [
"f",
"n",
"nok",
"ko",
"false",
"no",
]:
condition = False
else:
try:
cond = eval(c)
cond = eval_exec(c)
condition = eval_to_boolean(cond)
except Exception as e:
print("eval with c: {}".format(c))
raise e
elif type(c) is int:
condition = (c > 0)
condition = c > 0
else:
raise ETUMSyntaxError('c : {} not string, int or bool'.format(c))
raise ETUMSyntaxError("c : {} not string, int or bool".format(c))
return condition
def post_evaluate(post_eval, res):
"""This function is evaluating the result of a test,
therefore it may include a $(result) parameter.

1
src/testium/interpreter/utils/jrpc.py
View File

@@ -1,4 +1,3 @@
import sys
import socket
import json
import threading

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

@@ -4,13 +4,66 @@ import shutil
import subprocess
import socket
import libs.testium as tm
from interpreter.utils.paths import sys_lua_path
from interpreter.utils.paths import sys_app_path_lin, sys_app_path_win
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_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_path):
res = False
v = _lua_version(lua_path)
if (v[0] == "5") and (v[1] >= "1"):
res = True
return res
def _sys_lua_path():
sys_lua_path = tm.gd("_sys_lua_path", "")
if sys_lua_path != "":
return sys_lua_path
cur_os = tm.OS()
if cur_os == "Windows":
func = sys_app_path_win
else:
func = sys_app_path_lin
sys_lua_path = func("lua")
if (sys_lua_path != "") and not _is_lua51(sys_lua_path):
tm.print_debug(f"'{sys_lua_path}' not a lua 5.1 min.")
sys_lua_path = ""
tm.setgd("_sys_lua_path", sys_lua_path)
return sys_lua_path
def lua_func_call_init(lua_path, request_handler, timeout):
"""
@@ -87,7 +140,7 @@ class LuaFuncExecEngine:
f"The passed executable is not a lua interpreter: '{lua_path}'"
)
else:
lua_path = sys_lua_path()
lua_path = _sys_lua_path()
if lua_path == "":
raise ETUMRuntimeError(
f"No valid lua interpreter found"

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

@@ -1,7 +1,7 @@
import interpreter.utils.globdict as globdict
from interpreter.utils.eval import evaluate
from interpreter.utils.tum_except import ETUMSyntaxError, ETUMRuntimeError
glob_eval_func = None
class TestItemParams:
@@ -295,12 +295,6 @@ def _operate_param(glob, parent):
return treated, g
# def _dummy_eval(val):
# bla = evaluate(val)
# print("******** evaluate(" + str(val) + ") = " + str(bla[1]))
# return bla
def _preprocess_string(value, parent=None):
"""This function parses a string value to check if patterns corresponding
to $(xxx) exists.
@@ -318,7 +312,8 @@ def _eval_param(value):
content is done.
If it is not evaluable, not replaced.
"""
return _parse_and_process("$|", "|", value, evaluate)
global glob_eval_func
return _parse_and_process("$|", "|", value, glob_eval_func)
def _process_recursively(func, param_value, *fparams):
@@ -377,3 +372,9 @@ def expanse(param_value, parent=None):
result = tmp_res
n += 1
return result
def eval_func_init(eval_func):
global glob_eval_func
glob_eval_func = eval_func

138
src/testium/interpreter/utils/paths.py
View File

@@ -36,76 +36,7 @@ def abs_path_from_file(file):
return abs_file_path
def sys_encoding():
if tm.OS() == "Windows":
enc = 'oem'
else:
enc = 'utf-8'
return enc
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=sys_encoding(),
timeout=10
)
data = result.stdout
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired) as e:
tm.print_debug(str(e))
data = ""
return eval(data)
def _lua_version(path: str):
cmd = f'"{path}" -v'
try:
result = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
encoding=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_python3(python_path):
try:
v = _python_version(python_path)
if v[0] == 3:
res = True
except:
res = False
return res
def is_lua51(lua_path):
res = False
v = _lua_version(lua_path)
if (v[0] == "5") and (v[1] >= "1"):
res = True
return res
def _sys_app_path_win(app_name):
def sys_app_path_win(app_name):
try:
result = subprocess.run(
f"where {app_name}",
@@ -113,79 +44,28 @@ def _sys_app_path_win(app_name):
capture_output=True,
text=True,
encoding="oem",
timeout=10
timeout=10,
)
data = result.stdout
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
data = ""
sys_python_path = data.splitlines()
for l in sys_python_path:
sys_python_bin = data.splitlines()
for l in sys_python_bin:
if f"{app_name}.exe" in l:
return l
return ""
def _sys_app_path_lin(app_name):
def sys_app_path_lin(app_name):
try:
result = subprocess.run(
f"which {app_name}",
shell=True,
capture_output=True,
text=True,
timeout=10
f"which {app_name}", shell=True, capture_output=True, text=True, timeout=10
)
data = result.stdout
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
data = ""
sys_python_path = data.splitlines()
for l in sys_python_path:
if (
(f"{app_name}" in l)
and not l.startswith("which:")
):
sys_python_bin = data.splitlines()
for l in sys_python_bin:
if (f"{app_name}" in l) and not l.startswith("which:"):
return l
return ""
def sys_python_path():
sys_python_path = tm.gd("_sys_python_path", "")
if sys_python_path != "":
return sys_python_path
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_path = func(e)
if sys_python_path == "":
continue
if not is_python3(sys_python_path):
sys_python_path = ""
continue
tm.setgd("_sys_python_path", sys_python_path)
return sys_python_path
def sys_lua_path():
sys_lua_path = tm.gd("_sys_lua_path", "")
if sys_lua_path != "":
return sys_lua_path
cur_os = tm.OS()
if cur_os == "Windows":
func = _sys_app_path_win
else:
func = _sys_app_path_lin
sys_lua_path = func("lua")
if (sys_lua_path != "") and not is_lua51(sys_lua_path):
tm.print_debug(f"'{sys_lua_path}' not a lua 5.1 min.")
sys_lua_path = ""
tm.setgd("_sys_lua_path", sys_lua_path)
return sys_lua_path

49
src/testium/interpreter/utils/py_eval.py Normal file
View File

@@ -0,0 +1,49 @@
from interpreter.utils.py_process import PyProcessBase
from interpreter.utils.tum_except import ETUMRuntimeError
import libs.testium as tm
eval_process = None
def eval_process_init(python_bin, request_handler, timeout, python_path):
global eval_process
eval_process = EvalExecEngine(python_bin, request_handler, timeout, python_path)
return eval_process
class EvalExecEngine(PyProcessBase):
def eval(self, value):
if (self._rpc is not None) and self._rpc.is_alive():
answer = self._rpc.call(
"eval",
{
"value": value,
},
)
if "result" in answer:
return answer["result"]
# In case an error was encountered in the called function
elif "error" in answer:
raise ETUMRuntimeError(answer["result"])
else:
raise ETUMRuntimeError(
"Unexepected eval 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 eval_exec(value):
global eval_process
if eval_process is not None and eval_process.is_alive():
result = eval_process.eval(value)
else:
raise ETUMRuntimeError(
"No function execution process active. To be reported to testium support team."
)
return result

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

@@ -1,129 +1,10 @@
import os
import sys
import shutil
import subprocess
import socket
import libs.testium as tm
from interpreter.utils.paths import sys_python_path
from interpreter.utils.py_process import PyProcessBase
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 py_func_call_init(python_path, request_handler, timeout):
global function_call_process
function_call_process = PyFuncExecEngine(python_path, request_handler, timeout)
return function_call_process
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
class PyFuncExecEngine:
def __init__(self, python_path="", request_handler=None, timeout=10):
if python_path != "":
if shutil.which(python_path) is None:
raise ETUMRuntimeError(
f"The passed python path is not pointing to an executable: '{python_path}'"
)
if not is_python_interpreter(python_path):
raise ETUMRuntimeError(
f"The passed executable is not a python interpreter: '{python_path}'"
)
else:
python_path = sys_python_path()
if python_path == "":
raise ETUMRuntimeError(
f"No valid python interpreter found"
)
tm.setgd("python_path", python_path)
self._ppath = python_path
self._req_handler = request_handler
self._process = None
self._port = 0
self._timeout = timeout
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.")
# POpen config
CUST_ENV = {
"PATH": {"replace": False},
"PYTHONPATH": {"replace": True},
}
py_env = tm.gd("python_env", {})
env = os.environ.copy()
for k, v in CUST_ENV.items():
e = py_env.get(k, "")
if e != "":
if v["replace"]:
env[k] = e
else:
env[k] = e + ";" + env.get(k, "")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("localhost", 0))
self._port = sock.getsockname()[1]
func_proc_path = tm.gd("testium_path")
params = [self._ppath, "-m", "py_func", "-p", f"{self._port}", "-t", f"{self._timeout}"]
if tm.debug_enabled() and tm.gd("debug_rpc", False):
params.append("-v")
self._process = subprocess.Popen(
params, env=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("localhost", self._port, req_handler=self._req_handler)
if tm.debug_enabled() and tm.gd("debug_rpc", False):
self._rpc.dbg_out = sys.stdout
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()
class PyFuncExecEngine(PyProcessBase):
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():
@@ -159,19 +40,3 @@ class PyFuncExecEngine:
raise ETUMRuntimeError(
"No function execution process active. To be reported to testium support team."
)
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
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

184
src/testium/interpreter/utils/py_process.py Normal file
View File

@@ -0,0 +1,184 @@
import os
import shutil
import sys
import subprocess
import socket
import libs.testium as tm
from interpreter.utils.paths import sys_app_path_lin, sys_app_path_win
from interpreter.utils.tum_except import ETUMRuntimeError
from interpreter.utils.jrpc import JsonRpcClient
from interpreter.utils.paths import testium_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 = tm.gd("_sys_python_bin", "")
if sys_python_bin != "":
return 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 not _is_python3(sys_python_bin):
sys_python_bin = ""
continue
tm.setgd("_sys_python_bin", sys_python_bin)
return sys_python_bin
class PyProcessBase:
CUST_ENV = {
"PATH": {"replace": False},
"PYTHONPATH": {"replace": True},
}
def __init__(self, python_bin="", request_handler=None, timeout=10, python_path=""):
if (python_bin is not None) and (python_bin != ""):
if shutil.which(python_bin) is None:
raise ETUMRuntimeError(
f"The passed python path is not pointing to an executable: '{python_bin}'"
)
if not _is_python_interpreter(python_bin):
raise ETUMRuntimeError(
f"The passed executable is not a python interpreter: '{python_bin}'"
)
else:
python_bin = _sys_python_bin()
if python_bin == "":
raise ETUMRuntimeError(f"No valid python interpreter found")
tm.setgd("python_bin", python_bin)
self._pbin = python_bin
self._ppath = python_path
self._req_handler = request_handler
self._process = None
self._port = 0
self._timeout = timeout
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.")
# POpen config
py_env = tm.gd("python_env", {})
env = os.environ.copy()
for k, v in self.CUST_ENV.items():
e = py_env.get(k, "")
if e != "":
if v["replace"]:
env[k] = e
else:
env[k] = e + os.pathsep + env.get(k, "")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("localhost", 0))
self._port = sock.getsockname()[1]
# Add the path of the subprocess (root sources of testium)
func_proc_path = testium_path()
env["PYTHONPATH"] = func_proc_path + os.pathsep + self._ppath + os.pathsep + env.get("PYTHONPATH", "")
params = [
self._pbin,
"-m",
"py_func",
"-p",
f"{self._port}",
"-t",
f"{self._timeout}",
]
if tm.debug_enabled() and tm.gd("debug_rpc", False):
params.append("-v")
self._process = subprocess.Popen(params, env=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(
"localhost", self._port, req_handler=self._req_handler
)
if tm.debug_enabled() and tm.gd("debug_rpc", False):
self._rpc.dbg_out = sys.stdout
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 is_alive(self):
if self._rpc is not None:
return self._rpc.is_alive()
return False
def stop(self):
if self._rpc is not None:
self._rpc.stop()

8
src/testium/interpreter/utils/settings.py
View File

@@ -259,14 +259,14 @@ class TestiumSettings():
def git_supported(self, value):
self.set_value(self.SettingsGitSupported, value)
# SettingsPythonPath = 'pythonPath'
# SettingsPythonPath = 'python_bin'
@property
def python_path(self):
def python_bin(self):
r = self.value(self.SettingsPythonPath, "")
return r
@python_path.setter
def python_path(self, value):
@python_bin.setter
def python_bin(self, value):
self.set_value(self.SettingsPythonPath, value)
# SettingsLuaPath = 'luaPath'

10
src/testium/interpreter/utils/template.py
View File

@@ -1,7 +1,7 @@
import os
from sys import exc_info
from jinja2 import Template
from jinja2.exceptions import TemplateError, UndefinedError
from jinja2.exceptions import TemplateSyntaxError, TemplateError, UndefinedError
from tempfile import TemporaryFile
from interpreter.utils.yaml_load import print_yaml
from interpreter.utils.tum_except import ETUMSyntaxError
@@ -29,17 +29,17 @@ def template_to_test(filename: str, params: list):
params["include_directory"] = os.path.dirname(os.path.abspath(filename))
tmpf.write(j2_template.render(params))
except TemplateSyntaxError as e:
raise ECPTSyntaxError(f"""Template loading of file '{filename}' with following parameters '{str(params)}'
raise ETUMSyntaxError(f"""Template loading of file '{filename}' with following parameters '{str(params)}'
Syntax error in template: {e.message}""")
except UndefinedError as e:
raise ECPTSyntaxError(f"""Template loading of file '{filename}' with following parameters '{str(params)}'
raise ETUMSyntaxError(f"""Template loading of file '{filename}' with following parameters '{str(params)}'
Undefined variable error: {e.message}""")
except TemplateError as e:
raise ECPTSyntaxError(f"""Template loading of file '{filename}' with following parameters '{str(params)}'
raise ETUMSyntaxError(f"""Template loading of file '{filename}' with following parameters '{str(params)}'
Template rendering error: {e.message}""")
except Exception as e:
# Catch any other unexpected errors
raise ECPTSyntaxError(f"""Template loading of file '{filename}' with following parameters '{str(params)}'
raise ETUMSyntaxError(f"""Template loading of file '{filename}' with following parameters '{str(params)}'
Unexpected error: {str(e)}""")
# return to begining of the temp file

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

@@ -3,12 +3,9 @@ from pathlib import Path
import datetime
from socket import gethostname
import ast
import json
import yaml
import copy
import yaml
from interpreter.utils.constants import TestItemType as cst
import libs.testium as tm
import interpreter.utils.globdict as globdict
@@ -16,13 +13,14 @@ import interpreter.utils.settings as prefs
from interpreter.utils.paths import testium_path
from interpreter.utils.yaml_load import yaml_load
from interpreter.utils import clear_recursively
from interpreter.utils.include import TUMLoader, TUMLoaderNoIncludes, TUMLoaderRawIncludes
from interpreter.utils.tum_except import ETUMSyntaxError
from interpreter.utils.params import (expanse)
from interpreter.utils.version import (
get_version, get_testium_version, get_modifications)
from interpreter.utils.params import expanse, eval_func_init
from interpreter.utils.eval import evaluate
from interpreter.utils.template import template_to_test
from interpreter.utils.version import (
get_version,
get_testium_version,
get_modifications,
)
from interpreter.test_items.test_item import TestItem
from interpreter.test_items.test_item_sleep import TestItemSleep
@@ -73,33 +71,11 @@ def _constants_init():
cst.TYPE_VALUE_DLG.item_class = TestItemValueDialog
def _locate_config_files(test_dir, config_files, silent=False):
ret = []
pf = []
if len(config_files) == 0:
for p in ['param.yaml', 'param.yml']:
param_filename = os.path.join(test_dir, p)
if os.path.exists(param_filename):
pf.append(param_filename)
if not silent:
tm.print_info(f"Configuration file loaded: {p}.")
else:
if not silent:
tm.print_info(f"Default param file \"{p}\" does not exist.")
else:
pf = config_files
for p in pf:
ret.append(p)
return ret
def locate_report_file(rep_file):
# report file name treatment
if rep_file != '':
if rep_file != "":
if not os.path.isabs(rep_file):
rep_file = os.path.join(
os.getcwd(), rep_file)
rep_file = os.path.join(os.getcwd(), rep_file)
rep_file = os.path.normpath(rep_file)
if not os.path.exists(os.path.dirname(rep_file)):
os.makedirs(os.path.dirname(rep_file))
@@ -107,111 +83,9 @@ def locate_report_file(rep_file):
return rep_file
def _config_files_from_test(test_dict, config_files=None):
test_dir = tm.gd('test_directory')
pf = []
if isinstance(config_files, list) and len(config_files) == 0:
param_filename = test_dict.get('config_file', None)
if param_filename is None:
param_node = test_dict.get('param_file', None)
if param_node is not None:
if isinstance(param_node, dict):
p = param_node.get('file_name', None)
if p is not None:
param_filename = p
else:
param_filename = param_node
else:
param_filename = param_node
if param_filename is None:
pf = _locate_config_files(test_dir, [])
elif isinstance(param_filename, str):
pf.append(param_filename)
elif isinstance(param_filename, (list)):
pf = []
for p in param_filename:
if isinstance(p, list):
for pp in p:
pf.append(pp)
elif p is not None:
pf.append(p)
else:
raise ETUMSyntaxError(
'Unrecognized tum "param_file" : {}'.format(param_filename))
elif isinstance(config_files, list):
pf = config_files
elif isinstance(config_files, str):
pf = [config_files]
else:
raise ETUMSyntaxError(
'Unrecognized config_files parameter : {}'.format(config_files))
return pf
def _load_test_dict(test_file, variables: dict, no_include: bool = False, raw_include: bool = False):
loader = TUMLoader
loader = TUMLoaderRawIncludes if raw_include else loader
loader = TUMLoaderNoIncludes if no_include else loader
# Jinja template processing
tmpf = template_to_test(test_file, variables)
try:
d = yaml_load(tmpf, test_file, loader)
finally:
tmpf.close()
return d
def load_test(test_file, test_dir, cmdline_pfs, cmdline_defs, gui_defaults):
# First step: populate config files without includes considered
test_dict = _load_test_dict(test_file, {}, no_include=True)
_check_test_dict(test_dict)
prepare_global()
# Define the global builtin variables
set_standard_gd_keys(test_dict["main"].get(
"name", "Unnamed"), test_dir, test_file, cmdline_pfs)
# Include the content of the first config files into glob dict
old_pfs = _config_files_from_test(test_dict, cmdline_pfs)
# Variables updated
gd = update_global(old_pfs, cmdline_defs, gui_defaults, silent=True)
while True:
# Loop to check param files until all param files are identified
test_dict = _load_test_dict(test_file, gd, raw_include=True)
new_pfs = _config_files_from_test(test_dict, cmdline_pfs)
# Check if things have changed since previous evaluation of
# config files
new_stuff = False
if len(old_pfs) != len(new_pfs):
new_stuff = True
if not new_stuff:
for i in range(len(old_pfs)):
if old_pfs[i] != new_pfs[i]:
new_stuff = True
break
# If the param files are identical, we continue in loading process
if not new_stuff:
break
# Variables updated
gd = update_global(new_pfs, cmdline_defs, gui_defaults, silent=False)
old_pfs = copy.copy(new_pfs)
# Processing (with includes) for complete file loading
test_dict = _load_test_dict(test_file, gd)
return test_dict, new_pfs
def yamltodict(param_file, silent=True):
# load of the file
with open(param_file, 'r') as fd:
with open(param_file, "r") as fd:
dp = yaml_load(fd, param_file, yaml.Loader)
if dp is None:
@@ -229,9 +103,9 @@ def yamltodict(param_file, silent=True):
if not silent:
if not tm.debug_enabled():
tm.print_info(f"\"{param_file}\" loaded.")
tm.print_info(f'"{param_file}" loaded.')
else:
tm.print_debug(f"\"{param_file}\" loading:")
tm.print_debug(f'"{param_file}" loading:')
for k, v in dp.items():
tm.print_debug(f" {k}: {v}")
tm.print_debug(f"done.")
@@ -241,7 +115,7 @@ def yamltodict(param_file, silent=True):
def _feed_gd_with_params(param_file, silent=True):
test_dir = tm.gd('test_directory')
test_dir = tm.gd("test_directory")
# param files pre-processing
files = []
for p in param_file:
@@ -263,44 +137,35 @@ def _feed_gd_with_params(param_file, silent=True):
raise ETUMSyntaxError(f'Parameter file "{pf}" not found')
ext = os.path.splitext(pf)[1]
if (ext == '.yaml') or (ext == '.yml'):
if (ext == ".yaml") or (ext == ".yml"):
yamltodict(pf, silent)
else:
raise ETUMSyntaxError(
'config files must be "*.yaml" or "*.yml"')
raise ETUMSyntaxError('config files must be "*.yaml" or "*.yml"')
def set_standard_gd_keys(test_name, test_dir, test_file, config_files):
tm.setgd('testium_version', get_testium_version())
tm.setgd('testium_path', testium_path())
tm.setgd('test_name', test_name)
tm.setgd('test_directory', test_dir)
tm.setgd('test_main_file', test_file)
tm.setgd('config_files', config_files)
tm.setgd('host_name', gethostname())
tm.setgd('home', str(Path.home()))
tm.setgd('os', tm.OS())
tm.setgd("testium_version", get_testium_version())
tm.setgd("testium_path", testium_path())
tm.setgd("test_name", test_name)
tm.setgd("test_directory", test_dir)
tm.setgd("test_main_file", test_file)
tm.setgd("config_files", config_files)
tm.setgd("host_name", gethostname())
tm.setgd("home", str(Path.home()))
tm.setgd("os", tm.OS())
def env_init():
if not hasattr(prefs, "settings"):
prefs.init()
eval_func_init(evaluate)
_constants_init()
def _check_test_dict(test_dict):
if not isinstance(test_dict, dict):
raise ETUMSyntaxError(
"The tum file has a major problem. Please check the documentation for syntax.")
if not 'main' in test_dict.keys():
raise ETUMSyntaxError(
"The tum file has a major problem. The 'main' section could not be found.")
def update_global(config_files, defines, gui_defaults, silent=False):
'''Global dict updated with the content of the config file and a dict provided.
"""Global dict updated with the content of the config file and a dict provided.
this function returns the resulting dict.
'''
"""
# GUI preferences applied first
for k, v in gui_defaults.items():
try:
@@ -332,7 +197,9 @@ def update_global(config_files, defines, gui_defaults, silent=False):
conf_val = tm.gd(k)
if val != conf_val:
if not silent:
tm.print_info(f"Variable $({k}) overloaded by command line arg --> \"{val}\".")
tm.print_info(
f'Variable $({k}) overloaded by command line arg --> "{val}".'
)
tm.setgd(k, val)
return globdict.global_dict
@@ -355,48 +222,52 @@ def restore_gd(dict):
def test_run_init():
tm.init_timestamp()
test_dir = tm.gd('test_directory')
tm.setgd('test_version', get_version(test_dir))
tm.setgd('test_modifs', get_modifications(test_dir))
test_dir = tm.gd("test_directory")
tm.setgd("test_version", get_version(test_dir))
tm.setgd("test_modifs", get_modifications(test_dir))
start_test_date = datetime.datetime.now()
tm.setgd('start_test_date', start_test_date)
tm.setgd('testrun_date', start_test_date.strftime("%Y-%m-%d"))
tm.setgd('testrun_time', start_test_date.strftime("%H:%M:%S"))
tm.setgd("start_test_date", start_test_date)
tm.setgd("testrun_date", start_test_date.strftime("%Y-%m-%d"))
tm.setgd("testrun_time", start_test_date.strftime("%H:%M:%S"))
def test_run_header():
tool_version = tm.gd('testium_version')
test_file = tm.gd('test_main_file', '')
has_test_file = (tm.gd('test_main_file') != '')
tool_version = tm.gd("testium_version")
test_file = tm.gd("test_main_file", "")
has_test_file = tm.gd("test_main_file") != ""
s = ''
s += (80*'=') + '\n'
s += '====== Test overview' + '\n'
s += (80*'=') + '\n'
s = ""
s += (80 * "=") + "\n"
s += "====== Test overview" + "\n"
s += (80 * "=") + "\n"
if has_test_file:
s += ('Executed test file : ' + test_file) + '\n'
for cf in tm.gd('config_files'):
s += ('With param file : {}'.format(cf)) + '\n'
s += ('Test started : ' + tm.gd('testrun_date') + ' ' +
tm.gd('testrun_time')) + '\n'
s += ("Executed test file : " + test_file) + "\n"
for cf in tm.gd("config_files"):
s += ("With param file : {}".format(cf)) + "\n"
s += (
"Test started : "
+ tm.gd("testrun_date")
+ " "
+ tm.gd("testrun_time")
) + "\n"
s += (80*'=') + '\n'
s += ('====== Test configuration') + '\n'
s += (80*'=') + '\n'
s += ('Test executed with testium : ' +
tool_version.splitlines()[0]) + '\n'
s += (80 * "=") + "\n"
s += ("====== Test configuration") + "\n"
s += (80 * "=") + "\n"
s += ("Test executed with testium : " + tool_version.splitlines()[0]) + "\n"
for l in tool_version.splitlines()[1:]:
s += (32*' ' + ': ' + l) + '\n'
s += (' \n')
s += (32 * " " + ": " + l) + "\n"
s += " \n"
if has_test_file:
test_version = tm.gd('test_version')
test_modifs = tm.gd('test_modifs')
s += ('Test scripts revision : ' +
test_version.splitlines()[0]) + '\n'
test_version = tm.gd("test_version")
test_modifs = tm.gd("test_modifs")
s += (
"Test scripts revision : " + test_version.splitlines()[0]
) + "\n"
for l in test_version.splitlines()[1:]:
s += (32*' ' + ': ' + l) + '\n'
s += (32 * " " + ": " + l) + "\n"
for l in test_modifs.splitlines():
s += (' '+l) + '\n'
s += (" " + l) + "\n"
return s

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

@@ -16,9 +16,11 @@ import numpy as np
import matplotlib.dates as mdates
from datetime import datetime, timedelta, timezone
import libs.testium as tm
from interpreter.test_items.test_result import TestValue
from interpreter.utils.tum_except import ETUMRuntimeError
from interpreter.utils.py_func_exec import py_func_exec
from interpreter.utils.py_func_exec import PyFuncExecEngine
from interpreter.utils.api_srv import api_request
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
@@ -268,11 +270,19 @@ 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.start()
if not self.proc.wait_ready(10):
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")}"""
)
self.start()
self.on_timer_event()
def on_timer_event(self):
succ, ret = py_func_exec(self.file, self.func_name, self.args)
succ, ret = self.proc.func_call(self.file, self.func_name, self.args)
if succ == TestValue.SUCCESS:
res, _ = ret
res = post_evaluate(self.post_eval, res)
@@ -280,6 +290,11 @@ class RuntimePlotPeriodic(PeriodicTimer):
else:
print("Plot periodic timer function ({self.file}/{self.func_name}) failed: \"{ret}\"")
def stop(self):
self.proc.stop()
self.proc.join()
super().stop()
class RuntimePlot:
EXPORTS = [".pdf", ".csv"]

30
src/testium/libs/testium.py
View File

@@ -143,21 +143,21 @@ def last_plot_value(name: str) -> dict:
###############################################################################
class FunctionItem():
"""Class allowing extended capabilities of function."""
module_count = 0
# class FunctionItem():
# """Class allowing extended capabilities of function."""
# module_count = 0
def __init__(self):
self._reported_value = {}
# def __init__(self):
# self._reported_value = {}
def reportValue(self, key, value):
self._reported_value[key] = value
# def reportValue(self, key, value):
# self._reported_value[key] = value
def reportedValues(self):
return self._reported_value
# def reportedValues(self):
# return self._reported_value
def exec(self):
pass
# def exec(self):
# pass
def get_main_dir():
@@ -209,6 +209,14 @@ def OS():
return platform.system()
def sys_encoding():
if OS() == "Windows":
enc = "oem"
else:
enc = "utf-8"
return enc
def line_number(phrase, filename):
with open(filename, 'r') as f:
for (i, line) in enumerate(f):

2
src/testium/main_win/preference_win/preference_win.py
View File

@@ -101,7 +101,7 @@ class PrefWindow(QDialog):
prefs.settings.SettingsPythonPath: {
"type": "text",
"widget": self.ui.editPythonPath,
"value": prefs.settings.python_path,
"value": prefs.settings.python_bin,
"default": "",
"changed": False,
},

4
src/testium/main_win/testium_win.py
View File

@@ -977,9 +977,9 @@ class MainWindow(QMainWindow, Ui_MainWindow):
def defaults_for_process(self):
d = {}
pp = prefs.settings.python_path
pp = prefs.settings.python_bin
if pp != "":
d["python_path"] = pp
d["python_bin"] = pp
pp = prefs.settings.lua_path
if pp != "":

17
src/testium/py_func/__main__.py
View File

@@ -1,24 +1,21 @@
import os
from pathlib import Path
import sys
import logging
import traceback
logging.basicConfig(
level=logging.ERROR,
filename=os.path.join(os.path.normpath(os.getcwd()), "crash.txt"),
format="%(asctime)s - %(levelname)s - %(message)s"
)
def exception_handler(typ_exc, value, trbk):
"""Testium Exception handling"""
logging.error("An unmanaged exception occured", exc_info=(typ_exc, value, trbk))
print("An unmanaged exception occured", exc_info=(typ_exc, value, trbk))
print(f"Critical failure : '{value}'.")
tb = traceback.format_exception(typ_exc, value, trbk)
print("".join(tb))
sys.excepthook = exception_handler
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'))
p = Path(__file__)
p = p.parent / ".."
p = p.resolve()
sys.path.append(p)
from py_func import main

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

@@ -1,4 +1,10 @@
import random
import os
import sys
import time
import platform
import math
import json
import traceback
from interpreter.utils.jrpc import JsonRpcSrv
from interpreter.utils.tum_except import ETUMRuntimeError, print_exception
@@ -35,6 +41,22 @@ class FuncHandler(JsonRpcSrv):
return {
"error": f"bad jrpc req handler 'func_call' arguments ({"\n".join(tb.splitlines())}). To be reported to testium support team."
}
if method == "eval":
try:
value = params["value"]
try:
res = eval(value)
return {"result": res}
except Exception as e:
# eval can crash
return {
"error": f"Evaluation of '{value}' failed with message:\n "+str(e)
}
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."
}
else:
return {
"error": f"unknown RPC request ({method}). To be reported to testium support team."

4
test/validation/items/check/test.tum
View File

@@ -19,10 +19,10 @@
name: Check condition on existing variable (PASS)
key: $(test)_PASS
values:
- $(fn_Dummy_int) > 1
- $(pfn_Dummy_int) > 1
- check:
name: Check condition on existing variable (FAIL)
key: $(test)_FAIL
values:
- '"tailor" in "$(fn_Dummy_str)"'
- '"tailor" in "$(pfn_Dummy_str)"'

22
test/validation/items/common/conditional/test.tum
View File

@@ -16,17 +16,17 @@
- $(loop_param)
exit_condition:
value: "$(fn_Echo function) > 3"
value: $| $(pfn_Echo function) > 3 |
- let:
name: let
key: $(test)_PASS
eval:
- conditional_exec: "random.randint(1, 2)"
values:
- conditional_exec: $| random.randint(1, 2) |
- console:
name: Console creation
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
doc: Opening the console
key: $(test)_PASS
@@ -37,7 +37,7 @@
- console:
name: Console read_until with timeout
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
key: $(test)_PASS
steps:
@@ -45,7 +45,7 @@
- console:
name: Console write
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
key: $(test)_PASS
steps:
@@ -53,12 +53,12 @@
- sleep:
name: sleep item
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
timeout: 1
- console:
name: Console read_until immediate
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
key: $(test)_PASS
steps:
@@ -66,7 +66,7 @@
- console:
name: Console read_until immediate (2)
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
key: $(test)_PASS
steps:
@@ -74,7 +74,7 @@
- console:
name: Console closure
condition: "$(conditional_exec) == 1"
condition: $| $(conditional_exec) == 1 |
console_name: consname
key: $(test)_PASS
steps:
@@ -82,5 +82,5 @@
- sleep:
name: sleep item
condition: "$(conditional_exec) == 2"
condition: $| $(conditional_exec) == 2 |
timeout: 1

2
test/validation/items/jsonrpc/test.tum
View File

@@ -13,7 +13,7 @@
- group:
name: jsonrpc tests
condition: "'/jrpces' in r'''$(cn_json rpc echo server)'''"
condition: $| '/jrpces' in r'''$(cn_json rpc echo server)''' |
steps:
- console:
name: Start the json rpc echo server

14
test/validation/items/let/test.tum
View File

@@ -12,8 +12,7 @@
name: Let it be
values:
it: $(loop_param)
eval:
- be: "$(loop_param) == $(it)"
be: $| $(loop_param) == $(it) |
- loop:
name: Cycle iterating on list
@@ -31,8 +30,7 @@
name: Let it be
values:
- it: $(loop_param)
eval:
- be: "$(loop_param) == $(it)"
- be: $| $(loop_param) == $(it) |
- let:
name: Get time
@@ -45,8 +43,8 @@
- let:
name: Get parameter file value
key: $(test)_PASS
eval:
- test_overwrite_me: "$(overwrite_me) == True"
values:
- test_overwrite_me: $| $(overwrite_me) == True |
- py_func:
name: Check global dic pass
@@ -85,8 +83,8 @@
- let:
name: Evaluate Overwriting parameter file value
key: $(test)_PASS
eval:
- test_overwrite_me: '"$(overwrite_me)" == True'
values:
- test_overwrite_me: $| "$(overwrite_me)" == True |
- check:
name: Check Overwriting parameter file value

4
test/validation/main.tum
View File

@@ -13,14 +13,14 @@ main:
- let:
name: Set test variables for Linux
condition: "'$(os)' == 'Linux'"
condition: $| "$(os)" == "Linux" |
values:
- terminal_prompt: $(linux_prompt)
- psep: "/"
- let:
name: Set test variables for Windows
condition: "'$(os)' == 'Windows'"
condition: $| "$(os)" == "Windows" |
values:
- terminal_prompt: $(windows_prompt)
- psep: "\\"