diff --git a/tests/gateway/test_gateway_platform_event_hook.py b/tests/gateway/test_gateway_platform_event_hook.py index 694d39244113..e65dc1418f52 100644 --- a/tests/gateway/test_gateway_platform_event_hook.py +++ b/tests/gateway/test_gateway_platform_event_hook.py @@ -36,26 +36,6 @@ 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 103ea2fb54a7..374dc1dd5c2e 100644 --- a/tests/gateway/test_send_image_file.py +++ b/tests/gateway/test_send_image_file.py @@ -42,27 +42,6 @@ def test_png_image_extracted(self): # --------------------------------------------------------------------------- # 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 3bff6b79d091..2773b9f770e5 100644 --- a/tests/gateway/test_send_multiple_images.py +++ b/tests/gateway/test_send_multiple_images.py @@ -88,27 +88,6 @@ def test_loops_per_image_by_default(self): 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 0f6ba60db35a..c9dcf38edd4d 100644 --- a/tests/gateway/test_telegram_approval_buttons.py +++ b/tests/gateway/test_telegram_approval_buttons.py @@ -16,36 +16,6 @@ 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 7d8bdbd0d69d..a8a02f07659e 100644 --- a/tests/gateway/test_telegram_clarify_buttons.py +++ b/tests/gateway/test_telegram_clarify_buttons.py @@ -23,30 +23,6 @@ # 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 3c46bf5760a4..cc3e71df8f9b 100644 --- a/tests/gateway/test_telegram_closewait_limits_31599.py +++ b/tests/gateway/test_telegram_closewait_limits_31599.py @@ -26,31 +26,12 @@ """ 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 711f87485417..4fc9c6db9519 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 801d9a406ab0..721323f6ec72 100644 --- a/tests/gateway/test_telegram_connect.py +++ b/tests/gateway/test_telegram_connect.py @@ -5,37 +5,10 @@ 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 6a537fd38b4b..a82003ce9d0e 100644 --- a/tests/gateway/test_telegram_documents.py +++ b/tests/gateway/test_telegram_documents.py @@ -10,7 +10,6 @@ import asyncio import os -import sys from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -28,28 +27,6 @@ # --------------------------------------------------------------------------- # 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 ab6fb4d4996c..2195179b28f2 100644 --- a/tests/gateway/test_telegram_format.py +++ b/tests/gateway/test_telegram_format.py @@ -6,7 +6,6 @@ """ import re -import sys from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -18,23 +17,6 @@ # --------------------------------------------------------------------------- # 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 643187c552e8..c62882947d73 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 34aab9a01ff5..8d361d229ced 100644 --- a/tests/gateway/test_telegram_max_doc_bytes.py +++ b/tests/gateway/test_telegram_max_doc_bytes.py @@ -5,30 +5,8 @@ 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 941faa5b775b..017245a6d8a6 100644 --- a/tests/gateway/test_telegram_media_read_timeout.py +++ b/tests/gateway/test_telegram_media_read_timeout.py @@ -15,38 +15,9 @@ 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 72a0891245f8..259aaa072a51 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 20681d6683ed..73b4b1611eef 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -9,32 +9,11 @@ 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 cd8d3ba57117..e5f2ac97425c 100644 --- a/tests/gateway/test_telegram_pending_update_probe.py +++ b/tests/gateway/test_telegram_pending_update_probe.py @@ -12,28 +12,11 @@ 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 74be6364feba..612ad8467d6f 100644 --- a/tests/gateway/test_telegram_polling_health_confirmation.py +++ b/tests/gateway/test_telegram_polling_health_confirmation.py @@ -10,28 +10,6 @@ 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 f37a5b8f898f..ce96d0f791d8 100644 --- a/tests/gateway/test_telegram_reply_mode.py +++ b/tests/gateway/test_telegram_reply_mode.py @@ -6,31 +6,11 @@ - "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 4238192673ea..7150ed1dc8e4 100644 --- a/tests/gateway/test_telegram_reply_quote.py +++ b/tests/gateway/test_telegram_reply_quote.py @@ -8,31 +8,9 @@ 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 1588901615f9..91e2a2ec8673 100644 --- a/tests/gateway/test_telegram_send_draft_format.py +++ b/tests/gateway/test_telegram_send_draft_format.py @@ -13,28 +13,11 @@ 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 b449c278684c..f172021f626e 100644 --- a/tests/gateway/test_telegram_send_path_health.py +++ b/tests/gateway/test_telegram_send_path_health.py @@ -5,28 +5,11 @@ 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 bed23d4cd0fb..7d7f6a27a608 100644 --- a/tests/gateway/test_telegram_slash_confirm.py +++ b/tests/gateway/test_telegram_slash_confirm.py @@ -10,30 +10,6 @@ _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 8228eaae9752..c2037cc2a35e 100644 --- a/tests/gateway/test_telegram_start_polling_timeout.py +++ b/tests/gateway/test_telegram_start_polling_timeout.py @@ -14,31 +14,9 @@ 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 28486bad945b..16e57653bd99 100644 --- a/tests/gateway/test_telegram_status_indicator.py +++ b/tests/gateway/test_telegram_status_indicator.py @@ -7,32 +7,11 @@ 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 efaa2b95de2e..20379de8d428 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 e0c62899964b..33096cdaf06c 100644 --- a/tests/gateway/test_telegram_voice_caption_markdown.py +++ b/tests/gateway/test_telegram_voice_caption_markdown.py @@ -7,28 +7,11 @@ 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 ef45593b74a3..f87781c5dcab 100644 --- a/tests/gateway/test_telegram_voice_duration.py +++ b/tests/gateway/test_telegram_voice_duration.py @@ -24,22 +24,6 @@ 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,