houhuan
c7b8b01fe2
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>
85 lines
2.1 KiB
Python
85 lines
2.1 KiB
Python
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"]
|