From 202be02ac9a1b6ac8b640e9b2e0917fae6806f4b Mon Sep 17 00:00:00 2001 From: Roseyco-management Date: Mon, 13 Jul 2026 03:57:50 +0100 Subject: [PATCH] test(telegram): define polling progress contract --- plugins/platforms/telegram/adapter.py | 42 +++++ .../gateway/test_telegram_polling_progress.py | 171 ++++++++++++++++++ 2 files changed, 213 insertions(+) create mode 100644 tests/gateway/test_telegram_polling_progress.py diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 05506c79da48..e617b7b0bd23 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -633,6 +633,10 @@ class TelegramAdapter(BasePlatformAdapter): self._polling_error_task: Optional[asyncio.Task] = None self._polling_conflict_count: int = 0 self._polling_network_error_count: int = 0 + self._polling_generation: int = 0 + self._polling_progress_event = asyncio.Event() + self._polling_progress_accepting: bool = False + self._polling_progress_verifier_task: Optional[asyncio.Task] = None self._polling_error_callback_ref = None self._polling_heartbeat_task: Optional[asyncio.Task] = None # Consecutive heartbeat probes that saw queued updates the running @@ -1944,6 +1948,43 @@ class TelegramAdapter(BasePlatformAdapter): self.name, exc_info=True, ) + def _begin_polling_generation(self) -> tuple[int, asyncio.Event]: + """Start accepting progress for a new getUpdates polling generation.""" + verifier = self._polling_progress_verifier_task + if verifier is not None and not verifier.done(): + verifier.cancel() + self._polling_progress_verifier_task = None + self._polling_generation += 1 + self._polling_progress_event = asyncio.Event() + self._polling_progress_accepting = True + self._send_path_degraded = True + return self._polling_generation, self._polling_progress_event + + def _record_polling_progress(self, generation: int) -> None: + """Record successful getUpdates I/O for the current generation only.""" + if not self._polling_progress_accepting: + return + if generation != self._polling_generation: + return + self._polling_progress_event.set() + self._polling_network_error_count = 0 + self._send_path_degraded = False + + def _instrument_polling_request(self, request): + """Wrap one dedicated PTB getUpdates request with progress tracking.""" + do_request = request.do_request + + async def _do_request(*args, **kwargs): + generation = self._polling_generation + result = await do_request(*args, **kwargs) + status_code, _ = result + if 200 <= status_code < 300: + self._record_polling_progress(generation) + return result + + request.do_request = _do_request + return request + def _get_general_request_drain_lock(self) -> asyncio.Lock: lock = getattr(self, "_general_request_drain_lock", None) if lock is None: @@ -3246,6 +3287,7 @@ class TelegramAdapter(BasePlatformAdapter): **request_kwargs, httpx_kwargs=_with_limits() ) + get_updates_request = self._instrument_polling_request(get_updates_request) builder = builder.request(request).get_updates_request(get_updates_request) self._app = builder.build() self._bot = self._app.bot diff --git a/tests/gateway/test_telegram_polling_progress.py b/tests/gateway/test_telegram_polling_progress.py new file mode 100644 index 000000000000..717eaaf868a5 --- /dev/null +++ b/tests/gateway/test_telegram_polling_progress.py @@ -0,0 +1,171 @@ +"""Behavior contract for generation-safe Telegram polling progress.""" + +import asyncio + +import pytest + +from gateway.config import PlatformConfig +from plugins.platforms.telegram import adapter as tg_adapter +from plugins.platforms.telegram.adapter import TelegramAdapter + + +class _ControlledRequest: + """Minimal PTB request double with controllable completion.""" + + instances = [] + + def __init__(self, *args, result=None, error=None, entered=None, release=None, **kwargs): + self.result = result + self.error = error + self.entered = entered + self.release = release + self.args = args + self.kwargs = kwargs + type(self).instances.append(self) + + async def do_request(self, *args, **kwargs): + if self.entered is not None: + self.entered.set() + if self.release is not None: + await self.release.wait() + if self.error is not None: + raise self.error + return self.result + + +def _make_adapter() -> TelegramAdapter: + return TelegramAdapter(PlatformConfig(enabled=True, token="test-token")) + + +@pytest.mark.asyncio +async def test_current_polling_generation_success_records_progress(): + adapter = _make_adapter() + generation, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 3 + request = _ControlledRequest(result=(200, b'{"ok":true}')) + + instrumented = adapter._instrument_polling_request(request) + result = await instrumented.do_request("https://api.telegram.org/getUpdates") + + assert instrumented is request + assert result == (200, b'{"ok":true}') + assert progress.is_set() + assert adapter._polling_network_error_count == 0 + assert adapter._send_path_degraded is False + assert generation > 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error_type", [RuntimeError, asyncio.CancelledError]) +async def test_unsuccessful_polling_request_does_not_record_progress(error_type): + adapter = _make_adapter() + _, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 3 + request = adapter._instrument_polling_request( + _ControlledRequest(error=error_type("request did not complete")) + ) + + with pytest.raises(error_type): + await request.do_request("https://api.telegram.org/getUpdates") + + assert not progress.is_set() + assert adapter._polling_network_error_count == 3 + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_http_error_response_does_not_record_polling_progress(): + adapter = _make_adapter() + _, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 3 + request = adapter._instrument_polling_request( + _ControlledRequest(result=(500, b"bad")) + ) + + result = await request.do_request("https://api.telegram.org/getUpdates") + + assert result == (500, b"bad") + assert not progress.is_set() + assert adapter._polling_network_error_count == 3 + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_general_request_success_cannot_record_polling_progress(monkeypatch): + class _StopConnect(Exception): + pass + + class _Builder: + def __init__(self): + self.general_request = None + self.polling_request = None + + def token(self, _token): + return self + + def request(self, request): + self.general_request = request + return self + + def get_updates_request(self, request): + self.polling_request = request + return self + + def build(self): + raise _StopConnect + + builder = _Builder() + + class _Application: + @staticmethod + def builder(): + return builder + + _ControlledRequest.instances = [] + + async def _no_fallback_ips(): + return [] + + monkeypatch.setattr(tg_adapter, "Application", _Application) + monkeypatch.setattr(tg_adapter, "HTTPXRequest", _ControlledRequest) + monkeypatch.setattr(tg_adapter, "discover_fallback_ips", _no_fallback_ips) + monkeypatch.setattr(tg_adapter, "resolve_proxy_url", lambda *args, **kwargs: None) + + adapter = _make_adapter() + monkeypatch.setattr(adapter, "_acquire_platform_lock", lambda *args, **kwargs: True) + monkeypatch.setattr(adapter, "_fallback_ips", lambda: []) + _, progress = adapter._begin_polling_generation() + + assert await adapter.connect() is False + assert builder.general_request is _ControlledRequest.instances[0] + assert builder.polling_request is _ControlledRequest.instances[1] + + builder.general_request.result = (200, b'{"ok":true}') + result = await builder.general_request.do_request("https://api.telegram.org/sendMessage") + + assert result == (200, b'{"ok":true}') + assert not progress.is_set() + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_late_previous_generation_completion_cannot_heal_current_generation(): + adapter = _make_adapter() + generation_1, _ = adapter._begin_polling_generation() + entered = asyncio.Event() + release = asyncio.Event() + request = adapter._instrument_polling_request( + _ControlledRequest(result=(200, b'{"ok":true}'), entered=entered, release=release) + ) + + completion = asyncio.create_task(request.do_request("getUpdates")) + await entered.wait() + generation_2, progress_2 = adapter._begin_polling_generation() + adapter._polling_network_error_count = 4 + release.set() + + assert await completion == (200, b'{"ok":true}') + assert generation_2 == generation_1 + 1 + assert not progress_2.is_set() + assert adapter._polling_network_error_count == 4 + assert adapter._send_path_degraded is True