qzl
49 lines
1.2 KiB
Python
49 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from core.agentscope.events.agui_codec import to_agui_wire_event
|
|
|
|
|
|
def test_to_agui_wire_event_supports_custom_internal_type() -> None:
|
|
event = {
|
|
"type": "DIVINATION_DERIVED",
|
|
"threadId": "t1",
|
|
"runId": "r1",
|
|
"divination": {"guaName": "山火贲"},
|
|
}
|
|
|
|
wire = to_agui_wire_event(event)
|
|
|
|
assert wire["type"] == "DIVINATION_DERIVED"
|
|
assert wire["threadId"] == "t1"
|
|
assert wire["runId"] == "r1"
|
|
assert wire["divination"] == {"guaName": "山火贲"}
|
|
|
|
|
|
def test_to_agui_wire_event_rejects_unknown_type() -> None:
|
|
event = {
|
|
"type": "UNCONTROLLED_CUSTOM_EVENT",
|
|
"threadId": "t1",
|
|
"runId": "r1",
|
|
}
|
|
|
|
try:
|
|
to_agui_wire_event(event)
|
|
except ValueError as exc:
|
|
assert "unsupported ag-ui event type" in str(exc)
|
|
return
|
|
raise AssertionError("expected ValueError for unsupported event type")
|
|
|
|
|
|
def test_to_agui_wire_event_maps_known_internal_type() -> None:
|
|
event = {
|
|
"type": "run.started",
|
|
"threadId": "t1",
|
|
"runId": "r1",
|
|
}
|
|
|
|
wire = to_agui_wire_event(event)
|
|
|
|
assert wire["type"] == "RUN_STARTED"
|
|
assert wire["threadId"] == "t1"
|
|
assert wire["runId"] == "r1"
|