feat(gateway): add native Weixin/WeChat support via iLink Bot API
Add first-class Weixin platform adapter for personal WeChat accounts: - Long-poll inbound delivery via iLink getupdates - AES-128-ECB encrypted CDN media upload/download - QR-code login flow for gateway setup wizard - context_token persistence for reply continuity - DM/group access policies with allowlists - Native text, image, video, file, voice handling - Markdown formatting with header rewriting and table-to-list conversion - Block-aware message chunking (preserves fenced code blocks) - Typing indicators via getconfig/sendtyping - SSRF protection on remote media downloads - Message deduplication with TTL Integration across all gateway touchpoints: - Platform enum, config, env overrides, connected platforms check - Adapter creation in gateway runner - Authorization maps (allowed users, allow all) - Cron delivery routing - send_message tool with native media support - Toolset definition (hermes-weixin) - Channel directory (session-based) - Platform hint in prompt builder - CLI status display - hermes tools default toolset mapping Co-authored-by: Zihan Huang <bravohenry@users.noreply.github.com>
This commit is contained in:
Zihan Huang
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,214 @@
|
||||
"""Tests for the Weixin platform adapter."""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from gateway.config import PlatformConfig
|
||||
from gateway.config import GatewayConfig, HomeChannel, Platform, _apply_env_overrides
|
||||
from gateway.platforms.weixin import WeixinAdapter
|
||||
from tools.send_message_tool import _parse_target_ref, _send_to_platform
|
||||
|
||||
|
||||
def _make_adapter() -> WeixinAdapter:
|
||||
return WeixinAdapter(
|
||||
PlatformConfig(
|
||||
enabled=True,
|
||||
token="test-token",
|
||||
extra={"account_id": "test-account"},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestWeixinFormatting:
|
||||
def test_format_message_preserves_markdown_and_rewrites_headers(self):
|
||||
adapter = _make_adapter()
|
||||
|
||||
content = "# Title\n\n## Plan\n\nUse **bold** and [docs](https://example.com)."
|
||||
|
||||
assert (
|
||||
adapter.format_message(content)
|
||||
== "【Title】\n\n**Plan**\n\nUse **bold** and [docs](https://example.com)."
|
||||
)
|
||||
|
||||
def test_format_message_rewrites_markdown_tables(self):
|
||||
adapter = _make_adapter()
|
||||
|
||||
content = (
|
||||
"| Setting | Value |\n"
|
||||
"| --- | --- |\n"
|
||||
"| Timeout | 30s |\n"
|
||||
"| Retries | 3 |\n"
|
||||
)
|
||||
|
||||
assert adapter.format_message(content) == (
|
||||
"- Setting: Timeout\n"
|
||||
" Value: 30s\n"
|
||||
"- Setting: Retries\n"
|
||||
" Value: 3"
|
||||
)
|
||||
|
||||
def test_format_message_preserves_fenced_code_blocks(self):
|
||||
adapter = _make_adapter()
|
||||
|
||||
content = "## Snippet\n\n```python\nprint('hi')\n```"
|
||||
|
||||
assert adapter.format_message(content) == "**Snippet**\n\n```python\nprint('hi')\n```"
|
||||
|
||||
def test_format_message_returns_empty_string_for_none(self):
|
||||
adapter = _make_adapter()
|
||||
|
||||
assert adapter.format_message(None) == ""
|
||||
|
||||
|
||||
class TestWeixinChunking:
|
||||
def test_split_text_sends_top_level_newlines_as_separate_messages(self):
|
||||
adapter = _make_adapter()
|
||||
|
||||
content = adapter.format_message("第一行\n第二行\n第三行")
|
||||
chunks = adapter._split_text(content)
|
||||
|
||||
assert chunks == ["第一行", "第二行", "第三行"]
|
||||
|
||||
def test_split_text_keeps_indented_followup_with_previous_line(self):
|
||||
adapter = _make_adapter()
|
||||
|
||||
content = adapter.format_message(
|
||||
"| Setting | Value |\n"
|
||||
"| --- | --- |\n"
|
||||
"| Timeout | 30s |\n"
|
||||
"| Retries | 3 |\n"
|
||||
)
|
||||
chunks = adapter._split_text(content)
|
||||
|
||||
assert chunks == [
|
||||
"- Setting: Timeout\n Value: 30s",
|
||||
"- Setting: Retries\n Value: 3",
|
||||
]
|
||||
|
||||
def test_split_text_keeps_complete_code_block_together_when_possible(self):
|
||||
adapter = _make_adapter()
|
||||
adapter.MAX_MESSAGE_LENGTH = 80
|
||||
|
||||
content = adapter.format_message(
|
||||
"## Intro\n\nShort paragraph.\n\n```python\nprint('hello world')\nprint('again')\n```\n\nTail paragraph."
|
||||
)
|
||||
chunks = adapter._split_text(content)
|
||||
|
||||
assert len(chunks) >= 2
|
||||
assert any(
|
||||
"```python\nprint('hello world')\nprint('again')\n```" in chunk
|
||||
for chunk in chunks
|
||||
)
|
||||
assert all(chunk.count("```") % 2 == 0 for chunk in chunks)
|
||||
|
||||
def test_split_text_safely_splits_long_code_blocks(self):
|
||||
adapter = _make_adapter()
|
||||
adapter.MAX_MESSAGE_LENGTH = 70
|
||||
|
||||
lines = "\n".join(f"line_{idx:02d} = {idx}" for idx in range(10))
|
||||
content = adapter.format_message(f"```python\n{lines}\n```")
|
||||
chunks = adapter._split_text(content)
|
||||
|
||||
assert len(chunks) > 1
|
||||
assert all(len(chunk) <= adapter.MAX_MESSAGE_LENGTH for chunk in chunks)
|
||||
assert all(chunk.count("```") >= 2 for chunk in chunks)
|
||||
|
||||
|
||||
class TestWeixinConfig:
|
||||
def test_apply_env_overrides_configures_weixin(self):
|
||||
config = GatewayConfig()
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"WEIXIN_ACCOUNT_ID": "bot-account",
|
||||
"WEIXIN_TOKEN": "bot-token",
|
||||
"WEIXIN_BASE_URL": "https://ilink.example.com/",
|
||||
"WEIXIN_CDN_BASE_URL": "https://cdn.example.com/c2c/",
|
||||
"WEIXIN_DM_POLICY": "allowlist",
|
||||
"WEIXIN_ALLOWED_USERS": "wxid_1,wxid_2",
|
||||
"WEIXIN_HOME_CHANNEL": "wxid_1",
|
||||
"WEIXIN_HOME_CHANNEL_NAME": "Primary DM",
|
||||
},
|
||||
clear=True,
|
||||
):
|
||||
_apply_env_overrides(config)
|
||||
|
||||
platform_config = config.platforms[Platform.WEIXIN]
|
||||
assert platform_config.enabled is True
|
||||
assert platform_config.token == "bot-token"
|
||||
assert platform_config.extra["account_id"] == "bot-account"
|
||||
assert platform_config.extra["base_url"] == "https://ilink.example.com"
|
||||
assert platform_config.extra["cdn_base_url"] == "https://cdn.example.com/c2c"
|
||||
assert platform_config.extra["dm_policy"] == "allowlist"
|
||||
assert platform_config.extra["allow_from"] == "wxid_1,wxid_2"
|
||||
assert platform_config.home_channel == HomeChannel(Platform.WEIXIN, "wxid_1", "Primary DM")
|
||||
|
||||
def test_get_connected_platforms_includes_weixin_with_token(self):
|
||||
config = GatewayConfig(
|
||||
platforms={
|
||||
Platform.WEIXIN: PlatformConfig(
|
||||
enabled=True,
|
||||
token="bot-token",
|
||||
extra={"account_id": "bot-account"},
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
assert config.get_connected_platforms() == [Platform.WEIXIN]
|
||||
|
||||
def test_get_connected_platforms_requires_account_id(self):
|
||||
config = GatewayConfig(
|
||||
platforms={
|
||||
Platform.WEIXIN: PlatformConfig(
|
||||
enabled=True,
|
||||
token="bot-token",
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
assert config.get_connected_platforms() == []
|
||||
|
||||
|
||||
class TestWeixinSendMessageIntegration:
|
||||
def test_parse_target_ref_accepts_weixin_ids(self):
|
||||
assert _parse_target_ref("weixin", "wxid_test123") == ("wxid_test123", None, True)
|
||||
assert _parse_target_ref("weixin", "filehelper") == ("filehelper", None, True)
|
||||
assert _parse_target_ref("weixin", "group@chatroom") == ("group@chatroom", None, True)
|
||||
|
||||
@patch("tools.send_message_tool._send_weixin", new_callable=AsyncMock)
|
||||
def test_send_to_platform_routes_weixin_media_to_native_helper(self, send_weixin_mock):
|
||||
send_weixin_mock.return_value = {"success": True, "platform": "weixin", "chat_id": "wxid_test123"}
|
||||
config = PlatformConfig(enabled=True, token="bot-token", extra={"account_id": "bot-account"})
|
||||
|
||||
result = asyncio.run(
|
||||
_send_to_platform(
|
||||
Platform.WEIXIN,
|
||||
config,
|
||||
"wxid_test123",
|
||||
"hello",
|
||||
media_files=[("/tmp/demo.png", False)],
|
||||
)
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
send_weixin_mock.assert_awaited_once_with(
|
||||
config,
|
||||
"wxid_test123",
|
||||
"hello",
|
||||
media_files=[("/tmp/demo.png", False)],
|
||||
)
|
||||
|
||||
|
||||
class TestWeixinRemoteMediaSafety:
|
||||
def test_download_remote_media_blocks_unsafe_urls(self):
|
||||
adapter = _make_adapter()
|
||||
|
||||
with patch("tools.url_safety.is_safe_url", return_value=False):
|
||||
try:
|
||||
asyncio.run(adapter._download_remote_media("http://127.0.0.1/private.png"))
|
||||
except ValueError as exc:
|
||||
assert "Blocked unsafe URL" in str(exc)
|
||||
else:
|
||||
raise AssertionError("expected ValueError for unsafe URL")
|
||||
Reference in New Issue
Block a user