"""Tests for DingTalk platform adapter.""" import asyncio import json from datetime import datetime, timezone from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock import pytest from gateway.config import Platform, PlatformConfig # --------------------------------------------------------------------------- # Requirements check # --------------------------------------------------------------------------- class TestDingTalkRequirements: def test_returns_false_when_sdk_missing(self, monkeypatch): with patch.dict("sys.modules", {"dingtalk_stream": None}): monkeypatch.setattr( "gateway.platforms.dingtalk.DINGTALK_STREAM_AVAILABLE", False ) from gateway.platforms.dingtalk import check_dingtalk_requirements assert check_dingtalk_requirements() is False def test_returns_false_when_env_vars_missing(self, monkeypatch): monkeypatch.setattr( "gateway.platforms.dingtalk.DINGTALK_STREAM_AVAILABLE", True ) monkeypatch.setattr("gateway.platforms.dingtalk.HTTPX_AVAILABLE", True) monkeypatch.delenv("DINGTALK_CLIENT_ID", raising=False) monkeypatch.delenv("DINGTALK_CLIENT_SECRET", raising=False) from gateway.platforms.dingtalk import check_dingtalk_requirements assert check_dingtalk_requirements() is False def test_returns_true_when_all_available(self, monkeypatch): monkeypatch.setattr( "gateway.platforms.dingtalk.DINGTALK_STREAM_AVAILABLE", True ) monkeypatch.setattr("gateway.platforms.dingtalk.HTTPX_AVAILABLE", True) monkeypatch.setenv("DINGTALK_CLIENT_ID", "test-id") monkeypatch.setenv("DINGTALK_CLIENT_SECRET", "test-secret") from gateway.platforms.dingtalk import check_dingtalk_requirements assert check_dingtalk_requirements() is True # --------------------------------------------------------------------------- # Adapter construction # --------------------------------------------------------------------------- class TestDingTalkAdapterInit: def test_reads_config_from_extra(self): from gateway.platforms.dingtalk import DingTalkAdapter config = PlatformConfig( enabled=True, extra={"client_id": "cfg-id", "client_secret": "cfg-secret"}, ) adapter = DingTalkAdapter(config) assert adapter._client_id == "cfg-id" assert adapter._client_secret == "cfg-secret" assert adapter.name == "Dingtalk" # base class uses .title() def test_falls_back_to_env_vars(self, monkeypatch): monkeypatch.setenv("DINGTALK_CLIENT_ID", "env-id") monkeypatch.setenv("DINGTALK_CLIENT_SECRET", "env-secret") from gateway.platforms.dingtalk import DingTalkAdapter config = PlatformConfig(enabled=True) adapter = DingTalkAdapter(config) assert adapter._client_id == "env-id" assert adapter._client_secret == "env-secret" # --------------------------------------------------------------------------- # Message text extraction # --------------------------------------------------------------------------- class TestExtractText: def test_extracts_dict_text(self): from gateway.platforms.dingtalk import DingTalkAdapter msg = MagicMock() msg.text = {"content": " hello world "} msg.rich_text = None assert DingTalkAdapter._extract_text(msg) == "hello world" def test_extracts_string_text(self): from gateway.platforms.dingtalk import DingTalkAdapter msg = MagicMock() msg.text = "plain text" msg.rich_text = None assert DingTalkAdapter._extract_text(msg) == "plain text" def test_falls_back_to_rich_text(self): from gateway.platforms.dingtalk import DingTalkAdapter msg = MagicMock() msg.text = "" msg.rich_text = [{"text": "part1"}, {"text": "part2"}, {"image": "url"}] assert DingTalkAdapter._extract_text(msg) == "part1 part2" def test_returns_empty_for_no_content(self): from gateway.platforms.dingtalk import DingTalkAdapter msg = MagicMock() msg.text = "" msg.rich_text = None assert DingTalkAdapter._extract_text(msg) == "" # --------------------------------------------------------------------------- # Deduplication # --------------------------------------------------------------------------- class TestDeduplication: def test_first_message_not_duplicate(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) assert adapter._dedup.is_duplicate("msg-1") is False def test_second_same_message_is_duplicate(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) adapter._dedup.is_duplicate("msg-1") assert adapter._dedup.is_duplicate("msg-1") is True def test_different_messages_not_duplicate(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) adapter._dedup.is_duplicate("msg-1") assert adapter._dedup.is_duplicate("msg-2") is False def test_cache_cleanup_on_overflow(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) max_size = adapter._dedup._max_size # Fill beyond max for i in range(max_size + 10): adapter._dedup.is_duplicate(f"msg-{i}") # Cache should have been pruned assert len(adapter._dedup._seen) <= max_size + 10 # --------------------------------------------------------------------------- # Send # --------------------------------------------------------------------------- class TestSend: @pytest.mark.asyncio async def test_send_posts_to_webhook(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) mock_response = MagicMock() mock_response.status_code = 200 mock_response.text = "OK" mock_client = AsyncMock() mock_client.post = AsyncMock(return_value=mock_response) adapter._http_client = mock_client result = await adapter.send( "chat-123", "Hello!", metadata={"session_webhook": "https://dingtalk.example/webhook"} ) assert result.success is True mock_client.post.assert_called_once() call_args = mock_client.post.call_args assert call_args[0][0] == "https://dingtalk.example/webhook" payload = call_args[1]["json"] assert payload["msgtype"] == "markdown" assert payload["markdown"]["title"] == "Hermes" assert payload["markdown"]["text"] == "Hello!" @pytest.mark.asyncio async def test_send_fails_without_webhook(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) adapter._http_client = AsyncMock() result = await adapter.send("chat-123", "Hello!") assert result.success is False assert "session_webhook" in result.error @pytest.mark.asyncio async def test_send_uses_cached_webhook(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) mock_response = MagicMock() mock_response.status_code = 200 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" result = await adapter.send("chat-123", "Hello!") assert result.success is True assert mock_client.post.call_args[0][0] == "https://cached.example/webhook" @pytest.mark.asyncio async def test_send_handles_http_error(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) mock_response = MagicMock() mock_response.status_code = 400 mock_response.text = "Bad Request" mock_client = AsyncMock() mock_client.post = AsyncMock(return_value=mock_response) adapter._http_client = mock_client result = await adapter.send( "chat-123", "Hello!", metadata={"session_webhook": "https://example/webhook"} ) assert result.success is False assert "400" in result.error # --------------------------------------------------------------------------- # Connect / disconnect # --------------------------------------------------------------------------- class TestConnect: @pytest.mark.asyncio async def test_disconnect_closes_session_websocket(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) websocket = AsyncMock() blocker = asyncio.Event() async def _run_forever(): try: await blocker.wait() except asyncio.CancelledError: return adapter._stream_client = SimpleNamespace(websocket=websocket) adapter._stream_task = asyncio.create_task(_run_forever()) adapter._running = True await adapter.disconnect() websocket.close.assert_awaited_once() assert adapter._stream_task is None @pytest.mark.asyncio async def test_connect_fails_without_sdk(self, monkeypatch): monkeypatch.setattr( "gateway.platforms.dingtalk.DINGTALK_STREAM_AVAILABLE", False ) from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) result = await adapter.connect() assert result is False @pytest.mark.asyncio async def test_connect_fails_without_credentials(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) adapter._client_id = "" adapter._client_secret = "" result = await adapter.connect() assert result is False @pytest.mark.asyncio async def test_disconnect_cleans_up(self): from gateway.platforms.dingtalk import DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) adapter._session_webhooks["a"] = "http://x" adapter._dedup._seen["b"] = 1.0 adapter._http_client = AsyncMock() adapter._stream_task = None await adapter.disconnect() assert len(adapter._session_webhooks) == 0 assert len(adapter._dedup._seen) == 0 assert adapter._http_client is None # --------------------------------------------------------------------------- # Platform enum # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # SDK compatibility regression tests (dingtalk-stream >= 0.20 / 0.24) # --------------------------------------------------------------------------- class TestWebhookDomainAllowlist: """Guard the webhook origin allowlist against regression. The SDK started returning reply webhooks on ``oapi.dingtalk.com`` in addition to ``api.dingtalk.com``. Both must be accepted, and hostile lookalikes must still be rejected (SSRF defence-in-depth). """ def test_api_domain_accepted(self): from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE assert _DINGTALK_WEBHOOK_RE.match( "https://api.dingtalk.com/robot/send?access_token=x" ) def test_oapi_domain_accepted(self): from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE assert _DINGTALK_WEBHOOK_RE.match( "https://oapi.dingtalk.com/robot/send?access_token=x" ) def test_http_rejected(self): from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE assert not _DINGTALK_WEBHOOK_RE.match("http://api.dingtalk.com/robot/send") def test_suffix_attack_rejected(self): from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE assert not _DINGTALK_WEBHOOK_RE.match( "https://api.dingtalk.com.evil.example/" ) def test_unsanctioned_subdomain_rejected(self): from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE # Only api.* and oapi.* are allowed — e.g. eapi.dingtalk.com must not slip through assert not _DINGTALK_WEBHOOK_RE.match("https://eapi.dingtalk.com/robot/send") class TestHandlerProcessIsAsync: """dingtalk-stream >= 0.20 requires ``process`` to be a coroutine.""" def test_process_is_coroutine_function(self): from gateway.platforms.dingtalk import _IncomingHandler assert asyncio.iscoroutinefunction(_IncomingHandler.process) class TestExtractText: """_extract_text must handle both legacy and current SDK payload shapes. Before SDK 0.20 ``message.text`` was a ``dict`` with a ``content`` key. From 0.20 onward it is a ``TextContent`` dataclass whose ``__str__`` returns ``"TextContent(content=...)"`` — falling back to ``str(text)`` leaks that repr into the agent's input. """ def test_text_as_dict_legacy(self): from gateway.platforms.dingtalk import DingTalkAdapter msg = MagicMock() msg.text = {"content": "hello world"} msg.rich_text_content = None msg.rich_text = None assert DingTalkAdapter._extract_text(msg) == "hello world" def test_text_as_textcontent_object(self): """SDK >= 0.20 shape: object with ``.content`` attribute.""" from gateway.platforms.dingtalk import DingTalkAdapter class FakeTextContent: content = "hello from new sdk" def __str__(self): # mimic real SDK repr return f"TextContent(content={self.content})" msg = MagicMock() msg.text = FakeTextContent() msg.rich_text_content = None msg.rich_text = None result = DingTalkAdapter._extract_text(msg) assert result == "hello from new sdk" assert "TextContent(" not in result def test_text_content_attr_with_empty_string(self): from gateway.platforms.dingtalk import DingTalkAdapter class FakeTextContent: content = "" msg = MagicMock() msg.text = FakeTextContent() msg.rich_text_content = None msg.rich_text = None assert DingTalkAdapter._extract_text(msg) == "" def test_rich_text_content_new_shape(self): """SDK >= 0.20 exposes rich text as ``message.rich_text_content.rich_text_list``.""" from gateway.platforms.dingtalk import DingTalkAdapter class FakeRichText: rich_text_list = [{"text": "hello "}, {"text": "world"}] msg = MagicMock() msg.text = None msg.rich_text_content = FakeRichText() msg.rich_text = None result = DingTalkAdapter._extract_text(msg) assert "hello" in result and "world" in result def test_rich_text_legacy_shape(self): """Legacy ``message.rich_text`` list remains supported.""" from gateway.platforms.dingtalk import DingTalkAdapter msg = MagicMock() msg.text = None msg.rich_text_content = None msg.rich_text = [{"text": "legacy "}, {"text": "rich"}] result = DingTalkAdapter._extract_text(msg) assert "legacy" in result and "rich" in result def test_empty_message(self): from gateway.platforms.dingtalk import DingTalkAdapter msg = MagicMock() msg.text = None msg.rich_text_content = None msg.rich_text = None assert DingTalkAdapter._extract_text(msg) == "" # --------------------------------------------------------------------------- # Group gating — require_mention + allowed_users (parity with other platforms) # --------------------------------------------------------------------------- def _make_gating_adapter(monkeypatch, *, extra=None, env=None): """Build a DingTalkAdapter with only the gating fields populated. Clears every DINGTALK_* gating env var before applying the caller's overrides so individual tests stay isolated. """ for key in ( "DINGTALK_REQUIRE_MENTION", "DINGTALK_MENTION_PATTERNS", "DINGTALK_FREE_RESPONSE_CHATS", "DINGTALK_ALLOWED_USERS", ): monkeypatch.delenv(key, raising=False) for key, value in (env or {}).items(): monkeypatch.setenv(key, value) from gateway.platforms.dingtalk import DingTalkAdapter return DingTalkAdapter(PlatformConfig(enabled=True, extra=extra or {})) class TestAllowedUsersGate: def test_empty_allowlist_allows_everyone(self, monkeypatch): adapter = _make_gating_adapter(monkeypatch) assert adapter._is_user_allowed("anyone", "any-staff") is True def test_wildcard_allowlist_allows_everyone(self, monkeypatch): adapter = _make_gating_adapter(monkeypatch, extra={"allowed_users": ["*"]}) assert adapter._is_user_allowed("anyone", "any-staff") is True def test_matches_sender_id_case_insensitive(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"allowed_users": ["SenderABC"]} ) assert adapter._is_user_allowed("senderabc", "") is True def test_matches_staff_id(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"allowed_users": ["staff_1234"]} ) assert adapter._is_user_allowed("", "staff_1234") is True def test_rejects_unknown_user(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"allowed_users": ["staff_1234"]} ) assert adapter._is_user_allowed("other-sender", "other-staff") is False def test_env_var_csv_populates_allowlist(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, env={"DINGTALK_ALLOWED_USERS": "alice,bob,carol"} ) assert adapter._is_user_allowed("alice", "") is True assert adapter._is_user_allowed("dave", "") is False class TestMentionPatterns: def test_empty_patterns_list(self, monkeypatch): adapter = _make_gating_adapter(monkeypatch) assert adapter._mention_patterns == [] assert adapter._message_matches_mention_patterns("anything") is False def test_pattern_matches_text(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"mention_patterns": ["^hermes"]} ) assert adapter._message_matches_mention_patterns("hermes please help") is True assert adapter._message_matches_mention_patterns("please hermes help") is False def test_pattern_is_case_insensitive(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"mention_patterns": ["^hermes"]} ) assert adapter._message_matches_mention_patterns("HERMES help") is True def test_invalid_regex_is_skipped_not_raised(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"mention_patterns": ["[unclosed", "^valid"]}, ) # Invalid pattern dropped, valid one kept assert len(adapter._mention_patterns) == 1 assert adapter._message_matches_mention_patterns("valid trigger") is True def test_env_var_json_populates_patterns(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, env={"DINGTALK_MENTION_PATTERNS": '["^bot", "^assistant"]'}, ) assert len(adapter._mention_patterns) == 2 assert adapter._message_matches_mention_patterns("bot ping") is True def test_env_var_newline_fallback_when_not_json(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, env={"DINGTALK_MENTION_PATTERNS": "^bot\n^assistant"}, ) assert len(adapter._mention_patterns) == 2 class TestShouldProcessMessage: def test_dm_always_accepted(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"require_mention": True} ) msg = MagicMock(is_in_at_list=False) assert adapter._should_process_message(msg, "hi", is_group=False, chat_id="dm1") is True def test_group_rejected_when_require_mention_and_no_trigger(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"require_mention": True} ) msg = MagicMock(is_in_at_list=False) assert adapter._should_process_message(msg, "hi", is_group=True, chat_id="grp1") is False def test_group_accepted_when_require_mention_disabled(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"require_mention": False} ) msg = MagicMock(is_in_at_list=False) assert adapter._should_process_message(msg, "hi", is_group=True, chat_id="grp1") is True def test_group_accepted_when_bot_is_mentioned(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"require_mention": True} ) msg = MagicMock(is_in_at_list=True) assert adapter._should_process_message(msg, "hi", is_group=True, chat_id="grp1") is True def test_group_accepted_when_text_matches_wake_word(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"require_mention": True, "mention_patterns": ["^hermes"]}, ) msg = MagicMock(is_in_at_list=False) assert adapter._should_process_message(msg, "hermes help", is_group=True, chat_id="grp1") is True def test_group_accepted_when_chat_in_free_response_list(self, monkeypatch): adapter = _make_gating_adapter( monkeypatch, extra={"require_mention": True, "free_response_chats": ["grp1"]}, ) msg = MagicMock(is_in_at_list=False) assert adapter._should_process_message(msg, "hi", is_group=True, chat_id="grp1") is True # Different group still blocked assert adapter._should_process_message(msg, "hi", is_group=True, chat_id="grp2") is False # --------------------------------------------------------------------------- # _IncomingHandler.process — session_webhook extraction & fire-and-forget # --------------------------------------------------------------------------- class TestIncomingHandlerProcess: """Verify that _IncomingHandler.process correctly converts callback data and dispatches message processing as a background task (fire-and-forget) so the SDK ACK is returned immediately.""" @pytest.mark.asyncio async def test_process_extracts_session_webhook(self): """session_webhook must be populated from callback data.""" from gateway.platforms.dingtalk import _IncomingHandler, DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) adapter._on_message = AsyncMock() handler = _IncomingHandler(adapter, asyncio.get_running_loop()) callback = MagicMock() callback.data = { "msgtype": "text", "text": {"content": "hello"}, "senderId": "user1", "conversationId": "conv1", "sessionWebhook": "https://oapi.dingtalk.com/robot/sendBySession?session=abc", "msgId": "msg-001", } result = await handler.process(callback) # Should return ACK immediately (STATUS_OK = 200) assert result[0] == 200 # Let the background task run await asyncio.sleep(0.05) # _on_message should have been called with a ChatbotMessage adapter._on_message.assert_called_once() chatbot_msg = adapter._on_message.call_args[0][0] assert chatbot_msg.session_webhook == "https://oapi.dingtalk.com/robot/sendBySession?session=abc" @pytest.mark.asyncio async def test_process_fallback_session_webhook_when_from_dict_misses_it(self): """If ChatbotMessage.from_dict does not map sessionWebhook (e.g. SDK version mismatch), the handler should fall back to extracting it directly from the raw data dict.""" from gateway.platforms.dingtalk import _IncomingHandler, DingTalkAdapter adapter = DingTalkAdapter(PlatformConfig(enabled=True)) adapter._on_message = AsyncMock() handler = _IncomingHandler(adapter, asyncio.get_running_loop()) callback = MagicMock() # Use a key that from_dict might not recognise in some SDK versions callback.data = { "msgtype": "text", "text": {"content": "hi"}, "senderId": "user2", "conversationId": "conv2", "session_webhook": "https://oapi.dingtalk.com/robot/sendBySession?session=def", "msgId": "msg-002", } await handler.process(callback) await asyncio.sleep(0.05) adapter._on_message.assert_called_once() chatbot_msg = adapter._on_message.call_args[0][0] assert chatbot_msg.session_webhook == "https://oapi.dingtalk.com/robot/sendBySession?session=def" @pytest.mark.asyncio async def test_process_returns_ack_immediately(self): """process() must not block on _on_message — it should return the ACK tuple before the message is fully processed.""" from gateway.platforms.dingtalk import _IncomingHandler, DingTalkAdapter processing_started = asyncio.Event() processing_gate = asyncio.Event() async def slow_on_message(msg): processing_started.set() await processing_gate.wait() # Block until we release adapter = DingTalkAdapter(PlatformConfig(enabled=True)) adapter._on_message = slow_on_message handler = _IncomingHandler(adapter, asyncio.get_running_loop()) callback = MagicMock() callback.data = { "msgtype": "text", "text": {"content": "test"}, "senderId": "u", "conversationId": "c", "sessionWebhook": "https://oapi.dingtalk.com/x", "msgId": "m", } # process() should return immediately even though _on_message blocks result = await handler.process(callback) assert result[0] == 200 # Clean up: release the gate so the background task finishes processing_gate.set() await asyncio.sleep(0.05)