izackp/takopi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(telegram): msgspec schemas and parsing (#156)

Browse Source
This commit is contained in:
banteg
2026-01-16 19:21:26 +04:00
committed by GitHub
parent 190b2f6d6e
commit c85ab2e2a2
10 changed files with 505 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/takopi/telegram/api_models.py
+25 -41
View File
@@ -1,53 +1,37 @@
from __future__ import annotations
from typing import Any
import msgspec
from .api_schemas import (
CallbackQuery,
CallbackQueryMessage,
Chat,
ChatMember,
Document,
File,
ForumTopic,
Message,
MessageReply,
PhotoSize,
Sticker,
Update,
User,
Video,
Voice,
)
__all__ = [
"CallbackQuery",
"CallbackQueryMessage",
"Chat",
"ChatMember",
"Document",
"File",
"ForumTopic",
"Message",
"MessageReply",
"PhotoSize",
"Sticker",
"Update",
"User",
"Video",
"Voice",
]
class User(msgspec.Struct, forbid_unknown_fields=False):
id: int
username: str | None = None
first_name: str | None = None
last_name: str | None = None
class Chat(msgspec.Struct, forbid_unknown_fields=False):
id: int
type: str
is_forum: bool | None = None
class ChatMember(msgspec.Struct, forbid_unknown_fields=False):
status: str
can_manage_topics: bool | None = None
class Message(msgspec.Struct, forbid_unknown_fields=False):
message_id: int
message_thread_id: int | None = None
text: str | None = None
class File(msgspec.Struct, forbid_unknown_fields=False):
file_path: str
class ForumTopic(msgspec.Struct, forbid_unknown_fields=False):
message_thread_id: int
class Update(msgspec.Struct, forbid_unknown_fields=False):
update_id: int
message: dict[str, Any] | None = None
callback_query: dict[str, Any] | None = None
src/takopi/telegram/api_schemas.py
+152
View File
@@ -0,0 +1,152 @@
"""Msgspec models for Telegram Bot API payloads (subset used by takopi).
Derived from telegram-api.html in the repository.
"""
from __future__ import annotations
import msgspec
__all__ = [
"CallbackQuery",
"CallbackQueryMessage",
"Chat",
"ChatMember",
"Document",
"File",
"ForumTopic",
"Message",
"MessageReply",
"PhotoSize",
"Sticker",
"Update",
"User",
"Video",
"Voice",
"decode_update",
"decode_updates",
]
class User(msgspec.Struct, forbid_unknown_fields=False):
id: int
is_bot: bool | None = None
username: str | None = None
first_name: str | None = None
last_name: str | None = None
class Chat(msgspec.Struct, forbid_unknown_fields=False):
id: int
type: str
title: str | None = None
username: str | None = None
first_name: str | None = None
last_name: str | None = None
is_forum: bool | None = None
class PhotoSize(msgspec.Struct, forbid_unknown_fields=False):
file_id: str
width: int
height: int
file_size: int | None = None
class Document(msgspec.Struct, forbid_unknown_fields=False):
file_id: str
file_name: str | None = None
mime_type: str | None = None
file_size: int | None = None
class Video(msgspec.Struct, forbid_unknown_fields=False):
file_id: str
file_name: str | None = None
mime_type: str | None = None
file_size: int | None = None
class Voice(msgspec.Struct, forbid_unknown_fields=False):
file_id: str
duration: int | None = None
mime_type: str | None = None
file_size: int | None = None
class Sticker(msgspec.Struct, forbid_unknown_fields=False):
file_id: str
file_size: int | None = None
class MessageReply(msgspec.Struct, forbid_unknown_fields=False):
message_id: int
text: str | None = None
from_: User | None = msgspec.field(default=None, name="from")
class Message(msgspec.Struct, forbid_unknown_fields=False):
message_id: int
chat: Chat
message_thread_id: int | None = None
from_: User | None = msgspec.field(default=None, name="from")
text: str | None = None
caption: str | None = None
reply_to_message: MessageReply | None = None
forward_from: User | None = None
forward_from_chat: Chat | None = None
forward_from_message_id: int | None = None
forward_sender_name: str | None = None
forward_signature: str | None = None
forward_date: int | None = None
media_group_id: str | None = None
is_automatic_forward: bool | None = None
is_topic_message: bool | None = None
voice: Voice | None = None
document: Document | None = None
video: Video | None = None
photo: list[PhotoSize] | None = None
sticker: Sticker | None = None
class CallbackQueryMessage(msgspec.Struct, forbid_unknown_fields=False):
message_id: int
chat: Chat
class CallbackQuery(msgspec.Struct, forbid_unknown_fields=False):
id: str
from_: User = msgspec.field(name="from")
message: CallbackQueryMessage | None = None
data: str | None = None
class Update(msgspec.Struct, forbid_unknown_fields=False):
update_id: int
message: Message | None = None
callback_query: CallbackQuery | None = None
class File(msgspec.Struct, forbid_unknown_fields=False):
file_path: str
class ChatMember(msgspec.Struct, forbid_unknown_fields=False):
status: str
can_manage_topics: bool | None = None
class ForumTopic(msgspec.Struct, forbid_unknown_fields=False):
message_thread_id: int
_UPDATE_DECODER = msgspec.json.Decoder(Update)
_UPDATES_DECODER = msgspec.json.Decoder(list[Update])
def decode_update(payload: str | bytes) -> Update:
return _UPDATE_DECODER.decode(payload)
def decode_updates(payload: str | bytes) -> list[Update]:
return _UPDATES_DECODER.decode(payload)
src/takopi/telegram/onboarding.py
+12 -22
View File
@@ -264,35 +264,25 @@ async def wait_for_chat(token: str) -> ChatInfo:
continue
if not updates:
continue
offset = updates[-1].update_id + 1
update = updates[-1]
offset = update.update_id + 1
msg = update.message
if not isinstance(msg, dict):
if msg is None:
continue
sender = msg.get("from")
if isinstance(sender, dict) and sender.get("is_bot") is True:
sender = msg.from_
if sender is not None and sender.is_bot is True:
continue
chat = msg.get("chat")
if not isinstance(chat, dict):
continue
chat_id = chat.get("id")
if not isinstance(chat_id, int):
chat = msg.chat
if chat is None:
continue
chat_id = chat.id
return ChatInfo(
chat_id=chat_id,
username=chat.get("username")
if isinstance(chat.get("username"), str)
else None,
title=chat.get("title") if isinstance(chat.get("title"), str) else None,
first_name=chat.get("first_name")
if isinstance(chat.get("first_name"), str)
else None,
last_name=chat.get("last_name")
if isinstance(chat.get("last_name"), str)
else None,
chat_type=chat.get("type")
if isinstance(chat.get("type"), str)
else None,
username=chat.username,
title=chat.title,
first_name=chat.first_name,
last_name=chat.last_name,
chat_type=chat.type,
)
finally:
await bot.close()
src/takopi/telegram/parsing.py
+117 -179
View File
@@ -1,12 +1,20 @@
from __future__ import annotations
from typing import Any
from collections.abc import AsyncIterator, Callable, Iterable
import anyio
import msgspec
from ..logging import get_logger
from .api_models import Update
from .api_schemas import (
CallbackQuery,
Document,
Message,
PhotoSize,
Sticker,
Update,
Video,
)
from .client_api import BotClient
from .types import (
TelegramCallbackQuery,
@@ -20,23 +28,20 @@ logger = get_logger(__name__)
def parse_incoming_update(
update: Update | dict[str, Any],
update: Update,
*,
chat_id: int | None = None,
chat_ids: set[int] | None = None,
) -> TelegramIncomingUpdate | None:
if isinstance(update, Update):
msg = update.message
callback_query = update.callback_query
else:
msg = update.get("message")
callback_query = update.get("callback_query")
if isinstance(msg, dict):
return _parse_incoming_message(msg, chat_id=chat_id, chat_ids=chat_ids)
if isinstance(callback_query, dict):
if update.message is not None:
return _parse_incoming_message(
update.message,
chat_id=chat_id,
chat_ids=chat_ids,
)
if update.callback_query is not None:
return _parse_callback_query(
callback_query,
update.callback_query,
chat_id=chat_id,
chat_ids=chat_ids,
)
@@ -44,167 +49,71 @@ def parse_incoming_update(
def _parse_incoming_message(
msg: dict[str, Any],
msg: Message,
*,
chat_id: int | None = None,
chat_ids: set[int] | None = None,
) -> TelegramIncomingMessage | None:
def _parse_document_payload(payload: dict[str, Any]) -> TelegramDocument | None:
file_id = payload.get("file_id")
if not isinstance(file_id, str) or not file_id:
return None
return TelegramDocument(
file_id=file_id,
file_name=payload.get("file_name")
if isinstance(payload.get("file_name"), str)
else None,
mime_type=payload.get("mime_type")
if isinstance(payload.get("mime_type"), str)
else None,
file_size=payload.get("file_size")
if isinstance(payload.get("file_size"), int)
and not isinstance(payload.get("file_size"), bool)
else None,
raw=payload,
)
raw_text = msg.get("text")
text = raw_text if isinstance(raw_text, str) else None
caption = msg.get("caption")
if text is None and isinstance(caption, str):
text = caption
raw_text = msg.text
caption = msg.caption
text = raw_text if raw_text is not None else caption
if text is None:
text = ""
file_command = False
if isinstance(text, str):
stripped = text.lstrip()
if stripped.startswith("/"):
token = stripped.split(maxsplit=1)[0]
file_command = token.startswith("/file")
stripped = text.lstrip()
if stripped.startswith("/"):
token = stripped.split(maxsplit=1)[0]
file_command = token.startswith("/file")
voice_payload: TelegramVoice | None = None
voice = msg.get("voice")
if isinstance(voice, dict):
file_id = voice.get("file_id")
if not isinstance(file_id, str) or not file_id:
file_id = None
if file_id is not None:
voice_payload = TelegramVoice(
file_id=file_id,
mime_type=voice.get("mime_type")
if isinstance(voice.get("mime_type"), str)
else None,
file_size=voice.get("file_size")
if isinstance(voice.get("file_size"), int)
and not isinstance(voice.get("file_size"), bool)
else None,
duration=voice.get("duration")
if isinstance(voice.get("duration"), int)
and not isinstance(voice.get("duration"), bool)
else None,
raw=voice,
)
if not isinstance(raw_text, str) and not isinstance(caption, str):
text = ""
if msg.voice is not None:
voice_payload = TelegramVoice(
file_id=msg.voice.file_id,
mime_type=msg.voice.mime_type,
file_size=msg.voice.file_size,
duration=msg.voice.duration,
raw=msgspec.to_builtins(msg.voice),
)
if raw_text is None and caption is None:
text = ""
document_payload: TelegramDocument | None = None
document = msg.get("document")
if isinstance(document, dict):
document_payload = _parse_document_payload(document)
if msg.document is not None:
document_payload = _document_from_media(msg.document)
if document_payload is None and msg.video is not None:
document_payload = _document_from_media(msg.video)
if document_payload is None:
video = msg.get("video")
if isinstance(video, dict):
document_payload = _parse_document_payload(video)
if document_payload is None:
photo = msg.get("photo")
if isinstance(photo, list):
best: dict[str, Any] | None = None
best_score = -1
for item in photo:
if not isinstance(item, dict):
continue
file_id = item.get("file_id")
if not isinstance(file_id, str) or not file_id:
continue
size = item.get("file_size")
if isinstance(size, int) and not isinstance(size, bool):
score = size
else:
width = item.get("width")
height = item.get("height")
if isinstance(width, int) and isinstance(height, int):
score = width * height
else:
score = 0
if score > best_score:
best_score = score
best = item
if best is not None:
document_payload = _parse_document_payload(best)
if document_payload is None and file_command:
sticker = msg.get("sticker")
if isinstance(sticker, dict):
document_payload = _parse_document_payload(sticker)
has_text = isinstance(raw_text, str) or isinstance(caption, str)
best = _best_photo(msg.photo)
if best is not None:
document_payload = _document_from_photo(best)
if document_payload is None and file_command and msg.sticker is not None:
document_payload = _document_from_sticker(msg.sticker)
has_text = raw_text is not None or caption is not None
if not has_text and voice_payload is None and document_payload is None:
return None
chat = msg.get("chat")
if not isinstance(chat, dict):
return None
msg_chat_id = chat.get("id")
if not isinstance(msg_chat_id, int):
return None
chat_type = chat.get("type") if isinstance(chat.get("type"), str) else None
is_forum = chat.get("is_forum")
if not isinstance(is_forum, bool):
is_forum = None
msg_chat_id = msg.chat.id
chat_type = msg.chat.type
is_forum = msg.chat.is_forum
allowed = chat_ids
if allowed is None and chat_id is not None:
allowed = {chat_id}
if allowed is not None and msg_chat_id not in allowed:
return None
message_id = msg.get("message_id")
if not isinstance(message_id, int):
return None
reply = msg.get("reply_to_message")
reply_to_message_id = None
reply_to_text = None
reply_to_is_bot = None
reply_to_username = None
if isinstance(reply, dict):
reply_to_message_id = (
reply.get("message_id")
if isinstance(reply.get("message_id"), int)
else None
)
reply_to_text = (
reply.get("text") if isinstance(reply.get("text"), str) else None
)
reply_from = reply.get("from")
if isinstance(reply_from, dict):
is_bot = reply_from.get("is_bot")
if isinstance(is_bot, bool):
reply_to_is_bot = is_bot
username = reply_from.get("username")
if isinstance(username, str):
reply_to_username = username
sender = msg.get("from")
sender_id = (
sender.get("id")
if isinstance(sender, dict) and isinstance(sender.get("id"), int)
else None
reply = msg.reply_to_message
reply_to_message_id = reply.message_id if reply is not None else None
reply_to_text = reply.text if reply is not None else None
reply_to_is_bot = (
reply.from_.is_bot if reply is not None and reply.from_ is not None else None
)
media_group_id = msg.get("media_group_id")
if not isinstance(media_group_id, str):
media_group_id = None
thread_id = msg.get("message_thread_id")
if isinstance(thread_id, bool) or not isinstance(thread_id, int):
thread_id = None
is_topic_message = msg.get("is_topic_message")
if not isinstance(is_topic_message, bool):
is_topic_message = None
reply_to_username = (
reply.from_.username if reply is not None and reply.from_ is not None else None
)
sender_id = msg.from_.id if msg.from_ is not None else None
media_group_id = msg.media_group_id
thread_id = msg.message_thread_id
is_topic_message = msg.is_topic_message
return TelegramIncomingMessage(
transport="telegram",
chat_id=msg_chat_id,
message_id=message_id,
message_id=msg.message_id,
text=text,
reply_to_message_id=reply_to_message_id,
reply_to_text=reply_to_text,
@@ -218,51 +127,80 @@ def _parse_incoming_message(
is_forum=is_forum,
voice=voice_payload,
document=document_payload,
raw=msg,
raw=msgspec.to_builtins(msg),
)
def _parse_callback_query(
query: dict[str, Any],
query: CallbackQuery,
*,
chat_id: int | None = None,
chat_ids: set[int] | None = None,
) -> TelegramCallbackQuery | None:
callback_id = query.get("id")
if not isinstance(callback_id, str) or not callback_id:
return None
msg = query.get("message")
if not isinstance(msg, dict):
return None
chat = msg.get("chat")
if not isinstance(chat, dict):
return None
msg_chat_id = chat.get("id")
if not isinstance(msg_chat_id, int):
callback_id = query.id
msg = query.message
if msg is None:
return None
msg_chat_id = msg.chat.id
allowed = chat_ids
if allowed is None and chat_id is not None:
allowed = {chat_id}
if allowed is not None and msg_chat_id not in allowed:
return None
message_id = msg.get("message_id")
if not isinstance(message_id, int):
return None
data = query.get("data") if isinstance(query.get("data"), str) else None
sender = query.get("from")
sender_id = (
sender.get("id")
if isinstance(sender, dict) and isinstance(sender.get("id"), int)
else None
)
data = query.data
sender_id = query.from_.id if query.from_ is not None else None
return TelegramCallbackQuery(
transport="telegram",
chat_id=msg_chat_id,
message_id=message_id,
message_id=msg.message_id,
callback_query_id=callback_id,
data=data,
sender_id=sender_id,
raw=query,
raw=msgspec.to_builtins(query),
)
def _best_photo(photos: list[PhotoSize] | None) -> PhotoSize | None:
if not photos:
return None
best = None
best_score = -1
for item in photos:
size = item.file_size
score = size if size is not None else item.width * item.height
if score > best_score:
best_score = score
best = item
return best
def _document_from_media(media: Document | Video) -> TelegramDocument:
return TelegramDocument(
file_id=media.file_id,
file_name=media.file_name,
mime_type=media.mime_type,
file_size=media.file_size,
raw=msgspec.to_builtins(media),
)
def _document_from_photo(photo: PhotoSize) -> TelegramDocument:
return TelegramDocument(
file_id=photo.file_id,
file_name=None,
mime_type=None,
file_size=photo.file_size,
raw=msgspec.to_builtins(photo),
)
def _document_from_sticker(sticker: Sticker) -> TelegramDocument:
return TelegramDocument(
file_id=sticker.file_id,
file_name=None,
mime_type=None,
file_size=sticker.file_size,
raw=msgspec.to_builtins(sticker),
)
tests/telegram_fakes.py
+3 -3
View File
@@ -141,7 +141,7 @@ class FakeBot(BotClient):
"replace_message_id": replace_message_id,
}
)
return Message(message_id=1)
return Message(message_id=1, chat=Chat(id=chat_id, type="private"))
async def send_document(
self,
@@ -164,7 +164,7 @@ class FakeBot(BotClient):
"caption": caption,
}
)
return Message(message_id=2)
return Message(message_id=2, chat=Chat(id=chat_id, type="private"))
async def edit_message_text(
self,
@@ -188,7 +188,7 @@ class FakeBot(BotClient):
"wait": wait,
}
)
return Message(message_id=message_id)
return Message(message_id=message_id, chat=Chat(id=chat_id, type="private"))
async def delete_message(self, chat_id: int, message_id: int) -> bool:
self.delete_calls.append({"chat_id": chat_id, "message_id": message_id})
tests/test_onboarding_helpers.py
+31 -9
View File
@@ -9,7 +9,7 @@ from rich.text import Text
from takopi.config import ConfigError
from takopi.telegram import onboarding
from takopi.telegram.api_models import Update, User
from takopi.telegram.api_models import Chat, Message, Update, User
from takopi.telegram.client import TelegramRetryAfter
@@ -311,35 +311,57 @@ async def test_get_bot_info_gives_up(monkeypatch) -> None:
@pytest.mark.anyio
async def test_wait_for_chat_filters_updates(monkeypatch) -> None:
updates = [
[Update(update_id=1, message={"from": {"is_bot": True}, "chat": {"id": 1}})],
[
Update(
update_id=1,
message=Message(
message_id=1,
from_=User(id=1, is_bot=True),
chat=Chat(id=1, type="private"),
),
)
],
None,
[],
[Update(update_id=2, message=None)],
[
Update(
update_id=3,
message={"from": {"is_bot": True}, "chat": {"id": 2}},
message=Message(
message_id=3,
from_=User(id=2, is_bot=True),
chat=Chat(id=2, type="private"),
),
)
],
[
Update(
update_id=4,
message={"from": {"is_bot": False}, "chat": "nope"},
message=Message(
message_id=4,
from_=User(id=3, is_bot=True),
chat=Chat(id=3, type="private"),
),
)
],
[
Update(
update_id=5,
message={"from": {"is_bot": False}, "chat": {"id": "bad"}},
message=Message(
message_id=5,
from_=User(id=4, is_bot=True),
chat=Chat(id=4, type="private"),
),
)
],
[
Update(
update_id=6,
message={
"from": {"is_bot": False},
"chat": {"id": 7, "username": "bob", "type": "private"},
},
message=Message(
message_id=6,
from_=User(id=5, is_bot=False),
chat=Chat(id=7, username="bob", type="private"),
),
)
],
]
tests/test_telegram_bridge.py
+2 -2
View File
@@ -14,7 +14,7 @@ from takopi.telegram.commands.topics import _handle_topic_command
import takopi.telegram.loop as telegram_loop
import takopi.telegram.topics as telegram_topics
from takopi.directives import parse_directives
from takopi.telegram.api_models import File, ForumTopic, Message, Update, User
from takopi.telegram.api_models import Chat, File, ForumTopic, Message, Update, User
from takopi.settings import TelegramFilesSettings, TelegramTopicsSettings
from takopi.telegram.bridge import (
TelegramBridgeConfig,
@@ -462,7 +462,7 @@ async def test_telegram_transport_edit_wait_false_returns_ref() -> None:
)
if not wait:
return None
return Message(message_id=message_id)
return Message(message_id=message_id, chat=Chat(id=chat_id, type="private"))
async def delete_message(
self,
tests/test_telegram_client_api.py
+3 -3
View File
@@ -71,9 +71,9 @@ async def test_client_methods_build_params_and_decode() -> None:
payloads = {
"getUpdates": [{"update_id": 1}],
"getFile": {"file_path": "path"},
"sendMessage": {"message_id": 1},
"sendDocument": {"message_id": 2},
"editMessageText": {"message_id": 3},
"sendMessage": {"message_id": 1, "chat": {"id": 1, "type": "private"}},
"sendDocument": {"message_id": 2, "chat": {"id": 1, "type": "private"}},
"editMessageText": {"message_id": 3, "chat": {"id": 1, "type": "private"}},
"deleteMessage": True,
"setMyCommands": True,
"getMe": {"id": 7},
tests/test_telegram_incoming.py
+156 -149
View File
@@ -3,23 +3,37 @@ from takopi.telegram import (
TelegramIncomingMessage,
parse_incoming_update,
)
from takopi.telegram.api_models import (
CallbackQuery,
CallbackQueryMessage,
Chat,
Document,
Message,
MessageReply,
PhotoSize,
Sticker,
Update,
User,
Video,
Voice,
)
def test_parse_incoming_update_maps_fields() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"text": "hello",
"chat": {"id": 123, "type": "supergroup", "is_forum": True},
"from": {"id": 99},
"reply_to_message": {
"message_id": 5,
"text": "prev",
"from": {"id": 77, "is_bot": True, "username": "ReplyBot"},
},
},
}
update = Update(
update_id=1,
message=Message(
message_id=10,
text="hello",
chat=Chat(id=123, type="supergroup", is_forum=True),
from_=User(id=99),
reply_to_message=MessageReply(
message_id=5,
text="prev",
from_=User(id=77, is_bot=True, username="ReplyBot"),
),
),
)
msg = parse_incoming_update(update, chat_id=123)
assert msg is not None
@@ -39,50 +53,49 @@ def test_parse_incoming_update_maps_fields() -> None:
assert msg.is_forum is True
assert msg.voice is None
assert msg.document is None
assert msg.raw == update["message"]
assert msg.raw
assert msg.raw["message_id"] == 10
def test_parse_incoming_update_filters_non_matching_chat() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"text": "hello",
"chat": {"id": 123},
},
}
update = Update(
update_id=1,
message=Message(
message_id=10,
text="hello",
chat=Chat(id=123, type="private"),
),
)
assert parse_incoming_update(update, chat_id=999) is None
def test_parse_incoming_update_filters_non_text_and_non_voice() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"chat": {"id": 123},
"location": {"latitude": 1.0, "longitude": 2.0},
},
}
update = Update(
update_id=1,
message=Message(
message_id=10,
chat=Chat(id=123, type="private"),
),
)
assert parse_incoming_update(update, chat_id=123) is None
def test_parse_incoming_update_voice_message() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"chat": {"id": 123},
"voice": {
"file_id": "voice-id",
"file_unique_id": "uniq",
"duration": 3,
"mime_type": "audio/ogg",
"file_size": 1234,
},
},
}
update = Update(
update_id=1,
message=Message(
message_id=10,
chat=Chat(id=123, type="private"),
voice=Voice(
file_id="voice-id",
duration=3,
mime_type="audio/ogg",
file_size=1234,
),
),
)
msg = parse_incoming_update(update, chat_id=123)
assert msg is not None
@@ -96,21 +109,20 @@ def test_parse_incoming_update_voice_message() -> None:
def test_parse_incoming_update_document_message() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"caption": "/file put incoming/doc.txt",
"chat": {"id": 123},
"document": {
"file_id": "doc-id",
"file_unique_id": "uniq",
"file_name": "doc.txt",
"mime_type": "text/plain",
"file_size": 4321,
},
},
}
update = Update(
update_id=1,
message=Message(
message_id=10,
caption="/file put incoming/doc.txt",
chat=Chat(id=123, type="private"),
document=Document(
file_id="doc-id",
file_name="doc.txt",
mime_type="text/plain",
file_size=4321,
),
),
)
msg = parse_incoming_update(update, chat_id=123)
assert msg is not None
@@ -124,30 +136,28 @@ def test_parse_incoming_update_document_message() -> None:
def test_parse_incoming_update_photo_message() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"caption": "/file put incoming/photo.jpg",
"chat": {"id": 123},
"photo": [
{
"file_id": "small",
"file_unique_id": "uniq-small",
"file_size": 100,
"width": 90,
"height": 90,
},
{
"file_id": "large",
"file_unique_id": "uniq-large",
"file_size": 1000,
"width": 800,
"height": 600,
},
update = Update(
update_id=1,
message=Message(
message_id=10,
caption="/file put incoming/photo.jpg",
chat=Chat(id=123, type="private"),
photo=[
PhotoSize(
file_id="small",
file_size=100,
width=90,
height=90,
),
PhotoSize(
file_id="large",
file_size=1000,
width=800,
height=600,
),
],
},
}
),
)
msg = parse_incoming_update(update, chat_id=123)
assert msg is not None
@@ -160,23 +170,22 @@ def test_parse_incoming_update_photo_message() -> None:
def test_parse_incoming_update_media_group_id() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"chat": {"id": 123},
"media_group_id": "group-1",
"photo": [
{
"file_id": "large",
"file_unique_id": "uniq-large",
"file_size": 1000,
"width": 800,
"height": 600,
}
update = Update(
update_id=1,
message=Message(
message_id=10,
chat=Chat(id=123, type="private"),
media_group_id="group-1",
photo=[
PhotoSize(
file_id="large",
file_size=1000,
width=800,
height=600,
)
],
},
}
),
)
msg = parse_incoming_update(update, chat_id=123)
assert msg is not None
@@ -185,21 +194,20 @@ def test_parse_incoming_update_media_group_id() -> None:
def test_parse_incoming_update_video_message() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"caption": "/file put incoming/video.mp4",
"chat": {"id": 123},
"video": {
"file_id": "video-id",
"file_unique_id": "uniq",
"file_name": "video.mp4",
"mime_type": "video/mp4",
"file_size": 4242,
},
},
}
update = Update(
update_id=1,
message=Message(
message_id=10,
caption="/file put incoming/video.mp4",
chat=Chat(id=123, type="private"),
video=Video(
file_id="video-id",
file_name="video.mp4",
mime_type="video/mp4",
file_size=4242,
),
),
)
msg = parse_incoming_update(update, chat_id=123)
assert msg is not None
@@ -213,19 +221,18 @@ def test_parse_incoming_update_video_message() -> None:
def test_parse_incoming_update_sticker_message() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"caption": "/file put incoming/sticker.webp",
"chat": {"id": 123},
"sticker": {
"file_id": "sticker-id",
"file_unique_id": "uniq",
"file_size": 2468,
},
},
}
update = Update(
update_id=1,
message=Message(
message_id=10,
caption="/file put incoming/sticker.webp",
chat=Chat(id=123, type="private"),
sticker=Sticker(
file_id="sticker-id",
file_size=2468,
),
),
)
msg = parse_incoming_update(update, chat_id=123)
assert msg is not None
@@ -239,18 +246,18 @@ def test_parse_incoming_update_sticker_message() -> None:
def test_parse_incoming_update_callback_query() -> None:
update = {
"update_id": 1,
"callback_query": {
"id": "cbq-1",
"data": "takopi:cancel",
"from": {"id": 321},
"message": {
"message_id": 55,
"chat": {"id": 123},
},
},
}
update = Update(
update_id=1,
callback_query=CallbackQuery(
id="cbq-1",
data="takopi:cancel",
from_=User(id=321),
message=CallbackQueryMessage(
message_id=55,
chat=Chat(id=123, type="private"),
),
),
)
msg = parse_incoming_update(update, chat_id=123)
assert isinstance(msg, TelegramCallbackQuery)
@@ -263,16 +270,16 @@ def test_parse_incoming_update_callback_query() -> None:
def test_parse_incoming_update_topic_fields() -> None:
update = {
"update_id": 1,
"message": {
"message_id": 10,
"text": "hello",
"message_thread_id": 77,
"is_topic_message": True,
"chat": {"id": -100, "type": "supergroup", "is_forum": True},
},
}
update = Update(
update_id=1,
message=Message(
message_id=10,
text="hello",
message_thread_id=77,
is_topic_message=True,
chat=Chat(id=-100, type="supergroup", is_forum=True),
),
)
msg = parse_incoming_update(update, chat_id=-100)
assert isinstance(msg, TelegramIncomingMessage)
tests/test_telegram_queue.py
+4 -4
View File
@@ -3,7 +3,7 @@ from typing import Any
import anyio
import pytest
from takopi.telegram.api_models import File, Message, Update, User
from takopi.telegram.api_models import Chat, File, Message, Update, User
from takopi.telegram.client import BotClient, TelegramClient, TelegramRetryAfter
@@ -39,7 +39,7 @@ class FakeBot(BotClient):
_ = reply_markup
_ = replace_message_id
self.calls.append("send_message")
return Message(message_id=1)
return Message(message_id=1, chat=Chat(id=chat_id, type="private"))
async def send_document(
self,
@@ -61,7 +61,7 @@ class FakeBot(BotClient):
caption,
)
self.calls.append("send_document")
return Message(message_id=1)
return Message(message_id=1, chat=Chat(id=chat_id, type="private"))
async def edit_message_text(
self,
@@ -86,7 +86,7 @@ class FakeBot(BotClient):
self._edit_attempts += 1
raise TelegramRetryAfter(self.retry_after)
self._edit_attempts += 1
return Message(message_id=message_id)
return Message(message_id=message_id, chat=Chat(id=chat_id, type="private"))
async def delete_message(
self,