Add WeWork XiaoAi TTS bot - WeChat Work long connection bridge

Receives messages from WeChat Work bot via WebSocket long connection
and speaks them through XiaoAi smart speaker TTS.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-03 13:52:04 +08:00
commit c7b8b01fe2
17 changed files with 762 additions and 0 deletions
+84
View File
@@ -0,0 +1,84 @@
import json
import pytest
from app.services.ws_client import _extract_text
def test_extract_text_text_message():
msg = {
"cmd": "aibot_msg_callback",
"headers": {"req_id": "req123"},
"body": {
"msgtype": "text",
"text": {"content": "你好世界"},
},
}
assert _extract_text(msg) == "你好世界"
def test_extract_text_voice_message():
msg = {
"cmd": "aibot_msg_callback",
"headers": {"req_id": "req123"},
"body": {
"msgtype": "voice",
"voice": {"content": "语音识别结果"},
},
}
assert _extract_text(msg) == "语音识别结果"
def test_extract_text_image_returns_none():
msg = {
"cmd": "aibot_msg_callback",
"body": {
"msgtype": "image",
"image": {"url": "https://example.com/pic.jpg"},
},
}
assert _extract_text(msg) is None
def test_extract_text_empty_content():
msg = {
"cmd": "aibot_msg_callback",
"body": {
"msgtype": "text",
"text": {"content": " "},
},
}
assert _extract_text(msg) is None
def test_extract_text_missing_body():
msg = {"cmd": "aibot_msg_callback"}
assert _extract_text(msg) is None
def test_subscribe_message_format():
"""Verify subscribe message matches documented format"""
msg = {
"cmd": "aibot_subscribe",
"headers": {"req_id": "test_req_id"},
"body": {
"bot_id": "test_bot_id",
"secret": "test_secret",
},
}
data = json.dumps(msg)
parsed = json.loads(data)
assert parsed["cmd"] == "aibot_subscribe"
assert parsed["body"]["bot_id"] == "test_bot_id"
assert parsed["body"]["secret"] == "test_secret"
def test_ping_message_format():
"""Verify ping message format matches documented format"""
msg = {
"cmd": "ping",
"headers": {"req_id": "test_req_id"},
}
data = json.dumps(msg)
parsed = json.loads(data)
assert parsed["cmd"] == "ping"
assert "req_id" in parsed["headers"]