ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor: remove 27 duplicate _ensure_telegram_mock() copies from gateway tests

Browse Source 
tests/gateway/conftest.py already installs a comprehensive telegram mock
at collection time (line 330), before any test module's imports run.
The per-file copies were fully redundant — each was a simpler subset
(plain strings, setdefault, fewer error classes) of the conftest version
(which uses _fake_str_enum for PTB-faithful StrEnum semantics, sys.modules
overwrite to win over partial/broken imports, and a full error hierarchy
including BadRequest, Forbidden, RetryAfter, Conflict, InvalidToken).

Removed: function def + module-level call + now-unused imports (sys,
MagicMock where no longer referenced) + dangling comment blocks that
referenced the deleted mock, in 27 test files.
Left untouched: tests/gateway/conftest.py (canonical source) and
tests/e2e/conftest.py (separate conftest tree that may run in isolation).

Found by /simplify-code review of PR #90560.
This commit is contained in:
kshitijk4poor
2026-08-21 14:03:23 +05:30
parent fc9cbc872d
commit c1693d7dcc
27 changed files with 2 additions and 599 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/gateway/test_gateway_platform_event_hook.py
-20
View File
@@ -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
tests/gateway/test_send_image_file.py
-21
View File
@@ -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
tests/gateway/test_send_multiple_images.py
-21
View File
@@ -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
tests/gateway/test_telegram_approval_buttons.py
-30
View File
@@ -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
tests/gateway/test_telegram_clarify_buttons.py
-24
View File
@@ -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
tests/gateway/test_telegram_closewait_limits_31599.py
+1 -20
View File
@@ -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
tests/gateway/test_telegram_conflict.py
-29
View File
@@ -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
tests/gateway/test_telegram_connect.py
-27
View File
@@ -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
tests/gateway/test_telegram_documents.py
-23
View File
@@ -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
tests/gateway/test_telegram_format.py
-18
View File
@@ -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,
tests/gateway/test_telegram_init_deadline.py
-24
View File
@@ -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
tests/gateway/test_telegram_max_doc_bytes.py
-22
View File
@@ -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
tests/gateway/test_telegram_media_read_timeout.py
-29
View File
@@ -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
tests/gateway/test_telegram_model_picker.py
-27
View File
@@ -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
tests/gateway/test_telegram_network_reconnect.py
-21
View File
@@ -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
tests/gateway/test_telegram_pending_update_probe.py
-17
View File
@@ -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
tests/gateway/test_telegram_polling_health_confirmation.py
-22
View File
@@ -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
tests/gateway/test_telegram_reply_mode.py
-20
View File
@@ -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
tests/gateway/test_telegram_reply_quote.py
-22
View File
@@ -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
tests/gateway/test_telegram_send_draft_format.py
-17
View File
@@ -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
tests/gateway/test_telegram_send_path_health.py
-17
View File
@@ -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
tests/gateway/test_telegram_slash_confirm.py
-24
View File
@@ -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
tests/gateway/test_telegram_start_polling_timeout.py
-22
View File
@@ -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
tests/gateway/test_telegram_status_indicator.py
-21
View File
@@ -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
tests/gateway/test_telegram_typing_backoff.py
+1 -28
View File
@@ -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
tests/gateway/test_telegram_voice_caption_markdown.py
-17
View File
@@ -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
tests/gateway/test_telegram_voice_duration.py
-16
View File
@@ -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,