diff --git a/tests/test_telegram_polling_progress_ptb.py b/tests/test_telegram_polling_progress_ptb.py index c25c50335820..1eab796354cd 100644 --- a/tests/test_telegram_polling_progress_ptb.py +++ b/tests/test_telegram_polling_progress_ptb.py @@ -92,6 +92,34 @@ class _EnvelopeRequest(BaseRequest): return 200, self.payload +class _SlottedEnvelopeRequest(BaseRequest): + """A getUpdates request with no instance ``__dict__``. + + Reproduces PTB's real HTTPXRequest shape on Python 3.13, where every + class in the MRO defines ``__slots__`` and instances therefore reject an + instance-attribute ``do_request`` monkey-patch as "read-only" (#64482). + ``__slots__`` names the payload so the double needs no ``__dict__``. + """ + + __slots__ = ("_payload",) + + def __init__(self, payload): + self._payload = payload + + @property + def read_timeout(self): + return 10 + + async def initialize(self): + return None + + async def shutdown(self): + return None + + async def do_request(self, url, method, request_data=None, **_kwargs): + return 200, self._payload + + async def _cancel_task(task): if task is None or task.done(): return @@ -226,6 +254,43 @@ async def test_real_base_request_valid_success_envelope_records_progress(): assert adapter._send_path_degraded is False +@pytest.mark.asyncio +async def test_slotted_request_instrumented_without_read_only_error(): + """Instrumenting a __slots__ request must not raise, and must still record. + + Regression for #64482: PTB's HTTPXRequest carries no instance ``__dict__`` + on Python 3.13, so the old ``request.do_request = wrapper`` monkey-patch + raised ``AttributeError: ... 'do_request' is read-only`` and broke every + Telegram connect. The subclass re-tag must instrument such an instance and + still observe getUpdates progress. + """ + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="123456:test-token")) + generation, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 4 + adapter._polling_conflict_count = 3 + + request = _SlottedEnvelopeRequest(b'{"ok":true,"result":[]}') + # Guard the premise: the slotted instance rejects the old monkey-patch, + # exactly as PTB's HTTPXRequest does on Python 3.13. + with pytest.raises(AttributeError, match="read-only"): + request.do_request = lambda *a, **k: None + + instrumented = adapter._instrument_polling_request(request) + context_token = tg_adapter._POLLING_GENERATION_CONTEXT.set(generation) + try: + result = await instrumented.post( + "https://api.telegram.org/bot-token/getUpdates" + ) + finally: + tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token) + + assert result == [] + assert progress.is_set() + assert adapter._polling_network_error_count == 0 + assert adapter._polling_conflict_count == 0 + assert adapter._send_path_degraded is False + + @pytest.mark.asyncio async def test_real_ptb_stop_cleanup_cannot_heal_recovery_generation(): assert tg_adapter.TELEGRAM_AVAILABLE is True