"""Runtime-прогон всех core/media MCP-тулов против фейкового Telethon-клиента.
Дополнение к test_login_flow (там только логин): здесь проверяется, что каждый
тул строит валидные вызовы Telethon (TL-запросы/kwargs из реальной версии),
сериализаторы справляются с настоящими TL-объектами, а ошибки возвращаются
данными ({"error": …}), а не падают. По мотивам Production-бага: в `_tool`
стоял не существующий класс errors.EntityNotFoundError (AttributeError в
except-клаузе), а messages_history шёл offset_id=None — TL-сериализация не
принимает None для int.
"""
import asyncio
import base64
import sys
import types
from datetime import datetime, timezone
from pathlib import Path
import pytest
import telethon.errors
from telethon.tl.types import (
Chat, ChatPhotoEmpty, Dialog, Document, DocumentAttributeAudio,
DocumentAttributeFilename, InputPeerChat, InputPeerUser, Message,
MessageMediaDocument, MessageMediaPhoto, PeerChat, PeerUser,
PhoneCallProtocol, PhoneCallWaiting, Photo, PhotoSize, UpdatePhoneCall, User,
)
BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))
from app.config import get_settings
from app.db import get_db, init_db
from app.mcp import tools
from app.tg.calls import _calls
NOW = datetime.now(timezone.utc)
class DummyMCP:
"""Ловит тулерегистрацию register_tools: тулы по имени, без FastMCP."""
def __init__(self):
self.registered = {}
def tool(self):
def deco(fn):
self.registered[fn.__name__] = fn
return fn
return deco
_TOOLS_CACHE = {}
def get_tools(module=None):
"""register_tools → dict name → функция (кэш на модуль tools)."""
dm = DummyMCP()
_T = dm
module.register_tools(_T)
return _T.registered
class FakeManager:
def __init__(self, clients):
self.clients = clients # account_id → client
self.saved = []
async def get_client(self, account_id):
return self.clients[account_id]
async def save_session(self, account_id):
self.saved.append(account_id)
def pending_client(self, login_id):
return None
class FakeClient:
"""Подмножество Telethon-клиента, которого хватает тулам."""
def __init__(self, *, me=None, dialogs=None, by_id=None, contacts=None,
entities=None):
self.me = me or User(id=42, first_name="Тест", last_name="Юзер",
username="tester", phone="+79991234567")
self.dialogs = dialogs or []
self.by_id = by_id or {} # raw peer id (InputPeerChat.chat_id) → list[Message]
self.contacts = contacts or []
# маркированный dialog_id → InputPeer (как в пуле manager)
self.input_entities = entities or {777: InputPeerChat(chat_id=777)}
async def get_me(self):
return self.me
async def get_input_entity(self, peer):
ent = self.input_entities.get(peer)
if ent is None:
raise ValueError(f"Could not find the input entity for {peer}")
return ent
async def get_entity(self, entity):
return entity
def iter_dialogs(self, *, limit=100):
async def it():
for d in self.dialogs[:limit]:
yield d
return it()
@staticmethod
def _peer_key(entity):
return getattr(entity, "chat_id", getattr(entity, "user_id", 777))
async def get_messages(self, entity, *, limit=100, offset_id=0, min_id=0,
ids=None, search=None, from_user=None):
pool = self.by_id.get(self._peer_key(entity), [])
if ids:
found = {m.id: m for m in pool}
return [found.get(i) for i in ids]
if search is not None:
pool = [m for m in pool if search.lower() in (m.message or "").lower()]
return list(reversed(pool))[:limit] # новые первыми
async def send_message(self, entity, text, **kwargs):
return Message(id=max(m.id for m in self.by_id.get(777, [Message(id=0, peer_id=None, date=NOW)])) + 1,
peer_id=PeerChat(chat_id=777), date=NOW, message=text, out=True)
async def send_file(self, entity, *a, **kw):
mid = max(m.id for m in self.by_id.get(777, [Message(id=0, peer_id=None, date=NOW)])) + 1
doc = Document(id=1, access_hash=2, file_reference=b"x", date=NOW,
mime_type=kw.get("mime_type", "application/octet-stream"),
size=64, dc_id=1, attributes=kw.get("attributes", []))
return Message(id=mid, peer_id=PeerChat(chat_id=777), date=NOW, message="",
media=MessageMediaDocument(document=doc), out=True)
async def edit_message(self, entity, mid, text):
return Message(id=mid, peer_id=PeerChat(chat_id=777), date=NOW, message=text, out=True)
async def delete_messages(self, entity, ids, *, revoke=True):
self.by_id[777] = [m for m in self.by_id.get(777, []) if m.id not in ids]
return True
async def download_media(self, message, target):
return b"mediabytes"
def get_participants(self, entity, *, limit=100):
async def none_():
return []
return none_()
async def __call__(self, request):
kind = type(request).__name__
if kind == "GetFullChatRequest":
return types.SimpleNamespace(full_chat=None, users=self.contacts)
if kind == "GetContactsRequest":
return types.SimpleNamespace(users=self.contacts)
if kind == "SearchGlobalRequest":
request._bytes() # сериализация не должна падать (None-args регресс)
msgs = [m for pool in self.by_id.values() for m in pool]
return types.SimpleNamespace(messages=msgs)
if kind == "GetDhConfigRequest":
from telethon.tl.types import messages as tl_messages
return tl_messages.DhConfig(g=3, p=b"\xfd" * 256, version=1, random=[b"\x00"])
if kind == "RequestCallRequest":
proto = PhoneCallProtocol(min_layer=92, max_layer=127, udp_p2p=True,
udp_reflector=True, library_versions=["x"])
waiting = PhoneCallWaiting(id=555, access_hash=9, date=NOW,
admin_id=42, participant_id=7, protocol=proto)
return types.SimpleNamespace(updates=[UpdatePhoneCall(phone_call=waiting)])
if kind == "DiscardCallRequest":
request._bytes()
return types.SimpleNamespace()
if kind == "ConfirmCallRequest":
return types.SimpleNamespace()
raise AssertionError(f"fake client: unexpected request {kind}")
async def setup(tmp_path, monkeypatch, *, user="mcp"):
monkeypatch.setenv("TGCLIENT_DB_PATH", str(tmp_path / "t.db"))
monkeypatch.setenv("TGCLIENT_API_ID", "1")
monkeypatch.setenv("TGCLIENT_API_HASH", "hash")
monkeypatch.setenv("TGCLIENT_AUTH_CLIENT_ID", "")
get_settings.cache_clear()
await init_db()
conn = get_db()
await conn.execute(
"INSERT OR IGNORE INTO users (user_id, email, created_at, updated_at)"
" VALUES (?, 'test@example.com', ?, ?)", (user, NOW.isoformat(), NOW.isoformat()))
await conn.commit()
return conn
async def insert_account(conn, account_id=1, user="mcp", status="active"):
await conn.execute(
"INSERT OR REPLACE INTO accounts"
" (id, user_id, phone, label, tg_user_id, username, display_name,"
" session_data, status, created_at, updated_at, last_used_at)"
" VALUES (?, ?, ?, '', 42, 'tester', 'Тест', 'sess', ?, ?, ?, ?)",
(account_id, user, "+79991234567", status, NOW.isoformat(), NOW.isoformat(), None))
await conn.commit()
def make_manager(monkeypatch, client):
manager = FakeManager({1: client})
import app.main as appmain
import app.tg.manager as mgr
monkeypatch.setattr(appmain, "get_account_manager", lambda: manager, raising=False)
monkeypatch.setattr(mgr, "report", lambda *a, **k: None)
return manager
def voice_message(mid=10):
doc = Document(id=1, access_hash=2, file_reference=b"x", date=NOW,
mime_type="audio/ogg", size=1234, dc_id=1,
attributes=[DocumentAttributeAudio(duration=5, voice=True, waveform=b"\x82\x82"),
DocumentAttributeFilename("voice.ogg")])
return Message(id=mid, peer_id=PeerChat(chat_id=777), date=NOW, message="голосовое",
media=MessageMediaDocument(document=doc), out=True)
def photo_message(mid=11, raw=778):
photo = Photo(id=1, access_hash=2, file_reference=b"", date=NOW,
sizes=[PhotoSize(type="x", w=10, h=10, size=555)], dc_id=1)
return Message(id=mid, peer_id=PeerChat(chat_id=raw), date=NOW, message="",
media=MessageMediaPhoto(photo=photo))
def text_message(mid=12, text="привет"):
return Message(id=mid, peer_id=PeerChat(chat_id=777), date=NOW, message=text, out=True)
pytestmark = pytest.mark.asyncio
@pytest.mark.asyncio
async def test_core_tools_smoke(monkeypatch, tmp_path):
conn = await setup(tmp_path, monkeypatch)
await insert_account(conn)
g = Chat(id=777, title="Группа", photo=ChatPhotoEmpty(), participants_count=2,
date=NOW, version=0)
# iter_dialogs отдаёт custom Dialog (не TL) — как в реальном Telethon
raw_dlg = Dialog(peer=PeerChat(chat_id=777), top_message=12, read_inbox_max_id=12,
read_outbox_max_id=0, unread_count=1, unread_mentions_count=0,
unread_reactions_count=0, unread_poll_votes_count=0, draft=None,
notify_settings=None, folder_id=None)
from telethon.tl.custom.dialog import Dialog as CustomDialog
last = text_message(12)
dlg = CustomDialog(client=None, dialog=raw_dlg, entities={-777: g}, message=last)
client = FakeClient(
dialogs=[dlg], by_id={777: [voice_message(10), photo_message(11), text_message(12)]},
contacts=[User(id=9, first_name="Друг")],
entities={777: InputPeerChat(chat_id=777), -777: InputPeerChat(chat_id=777),
42: InputPeerUser(user_id=42, access_hash=1)})
T = get_tools(tools)
make_manager(monkeypatch, client)
out = await T["accounts_list"]()
assert out["accounts"][0]["status"] == "active" and out["accounts"][0]["id"] == 1
me = await T["me_get"](1)
assert me["account_id"] == 1 and me["is_bot"] is False
dl_out = await T["dialogs_list"](1, 10)
d = dl_out["dialogs"][0]
assert d["id"] == -777 and d["unread_count"] == 1 and d["last_message"]["id"] == 12
hist = await T["messages_history"](1, -777, 10, 0, 0)
assert [m["id"] for m in hist["messages"]] == [12, 11, 10]
read = await T["messages_read"](1, -777, 10)
assert read["media"]["voice"] is True and read["media"]["waveform"]
ph = await T["messages_read"](1, -777, 11)
assert ph["media"]["size"] == 555 and "voice" not in ph["media"]
sr = await T["messages_search"](1, -777, "привет", "", 10)
assert [m["id"] for m in sr["messages"]] == [12]
glob = await T["messages_search"](1, 0, "голосовое", "", 10)
assert glob["messages"]
ci = await T["chat_info"](1, -777)
assert ci["id"] == -777 and ci["type"] == "chat"
ct = await T["contacts_list"](1, 10)
assert ct["contacts"][0]["id"] == 9
# не-свой dialog_id → 404 данными
nf = await T["messages_history"](1, -888, 10, 0, 0)
assert nf["error"] == 404
@pytest.mark.asyncio
async def test_mutation_tools_smoke(monkeypatch, tmp_path):
conn = await setup(tmp_path, monkeypatch)
await insert_account(conn)
client = FakeClient(
by_id={777: [voice_message(10), text_message(12)]},
entities={-777: InputPeerChat(chat_id=777), 42: InputPeerUser(user_id=42, access_hash=1)})
T = get_tools(tools)
make_manager(monkeypatch, client)
sent = await T["message_send"](1, -777, "текст", 0, False)
assert sent["text"] == "текст" and sent["out"] is True
rep = await T["message_reply"](1, -777, 12, "ответ")
assert rep["text"] == "ответ"
ed = await T["message_edit"](1, -777, 12, "правка")
assert ed["text"] == "правка"
de = await T["message_delete"](1, -777, [10], True)
assert de == {"status": "ok", "deleted": [10], "revoke": True}
# пустой text / пустые ids → 422 данными
assert (await T["message_send"](1, -777, " ", 0, False))["error"] == 422
assert (await T["message_delete"](1, -777, [], True))["error"] == 422
# чужой dialog → 404 (не найден на этом аккаунте)
nf = await T["message_send"](1, -999, "x", 0, False)
assert nf["error"] == 404
@pytest.mark.asyncio
async def test_media_tools_smoke(monkeypatch, tmp_path):
conn = await setup(tmp_path, monkeypatch)
await insert_account(conn)
client = FakeClient(
by_id={777: [voice_message(10), text_message(12)], 778: [photo_message(11)]},
entities={-777: InputPeerChat(chat_id=777), -778: InputPeerChat(chat_id=778)})
T = get_tools(tools)
make_manager(monkeypatch, client)
dm = await T["download_media"](1, -777, 10, 0)
assert dm["is_voice"] is True and dm["file_name"] == "voice.ogg"
assert dm["data_base64"] == base64.b64encode(b"mediabytes").decode("ascii")
ph_dl = await T["download_media"](1, -778, 11, 0)
assert ph_dl["is_round"] is False and ph_dl["data_base64"]
# файл больше лимита → 413
small = await T["download_media"](1, -778, 11, 8)
assert small["error"] == 413
# сообщение без медиа → 404
nomed = await T["download_media"](1, -777, 12, 0)
assert nomed["error"] == 404
up = await T["upload_file"](1, -777, base64.b64encode(b"doc").decode(), "",
"doc.txt", "подпись", False)
assert "media" in up
voc = await T["upload_voice"](1, -777, base64.b64encode(b"ogg").decode(), "", 5, None)
assert voc["media"]["voice"] is True
rnd = await T["upload_round"](1, -777, base64.b64encode(b"mp4").decode(), "", 10, 512)
assert rnd["media"]["round"] is True
toolong = await T["upload_round"](1, -777, base64.b64encode(b"mp4").decode(), "", 61, 0)
assert toolong["error"] == 422
@pytest.mark.asyncio
async def test_call_tools_smoke(monkeypatch, tmp_path):
conn = await setup(tmp_path, monkeypatch)
await insert_account(conn)
_calls.clear()
client = FakeClient(
entities={42: InputPeerUser(user_id=42, access_hash=1),
-777: InputPeerChat(chat_id=777)})
T = get_tools(tools)
make_manager(monkeypatch, client)
out = await T["call_start"](1, 42, False)
assert out["call_id"] == 555 and out["state"] == "ringing"
st = await T["call_status"](555)
assert st["state"] == "ringing"
disc = await T["call_discard"](555)
assert disc["state"] == "cancelled"
_calls.clear()
# юзер не найден → 404 данными
nf = await T["call_start"](1, 999, False)
assert nf["error"] == 404
# группе звонить нельзя → 422
grp = await T["call_start"](1, -777, False)
assert grp["error"] == 422
# неизвестный статус → 404
unk = await T["call_status"](1)
assert unk["error"] == 404
@pytest.mark.asyncio
async def test_entity_not_found_boundary(monkeypatch, tmp_path):
"""«Диалог не найден» от Telethon (ValueError из кеша) → 404-данные."""
conn = await setup(tmp_path, monkeypatch)
await insert_account(conn)
client = FakeClient(entities={})
T = get_tools(tools)
make_manager(monkeypatch, client)
async def get_input_entity_raises(x):
raise ValueError("Could not find the input entity for PeerChat")
client.get_input_entity = get_input_entity_raises
out = await T["messages_history"](1, -888, 10, 0, 0)
assert out["error"] == 404
