# # Copyright (c) 2026, Daily # # SPDX-License-Identifier: BSD 2-Clause License # """Smoke tests for the UI Worker 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, UICancelJobGroupData, UICancelJobGroupMessage, UICommandData, UICommandMessage, UIEventData, UIEventMessage, UIJobCompletedData, UIJobGroupCompletedData, UIJobGroupMessage, UIJobGroupStartedData, UIJobUpdateData, UISnapshotData, UISnapshotMessage, ) 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 = UICancelJobGroupMessage( id="m3", data=UICancelJobGroupData(job_id="t-99", reason="user") ) self.assertEqual( msg.model_dump(), { "label": "rtvi-ai", "type": "ui-cancel-job-group", "id": "m3", "data": {"job_id": "t-99", "reason": "user"}, }, ) def test_ui_job_group_started(self): msg = UIJobGroupMessage( data=UIJobGroupStartedData(job_id="t-1", workers=["a", "b"], label="Search", at=42) ) self.assertEqual(msg.type, "ui-job-group") self.assertEqual(msg.data.kind, "group_started") self.assertEqual(msg.data.job_id, "t-1") def test_ui_job_update(self): msg = UIJobGroupMessage( data=UIJobUpdateData(job_id="t-1", worker_name="a", data={"progress": 0.5}, at=43) ) self.assertEqual(msg.data.kind, "job_update") self.assertEqual(msg.data.worker_name, "a") def test_ui_job_completed(self): msg = UIJobGroupMessage( data=UIJobCompletedData( job_id="t-1", worker_name="a", status="completed", response={"ok": True}, at=44 ) ) self.assertEqual(msg.data.kind, "job_completed") self.assertEqual(msg.data.status, "completed") def test_ui_job_group_completed(self): msg = UIJobGroupMessage(data=UIJobGroupCompletedData(job_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()