github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-27 08:28:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(dingtalk): support inbound file and image attachments (#4423)

Browse Source 
* feat(dingtalk): support inbound file and image attachments

DingTalk previously dropped picture and file (document) messages because
`_on_chatbot_message` ignored any message with empty text, so users could
not send files to the agent. This adds inbound attachment support, mirroring
`FeishuChannel`:

- `_extract_files` parses `picture`/`richText` image downloadCodes and `file`
  (document) descriptors. `dingtalk_stream.ChatbotMessage.from_dict` does not
  parse `file` messages, so `_DingTalkMessageHandler.process` stashes the raw
  callback payload on the message (`_df_raw_data`) for the document descriptor.
- `receive_file` downloads each attachment by `downloadCode` via the robot
  `messageFiles/download` OpenAPI, persists it into the thread uploads bucket,
  syncs it into a non-local sandbox, and prepends the sandbox virtual path to
  the message text so the agent can read the file by path.
- Filenames go through the shared `uploads.normalize_filename` helper, which
  strips directory components and rejects traversal patterns.

Outbound `send_file` already existed; this completes DingTalk file parity with
Feishu on the inbound side. Adds 21 tests covering extraction, download-by-code,
persistence/sandbox sync, filename sanitization, and the handler raw-data stash.

* fix(dingtalk): address inbound-file review feedback

Follow-up to the review on #4423:

- Make the fallback filename safe by construction. `download_code` is
  attacker-controllable webhook data and was embedded into `fallback_name`
  unsanitized; it only avoided escaping the uploads directory because the
  resulting write failed with OSError. It is now restricted to
  `[A-Za-z0-9_-]` before use. Covered by a test that reproduces the old
  behaviour (`uploads/dingtalk_../../evil.png`) and by a test that actually
  exercises the previously untested `except ValueError` branch (`".."`,
  whose basename — unlike `../../etc/passwd` — does raise).
- Log the swallowed `get_image_list()` failure instead of silently returning
  no images, so an SDK parse failure is distinguishable from a richText
  message that genuinely has no inline images.
- Surface failed downloads to the agent as a short `[failed to load ...]`
  marker rather than silently omitting the attachment, so a user whose file
  did not load does not simply appear to be ignored. Keeps the cleaner text
  shape while restoring the signal Feishu provides.

Tests: 119 passed (was 115).

* fix(dingtalk): claim unique upload names and refuse symlinked destinations

Round 2 review follow-up on #4423. Both findings reproduce as failing tests
against the previous head.

- Inbound attachments no longer overwrite each other. Generated names repeat
  across messages (every picture message yields "image.png", richText yields
  "image_0.png"), so a later attachment silently replaced an earlier one whose
  virtual path had already been prepended to the message text — the agent could
  read bytes that were not the ones its prompt referenced. The destination name
  is now claimed with the shared `claim_unique_filename` against the live
  directory contents, which also covers a real filename sent twice
  (`quote.xlsx`), a case Feishu's inline naming does not handle either. The
  claim and the write happen under one lock so two attachments cannot resolve
  to the same free name.
- Writes go through the shared `write_upload_file_no_symlink` instead of
  `Path.write_bytes`. Uploads dirs may be mounted into local sandboxes, so a
  sandbox process could leave a symlink at a future upload name and redirect a
  gateway-privileged write outside the bucket; the regression test shows the
  old code creating the out-of-bucket target.

Tests: 123 passed (was 119).

* fix(dingtalk): harden the inbound download path (self-audit)

Proactive hardening pass over the new inbound path; each fix reproduces as a
failing test against the previous head.

- Contain token failures. `_get_access_token()` sat outside the try in
  `_download_by_code`, and the manager awaits `receive_file` without one — a
  DingTalk auth hiccup during a file message aborted the whole chat turn with
  no reply. Token acquisition moves inside the try, and `receive_file` gains
  per-attachment isolation so no unforeseen error can escape past the marker.
- Cap inbound size. The download buffered arbitrary bytes in memory
  (`response.content`) with no limit, while outbound uploads already enforce
  one. The body is now streamed and dropped once it exceeds
  `_MAX_INBOUND_FILE_SIZE_BYTES` (50 MB), surfacing as a failed-load marker.
- Sanitize the failure marker. It embedded the raw webhook `fileName`; a
  newline could forge a standalone `/mnt/user-data/uploads/...` line inside
  msg.text and an over-long name bloated it. Markers now collapse whitespace
  and cap at 80 chars.
- Keep blocking IO off the event loop. `ensure_thread_dirs`, the uploads-dir
  resolve, sync `SandboxProvider.acquire`, and `sandbox.update_file` all ran on
  the loop; directory prep now lives inside the same `asyncio.to_thread` as the
  claim+write, and sandbox sync uses `acquire_async` + an offloaded
  `update_file`. Locked by a strict Blockbuster anchor
  (tests/blocking_io/test_dingtalk_receive_file.py), verified to fail with
  `BlockingError: Blocking call to os.mkdir` when the offload is reverted.

Tests: 127 + 1 blocking-io anchor (was 123); tests/blocking_io/ suite 55 passed.

* fix(dingtalk): surface missing-sandbox sync as a failed load

Round 3 follow-up on #4423:

- When a non-local sandbox acquire succeeds but the provider cannot resolve
  the instance, _receive_single_file returned the virtual path anyway — a
  path the agent's sandbox cannot read. Mirror Feishu: log and return "",
  so the [failed to load ...] marker fires instead. Red-first test:
  test_missing_sandbox_after_acquire_yields_marker.
- Drop the dead GetResponse / FakeClient.get scaffolding left in
  test_oversized_download_is_dropped from its red-first iteration.

Tests: 128 + 1 blocking-io anchor (was 127 + 1).

* fix(dingtalk): treat non-local sandbox sync failure as a failed load

Round 4 follow-up on #4423. The sync except-branch logged and still returned
the virtual path when acquire or update_file raised on a non-local sandbox —
the same handing-the-agent-an-unreadable-path failure mode the sandbox-is-None
branch was just fixed for, and exactly the leg the suite did not exercise.
Feishu's except-branch returns its failure marker; DingTalk now does the
equivalent (return "" so the failed-load marker fires). Red-first test:
test_update_file_failure_yields_marker.

Tests: 129 + 1 blocking-io anchor (was 128 + 1).

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Fgoll 2026-07-26 23:19:39 -07:00 committed by GitHub
parent 6091ce7576
commit 62b73fd2ea
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 1160 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/AGENTS.md
View File

@ -729,7 +729,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
A swallowed streaming failure publishes its final outbound before releasing the inbound dedupe key, so a provider redelivery can retry without overtaking the terminal reply.
- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle)
- `service.py` - Manages lifecycle of all configured channels from `config.yaml`
- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, and edits the "Working on it..." stream target in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured)
- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, and edits the "Working on it..." stream target in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`)
- `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR
- `app/gateway/routers/channel_connections.py` - Browser-facing user connection and disconnect APIs
- `deerflow.persistence.channel_connections` - SQL-backed user-owned connection, optional credential, connect state, and conversation store

