mypointvpn/src/Telegram/testing.py

"""Помощники для тестов приложений, которые строят интерфейсы ботов."""
from types import SimpleNamespace

from telebot import types

USER_ID = 7
CHAT_ID = 42


class FakeApi:
	"""Подменяет TeleBot: запоминает отправленные и удаленные сообщения, отдает присланные документы."""

	def __init__(self):
		self.sent = []
		self.deleted = []
		self.documents = []
		self.files = {}
		"""file_id -> содержимое присланного документа."""

	def get_file(self, file_id):
		from .api import WORK_DIR, uses_own_server

		# свой сервер отдает путь на своем диске, облако - относительный
		path = f'{WORK_DIR}/1:token/documents/{file_id}' if uses_own_server() else f'documents/{file_id}'
		return SimpleNamespace(file_id=file_id, file_path=path, file_size=len(self.files[file_id]))

	def download_file(self, file_path):
		return self.files[file_path.rsplit('/', 1)[-1]]

	def send_document(self, chat_id, document, caption=None, reply_markup=None, parse_mode=None):
		content = document.file.read()
		self.documents.append(SimpleNamespace(
			chat_id=chat_id, file_name=document.file_name, content=content, caption=caption, reply_markup=reply_markup,
		))
		self.sent.append(SimpleNamespace(chat_id=chat_id, text=caption, reply_markup=reply_markup, parse_mode=parse_mode))
		return SimpleNamespace(message_id=len(self.sent))

	def send_message(self, chat_id, text, reply_markup=None, parse_mode=None):
		self.sent.append(SimpleNamespace(chat_id=chat_id, text=text, reply_markup=reply_markup, parse_mode=parse_mode))
		return SimpleNamespace(message_id=len(self.sent))

	def delete_message(self, chat_id, message_id):
		self.deleted.append((chat_id, message_id))

	@property
	def last_text(self):
		return self.sent[-1].text if self.sent else None

	@property
	def last_keyboard(self) -> list[list[str]] | None:
		markup = self.sent[-1].reply_markup if self.sent else None
		keyboard = getattr(markup, 'keyboard', None)
		return [[button['text'] for button in row] for row in keyboard] if keyboard is not None else None


def message_update(update_id: int, text: str, chat_type: str = 'private', user_id: int = USER_ID) -> types.Update:
	return types.Update.de_json(message_update_json(update_id, text, chat_type, user_id))


def message_update_json(update_id: int, text: str, chat_type: str = 'private', user_id: int = USER_ID) -> dict:
	"""Обновление с сообщением в том виде, в каком его присылает Bot API."""
	return {
		'update_id': update_id,
		'message': {
			'message_id': update_id,
			'date': 0,
			'text': text,
			'chat': {'id': CHAT_ID if user_id == USER_ID else user_id, 'type': chat_type,
				'title': 'group' if chat_type != 'private' else None},
			'from': {'id': user_id, 'is_bot': False, 'first_name': 'Ann', 'username': f'user{user_id}'},
		},
	}


def document_update_json(
		update_id: int, file_id: str, file_name: str, size: int, caption: str = '', user_id: int = USER_ID,
) -> dict:
	"""Обновление с документом в том виде, в каком его присылает Bot API."""
	update = message_update_json(update_id, '', user_id=user_id)
	message = update['message']
	del message['text']
	message['document'] = {'file_id': file_id, 'file_unique_id': file_id, 'file_name': file_name, 'file_size': size}
	if caption:
		message['caption'] = caption
	return update


class Conversation:
	"""Диалог пользователя с ботом: say() отправляет сообщение и возвращает ответ бота."""

	def __init__(self, bot, user_id: int = USER_ID, api: FakeApi | None = None):
		from .dispatcher import handle_update

		self._handle_update = handle_update
		self.bot = bot
		self.user_id = user_id
		self.api = api or FakeApi()
		self._update_id = 0

	def say(self, text: str) -> str | None:
		self._update_id += 1
		sent_before = len(self.api.sent)
		self._handle_update(self.api, self.bot, message_update(self._update_id, text, user_id=self.user_id))
		return self.api.last_text if len(self.api.sent) > sent_before else None

	def send_document(self, file_name: str, content: bytes, caption: str = '') -> str | None:
		"""Присылает боту документ и возвращает ответ бота."""
		self._update_id += 1
		file_id = f'file{self._update_id}'
		self.api.files[file_id] = content
		sent_before = len(self.api.sent)
		update = document_update_json(self._update_id, file_id, file_name, len(content), caption, self.user_id)
		self._handle_update(self.api, self.bot, types.Update.de_json(update))
		return self.api.last_text if len(self.api.sent) > sent_before else None