3 Commits

Author SHA1 Message Date
354c5e12e8 bump to 0.2: declarative test item parameters
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 13:16:45 +02:00
b1a7dac0f3 item params: migrate every structured item to declarative PARAMS
Migrates the remaining test items to the ParamSet/Param declaration
introduced in d0721af:
  - dialogs: image, question, value, choices, tested_references
  - actions: check, run, report
  - console: parent + open/read_until actions
  - py_func / lua_func
  - containers: group, parallel + parallel_branch, unittest
  - complex: cycle (sub-block exit_condition documented in
    EXIT_CONDITION_PARAMS), git
  - runtime_plot: parent + open/close/periodic/last_value actions
  - json_rpc: parent + query/receive actions

Items intentionally without PARAMS (and therefore not validated) are
those whose body is the unstructured user value: console write/writeln,
plot add/export, and the json_rpc/console open & close actions. Same
for the internally-instantiated TestItemUnittestElement which passes
dict_item=None.

Behavior on valid .tum files is unchanged (validation suite source
mode: SUCCESS). Typos on declared params now surface as warnings
listing the accepted names; missing required params surface as load-
time errors with file context.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 11:14:42 +02:00
d0721af719 item params: declarative descriptor foundation + 4 pilot items
Adds utils/param_decl.py with Param/ParamSet/kind descriptors. TestItem
declares COMMON_PARAMS (name, doc, condition, key, skipped, …) and a
new _validate_declared_params() method that warns on unknown keys and
errors on missing required ones — opt-in per subclass (skipped while
PARAMS is None to keep the migration incremental).

Migrates sleep, let, msg_dialog, note_dialog as pilots. Behavior is
unchanged for any well-formed .tum; typos like 'timeoot' on a sleep
item now produce a clear WARN listing the accepted parameters.

The descriptor intentionally carries no Python type information —
parameter values that are $(…) / <|…|> expressions only acquire their
effective type after expansion, so a static type would be misleading.
Per-param post-expansion validators stay opt-in via validate=lambda.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 10:45:57 +02:00
26 changed files with 629 additions and 1 deletions

View File

