"""Typed in-memory Bot API request engine for disposable Telegram E2E.""" from __future__ import annotations import json from collections.abc import Mapping from dataclasses import dataclass from enum import StrEnum from typing import Protocol, final, override from pydantic import JsonValue as PydanticJsonValue, TypeAdapter, ValidationError from telegram import Bot from telegram.error import BadRequest, TelegramError, TimedOut from telegram.request import BaseRequest, RequestData from scripts.nutricoach_v140_runtime import JsonValue class ProviderMode(StrEnum): DELIVERED = "delivered" KNOWN_FAILURE = "known_failure" TIMEOUT = "timeout" UNKNOWN = "unknown" @dataclass(frozen=True, slots=True) class TelegramAction: sequence: int method: str chat_id: str topic_id: str | None message_id: str | None has_controls: bool @dataclass(frozen=True, slots=True) class CapturedTelegramText: method: str chat_id: str topic_id: str | None message_id: str text: str @dataclass(frozen=True, slots=True) class StoredMessage: chat_id: str topic_id: str | None message_id: str text: str callbacks: tuple[str, ...] class TimeoutDefault(Protocol): @property def value(self) -> float | None: ... TimeoutInput = float | None | TimeoutDefault class CallbackUnavailableError(KeyError): """The requested fake callback does not exist exactly once.""" class KnownProviderFailure(BadRequest): """The provider proves that no Telegram mutation occurred.""" class ProviderTimeout(TimedOut): """The provider deadline elapsed before a receipt was available.""" class UnknownProviderFailure(TelegramError): """The provider mutation outcome cannot be determined.""" @final class FakeTelegramEngine: """Capture serialized Bot API effects without sockets or credentials.""" def __init__(self) -> None: self.mode: ProviderMode = ProviderMode.DELIVERED self.actions: list[TelegramAction] = [] self.captured_texts: list[CapturedTelegramText] = [] self.attempts = 0 self.messages: dict[tuple[str, str], StoredMessage] = {} self._next_message_id = 100 def set_mode(self, mode: ProviderMode) -> None: self.mode = mode def dispatch(self, method: str, fields: Mapping[str, JsonValue]) -> bytes: self.attempts += 1 self._raise_provider_outcome() if method == "sendMessage": return self._send(fields) if method == "editMessageText": return self._edit(fields) return self._reply(True) def message(self, chat_id: str, message_id: str) -> StoredMessage: return self.messages[(chat_id, message_id)] def callback(self, chat_id: str, message_id: str, operation: str) -> str: callbacks = self.message(chat_id, message_id).callbacks matches = tuple(value for value in callbacks if f":{operation}:" in value) if len(matches) != 1: raise CallbackUnavailableError return matches[0] def callback_target( self, chat_id: str, operation: str, ) -> tuple[str, str]: """Return the sole current message carrying one operation.""" matches = tuple( (message_id, callback) for (current_chat, message_id), message in self.messages.items() if current_chat == chat_id for callback in message.callbacks if f":{operation}:" in callback ) if len(matches) != 1: raise CallbackUnavailableError return matches[0] def provider_calls(self) -> int: return len(self.actions) def _send(self, fields: Mapping[str, JsonValue]) -> bytes: chat_id = str(fields.get("chat_id", "")) topic = fields.get("message_thread_id") topic_id = None if topic is None else str(topic) message_id = str(self._next_message_id) self._next_message_id += 1 text = str(fields.get("text", "")) self.captured_texts.append(CapturedTelegramText( "sendMessage", chat_id, topic_id, message_id, text, )) callbacks = self._callbacks(fields.get("reply_markup")) self.messages[(chat_id, message_id)] = StoredMessage( chat_id, topic_id, message_id, text, callbacks ) self.actions.append(TelegramAction( len(self.actions) + 1, "sendMessage", chat_id, topic_id, message_id, bool(callbacks), )) return self._message_reply(chat_id, topic_id, message_id, text) def _edit(self, fields: Mapping[str, JsonValue]) -> bytes: chat_id = str(fields.get("chat_id", "")) message_id = str(fields.get("message_id", "")) current = self.messages.get((chat_id, message_id)) topic_id = None if current is None else current.topic_id text = str(fields.get("text", "")) callbacks = self._callbacks(fields.get("reply_markup")) self.captured_texts.append(CapturedTelegramText( "editMessageText", chat_id, topic_id, message_id, text, )) self.messages[(chat_id, message_id)] = StoredMessage( chat_id, topic_id, message_id, text, callbacks ) self.actions.append(TelegramAction( len(self.actions) + 1, "editMessageText", chat_id, topic_id, message_id, bool(callbacks), )) return self._message_reply(chat_id, topic_id, message_id, text) def _raise_provider_outcome(self) -> None: failures: dict[ProviderMode, Exception | None] = { ProviderMode.DELIVERED: None, ProviderMode.KNOWN_FAILURE: KnownProviderFailure( "provider rejected before mutation" ), ProviderMode.TIMEOUT: ProviderTimeout("provider timeout"), ProviderMode.UNKNOWN: UnknownProviderFailure("provider outcome unknown"), } failure = failures[self.mode] if failure is not None: raise failure @staticmethod def _callbacks(markup: JsonValue) -> tuple[str, ...]: if not isinstance(markup, dict): return () keyboard = markup.get("inline_keyboard") if not isinstance(keyboard, list): return () callbacks: list[str] = [] for row in keyboard: if not isinstance(row, list): continue for button in row: if not isinstance(button, dict): continue callback = button.get("callback_data") if isinstance(callback, str): callbacks.append(callback) return tuple(callbacks) @staticmethod def _message_reply( chat_id: str, topic_id: str | None, message_id: str, text: str, ) -> bytes: message: dict[str, JsonValue] = { "message_id": int(message_id), "date": 0, "text": text, "chat": {"id": int(chat_id), "type": "private"}, "from": {"id": 7_000_000_002, "is_bot": True, "first_name": "Bot"}, } if topic_id is not None: message["message_thread_id"] = int(topic_id) return FakeTelegramEngine._reply(message) @staticmethod def _reply(result: JsonValue) -> bytes: return json.dumps({"ok": True, "result": result}, separators=(",", ":")).encode() @final class FakeTelegramRequest(BaseRequest): """Translate PTB request data into the deterministic fake engine.""" def __init__(self, engine: FakeTelegramEngine) -> None: self.engine = engine @property @override def read_timeout(self) -> float | None: return None @override async def initialize(self) -> None: return None @override async def shutdown(self) -> None: return None @override async def do_request( self, url: str, method: str, request_data: RequestData | None = None, read_timeout: TimeoutInput = None, write_timeout: TimeoutInput = None, connect_timeout: TimeoutInput = None, pool_timeout: TimeoutInput = None, ) -> tuple[int, bytes]: _ = (method, read_timeout, write_timeout, connect_timeout, pool_timeout) fields: dict[str, JsonValue] = {} if request_data is not None: adapter: TypeAdapter[PydanticJsonValue] = TypeAdapter(PydanticJsonValue) fields = { name: _request_value(adapter, value) for name, value in request_data.json_parameters.items() } return 200, self.engine.dispatch(url.rsplit("/", 1)[-1], fields) def _request_value( adapter: TypeAdapter[PydanticJsonValue], raw: str, ) -> JsonValue: try: return adapter.validate_json(raw) except ValidationError: return raw def fake_bot(engine: FakeTelegramEngine) -> Bot: """Build a real PTB Bot using only the in-memory request adapter.""" request = FakeTelegramRequest(engine) return Bot("7000000002:disposable", request=request, get_updates_request=request)