github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-30 23:20:35 +00:00
Code Issues Packages Projects Releases Wiki Activity
deer-flow/backend/tests/test_dingtalk_channel.py
Hyeonsang Cho 48949482c5
fix(channels): give DingTalk AI cards the channel token, not the SDK fetch (#7597) 
* fix(channels): give DingTalk AI cards the channel token, not the SDK fetch

AICardReplier's async card calls read dingtalk_client.get_access_token()
inline, and dingtalk-stream refreshes it with requests.post and no
timeout. Card creation and card streaming both run on the Gateway loop,
so the first card and every SDK refresh after it (about every two hours)
blocked the loop for the whole token request, without bound if DingTalk
stalled.

Hand the replier a _CardTokenClient that serves the channel's own
httpx token (10s timeout), set before card creation and before every
streamed update. A token failure while streaming now raises into send's
existing sampleMarkdown fallback instead of being skipped silently.

Pin it with a strict blocking-I/O anchor over a real DingTalkStreamClient
and AICardReplier, a gate meta-check, and unit tests for token rotation
and failure. LocalHttpServer also records each request's method, path
and headers.

* docs(changelog): reference #7597 for the DingTalk AI card token fetch

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-10-28 10:50:30 +08:00

3221 lines
121 KiB
Python
Raw Permalink Blame History

"""Tests for the DingTalk channel implementation."""
from __future__ import annotations
import asyncio
import json
import logging
import threading
import time
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from support.symlinks import symlink_or_skip
from app.channels.commands import KNOWN_CHANNEL_COMMANDS
from app.channels.dingtalk import (
_CONVERSATION_TYPE_GROUP,
_CONVERSATION_TYPE_P2P,
DingTalkChannel,
_adapt_markdown_for_dingtalk,
_convert_markdown_table,
_DingTalkMessageHandler,
_extract_text_from_rich_text,
_is_dingtalk_command,
_normalize_conversation_type,
_parse_allowed_users,
)
from app.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage
from deerflow.config.paths import VIRTUAL_PATH_PREFIX
from deerflow.sandbox.lease import discard_sandbox_lease_manager, get_sandbox_lease_manager
def _run(coro):
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.close()
async def _take_inbound(bus: MessageBus):
inbound = await asyncio.wait_for(bus.get_inbound(), timeout=1)
bus.inbound_task_done()
return inbound
# ---------------------------------------------------------------------------
# Helper: build mock ChatbotMessage
# ---------------------------------------------------------------------------
def _make_chatbot_message(
*,
text: str = "hello",
message_type: str = "text",
conversation_type: str | int = _CONVERSATION_TYPE_P2P,
sender_staff_id: str = "user_001",
sender_nick: str = "Test User",
conversation_id: str = "conv_001",
message_id: str = "msg_001",
rich_text_list: list | None = None,
):
"""Build a minimal mock object mimicking dingtalk_stream.ChatbotMessage."""
msg = SimpleNamespace()
msg.message_type = message_type
msg.conversation_type = conversation_type
msg.sender_staff_id = sender_staff_id
msg.sender_nick = sender_nick
msg.conversation_id = conversation_id
msg.message_id = message_id
if message_type == "text":
msg.text = SimpleNamespace(content=text)
msg.rich_text_content = None
elif message_type == "richText":
msg.text = None
msg.rich_text_content = SimpleNamespace(rich_text_list=rich_text_list or [])
else:
msg.text = None
msg.rich_text_content = None
return msg
# ---------------------------------------------------------------------------
# _DingTalkMessageHandler SDK contract
# ---------------------------------------------------------------------------
class TestDingTalkMessageHandlerSdkContract:
def test_pre_start_exists_and_noop(self):
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
handler = _DingTalkMessageHandler(channel)
handler.pre_start()
def test_raw_process_returns_ack(self):
pytest.importorskip("dingtalk_stream")
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._on_chatbot_message = MagicMock()
handler = _DingTalkMessageHandler(channel)
cb = MagicMock()
cb.headers.message_id = "mid-1"
cb.data = {
"msgtype": "text",
"text": {"content": "hi"},
"senderStaffId": "u1",
"conversationType": "1",
"msgId": "m1",
}
ack = await handler.raw_process(cb)
assert ack.code == 200
assert ack.headers.message_id == "mid-1"
assert ack.data == {"response": "OK"}
channel._on_chatbot_message.assert_called_once()
_run(go())
# ---------------------------------------------------------------------------
# _parse_allowed_users tests
# ---------------------------------------------------------------------------
class TestParseAllowedUsers:
@pytest.mark.parametrize("allowed_users", [None, [], (), set(), frozenset(), "", " "])
def test_unset_or_empty_allows_everyone_quietly(self, allowed_users, caplog):
with caplog.at_level(logging.WARNING, logger="app.channels.dingtalk"):
assert _parse_allowed_users(allowed_users) is None
assert caplog.records == []
@pytest.mark.parametrize("collection", [list, tuple, set, frozenset])
def test_listed_staff_ids(self, collection):
assert _parse_allowed_users(collection(["user1", " user2 "])) == {"user1", "user2"}
def test_single_string_is_one_entry(self):
assert _parse_allowed_users("user1") == {"user1"}
def test_numeric_values_converted_to_string(self):
assert _parse_allowed_users([123, 456]) == {"123", "456"}
def test_numeric_scalar_is_one_entry(self):
assert _parse_allowed_users(12345) == {"12345"}
@pytest.mark.parametrize("allowed_users", ["user1,user2", ","])
def test_scalar_with_separators_stays_one_literal_id_and_warns(self, allowed_users, caplog):
with caplog.at_level(logging.WARNING, logger="app.channels.dingtalk"):
assert _parse_allowed_users(allowed_users) == {allowed_users}
assert "YAML list" in caplog.text
def test_invalid_entries_are_dropped_with_a_warning(self, caplog):
with caplog.at_level(logging.WARNING, logger="app.channels.dingtalk"):
assert _parse_allowed_users(["user1", True, 4.2, None, " "]) == {"user1"}
assert "True" in caplog.text
assert "4.2" in caplog.text
assert not any(record.levelno == logging.ERROR for record in caplog.records)
@pytest.mark.parametrize(
"allowed_users",
[[""], ["", " "], [None], [True], {"user1": True}, {}, True, 4.2],
ids=["blank-entry", "blank-entries", "null-entry", "bool-entry", "mapping", "empty-mapping", "bool-scalar", "float-scalar"],
)
def test_allowlist_without_a_valid_staff_id_denies_everyone(self, allowed_users, caplog):
# A mapping must not authorize its keys, and a list whose entries are
# all blank (e.g. ``[$DINGTALK_STAFF_ID]`` with the variable set to an
# empty string) must not turn into "allow all".
with caplog.at_level(logging.ERROR, logger="app.channels.dingtalk"):
assert _parse_allowed_users(allowed_users) == frozenset()
assert any(record.levelno == logging.ERROR for record in caplog.records)
# ---------------------------------------------------------------------------
# _normalize_conversation_type tests
# ---------------------------------------------------------------------------
class TestNormalizeConversationType:
def test_group_int_or_str(self):
assert _normalize_conversation_type(2) == _CONVERSATION_TYPE_GROUP
assert _normalize_conversation_type("2") == _CONVERSATION_TYPE_GROUP
def test_p2p_or_none(self):
assert _normalize_conversation_type(1) == _CONVERSATION_TYPE_P2P
assert _normalize_conversation_type(None) == _CONVERSATION_TYPE_P2P
# ---------------------------------------------------------------------------
# _is_dingtalk_command tests
# ---------------------------------------------------------------------------
class TestIsDingTalkCommand:
@pytest.mark.parametrize("command", sorted(KNOWN_CHANNEL_COMMANDS))
def test_known_commands_recognized(self, command):
assert _is_dingtalk_command(command) is True
@pytest.mark.parametrize(
"text",
[
"/unknown",
"/mnt/user-data/outputs/report.md",
"hello",
"",
"not a command",
],
)
def test_non_commands_rejected(self, text):
assert _is_dingtalk_command(text) is False
# ---------------------------------------------------------------------------
# _extract_text_from_rich_text tests
# ---------------------------------------------------------------------------
class TestExtractTextFromRichText:
def test_single_text_item(self):
result = _extract_text_from_rich_text([{"text": "hello"}])
assert result == "hello"
def test_multiple_text_items(self):
result = _extract_text_from_rich_text([{"text": "hello"}, {"text": "world"}])
assert result == "hello world"
def test_non_text_items_ignored(self):
result = _extract_text_from_rich_text(
[
{"downloadCode": "abc123"},
{"text": "caption"},
]
)
assert result == "caption"
def test_empty_list(self):
assert _extract_text_from_rich_text([]) == ""
# ---------------------------------------------------------------------------
# DingTalkChannel._extract_text tests
# ---------------------------------------------------------------------------
class TestExtractText:
def test_plain_text(self):
msg = _make_chatbot_message(text="Hello World")
assert DingTalkChannel._extract_text(msg) == "Hello World"
def test_plain_text_stripped(self):
msg = _make_chatbot_message(text=" Hello ")
assert DingTalkChannel._extract_text(msg) == "Hello"
def test_rich_text(self):
msg = _make_chatbot_message(
message_type="richText",
rich_text_list=[{"text": "Part 1"}, {"text": "Part 2"}],
)
assert DingTalkChannel._extract_text(msg) == "Part 1 Part 2"
def test_unknown_type_returns_empty(self):
msg = _make_chatbot_message(message_type="picture")
assert DingTalkChannel._extract_text(msg) == ""
# ---------------------------------------------------------------------------
# DingTalkChannel._on_chatbot_message tests (inbound parsing)
# ---------------------------------------------------------------------------
class TestOnChatbotMessage:
def test_p2p_message_produces_correct_inbound(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
text="hello from dingtalk",
conversation_type=_CONVERSATION_TYPE_P2P,
sender_staff_id="user_001",
message_id="msg_001",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.channel_name == "dingtalk"
assert inbound.chat_id == "user_001"
assert inbound.user_id == "user_001"
assert inbound.text == "hello from dingtalk"
assert inbound.topic_id is None
assert inbound.metadata["conversation_type"] == _CONVERSATION_TYPE_P2P
assert inbound.metadata["sender_staff_id"] == "user_001"
_run(go())
def test_group_message_produces_correct_inbound(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
text="hello group",
conversation_type=_CONVERSATION_TYPE_GROUP,
sender_staff_id="user_002",
conversation_id="conv_group_001",
message_id="msg_group_001",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.channel_name == "dingtalk"
assert inbound.chat_id == "conv_group_001"
assert inbound.user_id == "user_002"
assert inbound.text == "hello group"
assert inbound.topic_id == "msg_group_001"
assert inbound.metadata["conversation_type"] == _CONVERSATION_TYPE_GROUP
assert inbound.metadata["conversation_id"] == "conv_group_001"
_run(go())
@pytest.mark.parametrize("text", ["@bot /new", "@_user_1 /help", "@bot /goal ship it"])
def test_leading_mention_before_command_classifies_and_strips(self, text):
"""DingTalk group chats leave "@bot /new" in the text; classify as COMMAND
and strip the mention so ChannelManager receives the bare command."""
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
text=text,
conversation_type=_CONVERSATION_TYPE_GROUP,
sender_staff_id="user_002",
conversation_id="conv_group_001",
message_id="msg_mention_cmd",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.msg_type == InboundMessageType.COMMAND, f"{text!r} should be COMMAND"
assert not inbound.text.startswith("@"), "leading mention must be stripped for dispatch"
assert inbound.text.split(maxsplit=1)[0] in KNOWN_CHANNEL_COMMANDS
_run(go())
def test_leading_mention_before_chat_keeps_mention(self):
"""A mentioned non-command stays CHAT and keeps the mention for the agent."""
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
text="@bot please summarise this",
conversation_type=_CONVERSATION_TYPE_GROUP,
sender_staff_id="user_002",
conversation_id="conv_group_001",
message_id="msg_mention_chat",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.msg_type == InboundMessageType.CHAT
assert inbound.text == "@bot please summarise this"
_run(go())
def test_group_message_integer_conversation_type_normalized(self):
"""SDK may deliver conversationType as int 2 — must still route as group."""
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
text="hello group",
conversation_type=2,
sender_staff_id="user_002",
conversation_id="conv_group_001",
message_id="msg_group_002",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.chat_id == "conv_group_001"
assert inbound.topic_id == "msg_group_002"
assert inbound.metadata["conversation_type"] == _CONVERSATION_TYPE_GROUP
_run(go())
@pytest.mark.parametrize("sender_staff_id", ["", None])
def test_p2p_message_without_sender_is_dropped(self, sender_staff_id):
"""A P2P chat_id *is* the sender, so an empty one keys every user to one thread.
``binding_key`` builds ``f"{channel}:{chat_id}"`` for a topic-less conversation,
so publishing this would put every senderless P2P message under the literal
``"dingtalk:"`` — one shared thread, one shared history, across users.
"""
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
text="hello",
conversation_type=_CONVERSATION_TYPE_P2P,
sender_staff_id=sender_staff_id,
message_id="msg_no_sender",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
bus.publish_inbound.assert_not_awaited()
assert bus.inbound_queue.empty()
_run(go())
def test_group_message_without_conversation_id_is_dropped(self):
"""The group route reaches the same degenerate key from the other side."""
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
text="hello group",
conversation_type=_CONVERSATION_TYPE_GROUP,
sender_staff_id="user_002",
conversation_id="",
message_id="msg_no_conv",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
bus.publish_inbound.assert_not_awaited()
assert bus.inbound_queue.empty()
_run(go())
def test_group_message_with_unknown_sender_is_still_delivered(self):
"""Reverse anchor: the guard is on the conversation identity, not on the sender.
A group message identifies its conversation through ``conversation_id``, so an
unknown sender must not cost the whole message.
"""
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
text="hello group",
conversation_type=_CONVERSATION_TYPE_GROUP,
sender_staff_id="",
conversation_id="conv_group_003",
message_id="msg_group_003",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.chat_id == "conv_group_003"
assert inbound.topic_id == "msg_group_003"
_run(go())
def test_command_classified_correctly(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(text="/help")
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.msg_type == InboundMessageType.COMMAND
_run(go())
def test_non_command_classified_as_chat(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(text="just chatting")
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.msg_type == InboundMessageType.CHAT
_run(go())
def test_empty_text_ignored(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(text=" ")
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
bus.publish_inbound.assert_not_awaited()
assert bus.inbound_queue.empty()
_run(go())
# ---------------------------------------------------------------------------
# allowed_users filtering tests
# ---------------------------------------------------------------------------
class TestAllowedUsersFiltering:
def test_allowed_user_passes(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={"allowed_users": ["user_001"]})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(sender_staff_id="user_001")
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
await _take_inbound(bus)
_run(go())
def test_non_allowed_user_blocked(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={"allowed_users": ["user_001"]})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(sender_staff_id="user_blocked")
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
bus.publish_inbound.assert_not_awaited()
assert bus.inbound_queue.empty()
_run(go())
def test_non_allowed_user_message_content_not_logged(self, caplog):
import logging
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={"allowed_users": ["user_001"]})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(sender_staff_id="user_blocked", text="secret blocked content")
with caplog.at_level(logging.INFO, logger="app.channels.dingtalk"):
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
bus.publish_inbound.assert_not_awaited()
assert bus.inbound_queue.empty()
# The parsed-message INFO log (with message content) must not fire for
# a blocked sender — allowed_users still acts as a privacy/noise filter.
assert "parsed message" not in caplog.text
assert "secret blocked content" not in caplog.text
_run(go())
def test_connect_code_bypasses_allowed_users_filter(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={"allowed_users": ["user_001"], "connection_repo": object()})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
channel._bind_connection_from_connect_code = AsyncMock(return_value=True)
msg = _make_chatbot_message(sender_staff_id="user_blocked", text="/connect dingtalk-bind-code")
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
channel._bind_connection_from_connect_code.assert_awaited_once()
bus.publish_inbound.assert_not_awaited()
assert bus.inbound_queue.empty()
_run(go())
@pytest.mark.parametrize(
("allowed_users", "admitted"),
[([""], False), ({"user_001": True}, False), ([" user_001 "], True), (frozenset({"user_001"}), True)],
ids=["blank-entry", "mapping", "padded-entry", "frozenset"],
)
def test_allowlist_gate_uses_parsed_staff_ids(self, allowed_users, admitted):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={"allowed_users": allowed_users})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(_make_chatbot_message(sender_staff_id="user_001"))
await asyncio.sleep(0.1)
assert (not bus.inbound_queue.empty()) is admitted
_run(go())
def test_empty_allowed_users_allows_all(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={"allowed_users": []})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(sender_staff_id="anyone")
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
await _take_inbound(bus)
_run(go())
# ---------------------------------------------------------------------------
# send routing tests (P2P vs Group)
# ---------------------------------------------------------------------------
class TestMarkdownFallbackPropagation:
def test_fallback_raises_on_failure(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._cached_token = "tok"
channel._token_expires_at = float("inf")
channel._send_p2p_message = AsyncMock(side_effect=ConnectionError("send failed"))
with pytest.raises(ConnectionError, match="send failed"):
await channel._send_markdown_fallback("test_key", _CONVERSATION_TYPE_P2P, "user_001", "", "hello")
_run(go())
class TestSendRouting:
def test_p2p_send_uses_oto_endpoint(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
channel._send_p2p_message = AsyncMock()
channel._send_group_message = AsyncMock()
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="Hello P2P",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
},
)
await channel.send(msg)
channel._send_p2p_message.assert_awaited_once_with("test_key", "user_001", "Hello P2P")
channel._send_group_message.assert_not_awaited()
_run(go())
def test_group_send_uses_group_endpoint(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
channel._send_p2p_message = AsyncMock()
channel._send_group_message = AsyncMock()
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="conv_001",
thread_id="thread_001",
text="Hello Group",
metadata={
"conversation_type": _CONVERSATION_TYPE_GROUP,
"sender_staff_id": "user_001",
"conversation_id": "conv_001",
},
)
await channel.send(msg)
channel._send_group_message.assert_awaited_once_with("test_key", "conv_001", "Hello Group", at_user_ids=["user_001"])
channel._send_p2p_message.assert_not_awaited()
_run(go())
def test_default_metadata_uses_p2p(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
channel._send_p2p_message = AsyncMock()
channel._send_group_message = AsyncMock()
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="Hello",
metadata={},
)
await channel.send(msg)
channel._send_p2p_message.assert_awaited_once()
channel._send_group_message.assert_not_awaited()
_run(go())
# ---------------------------------------------------------------------------
# send retry tests
# ---------------------------------------------------------------------------
class TestSendRetry:
def test_retries_on_failure_then_succeeds(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
call_count = 0
async def flaky_send(robot_code, user_id, text):
nonlocal call_count
call_count += 1
if call_count < 3:
raise ConnectionError("network error")
channel._send_p2p_message = AsyncMock(side_effect=flaky_send)
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="hello",
metadata={"conversation_type": _CONVERSATION_TYPE_P2P, "sender_staff_id": "user_001"},
)
await channel.send(msg)
assert call_count == 3
_run(go())
def test_raises_after_all_retries_exhausted(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
channel._send_p2p_message = AsyncMock(side_effect=ConnectionError("fail"))
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="hello",
metadata={"conversation_type": _CONVERSATION_TYPE_P2P, "sender_staff_id": "user_001"},
)
with pytest.raises(ConnectionError):
await channel.send(msg)
assert channel._send_p2p_message.await_count == 3
_run(go())
def test_raises_runtime_error_when_no_attempts_configured(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="hello",
metadata={"conversation_type": _CONVERSATION_TYPE_P2P, "sender_staff_id": "user_001"},
)
with pytest.raises(RuntimeError, match="without an exception"):
await channel.send(msg, _max_retries=0)
_run(go())
# ---------------------------------------------------------------------------
# topic_id mapping tests
# ---------------------------------------------------------------------------
class TestTopicIdMapping:
def test_p2p_topic_is_none(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
conversation_type=_CONVERSATION_TYPE_P2P,
message_id="msg_p2p_001",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.topic_id is None
_run(go())
def test_group_topic_is_message_id(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
msg = _make_chatbot_message(
conversation_type=_CONVERSATION_TYPE_GROUP,
message_id="msg_group_001",
conversation_id="conv_001",
)
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.topic_id == "msg_group_001"
_run(go())
# ---------------------------------------------------------------------------
# Token caching tests
# ---------------------------------------------------------------------------
class TestAccessTokenValidation:
def test_rejects_non_dict_response(self):
async def go():
from unittest.mock import patch
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "k"
channel._client_secret = "s"
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
return "not a dict"
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
return FakeResponse()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
with pytest.raises(ValueError, match="JSON object"):
await channel._get_access_token()
_run(go())
def test_rejects_empty_access_token(self):
async def go():
from unittest.mock import patch
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "k"
channel._client_secret = "s"
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
return {"accessToken": "", "expireIn": 7200}
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
return FakeResponse()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
with pytest.raises(ValueError, match="usable accessToken"):
await channel._get_access_token()
_run(go())
@pytest.mark.parametrize("expire_in", ["invalid", float("inf"), 10**400, 0, -1, 1, 100, 299, 300, "300"])
def test_invalid_expire_in_uses_default(self, expire_in):
async def go():
from unittest.mock import patch
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "k"
channel._client_secret = "s"
call_count = 0
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
return {"accessToken": "tok_ok", "expireIn": expire_in}
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
nonlocal call_count
call_count += 1
return FakeResponse()
with (
patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()),
patch("app.channels.dingtalk.time.monotonic", return_value=1000.0) as clock,
):
token = await channel._get_access_token()
clock.return_value = 1005.0
cached_token = await channel._get_access_token()
assert token == "tok_ok"
assert cached_token == "tok_ok"
assert channel._token_expires_at == 7900.0
assert call_count == 1
_run(go())
class TestTokenCaching:
@pytest.mark.parametrize(("expire_in", "expected_expiry"), [(301, 1001.0), (3600, 4300.0), (7200, 7900.0)])
def test_token_is_cached_across_calls(self, expire_in, expected_expiry):
async def go():
from unittest.mock import patch
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
call_count = 0
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
return {"accessToken": "tok_abc", "expireIn": expire_in}
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
nonlocal call_count
call_count += 1
return FakeResponse()
with (
patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()),
patch("app.channels.dingtalk.time.monotonic", return_value=1000.0) as clock,
):
t1 = await channel._get_access_token()
clock.return_value = 1000.5
t2 = await channel._get_access_token()
assert channel._token_expires_at == expected_expiry
assert call_count == 1
clock.return_value = expected_expiry
t3 = await channel._get_access_token()
assert t1 == "tok_abc"
assert t2 == "tok_abc"
assert t3 == "tok_abc"
assert call_count == 2
_run(go())
# ---------------------------------------------------------------------------
# Group message @ mention format tests
# ---------------------------------------------------------------------------
class TestGroupMessageMarkdownFormat:
def test_at_user_ids_still_use_markdown(self):
"""groupMessages/send uses sampleMarkdown; @{userId} in body returns 400 so at_user_ids is ignored."""
async def go():
from unittest.mock import patch
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
channel._cached_token = "tok_test"
channel._token_expires_at = float("inf")
captured_json: list[dict] = []
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
return {"processQueryKey": "ok"}
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
captured_json.append(kwargs.get("json", {}))
return FakeResponse()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
await channel._send_group_message("bot", "conv1", "hello", at_user_ids=["staff_001"])
assert len(captured_json) == 1
payload = captured_json[0]
assert payload["msgKey"] == "sampleMarkdown"
import json
param = json.loads(payload["msgParam"])
assert param["text"] == "hello"
assert "@" not in json.dumps(param)
_run(go())
def test_no_at_user_ids_uses_markdown(self):
async def go():
from unittest.mock import patch
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._client_secret = "test_secret"
channel._cached_token = "tok_test"
channel._token_expires_at = float("inf")
captured_json: list[dict] = []
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
return {"processQueryKey": "ok"}
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
captured_json.append(kwargs.get("json", {}))
return FakeResponse()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
await channel._send_group_message("bot", "conv1", "hello")
assert len(captured_json) == 1
payload = captured_json[0]
assert payload["msgKey"] == "sampleMarkdown"
_run(go())
class TestAdaptMarkdownForDingtalk:
def test_fenced_code_block_to_blockquote(self):
text = "Hello\n```python\ndef foo():\n return 1\n```\nDone"
result = _adapt_markdown_for_dingtalk(text)
assert "```" not in result
assert "> **python**" in result
assert "> def foo():" in result
assert "> return 1" in result
def test_fenced_code_block_no_language(self):
text = "```\nplain code\n```"
result = _adapt_markdown_for_dingtalk(text)
assert "```" not in result
assert "> plain code" in result
def test_inline_code_to_bold(self):
text = "Use `pip install` to install"
result = _adapt_markdown_for_dingtalk(text)
assert result == "Use **pip install** to install"
def test_horizontal_rule_to_unicode(self):
text = "Above\n---\nBelow"
result = _adapt_markdown_for_dingtalk(text)
assert "───────────" in result
assert "---" not in result
def test_supported_markdown_preserved(self):
text = "# Title\n**bold** and *italic*\n- list item\n> quote\n[link](http://example.com)"
result = _adapt_markdown_for_dingtalk(text)
assert result == text
def test_plain_text_unchanged(self):
text = "Hello world, no markdown here."
assert _adapt_markdown_for_dingtalk(text) == text
def test_combined_elements(self):
text = "# Report\n\nRun `make test` then:\n\n```bash\npytest -v\n```\n\n---\n\nDone."
result = _adapt_markdown_for_dingtalk(text)
assert "# Report" in result
assert "**make test**" in result
assert "> **bash**" in result
assert "> pytest -v" in result
assert "───────────" in result
assert "Done." in result
class TestConvertMarkdownTable:
@pytest.mark.parametrize(
("header", "row", "expected"),
[
("| Rule | Meaning |", r"| foo\|bar | alternative |", "> **Rule**: foo\\|bar\n> **Meaning**: alternative\n"),
(r"| Rule\|Pattern | Meaning |", "| regex | alternative |", "> **Rule\\|Pattern**: regex\n> **Meaning**: alternative\n"),
("| Rule | Meaning |", "|| alternative |", "> **Rule**: \n> **Meaning**: alternative\n"),
("| Rule | Meaning |", "| regex ||", "> **Rule**: regex\n> **Meaning**: \n"),
("| Rule | Meaning |", r"| regex | value\|", "> **Rule**: regex\n> **Meaning**: value\\|\n"),
("| Rule | Meaning |", r"| foo\|bar\|baz | alternative |", "> **Rule**: foo\\|bar\\|baz\n> **Meaning**: alternative\n"),
],
ids=["escaped-body-pipe", "escaped-header-pipe", "leading-empty-cell", "trailing-empty-cell", "escaped-final-pipe", "multiple-escaped-pipes"],
)
def test_cell_boundaries_preserved(self, header, row, expected):
text = f"{header}\n|---|---|\n{row}"
assert _convert_markdown_table(text) == expected
def test_fallback_adapter_preserves_table_values(self):
text = "| Rule | Meaning |\n|---|---|\n| `foo\\|bar` | alternative |"
assert _adapt_markdown_for_dingtalk(text) == "> **Rule**: **foo\\|bar**\n> **Meaning**: alternative\n"
def test_simple_table(self):
text = "| Name | Age |\n|------|-----|\n| Alice | 30 |\n| Bob | 25 |"
result = _convert_markdown_table(text)
assert "> **Name**: Alice" in result
assert "> **Age**: 30" in result
assert "> **Name**: Bob" in result
assert "> **Age**: 25" in result
assert "|" not in result
def test_table_with_surrounding_text(self):
text = "Results:\n\n| Key | Value |\n|-----|-------|\n| a | 1 |\n\nEnd."
result = _convert_markdown_table(text)
assert "Results:" in result
assert "> **Key**: a" in result
assert "> **Value**: 1" in result
assert "End." in result
def test_no_table(self):
text = "Just plain text\nwith lines"
assert _convert_markdown_table(text) == text
def test_alignment_separators(self):
text = "| Left | Center | Right |\n|:-----|:------:|------:|\n| a | b | c |"
result = _convert_markdown_table(text)
assert "> **Left**: a" in result
assert "> **Center**: b" in result
assert "> **Right**: c" in result
class TestUploadMediaValidation:
def test_non_dict_response_returns_none(self):
async def go():
from unittest.mock import patch
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "k"
channel._client_secret = "s"
channel._cached_token = "tok"
channel._token_expires_at = float("inf")
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
return ["not", "a", "dict"]
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
return FakeResponse()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
result = await channel._upload_media("/tmp/test.png", "image")
assert result is None
_run(go())
def test_json_decode_error_returns_none(self):
async def go():
import json as json_mod
from unittest.mock import patch
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
channel._client_id = "k"
channel._client_secret = "s"
channel._cached_token = "tok"
channel._token_expires_at = float("inf")
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
raise json_mod.JSONDecodeError("err", "", 0)
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
return FakeResponse()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
result = await channel._upload_media("/tmp/test.png", "image")
assert result is None
_run(go())
class TestChannelRegistration:
def test_dingtalk_in_channel_registry(self):
from app.channels.service import _CHANNEL_REGISTRY
assert "dingtalk" in _CHANNEL_REGISTRY
assert _CHANNEL_REGISTRY["dingtalk"] == "app.channels.dingtalk:DingTalkChannel"
def test_dingtalk_in_credential_keys(self):
from app.channels.service import _CHANNEL_CREDENTIAL_KEYS
assert "dingtalk" in _CHANNEL_CREDENTIAL_KEYS
assert "client_id" in _CHANNEL_CREDENTIAL_KEYS["dingtalk"]
assert "client_secret" in _CHANNEL_CREDENTIAL_KEYS["dingtalk"]
def test_dingtalk_in_channel_capabilities(self):
from app.channels.manager import CHANNEL_CAPABILITIES
assert "dingtalk" in CHANNEL_CAPABILITIES
assert CHANNEL_CAPABILITIES["dingtalk"]["supports_streaming"] is False
# ---------------------------------------------------------------------------
# AI Card streaming mode tests
# ---------------------------------------------------------------------------
class TestCardMode:
def test_card_mode_enabled_supports_streaming(self):
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
assert channel.supports_streaming is True
def test_non_card_mode_no_streaming(self):
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
assert channel.supports_streaming is False
def test_non_card_mode_unchanged(self):
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
assert channel._card_template_id == ""
assert channel._card_track_ids == {}
assert channel._card_repliers == {}
assert channel._incoming_messages == {}
assert channel._dingtalk_client is None
def test_card_source_key_matches_inbound_using_message_id_metadata(self):
"""Outbound correlation must match inbound ``message_id`` even if ``thread_ts`` drifts."""
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
inbound = channel._make_inbound(
chat_id="x",
user_id="u",
text="hi",
thread_ts="ts_fallback",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
"message_id": "msg_real",
},
)
out = OutboundMessage(
channel_name="dingtalk",
chat_id="x",
thread_id="t",
text="ok",
thread_ts="wrong_ts",
metadata=dict(inbound.metadata),
)
assert channel._make_card_source_key(inbound) == channel._make_card_source_key_from_outbound(out)
_run(go())
def test_running_reply_creates_card(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._client_id = "test_key"
channel._create_and_deliver_card = AsyncMock(return_value="track_001")
inbound = channel._make_inbound(
chat_id="user_001",
user_id="user_001",
text="hello",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
"message_id": "msg_001",
},
)
mock_chatbot_msg = MagicMock()
source_key = channel._make_card_source_key(inbound)
channel._incoming_messages[source_key] = mock_chatbot_msg
await channel._send_running_reply("user_001", inbound)
channel._create_and_deliver_card.assert_awaited_once_with(
"\u23f3 Working on it...",
chatbot_message=mock_chatbot_msg,
)
assert channel._card_track_ids[source_key] == "track_001"
_run(go())
def test_send_streams_to_card(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._client_id = "test_key"
channel._stream_update_card = AsyncMock()
# Pre-populate card tracking
source_key = f"{_CONVERSATION_TYPE_P2P}:user_001::msg_001"
channel._card_track_ids[source_key] = "track_001"
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="Partial response...",
is_final=False,
thread_ts="msg_001",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
},
)
await channel.send(msg)
channel._stream_update_card.assert_awaited_once_with(
"track_001",
"Partial response...",
is_finalize=False,
)
# Track ID should still exist (not final)
assert source_key in channel._card_track_ids
_run(go())
def test_send_finalizes_card(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._client_id = "test_key"
channel._stream_update_card = AsyncMock()
source_key = f"{_CONVERSATION_TYPE_P2P}:user_001::msg_001"
channel._card_track_ids[source_key] = "track_001"
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="Final answer.",
is_final=True,
thread_ts="msg_001",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
},
)
await channel.send(msg)
channel._stream_update_card.assert_awaited_once_with(
"track_001",
"Final answer.",
is_finalize=True,
)
# Track ID should be cleaned up after final
assert source_key not in channel._card_track_ids
_run(go())
def test_card_mode_skips_markdown_adaptation(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._client_id = "test_key"
raw_markdown = "```python\ndef foo():\n pass\n```"
captured_content: list[str] = []
async def capture_stream(out_track_id, content, *, is_finalize=False, is_error=False):
captured_content.append(content)
channel._stream_update_card = AsyncMock(side_effect=capture_stream)
source_key = f"{_CONVERSATION_TYPE_P2P}:user_001::msg_001"
channel._card_track_ids[source_key] = "track_001"
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text=raw_markdown,
is_final=True,
thread_ts="msg_001",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
},
)
await channel.send(msg)
# Raw markdown should be passed through without adaptation
assert captured_content[0] == raw_markdown
_run(go())
def test_card_fallback_on_creation_failure(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._client_id = "test_key"
# Card creation returns None (failure)
channel._create_and_deliver_card = AsyncMock(return_value=None)
channel._send_text_message_to_user = AsyncMock()
inbound = channel._make_inbound(
chat_id="user_001",
user_id="user_001",
text="hello",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
"message_id": "msg_001",
},
)
source_key = channel._make_card_source_key(inbound)
channel._incoming_messages[source_key] = MagicMock()
await channel._send_running_reply("user_001", inbound)
# Should fall through to text message
channel._send_text_message_to_user.assert_awaited_once()
assert len(channel._card_track_ids) == 0
_run(go())
def test_send_skips_non_final_without_card_track_when_template_configured(self):
"""Without a live card track, Manager streaming would duplicate sampleMarkdown sends."""
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._client_id = "test_key"
channel._send_group_message = AsyncMock()
channel._send_p2p_message = AsyncMock()
meta = {
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
}
await channel.send(
OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="t1",
text="partial",
is_final=False,
thread_ts="msg_001",
metadata=meta,
)
)
channel._send_p2p_message.assert_not_called()
channel._send_group_message.assert_not_called()
await channel.send(
OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="t1",
text="final answer",
is_final=True,
thread_ts="msg_001",
metadata=meta,
)
)
channel._send_p2p_message.assert_awaited_once()
_run(go())
def test_card_fallback_on_stream_failure(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._client_id = "test_key"
channel._stream_update_card = AsyncMock(side_effect=ConnectionError("stream failed"))
channel._send_markdown_fallback = AsyncMock()
source_key = f"{_CONVERSATION_TYPE_P2P}:user_001::msg_001"
channel._card_track_ids[source_key] = "track_001"
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="Final answer.",
is_final=True,
thread_ts="msg_001",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
},
)
await channel.send(msg)
# Should fallback to markdown
channel._send_markdown_fallback.assert_awaited_once_with(
"test_key",
_CONVERSATION_TYPE_P2P,
"user_001",
"",
"Final answer.",
)
# Track ID should be cleaned up
assert source_key not in channel._card_track_ids
_run(go())
def test_pre_start_stores_dingtalk_client(self):
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
handler = _DingTalkMessageHandler(channel)
mock_client = MagicMock()
handler.dingtalk_client = mock_client
handler.pre_start()
assert channel._dingtalk_client is mock_client
def test_chatbot_message_stored_for_card_mode(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
mock_message = MagicMock()
mock_message.sender_staff_id = "user_001"
mock_message.conversation_type = "1"
mock_message.conversation_id = ""
mock_message.message_id = "msg_001"
mock_message.sender_nick = "TestUser"
mock_message.message_type = "text"
mock_message.text = MagicMock(content="hello")
mock_message.rich_text_content = None
channel._main_loop = asyncio.get_running_loop()
channel._allowed_users = None
channel._running = True
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(mock_message)
await asyncio.sleep(0.1)
assert len(channel._incoming_messages) == 1
stored_msg = list(channel._incoming_messages.values())[0]
assert stored_msg is mock_message
await _take_inbound(bus)
_run(go())
def test_card_replier_cleanup_on_final(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._client_id = "test_key"
channel._stream_update_card = AsyncMock()
source_key = f"{_CONVERSATION_TYPE_P2P}:user_001::msg_001"
channel._card_track_ids[source_key] = "track_001"
channel._card_repliers["track_001"] = MagicMock()
msg = OutboundMessage(
channel_name="dingtalk",
chat_id="user_001",
thread_id="thread_001",
text="Final answer.",
is_final=True,
thread_ts="msg_001",
metadata={
"conversation_type": _CONVERSATION_TYPE_P2P,
"sender_staff_id": "user_001",
"conversation_id": "",
},
)
await channel.send(msg)
assert source_key not in channel._card_track_ids
assert "track_001" not in channel._card_repliers
_run(go())
def test_card_creation_without_sdk_client_returns_none(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._dingtalk_client = None
result = await channel._create_and_deliver_card(
"test",
chatbot_message=MagicMock(),
)
assert result is None
_run(go())
def test_card_creation_without_chatbot_message_returns_none(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._dingtalk_client = MagicMock()
result = await channel._create_and_deliver_card(
"test",
chatbot_message=None,
)
assert result is None
_run(go())
def test_stream_update_card_raises_without_replier(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
with pytest.raises(RuntimeError, match="No AICardReplier found"):
await channel._stream_update_card("nonexistent_track", "content")
_run(go())
def test_card_calls_use_the_channel_token_not_the_sdk_fetch(self, monkeypatch):
"""The AI card replier sends the channel's httpx token; the SDK never fetches one.
dingtalk-stream's ``get_access_token()`` posts with ``requests`` and no
timeout, inline in the replier's async methods on the Gateway loop.
"""
pytest.importorskip("dingtalk_stream")
import dingtalk_stream
from dingtalk_stream import card_replier
from support.local_sockets import LocalHttpServer
async def go():
async with LocalHttpServer(body={"accessToken": "tok-1", "expireIn": 7200}) as api:
base = api.url.removesuffix("/open")
monkeypatch.setattr("app.channels.dingtalk.DINGTALK_API_BASE", base)
monkeypatch.setattr(card_replier, "DINGTALK_OPENAPI_ENDPOINT", base)
channel = DingTalkChannel(MessageBus(), config={"card_template_id": "tpl_123"})
channel._client_id, channel._client_secret = "id", "secret"
sdk_client = MagicMock()
sdk_client.credential.client_id = "id"
sdk_client.get_access_token.side_effect = AssertionError("the SDK must not fetch a token")
channel._dingtalk_client = sdk_client
incoming = dingtalk_stream.ChatbotMessage()
incoming.conversation_type = _CONVERSATION_TYPE_P2P
incoming.sender_staff_id = "user_001"
track = await channel._create_and_deliver_card("Working on it...", chatbot_message=incoming)
assert track
# An expired channel token is refreshed before the streamed update.
channel._token_expires_at = 0.0
api.body = {"accessToken": "tok-2", "expireIn": 7200}
await channel._stream_update_card(track, "done", is_finalize=True)
sdk_client.get_access_token.assert_not_called()
assert [(method, path, headers.get("x-acs-dingtalk-access-token")) for method, path, headers in api.calls] == [
("POST", "/v1.0/oauth2/accessToken", None),
("POST", "/v1.0/card/instances/createAndDeliver", "tok-1"),
("POST", "/v1.0/oauth2/accessToken", None),
("PUT", "/v1.0/card/streaming", "tok-2"),
]
assert api.requests[0] == {"appKey": "id", "appSecret": "secret"}
assert api.requests[3]["outTrackId"] == track
assert api.requests[3]["isFinalize"] is True
_run(go())
def test_stream_update_card_raises_when_the_channel_token_fails(self):
"""A token failure surfaces, so ``send`` falls back instead of dropping the update."""
async def go():
channel = DingTalkChannel(MessageBus(), config={"card_template_id": "tpl_123"})
replier = MagicMock()
replier.async_streaming = AsyncMock()
channel._card_repliers["track_001"] = replier
channel._get_access_token = AsyncMock(side_effect=ValueError("bad token response"))
with pytest.raises(ValueError, match="bad token response"):
await channel._stream_update_card("track_001", "content", is_finalize=True)
replier.async_streaming.assert_not_awaited()
_run(go())
def test_stop_clears_card_state(self):
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"})
channel._running = True
channel._dingtalk_client = MagicMock()
channel._incoming_messages["key"] = MagicMock()
channel._card_repliers["track"] = MagicMock()
channel._card_track_ids["source"] = "track"
await channel.stop()
assert channel._dingtalk_client is None
assert channel._incoming_messages == {}
assert channel._card_repliers == {}
assert channel._card_track_ids == {}
_run(go())
class TestStreamTaskLifecycle:
@pytest.mark.parametrize("outcome", ["success", "failure", "cancelled"])
def test_tracks_and_finalizes_message_tasks(self, outcome, monkeypatch, caplog):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._running = True
frames = [{"sequence": index} for index in range(50 if outcome == "success" else 1)]
received = []
handler_tasks = []
keepalive_tasks = []
handlers_started = asyncio.Event()
release_handlers = asyncio.Event()
handlers_finalized = asyncio.Event()
close_stream = asyncio.Event()
keepalive_started = asyncio.Event()
keepalive_stopped = asyncio.Event()
failure = RuntimeError("stream handler failed")
loop = asyncio.get_running_loop()
previous_exception_handler = loop.get_exception_handler()
loop_errors = []
loop.set_exception_handler(lambda _loop, context: loop_errors.append(context))
async def background_task(frame):
received.append(frame)
handler_tasks.append(asyncio.current_task())
if len(received) == len(frames):
handlers_started.set()
await release_handlers.wait()
if outcome == "failure":
raise failure
async def keepalive(_websocket):
keepalive_tasks.append(asyncio.current_task())
keepalive_started.set()
try:
await asyncio.Event().wait()
finally:
keepalive_stopped.set()
class WebSocket:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return None
async def __aiter__(self):
for frame in frames:
yield json.dumps(frame)
await close_stream.wait()
channel._running = False
client = SimpleNamespace(pre_start=MagicMock(), background_task=background_task, keepalive=keepalive)
monkeypatch.setattr(channel, "_open_stream_connection", AsyncMock(return_value={"endpoint": "wss://example.invalid/stream", "ticket": "test-ticket"}))
monkeypatch.setattr("websockets.connect", lambda _uri: WebSocket())
stream_task = asyncio.create_task(channel._serve_stream(client))
try:
await asyncio.wait_for(handlers_started.wait(), timeout=1)
await asyncio.wait_for(keepalive_started.wait(), timeout=1)
assert received == frames
assert channel._background_tasks == set(handler_tasks)
assert all(not task.done() for task in handler_tasks)
remaining = set(handler_tasks)
def finalized(task):
remaining.remove(task)
if not remaining:
handlers_finalized.set()
# These observers run after the production done callbacks.
for task in handler_tasks:
task.add_done_callback(finalized)
if outcome == "cancelled":
task.cancel()
release_handlers.set()
await asyncio.wait_for(handlers_finalized.wait(), timeout=1)
assert channel._background_tasks == set()
if outcome == "cancelled":
assert all(task.cancelled() for task in handler_tasks)
close_stream.set()
await asyncio.wait_for(stream_task, timeout=1)
await asyncio.wait_for(keepalive_stopped.wait(), timeout=1)
assert channel._background_tasks == set()
client.pre_start.assert_called_once_with()
errors = [record for record in caplog.records if record.name == "app.channels.dingtalk" and record.levelno >= logging.ERROR]
if outcome == "failure":
assert len(errors) == 1
assert "stream message task" in errors[0].getMessage()
assert errors[0].exc_info is not None
assert errors[0].exc_info[1] is failure
else:
assert errors == []
assert loop_errors == []
finally:
channel._running = False
close_stream.set()
release_handlers.set()
tasks = [stream_task, *handler_tasks, *keepalive_tasks]
for task in tasks:
if not task.done():
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
loop.set_exception_handler(previous_exception_handler)
_run(go())
class TestStop:
def test_stop_retains_live_stream_thread_after_join_timeout(self):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
thread = MagicMock()
thread.is_alive.return_value = True
channel._thread = thread
channel._running = True
with pytest.raises(RuntimeError, match="still running after stop timeout"):
await channel.stop()
thread.join.assert_called_once_with(timeout=5)
assert channel._thread is thread
_run(go())
@staticmethod
def _channel() -> DingTalkChannel:
return DingTalkChannel(MessageBus(), config={"client_id": "id", "client_secret": "secret"})
def test_stop_closes_live_stream_connection_and_exits_thread(self, monkeypatch):
pytest.importorskip("dingtalk_stream")
import dingtalk_stream
from support.local_sockets import LocalHttpServer
from websockets.asyncio.server import serve
async def go():
connected = asyncio.Event()
closed = asyncio.Event()
async def handler(websocket):
connected.set()
try:
async for _ in websocket:
pass
finally:
closed.set()
async with serve(handler, "127.0.0.1", 0) as server:
port = server.sockets[0].getsockname()[1]
async with LocalHttpServer(body={"endpoint": f"ws://127.0.0.1:{port}", "ticket": "ticket"}) as tickets:
monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url)
channel = self._channel()
await channel.start()
thread = channel._thread
assert thread is not None and thread.is_alive()
await asyncio.wait_for(connected.wait(), 5)
started = time.monotonic()
await channel.stop()
elapsed = time.monotonic() - started
assert elapsed < 5, f"stop() took {elapsed:.2f}s"
await asyncio.wait_for(closed.wait(), 5)
assert not thread.is_alive()
assert channel._thread is None
assert channel._stream_client is None
assert channel.is_running is False
# The ticket request mirrors DingTalkStreamClient.open_connection().
request = tickets.requests[0]
assert request["clientId"] == "id"
assert request["clientSecret"] == "secret"
assert {"type": "CALLBACK", "topic": dingtalk_stream.chatbot.ChatbotMessage.TOPIC} in request["subscriptions"]
assert request["ua"].startswith("dingtalk-sdk-python/v")
assert "localIp" in request
_run(go())
def test_stop_aborts_transport_when_peer_never_acks_close(self, monkeypatch, caplog):
pytest.importorskip("dingtalk_stream")
import dingtalk_stream
from support.local_sockets import LocalHttpServer, StalledWebSocketServer, transport_socket, wait_until
from app.channels import dingtalk as dingtalk_module
caplog.set_level(logging.WARNING, logger="app.channels.dingtalk")
async def go():
async with StalledWebSocketServer() as peer, LocalHttpServer(body={"endpoint": peer.url, "ticket": "ticket"}) as tickets:
monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url)
monkeypatch.setattr(dingtalk_module, "DINGTALK_STREAM_CLOSE_TIMEOUT_SECONDS", 0.5)
channel = self._channel()
await channel.start()
thread = channel._thread
await asyncio.wait_for(peer.accepted.wait(), 5)
websocket = await wait_until(lambda: channel._stream_client is not None and channel._stream_client.websocket)
sock = transport_socket(websocket)
assert sock.fileno() != -1
started = time.monotonic()
await channel.stop()
elapsed = time.monotonic() - started
assert elapsed < 5, f"stop() took {elapsed:.2f}s"
assert not thread.is_alive()
assert channel._thread is None
# The socket is released by stop() itself, not by a later GC pass.
assert sock.fileno() == -1
assert "transport aborted" in caplog.text
await asyncio.wait_for(peer.closed.wait(), 5)
_run(go())
def test_stop_interrupts_open_connection_retry_sleep(self, monkeypatch):
pytest.importorskip("dingtalk_stream")
import dingtalk_stream
from support.local_sockets import LocalHttpServer
async def go():
async with LocalHttpServer(status=500, body={"error": "nope"}) as tickets:
monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url)
channel = self._channel()
await channel.start()
thread = channel._thread
await asyncio.wait_for(tickets.accepted.wait(), 5)
started = time.monotonic()
await channel.stop()
elapsed = time.monotonic() - started
assert elapsed < 5, f"stop() took {elapsed:.2f}s"
assert not thread.is_alive()
assert channel._thread is None
_run(go())
def test_stop_cancels_stalled_ticket_request_without_lingering_threads(self, monkeypatch):
pytest.importorskip("dingtalk_stream")
import dingtalk_stream
from support.local_sockets import LocalHttpServer
def forbidden_open_connection(self):
raise AssertionError("the SDK's timeout-less open_connection() must not run")
async def go():
async with LocalHttpServer(stall=True) as tickets:
monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url)
monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "open_connection", forbidden_open_connection)
await asyncio.to_thread(lambda: None) # warm the test loop's own executor before the baseline
baseline = {thread.ident for thread in threading.enumerate()}
channel = self._channel()
await channel.start()
thread = channel._thread
await asyncio.wait_for(tickets.accepted.wait(), 5)
started = time.monotonic()
await channel.stop()
elapsed = time.monotonic() - started
assert elapsed < 5, f"stop() took {elapsed:.2f}s"
assert not thread.is_alive()
# The cancelled request released its connection ...
await asyncio.wait_for(tickets.closed.wait(), 5)
# ... and nothing non-daemon is left to block interpreter exit.
lingering = [t.name for t in threading.enumerate() if t.ident not in baseline and t.is_alive() and not t.daemon]
assert lingering == []
_run(go())
def test_stop_before_stream_client_exists_never_connects(self, monkeypatch):
pytest.importorskip("dingtalk_stream")
import dingtalk_stream
from support.local_sockets import LocalHttpServer
gate = threading.Event()
class _GatedStreamClient(dingtalk_stream.DingTalkStreamClient):
def __init__(self, *args, **kwargs):
gate.wait(timeout=5)
super().__init__(*args, **kwargs)
monkeypatch.setattr(dingtalk_stream, "DingTalkStreamClient", _GatedStreamClient)
async def go():
async with LocalHttpServer(body={"endpoint": "ws://127.0.0.1:1", "ticket": "ticket"}) as tickets:
monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url)
channel = self._channel()
await channel.start()
thread = channel._thread
stop_task = asyncio.create_task(channel.stop())
await asyncio.sleep(0.05)
assert not stop_task.done()
gate.set()
await stop_task
assert not thread.is_alive()
assert channel._thread is None
assert tickets.requests == []
_run(go())
def test_stop_joins_stream_thread_off_the_event_loop(self):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
release = threading.Event()
# dingtalk-stream's start_forever() never returns, so the join in
# stop() waits out its timeout; this thread blocks the same way,
# bounded so a join run on the event loop fails the assertion
# below instead of hanging the test.
stream_thread = threading.Thread(target=release.wait, args=(2,), daemon=True)
stream_thread.start()
channel._thread = stream_thread
channel._running = True
stop_task = asyncio.create_task(channel.stop())
await asyncio.sleep(0.05)
assert not stop_task.done()
release.set()
await stop_task
assert channel._thread is None
_run(go())
# ---------------------------------------------------------------------------
# Inbound file support
# ---------------------------------------------------------------------------
def _make_picture_message(*, download_code: str = "dc_img", **kwargs):
"""Build a mock DingTalk ``picture`` message carrying an ImageContent."""
msg = _make_chatbot_message(text="", message_type="picture", **kwargs)
msg.image_content = SimpleNamespace(download_code=download_code)
return msg
def _make_file_message(*, download_code: str = "dc_file", file_name: str | None = "report.xlsx", **kwargs):
"""Build a mock DingTalk ``file`` (document) message.
dingtalk_stream does not parse ``file`` messages, so the descriptor lives on
the raw callback payload the handler stashes as ``_df_raw_data``.
"""
msg = _make_chatbot_message(text="", message_type="file", **kwargs)
content: dict = {"downloadCode": download_code}
if file_name is not None:
content["fileName"] = file_name
msg._df_raw_data = {"msgtype": "file", "content": content}
return msg
def _patch_uploads(monkeypatch, uploads_dir, *, sandbox_id="local", sandbox=None):
monkeypatch.setattr(
"app.channels.dingtalk.get_paths",
lambda: SimpleNamespace(
ensure_thread_dirs=lambda thread_id, user_id=None: None,
sandbox_uploads_dir=lambda thread_id, user_id=None: uploads_dir,
),
)
monkeypatch.setattr("app.channels.dingtalk.get_effective_user_id", lambda: "default")
async def _acquire_async(thread_id, user_id=None):
return sandbox_id
provider = SimpleNamespace(
uses_thread_data_mounts=sandbox_id == "local" or sandbox_id.startswith("local:"),
acquire_async=_acquire_async,
get=lambda sid: sandbox,
release=lambda sid: None,
)
monkeypatch.setattr("app.channels.dingtalk.get_sandbox_provider", lambda: provider)
return provider
@pytest.fixture
def non_local_upload(tmp_path, monkeypatch):
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"BYTES")
uploads = tmp_path / "uploads"
uploads.mkdir()
sandbox = SimpleNamespace(update_file=MagicMock(), release_command_scope=MagicMock())
provider = _patch_uploads(monkeypatch, uploads, sandbox_id="aio:box1", sandbox=sandbox)
provider.release = MagicMock()
message = channel._make_inbound(
chat_id="c",
user_id="u",
text="hi",
thread_ts="m",
files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}],
)
try:
yield SimpleNamespace(channel=channel, provider=provider, sandbox=sandbox, message=message)
finally:
discard_sandbox_lease_manager(provider)
class TestExtractFiles:
def test_text_message_has_no_files(self):
assert DingTalkChannel._extract_files(_make_chatbot_message(text="hi")) == []
def test_picture_message(self):
files = DingTalkChannel._extract_files(_make_picture_message(download_code="dc_1"))
assert files == [{"type": "image", "download_code": "dc_1", "filename": "image.png"}]
def test_picture_message_missing_download_code_ignored(self):
msg = _make_chatbot_message(text="", message_type="picture")
msg.image_content = SimpleNamespace(download_code=None)
assert DingTalkChannel._extract_files(msg) == []
def test_rich_text_images(self):
msg = _make_chatbot_message(text="", message_type="richText")
msg.get_image_list = lambda: ["dc_a", "dc_b"]
files = DingTalkChannel._extract_files(msg)
assert files == [
{"type": "image", "download_code": "dc_a", "filename": "image_0.png"},
{"type": "image", "download_code": "dc_b", "filename": "image_1.png"},
]
def test_rich_text_get_image_list_failure_is_logged_not_silent(self, caplog):
"""An SDK parse failure must be distinguishable from "no inline images"."""
msg = _make_chatbot_message(text="", message_type="richText")
def _boom():
raise RuntimeError("sdk failure")
msg.get_image_list = _boom
with caplog.at_level(logging.WARNING, logger="app.channels.dingtalk"):
assert DingTalkChannel._extract_files(msg) == []
assert any("failed to read inline images" in r.message for r in caplog.records)
def test_file_message_from_raw_data(self):
files = DingTalkChannel._extract_files(_make_file_message(download_code="dc_doc", file_name="报价.xlsx"))
assert files == [{"type": "file", "download_code": "dc_doc", "filename": "报价.xlsx"}]
def test_file_message_default_filename(self):
files = DingTalkChannel._extract_files(_make_file_message(download_code="dc_doc", file_name=None))
assert files == [{"type": "file", "download_code": "dc_doc", "filename": "file.bin"}]
def test_file_message_without_raw_data_ignored(self):
msg = _make_chatbot_message(text="", message_type="file")
assert DingTalkChannel._extract_files(msg) == []
class TestOnChatbotMessageFiles:
def test_picture_message_publishes_inbound_with_files(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(_make_picture_message(download_code="dc_pic", sender_staff_id="u1", message_id="m1"))
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.msg_type == InboundMessageType.CHAT
assert inbound.files == [{"type": "image", "download_code": "dc_pic", "filename": "image.png"}]
_run(go())
def test_file_message_publishes_inbound_with_files(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
channel._send_running_reply = AsyncMock()
channel._on_chatbot_message(_make_file_message(download_code="dc_doc", file_name="data.csv", sender_staff_id="u1", message_id="m1"))
await asyncio.sleep(0.1)
inbound = await _take_inbound(bus)
assert inbound.files == [{"type": "file", "download_code": "dc_doc", "filename": "data.csv"}]
_run(go())
def test_message_without_text_or_files_dropped(self):
async def go():
bus = MessageBus()
bus.publish_inbound = AsyncMock()
channel = DingTalkChannel(bus, config={})
channel._client_id = "test_key"
channel._main_loop = asyncio.get_event_loop()
channel._running = True
# picture message whose image carries no download_code -> no files, no text
msg = _make_chatbot_message(text="", message_type="picture")
msg.image_content = SimpleNamespace(download_code=None)
channel._on_chatbot_message(msg)
await asyncio.sleep(0.1)
bus.publish_inbound.assert_not_awaited()
assert bus.inbound_queue.empty()
_run(go())
class TestReceiveFile:
def test_no_files_is_noop(self):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
msg = channel._make_inbound(chat_id="c", user_id="u", text="hi", thread_ts="m")
out = await channel.receive_file(msg, "t1")
assert out is msg
assert out.text == "hi"
_run(go())
def test_downloads_persists_and_prepends_path(self, tmp_path, monkeypatch):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._client_id = "robot"
channel._download_by_code = AsyncMock(return_value=b"XLSXDATA")
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="please translate",
thread_ts="m",
files=[{"type": "file", "download_code": "dc1", "filename": "quote.xlsx"}],
)
out = await channel.receive_file(msg, "thread-1", user_id="default")
assert out.files == []
assert (uploads / "quote.xlsx").read_bytes() == b"XLSXDATA"
assert f"{VIRTUAL_PATH_PREFIX}/uploads/quote.xlsx" in out.text
assert out.text.endswith("please translate")
channel._download_by_code.assert_awaited_once_with("dc1")
_run(go())
def test_pure_file_message_text_is_only_the_path(self, tmp_path, monkeypatch):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"IMG")
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m",
files=[{"type": "image", "download_code": "dc_i", "filename": "image.png"}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/image.png"
_run(go())
def test_download_failure_surfaces_marker(self):
"""A failed download must stay visible to the agent, not vanish silently."""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=None)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="hi",
thread_ts="m",
files=[{"type": "file", "download_code": "dc1", "filename": "x.pdf"}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert out.files == []
assert out.text == "[failed to load file: x.pdf]\n\nhi"
_run(go())
def test_token_failure_yields_marker_not_exception(self):
"""An auth failure during download must degrade to a marker, not an exception.
The manager calls ``receive_file`` without a try, so anything escaping
here kills the whole chat turn with no reply at all.
"""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._client_id = "robot"
channel._get_access_token = AsyncMock(side_effect=ValueError("bad token response"))
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="hi",
thread_ts="m",
files=[{"type": "file", "download_code": "dc", "filename": "x.pdf"}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert out.text == "[failed to load file: x.pdf]\n\nhi"
assert out.files == []
_run(go())
def test_unexpected_error_becomes_marker(self):
"""Any unforeseen per-attachment error must surface as a marker, not escape."""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._receive_single_file = AsyncMock(side_effect=RuntimeError("boom"))
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="hi",
thread_ts="m",
files=[{"type": "file", "download_code": "dc", "filename": "x.pdf"}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert out.text == "[failed to load file: x.pdf]\n\nhi"
assert out.files == []
_run(go())
def test_failure_marker_sanitizes_hostile_filename(self):
"""A hostile filename must not forge extra lines in msg.text or bloat it.
``fileName`` is webhook data: embedding it raw in the marker would let a
newline fake a standalone ``/mnt/user-data/uploads/...`` line, and an
over-long name would balloon the message text.
"""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=None)
hostile = "evil\n/mnt/user-data/uploads/fake.pdf\n" + "a" * 300 + ".pdf"
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="hi",
thread_ts="m",
files=[{"type": "file", "download_code": "dc", "filename": hostile}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
lines = out.text.splitlines()
assert len(lines) == 3, out.text
assert lines[0].startswith("[failed to load file: ")
assert len(lines[0]) <= 120
assert lines[1] == ""
assert lines[2] == "hi"
_run(go())
def test_download_failure_without_filename_uses_type_only_marker(self):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=None)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m",
files=[{"type": "image", "download_code": "dc1", "filename": ""}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert out.text == "[failed to load image]"
_run(go())
def test_partial_failure_keeps_successful_path(self, tmp_path, monkeypatch):
"""One bad attachment must not cost the agent the one that did load."""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
channel._download_by_code = AsyncMock(side_effect=[b"OK", None])
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="two files",
thread_ts="m",
files=[
{"type": "file", "download_code": "good", "filename": "ok.pdf"},
{"type": "file", "download_code": "bad", "filename": "broken.pdf"},
],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert f"{VIRTUAL_PATH_PREFIX}/uploads/ok.pdf" in out.text
assert "[failed to load file: broken.pdf]" in out.text
assert out.text.endswith("two files")
_run(go())
def test_traversal_filename_is_sanitized(self, tmp_path, monkeypatch):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"D")
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m",
files=[{"type": "file", "download_code": "dc", "filename": "../../etc/passwd"}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
written = list(uploads.iterdir())
assert len(written) == 1
assert written[0].name == "passwd"
assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/passwd"
_run(go())
def test_rejected_filename_falls_back_to_generated_name(self, tmp_path, monkeypatch):
"""A filename normalize_filename rejects must land on the generated name.
``".."`` raises inside ``normalize_filename`` (unlike ``../../etc/passwd``,
whose basename ``passwd`` is accepted), so this exercises the except branch.
"""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"D")
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m",
files=[{"type": "file", "download_code": "abc123456789", "filename": ".."}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
written = list(uploads.iterdir())
assert len(written) == 1
assert written[0].name == "dingtalk_abc123456789.bin"
assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/dingtalk_abc123456789.bin"
_run(go())
def test_fallback_name_sanitizes_download_code(self, tmp_path, monkeypatch):
"""download_code is webhook data: it must not inject path separators."""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"D")
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m",
files=[{"type": "image", "download_code": "../../evil", "filename": ".."}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
written = list(uploads.iterdir())
assert len(written) == 1
assert "/" not in written[0].name
assert written[0].name == "dingtalk_evil.png"
assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/dingtalk_evil.png"
_run(go())
def test_second_picture_does_not_overwrite_first(self, tmp_path, monkeypatch):
"""Two picture messages both generate "image.png" and must not collide.
The path handed to the agent must still hold the bytes it referred to.
"""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
channel._download_by_code = AsyncMock(return_value=b"FIRST")
first = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m1",
files=[{"type": "image", "download_code": "dc1", "filename": "image.png"}],
)
out_first = await channel.receive_file(first, "t1", user_id="default")
channel._download_by_code = AsyncMock(return_value=b"SECOND")
second = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m2",
files=[{"type": "image", "download_code": "dc2", "filename": "image.png"}],
)
out_second = await channel.receive_file(second, "t1", user_id="default")
assert out_first.text != out_second.text
first_name = out_first.text.rsplit("/", 1)[-1]
second_name = out_second.text.rsplit("/", 1)[-1]
# Each advertised path must still resolve to its own bytes.
assert (uploads / first_name).read_bytes() == b"FIRST"
assert (uploads / second_name).read_bytes() == b"SECOND"
_run(go())
def test_multiple_images_in_one_message_get_distinct_paths(self, tmp_path, monkeypatch):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
channel._download_by_code = AsyncMock(side_effect=[b"A", b"B"])
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m",
files=[
{"type": "image", "download_code": "dc1", "filename": "image_0.png"},
{"type": "image", "download_code": "dc2", "filename": "image_0.png"},
],
)
out = await channel.receive_file(msg, "t1", user_id="default")
paths = out.text.split("\n")
assert len(paths) == 2
assert paths[0] != paths[1]
assert len(list(uploads.iterdir())) == 2
assert {p.read_bytes() for p in uploads.iterdir()} == {b"A", b"B"}
_run(go())
def test_duplicate_document_names_do_not_collide(self, tmp_path, monkeypatch):
"""The same real filename sent twice must not clobber the earlier upload."""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads)
channel._download_by_code = AsyncMock(return_value=b"V1")
m1 = channel._make_inbound(chat_id="c", user_id="u", text="", thread_ts="m1", files=[{"type": "file", "download_code": "d1", "filename": "quote.xlsx"}])
await channel.receive_file(m1, "t1", user_id="default")
channel._download_by_code = AsyncMock(return_value=b"V2")
m2 = channel._make_inbound(chat_id="c", user_id="u", text="", thread_ts="m2", files=[{"type": "file", "download_code": "d2", "filename": "quote.xlsx"}])
await channel.receive_file(m2, "t1", user_id="default")
assert (uploads / "quote.xlsx").read_bytes() == b"V1"
assert len(list(uploads.iterdir())) == 2
_run(go())
def test_write_reserves_planted_symlink_name(self, tmp_path, monkeypatch):
"""A symlink planted at the requested name must force a unique suffix.
Upload dirs can be mounted into local sandboxes, so a sandbox process can
leave a symlink at a future upload name; following it would let a
gateway-privileged write land outside the bucket. The symlink name is
treated as occupied so the attachment still loads at the next suffix.
"""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
uploads = tmp_path / "uploads"
uploads.mkdir()
outside = tmp_path / "outside.txt"
symlink_or_skip(uploads / "image.png", outside)
_patch_uploads(monkeypatch, uploads)
channel._download_by_code = AsyncMock(return_value=b"PWNED")
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m",
files=[{"type": "image", "download_code": "dc", "filename": "image.png"}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert not outside.exists()
assert (uploads / "image.png").is_symlink()
assert (uploads / "image_1.png").read_bytes() == b"PWNED"
assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/image_1.png"
_run(go())
def test_missing_sandbox_after_acquire_yields_marker(self, tmp_path, monkeypatch):
"""A non-local sandbox that cannot be resolved must not yield a path.
The agent's sandbox cannot see the file in that case, so handing it the
virtual path would point at nothing readable; mirror Feishu and surface
a failed-load marker instead.
"""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"BYTES")
uploads = tmp_path / "uploads"
uploads.mkdir()
_patch_uploads(monkeypatch, uploads, sandbox_id="aio:box1", sandbox=None)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="hi",
thread_ts="m",
files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert out.text == "[failed to load file: a.pdf]\n\nhi"
_run(go())
def test_update_file_failure_yields_marker(self, tmp_path, monkeypatch):
"""A failed non-local sandbox sync must not yield a path either.
Same failure mode as the missing-sandbox case: the bytes never reached
the agent's sandbox, so the virtual path would read as nothing. Mirrors
Feishu, whose sync except-branch returns the failure marker.
"""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"BYTES")
uploads = tmp_path / "uploads"
uploads.mkdir()
broken_sandbox = MagicMock()
broken_sandbox.update_file.side_effect = RuntimeError("sandbox transport down")
_patch_uploads(monkeypatch, uploads, sandbox_id="aio:box1", sandbox=broken_sandbox)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="hi",
thread_ts="m",
files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}],
)
out = await channel.receive_file(msg, "t1", user_id="default")
assert out.text == "[failed to load file: a.pdf]\n\nhi"
_run(go())
def test_non_local_sandbox_is_synced(self, tmp_path, monkeypatch):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"BYTES")
uploads = tmp_path / "uploads"
uploads.mkdir()
fake_sandbox = MagicMock()
_patch_uploads(monkeypatch, uploads, sandbox_id="aio:box1", sandbox=fake_sandbox)
msg = channel._make_inbound(
chat_id="c",
user_id="u",
text="",
thread_ts="m",
files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}],
)
await channel.receive_file(msg, "t1", user_id="default")
fake_sandbox.update_file.assert_called_once_with(f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf", b"BYTES")
_run(go())
def test_non_local_upload_releases_standalone_sandbox(self, non_local_upload):
async def go():
upload = non_local_upload
result = await asyncio.wait_for(upload.channel.receive_file(upload.message, "t1", user_id="default"), timeout=5)
assert result.text == f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf\n\nhi"
upload.sandbox.update_file.assert_called_once_with(f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf", b"BYTES")
upload.sandbox.release_command_scope.assert_called_once()
upload.provider.release.assert_called_once_with("aio:box1")
_run(go())
@pytest.mark.parametrize("run_finishes_first", [False, True])
def test_non_local_upload_releases_only_after_upload_and_run_finish(self, non_local_upload, run_finishes_first):
async def go():
upload = non_local_upload
manager = get_sandbox_lease_manager(upload.provider)
await manager.retain_async("active-run", "aio:box1", thread_id="t1", user_id="default")
started = asyncio.Event()
finish_update = threading.Event()
update_finished = threading.Event()
loop = asyncio.get_running_loop()
def update_file(path, content):
loop.call_soon_threadsafe(started.set)
try:
if not finish_update.wait(timeout=10):
raise TimeoutError("test did not release sandbox update")
finally:
update_finished.set()
upload.sandbox.update_file.side_effect = update_file
task = asyncio.create_task(upload.channel.receive_file(upload.message, "t1", user_id="default"))
try:
await asyncio.wait_for(started.wait(), timeout=5)
upload.provider.release.assert_not_called()
assert not update_finished.is_set()
if run_finishes_first:
await manager.release_async("active-run")
upload.provider.release.assert_not_called()
assert not update_finished.is_set()
finish_update.set()
result = await asyncio.wait_for(task, timeout=5)
assert result.text == f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf\n\nhi"
assert update_finished.is_set()
if not run_finishes_first:
upload.provider.release.assert_not_called()
assert manager.binding_for("active-run") == "aio:box1"
await manager.release_async("active-run")
upload.provider.release.assert_called_once_with("aio:box1")
assert manager.binding_for("active-run") is None
assert upload.sandbox.release_command_scope.call_count == 2
finally:
finish_update.set()
await asyncio.gather(task, return_exceptions=True)
await manager.release_async("active-run")
_run(go())
def test_failed_non_local_upload_releases_standalone_sandbox(self, non_local_upload):
async def go():
upload = non_local_upload
upload.sandbox.update_file.side_effect = RuntimeError("sandbox transport down")
result = await asyncio.wait_for(upload.channel.receive_file(upload.message, "t1", user_id="default"), timeout=5)
assert result.text == "[failed to load file: a.pdf]\n\nhi"
upload.sandbox.update_file.assert_called_once_with(f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf", b"BYTES")
upload.sandbox.release_command_scope.assert_called_once()
upload.provider.release.assert_called_once_with("aio:box1")
_run(go())
class TestDownloadByCode:
def test_two_step_download(self):
async def go():
from unittest.mock import patch
channel = DingTalkChannel(MessageBus(), config={})
channel._client_id = "robot_x"
channel._get_access_token = AsyncMock(return_value="tok")
class PostResponse:
status_code = 200
@staticmethod
def json():
return {"downloadUrl": "https://dl.dingtalk/xyz"}
captured: dict = {}
class FakeStream:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
def raise_for_status(self):
pass
async def aiter_bytes(self):
yield b"FILE"
yield b"BYTES"
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
captured["post_url"] = url
captured["json"] = kwargs.get("json")
return PostResponse()
def stream(self, method, url, **kwargs):
captured["get_url"] = url
return FakeStream()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
data = await channel._download_by_code("dl_code")
assert data == b"FILEBYTES"
assert captured["post_url"].endswith("/v1.0/robot/messageFiles/download")
assert captured["json"] == {"downloadCode": "dl_code", "robotCode": "robot_x"}
assert captured["get_url"] == "https://dl.dingtalk/xyz"
_run(go())
def test_oversized_download_is_dropped(self, monkeypatch):
"""Inbound bytes are buffered in memory; a file over the cap must be refused.
Outbound uploads already enforce a size cap — without an inbound one, a
single large chat attachment balloons gateway memory.
"""
async def go():
from unittest.mock import patch
channel = DingTalkChannel(MessageBus(), config={})
channel._client_id = "robot_x"
channel._get_access_token = AsyncMock(return_value="tok")
monkeypatch.setattr("app.channels.dingtalk._MAX_INBOUND_FILE_SIZE_BYTES", 10)
class PostResponse:
status_code = 200
@staticmethod
def json():
return {"downloadUrl": "https://dl.dingtalk/big"}
class FakeStream:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
def raise_for_status(self):
pass
async def aiter_bytes(self):
for _ in range(4):
yield b"x" * 8
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
return PostResponse()
def stream(self, method, url, **kwargs):
return FakeStream()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
assert await channel._download_by_code("dl") is None
_run(go())
def test_post_non_200_returns_none(self):
async def go():
from unittest.mock import patch
channel = DingTalkChannel(MessageBus(), config={})
channel._client_id = "robot_x"
channel._get_access_token = AsyncMock(return_value="tok")
class PostResponse:
status_code = 403
text = "forbidden"
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
return PostResponse()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
assert await channel._download_by_code("dl") is None
_run(go())
def test_missing_download_url_returns_none(self):
async def go():
from unittest.mock import patch
channel = DingTalkChannel(MessageBus(), config={})
channel._client_id = "robot_x"
channel._get_access_token = AsyncMock(return_value="tok")
class PostResponse:
status_code = 200
@staticmethod
def json():
return {}
class FakeClient:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def post(self, url, **kwargs):
return PostResponse()
with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()):
assert await channel._download_by_code("dl") is None
_run(go())
class TestHandlerStashesRawData:
def test_process_stashes_raw_callback_payload(self):
pytest.importorskip("dingtalk_stream")
async def go():
bus = MessageBus()
channel = DingTalkChannel(bus, config={})
captured: dict = {}
channel._on_chatbot_message = lambda m: captured.update(msg=m)
handler = _DingTalkMessageHandler(channel)
cb = MagicMock()
cb.data = {
"msgtype": "file",
"content": {"downloadCode": "dc_doc", "fileName": "a.xlsx"},
"senderStaffId": "u1",
"conversationType": "1",
"msgId": "m1",
}
await handler.process(cb)
msg = captured["msg"]
assert getattr(msg, "_df_raw_data", None) == cb.data
# _extract_files can now read the document descriptor from the stash.
assert DingTalkChannel._extract_files(msg) == [{"type": "file", "download_code": "dc_doc", "filename": "a.xlsx"}]
_run(go())
class TestDingTalkDownloadGuardLogging:
"""A None (or empty) download result used to drop the attachment with
zero log lines at the receive level — the accurate reason lines inside
_download_by_code fired, but nothing tied them to the file being
received. The caller now logs a neutral guard line, mirroring the
wechat channel's callers and the manager reader."""
def test_none_download_logs_neutral_guard_line(self, caplog):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=None)
with caplog.at_level(logging.WARNING, logger="app.channels.dingtalk"):
result = await channel._receive_single_file("dc1", "file", "report.pdf", "t1", user_id="default")
assert result == ""
_run(go())
assert any("inbound file download returned no content" in r.message and "report.pdf" in r.message for r in caplog.records)
def test_empty_download_logs_neutral_guard_line(self, caplog):
async def go():
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"")
with caplog.at_level(logging.WARNING, logger="app.channels.dingtalk"):
result = await channel._receive_single_file("dc2", "image", "photo.png", "t1", user_id="default")
assert result == ""
_run(go())
assert any("inbound file download returned no content" in r.message and "photo.png" in r.message for r in caplog.records)
Reference in New Issue View Git Blame Copy Permalink