2026-03-16 16:11:06 +08:00
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
|
from uuid import uuid4
|
|
|
|
|
|
|
|
|
|
from v1.agent.utils import convert_message_to_history
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class _FakeMessage:
|
|
|
|
|
def __init__(self, *, role: str, metadata: dict[str, object] | None) -> None:
|
|
|
|
|
self.id = uuid4()
|
|
|
|
|
self.seq = 1
|
|
|
|
|
self.role = role
|
|
|
|
|
self.content = "content"
|
|
|
|
|
self.metadata = metadata
|
|
|
|
|
self.timestamp = datetime.now(timezone.utc)
|
|
|
|
|
|
|
|
|
|
|
2026-04-23 12:12:41 +08:00
|
|
|
def test_convert_message_to_history_attaches_ui_schema_for_tool_message() -> None:
|
2026-03-16 16:11:06 +08:00
|
|
|
message = _FakeMessage(
|
|
|
|
|
role="tool",
|
2026-04-23 12:12:41 +08:00
|
|
|
metadata={
|
|
|
|
|
"tool_agent_output": {
|
|
|
|
|
"result": {"status": "success"},
|
|
|
|
|
"ui_hints": {
|
|
|
|
|
"intent": "status",
|
|
|
|
|
"status": "success",
|
|
|
|
|
"title": "完成",
|
|
|
|
|
"items": [],
|
|
|
|
|
"listItems": [],
|
|
|
|
|
"sections": [],
|
|
|
|
|
"actions": [],
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
},
|
2026-03-16 16:11:06 +08:00
|
|
|
)
|
|
|
|
|
|
|
|
|
|
result = convert_message_to_history(message) # type: ignore[arg-type]
|
|
|
|
|
|
2026-04-23 12:12:41 +08:00
|
|
|
assert "ui_schema" in result
|
2026-03-16 16:11:06 +08:00
|
|
|
|
|
|
|
|
|
2026-04-23 12:12:41 +08:00
|
|
|
def test_convert_message_to_history_adds_suggested_actions_for_assistant_message() -> None:
|
2026-03-16 16:11:06 +08:00
|
|
|
message = _FakeMessage(
|
|
|
|
|
role="assistant",
|
|
|
|
|
metadata={
|
2026-04-23 12:12:41 +08:00
|
|
|
"agent_output": {
|
|
|
|
|
"suggested_actions": ["查今天日程", "创建会议"]
|
|
|
|
|
}
|
2026-03-16 16:11:06 +08:00
|
|
|
},
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
result = convert_message_to_history(message) # type: ignore[arg-type]
|
|
|
|
|
|
2026-04-23 12:12:41 +08:00
|
|
|
assert result["suggestedActions"] == ["查今天日程", "创建会议"]
|
2026-03-16 18:32:09 +08:00
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_convert_message_to_history_returns_multiple_user_attachments() -> None:
|
|
|
|
|
message = _FakeMessage(
|
|
|
|
|
role="user",
|
|
|
|
|
metadata={
|
|
|
|
|
"user_message_attachments": [
|
|
|
|
|
{"bucket": "bucket-a", "path": "path/a.png", "mime_type": "image/png"},
|
|
|
|
|
{"bucket": "bucket-a", "path": "path/b.jpg", "mime_type": "image/jpeg"},
|
|
|
|
|
]
|
|
|
|
|
},
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
def _signed(payload: dict[str, str]) -> str:
|
|
|
|
|
return f"https://signed.example/{payload['bucket']}/{payload['path']}"
|
|
|
|
|
|
|
|
|
|
result = convert_message_to_history(message, _signed) # type: ignore[arg-type]
|
|
|
|
|
|
|
|
|
|
attachments = result.get("attachments")
|
|
|
|
|
assert isinstance(attachments, list)
|
|
|
|
|
assert len(attachments) == 2
|
|
|
|
|
assert attachments[0]["mimeType"] == "image/png"
|
|
|
|
|
assert attachments[1]["mimeType"] == "image/jpeg"
|