fix(weixin): streaming cursor, media uploads, markdown links, blank messages (#8665)

Four fixes for the Weixin/WeChat adapter, synthesized from the best
aspects of community PRs #8407, #8521, #8360, #7695, #8308, #8525,
#7531, #8144, #8251.

1. Streaming cursor (▉) stuck permanently — WeChat doesn't support
   message editing, so the cursor appended during streaming can never
   be removed.  Add SUPPORTS_MESSAGE_EDITING = False to WeixinAdapter
   and check it in gateway/run.py to use an empty cursor for non-edit
   platforms.  (Fixes #8307, #8326)

2. Media upload failures — two bugs in _send_file():
   a) upload_full_url path used PUT (404 on WeChat CDN); now uses POST.
   b) aes_key was base64(raw_bytes) but the iLink API expects
      base64(hex_string); images showed as grey boxes.  (Fixes #8352, #7529)
   Also: unified both upload paths into _upload_ciphertext(), preferring
   upload_full_url.  Added send_video/send_voice methods and voice_item
   media builder for audio/.silk files.  Added video_md5 field.

3. Markdown links stripped — WeChat can't render [text](url), so
   format_message() now converts them to 'text (url)' plaintext.
   Code blocks are preserved.  (Fixes #7617)

4. Blank message prevention — three guards:
   a) _split_text_for_weixin_delivery('') returns [] not ['']
   b) send() filters empty/whitespace chunks before _send_text_chunk
   c) _send_message() raises ValueError for empty text as safety net

Community credit: joei4cm (#8407), lyonDan (#8521), SKFDJKLDG (#8360),
tomqiaozc (#7695), joshleeeeee (#8308), luoxiao6645(#8525),
longsizhuo (#7531), Astral-Yang (#8144), QingWei-Li (#8251).
This commit is contained in:
Teknium
2026-04-12 16:43:25 -07:00
committed by GitHub
parent c83674dd77
commit a266238e1e
3 changed files with 253 additions and 46 deletions

View File

@@ -30,7 +30,7 @@ class TestWeixinFormatting:
assert (
adapter.format_message(content)
== "【Title】\n\n**Plan**\n\nUse **bold** and [docs](https://example.com)."
== "【Title】\n\n**Plan**\n\nUse **bold** and docs (https://example.com)."
)
def test_format_message_rewrites_markdown_tables(self):
@@ -374,3 +374,149 @@ class TestWeixinRemoteMediaSafety:
assert "Blocked unsafe URL" in str(exc)
else:
raise AssertionError("expected ValueError for unsafe URL")
class TestWeixinMarkdownLinks:
"""Markdown links should be converted to plaintext since WeChat can't render them."""
def test_format_message_converts_markdown_links_to_plain_text(self):
adapter = _make_adapter()
content = "Check [the docs](https://example.com) and [GitHub](https://github.com) for details"
assert (
adapter.format_message(content)
== "Check the docs (https://example.com) and GitHub (https://github.com) for details"
)
def test_format_message_preserves_links_inside_code_blocks(self):
adapter = _make_adapter()
content = "See below:\n\n```\n[link](https://example.com)\n```\n\nDone."
result = adapter.format_message(content)
assert "[link](https://example.com)" in result
class TestWeixinBlankMessagePrevention:
"""Regression tests for the blank-bubble bugs.
Three separate guards now prevent a blank WeChat message from ever being
dispatched:
1. ``_split_text_for_weixin_delivery("")`` returns ``[]`` — not ``[""]``.
2. ``send()`` filters out empty/whitespace-only chunks before calling
``_send_text_chunk``.
3. ``_send_message()`` raises ``ValueError`` for empty text as a last-resort
safety net.
"""
def test_split_text_returns_empty_list_for_empty_string(self):
adapter = _make_adapter()
assert adapter._split_text("") == []
def test_split_text_returns_empty_list_for_empty_string_split_per_line(self):
adapter = WeixinAdapter(
PlatformConfig(
enabled=True,
extra={
"account_id": "acct",
"token": "test-tok",
"split_multiline_messages": True,
},
)
)
assert adapter._split_text("") == []
@patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock)
def test_send_empty_content_does_not_call_send_message(self, send_message_mock):
adapter = _make_adapter()
adapter._session = object()
adapter._token = "test-token"
adapter._base_url = "https://weixin.example.com"
adapter._token_store.get = lambda account_id, chat_id: "ctx-token"
result = asyncio.run(adapter.send("wxid_test123", ""))
# Empty content → no chunks → no _send_message calls
assert result.success is True
send_message_mock.assert_not_awaited()
def test_send_message_rejects_empty_text(self):
"""_send_message raises ValueError for empty/whitespace text."""
import pytest
with pytest.raises(ValueError, match="text must not be empty"):
asyncio.run(
weixin._send_message(
AsyncMock(),
base_url="https://example.com",
token="tok",
to="wxid_test",
text="",
context_token=None,
client_id="cid",
)
)
class TestWeixinStreamingCursorSuppression:
"""WeChat doesn't support message editing — cursor must be suppressed."""
def test_supports_message_editing_is_false(self):
adapter = _make_adapter()
assert adapter.SUPPORTS_MESSAGE_EDITING is False
class TestWeixinMediaBuilder:
"""Media builder uses base64(hex_key), not base64(raw_bytes) for aes_key."""
def test_image_builder_aes_key_is_base64_of_hex(self):
import base64
adapter = _make_adapter()
media_type, builder = adapter._outbound_media_builder("photo.jpg")
assert media_type == weixin.MEDIA_IMAGE
fake_hex_key = "0123456789abcdef0123456789abcdef"
expected_aes = base64.b64encode(fake_hex_key.encode("ascii")).decode("ascii")
item = builder(
encrypt_query_param="eq",
aes_key_for_api=expected_aes,
ciphertext_size=1024,
plaintext_size=1000,
filename="photo.jpg",
rawfilemd5="abc123",
)
assert item["image_item"]["media"]["aes_key"] == expected_aes
def test_video_builder_includes_md5(self):
adapter = _make_adapter()
media_type, builder = adapter._outbound_media_builder("clip.mp4")
assert media_type == weixin.MEDIA_VIDEO
item = builder(
encrypt_query_param="eq",
aes_key_for_api="fakekey",
ciphertext_size=2048,
plaintext_size=2000,
filename="clip.mp4",
rawfilemd5="deadbeef",
)
assert item["video_item"]["video_md5"] == "deadbeef"
def test_voice_builder_for_audio_files(self):
adapter = _make_adapter()
media_type, builder = adapter._outbound_media_builder("note.mp3")
assert media_type == weixin.MEDIA_VOICE
item = builder(
encrypt_query_param="eq",
aes_key_for_api="fakekey",
ciphertext_size=512,
plaintext_size=500,
filename="note.mp3",
rawfilemd5="abc",
)
assert item["type"] == weixin.ITEM_VOICE
assert "voice_item" in item
def test_voice_builder_for_silk_files(self):
adapter = _make_adapter()
media_type, builder = adapter._outbound_media_builder("recording.silk")
assert media_type == weixin.MEDIA_VOICE