forked from cleveragents/cleveragents-core
freemo
a5361a93ae
Implement the TuiMaterializer — the glue layer between the Output Rendering Framework and the Textual widget tree. It maps ElementHandle events from A2A task updates to Textual widgets. Key design: - TuiMaterializer implements the MaterializationStrategy protocol - Widget registry maps element kinds to Textual widget factories: panel → RichLog, table → DataTable, progress/status → Static - Per-session isolation: each tab gets its own materializer instance with independent widget map - Streaming via on_element_updated pushes incremental deltas directly into mounted widgets - Same widget tree serves TUI, Web (Textual Web), and IDE plugin - Callbacks (on_widget_created, on_widget_removed) let the host screen mount/unmount widgets into the DOM - Thread-safe via internal lock on widget/snapshot maps New files: - src/cleveragents/tui/materializer.py — TuiMaterializer class - features/tui_materializer.feature — 20 BDD scenarios - features/steps/tui_materializer_steps.py — Behave step definitions - robot/tui_materializer.robot — 12 Robot Framework integration tests - robot/helper_tui_materializer.py — Robot helper script Quality checks: typecheck (0 errors), lint (clean), format (clean), unit_tests (20/20 pass), integration_tests (12/12 pass) ISSUES CLOSED: #696
438 lines
13 KiB
Python
438 lines
13 KiB
Python
"""Step definitions for the TuiMaterializer feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.cli.output.handles import (
|
|
ColumnDef,
|
|
ElementClosed,
|
|
ElementCreated,
|
|
ElementUpdated,
|
|
Panel,
|
|
PanelEntry,
|
|
ProgressIndicator,
|
|
SessionEnd,
|
|
StatusMessage,
|
|
Table,
|
|
)
|
|
from cleveragents.cli.output.session import OutputSession
|
|
from cleveragents.tui.materializer import TuiMaterializer
|
|
|
|
# ---- helpers ----
|
|
|
|
_COUNTER = 0
|
|
|
|
|
|
def _next_handle_id() -> str:
|
|
global _COUNTER
|
|
_COUNTER += 1
|
|
return f"test-hdl-{_COUNTER:04d}"
|
|
|
|
|
|
# ---- Instantiation ----
|
|
|
|
|
|
@given('a TuiMaterializer with session id "{sid}"')
|
|
def step_given_materializer(context: Context, sid: str) -> None:
|
|
context.materializer = TuiMaterializer(session_id=sid)
|
|
|
|
|
|
@when("a TuiMaterializer is created with a non-string session id")
|
|
def step_when_non_string_sid(context: Context) -> None:
|
|
context.creation_error = None
|
|
try:
|
|
TuiMaterializer(session_id=42) # type: ignore[arg-type]
|
|
except TypeError as exc:
|
|
context.creation_error = exc
|
|
|
|
|
|
@then("a TypeError should be raised for materializer session id")
|
|
def step_then_type_error_sid(context: Context) -> None:
|
|
assert context.creation_error is not None
|
|
assert isinstance(context.creation_error, TypeError)
|
|
|
|
|
|
@then('the materializer session id should be "{sid}"')
|
|
def step_then_session_id(context: Context, sid: str) -> None:
|
|
assert context.materializer.session_id == sid
|
|
|
|
|
|
@then('the materializer strategy name should be "{name}"')
|
|
def step_then_strategy_name(context: Context, name: str) -> None:
|
|
assert context.materializer.strategy_name == name
|
|
|
|
|
|
@then("the materializer should not be closed")
|
|
def step_then_not_closed(context: Context) -> None:
|
|
assert not context.materializer.is_closed
|
|
|
|
|
|
@then("the materializer widget count should be {n:d}")
|
|
def step_then_widget_count(context: Context, n: int) -> None:
|
|
assert context.materializer.widget_count == n
|
|
|
|
|
|
# ---- Registry ----
|
|
|
|
|
|
@then('the materializer registry should contain kind "{kind}"')
|
|
def step_then_registry_contains(context: Context, kind: str) -> None:
|
|
assert kind in context.materializer._registry
|
|
|
|
|
|
@when('a custom widget factory is registered for kind "{kind}"')
|
|
def step_when_register_custom(context: Context, kind: str) -> None:
|
|
from textual.widgets import Static
|
|
|
|
def factory(hid: str, snap: Any) -> Static:
|
|
return Static("custom", id=f"el-{hid}")
|
|
|
|
context.materializer.register_widget_factory(kind, factory)
|
|
|
|
|
|
@when("an empty element kind is registered on the materializer")
|
|
def step_when_empty_kind(context: Context) -> None:
|
|
context.registry_error = None
|
|
try:
|
|
context.materializer.register_widget_factory("", lambda h, s: None)
|
|
except ValueError as exc:
|
|
context.registry_error = exc
|
|
|
|
|
|
@then("a ValueError should be raised for materializer empty kind")
|
|
def step_then_value_error_kind(context: Context) -> None:
|
|
assert context.registry_error is not None
|
|
assert isinstance(context.registry_error, ValueError)
|
|
|
|
|
|
@when("a non-callable factory is registered on the materializer")
|
|
def step_when_non_callable(context: Context) -> None:
|
|
context.factory_error = None
|
|
try:
|
|
context.materializer.register_widget_factory("x", "not_callable") # type: ignore[arg-type]
|
|
except TypeError as exc:
|
|
context.factory_error = exc
|
|
|
|
|
|
@then("a TypeError should be raised for materializer non-callable")
|
|
def step_then_type_error_factory(context: Context) -> None:
|
|
assert context.factory_error is not None
|
|
assert isinstance(context.factory_error, TypeError)
|
|
|
|
|
|
# ---- Element creation ----
|
|
|
|
|
|
@when('a panel element is created on the materializer with title "{title}"')
|
|
def step_when_panel_created(context: Context, title: str) -> None:
|
|
hid = _next_handle_id()
|
|
context.last_handle_id = hid
|
|
panel = Panel(title=title)
|
|
event = ElementCreated(
|
|
event_type="created",
|
|
handle_id=hid,
|
|
element_kind="panel",
|
|
declaration_index=0,
|
|
initial_state=panel,
|
|
)
|
|
context.materializer.on_element_created(event)
|
|
|
|
|
|
@when('a table element is created on the materializer with columns "{c1}" and "{c2}"')
|
|
def step_when_table_created(context: Context, c1: str, c2: str) -> None:
|
|
hid = _next_handle_id()
|
|
context.last_handle_id = hid
|
|
table = Table(
|
|
title=None,
|
|
columns=[ColumnDef(name=c1), ColumnDef(name=c2)],
|
|
)
|
|
event = ElementCreated(
|
|
event_type="created",
|
|
handle_id=hid,
|
|
element_kind="table",
|
|
declaration_index=0,
|
|
initial_state=table,
|
|
)
|
|
context.materializer.on_element_created(event)
|
|
|
|
|
|
@when('a progress element is created on the materializer with label "{label}"')
|
|
def step_when_progress_created(context: Context, label: str) -> None:
|
|
hid = _next_handle_id()
|
|
context.last_handle_id = hid
|
|
progress = ProgressIndicator(label=label)
|
|
event = ElementCreated(
|
|
event_type="created",
|
|
handle_id=hid,
|
|
element_kind="progress",
|
|
declaration_index=0,
|
|
initial_state=progress,
|
|
)
|
|
context.materializer.on_element_created(event)
|
|
|
|
|
|
@when('a status element is created on the materializer with message "{msg}"')
|
|
def step_when_status_created(context: Context, msg: str) -> None:
|
|
hid = _next_handle_id()
|
|
context.last_handle_id = hid
|
|
status = StatusMessage(message=msg)
|
|
event = ElementCreated(
|
|
event_type="created",
|
|
handle_id=hid,
|
|
element_kind="status",
|
|
declaration_index=0,
|
|
initial_state=status,
|
|
)
|
|
context.materializer.on_element_created(event)
|
|
|
|
|
|
@when('an element with unknown kind "{kind}" is created on the materializer')
|
|
def step_when_unknown_kind(context: Context, kind: str) -> None:
|
|
hid = _next_handle_id()
|
|
event = ElementCreated(
|
|
event_type="created",
|
|
handle_id=hid,
|
|
element_kind=kind,
|
|
declaration_index=0,
|
|
initial_state=StatusMessage(message="x"),
|
|
)
|
|
context.materializer.on_element_created(event)
|
|
|
|
|
|
@when("an element is created on the materializer without initial state")
|
|
def step_when_no_initial_state(context: Context) -> None:
|
|
hid = _next_handle_id()
|
|
event = ElementCreated(
|
|
event_type="created",
|
|
handle_id=hid,
|
|
element_kind="panel",
|
|
declaration_index=0,
|
|
initial_state=None,
|
|
)
|
|
context.materializer.on_element_created(event)
|
|
|
|
|
|
@then("a materializer widget should exist for the last handle")
|
|
def step_then_widget_exists(context: Context) -> None:
|
|
assert context.materializer.get_widget(context.last_handle_id) is not None
|
|
|
|
|
|
@then("the materializer widget should be a RichLog")
|
|
def step_then_widget_is_richlog(context: Context) -> None:
|
|
from textual.widgets import RichLog
|
|
|
|
widget = context.materializer.get_widget(context.last_handle_id)
|
|
assert isinstance(widget, RichLog)
|
|
|
|
|
|
@then("the materializer widget should be a DataTable")
|
|
def step_then_widget_is_datatable(context: Context) -> None:
|
|
from textual.widgets import DataTable
|
|
|
|
widget = context.materializer.get_widget(context.last_handle_id)
|
|
assert isinstance(widget, DataTable)
|
|
|
|
|
|
@then("the materializer widget should be a Static widget")
|
|
def step_then_widget_is_static(context: Context) -> None:
|
|
from textual.widgets import Static
|
|
|
|
widget = context.materializer.get_widget(context.last_handle_id)
|
|
assert isinstance(widget, Static)
|
|
|
|
|
|
# ---- Callbacks ----
|
|
|
|
|
|
@given("a TuiMaterializer with a widget-created callback")
|
|
def step_given_with_created_cb(context: Context) -> None:
|
|
context.cb_created_calls = []
|
|
|
|
def on_created(hid: str, w: Any) -> None:
|
|
context.cb_created_calls.append((hid, w))
|
|
|
|
context.materializer = TuiMaterializer(
|
|
session_id="cb-test",
|
|
on_widget_created=on_created,
|
|
)
|
|
|
|
|
|
@then("the materializer widget-created callback should have been called")
|
|
def step_then_created_cb_called(context: Context) -> None:
|
|
assert len(context.cb_created_calls) > 0
|
|
|
|
|
|
@given("a TuiMaterializer with a widget-removed callback")
|
|
def step_given_with_removed_cb(context: Context) -> None:
|
|
context.cb_removed_calls = []
|
|
|
|
def on_removed(hid: str, w: Any) -> None:
|
|
context.cb_removed_calls.append((hid, w))
|
|
|
|
context.materializer = TuiMaterializer(
|
|
session_id="cb-test",
|
|
on_widget_removed=on_removed,
|
|
)
|
|
|
|
|
|
@when("a status element is created on the materializer and then closed")
|
|
def step_when_created_then_closed(context: Context) -> None:
|
|
hid = _next_handle_id()
|
|
context.last_handle_id = hid
|
|
status = StatusMessage(message="temp")
|
|
create_event = ElementCreated(
|
|
event_type="created",
|
|
handle_id=hid,
|
|
element_kind="status",
|
|
declaration_index=0,
|
|
initial_state=status,
|
|
)
|
|
context.materializer.on_element_created(create_event)
|
|
close_event = ElementClosed(
|
|
event_type="closed",
|
|
handle_id=hid,
|
|
element_kind="status",
|
|
final_state=status,
|
|
)
|
|
context.materializer.on_element_closed(close_event)
|
|
|
|
|
|
@then("the materializer widget-removed callback should have been called")
|
|
def step_then_removed_cb_called(context: Context) -> None:
|
|
assert len(context.cb_removed_calls) > 0
|
|
|
|
|
|
# ---- Updates ----
|
|
|
|
|
|
@when('the materializer panel snapshot is updated with entry "{key}" "{value}"')
|
|
def step_when_panel_updated(context: Context, key: str, value: str) -> None:
|
|
hid = context.last_handle_id
|
|
panel = Panel(title="Info", entries=[PanelEntry(key=key, value=value)])
|
|
event = ElementUpdated(
|
|
event_type="updated",
|
|
handle_id=hid,
|
|
element_kind="panel",
|
|
update_type="entry_set",
|
|
delta={"key": key, "value": value},
|
|
element_snapshot=panel,
|
|
)
|
|
context.materializer.on_element_updated(event)
|
|
|
|
|
|
@then('the materializer stored snapshot should have entry "{key}"')
|
|
def step_then_snapshot_has_entry(context: Context, key: str) -> None:
|
|
snap = context.materializer.get_snapshot(context.last_handle_id)
|
|
assert snap is not None
|
|
assert isinstance(snap, Panel)
|
|
keys = [e.key for e in snap.entries]
|
|
assert key in keys
|
|
|
|
|
|
@when('the materializer status snapshot is updated to "{msg}"')
|
|
def step_when_status_updated(context: Context, msg: str) -> None:
|
|
hid = context.last_handle_id
|
|
status = StatusMessage(message=msg)
|
|
event = ElementUpdated(
|
|
event_type="updated",
|
|
handle_id=hid,
|
|
element_kind="status",
|
|
update_type="message_changed",
|
|
delta={"message": msg},
|
|
element_snapshot=status,
|
|
)
|
|
context.materializer.on_element_updated(event)
|
|
|
|
|
|
@then('the materializer stored status snapshot message should be "{msg}"')
|
|
def step_then_status_snapshot(context: Context, msg: str) -> None:
|
|
snap = context.materializer.get_snapshot(context.last_handle_id)
|
|
assert snap is not None
|
|
assert isinstance(snap, StatusMessage)
|
|
assert snap.message == msg
|
|
|
|
|
|
# ---- Session lifecycle ----
|
|
|
|
|
|
@when("a materializer session end event is dispatched")
|
|
def step_when_session_end(context: Context) -> None:
|
|
event = SessionEnd(
|
|
event_type="session_end",
|
|
handle_id="",
|
|
element_kind="session",
|
|
exit_code=0,
|
|
)
|
|
context.materializer.on_session_end(event)
|
|
|
|
|
|
@then("the materializer should be closed")
|
|
def step_then_is_closed(context: Context) -> None:
|
|
assert context.materializer.is_closed
|
|
|
|
|
|
@when("the materializer is explicitly closed")
|
|
def step_when_close(context: Context) -> None:
|
|
context.materializer.close()
|
|
|
|
|
|
# ---- Multi-session ----
|
|
|
|
|
|
@given('two TuiMaterializer instances "{a}" and "{b}"')
|
|
def step_given_two_materializers(context: Context, a: str, b: str) -> None:
|
|
context.materializers = {
|
|
a: TuiMaterializer(session_id=a),
|
|
b: TuiMaterializer(session_id=b),
|
|
}
|
|
|
|
|
|
@when('a panel is created on materializer instance "{name}"')
|
|
def step_when_panel_on_mat(context: Context, name: str) -> None:
|
|
hid = _next_handle_id()
|
|
panel = Panel(title="Multi")
|
|
event = ElementCreated(
|
|
event_type="created",
|
|
handle_id=hid,
|
|
element_kind="panel",
|
|
declaration_index=0,
|
|
initial_state=panel,
|
|
)
|
|
context.materializers[name].on_element_created(event)
|
|
|
|
|
|
@then('materializer instance "{name}" should have {n:d} widget')
|
|
def step_then_mat_widget_count_singular(context: Context, name: str, n: int) -> None:
|
|
assert context.materializers[name].widget_count == n
|
|
|
|
|
|
@then('materializer instance "{name}" should have {n:d} widgets')
|
|
def step_then_mat_widget_count_plural(context: Context, name: str, n: int) -> None:
|
|
assert context.materializers[name].widget_count == n
|
|
|
|
|
|
# ---- OutputSession integration ----
|
|
|
|
|
|
@given("an OutputSession using a TuiMaterializer strategy")
|
|
def step_given_output_session(context: Context) -> None:
|
|
context.materializer = TuiMaterializer(session_id="int-test")
|
|
context.output_session = OutputSession(
|
|
strategy=context.materializer,
|
|
command="test",
|
|
)
|
|
|
|
|
|
@when('a panel is created via the output session with title "{title}"')
|
|
def step_when_session_panel(context: Context, title: str) -> None:
|
|
context.panel_handle = context.output_session.panel(title)
|
|
|
|
|
|
@then("the materializer strategy should have {n:d} widget")
|
|
def step_then_materializer_count(context: Context, n: int) -> None:
|
|
assert context.materializer.widget_count == n
|