Blame for tests/test_wire.py

"""Check that every Telegram call the bot makes builds a valid, serialisable MTProto request.

Telethon's network layer is replaced by a function that captures the request (after Telethon has
resolved it) and serialises it, so wrong argument types or shapes fail here instead of in production.
"""
import asyncio
import datetime

import pytest
from telethon import TelegramClient, functions, types, utils
from telethon.sessions import MemorySession

from app.common import push_edit, send_post
from app.tgutil import (
    build_markup,
    de_entities,
    edit_raw,
    media_from_ref,
    media_ref,
    peer_of,
    ser_entities,
)
from app.ui import button_keyboard, panel_keyboard

from .fakes import make_msg

NOW = datetime.datetime.now(datetime.timezone.utc)
CH = type("Ch", (), {"id": 1234, "access_hash": 5678})()


class Captured(Exception):
    def __init__(self, req):
        self.req = req


@pytest.fixture()
def client(monkeypatch):
    async def fake_call(self, request, ordered=False, flood_sleep_threshold=None):
        reqs = list(request) if isinstance(request, (list, tuple)) else [request]
        for r in reqs:
            if hasattr(r, "resolve"):
                await r.resolve(self, utils)
            assert isinstance(bytes(r), bytes)  # must serialise
        raise Captured(reqs[0])

    monkeypatch.setattr(TelegramClient, "__call__", fake_call)
    return TelegramClient(MemorySession(), 1, "h")


def capture(coro):
    async def go():
        try:
            await coro
        except Captured as c:
            return c.req
        raise AssertionError("no request was made")

    return asyncio.run(go())


def post(**kw):
    base = dict(
        id=1, message_id=50, text="Hello 😀", entities=[{"k": "bold", "o": 0, "l": 5}, {"k": "url", "o": 6, "l": 2, "u": "https://x.com"}],
        media_kind=None, media_file_id=None, link_preview=False,
        buttons=[[{"t": "Go", "u": "https://t.me/bot?start=a"}, {"t": "Two", "u": "https://t.me/bot?start=b"}]],
    )
    base.update(kw)
    return type("P", (), base)()


def photo_ref():
    photo = types.Photo(11, 22, b"ref", NOW, [types.PhotoSize("x", 100, 100, 999)], 2)
    return media_ref(make_msg(1, "", media=types.MessageMediaPhoto(photo=photo)))


def test_send_text_post(client):
    req = capture(send_post(client, peer_of(CH), post()))
    assert isinstance(req, functions.messages.SendMessageRequest)
    assert req.message == "Hello 😀" and req.no_webpage is True
    assert [type(e).__name__ for e in req.entities] == ["MessageEntityBold", "MessageEntityTextUrl"]
    assert req.reply_markup.rows[0].buttons[1].text == "Two"
    assert isinstance(req.peer, types.InputPeerChannel)


def test_send_media_post_with_and_without_file_reference(client):
    ref = photo_ref()
    req = capture(send_post(client, peer_of(CH), post(media_kind="photo", media_file_id=ref)))
    assert isinstance(req, functions.messages.SendMediaRequest)
    assert isinstance(req.media, types.InputMediaPhoto) and req.media.id.id == 11 and req.media.id.file_reference == b"ref"
    assert req.message == "Hello 😀" and req.reply_markup is not None

    async def blank():
        c = client
        _, media, _ = await c._file_to_media(media_from_ref(ref, blank_reference=True))
        return media

    assert asyncio.run(blank()).id.file_reference == b""


def test_raw_edits(client):
    p = post()
    req = capture(push_edit(client, CH, p))
    assert isinstance(req, functions.messages.EditMessageRequest)
    assert req.id == 50 and req.message == "Hello 😀" and req.no_webpage is True and req.reply_markup is not None
    # no buttons -> no reply_markup (Telegram then removes the keyboard)
    req = capture(push_edit(client, CH, post(buttons=[])))
    assert req.reply_markup is None and req.message == "Hello 😀"
    # markup-only edit leaves the text alone
    req = capture(edit_raw(client, peer_of(CH), 9, markup=build_markup([[{"t": "A", "u": "https://a.com"}]])))
    assert req.message is None and req.entities is None and req.reply_markup.rows[0].buttons[0].text == "A"


def test_media_replacement_edit(client):
    ref = photo_ref()
    req = capture(push_edit(client, CH, post(media_kind="photo", media_file_id=ref), replace_media=True))
    assert isinstance(req, functions.messages.EditMessageRequest)
    assert isinstance(req.media, types.InputMediaPhoto) and req.message == "Hello 😀"
    assert req.reply_markup is not None


def test_reads_deletes_and_rights_requests(client):
    peer = peer_of(CH)
    req = capture(client.get_messages(peer, ids=[1, 2, 3]))
    assert isinstance(req, functions.channels.GetMessagesRequest)
    assert isinstance(req.channel, types.InputChannel) and [i.id for i in req.id] == [1, 2, 3]
    req = capture(client.delete_messages(peer, [7]))
    assert isinstance(req, functions.channels.DeleteMessagesRequest) and req.id == [7]
    req = capture(client(functions.channels.GetFullChannelRequest(peer)))
    assert isinstance(req.channel, types.InputChannel)
    req = capture(client(functions.channels.GetParticipantRequest(channel=peer, participant=types.InputPeerSelf())))
    assert isinstance(req.participant, types.InputPeerSelf)


def test_entities_with_every_stored_kind_serialise():
    ents = [
        types.MessageEntityBold(0, 1), types.MessageEntityItalic(1, 1), types.MessageEntityUnderline(2, 1),
        types.MessageEntityStrike(3, 1), types.MessageEntitySpoiler(4, 1), types.MessageEntityCode(5, 1),
        types.MessageEntityPre(6, 1, "py"), types.MessageEntityTextUrl(7, 1, "https://x.com"),
        types.MessageEntityBlockquote(8, 1, True), types.MessageEntityCustomEmoji(9, 2, 123456789),
    ]
    back = de_entities(ser_entities(ents))
    assert [type(e) for e in back] == [type(e) for e in ents]
    assert back[8].collapsed is True
    for e in back:
        bytes(e)


def test_ui_keyboards_serialise(client):
    p = post()
    p.status, p.channel_id, p.source = "draft", 1, "bot"
    p.buttons = [[{"t": "A", "u": "https://a.com"}, {"t": "⚠️ no link", "u": None}], [{"t": "👍", "raw": ""}]]
    for kb in (panel_keyboard(p), button_keyboard(p, 0, 0), button_keyboard(p, 1, 0)):
        mk = client.build_reply_markup(kb)
        assert isinstance(bytes(mk), bytes)
    for label_row in panel_keyboard(p):
        for b in label_row:
            assert len(b.type.data) <= 64  # Telegram's limit for callback data


def test_button_callback_data_fits_for_large_ids():
    p = post(id=2_000_000_000)
    p.status, p.channel_id, p.source = "sent", -1, "bot"
    p.buttons = [[{"t": "x", "u": "https://a.com"}] * 8] * 12
    for row in panel_keyboard(p):
        for b in row:
            assert len(b.type.data) <= 64