@@ -1,3 +1,11 @@
version 0.2
==============
- Test items: each item type now declares its accepted parameters
(``PARAMS = ParamSet(...)``). Typos in a ``.tum`` are surfaced as a
WARN listing the accepted names instead of being silently ignored;
missing required parameters error out at load time with the source
``.tum`` file as context. No change to valid existing tests.
version 0.1.3
==============
- Stop interrupts engaged blocking steps (console, py_func, lua_func,

View File

@@ -1 +1 @@
0.1.3
0.2

View File

@@ -5,6 +5,9 @@ from copy import deepcopy
from interpreter.test_items.test_result import TestResult, TestValue
import api.testium as tm
from interpreter.utils.params import TestItemParams
from interpreter.utils.param_decl import (
Param, ParamSet, LIST, BLOCK, unknown_keys, missing_required,
)
from interpreter.utils.constants import TestItemType as cst_type
from interpreter.utils.eval import eval_to_boolean, evaluate, post_evaluate
from runtime.tum_except import ETUMSyntaxError, item_load_context
@@ -13,6 +16,32 @@ LOG_TEST_STOP = '<----- step "{}" finished'
LOG_TEST_START = '-----> step "{}" started'
# Parameters accepted by every test item, regardless of its type. Subclasses
# concatenate their own ``PARAMS`` to this set; the merged result drives
# unknown-param warnings and (later) the LSP schema export.
COMMON_PARAMS = ParamSet(
Param("name", doc="Display name shown in the GUI tree and reports."),
Param("doc", doc="Free-form documentation; surfaced in tooltips."),
Param("skipped", doc="If truthy, the step is skipped (static expression, "
"evaluated at load time)."),
Param("key", doc="Report key used to classify the result "
"(typically <test>_PASS or <test>_FAIL)."),
Param("stop_on_failure", doc="If true, abort the surrounding container on failure."),
Param("execute_on_stop", doc="If true, run this step even when its container "
"is being stopped (cleanup)."),
Param("process_result", doc="Post-evaluation expression applied to the test result."),
Param("store_result", doc="Global-dict key in which to store the test result."),
Param("expected_result", doc="Expected outcome; the step is failed if it doesn't match."),
Param("no_fail", doc="If truthy, never report a FAILURE for this step."),
Param("report", doc="Per-step reporting override."),
Param("condition", doc="Optional gating expression evaluated before each "
"run; false ⇒ the step is skipped."),
Param("steps", kind=LIST, doc="Children (for container items)."),
Param("seq_filename", doc="(internal) source .tum file of this step; injected "
"by the loader."),
)
class TestItem:
pass
@@ -97,6 +126,11 @@ def test_data(item: TestItem, child: dict) -> dict:
class TestItem:
# Subclasses override with their own ParamSet to opt into the declarative
# validation. While ``PARAMS`` is empty / unset, the base class skips the
# unknown-param check for this item type — keeps the migration incremental.
PARAMS = None
def __init__(
self, dict_item: dict = None, parent: TestItem = None, status_queue=None, filename = ""
):
@@ -134,6 +168,13 @@ class TestItem:
# creation of the params object
self._prms = TestItemParams(dict_item, parent)
# Declarative-params validation. Only kicks in when the concrete
# subclass declares ``PARAMS`` — items not yet migrated stay
# silent. Warnings (not errors) during the migration window so
# existing .tum files don't break suddenly; will be flipped to
# errors once every item has migrated.
self._validate_declared_params(dict_item)
# getting parameters for the test item
try:
self._name = self._prms.getParam("name", default="", processed=True)
@@ -190,6 +231,36 @@ class TestItem:
self.result = TestResult(self, TestValue.FAILURE, "Failure by default")
def _validate_declared_params(self, dict_item):
"""Warn on unknown / missing-required params, if PARAMS is declared.
The check is opt-in per subclass: it only runs when the concrete
class sets a non-empty ``PARAMS`` attribute. Items not yet migrated
produce no diagnostics — preserving the historical "silently accept
anything" behavior until they get their declaration.
"""
if not self.PARAMS:
return
# ``self._type`` is the parent root type at this point (subclasses set
# it after super().__init__), so use the class name as a stable label
# in diagnostics. ``self._name`` was preset to the type name by every
# subclass before super() ran, which gives a useful prefix.
label = f"{type(self).__name__} '{self._name}'"
declared = COMMON_PARAMS + self.PARAMS
unknown = unknown_keys(declared, dict_item)
if unknown:
accepted = ", ".join(sorted(declared.names()))
for k in unknown:
tm.print_warn(
f"{label}: unknown parameter '{k}'. Accepted: {accepted}."
)
missing = missing_required(declared, dict_item)
for k in missing:
raise ETUMSyntaxError(
f"{label}: required parameter '{k}' is missing.",
self._seq_filename,
)
def _filter_dict_item(self, dict_item):
# Stores the content of the step to be displayed
# in the GUI

View File

@@ -5,11 +5,22 @@ from runtime.tum_except import ETUMSyntaxError, item_load_context
import api.testium as tm
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import evaluate
from interpreter.utils.param_decl import Param, ParamSet, LIST
class TestItemCheckValue(TestItem):
"""check item usage.
check usage:{check: {name: check my func output, steps: ['$(pfn_echo) < 5']}}
"""
PARAMS = ParamSet(
Param("values", kind=LIST, required=True,
doc="List of expressions to evaluate. Each is expanded then "
"evaluated; non-truthy results fail the check."),
# 'steps' is intentionally not redeclared here — it's the deprecated
# alias of 'values' and is already accepted by COMMON_PARAMS for
# container items. A runtime warning is emitted when 'steps' is used.
)
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_CHECK.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -2,11 +2,25 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, BLOCK
from runtime.tum_except import item_load_context
import api.testium as tm
class TestItemChoicesDialog(TestItemDialogBase):
PARAMS = ParamSet(
Param("question", required=True,
doc="Prompt shown above the list of choices."),
Param("choices", kind=BLOCK, required=True,
doc="Tree of choices: either a list of strings, or a nested "
"mapping {label: subchoices, ...} to build a multi-level menu."),
Param("icon", default=None,
doc="Default icon name shown next to each choice."),
Param("auto_result", default=None,
doc="Batch-mode selection (path or label). None ⇒ FAILURE."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_CHOICES_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -10,6 +10,7 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.item_actions import TestItemActions
from interpreter.test_items.item_actions.action import TestItemAction
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from interpreter.test_items.test_result import TestResult, TestValue
@@ -21,6 +22,38 @@ class TestItemConsoleAction(TestItemAction):
class TestItemConsoleOpen(TestItemConsoleAction):
PARAMS = ParamSet(
Param("protocol", required=True,
doc="Transport: 'telnet', 'ssh', 'rawtcp', 'serial' or 'terminal'."),
Param("write_delay", default=0,
doc="Inter-character write delay in ms (slow devices)."),
Param("log", doc="Path to a log file capturing the console traffic."),
Param("overwrite_log", default=True,
doc="If true, truncate the log file at open; else append."),
# telnet
Param("telnet_host", doc="Hostname/IP for the telnet target."),
Param("telnet_port", default=69, doc="TCP port for telnet."),
# ssh
Param("ssh_host", doc="Hostname/IP for the SSH target."),
Param("ssh_user", doc="SSH login user."),
Param("ssh_pwd", doc="SSH password (if key-based auth is not used)."),
# rawtcp
Param("tcp_host", doc="Hostname/IP for a raw-TCP connection."),
Param("tcp_port", doc="TCP port for a raw-TCP connection."),
# serial
Param("serial_port", doc="Serial device path (e.g. /dev/ttyUSB0 or COM3)."),
Param("serial_baudrate", doc="Serial baudrate."),
Param("buffered", default=True,
doc="If true, the serial console buffers received bytes between reads."),
# terminal
Param("terminal_path",
doc="Working directory for the local terminal protocol."),
Param("shell",
doc="Shell command used for the local terminal protocol "
"(default: 'cmd.exe' on Windows, '/usr/bin/env bash' elsewhere)."),
)
def __init__(
self, action_name, dict_item, parent=None, status_queue=None, filename=""
):
@@ -283,6 +316,17 @@ class TestItemConsoleWriteLn(TestItemConsoleAction):
class TestItemConsoleReadUntil(TestItemConsoleAction):
PARAMS = ParamSet(
Param("expected", required=True,
doc="Regex matched against incoming console output until found "
"or until timeout."),
Param("timeout", default=-1,
doc="Seconds before giving up. Negative means infinite."),
Param("mute", default=False,
doc="If true, don't echo received bytes to testium's stdout/log."),
)
def __init__(
self, action_name, dict_item, parent=None, status_queue=None, filename=""
):
@@ -336,6 +380,14 @@ class TestItemConsoleReadUntil(TestItemConsoleAction):
class TestItemConsole(TestItemActions):
PARAMS = ParamSet(
Param("console_name", required=True,
doc="Identifier of the console — used by every nested action to "
"reach back the same transport. Multiple consoles can coexist "
"as long as their names differ."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
super().__init__(
cst.TYPE_CONSOLE, dict_item, parent, status_queue, filename=filename

View File

@@ -8,9 +8,36 @@ from interpreter.test_items.test_result import TestResult, TestValue
import api.testium as tm
from interpreter.utils.params import TestItemParams
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, BLOCK
# Sub-block validation: 'cycle' accepts an 'exit_condition:' mapping whose
# own params are reported here so unknown keys inside it can be flagged
# during a future Block-aware diagnostic pass. For now the parent only
# declares that 'exit_condition' is an accepted top-level key.
EXIT_CONDITION_PARAMS = ParamSet(
Param("time", doc="HH:MM time of day after which the loop exits."),
Param("value", doc="Expression; when truthy the loop exits."),
Param("file", doc="Python file containing the exit-condition function."),
Param("func_name", doc="Function name in 'file' returning the exit value."),
Param("param", doc="Arguments passed to the exit function."),
Param("eval", default="",
doc="Post-evaluation expression applied to the function's return."),
)
class TestItemCycle(TestItem):
PARAMS = ParamSet(
Param("iterator",
doc="Iterable (or string expanding to one) driving the loop. "
"The current value is exposed as $(loop_param)."),
Param("exit_condition", kind=BLOCK,
doc="Optional block stopping the loop early: combine 'time', "
"'value', or a 'file'+'func_name' pair (with optional "
"'param' and 'eval')."),
)
def __init__(self, dict_cycle, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_CYCLE.item_name
super().__init__(dict_cycle, parent, status_queue, filename=filename)

View File

@@ -1,6 +1,7 @@
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, LIST
from runtime.tum_except import ETUMParamError, ETUMSyntaxError
import interpreter.utils.version as git
@@ -8,6 +9,13 @@ class TestItemGit(TestItem):
"""
This item expect only one parameter which is a string or list of string being the path to the git folder
"""
PARAMS = ParamSet(
Param("repo", kind=LIST, required=True,
doc="Path to a git checkout, or list of such paths. Each is "
"reported with its current version (tag + dirty state)."),
)
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_GIT.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -1,10 +1,17 @@
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import ParamSet
from runtime.tum_except import ETUMSyntaxError
import api.testium as tm
class TestItemGroup(TestItem):
# 'group' has no item-specific parameters; 'steps' is handled by COMMON_PARAMS.
# Declaring an empty ParamSet still opts in to unknown-param validation
# (e.g. typo 'stop_on_failures').
PARAMS = ParamSet()
def __init__(self, dict_cycle, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_GROUP.item_name
super().__init__(dict_cycle, parent, status_queue, filename=filename)

View File

@@ -4,6 +4,7 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from runtime.tum_except import item_load_context
import api.testium as tm
@@ -12,6 +13,17 @@ class TestItemImageDialog(TestItemDialogBase):
"""dialog_image item usage.
dialog_image name: Nice image, question: could you press the red button, filename: img.jpg
"""
PARAMS = ParamSet(
Param("question", required=True,
doc="Prompt shown above the image."),
Param("filename", required=True,
doc="Path to the image file (relative to the test directory or absolute)."),
Param("auto_result", default=None,
doc="Outcome used in batch/non-interactive mode. Truthy ⇒ SUCCESS, "
"None ⇒ FAILURE."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_IMAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -11,6 +11,7 @@ from interpreter.test_items.item_actions.action import TestItemAction
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import evaluate
from interpreter.utils.param_decl import Param, ParamSet, BLOCK
from interpreter.test_items.test_item_json_rpc.jsonrpc_adapters import (
JrpcAdapter,
@@ -76,6 +77,20 @@ class TestItemJSRPCActionClose(TestItemAction):
class TestItemJSRPCActionQuery(TestItemAction):
PARAMS = ParamSet(
Param("method", required=True,
doc="JSON-RPC method name to call."),
Param("params",
doc="Parameters payload (list, dict or scalar) sent to the method."),
Param("id", default="rand",
doc="JSON-RPC request id. 'rand' (default) ⇒ a random integer is used."),
Param("no_wait", default=False,
doc="If true, send the request without waiting for a response."),
Param("timeout", default=None,
doc="Seconds to wait for a response. None ⇒ inherits the transport "
"default."),
)
def __init__(
self, action_name, dict_item, parent=None, status_queue=None, filename=""
):
@@ -129,6 +144,13 @@ class TestItemJSRPCActionQuery(TestItemAction):
class TestItemJSRPCActionReceive(TestItemAction):
PARAMS = ParamSet(
Param("id", required=True,
doc="JSON-RPC request id whose response we expect."),
Param("timeout", default=None,
doc="Seconds to wait for the response. None ⇒ transport default."),
)
def __init__(
self, action_name, dict_item, parent=None, status_queue=None, filename=""
):
@@ -172,6 +194,22 @@ class TestItemJSON_RPC(TestItemActions):
This item TBD
"""
PARAMS = ParamSet(
Param("console", kind=BLOCK,
doc="Console-transport block: {console_name, …}. Either 'console' "
"or 'udp' must be set."),
Param("udp", kind=BLOCK,
doc="UDP-transport block: {host, port, …}. Either 'console' or "
"'udp' must be set."),
Param("version", default="1.0",
doc="JSON-RPC protocol version ('1.0' or '2.0')."),
Param("timeout", required=True,
doc="Default seconds to wait for a JSON-RPC response across all "
"child query/receive actions."),
Param("mute", default=False,
doc="If true, don't echo wire traffic to the log."),
)
def __init__(
self, dict_item: dict, parent: TestItem = None, status_queue=None, filename=""
):

View File

@@ -8,12 +8,20 @@ from interpreter.test_items.test_result import (TestResult, TestValue)
from runtime.tum_except import ETUMSyntaxError, item_load_context
import api.testium as tm
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, LIST
class TestItemLet(TestItem):
"""let item usage.
let values: {variable1: a, variable2: /dev/ttyUSB0, variable3: 115200}
let eval: {conditional_exec: "random.randint(1, 4)"}
"""
PARAMS = ParamSet(
Param("values", kind=LIST, required=True,
doc="Mapping (or list of single-pair mappings) of global-dict "
"key → value to set. Values are expanded at execution time."),
)
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_LET.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -11,6 +11,7 @@ import api.testium as tm
from interpreter.utils.lua_func_exec import LuaFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, LIST
_LUA_FUNC_CONTEXTS_KEY = "_lua_func_contexts"
@@ -21,6 +22,21 @@ class TestItemLuaFunc(TestItem):
Optional: context_id: <id> — share a persistent process with other lua_func items using the same id.
"""
PARAMS = ParamSet(
Param("file", required=True,
doc="Path to the .lua file containing the function."),
Param("func_name", required=True,
doc="Name of the function to call in the file."),
Param("param", kind=LIST,
doc="Arguments passed to the function. Each entry is expanded "
"before the call. Special tokens $(loop_param) / $(loop_index) "
"resolve from the surrounding cycle."),
Param("context_id", default=None,
doc="If set, the lua_func subprocess is kept alive and reused by "
"every other lua_func item with the same context_id — enables "
"shared in-memory state between successive calls."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_LUA_FUNCTION.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -5,6 +5,7 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from runtime.tum_except import item_load_context
@@ -12,6 +13,15 @@ class TestItemMsgDialog(TestItemDialogBase):
"""dialog_message item usage.
dialog_message name: Nice message, question: Open the door and press OK
"""
PARAMS = ParamSet(
Param("question", required=True,
doc="Message body shown to the user. Multi-line strings are supported."),
Param("auto_result", default=None,
doc="Outcome used in batch/non-interactive mode instead of waiting "
"for the user. Truthy ⇒ SUCCESS, None ⇒ FAILURE."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_MESSAGE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -2,11 +2,23 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from runtime.tum_except import item_load_context
import api.testium as tm
class TestItemNoteDialog(TestItemDialogBase):
PARAMS = ParamSet(
Param("question", required=True,
doc="Prompt shown above the note input field."),
Param("auto_result", default=None,
doc="Batch-mode outcome: None ⇒ FAILURE, 'cancel' ⇒ cancelled, "
"any other truthy ⇒ SUCCESS with auto_value."),
Param("auto_value", default=None,
doc="Note text used in batch mode when auto_result is set."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_NOTE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -6,6 +6,7 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestResult, TestValue
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import eval_to_boolean
from interpreter.utils.param_decl import Param, ParamSet, LIST, BLOCK, Enum
from runtime.tum_except import ETUMSyntaxError
from runtime.string_queue import StringQueue
from runtime.stdout_redirect import stdio_redir
@@ -15,6 +16,12 @@ class TestItemParallelBranch(TestItemContainer):
"""One branch of a parallel item. Runs its children sequentially,
optionally waiting for a condition before starting."""
PARAMS = ParamSet(
Param("wait_for", kind=BLOCK,
doc="Optional block {condition, timeout} that defers the branch "
"start until the condition is truthy (or the timeout elapses)."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
super().__init__(cst.TYPE_PARALLEL_BRANCH, dict_item, parent, status_queue, filename=filename)
self._wait_condition = None
@@ -87,6 +94,15 @@ class TestItemParallel(TestItemContainer):
- ...
"""
PARAMS = ParamSet(
Param("branches", kind=LIST, required=True,
doc="List of branch blocks (each branch holds its own 'steps' "
"and optional 'wait_for')."),
Param("sync", kind=Enum("all", "any"), default="all",
doc="'all' (default) waits for every branch; 'any' returns as "
"soon as the first branch completes."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
branches = dict_item.get("branches", [])
if not branches:

View File

@@ -11,6 +11,7 @@ import api.testium as tm
from interpreter.utils.py_func_exec import PyFuncExecEngine
from interpreter.utils.api_srv import api_request
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, LIST
_PY_FUNC_CONTEXTS_KEY = "_py_func_contexts"
@@ -21,6 +22,21 @@ class TestItemPyFunc(TestItem):
Optional: context_id: <id> — share a persistent process with other py_func items using the same id.
"""
PARAMS = ParamSet(
Param("file", required=True,
doc="Path to the .py file containing the function."),
Param("func_name", required=True,
doc="Name of the function to call in the file."),
Param("param", kind=LIST,
doc="Arguments passed to the function. Each entry is expanded "
"before the call. Special tokens $(loop_param) / $(loop_index) "
"resolve from the surrounding cycle."),
Param("context_id", default=None,
doc="If set, the py_func subprocess is kept alive and reused by "
"every other py_func item with the same context_id — enables "
"shared in-memory state between successive calls."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_PY_FUNCTION.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -2,6 +2,7 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from runtime.tum_except import item_load_context
@@ -9,6 +10,14 @@ class TestItemQuestionDialog(TestItemDialogBase):
"""dialog_question item usage.
dialog_question name: Nice question, question: "If OK, press OK, If not, press cancel"
"""
PARAMS = ParamSet(
Param("question", required=True,
doc="Yes/No prompt presented to the user."),
Param("auto_result", default=None,
doc="Batch-mode answer ('yes'/'no' or truthy/falsy). None ⇒ FAILURE."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_QUESTION_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -3,9 +3,17 @@ from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
from runtime.tum_except import ETUMSyntaxError
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, LIST
from interpreter.test_report.test_report import Export
class TestItemReport(TestItem):
PARAMS = ParamSet(
Param("export", kind=LIST, required=True,
doc="List of exporters to run (junit, sqlite, …). Each entry is a "
"mapping describing the exporter type and its parameters."),
)
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_REPORT.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -10,6 +10,7 @@ from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
import api.testium as tm
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from runtime.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
@@ -57,6 +58,25 @@ def nowInBetween(start, end):
class TestItemRun(TestItem):
PARAMS = ParamSet(
Param("tum", required=True,
doc="Path to the .tum file launched in a fresh testium instance."),
Param("param_file", default="",
doc="Optional path to a param.yaml passed to the sub-instance."),
Param("log_file", default="",
doc="Path where the sub-instance writes its log."),
Param("report_file", default="",
doc="Path where the sub-instance writes its report."),
Param("start_time",
doc="HH:MM time of day after which the sub-instance may run."),
Param("end_time",
doc="HH:MM time of day after which the sub-instance no longer runs."),
Param("wait_for_exec",
doc="If true, block until the time window opens. Requires both "
"start_time and end_time."),
)
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_RUN.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -11,6 +11,7 @@ from interpreter.test_items.item_actions import TestItemActions
from interpreter.test_items.item_actions.action import TestItemAction
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.eval import evaluate
from interpreter.utils.param_decl import Param, ParamSet, LIST
class TestItemPlotAction(TestItemAction):
@@ -21,6 +22,12 @@ class TestItemPlotAction(TestItemAction):
class TestItemPlotActionOpen(TestItemPlotAction):
PARAMS = ParamSet(
Param("log_path", default=None,
doc="Optional file to which the plot data are appended."),
)
def __init__(
self, action_name, dict_item, parent=None, status_queue=None, filename=""
):
@@ -57,6 +64,15 @@ class TestItemPlotActionOpen(TestItemPlotAction):
class TestItemPlotActionClose(TestItemPlotAction):
PARAMS = ParamSet(
Param("wait_dialog_exit", default=False,
doc="If true, the close action blocks until the user closes the "
"plot window (or timeout)."),
Param("timeout", default=-1,
doc="Seconds to wait when wait_dialog_exit is true. Negative ⇒ infinite."),
)
def __init__(
self, action_name, dict_item, parent=None, status_queue=None, filename=""
):
@@ -96,6 +112,20 @@ class TestItemPlotActionClose(TestItemPlotAction):
class TestItemPlotActionPeriodic(TestItemPlotAction):
PARAMS = ParamSet(
Param("period", required=True,
doc="Seconds between two calls of the periodic function."),
Param("file", required=True,
doc="Path to the .py file holding the periodic function."),
Param("func_name", required=True,
doc="Name of the periodic function."),
Param("param", kind=LIST,
doc="Arguments passed to the periodic function on each call."),
Param("eval", default="",
doc="Post-evaluation applied to the function's return value."),
)
def __init__(
self, action_name, dict_item, parent=None, status_queue=None, filename=""
):
@@ -169,6 +199,13 @@ class TestItemPlotActionAdd(TestItemPlotAction):
class TestItemPlotActionLastValues(TestItemPlotAction):
PARAMS = ParamSet(
Param("name", kind=LIST,
doc="List of plot variable names whose last sample is returned. "
"Result is stored in $(plv_<plot_name>) as a dict."),
)
def __init__(self, action_name, dict_item, parent=None, status_queue=None, filename=""):
super().__init__(
action_name, cst.TYPE_GRAPH_ACTION, dict_item, parent, status_queue, filename=filename
@@ -219,6 +256,13 @@ class TestItemPlotActionExport(TestItemPlotAction):
class TestItemPlot(TestItemActions):
PARAMS = ParamSet(
Param("plot_name", required=True,
doc="Identifier of the plot window — referenced by every nested "
"action and by $(plv_<plot_name>) for last-values output."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
super().__init__(
cst.TYPE_GRAPH, dict_item, parent, status_queue, filename=filename

View File

@@ -7,6 +7,7 @@ import api.testium as tm
from interpreter.test_items.test_item import (TestItem, test_run)
from interpreter.test_items.test_result import (TestValue)
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from runtime.tum_except import ETUMSyntaxError, ETUMRuntimeError, item_load_context
class TestItemSleep(TestItem):
@@ -14,6 +15,15 @@ class TestItemSleep(TestItem):
sleep timeout: 10
"""
PARAMS = ParamSet(
Param("timeout", required=True,
doc="Duration to sleep. Number of seconds, or a string "
"like '1d 2h 30m 15s'."),
Param("dialog", default=False,
doc="If true, show a cancel dialog (GUI mode) or an interactive "
"Ctrl+C-able countdown (text mode)."),
)
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_SLEEP.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -2,11 +2,23 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, LIST
from runtime.tum_except import item_load_context
import api.testium as tm
class TestItemTestedRefsDialog(TestItemDialogBase):
PARAMS = ParamSet(
Param("question", required=True,
doc="Prompt asking the operator to enter the tested references."),
Param("reference", kind=LIST,
doc="Pre-filled list of references shown in the dialog."),
Param("auto_result", default=None,
doc="Batch-mode outcome: None ⇒ FAILURE, truthy ⇒ SUCCESS with "
"the pre-filled references."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_REFERENCE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -11,6 +11,7 @@ from interpreter.test_items.test_item import (TestItem, test_run, LOG_TEST_STOP,
from interpreter.test_items.test_result import (TestResult, TestValue)
from interpreter.test_items.test_item import test_data
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet, LIST
from runtime.stdout_redirect import stdio_redir
class UnittestResult(TextTestResult):
@@ -95,6 +96,15 @@ class TestItemUnittestElement(TestItem):
class TestItemUnittestFile(TestItem):
PARAMS = ParamSet(
Param("test_file", required=True,
doc="Path to the Python unittest file (TestCase subclass)."),
Param("test_method", kind=LIST,
doc="Optional list of method names to restrict the run to. "
"When empty, every test_* method in the file is run."),
)
def __init__(self, dict_item, parent = None, status_queue=None, filename=""):
self._name = cst.TYPE_UNITTEST.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -2,6 +2,7 @@ from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase, _is_text_mode, _is_interactive
from interpreter.utils.constants import TestItemType as cst
from interpreter.utils.param_decl import Param, ParamSet
from runtime.tum_except import item_load_context
import api.testium as tm
@@ -10,6 +11,19 @@ class TestItemValueDialog(TestItemDialogBase):
"""dialog_value item usage.
dialog_value name: Enter value, question: "Which value did you measure?"
"""
PARAMS = ParamSet(
Param("question", required=True,
doc="Prompt shown above the value input field."),
Param("default", default="",
doc="Pre-filled value of the input field."),
Param("auto_result", default=None,
doc="Batch-mode outcome: None ⇒ FAILURE, 'cancel' ⇒ cancelled, "
"any other truthy ⇒ SUCCESS with auto_value."),
Param("auto_value", default=None,
doc="Value used in batch mode when auto_result is set."),
)
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_VALUE_DLG.item_name
super().__init__(dict_item, parent, status_queue, filename=filename)

View File

@@ -0,0 +1,175 @@
"""Declarative description of a test item's accepted parameters.
Each ``TestItem`` subclass declares its parameter surface as a class
attribute::
class TestItemFoo(TestItem):
PARAMS = ParamSet(
Param("bar", required=True, doc="The bar value."),
Param("baz", default=0, doc="Optional baz."),
Param("modes", kind=LIST, doc="Iterable of modes."),
Param("strategy", kind=ENUM("a", "b"), doc="..."),
Param("opts", kind=BLOCK, doc="Sub-block."),
)
The base ``TestItem.__init__`` consumes both ``COMMON_PARAMS`` (defined
in ``test_item.py``) and the subclass ``PARAMS`` to:
* warn on any key in the user's YAML that isn't declared anywhere
(catches typos like ``param_filee``);
* expose a machine-readable schema for documentation generation and,
eventually, an LSP server.
The descriptor is **purely about shape and naming**. Type coercion and
runtime checking of expanded values remain the responsibility of each
item's ``execute()`` method — most parameters are expressions
(``$(...)`` / ``<| ... |>``) whose effective type is only known after
expansion, so a static type would be misleading.
Validation of *values* (e.g. ``start_time`` must match HH:MM) can be
attached per-param via ``validate=lambda v: ...`` and is applied at
execution time on the expanded value, not at load time.
"""
from dataclasses import dataclass, field
from typing import Any, Callable, Optional, Union
# ---------- Parameter "kinds" -------------------------------------------------
#
# These describe the YAML *shape* expected for a parameter, not its
# semantic type. They drive the LSP completion (do we suggest a single
# value, a list, a sub-block, an enum picker?) and the unknown-param
# diagnostic; nothing more.
SCALAR = "scalar" # single value (string, number, bool, expression, ...)
LIST = "list" # YAML list — the historical ``getParamAll`` case
BLOCK = "block" # nested dict — e.g. ``cycle.exit:``
@dataclass(frozen=True)
class Enum:
"""Closed enumeration of acceptable scalar values."""
values: tuple
def __init__(self, *values):
# frozen=True forbids assignment; bypass via object.__setattr__.
object.__setattr__(self, "values", tuple(values))
def __repr__(self):
return f"Enum({', '.join(repr(v) for v in self.values)})"
Kind = Union[str, Enum]
# ---------- The descriptor ----------------------------------------------------
_MISSING = object()
@dataclass(frozen=True)
class Param:
"""Declarative description of one accepted parameter.
Attributes
----------
name : str
The YAML key.
kind : ``SCALAR`` (default) | ``LIST`` | ``BLOCK`` | ``Enum(...)``
The YAML shape expected.
required : bool
If True, missing the parameter is a load-time error.
default : Any
Default value when the parameter is absent. ``_MISSING`` when no
default was set (used to distinguish "absent" from "None").
doc : str
Free-form description used for hover / generated documentation.
validate : Optional[Callable[[Any], bool]]
Optional post-expansion validator, evaluated at ``execute()``
time on the effective (expanded) value. Returning ``False``
raises a clear error pointing at the param.
"""
name: str
kind: Kind = SCALAR
required: bool = False
default: Any = _MISSING
doc: str = ""
validate: Optional[Callable[[Any], bool]] = None
def has_default(self):
return self.default is not _MISSING
def to_schema(self):
"""Return a dict suitable for JSON Schema generation."""
s = {"name": self.name, "required": self.required, "doc": self.doc}
if isinstance(self.kind, Enum):
s["kind"] = "enum"
s["enum"] = list(self.kind.values)
else:
s["kind"] = self.kind
if self.has_default():
s["default"] = self.default
return s
class ParamSet:
"""Ordered, name-indexed collection of ``Param`` descriptors.
Supports concatenation (``COMMON_PARAMS + SUBCLASS_PARAMS``) to
merge the common surface with each item's own params. Later
declarations override earlier ones (so a subclass can tighten a
common param's docstring without redeclaring everything).
"""
def __init__(self, *params):
self._params = {}
for p in params:
self.add(p)
def add(self, param):
if not isinstance(param, Param):
raise TypeError(f"ParamSet only accepts Param instances, got {type(param).__name__}")
self._params[param.name] = param
def __iter__(self):
return iter(self._params.values())
def __contains__(self, name):
return name in self._params
def __getitem__(self, name):
return self._params[name]
def names(self):
return tuple(self._params.keys())
def __add__(self, other):
if not isinstance(other, ParamSet):
return NotImplemented
merged = ParamSet()
merged._params = {**self._params, **other._params}
return merged
def to_schema(self):
return [p.to_schema() for p in self._params.values()]
# ---------- Validation primitives --------------------------------------------
def unknown_keys(declared, user_dict):
"""Return the user-provided keys that are not declared in *declared*.
*declared* is a ``ParamSet``; *user_dict* is the raw YAML mapping
for the item. Unknown keys catch typos and obsolete parameters.
"""
if not isinstance(user_dict, dict):
return ()
return tuple(k for k in user_dict.keys() if k not in declared)
def missing_required(declared, user_dict):
"""Return the names of declared required params absent from *user_dict*."""
if not isinstance(user_dict, dict):
return tuple(p.name for p in declared if p.required)
return tuple(p.name for p in declared if p.required and p.name not in user_dict)