|
|
|
@@ -6,9 +6,7 @@ import pytest
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_telegram_platform_init_raises_when_dependency_missing():
|
|
|
|
|
from messaging import telegram as telegram_mod
|
|
|
|
|
|
|
|
|
|
with patch.object(telegram_mod, "TELEGRAM_AVAILABLE", False):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", False):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
with pytest.raises(ImportError):
|
|
|
|
@@ -19,7 +17,7 @@ def test_telegram_platform_init_raises_when_dependency_missing():
|
|
|
|
|
async def test_telegram_platform_start_requires_token():
|
|
|
|
|
with (
|
|
|
|
|
patch.dict("os.environ", {}, clear=True),
|
|
|
|
|
patch("messaging.telegram.TELEGRAM_AVAILABLE", True),
|
|
|
|
|
patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True),
|
|
|
|
|
):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
@@ -30,7 +28,7 @@ async def test_telegram_platform_start_requires_token():
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_telegram_platform_stop_no_application_is_noop():
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -42,7 +40,7 @@ async def test_telegram_platform_stop_no_application_is_noop():
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_with_retry_returns_none_when_message_not_modified_network_error():
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform, NetworkError
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -55,7 +53,7 @@ async def test_with_retry_returns_none_when_message_not_modified_network_error()
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_with_retry_retries_network_error_then_succeeds(monkeypatch):
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform, NetworkError
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -76,7 +74,7 @@ async def test_with_retry_retries_network_error_then_succeeds(monkeypatch):
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_with_retry_honors_retry_after_timedelta(monkeypatch):
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform, RetryAfter
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -97,7 +95,7 @@ async def test_with_retry_honors_retry_after_timedelta(monkeypatch):
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_with_retry_drops_parse_mode_on_markdown_entity_error():
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform, TelegramError
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -116,7 +114,7 @@ async def test_with_retry_drops_parse_mode_on_markdown_entity_error():
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_queue_send_message_without_limiter_calls_send_message():
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -131,7 +129,7 @@ async def test_queue_send_message_without_limiter_calls_send_message():
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_queue_edit_message_without_limiter_calls_edit_message():
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -144,7 +142,7 @@ async def test_queue_edit_message_without_limiter_calls_edit_message():
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_fire_and_forget_non_coroutine_uses_ensure_future(monkeypatch):
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -158,7 +156,7 @@ def test_fire_and_forget_non_coroutine_uses_ensure_future(monkeypatch):
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_on_start_command_replies_and_forwards():
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -175,7 +173,7 @@ async def test_on_start_command_replies_and_forwards():
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_on_telegram_message_handler_error_sends_error_message():
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t", allowed_user_id="123")
|
|
|
|
@@ -201,7 +199,7 @@ async def test_on_telegram_message_handler_error_sends_error_message():
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_telegram_start_retries_on_network_error(monkeypatch):
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform, NetworkError
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="token", allowed_user_id=None)
|
|
|
|
@@ -226,7 +224,7 @@ async def test_telegram_start_retries_on_network_error(monkeypatch):
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_edit_message_with_text_exceeding_4096_raises():
|
|
|
|
|
"""edit_message with text > 4096 raises TelegramError (BadRequest)."""
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform, TelegramError
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -243,7 +241,7 @@ async def test_edit_message_with_text_exceeding_4096_raises():
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_edit_message_empty_string():
|
|
|
|
|
"""edit_message with empty string - Telegram accepts (no-op edit)."""
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -260,7 +258,7 @@ async def test_edit_message_empty_string():
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_send_message_empty_string():
|
|
|
|
|
"""send_message with empty string - Telegram may reject; we pass through."""
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
@@ -278,7 +276,7 @@ async def test_send_message_empty_string():
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_on_telegram_message_non_text_update_ignored():
|
|
|
|
|
"""Update with message.photo but no text returns early without calling handler."""
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t", allowed_user_id="123")
|
|
|
|
@@ -300,7 +298,7 @@ async def test_on_telegram_message_non_text_update_ignored():
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_with_retry_message_not_found_returns_none():
|
|
|
|
|
"""'message to edit not found' returns None without retry."""
|
|
|
|
|
with patch("messaging.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
|
|
|
|
|
from messaging.telegram import TelegramPlatform, TelegramError
|
|
|
|
|
|
|
|
|
|
platform = TelegramPlatform(bot_token="t")
|
|
|
|
|