HAL9000
3cb3815944
Implemented the TuiMaterializer class that bridges the Output Rendering Framework to Textual UI widgets, enabling all CLI command producers to render in the TUI without modification. Split the module into three files to stay under the 500-line file limit: main materializer.py (TuiMaterializer class), _tui_events.py (event type constants and event model), and _tui_renderers.py (all rendering helper functions). The TuiMaterializer implements the MaterializationStrategy protocol and maps ElementHandle events (Panel, Table, Status, Progress, Tree, Code, Diff, Separator, ActionHint, Text) to plain-text renderings for TUI display. Supports real-time streaming updates and A2A event routing for PermissionRequest and ThoughtBlock events. Thread-safe with lock guards on all shared state mutations. Added comprehensive Behave BDD test suite covering all element types, callback invocation, rendered output accumulation, A2A routing logic, and concurrent thread safety verification. ISSUES CLOSED: #5326
549 lines
20 KiB
Python
549 lines
20 KiB
Python
"""Step definitions for tui_materializer.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from typing import Any
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the tui materializer module is imported")
|
|
def step_import_tui_materializer(context: Any) -> None:
|
|
from cleveragents.tui.materializer import (
|
|
TuiMaterializer,
|
|
TuiWidgetEvent,
|
|
TuiWidgetEventType,
|
|
render_element_for_tui,
|
|
)
|
|
|
|
context.TuiMaterializer = TuiMaterializer
|
|
context.TuiWidgetEvent = TuiWidgetEvent
|
|
context.TuiWidgetEventType = TuiWidgetEventType
|
|
context.render_element_for_tui = render_element_for_tui
|
|
context.materializer = None
|
|
context.session = None
|
|
context.panel_handle = None
|
|
context.status_handle = None
|
|
context.rendered_text = ""
|
|
context.callback_events: list[TuiWidgetEvent] = []
|
|
context.last_event: TuiWidgetEvent | None = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module exports
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("TuiMaterializer should be importable from tui.materializer")
|
|
def step_check_tui_materializer_importable(context: Any) -> None:
|
|
from cleveragents.tui.materializer import TuiMaterializer
|
|
|
|
assert TuiMaterializer is not None
|
|
|
|
|
|
@then("TuiWidgetEvent should be importable from tui.materializer")
|
|
def step_check_tui_widget_event_importable(context: Any) -> None:
|
|
from cleveragents.tui.materializer import TuiWidgetEvent
|
|
|
|
assert TuiWidgetEvent is not None
|
|
|
|
|
|
@then("TuiWidgetEventType should be importable from tui.materializer")
|
|
def step_check_tui_widget_event_type_importable(context: Any) -> None:
|
|
from cleveragents.tui.materializer import TuiWidgetEventType
|
|
|
|
assert TuiWidgetEventType is not None
|
|
|
|
|
|
@then("render_element_for_tui should be importable from tui.materializer")
|
|
def step_check_render_element_for_tui_importable(context: Any) -> None:
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
assert render_element_for_tui is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TuiMaterializer instantiation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a TuiMaterializer without a callback")
|
|
def step_create_materializer_no_callback(context: Any) -> None:
|
|
from cleveragents.tui.materializer import TuiMaterializer
|
|
|
|
context.materializer = TuiMaterializer()
|
|
|
|
|
|
@when("I create a TuiMaterializer with a callback")
|
|
def step_create_materializer_with_callback(context: Any) -> None:
|
|
from cleveragents.tui.materializer import TuiMaterializer, TuiWidgetEvent
|
|
|
|
context.callback_events = []
|
|
|
|
def _callback(event: TuiWidgetEvent) -> None:
|
|
context.callback_events.append(event)
|
|
|
|
context.materializer = TuiMaterializer(on_event=_callback)
|
|
|
|
|
|
@then('the materializer strategy_name should be "tui"')
|
|
def step_check_strategy_name(context: Any) -> None:
|
|
assert context.materializer.strategy_name == "tui"
|
|
|
|
|
|
@then("the materializer supports_incremental_updates should be True")
|
|
def step_check_supports_incremental(context: Any) -> None:
|
|
assert context.materializer.supports_incremental_updates is True
|
|
|
|
|
|
@then("the materializer events list should be empty")
|
|
def step_check_events_empty(context: Any) -> None:
|
|
assert context.materializer.events == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# the materializer should still be running (alias for strategy checks)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the materializer should still be running")
|
|
def step_materializer_still_running(context: Any) -> None:
|
|
"""The materializer is considered 'running' when it reports a valid
|
|
strategy name and supports incremental updates."""
|
|
assert context.materializer.strategy_name == "tui"
|
|
assert context.materializer.supports_incremental_updates is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# on_session_begin
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I call on_session_begin with a mock session")
|
|
def step_call_on_session_begin(context: Any) -> None:
|
|
context.materializer.on_session_begin(None) # type: ignore[arg-type]
|
|
|
|
|
|
@then("no error should be raised")
|
|
def step_no_error_raised(context: Any) -> None:
|
|
pass # If we got here, no exception was raised
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OutputSession integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I use the materializer with an OutputSession")
|
|
def step_use_with_output_session(context: Any) -> None:
|
|
from cleveragents.cli.output.session import OutputSession
|
|
|
|
context.session = OutputSession(strategy=context.materializer)
|
|
|
|
|
|
@when('I create a panel handle titled "{title}"')
|
|
def step_create_panel_handle(context: Any, title: str) -> None:
|
|
context.panel_handle = context.session.panel(title)
|
|
|
|
|
|
@when('I create a table handle titled "{title}"')
|
|
def step_create_table_handle(context: Any, title: str) -> None:
|
|
context.table_handle = context.session.table(title)
|
|
|
|
|
|
@when('I create a status handle with message "{message}"')
|
|
def step_create_status_handle(context: Any, message: str) -> None:
|
|
context.status_handle = context.session.status(message)
|
|
|
|
|
|
@when('I create a progress handle with label "{label}"')
|
|
def step_create_progress_handle(context: Any, label: str) -> None:
|
|
context.progress_handle = context.session.progress(label, indeterminate=True)
|
|
|
|
|
|
@when('I create a code handle with content "{content}" and language "{language}"')
|
|
def step_create_code_handle(context: Any, content: str, language: str) -> None:
|
|
context.code_handle = context.session.code(content, language=language)
|
|
|
|
|
|
@when("I create a separator handle")
|
|
def step_create_separator_handle(context: Any) -> None:
|
|
context.separator_handle = context.session.separator()
|
|
|
|
|
|
@when('I create an action hint handle with command "{command}"')
|
|
def step_create_action_hint_handle(context: Any, command: str) -> None:
|
|
context.action_hint_handle = context.session.action_hint([command])
|
|
|
|
|
|
@when("I close the panel handle")
|
|
def step_close_panel_handle(context: Any) -> None:
|
|
context.panel_handle.close()
|
|
|
|
|
|
@when("I close the status handle")
|
|
def step_close_status_handle(context: Any) -> None:
|
|
context.status_handle.close()
|
|
|
|
|
|
@when("I close the session")
|
|
def step_close_session(context: Any) -> None:
|
|
context.session.close()
|
|
|
|
|
|
@when('I create and close a status handle with message "{message}"')
|
|
def step_create_and_close_status_handle(context: Any, message: str) -> None:
|
|
handle = context.session.status(message)
|
|
handle.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Event assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("an element_created event should be emitted")
|
|
def step_check_element_created_event(context: Any) -> None:
|
|
events = context.materializer.events
|
|
created_events = [e for e in events if e.event_type == "element_created"]
|
|
assert len(created_events) > 0, f"No element_created events found. Events: {events}"
|
|
context.last_event = created_events[-1]
|
|
|
|
|
|
@then('the event element_kind should be "{kind}"')
|
|
def step_check_event_element_kind(context: Any, kind: str) -> None:
|
|
assert context.last_event is not None
|
|
assert context.last_event.element_kind == kind, (
|
|
f"Expected element_kind={kind!r}, got {context.last_event.element_kind!r}"
|
|
)
|
|
|
|
|
|
@then('the event rendered_text should contain "{text}"')
|
|
def step_check_event_rendered_text_contains(context: Any, text: str) -> None:
|
|
assert context.last_event is not None
|
|
assert text in context.last_event.rendered_text, (
|
|
f"Expected {text!r} in rendered_text={context.last_event.rendered_text!r}"
|
|
)
|
|
|
|
|
|
@then("an element_closed event should be emitted")
|
|
def step_check_element_closed_event(context: Any) -> None:
|
|
events = context.materializer.events
|
|
closed_events = [e for e in events if e.event_type == "element_closed"]
|
|
assert len(closed_events) > 0, f"No element_closed events found. Events: {events}"
|
|
context.last_closed_event = closed_events[-1]
|
|
|
|
|
|
@then('the closed event rendered_text should contain "{text}"')
|
|
def step_check_closed_event_rendered_text(context: Any, text: str) -> None:
|
|
assert context.last_closed_event is not None
|
|
assert text in context.last_closed_event.rendered_text, (
|
|
f"Expected {text!r} in rendered_text={context.last_closed_event.rendered_text!r}"
|
|
)
|
|
|
|
|
|
@then("a session_end event should be emitted")
|
|
def step_check_session_end_event(context: Any) -> None:
|
|
events = context.materializer.events
|
|
end_events = [e for e in events if e.event_type == "session_end"]
|
|
assert len(end_events) > 0, f"No session_end events found. Events: {events}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Callback assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the callback should have been called")
|
|
def step_check_callback_called(context: Any) -> None:
|
|
assert len(context.callback_events) > 0, "Callback was not called"
|
|
|
|
|
|
@then('the callback event type should be "element_created"')
|
|
def step_check_callback_event_type_created(context: Any) -> None:
|
|
created = [e for e in context.callback_events if e.event_type == "element_created"]
|
|
assert len(created) > 0, (
|
|
f"No element_created callback events. Got: {[e.event_type for e in context.callback_events]}"
|
|
)
|
|
|
|
|
|
@then('the callback should have been called with event_type "{event_type}"')
|
|
def step_check_callback_event_type(context: Any, event_type: str) -> None:
|
|
matching = [e for e in context.callback_events if e.event_type == event_type]
|
|
assert len(matching) > 0, (
|
|
f"No {event_type!r} callback events. Got: {[e.event_type for e in context.callback_events]}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# rendered_output property
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the materializer rendered_output should be empty")
|
|
def step_check_rendered_output_empty(context: Any) -> None:
|
|
assert context.materializer.rendered_output == "", (
|
|
f"Expected empty rendered_output, got: {context.materializer.rendered_output!r}"
|
|
)
|
|
|
|
|
|
@then('the materializer rendered_output should contain "{text}"')
|
|
def step_check_rendered_output_contains(context: Any, text: str) -> None:
|
|
output = context.materializer.rendered_output
|
|
assert text in output, f"Expected {text!r} in rendered_output={output!r}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# render_element_for_tui
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I render a Panel element with title "{title}" and entry "{key}" "{value}"')
|
|
def step_render_panel_element(context: Any, title: str, key: str, value: str) -> None:
|
|
from cleveragents.cli.output.handles._models import Panel, PanelEntry
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = Panel(title=title, entries=[PanelEntry(key=key, value=value)])
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when('I render a Table element with title "{title}" and column "{column}"')
|
|
def step_render_table_element(context: Any, title: str, column: str) -> None:
|
|
from cleveragents.cli.output.handles._models import ColumnDef, Table
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = Table(title=title, columns=[ColumnDef(name=column)])
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when('I render a StatusMessage element with level "{level}" and message "{message}"')
|
|
def step_render_status_element(context: Any, level: str, message: str) -> None:
|
|
from cleveragents.cli.output.handles._models import StatusMessage
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = StatusMessage(message=message, level=level)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when('I render an indeterminate ProgressIndicator with label "{label}"')
|
|
def step_render_indeterminate_progress(context: Any, label: str) -> None:
|
|
from cleveragents.cli.output.handles._models import ProgressIndicator
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = ProgressIndicator(label=label, indeterminate=True)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when(
|
|
'I render a determinate ProgressIndicator with label "{label}" current {current:d} total {total:d}'
|
|
)
|
|
def step_render_determinate_progress(
|
|
context: Any, label: str, current: int, total: int
|
|
) -> None:
|
|
from cleveragents.cli.output.handles._models import ProgressIndicator
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = ProgressIndicator(label=label, current=current, total=total)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when('I render a Tree element with root "{root}" and child "{child}"')
|
|
def step_render_tree_element(context: Any, root: str, child: str) -> None:
|
|
from cleveragents.cli.output.handles._models import Tree, TreeNode
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
root_node = TreeNode(label=root, children=[TreeNode(label=child)])
|
|
element = Tree(root=root_node)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when(
|
|
'I render a CodeBlock element with language "{language}" and content "{content}"'
|
|
)
|
|
def step_render_code_element(context: Any, language: str, content: str) -> None:
|
|
from cleveragents.cli.output.handles._models import CodeBlock
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = CodeBlock(content=content, language=language)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when(
|
|
'I render a DiffBlock element with file_a "{file_a}" and file_b "{file_b}"'
|
|
)
|
|
def step_render_diff_element(context: Any, file_a: str, file_b: str) -> None:
|
|
from cleveragents.cli.output.handles._models import DiffBlock
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = DiffBlock(file_a=file_a, file_b=file_b)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when('I render a Separator element with style "{style}"')
|
|
def step_render_separator_element(context: Any, style: str) -> None:
|
|
from cleveragents.cli.output.handles._models import Separator
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = Separator(style=style)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when('I render an ActionHint element with command "{command}"')
|
|
def step_render_action_hint_element(context: Any, command: str) -> None:
|
|
from cleveragents.cli.output.handles._models import ActionHint
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = ActionHint(commands=[command])
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when('I render a TextBlock element with content "{content}"')
|
|
def step_render_text_element(context: Any, content: str) -> None:
|
|
from cleveragents.cli.output.handles._models import TextBlock
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = TextBlock(content=content)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@when('I render a TextBlock element with content "{content}" and indent {indent:d}')
|
|
def step_render_text_element_with_indent(
|
|
context: Any, content: str, indent: int
|
|
) -> None:
|
|
from cleveragents.cli.output.handles._models import TextBlock
|
|
from cleveragents.tui.materializer import render_element_for_tui
|
|
|
|
element = TextBlock(content=content, indent=indent)
|
|
context.rendered_text = render_element_for_tui(element)
|
|
|
|
|
|
@then('the rendered text should contain "{text}"')
|
|
def step_check_rendered_text_contains(context: Any, text: str) -> None:
|
|
assert text in context.rendered_text, (
|
|
f"Expected {text!r} in rendered_text={context.rendered_text!r}"
|
|
)
|
|
|
|
|
|
@then("the rendered text should be empty")
|
|
def step_check_rendered_text_empty(context: Any) -> None:
|
|
assert context.rendered_text == "", (
|
|
f"Expected empty rendered_text, got: {context.rendered_text!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A2A event routing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I route a permission request for "{file_path}" with type "{request_type}"'
|
|
)
|
|
def step_route_permission_request(
|
|
context: Any, file_path: str, request_type: str
|
|
) -> None:
|
|
from cleveragents.domain.models.core.inline_permission_question import (
|
|
InlinePermissionQuestion,
|
|
PermissionRequestType,
|
|
)
|
|
|
|
question = InlinePermissionQuestion(
|
|
file_path=file_path,
|
|
request_type=PermissionRequestType(request_type),
|
|
)
|
|
context.permission_question = question
|
|
context.materializer.route_permission_request(question)
|
|
|
|
|
|
@then("a permission_request event should be emitted")
|
|
def step_check_permission_request_event(context: Any) -> None:
|
|
events = context.materializer.events
|
|
perm_events = [e for e in events if e.event_type == "permission_request"]
|
|
assert len(perm_events) > 0, f"No permission_request events found. Events: {events}"
|
|
context.last_perm_event = perm_events[-1]
|
|
|
|
|
|
@then("the permission event extra should be the InlinePermissionQuestion")
|
|
def step_check_permission_event_extra(context: Any) -> None:
|
|
assert context.last_perm_event.extra is context.permission_question
|
|
|
|
|
|
@then('the permission event rendered_text should contain "{text}"')
|
|
def step_check_permission_event_rendered_text(context: Any, text: str) -> None:
|
|
assert text in context.last_perm_event.rendered_text, (
|
|
f"Expected {text!r} in rendered_text={context.last_perm_event.rendered_text!r}"
|
|
)
|
|
|
|
|
|
@when('I route a thought block with content "{content}"')
|
|
def step_route_thought_block(context: Any, content: str) -> None:
|
|
from cleveragents.domain.models.thought.thought_block import ThoughtBlock
|
|
|
|
thought = ThoughtBlock(content=content)
|
|
context.thought_block = thought
|
|
context.materializer.route_thought_block(thought)
|
|
|
|
|
|
@then("a thought_block event should be emitted")
|
|
def step_check_thought_block_event(context: Any) -> None:
|
|
events = context.materializer.events
|
|
thought_events = [e for e in events if e.event_type == "thought_block"]
|
|
assert len(thought_events) > 0, f"No thought_block events found. Events: {events}"
|
|
context.last_thought_event = thought_events[-1]
|
|
|
|
|
|
@then("the thought event extra should be the ThoughtBlock")
|
|
def step_check_thought_event_extra(context: Any) -> None:
|
|
assert context.last_thought_event.extra is context.thought_block
|
|
|
|
|
|
@then('the thought event rendered_text should contain "{text}"')
|
|
def step_check_thought_event_rendered_text(context: Any, text: str) -> None:
|
|
assert text in context.last_thought_event.rendered_text, (
|
|
f"Expected {text!r} in rendered_text={context.last_thought_event.rendered_text!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thread safety
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create multiple status handles concurrently")
|
|
def step_create_multiple_status_handles_concurrently(context: Any) -> None:
|
|
errors: list[Exception] = []
|
|
|
|
def _create_handle(i: int) -> None:
|
|
try:
|
|
handle = context.session.status(f"Message {i}")
|
|
handle.close()
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
threads = [threading.Thread(target=_create_handle, args=(i,)) for i in range(10)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
context.thread_errors = errors
|
|
|
|
|
|
@then("all events should be recorded without data corruption")
|
|
def step_check_thread_safety(context: Any) -> None:
|
|
assert context.thread_errors == [], (
|
|
f"Thread errors: {context.thread_errors}"
|
|
)
|
|
events = context.materializer.events
|
|
# We should have at least 10 element_created + 10 element_closed events
|
|
created = [e for e in events if e.event_type == "element_created"]
|
|
closed = [e for e in events if e.event_type == "element_closed"]
|
|
assert len(created) >= 10, f"Expected >= 10 created events, got {len(created)}"
|
|
assert len(closed) >= 10, f"Expected >= 10 closed events, got {len(closed)}"
|