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>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
175
src/testium/interpreter/utils/param_decl.py
Normal file
175
src/testium/interpreter/utils/param_decl.py
Normal 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)
|
||||
Reference in New Issue
Block a user