test(wecomcs): cover bounded message IDs and images

Author: dadachann

src/langbot/libs/wecom_customer_service_api/api.py

@@ -295,6 +295,34 @@ async def send_text_msg(self, open_kfid: str, external_userid: str, msgid: str,
                 raise Exception('Failed to send message')
             return data
 
+    @_bounded_token_retry
+    async def send_image_msg(self, open_kfid: str, external_userid: str, msgid: str, media_id: str):
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+
+        url = f'{self.base_url}/kf/send_msg?access_token={self.access_token}'
+        payload = {
+            'touser': external_userid,
+            'open_kfid': open_kfid,
+            'msgid': msgid,
+            'msgtype': 'image',
+            'image': {
+                'media_id': media_id,
+            },
+        }
+
+        async with self._http_client_context() as client:
+            response = await client.post(url, json=payload)
+
+            data = await httpclient.parse_json_response(response)
+            if data['errcode'] == 40014 or data['errcode'] == 42001:
+                self.access_token = await self.get_access_token(self.secret)
+                return await self.send_image_msg(open_kfid, external_userid, msgid, media_id)
+            if data['errcode'] != 0:
+                await self.logger.error(f'发送图片失败：{data}')
+                raise Exception('Failed to send image message')
+            return data
+
     async def handle_callback_request(self):
         """处理回调请求（独立端口模式，使用全局 request）。"""
         return await self._handle_callback_internal(request)

tests/unit_tests/platform/test_wecomcs_adapter.py

@@ -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

tests/unit_tests/platform/test_wecomcs_api.py

@@ -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"}}'
+    )