266
backend/app/channels/dingtalk.py
View File

@ -17,6 +17,10 @@ from app.channels.base import Channel
from app.channels.commands import is_known_channel_command
from app.channels.connection_identity import attach_connection_identity
from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment
from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.sandbox.sandbox_provider import get_sandbox_provider
from deerflow.uploads.manager import UnsafeUploadPathError, claim_unique_filename, normalize_filename, write_upload_file_no_symlink
logger = logging.getLogger(__name__)
@ -29,6 +33,11 @@ _CONVERSATION_TYPE_GROUP = "2"
_MAX_UPLOAD_SIZE_BYTES = 20 * 1024 * 1024
# Inbound attachments are buffered in memory before being persisted and synced
# into the sandbox, so bound them. DingTalk chats accept far larger files than
# an agent can usefully read; oversized ones surface as a failed-load marker.
_MAX_INBOUND_FILE_SIZE_BYTES = 50 * 1024 * 1024
def _normalize_conversation_type(raw: Any) -> str:
"""Normalize ``conversationType`` to ``"1"`` (P2P) or ``"2"`` (group).
@ -63,6 +72,16 @@ def _is_dingtalk_command(text: str) -> bool:
return is_known_channel_command(text)
def _display_filename(filename: str) -> str:
"""Collapse whitespace and cap length for embedding a filename in message text.
``fileName`` is webhook-supplied data: embedded raw in a failed-load marker,
a newline could forge a standalone ``/mnt/user-data/uploads/...`` line and an
over-long name would bloat the message text.
"""
return re.sub(r"\s+", " ", filename).strip()[:80]
def _extract_text_from_rich_text(rich_text_list: list) -> str:
parts: list[str] = []
for item in rich_text_list:
@ -137,6 +156,9 @@ class DingTalkChannel(Channel):
self._incoming_messages: dict[str, Any] = {}
self._incoming_messages_lock = threading.Lock()
self._card_repliers: dict[str, Any] = {}
# Serialize inbound-file writes into the uploads directory to avoid
# racing writers clobbering one another (mirrors FeishuChannel).
self._file_write_lock = threading.Lock()
@property
def supports_streaming(self) -> bool:
@ -366,8 +388,9 @@ class DingTalkChannel(Channel):
sender_nick = message.sender_nick or ""
text = self._extract_text(message)
if not text:
logger.info("[DingTalk] empty text, ignoring message")
files = self._extract_files(message)
if not text and not files:
logger.info("[DingTalk] empty message with no files, ignoring")
return
connect_code = self._pending_connect_code(text)
@ -396,12 +419,13 @@ class DingTalkChannel(Channel):
# INFO logs, and only after the allowed_users gate so blocked senders are
# not logged at all.
logger.info(
"[DingTalk] parsed message: conv_type=%s, msg_id=%s, sender=%s(%s), text_len=%d",
"[DingTalk] parsed message: conv_type=%s, msg_id=%s, sender=%s(%s), text_len=%d, files=%d",
conversation_type,
msg_id,
sender_staff_id,
sender_nick,
len(text or ""),
len(files),
)
if _is_dingtalk_command(text):
@ -435,6 +459,7 @@ class DingTalkChannel(Channel):
text=text,
msg_type=msg_type,
thread_ts=msg_id,
files=files,
metadata={
"conversation_type": conversation_type,
"conversation_id": conversation_id,
@ -471,6 +496,237 @@ class DingTalkChannel(Channel):
return _extract_text_from_rich_text(message.rich_text_content.rich_text_list).strip()
return ""
# -- inbound: file attachments -----------------------------------------
@staticmethod
def _extract_files(message: Any) -> list[dict[str, Any]]:
"""Extract inbound file/image descriptors from a DingTalk message.
Returns a list of dicts shaped for ``InboundMessage.files``; each carries
``type`` (``"image"`` or ``"file"``), ``download_code``, and ``filename``.
Images arrive as ``picture`` messages (a single ``downloadCode``) or as
inline images inside a ``richText`` message. Documents arrive as ``file``
messages, which ``dingtalk_stream.ChatbotMessage.from_dict`` does not
parse — the descriptor (``downloadCode`` / ``fileName``) is read from the
raw callback payload stashed on the message as ``_df_raw_data`` by
``_DingTalkMessageHandler.process``.
"""
msg_type = getattr(message, "message_type", None)
files: list[dict[str, Any]] = []
if msg_type == "picture":
image_content = getattr(message, "image_content", None)
code = getattr(image_content, "download_code", None)
if code:
files.append({"type": "image", "download_code": code, "filename": "image.png"})
elif msg_type == "richText":
try:
codes = message.get_image_list() or []
except Exception:
# Log rather than swallow: without this, a richText message whose
# SDK image parse fails looks identical to one that simply has no
# inline images, which is hard to diagnose if the SDK shape changes.
logger.warning("[DingTalk] failed to read inline images from richText message", exc_info=True)
codes = []
for idx, code in enumerate(codes):
if code:
files.append({"type": "image", "download_code": code, "filename": f"image_{idx}.png"})
elif msg_type == "file":
raw = getattr(message, "_df_raw_data", None)
content = raw.get("content") if isinstance(raw, dict) else None
if isinstance(content, dict):
code = content.get("downloadCode")
if code:
filename = content.get("fileName") or "file.bin"
files.append({"type": "file", "download_code": code, "filename": filename})
return files
async def receive_file(self, msg: InboundMessage, thread_id: str, *, user_id: str | None = None) -> InboundMessage:
"""Download inbound DingTalk files into the thread uploads directory.
Mirrors :meth:`FeishuChannel.receive_file`: each descriptor in
``msg.files`` is downloaded by its ``downloadCode``, persisted under the
thread's uploads bucket, synced into a non-local sandbox, and its sandbox
virtual path is prepended to ``msg.text`` so downstream models can read
the file by path. Descriptors are then cleared so the generic
URL-based ``_ingest_inbound_files`` pass does not try to re-fetch them
(DingTalk files are downloaded by code, not by URL).
An attachment that fails to download contributes a short
``[failed to load ...]`` marker instead of a path, so the agent can tell
the user something was sent but could not be read rather than silently
ignoring it.
"""
if not msg.files:
return msg
virtual_paths: list[str] = []
failures: list[str] = []
for f in msg.files:
if not isinstance(f, dict):
continue
download_code = f.get("download_code")
if not download_code:
continue
file_type = "image" if f.get("type") == "image" else "file"
filename = f.get("filename") if isinstance(f.get("filename"), str) else ""
# Per-attachment isolation: the manager awaits receive_file without a
# try, so anything escaping here would kill the chat turn with no
# reply — the silent-drop failure mode this feature exists to fix.
try:
virtual_path = await self._receive_single_file(download_code, file_type, filename, thread_id, user_id=user_id)
except Exception:
logger.exception("[DingTalk] unexpected error receiving inbound %s", file_type)
virtual_path = ""
if virtual_path:
virtual_paths.append(virtual_path)
else:
display = _display_filename(filename)
failures.append(f"[failed to load {file_type}: {display}]" if display else f"[failed to load {file_type}]")
prefix_lines = virtual_paths + failures
if prefix_lines:
block = "\n".join(prefix_lines)
msg.text = f"{block}\n\n{msg.text}".strip() if msg.text else block
msg.files = []
return msg
async def _receive_single_file(
self,
download_code: str,
file_type: str,
filename: str,
thread_id: str,
*,
user_id: str | None = None,
) -> str:
"""Download one file by ``downloadCode`` and persist it into uploads.
Returns the sandbox virtual path on success, or ``""`` on failure.
"""
content = await self._download_by_code(download_code)
if not content:
return ""
paths = get_paths()
effective_user_id = user_id or get_effective_user_id()
default_ext = "png" if file_type == "image" else "bin"
# ``download_code`` is attacker-controllable webhook data, so restrict it to
# a safe character set before embedding it in the fallback name. This keeps
# the fallback safe by construction instead of relying on a later write
# failure to reject a name that escaped the uploads directory.
code_token = re.sub(r"[^A-Za-z0-9_-]", "", download_code)[-12:] or "attachment"
fallback_name = f"dingtalk_{code_token}.{default_ext}"
# normalize_filename strips directory components and rejects traversal
# patterns ("..", backslash paths, over-long names); fall back to the
# sanitized generated name if the platform-supplied filename is rejected.
try:
safe_filename = normalize_filename(filename or fallback_name)
except ValueError:
safe_filename = fallback_name
def _persist() -> Path:
# Directory prep, the uniqueness claim, and the write are blocking
# filesystem IO — the whole sequence stays off the event loop. The
# claim and the write share one lock because generated names repeat
# across messages ("image.png" for every picture message): without a
# claim a later attachment silently overwrites an earlier one whose
# path was already handed to the agent, and letting the claim and
# write interleave would resolve two attachments to the same free name.
paths.ensure_thread_dirs(thread_id, user_id=effective_user_id)
uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve()
with self._file_write_lock:
seen = {entry.name for entry in uploads_dir.iterdir() if entry.is_file()}
unique_name = claim_unique_filename(safe_filename, seen)
# write_upload_file_no_symlink refuses a symlinked destination:
# uploads dirs can be mounted into local sandboxes, so a sandbox
# process could otherwise redirect this privileged write outside
# the bucket.
return write_upload_file_no_symlink(uploads_dir, unique_name, content)
try:
resolved_target = await asyncio.to_thread(_persist)
except (OSError, UnsafeUploadPathError):
logger.exception("[DingTalk] failed to persist downloaded file: %s", safe_filename)
return ""
virtual_path = f"{VIRTUAL_PATH_PREFIX}/uploads/{resolved_target.name}"
try:
sandbox_provider = get_sandbox_provider()
# acquire_async keeps provider lifecycle work (Docker discovery,
# readiness polls) off the event loop; update_file is blocking
# transport IO on remote sandboxes, so it is offloaded too.
sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
if sandbox_id != "local":
sandbox = sandbox_provider.get(sandbox_id)
if sandbox is None:
# Mirror Feishu: the agent's non-local sandbox cannot see this
# file, so returning the virtual path would hand the model a
# path that reads as nothing — surface a failed-load marker.
logger.warning("[DingTalk] sandbox %s not found after acquire, dropping attachment: %s", sandbox_id, virtual_path)
return ""
await asyncio.to_thread(sandbox.update_file, virtual_path, content)
except Exception:
# Same failure mode as the sandbox-is-None branch: the bytes never
# reached the agent's sandbox, so the virtual path would read as
# nothing. Mirror Feishu and surface a failed-load marker.
logger.exception("[DingTalk] failed to sync downloaded file into non-local sandbox: %s", virtual_path)
return ""
return virtual_path
async def _download_by_code(self, download_code: str) -> bytes | None:
"""Exchange a DingTalk ``downloadCode`` for the raw file bytes.
Two steps per the DingTalk robot OpenAPI:
1. ``POST /v1.0/robot/messageFiles/download`` -> ``{downloadUrl}``
2. ``GET downloadUrl`` -> binary content
"""
try:
# Token acquisition stays inside the try: the manager awaits
# receive_file without a try, so a token failure escaping here would
# abort the whole chat turn instead of degrading to a failed-load marker.
token = await self._get_access_token()
async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/robot/messageFiles/download",
headers={
"x-acs-dingtalk-access-token": token,
"Content-Type": "application/json",
},
json={"downloadCode": download_code, "robotCode": self._client_id},
)
if response.status_code != 200:
logger.warning("[DingTalk] messageFiles/download failed: status=%d, body=%s", response.status_code, response.text[:300])
return None
download_url = response.json().get("downloadUrl")
if not download_url:
logger.warning("[DingTalk] messageFiles/download returned no downloadUrl")
return None
# Stream with a size cap: the bytes are buffered in memory before
# being persisted, so an oversized attachment must be refused
# before it is fully read, not after.
chunks: list[bytes] = []
total = 0
async with client.stream("GET", download_url, follow_redirects=True) as file_response:
file_response.raise_for_status()
async for chunk in file_response.aiter_bytes():
total += len(chunk)
if total > _MAX_INBOUND_FILE_SIZE_BYTES:
logger.warning("[DingTalk] inbound file exceeds %d bytes, dropping", _MAX_INBOUND_FILE_SIZE_BYTES)
return None
chunks.append(chunk)
return b"".join(chunks)
except (httpx.HTTPError, ValueError):
logger.exception("[DingTalk] failed to download file by code")
return None
async def _prepare_inbound(self, chat_id: str, inbound: InboundMessage) -> None:
inbound = await self._attach_connection_identity(inbound)
# Running reply must finish before publish_inbound so AI card tracks are
@ -831,5 +1087,9 @@ class _DingTalkMessageHandler:
import dingtalk_stream
incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
# Stash the raw callback payload: dingtalk_stream does not parse ``file``
# (document) messages, so DingTalkChannel._extract_files reads the file
# descriptor (downloadCode / fileName) from it.
incoming_message._df_raw_data = callback.data
self._channel._on_chatbot_message(incoming_message)
return dingtalk_stream.AckMessage.STATUS_OK, "OK"

56
backend/tests/blocking_io/test_dingtalk_receive_file.py Normal file
View File

@ -0,0 +1,56 @@
"""Regression anchor: DingTalk ``receive_file`` must not block the event loop.
``_receive_single_file`` prepares the thread directories, resolves the uploads
dir, scans it for the uniqueness claim, and writes the attachment — all blocking
filesystem IO that must run inside ``asyncio.to_thread`` (and sandbox sync must
go through ``acquire_async`` + an offloaded ``update_file``). This anchor drives
the real ``receive_file`` under the strict Blockbuster gate; if any of that
regresses back onto the event loop, Blockbuster raises ``BlockingError``.
The ``Paths`` construction is offloaded only because ``Paths.__init__`` resolves
paths synchronously; the surface under test (``receive_file``'s persist path) is
exercised on the event loop, not bypassed. The download itself is mocked — the
network leg is httpx-async and not the subject here.
"""
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
pytestmark = pytest.mark.asyncio
async def test_receive_file_persist_does_not_block_event_loop(tmp_path, monkeypatch) -> None:
from app.channels.dingtalk import DingTalkChannel
from app.channels.message_bus import MessageBus
from deerflow.config.paths import Paths
paths = await asyncio.to_thread(Paths, str(tmp_path))
monkeypatch.setattr("app.channels.dingtalk.get_paths", lambda: paths)
async def _acquire_async(thread_id, user_id=None):
return "local"
monkeypatch.setattr(
"app.channels.dingtalk.get_sandbox_provider",
lambda: SimpleNamespace(acquire_async=_acquire_async, get=lambda sid: None),
)
channel = DingTalkChannel(MessageBus(), config={})
channel._download_by_code = AsyncMock(return_value=b"DATA")
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 "/uploads/a.pdf" in out.text
assert out.files == []

840
backend/tests/test_dingtalk_channel.py
View File

@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
import logging
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
@ -22,6 +23,7 @@ from app.channels.dingtalk import (
_normalize_conversation_type,
)
from app.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage
from deerflow.config.paths import VIRTUAL_PATH_PREFIX
def _run(coro):
@ -1691,3 +1693,841 @@ class TestCardMode:
assert channel._card_track_ids == {}
_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
monkeypatch.setattr(
"app.channels.dingtalk.get_sandbox_provider",
lambda: SimpleNamespace(acquire_async=_acquire_async, get=lambda sid: sandbox),
)
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)
bus.publish_inbound.assert_awaited_once()
inbound = bus.publish_inbound.await_args.args[0]
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)
bus.publish_inbound.assert_awaited_once()
inbound = bus.publish_inbound.await_args.args[0]
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()
_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_does_not_follow_planted_symlink(self, tmp_path, monkeypatch):
"""A symlink planted at the destination must not be written through.
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.
"""
async def go():
channel = DingTalkChannel(MessageBus(), config={})
uploads = tmp_path / "uploads"
uploads.mkdir()
outside = tmp_path / "outside.txt"
(uploads / "image.png").symlink_to(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 "[failed to load image: image.png]" in out.text
_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())
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())