test(telegram): mirror PTB errors in heartbeat recovery

tests/gateway/conftest.py

@@ -69,15 +69,21 @@ def _ensure_telegram_mock() -> None:
     mod.constants.ChatType.SUPERGROUP = "supergroup"
     mod.constants.ChatType.CHANNEL = "channel"
 
-    # Real exception classes so ``except (NetworkError, ...)`` clauses
-    # in production code don't blow up with TypeError.
-    mod.error.NetworkError = type("NetworkError", (OSError,), {})
-    mod.error.TimedOut = type("TimedOut", (OSError,), {})
-    mod.error.BadRequest = type("BadRequest", (Exception,), {})
-    mod.error.Forbidden = type("Forbidden", (Exception,), {})
-    mod.error.InvalidToken = type("InvalidToken", (Exception,), {})
-    mod.error.RetryAfter = type("RetryAfter", (Exception,), {"retry_after": 1})
-    mod.error.Conflict = type("Conflict", (Exception,), {})
+    # Mirror PTB's exception hierarchy: BadRequest is a semantic API error,
+    # but inherits from NetworkError in python-telegram-bot 22.x.
+    mod.error.TelegramError = type("TelegramError", (Exception,), {})
+    mod.error.NetworkError = type("NetworkError", (mod.error.TelegramError,), {})
+    mod.error.TimedOut = type("TimedOut", (mod.error.NetworkError,), {})
+    mod.error.BadRequest = type("BadRequest", (mod.error.NetworkError,), {})
+    mod.error.Forbidden = type("Forbidden", (mod.error.TelegramError,), {})
+    mod.error.InvalidToken = type("InvalidToken", (mod.error.TelegramError,), {})
+
+    class RetryAfter(mod.error.TelegramError):
+        def __init__(self, retry_after=1):
+            self.retry_after = retry_after
+
+    mod.error.RetryAfter = RetryAfter
+    mod.error.Conflict = type("Conflict", (mod.error.TelegramError,), {})
 
     # Update.ALL_TYPES used in start_polling()
     mod.Update.ALL_TYPES = []

tests/gateway/test_telegram_network_reconnect.py

@@ -6,7 +6,9 @@ network error, the adapter must self-reschedule the next reconnect attempt
 rather than silently leaving polling dead.
 """
 
+import ast
 import asyncio
+from pathlib import Path
 import sys
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -678,6 +680,123 @@ async def test_heartbeat_loop_ignores_non_connectivity_errors():
     adapter._handle_polling_network_error.assert_not_awaited()
 
 
+async def _heartbeat_exception_case(exc, *, pending_probe=False):
+    adapter = _make_adapter()
+    reconnect_handler = AsyncMock()
+    adapter._handle_polling_network_error = reconnect_handler  # type: ignore[method-assign]
+    mock_app = MagicMock()
+    mock_app.updater.running = True
+    if pending_probe:
+        mock_app.bot.get_me = AsyncMock(return_value=MagicMock())
+        mock_app.bot.get_webhook_info = AsyncMock(side_effect=exc)
+    else:
+        mock_app.bot.get_me = AsyncMock(side_effect=exc)
+    adapter._app = mock_app
+
+    sleep_calls = 0
+
+    async def fast_sleep(_seconds):
+        nonlocal sleep_calls
+        sleep_calls += 1
+        if sleep_calls >= 2:
+            raise asyncio.CancelledError()
+
+    with patch("asyncio.sleep", side_effect=fast_sleep):
+        await adapter._polling_heartbeat_loop()
+    await asyncio.sleep(0)
+    return adapter
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("pending_probe", [False, True])
+async def test_heartbeat_routes_ptb_transport_errors_to_reconnect(pending_probe):
+    from telegram.error import NetworkError, TimedOut
+
+    for exc in (NetworkError("network"), TimedOut("timeout")):
+        adapter = await _heartbeat_exception_case(exc, pending_probe=pending_probe)
+        reconnect_handler = adapter._handle_polling_network_error
+        assert isinstance(reconnect_handler, AsyncMock)
+        reconnect_handler.assert_awaited_once_with(exc)
+        assert adapter._polling_error_task is not None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("pending_probe", [False, True])
+async def test_heartbeat_ignores_ptb_semantic_errors(pending_probe):
+    from telegram.error import BadRequest, Forbidden, InvalidToken, RetryAfter
+
+    for exc in (
+        BadRequest("bad request"),
+        Forbidden("forbidden"),
+        InvalidToken("invalid token"),
+        RetryAfter(1),
+    ):
+        adapter = await _heartbeat_exception_case(exc, pending_probe=pending_probe)
+        reconnect_handler = adapter._handle_polling_network_error
+        assert isinstance(reconnect_handler, AsyncMock)
+        reconnect_handler.assert_not_awaited()
+        assert adapter._polling_error_task is None
+
+
+@pytest.mark.parametrize(
+    ("error_name", "expected"),
+    [
+        ("NetworkError", True),
+        ("TimedOut", True),
+        ("BadRequest", False),
+        ("Forbidden", False),
+        ("InvalidToken", False),
+        ("RetryAfter", False),
+    ],
+)
+def test_network_error_classifier_matches_ptb_semantics(error_name, expected):
+    import telegram.error as telegram_error
+
+    error_type = getattr(telegram_error, error_name)
+    error = error_type(1) if error_name == "RetryAfter" else error_type(error_name)
+    assert TelegramAdapter._looks_like_network_error(error) is expected
+
+
+def _calls_shared_network_classifier(node):
+    return any(
+        isinstance(child, ast.Call)
+        and isinstance(child.func, ast.Attribute)
+        and child.func.attr == "_looks_like_network_error"
+        for child in ast.walk(node)
+    )
+
+
+def test_polling_error_callback_uses_shared_network_classifier():
+    source = Path(TelegramAdapter.connect.__code__.co_filename).read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    callbacks = [
+        node
+        for node in ast.walk(tree)
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        and node.name == "_polling_error_callback"
+    ]
+    assert len(callbacks) == 1
+    assert _calls_shared_network_classifier(callbacks[0])
+
+
+def test_connect_initialize_retry_uses_shared_network_classifier():
+    source = Path(TelegramAdapter.connect.__code__.co_filename).read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    connect = next(
+        node
+        for node in ast.walk(tree)
+        if isinstance(node, ast.AsyncFunctionDef) and node.name == "connect"
+    )
+    exception_handlers = [
+        node
+        for node in ast.walk(connect)
+        if isinstance(node, ast.ExceptHandler)
+        and isinstance(node.type, ast.Name)
+        and node.type.id == "Exception"
+    ]
+    assert any(_calls_shared_network_classifier(handler) for handler in exception_handlers)
+
+
 @pytest.mark.asyncio
 async def test_heartbeat_loop_exits_on_fatal_error():
     """A fatal error short-circuits the loop before probing get_me()."""