mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-09-16 14:57:15 +00:00
Merge remote-tracking branch 'origin/master' into dev/4.11.x
# Conflicts: # src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py # src/langbot/pkg/api/http/service/bot.py # src/langbot/pkg/provider/runners/localagent.py # src/langbot/templates/metadata/pipeline/ai.yaml # tests/unit_tests/api/service/test_bot_service.py # tests/unit_tests/provider/runners/test_difysvapi_runner.py # tests/unit_tests/utils/test_safe_regex.py # web/src/app/infra/entities/adapter-categories.ts # web/src/app/wizard/page.tsx # web/src/i18n/locales/en-US.ts # web/src/i18n/locales/ja-JP.ts # web/src/i18n/locales/zh-Hans.ts # web/tests/e2e/plugin-page-auth.spec.ts
This commit is contained in:
Hyu
@@ -0,0 +1,259 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from linebot.v3.webhooks import TextMessageContent, UserMentionee, AllMentionee
|
||||
|
||||
from langbot.pkg.platform import botmgr as _botmgr # noqa: F401
|
||||
from langbot.pkg.platform.sources import line
|
||||
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
||||
|
||||
BOT_ACCOUNT_ID = 'line-bot-account'
|
||||
|
||||
|
||||
def _make_event(
|
||||
*, source_type: str, user_id, group_id=None, room_id=None, message_id: str, text: str = 'hi', mention=None
|
||||
):
|
||||
event = MagicMock()
|
||||
event.timestamp = 1700000000000
|
||||
message = MagicMock(spec=TextMessageContent)
|
||||
message.id = message_id
|
||||
message.text = text
|
||||
message.mention = mention
|
||||
event.message = message
|
||||
event.message.webhook_event_id = f'webhook-{message_id}'
|
||||
event.message.timestamp = event.timestamp
|
||||
|
||||
source = MagicMock()
|
||||
source.type = source_type
|
||||
source.user_id = user_id
|
||||
if group_id is not None:
|
||||
source.group_id = group_id
|
||||
if room_id is not None:
|
||||
source.room_id = room_id
|
||||
event.source = source
|
||||
|
||||
return event
|
||||
|
||||
|
||||
def _make_converter(bot_account_id: str = BOT_ACCOUNT_ID) -> line.LINEEventConverter:
|
||||
return line.LINEEventConverter(bot_account_id=bot_account_id)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_message_launcher_id_stable_across_messages() -> None:
|
||||
"""Two distinct messages from the same LINE user must resolve to the same
|
||||
sender id, otherwise every message starts a brand new session (context loss).
|
||||
"""
|
||||
converter = _make_converter()
|
||||
event1 = _make_event(source_type='user', user_id='U-stable-user', message_id='msg-1')
|
||||
event2 = _make_event(source_type='user', user_id='U-stable-user', message_id='msg-2')
|
||||
|
||||
result1 = await converter.target2yiri(event1, bot_client=None)
|
||||
result2 = await converter.target2yiri(event2, bot_client=None)
|
||||
|
||||
assert result1.sender.id == 'U-stable-user'
|
||||
assert result1.sender.id == result2.sender.id
|
||||
assert result1.sender.id != event1.message.id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_group_message_uses_group_id_not_message_id() -> None:
|
||||
converter = _make_converter()
|
||||
event1 = _make_event(source_type='group', user_id='U-member', group_id='G-stable-group', message_id='msg-1')
|
||||
event2 = _make_event(source_type='group', user_id='U-member', group_id='G-stable-group', message_id='msg-2')
|
||||
|
||||
result1 = await converter.target2yiri(event1, bot_client=None)
|
||||
result2 = await converter.target2yiri(event2, bot_client=None)
|
||||
|
||||
assert result1.sender.group.id == 'G-stable-group'
|
||||
assert result1.sender.group.id == result2.sender.group.id
|
||||
assert result1.sender.id == 'U-member'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_room_message_uses_room_id_and_falls_back_when_user_id_missing() -> None:
|
||||
converter = _make_converter()
|
||||
event = _make_event(source_type='room', user_id=None, room_id='R-stable-room', message_id='msg-1')
|
||||
|
||||
result = await converter.target2yiri(event, bot_client=None)
|
||||
|
||||
assert result.sender.group.id == 'R-stable-room'
|
||||
assert result.sender.id == 'R-stable-room'
|
||||
|
||||
|
||||
def _plain_texts(chain: platform_message.MessageChain) -> list[str]:
|
||||
return [c.text for c in chain if isinstance(c, platform_message.Plain)]
|
||||
|
||||
|
||||
def _ats(chain: platform_message.MessageChain) -> list[platform_message.At]:
|
||||
return [c for c in chain if isinstance(c, platform_message.At)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_mention_keeps_plain_text() -> None:
|
||||
converter = _make_converter()
|
||||
event = _make_event(source_type='group', user_id='U-member', group_id='G1', message_id='m1', text='hello world')
|
||||
|
||||
chain = await converter.message_converter.target2yiri(event, bot_client=None)
|
||||
|
||||
assert _plain_texts(chain) == ['hello world']
|
||||
assert _ats(chain) == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bot_mention_maps_to_at_with_bot_account_id() -> None:
|
||||
"""A @bot mention must become At(target=bot_account_id) so the 'at-bot'
|
||||
group respond rule matches (previously the mention was lost and the message
|
||||
was silently dropped in groups with at-only rules).
|
||||
"""
|
||||
mention = MagicMock()
|
||||
mention.mentionees = [
|
||||
UserMentionee(type='user', index=0, length=4, userId='U-bot-user-id', isSelf=True),
|
||||
]
|
||||
converter = _make_converter()
|
||||
event = _make_event(
|
||||
source_type='group',
|
||||
user_id='U-member',
|
||||
group_id='G1',
|
||||
message_id='m1',
|
||||
text='@BOT hey',
|
||||
mention=mention,
|
||||
)
|
||||
|
||||
chain = await converter.message_converter.target2yiri(event, bot_client=None)
|
||||
|
||||
ats = _ats(chain)
|
||||
assert len(ats) == 1
|
||||
assert ats[0].target == BOT_ACCOUNT_ID
|
||||
assert _plain_texts(chain) == [' hey']
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_other_user_mention_keeps_display_text() -> None:
|
||||
"""Mentions of other users keep their display text in the message string,
|
||||
so prefix/regexp rules that match the raw '@Name ...' text still work.
|
||||
"""
|
||||
mention = MagicMock()
|
||||
mention.mentionees = [
|
||||
UserMentionee(type='user', index=0, length=6, userId='U-other', isSelf=False),
|
||||
]
|
||||
converter = _make_converter()
|
||||
event = _make_event(
|
||||
source_type='group',
|
||||
user_id='U-member',
|
||||
group_id='G1',
|
||||
message_id='m1',
|
||||
text='@Alice hello',
|
||||
mention=mention,
|
||||
)
|
||||
|
||||
chain = await converter.message_converter.target2yiri(event, bot_client=None)
|
||||
|
||||
ats = _ats(chain)
|
||||
assert len(ats) == 1
|
||||
assert ats[0].target == 'U-other'
|
||||
# str() of the At component falls back to display when set
|
||||
assert str(chain) == '@Alice hello'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bot_mention_triggers_atbot_rule() -> None:
|
||||
"""End-to-end: a group message that @mentions the bot must be accepted by
|
||||
the at-bot respond rule (this is the regression that silently dropped
|
||||
'@bot' messages in LINE groups).
|
||||
"""
|
||||
from langbot.pkg.pipeline.resprule.rules.atbot import AtBotRule
|
||||
|
||||
mention = MagicMock()
|
||||
mention.mentionees = [
|
||||
UserMentionee(type='user', index=0, length=6, userId='U-bot-user-id', isSelf=True),
|
||||
]
|
||||
converter = _make_converter()
|
||||
event = _make_event(
|
||||
source_type='group',
|
||||
user_id='U-member',
|
||||
group_id='G1',
|
||||
message_id='m1',
|
||||
text='@RAIQt hi',
|
||||
mention=mention,
|
||||
)
|
||||
|
||||
chain = await converter.message_converter.target2yiri(event, bot_client=None)
|
||||
|
||||
query = MagicMock()
|
||||
query.adapter = MagicMock()
|
||||
query.adapter.bot_account_id = BOT_ACCOUNT_ID
|
||||
|
||||
rule = AtBotRule(ap=MagicMock())
|
||||
result = await rule.match(str(chain), chain, {'at': True}, query)
|
||||
|
||||
assert result.matching is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_group_without_bot_mention_still_dropped_by_atbot_rule() -> None:
|
||||
from langbot.pkg.pipeline.resprule.rules.atbot import AtBotRule
|
||||
|
||||
converter = _make_converter()
|
||||
event = _make_event(source_type='group', user_id='U-member', group_id='G1', message_id='m1', text='hello')
|
||||
|
||||
chain = await converter.message_converter.target2yiri(event, bot_client=None)
|
||||
|
||||
query = MagicMock()
|
||||
query.adapter = MagicMock()
|
||||
query.adapter.bot_account_id = BOT_ACCOUNT_ID
|
||||
|
||||
rule = AtBotRule(ap=MagicMock())
|
||||
result = await rule.match(str(chain), chain, {'at': True}, query)
|
||||
|
||||
assert result.matching is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_at_all_mention_preserved_as_at_component() -> None:
|
||||
mention = MagicMock()
|
||||
mention.mentionees = [
|
||||
AllMentionee(type='all', index=0, length=4),
|
||||
]
|
||||
converter = _make_converter()
|
||||
event = _make_event(
|
||||
source_type='group',
|
||||
user_id='U-member',
|
||||
group_id='G1',
|
||||
message_id='m1',
|
||||
text='@All hello',
|
||||
mention=mention,
|
||||
)
|
||||
|
||||
chain = await converter.message_converter.target2yiri(event, bot_client=None)
|
||||
|
||||
ats = _ats(chain)
|
||||
assert len(ats) == 1
|
||||
assert str(chain) == '@All hello'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_mentions_sorted_by_position() -> None:
|
||||
mention = MagicMock()
|
||||
# Intentionally out of order to exercise sorting
|
||||
mention.mentionees = [
|
||||
UserMentionee(type='user', index=9, length=4, userId='U-b', isSelf=False),
|
||||
UserMentionee(type='user', index=0, length=4, userId='U-a', isSelf=False),
|
||||
]
|
||||
converter = _make_converter()
|
||||
event = _make_event(
|
||||
source_type='group',
|
||||
user_id='U-member',
|
||||
group_id='G1',
|
||||
message_id='m1',
|
||||
text='@aaa mid @bbb tail',
|
||||
mention=mention,
|
||||
)
|
||||
|
||||
chain = await converter.message_converter.target2yiri(event, bot_client=None)
|
||||
|
||||
ats = _ats(chain)
|
||||
assert [a.target for a in ats] == ['U-a', 'U-b']
|
||||
assert str(chain) == '@aaa mid @bbb tail'
|
||||
@@ -1,9 +1,11 @@
|
||||
"""Tests for QQ Official keyboard payload helpers."""
|
||||
"""Tests for QQ Official message and keyboard payload helpers."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import time
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
||||
@@ -99,6 +101,12 @@ def _stream_test_adapter():
|
||||
adapter.bot = MagicMock()
|
||||
adapter.bot.send_stream_msg = AsyncMock(return_value={'id': 'stream-1'})
|
||||
adapter.bot.send_markdown_keyboard = AsyncMock(return_value={'id': 'message-1'})
|
||||
adapter.bot.send_private_text_msg = AsyncMock()
|
||||
adapter.bot.send_group_text_msg = AsyncMock()
|
||||
adapter.bot.send_private_markdown_msg = AsyncMock()
|
||||
adapter.bot.send_group_markdown_msg = AsyncMock()
|
||||
adapter.bot.send_channle_group_text_msg = AsyncMock()
|
||||
adapter.bot.send_channle_private_text_msg = AsyncMock()
|
||||
adapter.ap = None
|
||||
adapter._stream_ctx = {}
|
||||
adapter._stream_ctx_ts = {}
|
||||
@@ -108,7 +116,7 @@ def _stream_test_adapter():
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_qq_stream_uses_cumulative_chunks_as_snapshots():
|
||||
async def test_qq_stream_replace_mode_sends_complete_snapshots():
|
||||
adapter = _stream_test_adapter()
|
||||
adapter._stream_ctx['message-1'] = {
|
||||
'user_openid': 'user-1',
|
||||
@@ -138,10 +146,109 @@ async def test_qq_stream_uses_cumulative_chunks_as_snapshots():
|
||||
|
||||
assert [call.kwargs['content'] for call in adapter.bot.send_stream_msg.await_args_list] == [
|
||||
'<think>one',
|
||||
' two',
|
||||
'<think>one two',
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_qq_markdown_messages_use_markdown_payloads():
|
||||
requests = []
|
||||
|
||||
def capture_request(request: httpx.Request) -> httpx.Response:
|
||||
requests.append((str(request.url), json.loads(request.content)))
|
||||
return httpx.Response(200, json={})
|
||||
|
||||
client = QQOfficialClient('secret', 'token', 'app-id', AsyncMock())
|
||||
client.access_token = 'access-token'
|
||||
client.access_token_expiry_time = time.time() + 3600
|
||||
client._http_clients[None] = httpx.AsyncClient(transport=httpx.MockTransport(capture_request))
|
||||
|
||||
try:
|
||||
await client.send_private_markdown_msg('user-1', '# Hello', msg_id='message-1', msg_seq=2)
|
||||
await client.send_group_markdown_msg('group-1', '* Hello', event_id='event-1', msg_seq=3)
|
||||
finally:
|
||||
await client.close()
|
||||
|
||||
assert requests == [
|
||||
(
|
||||
'https://api.sgroup.qq.com/v2/users/user-1/messages',
|
||||
{'msg_type': 2, 'markdown': {'content': '# Hello'}, 'msg_seq': 2, 'msg_id': 'message-1'},
|
||||
),
|
||||
(
|
||||
'https://api.sgroup.qq.com/v2/groups/group-1/messages',
|
||||
{'msg_type': 2, 'markdown': {'content': '* Hello'}, 'msg_seq': 3, 'event_id': 'event-1'},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_qq_markdown_rendering_switches_c2c_and_group_text_replies():
|
||||
adapter = _stream_test_adapter()
|
||||
adapter.config = {'enable-markdown-rendering': True}
|
||||
|
||||
await adapter._send_c2c_or_group_text_reply('c2c', 'user-1', '# Hello', msg_id='message-1')
|
||||
await adapter._send_c2c_or_group_text_reply('group', 'group-1', '* Hello', event_id='event-1')
|
||||
|
||||
adapter.bot.send_private_markdown_msg.assert_awaited_once_with(
|
||||
user_openid='user-1',
|
||||
content='# Hello',
|
||||
msg_id='message-1',
|
||||
event_id=None,
|
||||
msg_seq=1,
|
||||
)
|
||||
adapter.bot.send_group_markdown_msg.assert_awaited_once_with(
|
||||
group_openid='group-1',
|
||||
content='* Hello',
|
||||
msg_id=None,
|
||||
event_id='event-1',
|
||||
msg_seq=1,
|
||||
)
|
||||
adapter.bot.send_private_text_msg.assert_not_awaited()
|
||||
adapter.bot.send_group_text_msg.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_qq_markdown_rendering_defaults_to_plain_text_replies():
|
||||
adapter = _stream_test_adapter()
|
||||
adapter.config = {}
|
||||
|
||||
await adapter._send_c2c_or_group_text_reply('c2c', 'user-1', 'Hello')
|
||||
await adapter._send_c2c_or_group_text_reply('group', 'group-1', 'Hello')
|
||||
|
||||
adapter.bot.send_private_text_msg.assert_awaited_once()
|
||||
adapter.bot.send_group_text_msg.assert_awaited_once()
|
||||
adapter.bot.send_private_markdown_msg.assert_not_awaited()
|
||||
adapter.bot.send_group_markdown_msg.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_qq_markdown_rendering_does_not_affect_channel_messages():
|
||||
adapter = _stream_test_adapter()
|
||||
adapter.config = {'enable-markdown-rendering': True}
|
||||
message = platform_message.MessageChain([platform_message.Plain(text='# Hello')])
|
||||
|
||||
channel_source = MagicMock()
|
||||
channel_source.t = 'AT_MESSAGE_CREATE'
|
||||
channel_source.channel_id = 'channel-1'
|
||||
channel_source.d_id = 'message-1'
|
||||
channel_event = MagicMock()
|
||||
channel_event.source_platform_object = channel_source
|
||||
await adapter.reply_message(channel_event, message)
|
||||
|
||||
dm_source = MagicMock()
|
||||
dm_source.t = 'DIRECT_MESSAGE_CREATE'
|
||||
dm_source.guild_id = 'guild-1'
|
||||
dm_source.d_id = 'message-2'
|
||||
dm_event = MagicMock()
|
||||
dm_event.source_platform_object = dm_source
|
||||
await adapter.reply_message(dm_event, message)
|
||||
|
||||
adapter.bot.send_channle_group_text_msg.assert_awaited_once_with('channel-1', '# Hello', 'message-1')
|
||||
adapter.bot.send_channle_private_text_msg.assert_awaited_once_with('guild-1', '# Hello', 'message-2')
|
||||
adapter.bot.send_private_markdown_msg.assert_not_awaited()
|
||||
adapter.bot.send_group_markdown_msg.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_qq_non_streaming_fallback_keeps_latest_snapshot_only():
|
||||
from langbot.pkg.platform.sources.qqofficial import QQOfficialAdapter
|
||||
|
||||
@@ -44,6 +44,86 @@ def test_webhook_dispatch_tasks_are_bounded():
|
||||
assert len(client._dispatch_tasks) == 100
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ws_initial_stream_frame_precedes_pipeline_dispatch(monkeypatch):
|
||||
from langbot.libs.wecom_ai_bot_api import ws_client as ws_client_module
|
||||
|
||||
order = []
|
||||
logger = types.SimpleNamespace(
|
||||
debug=Mock(),
|
||||
error=Mock(),
|
||||
warning=Mock(),
|
||||
)
|
||||
client = WecomBotWsClient('bot-id', 'secret', logger)
|
||||
|
||||
async def parse_message(*args, **kwargs):
|
||||
del args, kwargs
|
||||
return {'msgid': 'msg-1', 'type': 'single', 'userid': 'user-1'}
|
||||
|
||||
async def reply_stream(*args, **kwargs):
|
||||
del args, kwargs
|
||||
order.append('initial-frame')
|
||||
return {}
|
||||
|
||||
async def dispatch_event(event):
|
||||
del event
|
||||
order.append('pipeline-dispatch')
|
||||
|
||||
monkeypatch.setattr(ws_client_module, 'parse_wecom_bot_message', parse_message)
|
||||
monkeypatch.setattr(ws_client_module.wecombotevent, 'WecomBotEvent', lambda data: data)
|
||||
client.reply_stream = reply_stream
|
||||
client._dispatch_event = dispatch_event
|
||||
|
||||
await client._handle_message_callback({'headers': {'req_id': 'req-1'}, 'body': {}})
|
||||
|
||||
assert order == ['initial-frame', 'pipeline-dispatch']
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ws_initial_stream_failure_still_dispatches_message(monkeypatch):
|
||||
from langbot.libs.wecom_ai_bot_api import ws_client as ws_client_module
|
||||
|
||||
dispatched = []
|
||||
|
||||
class Logger:
|
||||
def __init__(self):
|
||||
self.warnings = []
|
||||
|
||||
async def debug(self, message):
|
||||
del message
|
||||
|
||||
async def error(self, message):
|
||||
raise AssertionError(message)
|
||||
|
||||
async def warning(self, message):
|
||||
self.warnings.append(message)
|
||||
|
||||
logger = Logger()
|
||||
client = WecomBotWsClient('bot-id', 'secret', logger)
|
||||
|
||||
async def parse_message(*args, **kwargs):
|
||||
del args, kwargs
|
||||
return {'msgid': 'msg-1', 'type': 'single', 'userid': 'user-1'}
|
||||
|
||||
async def reply_stream(*args, **kwargs):
|
||||
del args, kwargs
|
||||
raise ConnectionError('simulated reply failure')
|
||||
|
||||
async def dispatch_event(event):
|
||||
dispatched.append(event)
|
||||
|
||||
monkeypatch.setattr(ws_client_module, 'parse_wecom_bot_message', parse_message)
|
||||
monkeypatch.setattr(ws_client_module.wecombotevent, 'WecomBotEvent', lambda data: data)
|
||||
client.reply_stream = reply_stream
|
||||
client._dispatch_event = dispatch_event
|
||||
|
||||
await client._handle_message_callback({'headers': {'req_id': 'req-1'}, 'body': {}})
|
||||
|
||||
assert len(dispatched) == 1
|
||||
assert len(logger.warnings) == 1
|
||||
assert 'simulated reply failure' in logger.warnings[0]
|
||||
|
||||
|
||||
def test_extract_template_card_action_supports_nested_button_key():
|
||||
task_id, event_key, card_type = extract_template_card_action(
|
||||
{
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
@@ -49,7 +50,29 @@ async def test_send_message_sends_text_to_customer_service_user():
|
||||
assert kwargs['open_kfid'] == 'kf-test'
|
||||
assert kwargs['external_userid'] == 'external-user'
|
||||
assert kwargs['content'] == 'hello'
|
||||
assert kwargs['msgid'].startswith('langbot_')
|
||||
assert len(kwargs['msgid'].encode()) <= 32
|
||||
assert uuid.UUID(hex=kwargs['msgid']).hex == kwargs['msgid']
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_message_sends_image_to_customer_service_user():
|
||||
adapter = make_adapter()
|
||||
adapter.bot_account_id = 'kf-test'
|
||||
adapter.bot = SimpleNamespace(
|
||||
get_media_id=AsyncMock(return_value='media-id'),
|
||||
send_image_msg=AsyncMock(),
|
||||
)
|
||||
|
||||
message = platform_message.MessageChain([platform_message.Image(base64='aW1hZ2U=')])
|
||||
|
||||
await adapter.send_message('person', 'uexternal-user', message)
|
||||
|
||||
adapter.bot.send_image_msg.assert_awaited_once()
|
||||
kwargs = adapter.bot.send_image_msg.await_args.kwargs
|
||||
assert kwargs['open_kfid'] == 'kf-test'
|
||||
assert kwargs['external_userid'] == 'external-user'
|
||||
assert kwargs['media_id'] == 'media-id'
|
||||
assert len(kwargs['msgid'].encode()) <= 32
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from langbot.libs.wecom_customer_service_api.api import WecomCSClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_image_msg_posts_customer_service_image_payload() -> None:
|
||||
captured_request: httpx.Request | None = None
|
||||
|
||||
def handle_request(request: httpx.Request) -> httpx.Response:
|
||||
nonlocal captured_request
|
||||
captured_request = request
|
||||
return httpx.Response(200, json={'errcode': 0})
|
||||
|
||||
client = WecomCSClient(
|
||||
corpid='corp-id',
|
||||
secret='secret',
|
||||
token='token',
|
||||
EncodingAESKey='encoding-key',
|
||||
logger=None,
|
||||
unified_mode=True,
|
||||
)
|
||||
client.access_token = 'access-token'
|
||||
client._http_client = httpx.AsyncClient(transport=httpx.MockTransport(handle_request))
|
||||
|
||||
try:
|
||||
await client.send_image_msg(
|
||||
open_kfid='kf-test',
|
||||
external_userid='external-user',
|
||||
msgid='a' * 32,
|
||||
media_id='media-id',
|
||||
)
|
||||
finally:
|
||||
await client.close()
|
||||
|
||||
assert captured_request is not None
|
||||
assert captured_request.url.path == '/cgi-bin/kf/send_msg'
|
||||
assert captured_request.url.params['access_token'] == 'access-token'
|
||||
assert captured_request.method == 'POST'
|
||||
assert captured_request.read().decode() == (
|
||||
'{"touser":"external-user","open_kfid":"kf-test","msgid":"'
|
||||
+ 'a' * 32
|
||||
+ '","msgtype":"image","image":{"media_id":"media-id"}}'
|
||||
)
|
||||
Reference in New Issue
Block a user