feat(dingtalk): AI Cards streaming, emoji reactions, and media handling
Cherry-picked from #10985 by pedh, adapted to current main: * Keeps main's full group-chat gating (require_mention + allowed_users + free_response_chats + mention_patterns) — PR's simpler subset dropped. * Keeps main's fire-and-forget process() dispatch + session_webhook fallback for SDK >= 0.24. * Picks up PR's REQUIRES_EDIT_FINALIZE capability flag on BasePlatformAdapter + finalize kwarg on edit_message(), plumbed through stream_consumer. Default False so Telegram/Slack/Discord/Matrix stay on the zero-overhead fast path. * DingTalk AI Card lifecycle: per-chat _message_contexts, two-card flow (tool-progress + final response) with sibling auto-close driven by reply_to, idempotent 🤔Thinking → 🥳Done swap, $alibabacloud-dingtalk$ for media URL resolution (replaces raw HTTP that was 403-ing). * pyproject: dingtalk extra now dingtalk-stream>=0.20,<1 + alibabacloud-dingtalk>=2.0.0 + qrcode. Closes #10991 Co-authored-by: pedh
This commit is contained in:
pedh
@@ -198,7 +198,7 @@ class TestSend:
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post = AsyncMock(return_value=mock_response)
|
||||
adapter._http_client = mock_client
|
||||
adapter._session_webhooks["chat-123"] = "https://cached.example/webhook"
|
||||
adapter._session_webhooks["chat-123"] = ("https://cached.example/webhook", 9999999999999)
|
||||
|
||||
result = await adapter.send("chat-123", "Hello!")
|
||||
assert result.success is True
|
||||
@@ -681,3 +681,290 @@ class TestIncomingHandlerProcess:
|
||||
processing_gate.set()
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Text extraction — mention preservation + platform sanity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestExtractTextMentions:
|
||||
|
||||
def test_preserves_at_mentions_in_text(self):
|
||||
"""@mentions are routing signals (via isInAtList), not text to strip.
|
||||
|
||||
Stripping all @handles collateral-damages emails, SSH URLs, and
|
||||
literal references the user wrote.
|
||||
"""
|
||||
from gateway.platforms.dingtalk import DingTalkAdapter
|
||||
cases = [
|
||||
("@bot hello", "@bot hello"),
|
||||
("contact alice@example.com", "contact alice@example.com"),
|
||||
("git@github.com:foo/bar.git", "git@github.com:foo/bar.git"),
|
||||
("what does @openai think", "what does @openai think"),
|
||||
("@机器人 转发给 @老王", "@机器人 转发给 @老王"),
|
||||
]
|
||||
for text, expected in cases:
|
||||
msg = MagicMock()
|
||||
msg.text = text
|
||||
msg.rich_text = None
|
||||
msg.rich_text_content = None
|
||||
assert DingTalkAdapter._extract_text(msg) == expected, (
|
||||
f"mangled: {text!r} -> {DingTalkAdapter._extract_text(msg)!r}"
|
||||
)
|
||||
|
||||
def test_dingtalk_in_platform_enum(self):
|
||||
assert Platform.DINGTALK.value == "dingtalk"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Concurrency — chat-scoped message context
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMessageContextIsolation:
|
||||
|
||||
def test_contexts_keyed_by_chat_id(self):
|
||||
"""Two concurrent chats must not clobber each other's context."""
|
||||
from gateway.platforms.dingtalk import DingTalkAdapter
|
||||
adapter = DingTalkAdapter(PlatformConfig(enabled=True))
|
||||
|
||||
msg_a = MagicMock(conversation_id="chat-A", sender_staff_id="user-A")
|
||||
msg_b = MagicMock(conversation_id="chat-B", sender_staff_id="user-B")
|
||||
adapter._message_contexts["chat-A"] = msg_a
|
||||
adapter._message_contexts["chat-B"] = msg_b
|
||||
|
||||
assert adapter._message_contexts["chat-A"] is msg_a
|
||||
assert adapter._message_contexts["chat-B"] is msg_b
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Card lifecycle: finalize via metadata["streaming"]
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCardLifecycle:
|
||||
|
||||
@pytest.fixture
|
||||
def adapter_with_card(self):
|
||||
from gateway.platforms.dingtalk import DingTalkAdapter
|
||||
a = DingTalkAdapter(PlatformConfig(
|
||||
enabled=True,
|
||||
extra={"card_template_id": "tmpl-1"},
|
||||
))
|
||||
a._card_sdk = MagicMock()
|
||||
a._card_sdk.create_card_with_options_async = AsyncMock()
|
||||
a._card_sdk.deliver_card_with_options_async = AsyncMock()
|
||||
a._card_sdk.streaming_update_with_options_async = AsyncMock()
|
||||
a._http_client = AsyncMock()
|
||||
a._get_access_token = AsyncMock(return_value="token")
|
||||
# Minimal message context
|
||||
msg = MagicMock(
|
||||
conversation_id="chat-1",
|
||||
conversation_type="1",
|
||||
sender_staff_id="staff-1",
|
||||
message_id="user-msg-1",
|
||||
)
|
||||
a._message_contexts["chat-1"] = msg
|
||||
a._session_webhooks["chat-1"] = (
|
||||
"https://api.dingtalk.com/x", 9999999999999,
|
||||
)
|
||||
return a
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_final_reply_finalizes_card(self, adapter_with_card):
|
||||
"""send(reply_to=...) creates a closed card (final response path)."""
|
||||
a = adapter_with_card
|
||||
result = await a.send("chat-1", "Hello", reply_to="user-msg-1")
|
||||
assert result.success
|
||||
call = a._card_sdk.streaming_update_with_options_async.call_args
|
||||
assert call[0][0].is_finalize is True
|
||||
# Not tracked as streaming — it's already closed.
|
||||
assert "chat-1" not in a._streaming_cards
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_intermediate_send_stays_streaming(self, adapter_with_card):
|
||||
"""send() without reply_to creates an OPEN card (tool progress /
|
||||
commentary / streaming first chunk). No flicker closed→streaming
|
||||
when edit_message follows."""
|
||||
a = adapter_with_card
|
||||
result = await a.send("chat-1", "💻 terminal: ls")
|
||||
assert result.success
|
||||
call = a._card_sdk.streaming_update_with_options_async.call_args
|
||||
assert call[0][0].is_finalize is False
|
||||
# Tracked for sibling cleanup.
|
||||
assert result.message_id in a._streaming_cards.get("chat-1", {})
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_done_fires_only_when_reply_to_is_set(self, adapter_with_card):
|
||||
"""reply_to distinguishes final response (base.py) from tool-progress
|
||||
sends (run.py). Done must only fire for the former."""
|
||||
a = adapter_with_card
|
||||
fired: list[str] = []
|
||||
a._fire_done_reaction = lambda cid: fired.append(cid)
|
||||
|
||||
# Tool-progress / commentary path: no reply_to — no Done.
|
||||
await a.send("chat-1", "tool line")
|
||||
assert fired == []
|
||||
|
||||
# Final response path: reply_to set — Done fires.
|
||||
await a.send("chat-1", "final", reply_to="user-msg-1")
|
||||
assert fired == ["chat-1"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edit_message_finalize_fires_done(self, adapter_with_card):
|
||||
"""Stream consumer's final edit_message(finalize=True) fires Done."""
|
||||
a = adapter_with_card
|
||||
fired: list[str] = []
|
||||
a._fire_done_reaction = lambda cid: fired.append(cid)
|
||||
|
||||
await a.send("chat-1", "initial")
|
||||
# Reopen via edit_message(finalize=False) then close.
|
||||
await a.edit_message(
|
||||
chat_id="chat-1", message_id="track-X",
|
||||
content="streaming...", finalize=False,
|
||||
)
|
||||
await a.edit_message(
|
||||
chat_id="chat-1", message_id="track-X",
|
||||
content="final", finalize=True,
|
||||
)
|
||||
assert "chat-1" in fired
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edit_message_finalize_false_tracks_sibling(self, adapter_with_card):
|
||||
"""After edit_message(finalize=False), card is tracked as open."""
|
||||
a = adapter_with_card
|
||||
await a.edit_message(
|
||||
chat_id="chat-1", message_id="track-1",
|
||||
content="partial", finalize=False,
|
||||
)
|
||||
assert "chat-1" in a._streaming_cards
|
||||
assert a._streaming_cards["chat-1"].get("track-1") == "partial"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_next_send_auto_closes_sibling_streaming_cards(
|
||||
self, adapter_with_card,
|
||||
):
|
||||
"""Tool-progress card left open (send without reply_to + edits) must
|
||||
be auto-closed when the final-reply send arrives."""
|
||||
a = adapter_with_card
|
||||
# First tool: intermediate send — card stays open.
|
||||
r1 = await a.send("chat-1", "💻 tool1")
|
||||
# Second tool: edit_message(finalize=False) — keeps streaming.
|
||||
await a.edit_message(
|
||||
chat_id="chat-1", message_id=r1.message_id,
|
||||
content="💻 tool1\n💻 tool2", finalize=False,
|
||||
)
|
||||
assert r1.message_id in a._streaming_cards.get("chat-1", {})
|
||||
a._card_sdk.streaming_update_with_options_async.reset_mock()
|
||||
|
||||
# Final response send auto-closes the sibling.
|
||||
await a.send("chat-1", "final answer", reply_to="user-msg")
|
||||
|
||||
calls = a._card_sdk.streaming_update_with_options_async.call_args_list
|
||||
assert len(calls) >= 2
|
||||
# First call was the sibling close with last-seen tool-progress content.
|
||||
first_req = calls[0][0][0]
|
||||
assert first_req.out_track_id == r1.message_id
|
||||
assert first_req.is_finalize is True
|
||||
assert "tool1" in first_req.content
|
||||
# Streaming tracking is cleared after close.
|
||||
assert "chat-1" not in a._streaming_cards
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edit_message_requires_message_id(self, adapter_with_card):
|
||||
a = adapter_with_card
|
||||
result = await a.edit_message(
|
||||
chat_id="chat-1", message_id="", content="x", finalize=True,
|
||||
)
|
||||
assert result.success is False
|
||||
a._card_sdk.streaming_update_with_options_async.assert_not_called()
|
||||
|
||||
def test_fire_done_reaction_is_idempotent(self, adapter_with_card):
|
||||
a = adapter_with_card
|
||||
captured = []
|
||||
def _capture(coro):
|
||||
captured.append(coro)
|
||||
a._spawn_bg = _capture
|
||||
|
||||
a._fire_done_reaction("chat-1")
|
||||
a._fire_done_reaction("chat-1")
|
||||
assert len(captured) == 1
|
||||
captured[0].close()
|
||||
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AI Card Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestDingTalkAdapterAICards:
|
||||
@pytest.fixture
|
||||
def config(self):
|
||||
return PlatformConfig(
|
||||
enabled=True,
|
||||
extra={
|
||||
"client_id": "test_id",
|
||||
"client_secret": "test_secret",
|
||||
"card_template_id": "test_card_template",
|
||||
},
|
||||
)
|
||||
|
||||
@pytest.fixture
|
||||
def mock_stream_client(self):
|
||||
client = MagicMock()
|
||||
client.get_access_token = MagicMock(return_value="test_token")
|
||||
return client
|
||||
|
||||
@pytest.fixture
|
||||
def mock_http_client(self):
|
||||
return AsyncMock()
|
||||
|
||||
@pytest.fixture
|
||||
def mock_message(self):
|
||||
msg = MagicMock()
|
||||
msg.message_id = "test_msg_id"
|
||||
msg.conversation_id = "test_conv_id"
|
||||
msg.conversation_type = "1"
|
||||
msg.sender_id = "sender1"
|
||||
msg.sender_nick = "Test User"
|
||||
msg.sender_staff_id = "staff1"
|
||||
msg.text = MagicMock(content="Hello")
|
||||
msg.session_webhook = "https://api.dingtalk.com/robot/sendBySession?session=test"
|
||||
msg.session_webhook_expired_time = 999999999999
|
||||
msg.create_at = int(datetime.now(tz=timezone.utc).timestamp() * 1000)
|
||||
msg.at_users = []
|
||||
return msg
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_uses_ai_card_if_configured(self, config, mock_stream_client, mock_http_client, mock_message):
|
||||
from gateway.platforms.dingtalk import DingTalkAdapter
|
||||
|
||||
adapter = DingTalkAdapter(config)
|
||||
adapter._stream_client = mock_stream_client
|
||||
adapter._http_client = mock_http_client
|
||||
adapter._message_contexts["test_conv_id"] = mock_message
|
||||
adapter._session_webhooks = {"test_conv_id": ("https://api.dingtalk.com/robot/sendBySession?session=test", 9999999999999)}
|
||||
adapter._card_template_id = "test_card_template"
|
||||
|
||||
# Mock the card SDK with proper async methods
|
||||
mock_card_sdk = MagicMock()
|
||||
mock_card_sdk.create_card_with_options_async = AsyncMock()
|
||||
mock_card_sdk.deliver_card_with_options_async = AsyncMock()
|
||||
mock_card_sdk.streaming_update_with_options_async = AsyncMock()
|
||||
adapter._card_sdk = mock_card_sdk
|
||||
|
||||
# Mock access token
|
||||
adapter._get_access_token = AsyncMock(return_value="test_token")
|
||||
|
||||
result = await adapter.send("test_conv_id", "Hello World")
|
||||
|
||||
mock_card_sdk.create_card_with_options_async.assert_called_once()
|
||||
mock_card_sdk.deliver_card_with_options_async.assert_called_once()
|
||||
mock_card_sdk.streaming_update_with_options_async.assert_called_once()
|
||||
assert result.success is True
|
||||
|
||||
@@ -88,6 +88,51 @@ class TestCleanForDisplay:
|
||||
# ── Integration: _send_or_edit strips MEDIA: ─────────────────────────────
|
||||
|
||||
|
||||
class TestFinalizeCapabilityGate:
|
||||
"""Verify REQUIRES_EDIT_FINALIZE gates the redundant final edit.
|
||||
|
||||
Platforms that don't need an explicit finalize signal (Telegram,
|
||||
Slack, Matrix, …) should skip the redundant final edit when the
|
||||
mid-stream edit already delivered the final content. Platforms that
|
||||
*do* need it (DingTalk AI Cards) must always receive a finalize=True
|
||||
edit at the end of the stream.
|
||||
"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_identical_text_skip_respects_adapter_flag(self):
|
||||
"""_send_or_edit short-circuits identical-text only when the
|
||||
adapter doesn't require an explicit finalize signal."""
|
||||
# Adapter without finalize requirement — should skip identical edit.
|
||||
plain = MagicMock()
|
||||
plain.REQUIRES_EDIT_FINALIZE = False
|
||||
plain.send = AsyncMock(return_value=SimpleNamespace(
|
||||
success=True, message_id="m1",
|
||||
))
|
||||
plain.edit_message = AsyncMock()
|
||||
plain.MAX_MESSAGE_LENGTH = 4096
|
||||
c1 = GatewayStreamConsumer(plain, "chat_1")
|
||||
await c1._send_or_edit("hello") # first send
|
||||
await c1._send_or_edit("hello", finalize=True) # identical → skip
|
||||
plain.edit_message.assert_not_called()
|
||||
|
||||
# Adapter that requires finalize — must still fire the edit.
|
||||
picky = MagicMock()
|
||||
picky.REQUIRES_EDIT_FINALIZE = True
|
||||
picky.send = AsyncMock(return_value=SimpleNamespace(
|
||||
success=True, message_id="m1",
|
||||
))
|
||||
picky.edit_message = AsyncMock(return_value=SimpleNamespace(
|
||||
success=True, message_id="m1",
|
||||
))
|
||||
picky.MAX_MESSAGE_LENGTH = 4096
|
||||
c2 = GatewayStreamConsumer(picky, "chat_1")
|
||||
await c2._send_or_edit("hello")
|
||||
await c2._send_or_edit("hello", finalize=True)
|
||||
# Finalize edit must go through even on identical content.
|
||||
picky.edit_message.assert_called_once()
|
||||
assert picky.edit_message.call_args[1]["finalize"] is True
|
||||
|
||||
|
||||
class TestSendOrEditMediaStripping:
|
||||
"""Verify _send_or_edit strips MEDIA: before sending to the platform."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user