mirror of
https://github.com/hansjone/oclaw.git
synced 2026-10-09 07:20:44 +08:00
oliver
1d46356d9e
保持入站到 agent 流程不变，在回送阶段支持 media_path/media_url 与 base64 附件直发，同时让通道回包沿用会话中助手附件记录以对齐 chat 历史逻辑。 Made-with: Cursor
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from oclaw.interfaces.http.fastapi_app import create_app
|
|
from oclaw.interfaces.http import fastapi_app
|
|
|
|
|
|
def test_inbound_whatsapp_accepts_basic_payload_and_returns_replies() -> None:
|
|
client = TestClient(create_app())
|
|
old_usecase = fastapi_app.process_inbound_payload_usecase
|
|
|
|
def _fake_usecase(payload: dict[str, object]) -> dict[str, object]:
|
|
assert str(payload.get("channel") or "") == "whatsapp"
|
|
assert str(payload.get("account_id") or "") == "wa-default"
|
|
assert str(payload.get("user_id") or "") == "111@s.whatsapp.net"
|
|
assert str(payload.get("chat_id") or "") == "111@s.whatsapp.net"
|
|
return {
|
|
"ok": True,
|
|
"replies": [
|
|
{
|
|
"chat_id": payload.get("chat_id"),
|
|
"text": "ok",
|
|
"attachments": [
|
|
{
|
|
"type": "binary_ref",
|
|
"name": "hello.txt",
|
|
"mime_type": "text/plain",
|
|
"data_base64": "aGVsbG8=",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
|
|
try:
|
|
fastapi_app.process_inbound_payload_usecase = _fake_usecase # type: ignore[assignment]
|
|
r = client.post(
|
|
"/inbound/whatsapp",
|
|
json={
|
|
"account_id": "wa-default",
|
|
"user_id": "111@s.whatsapp.net",
|
|
"chat_id": "111@s.whatsapp.net",
|
|
"text": "hello",
|
|
"metadata": {"source": "test"},
|
|
},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
data = r.json() or {}
|
|
assert data.get("ok") is True
|
|
replies = data.get("replies") if isinstance(data.get("replies"), list) else []
|
|
assert replies and isinstance(replies[0], dict)
|
|
assert str(replies[0].get("text") or "") == "ok"
|
|
atts = replies[0].get("attachments") if isinstance(replies[0].get("attachments"), list) else []
|
|
assert len(atts) == 1
|
|
assert str(atts[0].get("name") or "") == "hello.txt"
|
|
finally:
|
|
fastapi_app.process_inbound_payload_usecase = old_usecase # type: ignore[assignment]
|
|
|