mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 12:25:29 +08:00
746 lines
27 KiB
Python
746 lines
27 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import types
|
|
from collections.abc import AsyncGenerator
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from langbot_plugin.api.entities.builtin.provider.message import ContentElement
|
|
from langbot_plugin.api.entities.builtin.runner import (
|
|
ActorContext,
|
|
AgentEventContext,
|
|
AgentInput,
|
|
AgentResources,
|
|
AgentRunState,
|
|
AgentRuntimeContext,
|
|
AgentTrigger,
|
|
ConversationContext,
|
|
DeliveryContext,
|
|
InteractionSubmission,
|
|
RunnerContext,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _load_runner_module(plugin_dir: str, stubs: dict[str, Any] | None = None):
|
|
for module_name in list(sys.modules):
|
|
if module_name == "pkg" or module_name.startswith("pkg."):
|
|
del sys.modules[module_name]
|
|
|
|
installed_stubs: dict[str, Any] = {}
|
|
for name, value in (stubs or {}).items():
|
|
if name not in sys.modules:
|
|
installed_stubs[name] = value
|
|
sys.modules[name] = value
|
|
|
|
plugin_root = ROOT / plugin_dir
|
|
sys.path.insert(0, str(plugin_root))
|
|
try:
|
|
module_path = plugin_root / "components" / "runner" / "default.py"
|
|
spec = importlib.util.spec_from_file_location(
|
|
f"test_traditional_{plugin_dir.replace('-', '_')}_runner",
|
|
module_path,
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
finally:
|
|
sys.path.remove(str(plugin_root))
|
|
for name in installed_stubs:
|
|
sys.modules.pop(name, None)
|
|
|
|
|
|
async def _collect_async(generator):
|
|
return [item async for item in generator]
|
|
|
|
|
|
def _type(result) -> str:
|
|
return getattr(result.type, "value", str(result.type))
|
|
|
|
|
|
def _ctx(
|
|
*,
|
|
config: dict[str, Any] | None = None,
|
|
text: str = "hello",
|
|
contents: list[ContentElement] | None = None,
|
|
conversation_state: dict[str, Any] | None = None,
|
|
event_type: str = "message.received",
|
|
interaction: InteractionSubmission | None = None,
|
|
) -> RunnerContext:
|
|
return RunnerContext(
|
|
run_id="run_traditional",
|
|
trigger=AgentTrigger(type=event_type),
|
|
event=AgentEventContext(event_id="evt_1", event_type=event_type, source="test"),
|
|
conversation=ConversationContext(
|
|
conversation_id="langbot-conv-1",
|
|
session_id="langbot-session-1",
|
|
),
|
|
input=AgentInput(text=text, contents=contents or [], interaction=interaction),
|
|
delivery=DeliveryContext(surface="test", supports_streaming=True),
|
|
resources=AgentResources(),
|
|
state=AgentRunState(conversation=conversation_state or {}),
|
|
runtime=AgentRuntimeContext(),
|
|
actor=ActorContext(actor_type="user", actor_id="traditional-sender"),
|
|
config=config or {},
|
|
)
|
|
|
|
|
|
def _text_and_image_ctx(config: dict[str, Any]) -> RunnerContext:
|
|
return _ctx(
|
|
config=config,
|
|
text="describe this",
|
|
contents=[ContentElement.from_image_base64("data:image/png;base64,aGVsbG8=")],
|
|
)
|
|
|
|
|
|
def test_coze_does_not_reuse_langbot_conversation_id_as_external_id() -> None:
|
|
module = _load_runner_module("coze-agent")
|
|
runner = object.__new__(module.DefaultRunner)
|
|
|
|
assert runner._get_external_conversation_id(_ctx()) is None
|
|
|
|
ctx = _ctx(conversation_state={"external.conversation_id": "coze-conv-1"})
|
|
assert runner._get_external_conversation_id(ctx) == "coze-conv-1"
|
|
|
|
|
|
def test_n8n_uses_runner_owned_conversation_and_session_ids() -> None:
|
|
module = _load_runner_module("n8n-agent")
|
|
runner = object.__new__(module.DefaultRunner)
|
|
ctx = _ctx()
|
|
|
|
conversation_id, conversation_created = runner._get_or_create_state_id(
|
|
ctx,
|
|
"external.conversation_id",
|
|
"n8n_conversation",
|
|
)
|
|
session_id, session_created = runner._get_or_create_state_id(
|
|
ctx,
|
|
"external.session_id",
|
|
"n8n_session",
|
|
)
|
|
payload = runner._build_payload(ctx, "hello", conversation_id, session_id)
|
|
|
|
assert conversation_created is True
|
|
assert session_created is True
|
|
assert payload["conversation_id"].startswith("n8n_conversation_")
|
|
assert payload["session_id"].startswith("n8n_session_")
|
|
assert payload["conversation_id"] != "langbot-conv-1"
|
|
assert payload["session_id"] != "langbot-session-1"
|
|
|
|
|
|
def test_dify_workflow_uses_runner_owned_state_ids() -> None:
|
|
module = _load_runner_module("dify-agent")
|
|
runner = object.__new__(module.DefaultRunner)
|
|
captured_inputs: dict[str, Any] = {}
|
|
|
|
class FakeClient:
|
|
async def workflow_run(self, *, inputs, user, files):
|
|
captured_inputs.update(inputs)
|
|
yield {
|
|
"event": "workflow_finished",
|
|
"data": {"outputs": {"summary": "workflow ok"}},
|
|
}
|
|
|
|
ctx = _ctx()
|
|
results = asyncio.run(
|
|
_collect_async(
|
|
runner._run_workflow(
|
|
ctx,
|
|
FakeClient(),
|
|
{},
|
|
"hello",
|
|
"user_1",
|
|
[],
|
|
False,
|
|
)
|
|
)
|
|
)
|
|
|
|
assert captured_inputs["langbot_session_id"].startswith("dify_session_")
|
|
assert captured_inputs["langbot_conversation_id"].startswith("dify_conversation_")
|
|
assert captured_inputs["langbot_session_id"] != "langbot-session-1"
|
|
assert captured_inputs["langbot_conversation_id"] != "langbot-conv-1"
|
|
assert ("state.updated", "external.workflow_session_id") in {
|
|
(_type(item), item.data.get("key")) for item in results
|
|
}
|
|
assert ("state.updated", "external.workflow_conversation_id") in {
|
|
(_type(item), item.data.get("key")) for item in results
|
|
}
|
|
assert _type(results[-1]) == "run.completed"
|
|
|
|
|
|
class _FakePluginStorage:
|
|
def __init__(self) -> None:
|
|
self.values: dict[str, bytes] = {}
|
|
self.deleted: list[str] = []
|
|
|
|
async def set_plugin_storage(self, key: str, value: bytes) -> None:
|
|
self.values[key] = value
|
|
|
|
async def get_plugin_storage(self, key: str) -> bytes:
|
|
return self.values[key]
|
|
|
|
async def delete_plugin_storage(self, key: str) -> None:
|
|
self.deleted.append(key)
|
|
self.values.pop(key, None)
|
|
|
|
|
|
def test_dify_workflow_pause_requests_host_interaction_and_hides_provider_tokens() -> None:
|
|
module = _load_runner_module("dify-agent")
|
|
runner = object.__new__(module.DefaultRunner)
|
|
storage = _FakePluginStorage()
|
|
runner.get_run_api = lambda ctx: storage
|
|
|
|
class FakeClient:
|
|
async def workflow_run(self, *, inputs, user, files):
|
|
yield {"event": "workflow_started", "data": {"workflow_run_id": "workflow-private"}}
|
|
yield {
|
|
"event": "node_started",
|
|
"data": {
|
|
"node_id": "human-input-node",
|
|
"node_type": "human-input",
|
|
"title": "Manual review",
|
|
},
|
|
}
|
|
yield {
|
|
"event": "workflow_paused",
|
|
"data": {
|
|
"reasons": [
|
|
{
|
|
"TYPE": "human_input_required",
|
|
"form_token": "form-private",
|
|
"node_title": "Manual review",
|
|
"form_content": "Choose a priority\n{{#$output.priority#}}",
|
|
"inputs": [
|
|
{
|
|
"output_variable_name": "priority",
|
|
"type": "select",
|
|
"option_source": {"type": "constant", "value": ["high", "low"]},
|
|
}
|
|
],
|
|
"actions": [{"id": "approve-private", "title": "Approve"}],
|
|
}
|
|
],
|
|
},
|
|
}
|
|
|
|
results = asyncio.run(_collect_async(runner._run_workflow(_ctx(), FakeClient(), {}, "hello", "user_1", [], False)))
|
|
|
|
interaction_result = results[-1]
|
|
assert _type(results[0]) == "tool.call.started"
|
|
assert results[0].data["tool_name"] == "Manual review"
|
|
assert all(_type(result) != "message.delta" for result in results)
|
|
assert _type(interaction_result) == "action.requested"
|
|
assert interaction_result.data["action"] == "interaction.requested"
|
|
request = interaction_result.data["payload"]
|
|
assert request["fields"][0]["options"][0]["value"] == "high"
|
|
assert request["actions"] == []
|
|
assert "form-private" not in str(request)
|
|
assert "workflow-private" not in str(request)
|
|
assert len(storage.values) == 1
|
|
continuation = json.loads(next(iter(storage.values.values())))
|
|
assert continuation["form_token"] == "form-private"
|
|
assert continuation["workflow_run_id"] == "workflow-private"
|
|
assert continuation["interaction_actions"][0]["label"] == "Approve"
|
|
assert all(_type(result) != "run.completed" for result in results)
|
|
|
|
|
|
def test_dify_interaction_resume_submits_mapped_values_and_clears_continuation() -> None:
|
|
module = _load_runner_module("dify-agent")
|
|
runner = object.__new__(module.DefaultRunner)
|
|
storage = _FakePluginStorage()
|
|
runner.get_run_api = lambda ctx: storage
|
|
interaction_id = "dify-test"
|
|
storage_key = module._interaction_storage_key(interaction_id)
|
|
storage.values[storage_key] = json.dumps(
|
|
{
|
|
"version": 1,
|
|
"interaction_id": interaction_id,
|
|
"form_token": "form-private",
|
|
"workflow_run_id": "workflow-private",
|
|
"user": "user_1",
|
|
"field_map": {"field_1": "priority"},
|
|
"action_map": {"action_1": "approve-private"},
|
|
"default_inputs": {"comment": "default"},
|
|
}
|
|
).encode()
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeClient:
|
|
async def workflow_submit(self, **kwargs):
|
|
captured.update(kwargs)
|
|
yield {
|
|
"event": "workflow_finished",
|
|
"data": {"error": None, "outputs": {"summary": "Approved"}},
|
|
}
|
|
|
|
submission = InteractionSubmission(
|
|
interaction_id=interaction_id,
|
|
action_id="action_1",
|
|
values={"field_1": "high"},
|
|
)
|
|
# Persist through the real contract: owner scope, expiry and pending status.
|
|
asyncio.run(
|
|
runner._store_interaction_continuation(
|
|
_ctx(), json.loads(storage.values[module._interaction_storage_key(interaction_id)])
|
|
)
|
|
)
|
|
results = asyncio.run(_collect_async(runner._resume_workflow(_ctx(), FakeClient(), submission, False)))
|
|
|
|
assert captured == {
|
|
"form_token": "form-private",
|
|
"workflow_run_id": "workflow-private",
|
|
"inputs": {"comment": "default", "priority": "high"},
|
|
"user": "user_1",
|
|
"action": "approve-private",
|
|
}
|
|
assert [_type(result) for result in results] == ["message.delta", "run.completed"]
|
|
assert storage_key in storage.deleted
|
|
assert storage_key not in storage.values
|
|
|
|
|
|
def test_dify_field_submission_advances_to_action_without_calling_provider() -> None:
|
|
module = _load_runner_module("dify-agent")
|
|
runner = object.__new__(module.DefaultRunner)
|
|
storage = _FakePluginStorage()
|
|
runner.get_run_api = lambda ctx: storage
|
|
interaction_id = "dify-field"
|
|
storage.values[module._interaction_storage_key(interaction_id)] = json.dumps(
|
|
{
|
|
"version": 1,
|
|
"interaction_id": interaction_id,
|
|
"form_token": "form-private",
|
|
"workflow_run_id": "workflow-private",
|
|
"user": "user_1",
|
|
"field_map": {"field_1": "priority"},
|
|
"action_map": {"action_1": "approve-private"},
|
|
"default_inputs": {},
|
|
"title": "Manual review",
|
|
"description": "Review it",
|
|
"fallback_text": "Review it",
|
|
"phase": "field",
|
|
"current_fields": [
|
|
{
|
|
"id": "field_1",
|
|
"label": "Priority",
|
|
"type": "select",
|
|
"required": True,
|
|
"options": [{"value": "high", "label": "High"}],
|
|
}
|
|
],
|
|
"remaining_fields": [],
|
|
"interaction_actions": [{"id": "action_1", "label": "Approve", "style": "primary"}],
|
|
}
|
|
).encode()
|
|
|
|
class ProviderMustNotRun:
|
|
def workflow_submit(self, **kwargs):
|
|
raise AssertionError("provider must not resume before the action step")
|
|
|
|
submission = InteractionSubmission(interaction_id=interaction_id, values={"field_1": "high"})
|
|
# Persist through the real contract: owner scope, expiry and pending status.
|
|
asyncio.run(
|
|
runner._store_interaction_continuation(
|
|
_ctx(), json.loads(storage.values[module._interaction_storage_key(interaction_id)])
|
|
)
|
|
)
|
|
results = asyncio.run(_collect_async(runner._resume_workflow(_ctx(), ProviderMustNotRun(), submission, False)))
|
|
|
|
assert [_type(result) for result in results] == ["action.requested"]
|
|
request = results[0].data["payload"]
|
|
assert request["fields"] == []
|
|
assert request["actions"][0]["label"] == "Approve"
|
|
continuation = json.loads(storage.values[module._interaction_storage_key(request["interaction_id"])])
|
|
assert continuation["default_inputs"] == {"priority": "high"}
|
|
|
|
|
|
def test_dify_form_content_is_revealed_with_its_current_field() -> None:
|
|
module = _load_runner_module("dify-agent")
|
|
runner = object.__new__(module.DefaultRunner)
|
|
storage = _FakePluginStorage()
|
|
runner.get_run_api = lambda ctx: storage
|
|
request, continuation = runner._build_interaction_request(
|
|
{
|
|
"form_token": "form-private",
|
|
"node_title": "人工介入",
|
|
"form_content": ("1\n请输入你的问题\n{{#$output.us_input#}}\n\n请选择你的答案\n{{#$output.xiala#}}"),
|
|
"inputs": [
|
|
{"output_variable_name": "us_input", "type": "paragraph", "required": True},
|
|
{
|
|
"output_variable_name": "xiala",
|
|
"type": "select",
|
|
"required": True,
|
|
"option_source": {"type": "constant", "value": ["1", "2"]},
|
|
},
|
|
],
|
|
"actions": [{"id": "or-private", "title": "or"}],
|
|
},
|
|
"workflow-private",
|
|
"user-1",
|
|
)
|
|
|
|
assert request.description == "1\n请输入你的问题"
|
|
assert request.fields[0].label == "us_input"
|
|
assert "请选择你的答案" not in request.description
|
|
|
|
asyncio.run(runner._store_interaction_continuation(_ctx(), continuation))
|
|
second_result = asyncio.run(
|
|
runner._advance_field_interaction(
|
|
_ctx(),
|
|
continuation,
|
|
InteractionSubmission(interaction_id=request.interaction_id, values={"field_1": "问题内容"}),
|
|
)
|
|
)
|
|
second_request = second_result.data["payload"]
|
|
assert second_request["description"] == "请选择你的答案"
|
|
assert second_request["fields"][0]["label"] == "xiala"
|
|
assert "请输入你的问题" not in second_request["description"]
|
|
|
|
second_continuation = json.loads(storage.values[module._interaction_storage_key(second_request["interaction_id"])])
|
|
final_result = asyncio.run(
|
|
runner._advance_field_interaction(
|
|
_ctx(),
|
|
second_continuation,
|
|
InteractionSubmission(
|
|
interaction_id=second_request["interaction_id"],
|
|
values={"field_2": "1"},
|
|
),
|
|
)
|
|
)
|
|
final_request = final_result.data["payload"]
|
|
assert final_request["description"] is None
|
|
assert final_request["fields"] == []
|
|
assert final_request["actions"][0]["label"] == "or"
|
|
|
|
|
|
def test_dify_interaction_resume_can_pause_again() -> None:
|
|
module = _load_runner_module("dify-agent")
|
|
runner = object.__new__(module.DefaultRunner)
|
|
storage = _FakePluginStorage()
|
|
runner.get_run_api = lambda ctx: storage
|
|
interaction_id = "dify-first"
|
|
old_key = module._interaction_storage_key(interaction_id)
|
|
storage.values[old_key] = json.dumps(
|
|
{
|
|
"version": 1,
|
|
"interaction_id": interaction_id,
|
|
"form_token": "form-1",
|
|
"workflow_run_id": "workflow-1",
|
|
"user": "user_1",
|
|
"field_map": {},
|
|
"action_map": {"continue": ""},
|
|
"default_inputs": {},
|
|
}
|
|
).encode()
|
|
|
|
class FakeClient:
|
|
async def workflow_submit(self, **kwargs):
|
|
yield {
|
|
"event": "workflow_paused",
|
|
"data": {
|
|
"workflow_run_id": "workflow-2",
|
|
"reasons": [
|
|
{
|
|
"TYPE": "human_input_required",
|
|
"form_token": "form-2",
|
|
"node_title": "Second review",
|
|
"inputs": [],
|
|
"actions": [{"id": "done", "title": "Done"}],
|
|
}
|
|
],
|
|
},
|
|
}
|
|
|
|
submission = InteractionSubmission(interaction_id=interaction_id, action_id="continue")
|
|
# Persist through the real contract: owner scope, expiry and pending status.
|
|
asyncio.run(
|
|
runner._store_interaction_continuation(
|
|
_ctx(), json.loads(storage.values[module._interaction_storage_key(interaction_id)])
|
|
)
|
|
)
|
|
results = asyncio.run(_collect_async(runner._resume_workflow(_ctx(), FakeClient(), submission, False)))
|
|
|
|
assert [_type(result) for result in results] == ["action.requested"]
|
|
new_interaction_id = results[0].data["payload"]["interaction_id"]
|
|
assert new_interaction_id != interaction_id
|
|
assert old_key in storage.deleted
|
|
assert module._interaction_storage_key(new_interaction_id) in storage.values
|
|
|
|
|
|
def test_dify_client_submits_form_then_streams_resumed_events(monkeypatch) -> None:
|
|
module = _load_runner_module("dify-agent")
|
|
client_module = sys.modules["pkg.dify_client"]
|
|
real_async_client = httpx.AsyncClient
|
|
calls: list[httpx.Request] = []
|
|
expected_event = {
|
|
"event": "workflow_finished",
|
|
"data": {"outputs": {"summary": "done", "note": "完成"}},
|
|
}
|
|
event_bytes = b"data: " + json.dumps(expected_event, ensure_ascii=False).encode("utf-8")
|
|
# Split UTF-8 across the bounded reader's 8192-byte boundary as well as
|
|
# transport chunks. Keep the original blank line and unterminated EOF event.
|
|
padding = 8191 - len(b"\r\n:\r\n") - event_bytes.index("完".encode())
|
|
wire = b"\r\n:" + b" " * padding + b"\r\n" + event_bytes
|
|
assert wire[8191:8194] == "完".encode()
|
|
|
|
class ChunkedBytes(httpx.AsyncByteStream):
|
|
def __init__(self, chunks):
|
|
self.chunks = chunks
|
|
self.consumed = False
|
|
self.closed = False
|
|
|
|
async def __aiter__(self):
|
|
for chunk in self.chunks:
|
|
yield chunk
|
|
self.consumed = True
|
|
|
|
async def aclose(self):
|
|
self.closed = True
|
|
|
|
form_body = ChunkedBytes([b'{"res', b'ult":"success"}'])
|
|
# CR/LF, SSE prefix, JSON and multibyte characters may all span chunks.
|
|
event_body = ChunkedBytes([wire[:1], *[wire[i : i + 7] for i in range(1, len(wire), 7)]])
|
|
|
|
def handle_request(request):
|
|
calls.append(request)
|
|
if len(calls) == 1:
|
|
assert request.method == "POST"
|
|
return httpx.Response(200, headers={"content-type": "application/json"}, stream=form_body)
|
|
assert len(calls) == 2
|
|
assert request.method == "GET"
|
|
assert form_body.consumed and form_body.closed
|
|
return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=event_body)
|
|
|
|
transport = httpx.MockTransport(handle_request)
|
|
monkeypatch.setattr(
|
|
client_module.httpx,
|
|
"AsyncClient",
|
|
lambda **kwargs: real_async_client(transport=transport, **kwargs),
|
|
)
|
|
client = module.AsyncDifyClient(api_key="key", base_url="https://dify.example/v1")
|
|
events = asyncio.run(
|
|
_collect_async(
|
|
client.workflow_submit(
|
|
form_token="form-token",
|
|
workflow_run_id="workflow-run",
|
|
inputs={"priority": "high"},
|
|
user="user-1",
|
|
action="approve",
|
|
)
|
|
)
|
|
)
|
|
|
|
assert events[0]["event"] == "workflow_finished"
|
|
assert events == [expected_event]
|
|
assert [(request.method, request.url.path) for request in calls] == [
|
|
("POST", "/v1/form/human_input/form-token"),
|
|
("GET", "/v1/workflow/workflow-run/events"),
|
|
]
|
|
assert json.loads(calls[0].content) == {
|
|
"inputs": {"priority": "high"},
|
|
"user": "user-1",
|
|
"action": "approve",
|
|
}
|
|
assert calls[0].headers["content-type"] == "application/json"
|
|
assert dict(calls[1].url.params) == {"user": "user-1"}
|
|
assert calls[1].content == b""
|
|
for request in calls:
|
|
assert request.url.host == "dify.example"
|
|
assert request.headers["authorization"] == "Bearer key"
|
|
assert form_body.consumed and form_body.closed
|
|
assert event_body.consumed and event_body.closed
|
|
|
|
|
|
def test_dify_text_image_upload_failure_is_input_error() -> None:
|
|
module = _load_runner_module("dify-agent")
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def upload_file(self, *args, **kwargs):
|
|
raise module.DifyAPIError("provider upload failed", code="dify.http_error")
|
|
|
|
module.AsyncDifyClient = FakeClient
|
|
runner = object.__new__(module.DefaultRunner)
|
|
ctx = _text_and_image_ctx(
|
|
{
|
|
"base-url": "https://api.dify.ai/v1",
|
|
"api-key": "key",
|
|
"app-type": "chat",
|
|
}
|
|
)
|
|
|
|
results = asyncio.run(_collect_async(runner.run(ctx)))
|
|
|
|
assert [_type(item) for item in results] == ["run.failed"]
|
|
assert results[0].data["code"] == "dify.input_error"
|
|
assert results[0].data["retryable"] is False
|
|
|
|
|
|
def test_coze_text_image_upload_failure_is_input_error() -> None:
|
|
module = _load_runner_module("coze-agent")
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def upload_file(self, *args, **kwargs):
|
|
raise module.CozeAPIError("provider upload failed", code="coze.http_error")
|
|
|
|
async def close(self):
|
|
pass
|
|
|
|
module.AsyncCozeClient = FakeClient
|
|
runner = object.__new__(module.DefaultRunner)
|
|
ctx = _text_and_image_ctx({"api-key": "key", "bot-id": "bot"})
|
|
|
|
results = asyncio.run(_collect_async(runner.run(ctx)))
|
|
|
|
assert [_type(item) for item in results] == ["run.failed"]
|
|
assert results[0].data["code"] == "coze.input_error"
|
|
|
|
|
|
def test_dashscope_timeout_maps_to_retryable_run_failed() -> None:
|
|
dashscope_stub = types.SimpleNamespace(Application=types.SimpleNamespace(call=lambda **kwargs: iter(())))
|
|
module = _load_runner_module("dashscope-agent", stubs={"dashscope": dashscope_stub})
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def iter_agent(self, **kwargs):
|
|
raise module.DashScopeAPIError("timeout", code="dashscope.timeout", retryable=True)
|
|
yield {}
|
|
|
|
module.DashScopeClient = FakeClient
|
|
runner = object.__new__(module.DefaultRunner)
|
|
ctx = _ctx(config={"api-key": "key", "app-id": "app", "app-type": "agent"})
|
|
|
|
results = asyncio.run(_collect_async(runner.run(ctx)))
|
|
|
|
assert [_type(item) for item in results] == ["run.failed"]
|
|
assert results[0].data == {"code": "dashscope.timeout", "error": "timeout", "retryable": True}
|
|
|
|
|
|
def test_dashscope_empty_output_fails_instead_of_completing() -> None:
|
|
dashscope_stub = types.SimpleNamespace(Application=types.SimpleNamespace(call=lambda **kwargs: iter(())))
|
|
module = _load_runner_module("dashscope-agent", stubs={"dashscope": dashscope_stub})
|
|
|
|
class FakeClient:
|
|
references_quote = "refs:"
|
|
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def iter_agent(self, **kwargs):
|
|
if False:
|
|
yield {}
|
|
|
|
module.DashScopeClient = FakeClient
|
|
runner = object.__new__(module.DefaultRunner)
|
|
ctx = _ctx(config={"api-key": "key", "app-id": "app", "app-type": "agent"})
|
|
|
|
results = asyncio.run(_collect_async(runner.run(ctx)))
|
|
|
|
assert [_type(item) for item in results] == ["run.failed"]
|
|
assert results[0].data["code"] == "dashscope.empty_response"
|
|
|
|
|
|
def test_tbox_timeout_maps_to_retryable_run_failed() -> None:
|
|
module = _load_runner_module("tbox-agent")
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def chat(self, **kwargs):
|
|
raise module.TboxAPIError("timeout", code="tbox.timeout", retryable=True)
|
|
yield {}
|
|
|
|
module.AsyncTboxClient = FakeClient
|
|
runner = object.__new__(module.DefaultRunner)
|
|
ctx = _ctx(config={"api-key": "key", "app-id": "app"})
|
|
|
|
results = asyncio.run(_collect_async(runner.run(ctx)))
|
|
|
|
assert [_type(item) for item in results] == ["run.failed"]
|
|
assert results[0].data == {"code": "tbox.timeout", "error": "timeout", "retryable": True}
|
|
|
|
|
|
def test_deerflow_timeout_maps_to_retryable_run_failed() -> None:
|
|
module = _load_runner_module("deerflow-agent")
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def stream_run(self, **kwargs):
|
|
raise TimeoutError("provider timeout")
|
|
yield {}
|
|
|
|
module.AsyncDeerFlowClient = FakeClient
|
|
runner = object.__new__(module.DefaultRunner)
|
|
runner._ensure_thread_id = lambda ctx, client, timeout: _async_value(("thread-1", False))
|
|
ctx = _ctx(config={"api-base": "http://127.0.0.1:2026", "streaming": True, "timeout": 1})
|
|
|
|
results = asyncio.run(_collect_async(runner.run(ctx)))
|
|
|
|
assert [_type(item) for item in results] == ["run.failed"]
|
|
assert results[0].data["code"] == "deerflow.timeout"
|
|
assert results[0].data["retryable"] is True
|
|
|
|
|
|
async def _async_value(value):
|
|
return value
|
|
|
|
|
|
def test_weknora_timeout_maps_to_retryable_run_failed() -> None:
|
|
module = _load_runner_module("weknora-agent")
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def create_session(self, **kwargs):
|
|
raise module.WeKnoraAPIError("timeout", code="weknora.timeout", retryable=True)
|
|
|
|
module.AsyncWeKnoraClient = FakeClient
|
|
runner = object.__new__(module.DefaultRunner)
|
|
ctx = _ctx(config={"base-url": "http://weknora/api/v1", "api-key": "key"})
|
|
|
|
results = asyncio.run(_collect_async(runner.run(ctx)))
|
|
|
|
assert [_type(item) for item in results] == ["run.failed"]
|
|
assert results[0].data == {"code": "weknora.timeout", "error": "timeout", "retryable": True}
|
|
|
|
|
|
def test_weknora_empty_answer_fails_instead_of_completing() -> None:
|
|
module = _load_runner_module("weknora-agent")
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
async def agent_chat(self, **kwargs) -> AsyncGenerator[dict[str, Any], None]:
|
|
yield {"response_type": "answer", "content": "", "done": True}
|
|
|
|
module.AsyncWeKnoraClient = FakeClient
|
|
runner = object.__new__(module.DefaultRunner)
|
|
ctx = _ctx(
|
|
config={"base-url": "http://weknora/api/v1", "api-key": "key", "app-type": "agent"},
|
|
conversation_state={"external.session_id": "weknora-session-1"},
|
|
)
|
|
|
|
results = asyncio.run(_collect_async(runner.run(ctx)))
|
|
|
|
assert [_type(item) for item in results] == ["run.failed"]
|
|
assert results[0].data["code"] == "weknora.empty_response"
|