feat(aiocqhttp): 优化文件上传处理逻辑#8466
Open
NayukiChiba wants to merge 2 commits into
Open
Conversation
- 优先使用 NapCat 的 upload API 上传文件，上传成功后自动发送 - 增加异常处理，上传失败时回退到直接发送 - 提升文件发送的可靠性和用户体验
dosubot
Bot
added
size:S
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces a feature to prioritize uploading files using NapCat's upload_group_file and upload_private_file APIs when sending file segments in the aiocqhttp platform, falling back to direct sending if the upload fails. It also adds comprehensive unit tests to cover these scenarios. The feedback suggests checking if the file exists locally and converting it to an absolute path before uploading, as well as downgrading the error log to a warning during fallback to prevent log spam on unsupported clients.
Comment on lines
+160
to
+182
| file_path = seg.file | ||
| if file_path and session_id and session_id.isdigit(): | ||
| sid = int(session_id) | ||
| try: | ||
| if is_group: | ||
| await bot.call_action( | ||
| "upload_group_file", | ||
| group_id=sid, | ||
| file=file_path, | ||
| name=seg.name or "file", | ||
| ) | ||
| else: | ||
| await bot.call_action( | ||
| "upload_private_file", | ||
| user_id=sid, | ||
| file=file_path, | ||
| name=seg.name or "file", | ||
| ) | ||
| continue # 上传成功，upload API 自带发送功能 | ||
| except Exception as e: | ||
| from astrbot.api import logger | ||
|
|
||
| logger.error(f"上传文件失败，回退到直接发送: {e}") |
Contributor
There was a problem hiding this comment.
这里有几点可以优化：
-
路径存在性与绝对路径转换：
- 建议在调用
upload_group_file/upload_private_file之前，先使用os.path.exists(file_path)检查文件是否在本地存在。这样可以避免对网络 URL 或不存在的文件路径进行无意义的上传 API 调用。 - 建议使用
os.path.abspath(file_path)将文件路径转换为绝对路径。因为 OneBot 客户端（如 NapCat）可能运行在不同的工作目录下，使用相对路径可能会导致其无法正确找到并上传文件。
- 建议在调用
-
日志级别调整：
- 当前在上传失败回退到直接发送时，使用了
logger.error。由于对于不支持这些自定义上传 API 的 OneBot 客户端，每次发送文件都会触发此回退逻辑，使用logger.error会导致日志中充斥大量的“错误”信息。建议将其降级为logger.warning。
- 当前在上传失败回退到直接发送时，使用了
-
重构与测试（根据项目规范）：
- 避免代码重复：在实现类似功能（如直接发送与引用发送附件）时，建议将逻辑重构为共享的辅助函数（helper function），以避免代码重复。
- 单元测试：新增的附件处理功能应当伴随相应的单元测试，以确保其稳定性和正确性。
file_path = seg.file
import os
if file_path and os.path.exists(file_path) and session_id and session_id.isdigit():
file_path = os.path.abspath(file_path)
sid = int(session_id)
try:
if is_group:
await bot.call_action(
"upload_group_file",
group_id=sid,
file=file_path,
name=seg.name or "file",
)
else:
await bot.call_action(
"upload_private_file",
user_id=sid,
file=file_path,
name=seg.name or "file",
)
continue
except Exception as e:
from astrbot.api import logger
logger.warning(f"上传文件失败，回退到直接发送: {e}")References
- When implementing similar functionality for different cases (e.g., direct vs. quoted attachments), refactor the logic into a shared helper function to avoid code duplication.
- New functionality, such as handling attachments, should be accompanied by corresponding unit tests.
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 4 issues, and left some high level feedback:
- Consider catching a more specific exception than
Exceptionin the upload path (or at least usinglogger.exception) so that unexpected errors surface clearly while still logging a full traceback for genuine upload failures. - The
from astrbot.api import loggerinside the exception block will execute on every failure; if circular imports allow, moving this to module scope and reusing a module-level logger would avoid repeated imports in a hot path.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Consider catching a more specific exception than `Exception` in the upload path (or at least using `logger.exception`) so that unexpected errors surface clearly while still logging a full traceback for genuine upload failures.
- The `from astrbot.api import logger` inside the exception block will execute on every failure; if circular imports allow, moving this to module scope and reusing a module-level logger would avoid repeated imports in a hot path.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/aiocqhttp/aiocqhttp_message_event.py" line_range="179-182" />
<code_context>
+ file_path = seg.file
+ if file_path and session_id and session_id.isdigit():
+ sid = int(session_id)
+ try:
+ if is_group:
+ await bot.call_action(
+ "upload_group_file",
+ group_id=sid,
+ file=file_path,
+ name=seg.name or "file",
+ )
+ else:
+ await bot.call_action(
+ "upload_private_file",
+ user_id=sid,
+ file=file_path,
+ name=seg.name or "file",
+ )
+ continue # 上传成功，upload API 自带发送功能
+ except Exception as e:
+ from astrbot.api import logger
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Catching bare `Exception` may hide specific upload failures and stack traces.
Catching a broad `Exception` and only logging `e` drops the stack trace and obscures where the upload failed. Please narrow this to the expected exception type (e.g., the client’s error base class), or at least log the full traceback with something like `logger.exception` so failures remain diagnosable while still allowing the fallback path.
```suggestion
except Exception:
from astrbot.api import logger
logger.exception("上传文件失败，回退到直接发送")
```
</issue_to_address>
### Comment 2
<location path="tests/unit/test_aiocqhttp_file_message.py" line_range="20-29" />
<code_context>
+ assert "file:///" in str(fallbackMessage[0]["data"]["file"])
+
+
+@pytest.mark.asyncio
+async def test_no_local_file_falls_back_to_dispatch_send():
+ """无本地文件（纯 URL）时回退到 _from_segment_to_dict + _dispatch_send。"""
+ fileComp = Comp.File(name="remote.txt", url="https://example.com/remote.txt")
+ bot = MagicMock()
+ bot.send_group_msg = AsyncMock()
+ bot.send_private_msg = AsyncMock()
+ bot.call_action = AsyncMock()
+
+ await AiocqhttpMessageEvent.send_message(
+ bot=bot,
+ message_chain=MessageChain([fileComp]),
+ is_group=True,
+ session_id="123456",
+ )
+
+ # 无本地文件路径，不应调用 upload API
+ bot.call_action.assert_not_called()
+ # 回退到 send_group_msg
+ bot.send_group_msg.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_no_session_id_falls_back_to_dispatch_send(tmp_path):
+ """session_id 为 None 时回退到 _dispatch_send（会抛 ValueError）。"""
+ testFile = tmp_path / "test.txt"
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test for non-digit session_id to cover the `session_id.isdigit()` guard.
Currently we only cover `session_id=None`. Please also add a case where `session_id` is a non-empty, non-numeric string (e.g. `"abc123"`) to confirm the upload path is skipped and `_dispatch_send` is used instead. For example, a `test_non_digit_session_id_falls_back_to_dispatch_send` (for both group and private, if applicable) that:
- passes a `File` segment with `session_id="abc123"`;
- asserts `bot.call_action` is not called;
- asserts the appropriate fallback send path (`send_group_msg` / `send_private_msg`) or the expected exception from `_dispatch_send`.
Suggested implementation:
```python
bot.call_action.assert_not_called()
# 回退到 send_group_msg
bot.send_group_msg.assert_called_once()
@pytest.mark.asyncio
async def test_non_digit_session_id_group_falls_back_to_dispatch_send(tmp_path):
"""群聊：session_id 为非数字字符串时应回退到 _dispatch_send，不调用 upload API。"""
testFile = tmp_path / "test.txt"
testFile.write_text("hello world", encoding="utf-8")
fileComp = Comp.File(name="test.txt", file=str(testFile))
bot = MagicMock()
bot.send_group_msg = AsyncMock()
bot.send_private_msg = AsyncMock()
bot.call_action = AsyncMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([fileComp]),
is_group=True,
session_id="abc123",
)
# 非数字 session_id，上传 API 应被跳过
bot.call_action.assert_not_called()
# 回退到 send_group_msg
bot.send_group_msg.assert_called_once()
bot.send_private_msg.assert_not_called()
@pytest.mark.asyncio
async def test_non_digit_session_id_private_falls_back_to_dispatch_send(tmp_path):
"""私聊：session_id 为非数字字符串时应回退到 _dispatch_send，不调用 upload API。"""
testFile = tmp_path / "test.txt"
testFile.write_text("hello world", encoding="utf-8")
fileComp = Comp.File(name="test.txt", file=str(testFile))
bot = MagicMock()
bot.send_group_msg = AsyncMock()
bot.send_private_msg = AsyncMock()
bot.call_action = AsyncMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([fileComp]),
is_group=False,
session_id="abc123",
)
# 非数字 session_id，上传 API 应被跳过
bot.call_action.assert_not_called()
# 回退到 send_private_msg
bot.send_private_msg.assert_called_once()
bot.send_group_msg.assert_not_called()
```
These edits assume that:
1. The snippet with `bot.call_action.assert_not_called()` and `bot.send_group_msg.assert_called_once()` belongs to `test_no_local_file_falls_back_to_dispatch_send`, and that `pytest`, `MagicMock`, `AsyncMock`, `Comp`, `MessageChain`, and `AiocqhttpMessageEvent` are already imported (as shown in the partial file).
2. No additional fixtures or helpers are required for these tests; if your existing tests use shared fixtures for `bot` construction or message setup, you may want to refactor these new tests to reuse those fixtures instead of instantiating `MagicMock`/`AsyncMock` directly.
</issue_to_address>
### Comment 3
<location path="tests/unit/test_aiocqhttp_file_message.py" line_range="185-45" />
<code_context>
+async def test_multiple_files_each_uses_upload_api(tmp_path):
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding a private-chat variant for multiple-file uploads.
You already verify that each file in a group chat triggers a separate `upload_group_file` call. To fully exercise the new behavior, please add an analogous private-chat test (e.g. `test_multiple_files_each_uses_upload_private_api`) that checks:
- `bot.call_action` is invoked once per file with `"upload_private_file"` and the expected `user_id`/`file`/`name`.
- `send_private_msg` is never called.
This keeps group and private flows symmetric and protects against regressions in one path only.
Suggested implementation:
```python
@pytest.mark.asyncio
async def test_multiple_files_each_uses_upload_api(tmp_path):
"""多个 File 段：每个都独立调用 group upload API。"""
file1 = tmp_path / "a.txt"
file2 = tmp_path / "b.txt"
file1.write_text("aaa", encoding="utf-8")
file2.write_text("bbb", encoding="utf-8")
comp1 = Comp.File(name="a.txt", file=str(file1))
comp2 = Comp.File(name="b.txt", file=str(file2))
bot = MagicMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([comp1, comp2]),
is_group=True,
session_id="123456",
)
assert bot.call_action.call_count == 2
bot.call_action.assert_any_call(
"upload_group_file",
group_id="123456",
file=str(file1),
name="a.txt",
)
bot.call_action.assert_any_call(
"upload_group_file",
group_id="123456",
file=str(file2),
name="b.txt",
)
# 发送文件时不应直接调用发送消息接口
assert not bot.send_group_msg.called
@pytest.mark.asyncio
async def test_multiple_files_each_uses_upload_private_api(tmp_path):
"""多个 File 段：每个都独立调用 private upload API。"""
file1 = tmp_path / "a.txt"
file2 = tmp_path / "b.txt"
file1.write_text("aaa", encoding="utf-8")
file2.write_text("bbb", encoding="utf-8")
comp1 = Comp.File(name="a.txt", file=str(file1))
comp2 = Comp.File(name="b.txt", file=str(file2))
bot = MagicMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([comp1, comp2]),
is_group=False,
user_id=123456,
session_id="123456",
)
assert bot.call_action.call_count == 2
bot.call_action.assert_any_call(
"upload_private_file",
user_id=123456,
file=str(file1),
name="a.txt",
)
bot.call_action.assert_any_call(
"upload_private_file",
user_id=123456,
file=str(file2),
name="b.txt",
)
# 发送文件时不应直接调用发送私聊消息接口
assert not bot.send_private_msg.called
```
1. The exact arguments to `AiocqhttpMessageEvent.send_message` (e.g. `group_id`, `user_id`, `session_id`) and the `upload_group_file`/`upload_private_file` call signatures should be aligned with the rest of this test file. You may need to adjust `group_id="123456"`/`user_id=123456`/`session_id="123456"` to match existing conventions.
2. If `ANY` from `unittest.mock` is preferred in your tests (to avoid hard-coding IDs), you can replace `123456` with `ANY` and ensure `from unittest.mock import ANY` is present at the top of the file.
3. If the existing `test_multiple_files_each_uses_upload_api` already has assertions beyond the snippet you provided, merge the new body carefully to avoid duplicating or conflicting assertions.
</issue_to_address>
### Comment 4
<location path="tests/unit/test_aiocqhttp_file_message.py" line_range="50-46" />
<code_context>
+
+
+@pytest.mark.asyncio
+async def test_file_private_uses_upload_private_file_api(tmp_path):
+ """私聊发送本地文件：应调用 upload_private_file API，不走 send_private_msg。"""
+ testFile = tmp_path / "doc.pdf"
+ testFile.write_text("pdf content", encoding="utf-8")
+
+ fileComp = Comp.File(name="doc.pdf", file=str(testFile))
+ bot = MagicMock()
+ bot.send_group_msg = AsyncMock()
+ bot.send_private_msg = AsyncMock()
+ bot.call_action = AsyncMock()
+
+ await AiocqhttpMessageEvent.send_message(
+ bot=bot,
+ message_chain=MessageChain([fileComp]),
+ is_group=False,
+ session_id="789012",
+ )
+
+ bot.call_action.assert_called_once_with(
+ "upload_private_file",
+ user_id=789012,
+ file=str(testFile.resolve()),
+ name="doc.pdf",
+ )
+ bot.send_private_msg.assert_not_called()
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test for defaulting `name` to 'file' when `seg.name` is missing.
Right now all tests pass a `File` with an explicit `name`. Since the implementation uses `seg.name or "file"`, please add a test (e.g. `test_file_without_name_defaults_to_file`) that:
- creates `Comp.File(file=str(testFile))` without a `name`;
- calls `send_message` for group/private;
- asserts `bot.call_action` is invoked with `name="file"`.
This ensures the default-name behavior is covered and guarded against regressions.
Suggested implementation:
```python
bot.send_group_msg.assert_not_called()
bot.send_private_msg.assert_not_called()
@pytest.mark.asyncio
async def test_file_group_without_name_defaults_to_file(tmp_path):
"""群聊发送本地文件且未提供 name：应使用默认 name='file' 并调用 upload_group_file。"""
testFile = tmp_path / "test.txt"
testFile.write_text("hello world", encoding="utf-8")
fileComp = Comp.File(file=str(testFile))
bot = MagicMock()
bot.send_group_msg = AsyncMock()
bot.send_private_msg = AsyncMock()
bot.call_action = AsyncMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([fileComp]),
is_group=True,
session_id="123456",
)
bot.call_action.assert_called_once_with(
"upload_group_file",
group_id=123456,
file=str(testFile.resolve()),
name="file",
)
bot.send_group_msg.assert_not_called()
bot.send_private_msg.assert_not_called()
@pytest.mark.asyncio
async def test_file_private_without_name_defaults_to_file(tmp_path):
"""私聊发送本地文件且未提供 name：应使用默认 name='file' 并调用 upload_private_file。"""
testFile = tmp_path / "doc.pdf"
testFile.write_text("pdf content", encoding="utf-8")
fileComp = Comp.File(file=str(testFile))
bot = MagicMock()
bot.send_group_msg = AsyncMock()
bot.send_private_msg = AsyncMock()
bot.call_action = AsyncMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([fileComp]),
is_group=False,
session_id="789012",
)
bot.call_action.assert_called_once_with(
"upload_private_file",
user_id=789012,
file=str(testFile.resolve()),
name="file",
)
bot.send_private_msg.assert_not_called()
@pytest.mark.asyncio
```
These edits assume that `Comp`, `MessageChain`, and `AiocqhttpMessageEvent` are already imported earlier in `test_aiocqhttp_file_message.py`, consistent with the existing tests. If they are not, add the appropriate imports (mirroring how the existing file-upload tests reference them).
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+20
to
+29
| @pytest.mark.asyncio | ||
| async def test_file_group_uses_upload_group_file_api(tmp_path): | ||
| """群聊发送本地文件：应调用 upload_group_file API，不走 send_group_msg。""" | ||
| testFile = tmp_path / "test.txt" | ||
| testFile.write_text("hello world", encoding="utf-8") | ||
|
|
||
| fileComp = Comp.File(name="test.txt", file=str(testFile)) | ||
| bot = MagicMock() | ||
| bot.send_group_msg = AsyncMock() | ||
| bot.send_private_msg = AsyncMock() |
Contributor
There was a problem hiding this comment.
suggestion (testing): Add a test for non-digit session_id to cover the session_id.isdigit() guard.
Currently we only cover session_id=None. Please also add a case where session_id is a non-empty, non-numeric string (e.g. "abc123") to confirm the upload path is skipped and _dispatch_send is used instead. For example, a test_non_digit_session_id_falls_back_to_dispatch_send (for both group and private, if applicable) that:
- passes a
Filesegment withsession_id="abc123"; - asserts
bot.call_actionis not called; - asserts the appropriate fallback send path (
send_group_msg/send_private_msg) or the expected exception from_dispatch_send.
Suggested implementation:
bot.call_action.assert_not_called()
# 回退到 send_group_msg
bot.send_group_msg.assert_called_once()
@pytest.mark.asyncio
async def test_non_digit_session_id_group_falls_back_to_dispatch_send(tmp_path):
"""群聊：session_id 为非数字字符串时应回退到 _dispatch_send，不调用 upload API。"""
testFile = tmp_path / "test.txt"
testFile.write_text("hello world", encoding="utf-8")
fileComp = Comp.File(name="test.txt", file=str(testFile))
bot = MagicMock()
bot.send_group_msg = AsyncMock()
bot.send_private_msg = AsyncMock()
bot.call_action = AsyncMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([fileComp]),
is_group=True,
session_id="abc123",
)
# 非数字 session_id，上传 API 应被跳过
bot.call_action.assert_not_called()
# 回退到 send_group_msg
bot.send_group_msg.assert_called_once()
bot.send_private_msg.assert_not_called()
@pytest.mark.asyncio
async def test_non_digit_session_id_private_falls_back_to_dispatch_send(tmp_path):
"""私聊：session_id 为非数字字符串时应回退到 _dispatch_send，不调用 upload API。"""
testFile = tmp_path / "test.txt"
testFile.write_text("hello world", encoding="utf-8")
fileComp = Comp.File(name="test.txt", file=str(testFile))
bot = MagicMock()
bot.send_group_msg = AsyncMock()
bot.send_private_msg = AsyncMock()
bot.call_action = AsyncMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([fileComp]),
is_group=False,
session_id="abc123",
)
# 非数字 session_id，上传 API 应被跳过
bot.call_action.assert_not_called()
# 回退到 send_private_msg
bot.send_private_msg.assert_called_once()
bot.send_group_msg.assert_not_called()These edits assume that:
- The snippet with
bot.call_action.assert_not_called()andbot.send_group_msg.assert_called_once()belongs totest_no_local_file_falls_back_to_dispatch_send, and thatpytest,MagicMock,AsyncMock,Comp,MessageChain, andAiocqhttpMessageEventare already imported (as shown in the partial file). - No additional fixtures or helpers are required for these tests; if your existing tests use shared fixtures for
botconstruction or message setup, you may want to refactor these new tests to reuse those fixtures instead of instantiatingMagicMock/AsyncMockdirectly.
| file=str(testFile.resolve()), | ||
| name="test.txt", | ||
| ) | ||
| bot.send_group_msg.assert_not_called() |
Contributor
There was a problem hiding this comment.
suggestion (testing): Consider adding a private-chat variant for multiple-file uploads.
You already verify that each file in a group chat triggers a separate upload_group_file call. To fully exercise the new behavior, please add an analogous private-chat test (e.g. test_multiple_files_each_uses_upload_private_api) that checks:
bot.call_actionis invoked once per file with"upload_private_file"and the expecteduser_id/file/name.send_private_msgis never called.
This keeps group and private flows symmetric and protects against regressions in one path only.
Suggested implementation:
@pytest.mark.asyncio
async def test_multiple_files_each_uses_upload_api(tmp_path):
"""多个 File 段：每个都独立调用 group upload API。"""
file1 = tmp_path / "a.txt"
file2 = tmp_path / "b.txt"
file1.write_text("aaa", encoding="utf-8")
file2.write_text("bbb", encoding="utf-8")
comp1 = Comp.File(name="a.txt", file=str(file1))
comp2 = Comp.File(name="b.txt", file=str(file2))
bot = MagicMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([comp1, comp2]),
is_group=True,
session_id="123456",
)
assert bot.call_action.call_count == 2
bot.call_action.assert_any_call(
"upload_group_file",
group_id="123456",
file=str(file1),
name="a.txt",
)
bot.call_action.assert_any_call(
"upload_group_file",
group_id="123456",
file=str(file2),
name="b.txt",
)
# 发送文件时不应直接调用发送消息接口
assert not bot.send_group_msg.called
@pytest.mark.asyncio
async def test_multiple_files_each_uses_upload_private_api(tmp_path):
"""多个 File 段：每个都独立调用 private upload API。"""
file1 = tmp_path / "a.txt"
file2 = tmp_path / "b.txt"
file1.write_text("aaa", encoding="utf-8")
file2.write_text("bbb", encoding="utf-8")
comp1 = Comp.File(name="a.txt", file=str(file1))
comp2 = Comp.File(name="b.txt", file=str(file2))
bot = MagicMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([comp1, comp2]),
is_group=False,
user_id=123456,
session_id="123456",
)
assert bot.call_action.call_count == 2
bot.call_action.assert_any_call(
"upload_private_file",
user_id=123456,
file=str(file1),
name="a.txt",
)
bot.call_action.assert_any_call(
"upload_private_file",
user_id=123456,
file=str(file2),
name="b.txt",
)
# 发送文件时不应直接调用发送私聊消息接口
assert not bot.send_private_msg.called- The exact arguments to
AiocqhttpMessageEvent.send_message(e.g.group_id,user_id,session_id) and theupload_group_file/upload_private_filecall signatures should be aligned with the rest of this test file. You may need to adjustgroup_id="123456"/user_id=123456/session_id="123456"to match existing conventions. - If
ANYfromunittest.mockis preferred in your tests (to avoid hard-coding IDs), you can replace123456withANYand ensurefrom unittest.mock import ANYis present at the top of the file. - If the existing
test_multiple_files_each_uses_upload_apialready has assertions beyond the snippet you provided, merge the new body carefully to avoid duplicating or conflicting assertions.
| name="test.txt", | ||
| ) | ||
| bot.send_group_msg.assert_not_called() | ||
| bot.send_private_msg.assert_not_called() |
Contributor
There was a problem hiding this comment.
suggestion (testing): Add a test for defaulting name to 'file' when seg.name is missing.
Right now all tests pass a File with an explicit name. Since the implementation uses seg.name or "file", please add a test (e.g. test_file_without_name_defaults_to_file) that:
- creates
Comp.File(file=str(testFile))without aname; - calls
send_messagefor group/private; - asserts
bot.call_actionis invoked withname="file".
This ensures the default-name behavior is covered and guarded against regressions.
Suggested implementation:
bot.send_group_msg.assert_not_called()
bot.send_private_msg.assert_not_called()
@pytest.mark.asyncio
async def test_file_group_without_name_defaults_to_file(tmp_path):
"""群聊发送本地文件且未提供 name：应使用默认 name='file' 并调用 upload_group_file。"""
testFile = tmp_path / "test.txt"
testFile.write_text("hello world", encoding="utf-8")
fileComp = Comp.File(file=str(testFile))
bot = MagicMock()
bot.send_group_msg = AsyncMock()
bot.send_private_msg = AsyncMock()
bot.call_action = AsyncMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([fileComp]),
is_group=True,
session_id="123456",
)
bot.call_action.assert_called_once_with(
"upload_group_file",
group_id=123456,
file=str(testFile.resolve()),
name="file",
)
bot.send_group_msg.assert_not_called()
bot.send_private_msg.assert_not_called()
@pytest.mark.asyncio
async def test_file_private_without_name_defaults_to_file(tmp_path):
"""私聊发送本地文件且未提供 name：应使用默认 name='file' 并调用 upload_private_file。"""
testFile = tmp_path / "doc.pdf"
testFile.write_text("pdf content", encoding="utf-8")
fileComp = Comp.File(file=str(testFile))
bot = MagicMock()
bot.send_group_msg = AsyncMock()
bot.send_private_msg = AsyncMock()
bot.call_action = AsyncMock()
await AiocqhttpMessageEvent.send_message(
bot=bot,
message_chain=MessageChain([fileComp]),
is_group=False,
session_id="789012",
)
bot.call_action.assert_called_once_with(
"upload_private_file",
user_id=789012,
file=str(testFile.resolve()),
name="file",
)
bot.send_private_msg.assert_not_called()
@pytest.mark.asyncioThese edits assume that Comp, MessageChain, and AiocqhttpMessageEvent are already imported earlier in test_aiocqhttp_file_message.py, consistent with the existing tests. If they are not, add the appropriate imports (mirroring how the existing file-upload tests reference them).
- 增加对文件路径存在性的检查，确保文件可用 - 优化异常处理，记录详细错误信息 - 添加多个文件上传的测试用例，确保功能正常
dosubot
Bot
added
size:M
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Modifications / 改动点
close #8464
Screenshots or Test Results / 运行截图或测试结果
Checklist / 检查清单
😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
/ 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。
👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
/ 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。
🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in
requirements.txtandpyproject.toml./ 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到
requirements.txt和pyproject.toml文件相应位置。😮 My changes do not introduce malicious code.
/ 我的更改没有引入恶意代码。
Summary by Sourcery
Route aiocqhttp file message segments through NapCat upload APIs when possible and add fallbacks for robust file delivery.
New Features:
Bug Fixes:
Tests: