Merge pull request #4407 from pipecat-ai/mb/ui-agent-wire-format
feat(rtvi): add UI Agent Protocol as first-class RTVI message types
This commit is contained in:
Mark Backman
297
tests/test_rtvi_ui.py
Normal file
297
tests/test_rtvi_ui.py
Normal file
@@ -0,0 +1,297 @@
|
||||
#
|
||||
# Copyright (c) 2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
"""Smoke tests for the UI Agent Protocol wire format.
|
||||
|
||||
The module under test is data only (constants, payload models, and
|
||||
envelope classes), so the goal is to pin the shapes: any accidental
|
||||
rename or type change to a wire-format field would break compatibility
|
||||
with existing client code, and we want a test that fails loudly.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
|
||||
from pipecat.processors.frameworks.rtvi.frames import (
|
||||
RTVIUICommandFrame,
|
||||
RTVIUIEventFrame,
|
||||
)
|
||||
from pipecat.processors.frameworks.rtvi.models import (
|
||||
A11yNode,
|
||||
A11ySelection,
|
||||
A11ySnapshot,
|
||||
Click,
|
||||
Focus,
|
||||
Highlight,
|
||||
Navigate,
|
||||
ScrollTo,
|
||||
SelectText,
|
||||
SetInputValue,
|
||||
Toast,
|
||||
UICancelTaskData,
|
||||
UICancelTaskMessage,
|
||||
UICommandData,
|
||||
UICommandMessage,
|
||||
UIEventData,
|
||||
UIEventMessage,
|
||||
UISnapshotData,
|
||||
UISnapshotMessage,
|
||||
UITaskCompletedData,
|
||||
UITaskGroupCompletedData,
|
||||
UITaskGroupStartedData,
|
||||
UITaskMessage,
|
||||
UITaskUpdateData,
|
||||
)
|
||||
|
||||
|
||||
class TestEnvelopeMessages(unittest.TestCase):
|
||||
"""Pin the on-the-wire envelope shapes for each first-class UI message."""
|
||||
|
||||
def test_ui_event_envelope(self):
|
||||
msg = UIEventMessage(id="m1", data=UIEventData(event="nav_click", payload={"view": "home"}))
|
||||
self.assertEqual(
|
||||
msg.model_dump(),
|
||||
{
|
||||
"label": "rtvi-ai",
|
||||
"type": "ui-event",
|
||||
"id": "m1",
|
||||
"data": {"event": "nav_click", "payload": {"view": "home"}},
|
||||
},
|
||||
)
|
||||
|
||||
def test_ui_command_envelope_no_id(self):
|
||||
# Server-to-client push: no id field on the envelope (matches
|
||||
# ServerMessage / LLMFunctionCallMessage shape).
|
||||
msg = UICommandMessage(data=UICommandData(command="toast", payload={"title": "Saved"}))
|
||||
self.assertEqual(
|
||||
msg.model_dump(),
|
||||
{
|
||||
"label": "rtvi-ai",
|
||||
"type": "ui-command",
|
||||
"data": {"command": "toast", "payload": {"title": "Saved"}},
|
||||
},
|
||||
)
|
||||
|
||||
def test_ui_snapshot_envelope(self):
|
||||
msg = UISnapshotMessage(
|
||||
id="m2",
|
||||
data=UISnapshotData(
|
||||
tree=A11ySnapshot(
|
||||
root=A11yNode(
|
||||
ref="e1",
|
||||
role="main",
|
||||
children=[A11yNode(ref="e2", role="button", name="Save")],
|
||||
),
|
||||
captured_at=42,
|
||||
selection=A11ySelection(ref="e2", text="Save", start_offset=0, end_offset=4),
|
||||
)
|
||||
),
|
||||
)
|
||||
self.assertEqual(
|
||||
msg.model_dump(),
|
||||
{
|
||||
"label": "rtvi-ai",
|
||||
"type": "ui-snapshot",
|
||||
"id": "m2",
|
||||
"data": {
|
||||
"tree": {
|
||||
"root": {
|
||||
"ref": "e1",
|
||||
"role": "main",
|
||||
"name": None,
|
||||
"value": None,
|
||||
"state": None,
|
||||
"level": None,
|
||||
"colcount": None,
|
||||
"rowcount": None,
|
||||
"children": [
|
||||
{
|
||||
"ref": "e2",
|
||||
"role": "button",
|
||||
"name": "Save",
|
||||
"value": None,
|
||||
"state": None,
|
||||
"level": None,
|
||||
"colcount": None,
|
||||
"rowcount": None,
|
||||
"children": None,
|
||||
}
|
||||
],
|
||||
},
|
||||
"captured_at": 42,
|
||||
"selection": {
|
||||
"ref": "e2",
|
||||
"text": "Save",
|
||||
"start_offset": 0,
|
||||
"end_offset": 4,
|
||||
},
|
||||
}
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
def test_ui_snapshot_allows_future_client_fields(self):
|
||||
msg = UISnapshotMessage.model_validate(
|
||||
{
|
||||
"id": "m2",
|
||||
"data": {
|
||||
"tree": {
|
||||
"root": {
|
||||
"ref": "e1",
|
||||
"role": "main",
|
||||
"bounds": {"x": 1, "y": 2},
|
||||
"children": [
|
||||
{
|
||||
"ref": "e2",
|
||||
"role": "button",
|
||||
"name": "Save",
|
||||
"platform_state": {"pressed": False},
|
||||
}
|
||||
],
|
||||
},
|
||||
"captured_at": 42,
|
||||
"selection": {
|
||||
"ref": "e2",
|
||||
"text": "Save",
|
||||
"direction": "forward",
|
||||
},
|
||||
"viewport": {"width": 1024, "height": 768},
|
||||
}
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
dumped = msg.model_dump()
|
||||
tree = dumped["data"]["tree"]
|
||||
self.assertEqual(tree["root"]["bounds"], {"x": 1, "y": 2})
|
||||
self.assertEqual(tree["root"]["children"][0]["platform_state"], {"pressed": False})
|
||||
self.assertEqual(tree["selection"]["direction"], "forward")
|
||||
self.assertEqual(tree["viewport"], {"width": 1024, "height": 768})
|
||||
|
||||
def test_ui_cancel_task_envelope(self):
|
||||
msg = UICancelTaskMessage(id="m3", data=UICancelTaskData(task_id="t-99", reason="user"))
|
||||
self.assertEqual(
|
||||
msg.model_dump(),
|
||||
{
|
||||
"label": "rtvi-ai",
|
||||
"type": "ui-cancel-task",
|
||||
"id": "m3",
|
||||
"data": {"task_id": "t-99", "reason": "user"},
|
||||
},
|
||||
)
|
||||
|
||||
def test_ui_task_group_started(self):
|
||||
msg = UITaskMessage(
|
||||
data=UITaskGroupStartedData(task_id="t-1", agents=["a", "b"], label="Search", at=42)
|
||||
)
|
||||
self.assertEqual(msg.type, "ui-task")
|
||||
self.assertEqual(msg.data.kind, "group_started")
|
||||
self.assertEqual(msg.data.task_id, "t-1")
|
||||
|
||||
def test_ui_task_update(self):
|
||||
msg = UITaskMessage(
|
||||
data=UITaskUpdateData(task_id="t-1", agent_name="a", data={"progress": 0.5}, at=43)
|
||||
)
|
||||
self.assertEqual(msg.data.kind, "task_update")
|
||||
self.assertEqual(msg.data.agent_name, "a")
|
||||
|
||||
def test_ui_task_completed(self):
|
||||
msg = UITaskMessage(
|
||||
data=UITaskCompletedData(
|
||||
task_id="t-1", agent_name="a", status="completed", response={"ok": True}, at=44
|
||||
)
|
||||
)
|
||||
self.assertEqual(msg.data.kind, "task_completed")
|
||||
self.assertEqual(msg.data.status, "completed")
|
||||
|
||||
def test_ui_task_group_completed(self):
|
||||
msg = UITaskMessage(data=UITaskGroupCompletedData(task_id="t-1", at=45))
|
||||
self.assertEqual(msg.data.kind, "group_completed")
|
||||
|
||||
|
||||
class TestPayloadShapes(unittest.TestCase):
|
||||
"""Pin the on-the-wire dict shape of each command payload."""
|
||||
|
||||
def test_toast_required_only(self):
|
||||
self.assertEqual(
|
||||
dict(Toast(title="Saved")),
|
||||
{
|
||||
"title": "Saved",
|
||||
"subtitle": None,
|
||||
"description": None,
|
||||
"image_url": None,
|
||||
"duration_ms": None,
|
||||
},
|
||||
)
|
||||
|
||||
def test_navigate(self):
|
||||
self.assertEqual(
|
||||
dict(Navigate(view="home", params={"id": "42"})),
|
||||
{"view": "home", "params": {"id": "42"}},
|
||||
)
|
||||
|
||||
def test_scroll_to_with_ref(self):
|
||||
self.assertEqual(
|
||||
dict(ScrollTo(ref="e42", behavior="smooth")),
|
||||
{"ref": "e42", "target_id": None, "behavior": "smooth"},
|
||||
)
|
||||
|
||||
def test_highlight_with_target_id(self):
|
||||
self.assertEqual(
|
||||
dict(Highlight(target_id="cta", duration_ms=2000)),
|
||||
{"ref": None, "target_id": "cta", "duration_ms": 2000},
|
||||
)
|
||||
|
||||
def test_focus(self):
|
||||
self.assertEqual(
|
||||
dict(Focus(ref="e7")),
|
||||
{"ref": "e7", "target_id": None},
|
||||
)
|
||||
|
||||
def test_click(self):
|
||||
self.assertEqual(
|
||||
dict(Click(ref="e9")),
|
||||
{"ref": "e9", "target_id": None},
|
||||
)
|
||||
|
||||
def test_set_input_value_default_replace_true(self):
|
||||
self.assertEqual(
|
||||
dict(SetInputValue(ref="e3", value="Marie Curie")),
|
||||
{"value": "Marie Curie", "ref": "e3", "target_id": None, "replace": True},
|
||||
)
|
||||
|
||||
def test_set_input_value_append(self):
|
||||
payload = dict(SetInputValue(ref="e3", value="more", replace=False))
|
||||
self.assertFalse(payload["replace"])
|
||||
|
||||
def test_select_text_full(self):
|
||||
self.assertEqual(
|
||||
dict(SelectText(ref="e15", start_offset=4, end_offset=12)),
|
||||
{"ref": "e15", "target_id": None, "start_offset": 4, "end_offset": 12},
|
||||
)
|
||||
|
||||
def test_select_text_whole_target_when_offsets_omitted(self):
|
||||
payload = dict(SelectText(ref="e15"))
|
||||
self.assertIsNone(payload["start_offset"])
|
||||
self.assertIsNone(payload["end_offset"])
|
||||
|
||||
|
||||
class TestFrames(unittest.TestCase):
|
||||
"""Pin the frame API for named UI messages."""
|
||||
|
||||
def test_ui_command_frame_names_command_and_payload(self):
|
||||
frame = RTVIUICommandFrame(command="toast", payload={"title": "Saved"})
|
||||
self.assertEqual(frame.command, "toast")
|
||||
self.assertEqual(frame.payload, {"title": "Saved"})
|
||||
|
||||
def test_ui_event_frame_names_event_and_payload(self):
|
||||
frame = RTVIUIEventFrame(msg_id="m1", event="nav_click", payload={"view": "home"})
|
||||
self.assertEqual(frame.msg_id, "m1")
|
||||
self.assertEqual(frame.event, "nav_click")
|
||||
self.assertEqual(frame.payload, {"view": "home"})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user