From 9060f56f40d3bc363a58ff34ab57cb715f357881 Mon Sep 17 00:00:00 2001 From: chichuan Date: Wed, 7 Oct 2026 15:20:23 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=8F=AF=E9=80=89=20?= =?UTF-8?q?DWS=20A2UI=20=E5=8D=A1=E7=89=87=E5=8F=91=E9=80=81=E4=B8=8E?= =?UTF-8?q?=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- A2UI.md | 70 +++++++++++ README.md | 5 + README.zh-CN.md | 5 + example/a2ui-card.json | 41 +++++++ example/a2ui-update.json | 10 ++ example/a2ui.py | 25 ++++ src/dingtalk_channel_sdk/__init__.py | 6 + src/dingtalk_channel_sdk/a2ui.py | 173 +++++++++++++++++++++++++++ src/dingtalk_channel_sdk/channel.py | 15 ++- src/dingtalk_channel_sdk/config.py | 3 + tests/test_a2ui.py | 111 +++++++++++++++++ 11 files changed, 463 insertions(+), 1 deletion(-) create mode 100644 A2UI.md create mode 100644 example/a2ui-card.json create mode 100644 example/a2ui-update.json create mode 100644 example/a2ui.py create mode 100644 src/dingtalk_channel_sdk/a2ui.py create mode 100644 tests/test_a2ui.py diff --git a/A2UI.md b/A2UI.md new file mode 100644 index 0000000..9b938d3 --- /dev/null +++ b/A2UI.md @@ -0,0 +1,70 @@ +# 发送 A2UI 卡片 + +本 SDK 参考 [dingtalk-aicard](https://github.com/DingTalk-Real-AI/dingtalk-aicard/tree/d2c2e7f05592a346e72486f0e41b744c1189f6fa) 的公开 DWS 接入方式，增加可选发送通道。A2UI 消息使用 `version: "v1.0"`；V0.8 是钉钉规范版本，不能替换消息版本。 + +## 发送身份与准备 + +`DwsA2UIClient`（Go 为 `DWSA2UIClient`）调用本机 DWS，用其登录账号和 Profile 发送。显式配置后才启用，发送身份与 Channel 的机器人应用 Token 独立。建议指定固定的 `corpId:userId` Profile，创建和更新始终使用同一个 Profile；适配器拒绝同时选择多个 Profile。 + +先单独安装并登录 [DingTalk Workspace CLI](https://github.com/DingTalk-Real-AI/dingtalk-workspace-cli)，确认所用版本有以下两个命令： + +```bash +dws chat message send-a2ui-card --help +dws chat message update-a2ui-card --help +``` + +默认执行 `dws`，可配置命令路径、固定前缀参数和执行超时（默认 30 秒）。SDK 使用参数数组执行，始终不调用 shell。DWS 接收一个 `--content` 参数，本适配器将其限制为 UTF-8 64 KiB，回执限制为 8 MiB。 + +新增发送通道可以独立使用，也可以注入已有 Channel 的配置。自定义 A2UI 通道需实现同样的发送、更新接口；这为其他已验证的投递集成保留扩展点。 + +## 消息与生命周期 + +消息是非空数组，元素可为 JSON 对象或已经序列化的 JSON 字符串。适配器将其转换为 DWS 要求的字符串数组，保留消息顺序和业务数据。SDK 检查 JSON、`version`、单一操作和非空 `surfaceId`；组件属性、引用、初始数据与资源需按 dingtalk-aicard 校验。 + +仓库中的 `example/a2ui-card.json` 是完整创建示例，`example/a2ui-update.json` 是同一 Surface 的完成增量。可在 dingtalk-aicard 仓库内设置 Python 环境并验证创建示例： + +```bash +python3 skills/dingtalk-aicard/scripts/setup_env.py +.aicard-venv/bin/python skills/dingtalk-aicard/scripts/aicard_lint.py /path/to/sdk/example/a2ui-card.json --preflight new-card --format json +``` + +组件与事件定义以 [钉钉规范](https://github.com/DingTalk-Real-AI/dingtalk-aicard/tree/d2c2e7f05592a346e72486f0e41b744c1189f6fa/spec) 为准。 + +单聊必须提供 `openDingTalkId`，群聊提供 `openConversationId`，选择一个目标。不要把机器人 `staffId`、普通 `userId` 或 `senderId` 当作个人开放标识；适配器不会自动转换这些标识。 + +创建使用 `send-a2ui-card`，状态为 `PROCESSING`。发送结果保留完整 DWS 回执，并提取服务端 `bizId`。缺少它时返回更新警告；不能把请求侧 `bizCardId` 或 `openTaskId` 当作更新标识，也不要自动再创建一张卡片。 + +更新使用 `update-a2ui-card`，必须传原卡片 `bizId`、非空增量和 `flowStatus`。保持原 `surfaceId`、组件 ID 和用户表单数据，发送完整 A2UI 消息，不发送 JSON token 片段。静态卡片通过状态 `FINISH` 和非空、保留原状态的增量结束。 + +状态支持 `PROCESSING`、`INPUTTING`、`FINISH`、`EXECUTING`、`ERROR`、`ABORTED`、`TIMEOUT`、`CONFIRMING`、`CONFIRMED`，兼容字符串 `1` 至 `9`。 + +## 验证与回调 + +只有回执明确确认接受请求才返回成功；执行错误、非 JSON 输出、无确认标志、dry-run 及失败回执会返回错误。写入超时或进程失败后，服务端可能已经接收请求；SDK 明确报告结果可能未知，并且不会自动重试或降级为 Markdown 消息。请保留回执并核实实际状态。 + +DWS 命令成功和本地测试不能证明钉钉客户端渲染或用户点击回调。本次适配器负责发送与更新；DWS 身份下的业务回调须通过相应的 `user_card_action_triggered` 事件通道另外接入，不会自动进入 Channel 的机器人 `cardAction` 回调。 + +参考仓库没有给出可复用的机器人应用 Token A2UI OpenAPI 实现。本适配器实现它已经公开的 DWS 路径；需要机器人原生发送时，应提供已验证的接口契约并实现可注入发送通道。 + +## SDK 调用 + +```python +import json +import os +from pathlib import Path +from dingtalk_channel_sdk import Config, DingTalkChannel, DwsA2UIClient + +ch = DingTalkChannel(config=Config( + client_id=os.environ["DD_CLIENT_ID"], + client_secret=os.environ["DD_CLIENT_SECRET"], + a2ui_client=DwsA2UIClient(profile=os.environ["DWS_PROFILE"]), +)) +messages = json.loads(Path("example/a2ui-card.json").read_text(encoding="utf-8")) +result = await ch.send_a2ui_card({"open_dingtalk_id": os.environ["DWS_OPEN_DINGTALK_ID"]}, messages) +if not result.biz_id: + raise RuntimeError(result.update_warning) +delta = json.loads(Path("example/a2ui-update.json").read_text(encoding="utf-8")) +await ch.update_a2ui_card(result.biz_id, delta, "FINISH") +``` + +在异步函数中执行以上调用。群聊使用 `{"conversation_id": ""}`。独立使用时直接调用 `DwsA2UIClient.send_card` 和 `update_card`，不需要机器人凭据。 diff --git a/README.md b/README.md index 9809b04..147e647 100644 --- a/README.md +++ b/README.md @@ -103,3 +103,8 @@ Live check: `DD_CLIENT_ID=... DD_CLIENT_SECRET=... python example/livecheck.py` ## License MIT + + +## A2UI 卡片 + +支持通过显式配置的 DWS 通道发送和更新 A2UI 卡片，使用 DWS 登录身份。接口、完整示例、消息校验和回调范围见 [A2UI 接入说明](A2UI.md)。 diff --git a/README.zh-CN.md b/README.zh-CN.md index 68ad5d7..f51ee58 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -103,3 +103,8 @@ pytest # 108 个测试 ## License MIT + + +## A2UI 卡片 + +支持通过显式配置的 DWS 通道发送和更新 A2UI 卡片，使用 DWS 登录身份。接口、完整示例、消息校验和回调范围见 [A2UI 接入说明](A2UI.md)。 diff --git a/example/a2ui-card.json b/example/a2ui-card.json new file mode 100644 index 0000000..a8e8000 --- /dev/null +++ b/example/a2ui-card.json @@ -0,0 +1,41 @@ +[ + { + "version": "v1.0", + "createSurface": { + "surfaceId": "channel-sdk-example", + "catalogId": "https://dingtalk.com/card/a2ui/catalogs/public/catalog.json" + } + }, + { + "version": "v1.0", + "updateDataModel": { + "surfaceId": "channel-sdk-example", + "path": "/", + "value": { + "text": "这是 Channel SDK 的 A2UI 示例。" + } + } + }, + { + "version": "v1.0", + "updateComponents": { + "surfaceId": "channel-sdk-example", + "components": [ + { + "id": "root", + "component": "Column", + "children": [ + "body" + ] + }, + { + "id": "body", + "component": "Markdown", + "content": { + "path": "/text" + } + } + ] + } + } +] diff --git a/example/a2ui-update.json b/example/a2ui-update.json new file mode 100644 index 0000000..9a6c2db --- /dev/null +++ b/example/a2ui-update.json @@ -0,0 +1,10 @@ +[ + { + "version": "v1.0", + "updateDataModel": { + "surfaceId": "channel-sdk-example", + "path": "/text", + "value": "示例已完成；这条消息更新原卡片的数据。" + } + } +] diff --git a/example/a2ui.py b/example/a2ui.py new file mode 100644 index 0000000..33964a4 --- /dev/null +++ b/example/a2ui.py @@ -0,0 +1,25 @@ +"""使用 DWS 的指定 Profile 发送示例卡片并完成原卡片，不需要机器人凭据。""" +import asyncio +import json +import os +from pathlib import Path +from dingtalk_channel_sdk import DwsA2UIClient + + +async def main(): + profile, recipient = os.environ.get("DWS_PROFILE"), os.environ.get("DWS_OPEN_DINGTALK_ID") + if not profile or not recipient: + raise RuntimeError("请设置 DWS_PROFILE 和 DWS_OPEN_DINGTALK_ID") + client = DwsA2UIClient(command=[os.environ.get("DWS_BIN", "dws")], profile=profile) + folder = Path(__file__).resolve().parent + messages = json.loads((folder / "a2ui-card.json").read_text(encoding="utf-8")) + result = await client.send_card({"open_dingtalk_id": recipient}, messages) + if not result.biz_id: + raise RuntimeError(result.update_warning) + delta = json.loads((folder / "a2ui-update.json").read_text(encoding="utf-8")) + await client.update_card(result.biz_id, delta, "FINISH") + print(f"示例卡片已完成，bizId={result.biz_id}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/dingtalk_channel_sdk/__init__.py b/src/dingtalk_channel_sdk/__init__.py index 7e6cc4f..e7f725d 100644 --- a/src/dingtalk_channel_sdk/__init__.py +++ b/src/dingtalk_channel_sdk/__init__.py @@ -4,6 +4,7 @@ from .bot_identity import BotIdentity, BotIdentityProvider from .card import ApiError as _ApiErrorAlias # noqa: F401（re-export 由 httpx 提供） from .channel import DingTalkChannel +from .a2ui import A2UIClient, A2UICardResult, DwsA2UIClient, A2UI_FLOW_STATUSES, serialize_a2ui_messages from .config import ( ChatQueueConfig, Config, @@ -29,6 +30,11 @@ __all__ = [ "DingTalkChannel", + "A2UIClient", + "A2UICardResult", + "DwsA2UIClient", + "A2UI_FLOW_STATUSES", + "serialize_a2ui_messages", "Config", "Reply", "OapiClient", diff --git a/src/dingtalk_channel_sdk/a2ui.py b/src/dingtalk_channel_sdk/a2ui.py new file mode 100644 index 0000000..2d95a32 --- /dev/null +++ b/src/dingtalk_channel_sdk/a2ui.py @@ -0,0 +1,173 @@ +"""按 dingtalk-aicard 的公开接入方式，通过显式配置的 DWS 身份发送 A2UI。""" +from __future__ import annotations + +import asyncio +import json +import math +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Protocol, Sequence, Union + +A2UI_FLOW_STATUSES = ( + "PROCESSING", "INPUTTING", "FINISH", "EXECUTING", "ERROR", + "ABORTED", "TIMEOUT", "CONFIRMING", "CONFIRMED", +) +_OPERATIONS = ("createSurface", "updateComponents", "updateDataModel", "deleteSurface") +_MISSING_ID = "回执未包含可用的 bizId；请保留回执并核实服务端标识，不要自动重发创建请求。" +Messages = Sequence[Union[str, Dict[str, Any]]] + + +def _reject_json_constant(value: str): + raise ValueError("JSON 不允许非有限数值") + + +def _identifier(value: Any, name: str) -> str: + if not isinstance(value, str) or not value.strip() or any(ord(c) < 32 or ord(c) == 127 for c in value): + raise ValueError(f"{name} 必须是非空标识") + return value.strip() + + +def serialize_a2ui_messages(messages: Messages) -> str: + """只检查消息信封；组件和完整创建状态使用 dingtalk-aicard 校验。""" + if not isinstance(messages, (list, tuple)) or not messages: + raise ValueError("A2UI 消息必须是非空数组") + strings: List[str] = [] + for index, message in enumerate(messages): + try: + encoded = message if isinstance(message, str) else json.dumps(message, ensure_ascii=False, allow_nan=False, separators=(",", ":")) + parsed = json.loads(encoded, parse_constant=_reject_json_constant) + except (TypeError, ValueError, OverflowError): + raise ValueError(f"A2UI 消息 {index} 不是有效 JSON") from None + if not isinstance(parsed, dict) or parsed.get("version") != "v1.0": + raise ValueError(f"A2UI 消息 {index} 必须是 version=v1.0 的对象") + keys = [key for key in _OPERATIONS if key in parsed] + if len(keys) != 1 or not isinstance(parsed[keys[0]], dict): + raise ValueError(f"A2UI 消息 {index} 必须包含一个操作") + _identifier(parsed[keys[0]].get("surfaceId"), "surfaceId") + strings.append(encoded) + content = json.dumps(strings, ensure_ascii=False, separators=(",", ":")) + if len(content.encode("utf-8")) > 65536: + raise ValueError("DWS A2UI content 超过 64 KiB") + return content + + +def _envelope_chain(receipt: Dict[str, Any]) -> List[Dict[str, Any]]: + chain = [] + value: Any = receipt + for _ in range(5): + if not isinstance(value, dict): + break + chain.append(value) + if (value.get("success") is False or value.get("ok") is False or value.get("isError") is True or value.get("error") + or value.get("dry_run") is True or value.get("dryRun") is True + or ("outcome" in value and value["outcome"] not in ("success", "pending"))): + raise RuntimeError("DWS 返回失败回执") + value = value.get("data") if value.get("data") is not None else value.get("result") + if not any(item.get("success") is True or item.get("ok") is True for item in chain): + raise RuntimeError("DWS 回执未明确确认接受请求；发送结果可能未知，请核实后再重试") + return chain + + +@dataclass(frozen=True) +class A2UICardResult: + biz_id: Optional[str] + receipt: Dict[str, Any] + update_warning: Optional[str] = None + + +class A2UIClient(Protocol): + """可注入发送通道；接收目标使用 open_dingtalk_id 或 conversation_id。""" + + async def send_card(self, target: Dict[str, str], messages: Messages) -> A2UICardResult: ... + async def update_card(self, biz_id: str, messages: Messages, flow_status: str) -> Dict[str, Any]: ... + + +class DwsA2UIClient: + """可选 DWS 发送通道。DWS 单独安装、登录，profile 决定发送身份。""" + + def __init__(self, command: Sequence[str] = ("dws",), profile: Optional[str] = None, timeout_s: float = 30.0): + if isinstance(command, str) or not command or any(not isinstance(part, str) or not part or "\0" in part for part in command): + raise ValueError("command 必须是非空字符串数组") + if not math.isfinite(timeout_s) or timeout_s <= 0: + raise ValueError("timeout_s 必须大于 0") + self.command = tuple(command) + self.profile = _identifier(profile, "profile") if profile is not None else None + if self.profile and "," in self.profile: + raise ValueError("A2UI 发送只允许一个 DWS Profile") + self.timeout_s = timeout_s + + async def _invoke(self, args: List[str]) -> Dict[str, Any]: + argv = list(self.command) + ["chat", "message"] + args + ["--format=json", "--yes"] + if self.profile: + argv.append(f"--profile={self.profile}") + try: + process = await asyncio.create_subprocess_exec(*argv, stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL) + except OSError: + raise RuntimeError("DWS 无法启动；请检查安装和 command 配置") from None + + async def read_output(): + chunks = [] + size = 0 + while True: + chunk = await process.stdout.read(65536) + if not chunk: + break + size += len(chunk) + if size > 8 * 1024 * 1024: + raise RuntimeError("DWS 回执超过 8 MiB；发送结果可能未知") + chunks.append(chunk) + await process.wait() + return b"".join(chunks) + + try: + stdout = await asyncio.wait_for(read_output(), timeout=self.timeout_s) + except (asyncio.TimeoutError, asyncio.CancelledError) as error: + if process.returncode is None: + process.kill() + await process.wait() + if isinstance(error, asyncio.CancelledError): + raise + raise RuntimeError("DWS 执行超时；发送结果可能未知，请核实后再重试") from None + except RuntimeError: + if process.returncode is None: + process.kill() + await process.wait() + raise + if process.returncode: + raise RuntimeError("DWS 执行失败；发送结果可能未知，请核实后再重试") + try: + receipt = json.loads(stdout) + except (ValueError, UnicodeDecodeError): + raise RuntimeError("DWS 输出不是 JSON；发送结果可能未知，请保留现场核实") from None + if not isinstance(receipt, dict): + raise RuntimeError("DWS 回执必须是 JSON 对象") + _envelope_chain(receipt) + return receipt + + async def send_card(self, target: Dict[str, str], messages: Messages) -> A2UICardResult: + if not isinstance(target, dict) or any(key not in ("open_dingtalk_id", "conversation_id") for key in target): + raise ValueError("A2UI target 使用 open_dingtalk_id 或 conversation_id，不接受 user_id") + dm, group = "open_dingtalk_id" in target, "conversation_id" in target + if dm == group: + raise ValueError("A2UI target 必须恰好选择一个接收目标") + flag = "open-dingtalk-id" if dm else "conversation-id" + target_id = _identifier(target["open_dingtalk_id" if dm else "conversation_id"], flag) + content = serialize_a2ui_messages(messages) + receipt = await self._invoke(["send-a2ui-card", f"--{flag}={target_id}", f"--content={content}"]) + biz_id = None + for value in reversed(_envelope_chain(receipt)): + try: + biz_id = _identifier(value.get("bizId"), "bizId") + break + except ValueError: + pass + return A2UICardResult(biz_id, receipt, None if biz_id else _MISSING_ID) + + async def update_card(self, biz_id: str, messages: Messages, flow_status: str) -> Dict[str, Any]: + card_id = _identifier(biz_id, "bizId") + status = str(flow_status).strip().upper() + if status in tuple(str(i) for i in range(1, 10)): + status = A2UI_FLOW_STATUSES[int(status) - 1] + if status not in A2UI_FLOW_STATUSES: + raise ValueError("不支持的 A2UI flow_status") + content = serialize_a2ui_messages(messages) + return await self._invoke(["update-a2ui-card", f"--biz-id={card_id}", f"--content={content}", f"--flow-status={status}"]) diff --git a/src/dingtalk_channel_sdk/channel.py b/src/dingtalk_channel_sdk/channel.py index 52abaa1..d1c4404 100644 --- a/src/dingtalk_channel_sdk/channel.py +++ b/src/dingtalk_channel_sdk/channel.py @@ -10,13 +10,14 @@ import time import urllib.request from datetime import timedelta -from typing import Any, Awaitable, Callable, List, Optional +from typing import Any, Awaitable, Callable, Dict, List, Optional from .compat import to_thread from .safety.batching import BatchConfig, BatchedMessage, MessageBatcher from .bot_identity import BotIdentity, BotIdentityProvider from .card import CardClient +from .a2ui import A2UICardResult, Messages from .config import TOPIC_BOT_MESSAGE, TOPIC_CARD_CALLBACK, TRANSPORT_HTTP, USER_AGENT, Config from .safety.dedup import Deduper from .emotion import Emotion @@ -215,6 +216,18 @@ async def send_text(self, target: SendTarget, content: str) -> None: async def send_markdown(self, target: SendTarget, title: str, text: str) -> None: await self.sender.send_markdown(target, title, text) + async def send_a2ui_card(self, target: Dict[str, str], messages: Messages) -> A2UICardResult: + """通过显式配置的 A2UI 通道发送，返回 biz_id 与完整回执。""" + if self.cfg.a2ui_client is None: + raise RuntimeError("请先配置 a2ui_client（例如 DwsA2UIClient）") + return await self.cfg.a2ui_client.send_card(target, messages) + + async def update_a2ui_card(self, biz_id: str, messages: Messages, flow_status: str) -> Dict[str, Any]: + """用服务端 biz_id 更新同一卡片，flow_status 必填。""" + if self.cfg.a2ui_client is None: + raise RuntimeError("请先配置 a2ui_client（例如 DwsA2UIClient）") + return await self.cfg.a2ui_client.update_card(biz_id, messages, flow_status) + async def send_image(self, target: SendTarget, image_url: str) -> None: await self.sender.send_image(target, image_url) diff --git a/src/dingtalk_channel_sdk/config.py b/src/dingtalk_channel_sdk/config.py index 6a84bb3..e426178 100644 --- a/src/dingtalk_channel_sdk/config.py +++ b/src/dingtalk_channel_sdk/config.py @@ -7,6 +7,7 @@ from typing import Any, Callable, List, Optional from .safety.policy import PolicyConfig +from .a2ui import A2UIClient DEFAULT_API_BASE = "https://api.dingtalk.com" DEFAULT_OAPI_BASE = "https://oapi.dingtalk.com" @@ -119,6 +120,8 @@ class Config: outbound: Optional[OutboundConfig] = None #: SSRF 白名单：命中的主机名跳过公网校验（支持通配符 *.example.com）。 ssrf_allowlist: List[str] = field(default_factory=list) + #: 显式配置的 A2UI 发送通道；例如 DwsA2UIClient。 + a2ui_client: Optional[A2UIClient] = None def __post_init__(self) -> None: if self.transport not in (TRANSPORT_STREAM, TRANSPORT_HTTP): diff --git a/tests/test_a2ui.py b/tests/test_a2ui.py new file mode 100644 index 0000000..0626150 --- /dev/null +++ b/tests/test_a2ui.py @@ -0,0 +1,111 @@ +import json +import sys + +import pytest + +from dingtalk_channel_sdk import Config, DingTalkChannel, DwsA2UIClient, serialize_a2ui_messages + +MESSAGE = {"version": "v1.0", "updateDataModel": {"surfaceId": "sdk-card", "path": "/text", "value": '中文、引号"与 $(echo test) `test`'}} + + +def fixture(tmp_path, response=None, mode="ok", timeout_s=5): + if response is None: + response = {"ok": True, "outcome": "success", "data": {"success": True, "result": {"bizId": "server-biz"}}} + script = tmp_path / "模拟 dws.py" + trace = tmp_path / "trace.jsonl" + script.write_text('''import json, sys, time +trace, response, mode, *args = sys.argv[1:] +with open(trace, "a", encoding="utf-8") as output: + output.write(json.dumps(args, ensure_ascii=False) + "\\n") +if mode == "hang": time.sleep(10) +if mode == "fail": + sys.stderr.write("不应暴露的凭据占位符") + sys.exit(2) +sys.stdout.write(response) +''', encoding="utf-8") + client = DwsA2UIClient(command=[sys.executable, str(script), str(trace), response if isinstance(response, str) else json.dumps(response), mode], profile="corp:user", timeout_s=timeout_s) + return client, lambda: [json.loads(line) for line in trace.read_text(encoding="utf-8").splitlines()] + + +def test_对象和字符串均保留语义(): + encoded = json.dumps(MESSAGE, ensure_ascii=False) + strings = json.loads(serialize_a2ui_messages([MESSAGE, encoded])) + assert [json.loads(value) for value in strings] == [MESSAGE, MESSAGE] + assert strings[1] == encoded + + +@pytest.mark.parametrize("messages", [[], {}, ["not-json"], [None], [{"version": "v0.8"}], [{"version": "v1.0"}], [{"version": "v1.0", "deleteSurface": {"surfaceId": ""}}], [{**MESSAGE, "deleteSurface": {"surfaceId": "x"}}]]) +def test_无效信封在执行前拒绝(messages): + with pytest.raises(ValueError): + serialize_a2ui_messages(messages) + + +def test_超大参数在执行前拒绝(): + with pytest.raises(ValueError, match="64 KiB"): + serialize_a2ui_messages([{**MESSAGE, "updateDataModel": {**MESSAGE["updateDataModel"], "value": "中" * 24000}}]) + + +async def test_Channel显式通道发送单聊群聊并完成原卡片(tmp_path): + client, calls = fixture(tmp_path) + ch = DingTalkChannel(config=Config("unused", "unused", a2ui_client=client)) + result = await ch.send_a2ui_card({"open_dingtalk_id": "D-user"}, [MESSAGE]) + assert result.biz_id == "server-biz" + assert result.update_warning is None + assert result.receipt["data"]["result"]["bizId"] == "server-biz" + await ch.send_a2ui_card({"conversation_id": "--group-value"}, [MESSAGE]) + await ch.update_a2ui_card(result.biz_id, [MESSAGE], "3") + argv = calls() + assert len(argv) == 3 + assert "--open-dingtalk-id=D-user" in argv[0] + assert "--conversation-id=--group-value" in argv[1] + assert "--biz-id=server-biz" in argv[2] + assert "--flow-status=FINISH" in argv[2] + for args in argv: + assert "--profile=corp:user" in args and "--format=json" in args and "--yes" in args + assert not any("client-secret" in arg for arg in args) + content = next(arg.split("=", 1)[1] for arg in args if arg.startswith("--content=")) + assert [json.loads(value) for value in json.loads(content)] == [MESSAGE] + + +async def test_未启用时明确拒绝(): + ch = DingTalkChannel(config=Config("unused", "unused")) + with pytest.raises(RuntimeError, match="a2ui_client"): + await ch.send_a2ui_card({"conversation_id": "cid"}, [MESSAGE]) + with pytest.raises(RuntimeError, match="a2ui_client"): + await ch.update_a2ui_card("biz", [MESSAGE], "FINISH") + + +async def test_目标标识和状态错误不执行子进程(): + with pytest.raises(ValueError, match="一个 DWS Profile"): + DwsA2UIClient(profile="corp:user,other:user") + client = DwsA2UIClient(command=["不存在的 dws"]) + for target in [{}, {"user_id": "staff"}, {"conversation_id": "cid", "at_all": True}, {"conversation_id": "cid", "open_dingtalk_id": "D-user"}, {"open_dingtalk_id": ""}]: + with pytest.raises(ValueError): + await client.send_card(target, [MESSAGE]) + with pytest.raises(ValueError, match="bizId"): + await client.update_card("", [MESSAGE], "FINISH") + with pytest.raises(ValueError, match="flow_status"): + await client.update_card("biz", [MESSAGE], "unknown") + + +async def test_请求侧bizCardId不作为更新标识且不重发(tmp_path): + client, calls = fixture(tmp_path, {"success": True, "result": {"bizCardId": "request-only", "openTaskId": "task"}}) + result = await client.send_card({"conversation_id": "cid"}, [MESSAGE]) + assert result.biz_id is None and "不要自动重发" in result.update_warning + assert result.receipt["result"]["bizCardId"] == "request-only" + assert len(calls()) == 1 + + +@pytest.mark.parametrize("response,mode", [({"success": False}, "ok"), ({"success": True, "result": {"success": False}}, "ok"), ("not-json", "ok"), ({}, "ok"), ({"result": {"bizId": "unconfirmed"}}, "ok"), ({"ok": True, "outcome": "success", "dry_run": True}, "ok"), ({}, "fail")]) +async def test_失败回执和进程失败不泄露stderr(tmp_path, response, mode): + client, calls = fixture(tmp_path, response, mode) + with pytest.raises(RuntimeError) as error: + await client.send_card({"conversation_id": "cid"}, [MESSAGE]) + assert "不应暴露" not in str(error.value) + assert len(calls()) == 1 + + +async def test_超时明确报告未知结果(tmp_path): + client, _ = fixture(tmp_path, {}, "hang", timeout_s=0.2) + with pytest.raises(RuntimeError, match="结果可能未知"): + await client.send_card({"conversation_id": "cid"}, [MESSAGE])