test(telegram): cover slotted getUpdates request instrumentation (#64482)

Adds a __slots__ request double that has no instance __dict__, mirroring
PTB's HTTPXRequest shape on Python 3.13. The test asserts the old
instance monkey-patch is rejected as read-only and that
_instrument_polling_request instead re-tags the class and still records
getUpdates progress. Fails on the pre-fix adapter with the exact
"'do_request' is read-only" AttributeError from the report.
This commit is contained in:
HexLab98 2026-07-14 23:02:33 +07:00 committed by Teknium
parent f5f79ff1b4
commit 2dd27c7fcb

View file

@ -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