test(send_message): add thread-not-found retry tests for Telegram topics

Three tests covering the #27012 fix:
- test_is_thread_not_found_matches_expected_errors
- test_text_send_retries_without_thread_id_on_thread_not_found
- test_disable_web_page_preview_not_leaked_to_media_sends

116/116 existing tests still pass (no regressions).
This commit is contained in:
kunci115 2026-05-17 04:33:33 +08:00 committed by Teknium
parent 2bb04f6842
commit 4abaec18b8

View file

@ -23,6 +23,7 @@ def _reset_signal_scheduler():
from gateway.config import Platform
from tools.send_message_tool import (
_derive_forum_thread_name,
_is_telegram_thread_not_found,
_parse_target_ref,
_send_discord,
_send_matrix_via_adapter,
@ -2478,3 +2479,94 @@ class TestCheckSendMessage:
patch("gateway.status.is_gateway_running",
side_effect=ImportError("simulated")):
assert _check_send_message() is False
class TestSendTelegramThreadNotFoundRetry:
"""Tests for thread-not-found retry behaviour in _send_telegram (#27012)."""
def test_is_thread_not_found_matches_expected_errors(self):
"""_is_telegram_thread_not_found should detect thread-not-found errors."""
class FakeError(Exception):
pass
assert _is_telegram_thread_not_found(FakeError("message thread not found")) is True
assert _is_telegram_thread_not_found(FakeError("THREAD NOT FOUND")) is True
assert _is_telegram_thread_not_found(FakeError("Bad Request: thread not found")) is True
assert _is_telegram_thread_not_found(FakeError("chat not found")) is False
assert _is_telegram_thread_not_found(FakeError("parse error")) is False
assert _is_telegram_thread_not_found(FakeError("")) is False
def test_text_send_retries_without_thread_id_on_thread_not_found(self):
"""When thread is not found, the text send should retry without
message_thread_id."""
call_args = []
async def fake_retry(bot, *, chat_id, text, parse_mode, **kwargs):
call_args.append(dict(kwargs, chat_id=chat_id, text=text))
if len(call_args) == 1:
raise Exception("Bad Request: message thread not found")
return SimpleNamespace(message_id=42)
async def run_test():
with patch(
"tools.send_message_tool._send_telegram_message_with_retry",
fake_retry,
):
# _send_telegram imports Bot locally; we only need to mock
# the send path, not Bot itself (Bot import falls through
# normally since python-telegram-bot is installed).
return await _send_telegram(
"fake-token", "-100123", "hello from topic 17585",
thread_id="17585",
)
result = asyncio.run(run_test())
assert result["success"] is True
assert result["message_id"] == "42"
assert len(call_args) == 2, f"expected 2 calls, got {len(call_args)}"
# First call should have message_thread_id
assert call_args[0].get("message_thread_id") is not None
# Second call (retry) should NOT have message_thread_id
assert "message_thread_id" not in call_args[1], \
"retry should drop message_thread_id after thread-not-found"
def test_disable_web_page_preview_not_leaked_to_media_sends(self):
"""disable_web_page_preview should only appear in text send, not media sends."""
text_kwargs_seen = []
media_kwargs_seen = []
class FakeBot:
async def send_message(self, **kwargs):
text_kwargs_seen.append(kwargs)
return SimpleNamespace(message_id=1)
async def send_document(self, **kwargs):
media_kwargs_seen.append(kwargs)
return SimpleNamespace(message_id=2)
import tempfile
media_path = None
try:
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tf:
tf.write(b"%PDF-1.4 test content")
media_path = tf.name
async def run_test():
with patch("telegram.Bot", return_value=FakeBot()):
return await _send_telegram(
"fake-token", "-100123", "check preview",
media_files=[(media_path, False)],
disable_link_previews=True,
)
result = asyncio.run(run_test())
assert result["success"] is True
# Text send should have disable_web_page_preview
assert text_kwargs_seen[0].get("disable_web_page_preview") is True
# Media send should NOT have disable_web_page_preview
assert "disable_web_page_preview" not in media_kwargs_seen[0], \
"disable_web_page_preview leaked into send_document kwargs"
finally:
if media_path and os.path.exists(media_path):
os.unlink(media_path)