test(telegram): fold dispatch-stall recovery tests into the delivery-gap file
Gate finding (reuse + quality, Low): the new recovery file re-declared four helpers verbatim and its re-arm test duplicated test_dispatch_progress_rearms_the_report step for step. Move the fatal-handoff test next to the existing helpers, fold the recovery-scheduling counts (0 -> 1 -> 1 -> 2) into the existing re-arm test, and delete the new file and the cross-reference docstring paragraph. The stack now adds one test function.
This commit is contained in:
kshitijk4poor
@@ -3,18 +3,16 @@
|
||||
The transport probes prove getUpdates round-trips complete; these pin the one signal they cannot
|
||||
give — whether PTB's dispatcher hands the fetched updates to a handler — and the once-per-adapter
|
||||
report for an adapter with no gateway message handler at all.
|
||||
|
||||
Recovery escalation (warn + supervisor rebuild) is pinned separately in
|
||||
``test_telegram_ingress_dispatch_recovery.py``; the counting tests below stub
|
||||
``_schedule_polling_recovery`` so the real fatal handoff does not fire mid-test.
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.config import PlatformConfig
|
||||
from gateway.platforms.base import MessageEvent, MessageType, Platform, SessionSource
|
||||
from plugins.platforms.telegram import adapter as tg_adapter
|
||||
from plugins.platforms.telegram.adapter import TelegramAdapter
|
||||
|
||||
_DEAF = "healthy but deaf"
|
||||
@@ -74,16 +72,59 @@ async def test_stall_reported_once_on_backlog_regardless_of_update_age(caplog):
|
||||
async def test_dispatch_progress_rearms_the_report(caplog):
|
||||
adapter = _polling_adapter()
|
||||
caplog.set_level(logging.WARNING)
|
||||
with patch.object(adapter, "_schedule_polling_recovery"):
|
||||
with patch.object(adapter, "_schedule_polling_recovery") as sched:
|
||||
_receive(adapter, 3)
|
||||
_heartbeats(adapter, 5)
|
||||
assert len(_deaf_reports(caplog)) == 1
|
||||
_heartbeats(adapter, 3) # 270s: a slow-but-bounded (<=300s) sequential handler is not a wedge
|
||||
assert sched.call_count == 0
|
||||
_heartbeats(adapter, 2)
|
||||
assert len(_deaf_reports(caplog)) == 1 and sched.call_count == 1
|
||||
|
||||
await _dispatch(adapter, 1) # partial drain: progress, backlog remains
|
||||
adapter._check_ingress_dispatch_stall()
|
||||
assert len(_deaf_reports(caplog)) == 1
|
||||
assert len(_deaf_reports(caplog)) == 1 and sched.call_count == 1
|
||||
_heartbeats(adapter, 4)
|
||||
assert len(_deaf_reports(caplog)) == 2
|
||||
assert len(_deaf_reports(caplog)) == 2 and sched.call_count == 2
|
||||
for call in sched.call_args_list:
|
||||
assert isinstance(call.args[0], tg_adapter._IngressDispatchStallError)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatch_stall_marks_degraded_and_goes_fatal():
|
||||
"""A confirmed dispatch stall (#130407) is a handoff, not a retry: degraded status,
|
||||
retryable fatal, no in-place updater restart, no backoff sleep."""
|
||||
adapter = _polling_adapter()
|
||||
adapter._running = True
|
||||
adapter._mark_degraded = MagicMock()
|
||||
updater = MagicMock()
|
||||
updater.stop = AsyncMock(return_value=None)
|
||||
updater.start_polling = AsyncMock()
|
||||
adapter._app = MagicMock()
|
||||
adapter._app.updater = updater
|
||||
adapter._drain_polling_connections = AsyncMock()
|
||||
adapter._notify_fatal_error = AsyncMock()
|
||||
_receive(adapter, 2)
|
||||
|
||||
try:
|
||||
with patch("asyncio.sleep", new=AsyncMock()) as sleep:
|
||||
_heartbeats(adapter, 3)
|
||||
assert adapter._polling_error_task is None
|
||||
adapter._check_ingress_dispatch_stall()
|
||||
task = adapter._polling_error_task
|
||||
assert task is not None
|
||||
await task
|
||||
assert adapter.has_fatal_error
|
||||
assert adapter.fatal_error_retryable is True
|
||||
assert "PTB dispatcher made no progress" in adapter.fatal_error_message
|
||||
adapter._notify_fatal_error.assert_awaited_once()
|
||||
adapter._mark_degraded.assert_called_once()
|
||||
updater.start_polling.assert_not_awaited()
|
||||
updater.stop.assert_not_awaited()
|
||||
sleep.assert_not_awaited()
|
||||
adapter._drain_polling_connections.assert_not_awaited()
|
||||
finally:
|
||||
for pending in tuple(adapter._background_tasks):
|
||||
pending.cancel()
|
||||
await asyncio.gather(*tuple(adapter._background_tasks), return_exceptions=True)
|
||||
|
||||
|
||||
def test_new_generation_restarts_backlog_and_ignores_fenced_polls(caplog):
|
||||
|
||||
@@ -1,119 +0,0 @@
|
||||
"""Telegram ingress dispatch stall recovery (#130407).
|
||||
|
||||
Section 1 of #130407: getUpdates kept fetching inbound messages while the PTB
|
||||
Application dispatcher stopped draining its local queue. The adapter logged
|
||||
"healthy but deaf" yet kept reporting ``connected`` and never initiated
|
||||
recovery — a manual restart was required.
|
||||
|
||||
The fix: :meth:`_check_ingress_dispatch_stall` warns once per stall AND hands
|
||||
the adapter to the supervisor for a rebuild via ``_schedule_polling_recovery``
|
||||
with :class:`_IngressDispatchStallError` (a ``_PollingStallError`` subclass, so
|
||||
the ladder skips the in-place Updater restart — the wedged dispatcher lives on
|
||||
the same ``Application`` — and goes retryable-fatal). ``_schedule`` also marks
|
||||
the adapter degraded so ``gateway_state.json`` stops saying ``connected``.
|
||||
|
||||
These pin the recovery contract; the pure counting/reporting contract lives in
|
||||
``test_telegram_ingress_delivery_gap.py`` (which stubs recovery).
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.config import PlatformConfig
|
||||
from plugins.platforms.telegram import adapter as tg_adapter
|
||||
from plugins.platforms.telegram.adapter import TelegramAdapter
|
||||
|
||||
_DEAF = "healthy but deaf"
|
||||
|
||||
|
||||
def _polling_adapter() -> TelegramAdapter:
|
||||
adapter = TelegramAdapter(PlatformConfig(enabled=True, token="test-token"))
|
||||
adapter._webhook_mode = False
|
||||
adapter._begin_polling_generation()
|
||||
return adapter
|
||||
|
||||
|
||||
def _receive(adapter: TelegramAdapter, n: int) -> None:
|
||||
request = MagicMock()
|
||||
request.parse_json_payload = MagicMock(
|
||||
return_value={"ok": True, "result": [{"update_id": i} for i in range(n)]}
|
||||
)
|
||||
adapter._observe_polling_request_result(
|
||||
request, adapter._polling_generation, (200, b"{}")
|
||||
)
|
||||
|
||||
|
||||
async def _dispatch(adapter: TelegramAdapter, n: int) -> None:
|
||||
for _ in range(n):
|
||||
await adapter._on_platform_update(MagicMock(), MagicMock())
|
||||
|
||||
|
||||
def _deaf_reports(caplog) -> list[str]:
|
||||
return [r.getMessage() for r in caplog.records if _DEAF in r.message]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatch_stall_marks_degraded_and_goes_fatal():
|
||||
"""A confirmed dispatch stall is a handoff, not a retry: degraded status,
|
||||
retryable fatal, no in-place updater restart, no backoff sleep."""
|
||||
adapter = _polling_adapter()
|
||||
adapter._running = True
|
||||
adapter._mark_degraded = MagicMock()
|
||||
updater = MagicMock()
|
||||
updater.stop = AsyncMock(return_value=None)
|
||||
updater.start_polling = AsyncMock()
|
||||
adapter._app = MagicMock()
|
||||
adapter._app.updater = updater
|
||||
adapter._drain_polling_connections = AsyncMock()
|
||||
adapter._notify_fatal_error = AsyncMock()
|
||||
_receive(adapter, 2)
|
||||
|
||||
try:
|
||||
with patch("asyncio.sleep", new=AsyncMock()) as sleep:
|
||||
for _ in range(3):
|
||||
adapter._check_ingress_dispatch_stall()
|
||||
assert adapter._polling_error_task is None
|
||||
adapter._check_ingress_dispatch_stall()
|
||||
task = adapter._polling_error_task
|
||||
assert task is not None
|
||||
await task
|
||||
assert adapter.has_fatal_error
|
||||
assert adapter.fatal_error_retryable is True
|
||||
assert "PTB dispatcher made no progress" in adapter.fatal_error_message
|
||||
adapter._notify_fatal_error.assert_awaited_once()
|
||||
adapter._mark_degraded.assert_called_once()
|
||||
updater.start_polling.assert_not_awaited()
|
||||
updater.stop.assert_not_awaited()
|
||||
sleep.assert_not_awaited()
|
||||
adapter._drain_polling_connections.assert_not_awaited()
|
||||
finally:
|
||||
for pending in tuple(adapter._background_tasks):
|
||||
pending.cancel()
|
||||
await asyncio.gather(*tuple(adapter._background_tasks), return_exceptions=True)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatch_progress_rearms_recovery_scheduling(caplog):
|
||||
"""With the handoff stubbed (no fatal), partial dispatch progress re-arms the next escalation."""
|
||||
adapter = _polling_adapter()
|
||||
caplog.set_level(logging.WARNING)
|
||||
_receive(adapter, 3)
|
||||
with patch.object(adapter, "_schedule_polling_recovery") as sched:
|
||||
for _ in range(3): # 270s: a slow-but-bounded (<=300s) sequential handler is not a wedge
|
||||
adapter._check_ingress_dispatch_stall()
|
||||
assert sched.call_count == 0
|
||||
adapter._check_ingress_dispatch_stall()
|
||||
assert sched.call_count == 1
|
||||
assert len(_deaf_reports(caplog)) == 1
|
||||
await _dispatch(adapter, 1)
|
||||
adapter._check_ingress_dispatch_stall()
|
||||
assert sched.call_count == 1
|
||||
for _ in range(4):
|
||||
adapter._check_ingress_dispatch_stall()
|
||||
assert sched.call_count == 2
|
||||
assert len(_deaf_reports(caplog)) == 2
|
||||
for call in sched.call_args_list:
|
||||
(err,) = call.args
|
||||
assert isinstance(err, tg_adapter._IngressDispatchStallError)
|
||||
Reference in New Issue
Block a user