diff --git a/tests/gateway/test_gateway_platform_event_hook.py b/tests/gateway/test_gateway_platform_event_hook.py index 694d392441..e65dc1418f 100644 --- a/tests/gateway/test_gateway_platform_event_hook.py +++ b/tests/gateway/test_gateway_platform_event_hook.py @@ -36,26 +36,6 @@ if _repo not in sys.path: sys.path.insert(0, _repo) -# --------------------------------------------------------------------------- -# python-telegram-bot is an optional dep; mock it so the adapter imports -# (same shim as test_telegram_network_reconnect / test_telegram_plugin_handlers). -# --------------------------------------------------------------------------- -def _ensure_telegram_mock() -> None: - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 from gateway.run import GatewayRunner # noqa: E402 from gateway.profile_routing import ProfileRoute # noqa: E402 diff --git a/tests/gateway/test_send_image_file.py b/tests/gateway/test_send_image_file.py index 103ea2fb54..374dc1dd5c 100644 --- a/tests/gateway/test_send_image_file.py +++ b/tests/gateway/test_send_image_file.py @@ -42,27 +42,6 @@ class TestExtractMediaImages: # --------------------------------------------------------------------------- # Telegram send_image_file tests # --------------------------------------------------------------------------- - - -def _ensure_telegram_mock(): - """Install mock telegram modules so TelegramAdapter can be imported.""" - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_send_multiple_images.py b/tests/gateway/test_send_multiple_images.py index 3bff6b79d0..2773b9f770 100644 --- a/tests/gateway/test_send_multiple_images.py +++ b/tests/gateway/test_send_multiple_images.py @@ -88,27 +88,6 @@ class TestBaseDefaultLoop: assert a.sent_files[0][1] == "/tmp/foo.png" -# --------------------------------------------------------------------------- -# Telegram mocks setup (shared with test_send_image_file pattern) -# --------------------------------------------------------------------------- - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_approval_buttons.py b/tests/gateway/test_telegram_approval_buttons.py index 0f6ba60db3..c9dcf38edd 100644 --- a/tests/gateway/test_telegram_approval_buttons.py +++ b/tests/gateway/test_telegram_approval_buttons.py @@ -16,36 +16,6 @@ if _repo not in sys.path: sys.path.insert(0, _repo) -# --------------------------------------------------------------------------- -# Minimal Telegram mock so TelegramAdapter can be imported -# --------------------------------------------------------------------------- -def _ensure_telegram_mock(): - """Wire up the minimal mocks required to import TelegramAdapter.""" - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - mod = MagicMock() - mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - mod.constants.ParseMode.MARKDOWN = "Markdown" - mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - mod.constants.ParseMode.HTML = "HTML" - mod.constants.ChatType.PRIVATE = "private" - mod.constants.ChatType.GROUP = "group" - mod.constants.ChatType.SUPERGROUP = "supergroup" - mod.constants.ChatType.CHANNEL = "channel" - # Provide real exception classes so ``except (NetworkError, ...)`` in - # connect() doesn't blow up under xdist when this mock leaks. - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter from gateway.config import Platform, PlatformConfig diff --git a/tests/gateway/test_telegram_clarify_buttons.py b/tests/gateway/test_telegram_clarify_buttons.py index 7d8bdbd0d6..a8a02f0765 100644 --- a/tests/gateway/test_telegram_clarify_buttons.py +++ b/tests/gateway/test_telegram_clarify_buttons.py @@ -23,30 +23,6 @@ if _repo not in sys.path: # Minimal Telegram mock so TelegramAdapter can be imported (mirrors # test_telegram_approval_buttons.py) # --------------------------------------------------------------------------- -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - mod = MagicMock() - mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - mod.constants.ParseMode.MARKDOWN = "Markdown" - mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - mod.constants.ParseMode.HTML = "HTML" - mod.constants.ChatType.PRIVATE = "private" - mod.constants.ChatType.GROUP = "group" - mod.constants.ChatType.SUPERGROUP = "supergroup" - mod.constants.ChatType.CHANNEL = "channel" - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter from gateway.config import PlatformConfig diff --git a/tests/gateway/test_telegram_closewait_limits_31599.py b/tests/gateway/test_telegram_closewait_limits_31599.py index 3c46bf5760..cc3e71df8f 100644 --- a/tests/gateway/test_telegram_closewait_limits_31599.py +++ b/tests/gateway/test_telegram_closewait_limits_31599.py @@ -26,31 +26,12 @@ client-level limits when a custom transport is supplied. """ import asyncio -import sys -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock import httpx import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram import adapter as tg_adapter # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_conflict.py b/tests/gateway/test_telegram_conflict.py index 711f874854..4fc9c6db95 100644 --- a/tests/gateway/test_telegram_conflict.py +++ b/tests/gateway/test_telegram_conflict.py @@ -1,39 +1,10 @@ import asyncio -import sys from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - - # Provide real exception classes so ``except (NetworkError, ...)`` in - # connect() doesn't blow up with "catching classes that do not inherit - # from BaseException" when another xdist worker pollutes sys.modules. - telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {}) - telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {}) - telegram_mod.error.BadRequest = type("BadRequest", (Exception,), {}) - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - sys.modules.setdefault("telegram.error", telegram_mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_connect.py b/tests/gateway/test_telegram_connect.py index 801d9a406a..721323f6ec 100644 --- a/tests/gateway/test_telegram_connect.py +++ b/tests/gateway/test_telegram_connect.py @@ -5,37 +5,10 @@ must set a non-retryable fatal error so the gateway does not queue it for background reconnection (#31049). """ -import sys -from unittest.mock import MagicMock import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - - telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {}) - telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {}) - telegram_mod.error.BadRequest = type("BadRequest", (Exception,), {}) - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - sys.modules.setdefault("telegram.error", telegram_mod.error) - - -_ensure_telegram_mock() - import plugins.platforms.telegram.adapter as telegram_mod # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_documents.py b/tests/gateway/test_telegram_documents.py index 6a537fd38b..a82003ce9d 100644 --- a/tests/gateway/test_telegram_documents.py +++ b/tests/gateway/test_telegram_documents.py @@ -10,7 +10,6 @@ We mock the telegram module at import time to avoid collection errors. import asyncio import os -import sys from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -28,28 +27,6 @@ from gateway.platforms.base import ( # --------------------------------------------------------------------------- # Mock the telegram package if it's not installed # --------------------------------------------------------------------------- - -def _ensure_telegram_mock(): - """Install mock telegram modules so TelegramAdapter can be imported.""" - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - # Real library is installed — no mocking needed - return - - telegram_mod = MagicMock() - # ContextTypes needs DEFAULT_TYPE as an actual attribute for the annotation - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - # Now we can safely import from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_format.py b/tests/gateway/test_telegram_format.py index ab6fb4d499..2195179b28 100644 --- a/tests/gateway/test_telegram_format.py +++ b/tests/gateway/test_telegram_format.py @@ -6,7 +6,6 @@ or corrupt user-visible content. """ import re -import sys from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -18,23 +17,6 @@ from gateway.config import PlatformConfig # --------------------------------------------------------------------------- # Mock the telegram package if it's not installed # --------------------------------------------------------------------------- - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - mod = MagicMock() - mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - mod.constants.ChatType.GROUP = "group" - mod.constants.ChatType.SUPERGROUP = "supergroup" - mod.constants.ChatType.CHANNEL = "channel" - mod.constants.ChatType.PRIVATE = "private" - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import ( # noqa: E402 TelegramAdapter, _escape_mdv2, diff --git a/tests/gateway/test_telegram_init_deadline.py b/tests/gateway/test_telegram_init_deadline.py index 643187c552..c62882947d 100644 --- a/tests/gateway/test_telegram_init_deadline.py +++ b/tests/gateway/test_telegram_init_deadline.py @@ -1,30 +1,6 @@ -import sys -from unittest.mock import AsyncMock, MagicMock - import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {}) - telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - sys.modules.setdefault("telegram.error", telegram_mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram import adapter as tg_adapter # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_max_doc_bytes.py b/tests/gateway/test_telegram_max_doc_bytes.py index 34aab9a01f..8d361d229c 100644 --- a/tests/gateway/test_telegram_max_doc_bytes.py +++ b/tests/gateway/test_telegram_max_doc_bytes.py @@ -5,30 +5,8 @@ The public Telegram Bot API caps `getFile` at 20MB. A locally-hosted of `extra.base_url` as the explicit opt-in to the higher cap. """ -import sys -from unittest.mock import MagicMock from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_media_read_timeout.py b/tests/gateway/test_telegram_media_read_timeout.py index 941faa5b77..017245a6d8 100644 --- a/tests/gateway/test_telegram_media_read_timeout.py +++ b/tests/gateway/test_telegram_media_read_timeout.py @@ -15,38 +15,9 @@ that actually reaches the Bot API. from __future__ import annotations -import sys from unittest.mock import AsyncMock, MagicMock import pytest - - -def _ensure_telegram_mock(): - """Install mock telegram modules only when the real library is absent. - - Probe with a real import rather than inspecting ``sys.modules``: when PTB - is installed but not yet imported, a ``sys.modules`` check alone would - install the mock first and poison every later test that needs the real - library (``tests/test_telegram_polling_progress_ptb.py`` exercises the - genuine ``BaseRequest``). - """ - try: - import telegram # noqa: F401 - - return - except ImportError: - pass - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - for name in ("GROUP", "SUPERGROUP", "CHANNEL", "PRIVATE"): - setattr(telegram_mod.constants.ChatType, name, name.lower()) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from gateway.config import PlatformConfig # noqa: E402 from plugins.platforms.telegram import adapter as tg # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_model_picker.py b/tests/gateway/test_telegram_model_picker.py index 72a0891245..259aaa072a 100644 --- a/tests/gateway/test_telegram_model_picker.py +++ b/tests/gateway/test_telegram_model_picker.py @@ -1,36 +1,9 @@ """Tests for Telegram model picker thread fallback.""" -import sys from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - mod = MagicMock() - mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - mod.constants.ParseMode.MARKDOWN = "Markdown" - mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - mod.constants.ParseMode.HTML = "HTML" - mod.constants.ChatType.PRIVATE = "private" - mod.constants.ChatType.GROUP = "group" - mod.constants.ChatType.SUPERGROUP = "supergroup" - mod.constants.ChatType.CHANNEL = "channel" - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from gateway.config import PlatformConfig from plugins.platforms.telegram.adapter import TelegramAdapter diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index 20681d6683..73b4b1611e 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -9,32 +9,11 @@ rather than silently leaving polling dead. import ast import asyncio from pathlib import Path -import sys from unittest.mock import AsyncMock, MagicMock, patch import pytest from gateway.config import GatewayConfig, Platform, PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram import adapter as tg_adapter # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 from gateway.run import GatewayRunner # noqa: E402 diff --git a/tests/gateway/test_telegram_pending_update_probe.py b/tests/gateway/test_telegram_pending_update_probe.py index cd8d3ba571..e5f2ac9742 100644 --- a/tests/gateway/test_telegram_pending_update_probe.py +++ b/tests/gateway/test_telegram_pending_update_probe.py @@ -12,28 +12,11 @@ entirely (``running=False``) with no reconnect in flight — the long-poll task is gone, so the gateway silently stops receiving messages while the process stays alive (#55769) — and feeds it into the same recovery ladder. """ -import sys from unittest.mock import AsyncMock, MagicMock, patch import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - mod = MagicMock() - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_polling_health_confirmation.py b/tests/gateway/test_telegram_polling_health_confirmation.py index 74be6364fe..612ad8467d 100644 --- a/tests/gateway/test_telegram_polling_health_confirmation.py +++ b/tests/gateway/test_telegram_polling_health_confirmation.py @@ -10,28 +10,6 @@ reconnect is a reliable hung-poll signature. import asyncio import logging -import sys -from unittest.mock import MagicMock - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {}) - telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - sys.modules.setdefault("telegram.error", telegram_mod.error) - - -_ensure_telegram_mock() - from gateway.config import Platform # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_reply_mode.py b/tests/gateway/test_telegram_reply_mode.py index f37a5b8f89..ce96d0f791 100644 --- a/tests/gateway/test_telegram_reply_mode.py +++ b/tests/gateway/test_telegram_reply_mode.py @@ -6,31 +6,11 @@ Covers the threading behavior control for multi-chunk replies: - "all": All chunks thread to original message """ import os -import sys from unittest.mock import MagicMock, AsyncMock, patch import pytest from gateway.config import PlatformConfig, GatewayConfig, Platform, _apply_env_overrides, load_gateway_config - - -def _ensure_telegram_mock(): - """Mock the telegram package if it's not installed.""" - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - mod = MagicMock() - mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - mod.constants.ChatType.GROUP = "group" - mod.constants.ChatType.SUPERGROUP = "supergroup" - mod.constants.ChatType.CHANNEL = "channel" - mod.constants.ChatType.PRIVATE = "private" - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_reply_quote.py b/tests/gateway/test_telegram_reply_quote.py index 4238192673..7150ed1dc8 100644 --- a/tests/gateway/test_telegram_reply_quote.py +++ b/tests/gateway/test_telegram_reply_quote.py @@ -8,31 +8,9 @@ message as ``reply_to_text``, which can cause it to act on unrelated actionable-looking text the user did not quote (#22619). """ -import sys from types import SimpleNamespace -from unittest.mock import MagicMock from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_send_draft_format.py b/tests/gateway/test_telegram_send_draft_format.py index 1588901615..91e2a2ec86 100644 --- a/tests/gateway/test_telegram_send_draft_format.py +++ b/tests/gateway/test_telegram_send_draft_format.py @@ -13,28 +13,11 @@ These tests pin: killing draft streaming for the whole response. 3. A non-BadRequest failure propagates so the caller falls back to edit. """ -import sys from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - mod = MagicMock() - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - import plugins.platforms.telegram.adapter as tg_mod # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_send_path_health.py b/tests/gateway/test_telegram_send_path_health.py index b449c27868..f172021f62 100644 --- a/tests/gateway/test_telegram_send_path_health.py +++ b/tests/gateway/test_telegram_send_path_health.py @@ -5,28 +5,11 @@ can enter a wedged state where ``bot.send_message()`` returns a valid Message but nothing reaches the recipient. ``_send_path_degraded`` short-circuits ``send()`` so cron's live-adapter branch falls through to standalone HTTP. """ -import sys from unittest.mock import AsyncMock, MagicMock, patch import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - mod = MagicMock() - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_slash_confirm.py b/tests/gateway/test_telegram_slash_confirm.py index bed23d4cd0..7d7f6a27a6 100644 --- a/tests/gateway/test_telegram_slash_confirm.py +++ b/tests/gateway/test_telegram_slash_confirm.py @@ -10,30 +10,6 @@ import pytest _repo = str(Path(__file__).resolve().parents[2]) if _repo not in sys.path: sys.path.insert(0, _repo) - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - mod = MagicMock() - mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - mod.constants.ParseMode.MARKDOWN = "Markdown" - mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - mod.constants.ParseMode.HTML = "HTML" - mod.constants.ChatType.PRIVATE = "private" - mod.constants.ChatType.GROUP = "group" - mod.constants.ChatType.SUPERGROUP = "supergroup" - mod.constants.ChatType.CHANNEL = "channel" - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter from gateway.config import PlatformConfig diff --git a/tests/gateway/test_telegram_start_polling_timeout.py b/tests/gateway/test_telegram_start_polling_timeout.py index 8228eaae97..c2037cc2a3 100644 --- a/tests/gateway/test_telegram_start_polling_timeout.py +++ b/tests/gateway/test_telegram_start_polling_timeout.py @@ -14,31 +14,9 @@ feeds the existing retry ladder. These tests patch the timeout down to keep the suite fast. """ import asyncio -import sys from unittest.mock import AsyncMock, MagicMock, patch import pytest - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {}) - telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - sys.modules.setdefault("telegram.error", telegram_mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram import adapter as tg_adapter # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_status_indicator.py b/tests/gateway/test_telegram_status_indicator.py index 28486bad94..16e57653bd 100644 --- a/tests/gateway/test_telegram_status_indicator.py +++ b/tests/gateway/test_telegram_status_indicator.py @@ -7,32 +7,11 @@ is enabled, the adapter sets it to "Online" on connect and "Offline" on clean disconnect so users can tell whether the gateway is up. """ -import sys from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - telegram_mod = MagicMock() - telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - telegram_mod.constants.ChatType.GROUP = "group" - telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" - telegram_mod.constants.ChatType.CHANNEL = "channel" - telegram_mod.constants.ChatType.PRIVATE = "private" - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, telegram_mod) - - -_ensure_telegram_mock() - from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_typing_backoff.py b/tests/gateway/test_telegram_typing_backoff.py index efaa2b95de..20379de8d4 100644 --- a/tests/gateway/test_telegram_typing_backoff.py +++ b/tests/gateway/test_telegram_typing_backoff.py @@ -2,40 +2,13 @@ import sys from pathlib import Path -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock import pytest _repo = str(Path(__file__).resolve().parents[2]) if _repo not in sys.path: sys.path.insert(0, _repo) - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - - mod = MagicMock() - mod.ext.ContextTypes.DEFAULT_TYPE = type(None) - mod.constants.ParseMode.MARKDOWN = "Markdown" - mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" - mod.constants.ParseMode.HTML = "HTML" - mod.constants.ChatType.PRIVATE = "private" - mod.constants.ChatType.GROUP = "group" - mod.constants.ChatType.SUPERGROUP = "supergroup" - mod.constants.ChatType.CHANNEL = "channel" - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.RetryAfter = type("RetryAfter", (Exception,), {"__init__": lambda self, retry_after=1: setattr(self, "retry_after", retry_after)}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from gateway.config import PlatformConfig from plugins.platforms.telegram.adapter import TelegramAdapter diff --git a/tests/gateway/test_telegram_voice_caption_markdown.py b/tests/gateway/test_telegram_voice_caption_markdown.py index e0c6289996..33096cdaf0 100644 --- a/tests/gateway/test_telegram_voice_caption_markdown.py +++ b/tests/gateway/test_telegram_voice_caption_markdown.py @@ -7,28 +7,11 @@ caption to MarkdownV2 (when the formatted text fits the 1024-char caption cap) and falls back to the plain truncated caption when the Bot API rejects the entities or formatting overflows. """ -import sys from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - mod = MagicMock() - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram import adapter as telegram_mod # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 diff --git a/tests/gateway/test_telegram_voice_duration.py b/tests/gateway/test_telegram_voice_duration.py index ef45593b74..f87781c5dc 100644 --- a/tests/gateway/test_telegram_voice_duration.py +++ b/tests/gateway/test_telegram_voice_duration.py @@ -24,22 +24,6 @@ from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import PlatformConfig - - -def _ensure_telegram_mock(): - if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): - return - mod = MagicMock() - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): - sys.modules.setdefault(name, mod) - sys.modules.setdefault("telegram.error", mod.error) - - -_ensure_telegram_mock() - from plugins.platforms.telegram import adapter as telegram_mod # noqa: E402 from plugins.platforms.telegram.adapter import ( # noqa: E402 TelegramAdapter,