✅ Add test for Telegram Middleware

Author: j1g5awi

nonebot_plugin_all4one/middlewares/onebot/v12.py

@@ -89,7 +89,7 @@ async def send_message(
                 message=message,
                 **kwargs,
             )
-        return self.bot.send_message(
+        return await self.bot.send_message(
             detail_type=detail_type, message=message, **kwargs
         )  # type:ignore
 

nonebot_plugin_all4one/middlewares/telegram.py

@@ -1,7 +1,6 @@
 from pathlib import Path
 from typing import Any, Dict, List, Union, Literal, Optional
 
-from httpx import AsyncClient
 from pydantic import parse_obj_as
 from nonebot.adapters.onebot.v12 import UnsupportedSegment
 from nonebot.adapters.telegram.message import File, Entity
@@ -124,13 +123,14 @@ async def to_onebot_message(self, message: Message) -> OneBotMessage:
         for segment in message_list:
             if segment.type in ("image", "voice", "audio", "video", "file"):
                 file = await self.bot.get_file(segment.data["file_id"])
-                async with AsyncClient() as client:
-                    segment.data["file_id"] = await upload_file(
-                        Path(file.file_path).name,  # type:ignore
-                        self.get_platform(),
-                        file.file_id,
-                        url=f"https://api.telegram.org/file/bot{self.bot.bot_config.token}/{file.file_path}",
-                    )
+                if file.file_path is None:
+                    continue
+                segment.data["file_id"] = await upload_file(
+                    Path(file.file_path).name,
+                    self.get_platform(),
+                    file.file_id,
+                    url=f"https://api.telegram.org/file/bot{self.bot.bot_config.token}/{file.file_path}",
+                )
         return OneBotMessage(message_list)
 
     async def send(

tests/qqguild/test_event.py tests/qqguild/test_qqguild.pytests/qqguild/test_event.py renamed to tests/qqguild/test_qqguild.py

@@ -0,0 +1,60 @@
+import json
+from pathlib import Path
+
+from nonebug import App
+from nonebot.adapters.telegram import Bot, Event
+from nonebot.adapters.telegram.model import File
+from nonebot.adapters.telegram.config import BotConfig
+from nonebot.adapters.onebot.v12 import PrivateMessageEvent
+
+bot_config = BotConfig(token="1234567890:ABCDEFGHIJKLMNOPQRSTUVWXYZABCDEFGHI")
+
+
+async def test_to_onebot_event(app: App):
+    from nonebot_plugin_all4one.middlewares.telegram import Middleware
+
+    with (Path(__file__).parent / "updates.json").open("r", encoding="utf8") as f:
+        test_updates = json.load(f)
+
+    async with app.test_api() as ctx:
+        bot = ctx.create_bot(base=Bot, self_id=bot_config)  # type:ignore
+        middleware = Middleware(bot)
+
+        event = Event.parse_event(test_updates[0])
+        event = await middleware.to_onebot_event(event)
+        assert isinstance(event[0], PrivateMessageEvent)
+
+        event = Event.parse_event(test_updates[3])
+        event = await middleware.to_onebot_event(event)
+        assert isinstance(event[0], PrivateMessageEvent)
+        assert event[0].message[0].type == "reply"
+
+        event = Event.parse_event(test_updates[6])
+        ctx.should_call_api(
+            "get_file",
+            {"file_id": "AwADBAADbXXXXXXXXXXXGBdhD2l6_XX"},
+            File(file_id="AwADBAADbXXXXXXXXXXXGBdhD2l6_XX", file_unique_id=""),
+        )
+        event = await middleware.to_onebot_event(event)
+        assert isinstance(event[0], PrivateMessageEvent)
+        assert event[0].message[0].type == "audio"
+
+        event = Event.parse_event(test_updates[7])
+        ctx.should_call_api(
+            "get_file",
+            {"file_id": "AwADBAADbXXXXXXXXXXXGBdhD2l6_XX"},
+            File(file_id="AwADBAADbXXXXXXXXXXXGBdhD2l6_XX", file_unique_id=""),
+        )
+        event = await middleware.to_onebot_event(event)
+        assert isinstance(event[0], PrivateMessageEvent)
+        assert event[0].message[0].type == "voice"
+
+        event = Event.parse_event(test_updates[8])
+        ctx.should_call_api(
+            "get_file",
+            {"file_id": "AwADBAADbXXXXXXXXXXXGBdhD2l6_XX"},
+            File(file_id="AwADBAADbXXXXXXXXXXXGBdhD2l6_XX", file_unique_id=""),
+        )
+        event = await middleware.to_onebot_event(event)
+        assert isinstance(event[0], PrivateMessageEvent)
+        assert event[0].message[0].type == "file"