mirror of
https://github.com/hansjone/oclaw.git
synced 2026-10-09 03:30:48 +08:00
oliver
420abac9f1
- Rename platform/ to svc/ to avoid shadowing stdlib platform. - Replace from oclaw.* with from svc/runtime/interfaces; update -m CLI paths. - tests/conftest: prepend repo root to sys.path (no parent-folder package name). - CI: paths and offline_eval script under repo root. - Ops scripts: PYTHONPATH must be repo root for python -m runtime.* (fixes gateway/WhatsApp sidecar startup). - Fix default oclaw.json path in tabular/file attachment limits; stabilize attachment test config. Co-authored-by: Cursor <cursoragent@cursor.com>
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from svc.llm.transports.openai_responses import parse_openai_responses_stream_events
|
|
|
|
|
|
def test_openai_responses_stream_parses_text_deltas_and_final_response_tool_calls() -> None:
|
|
events = [
|
|
{"type": "response.created"},
|
|
{"type": "response.output_text.delta", "delta": "Hel"},
|
|
{"type": "response.output_text.delta", "delta": "lo"},
|
|
{
|
|
"type": "response.completed",
|
|
"response": {
|
|
"id": "resp_1",
|
|
"output_text": "Hello",
|
|
"output": [
|
|
{
|
|
"type": "function_call",
|
|
"call_id": "call_1",
|
|
"name": "query_route",
|
|
"arguments": {"destination": "1.1.1.1"},
|
|
}
|
|
],
|
|
},
|
|
},
|
|
]
|
|
buf: list[str] = []
|
|
text, tool_calls, final_resp, reasoning_text = parse_openai_responses_stream_events(events, on_token=buf.append)
|
|
assert text == "Hello"
|
|
assert "".join(buf) == "Hello"
|
|
assert reasoning_text == ""
|
|
assert final_resp and final_resp.get("id") == "resp_1"
|
|
assert len(tool_calls) == 1
|
|
assert tool_calls[0].id == "call_1"
|
|
assert tool_calls[0].name == "query_route"
|
|
assert tool_calls[0].arguments.get("destination") == "1.1.1.1"
|
|
|
|
|
|
def test_openai_responses_stream_emits_reasoning_summary_deltas() -> None:
|
|
events = [
|
|
{"type": "response.reasoning_summary_text.delta", "delta": "step "},
|
|
{"type": "response.reasoning_summary_text.delta", "delta": "one"},
|
|
{"type": "response.output_text.delta", "delta": "Hi"},
|
|
{"type": "response.completed", "response": {"id": "resp_2", "output_text": "Hi"}},
|
|
]
|
|
buf: list[str] = []
|
|
text, tool_calls, final_resp, reasoning_text = parse_openai_responses_stream_events(events, on_token=buf.append)
|
|
assert text == "Hi"
|
|
assert reasoning_text == "step one"
|
|
assert "".join(buf) == "step oneHi"
|
|
assert final_resp and final_resp.get("id") == "resp_2"
|
|
assert tool_calls == []
|
|
|
|
|
|
def test_openai_responses_stream_reasoning_fallback_from_completed_output() -> None:
|
|
events = [
|
|
{"type": "response.output_text.delta", "delta": "Z"},
|
|
{
|
|
"type": "response.completed",
|
|
"response": {
|
|
"id": "resp_3",
|
|
"output_text": "Z",
|
|
"output": [{"type": "reasoning", "summary": "silent plan"}],
|
|
},
|
|
},
|
|
]
|
|
text, _, _, reasoning_text = parse_openai_responses_stream_events(events)
|
|
assert text == "Z"
|
|
assert reasoning_text == "silent plan"
|
|
|