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

fix: expand parameters at run time, not at load

Browse Source 
Variable substitution ($(...)) must use the runtime global dict, so it
must happen at run time (execute), never at load (__init__).

- console telnet_port: was never expanded — `telnet_port: $(port)` stayed
  literal. Now expanded at run (processed=True in execute, like the other
  host/port params).
- test_item base: stop_on_failure / execute_on_stop are now stored raw and
  resolved at run time via properties (so a $(...) flag reflects the
  runtime value, not the load-time one).
- cycle iterator and git repo: drop the redundant load-time expansion
  (execute() already re-expands them).
- tested_references: fetch 'reference' raw, expand each value in execute().

Justified load-time exceptions kept: name, doc, skipped (static/GUI at
load) and unittest test_method (drives child loading at load).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Foué
2026-06-14 17:19:07 +02:00
parent 3661a71145
commit 8a498dd6ac
6 changed files with 34 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/testium/api/console.py
View File

@@ -127,13 +127,11 @@ A {classname}.close() is missing somewhere in your code !'.format(classname=type
# c = ''
return c
# Upper bound (in characters) of the accumulated buffer tail scanned in
# regex mode, so cost/memory stay bounded on long-running streams.
# Max chars of the buffer tail scanned in regex mode (bounds cost/memory).
REGEX_WINDOW = 65536
def _feed_match(self, data, search_deques, match_deques, matches):
"""Append *data* to every rolling window and return the first matched
pattern string, or None if none completed on this character."""
"""Append *data* to each window; return the first matched pattern or None."""
matched = None
for sd, md, m in zip(search_deques, match_deques, matches):
sd.append(data)
@@ -142,8 +140,7 @@ A {classname}.close() is missing somewhere in your code !'.format(classname=type
return matched
def _search_regex(self, read_data, compiled):
"""Search the (bounded) tail of *read_data* with each compiled regex;
return the matched text of the first hit, or None."""
"""Search the buffer tail with each regex; return the first hit's text or None."""
tail = read_data[-self.REGEX_WINDOW:]
for p in compiled:
m = p.search(tail)
@@ -171,8 +168,7 @@ A {classname}.close() is missing somewhere in your code !'.format(classname=type
if not match:
raise ETUMRuntimeError("'expected' pattern can not be empty")
# match may be a single string or a list/tuple of strings: the read
# succeeds as soon as ANY of them is seen in the stream.
# match: a string or list of strings; succeed as soon as any is seen.
if isinstance(match, (list, tuple)):
matches = [str(m) for m in match]
else:
@@ -195,8 +191,7 @@ A {classname}.close() is missing somewhere in your code !'.format(classname=type
raise ETUMRuntimeError(
"Invalid regular expression {!r}: {}".format(m, e)) from None
else:
# One fixed-length rolling window per match pattern, compared
# against the corresponding pattern deque.
# One fixed-length rolling window per literal pattern.
search_deques = [collections.deque(maxlen=len(m)) for m in matches]
match_deques = [collections.deque(m) for m in matches]
self._matched = None

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

@@ -145,7 +145,7 @@ class TestItem:
self._report_key = None
self._reported = None
self.status_queue = status_queue
self._execute_on_stop = False
self._execute_on_stop_raw = False
self._post_eval = None
self._store_result = None
self._expected_result = None
@@ -154,7 +154,7 @@ class TestItem:
self._is_running = False
self._is_breakpoint = False
self._is_paused = False
self._stop_on_failure = False
self._stop_on_failure_raw = False
self._doc = ""
self._name = ""
self.report = None
@@ -197,13 +197,14 @@ class TestItem:
self.skipped = False
self._report_key = self._prms.getParam("key", default=None)
self._stop_on_failure = self._prms.getParam(
"stop_on_failure", default=False, processed=True
# Kept raw: expanded at run time by the matching properties.
self._stop_on_failure_raw = self._prms.getParam(
"stop_on_failure", default=False
)
self._doc = self._prms.getParam("doc", default="", processed=True)
#
self._execute_on_stop = self._prms.getParam(
"execute_on_stop", default=False, processed=True
self._execute_on_stop_raw = self._prms.getParam(
"execute_on_stop", default=False
)
if "process_result" in dict_item:
@@ -570,6 +571,20 @@ class TestItem:
def setEnabled(self):
self.enabled = True
def _eval_flag(self, raw):
"""Run-time flag: bool as-is, otherwise expanded and coerced to bool."""
if isinstance(raw, bool):
return raw
return eval_to_boolean(self._prms.expanse(raw))
@property
def _stop_on_failure(self):
return self._eval_flag(self._stop_on_failure_raw)
@property
def _execute_on_stop(self):
return self._eval_flag(self._execute_on_stop_raw)
def executedOnStop(self):
return self._execute_on_stop

5
src/testium/interpreter/test_items/test_item_console.py
View File

@@ -88,7 +88,7 @@ class TestItemConsoleOpen(TestItemConsoleAction):
telnet_host = self._prms.getParam(
"telnet_host", required=True, processed=True
)
telnet_port = self._prms.getParam("telnet_port", default=69)
telnet_port = self._prms.getParam("telnet_port", default=69, processed=True)
elif self._protocol == "ssh":
if tm.OS() == "Windows":
@@ -226,8 +226,7 @@ class TestItemConsoleOpen(TestItemConsoleAction):
cons.open()
self.result.set(TestValue.SUCCESS)
except ETUMRuntimeError as e:
# Expected, user-facing console error (device missing, no permission,
# …): report a single clear line, no traceback.
# Expected console error (device missing, no permission…): one line.
msg = "Impossible to open the console '{}': {}".format(cname, e._message)
self.result.set(result=TestValue.FAILURE, message=msg)
print(msg)

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

@@ -51,11 +51,8 @@ class TestItemCycle(TestItem):
self._niter = None
if "iterator" in dict_cycle:
# Kept raw: expanded at run time in execute().
self._iter = dict_cycle["iterator"]
if isinstance(self._iter, str):
self._iter = self._prms.expanse(self._iter)
else:
self._iter = None

3
src/testium/interpreter/test_items/test_item_git.py
View File

@@ -21,7 +21,8 @@ class TestItemGit(TestItem):
super().__init__(dict_item, parent, status_queue, filename=filename)
self._type = cst.TYPE_GIT
self.is_container = False
self.repo = self._prms.getParamAll('repo', processed=True, required=True)
# Kept raw: each repo entry is expanded at run time in execute().
self.repo = self._prms.getParamAll('repo', required=True)
@test_run
def execute(self):

5
src/testium/interpreter/test_items/test_item_tested_references.py
View File

@@ -26,13 +26,14 @@ class TestItemTestedRefsDialog(TestItemDialogBase):
self.is_container = False
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
self._init_values = self._prms.getParamAll('reference', required=False, processed=True)
# Kept raw: expanded at run time in execute().
self._init_values = self._prms.getParamAll('reference', required=False)
self._auto_result = self._prms.getParam('auto_result', required=False, default=None)
@test_run
def execute(self):
q = self._prms.expanse(self._question)
init_values = ','.join(self._init_values)
init_values = ','.join(self._prms.expanse(v) for v in self._init_values)
if _is_text_mode():
print(f"References: {q}")
rows = init_values.split(',') if init_values else ['']