Blame for tests/fakes.py

"""Tiny in-memory stand-ins for Telethon, good enough to exercise the engine offline."""
import datetime
import itertools
import os
from types import SimpleNamespace

from telethon import errors, functions, types
from telethon.tl.custom import Message

from app.tgutil import make_url_button, media_key


def make_msg(i, text="", entities=None, markup=None, out=False, media=None, grouped_id=None, reply_to=None, pinned=False):
    return Message(
        id=i,
        peer_id=types.PeerChannel(1),
        date=datetime.datetime.now(datetime.timezone.utc),
        message=text,
        out=out,
        entities=entities,
        reply_markup=markup,
        media=media,
        grouped_id=grouped_id,
        reply_to=types.MessageReplyHeader(reply_to_msg_id=reply_to) if reply_to else None,
        pinned=True if pinned else None,
    )


def url_btn(text, url):
    return make_url_button(text, url)


def cb_btn(text, data):
    return types.KeyboardButton(text=text, type=types.InlineButtonTypeCallback(data))


def markup(*rows):
    return types.ReplyInlineMarkup([types.KeyboardButtonRow(list(r)) for r in rows])


class FakeClient:
    def __init__(self, msgs, pts=None):
        self.msgs = dict(msgs)
        self.pts = pts
        self.edits = []
        self.fail = {}  # message id -> exception to raise on edit

    async def __call__(self, req):
        if isinstance(req, functions.channels.GetFullChannelRequest):
            return SimpleNamespace(full_chat=SimpleNamespace(pts=self.pts))
        if isinstance(req, functions.messages.EditMessageRequest):
            if req.id in self.fail:
                raise self.fail[req.id]
            m = self.msgs[req.id]
            self.edits.append(req)
            if req.message is not None:
                m.message = req.message
                m.entities = list(req.entities or [])
                m.reply_markup = req.reply_markup  # like Telegram: no markup sent -> keyboard removed
            elif req.reply_markup is not None:
                m.reply_markup = req.reply_markup
            return None
        raise NotImplementedError(type(req))

    async def get_messages(self, peer, ids=None):
        return [self.msgs.get(i) for i in ids]


def rpc(name):
    return getattr(errors, name)(request=None)


# ======================================================================================================
# A fake Telegram "channel": enough of the API for the repost / delete / buttons / userbot code, offline.
# ======================================================================================================
NOW = datetime.datetime.now(datetime.timezone.utc)
USER_HASH = 424242  # the channel's access hash as the userbot account sees it (differs from the bot's)


def photo_media(pid=11):
    photo = types.Photo(pid, pid * 7, b"ref%d" % pid, NOW, [types.PhotoSize("x", 100, 100, 999)], 2)
    return types.MessageMediaPhoto(photo=photo)


def video_media(did=21, thumb=False):
    """A video; `thumb`: Telegram shows a picture for it (its bytes: b"THUMB OF <id>" unless the test sets `_media_data`)."""
    thumbs = [types.PhotoStrippedSize("i", b"\x01\x28\x28"), types.PhotoSize("m", 320, 180, 1234)] if thumb else None
    doc = types.Document(
        did, did * 7, b"fr%d" % did, NOW, "video/mp4", 1000, 2, [types.DocumentAttributeVideo(5.0, 320, 240)], thumbs=thumbs
    )
    return types.MessageMediaDocument(document=doc)


def as_the_userbot_holds_it(media):
    """The same photo / file with the access data of ANOTHER account: its id is the same everywhere, its access_hash and
    file_reference belong to one account (so one account's copy is refused when another presents it)."""
    if isinstance(media, types.MessageMediaPhoto):
        p = media.photo
        return types.MessageMediaPhoto(
            photo=types.Photo(p.id, p.access_hash + 1000, b"ub-" + p.file_reference, p.date, p.sizes, p.dc_id)
        )
    d = media.document
    return types.MessageMediaDocument(
        document=types.Document(
            d.id, d.access_hash + 1000, b"ub-" + d.file_reference, d.date, d.mime_type, d.size, d.dc_id, d.attributes,
            thumbs=d.thumbs,
        )
    )


class FakeDM:
    """The chat between the bot and the userbot account: what the bot put there and what the account holds of it."""

    def __init__(self):
        self.bot_user = types.User(id=9100, access_hash=31337, first_name="Post Bot", username="the_post_bot", bot=True)
        self.userbot_id = None  # the userbot account (set when its FakeUserClient is made) ...
        self.userbot_user = None  # ... and its types.User
        self.msgs = []  # the chat as the userbot sees it, oldest first
        self.sent = []  # (the bot's id of the message, its kwargs) for everything the bot sent there
        self.bot_ids = {}  # the bot's id of a message -> the userbot's message
        self.deleted = []  # the bot's ids of the messages it deleted
        self.error = None  # raised when the bot sends there (the account never started the bot, or blocked it)
        self.read_error = None  # raised when the userbot reads the chat
        self.reference_error = None  # raised when the userbot edits a post with a file it holds (not with an upload)
        self.late = 0  # the next N reads find nothing: the message has not reached the userbot yet
        self.known = {}  # ("photo" | "doc", id) -> (access_hash, file_reference) the userbot account holds
        self._ids = itertools.count(500)  # the userbot's message ids
        self._bot_ids = itertools.count(1000)  # the bot's message ids


def service_msg(i):
    return types.MessageService(id=i, peer_id=types.PeerChannel(1), date=NOW, action=types.MessageActionPinMessage())


def quote(offset, length, collapsed=None):
    return types.MessageEntityBlockquote(offset, length, collapsed)


def _bytes(obj):
    return bytes(obj) if obj is not None else None


def _empty_rights(r):
    return not any(v for k, v in r.to_dict().items() if k != "_")


class FakeChannelClient:
    """Telegram's side of ONE channel (peer id 1) as seen by the bot, with switches for the odd things Telegram does."""

    def __init__(self, msgs=None, *, username=None, cid=1, title="Test"):
        self.cid = cid  # this simulation is the channel with this id; add_channel() puts more channels next to it
        self.title = title
        self.registry = {cid: self}
        self.invite_hash = None  # a t.me/+hash invite link of this channel the bot "is in" (for CheckChatInvite)
        self.msgs = dict(msgs or {})
        self.username = username
        self.requests = []  # every request received, in order
        self.edits = []  # EditMessageRequests Telegram accepted
        self.delete_requests = []  # the id lists of channels.deleteMessages calls
        self._top = max(self.msgs, default=0)
        self._pts = self._top  # the channel's event counter: new messages, edits, deletes and pins all move it
        self._gid = 7000
        self._media = {}
        for m in self.msgs.values():
            self._remember_media(m)
        # --- switches
        self.restrict_forwards = False  # "Restrict saving content" is on
        self.hide_keyboard_on_create = 0  # the next N created posts show no keyboard until a real change (suspected bug)
        self.forward_hides_keyboard = False  # a forwarded copy keeps its keyboard "hidden": edits to the same keyboard are no-ops
        self.drop_all_markup = False  # Telegram never shows a keyboard (to test giving up)
        self.undeletable = set()  # ids the BOT may not delete (like posts older than 48 hours)
        self.delete_error = None  # raised by every delete request of the bot
        self.fail = {}  # request class name -> exception to raise
        self.rights = dict(admin=True, post=True, edit=True, delete=True, invite=False, add_admins=False)
        self.participants_visible = True
        self.known_users = {}  # user id -> types.User the bot has "met"
        self.members = set()  # user ids (other than the bot) that are in the channel
        self.admins = {}  # user id -> ChatAdminRights
        self.invites = {}  # hash -> dict(limit, uses, revoked)
        self.primary_invite = None  # the channel's own invite link
        self.require_approval = False
        self.users = {}  # user id -> types.User, for participant lists
        self.cant_delete_at_all = set()  # ids nobody can delete (userbot included)
        self.restriction = None  # text: Telegram restricts the whole channel (a copyright strike)
        self.restrict_new = None  # f(new message) -> text or None: Telegram itself restricts a NEW post (a blocked file ...)
        self.about = ""  # the channel's description
        self.photo_id = None  # the channel's profile photo (its bytes are in _photo_data)
        self._photo_data = {}  # photo id -> bytes; shared by all channels next to each other
        self._media_data = {}  # ("photo" | "doc", id) -> bytes of a message's photo or file; shared as well
        self._uploads = {}  # uploaded file id -> bytes; shared as well
        self.user_edit_keeps_markup = False  # an edit by a user account leaves the keyboard alone (else it is dropped)
        self.user_edit_frees_message = False  # after a user account edited it, the bot may edit the post again
        self.user_edits = []  # the EditMessageRequests the userbot account made
        self._counter = [9000]  # ids for photos and uploads; shared
        self._dm = FakeDM()  # the chat between the bot and the userbot account; shared
        self.pinned_order = []  # message ids in the order they were pinned
        self._n_invites = 0

    # ----------------------------------------------------------------------------------- helpers
    def add_channel(self, other):
        """Put another channel next to this one: requests that name it (by channel id) are answered by it."""
        other.registry = self.registry
        self.registry[other.cid] = other
        self._media.update(other._media)
        other._media = self._media  # a photo or file is known to the account, whichever channel it is sent to
        self._photo_data.update(other._photo_data)
        other._photo_data, other._uploads, other._counter = self._photo_data, self._uploads, self._counter
        self._media_data.update(other._media_data)
        other._media_data = self._media_data
        other._dm = self._dm
        return other

    def set_photo(self, data: bytes) -> int:
        """Give the channel a profile photo (what it looks like is the bytes)."""
        self._counter[0] += 1
        self.photo_id = self._counter[0]
        self._photo_data[self.photo_id] = data
        return self.photo_id

    def photo_bytes(self):
        return self._photo_data.get(self.photo_id) if self.photo_id else None

    def _chat_photo(self):
        if not self.photo_id:
            return types.PhotoEmpty(0)
        return types.Photo(self.photo_id, 1, b"ref", NOW, [types.PhotoSize("x", 100, 100, 999)], 2)

    def media_bytes(self, media):
        """What a photo / file of a message looks like (its bytes): set in `_media_data`, else made up from its id."""
        if isinstance(media, types.MessageMediaPhoto) and media.photo:
            key = ("photo", media.photo.id)
        elif isinstance(media, types.MessageMediaDocument) and media.document:
            key = ("doc", media.document.id)
        else:
            raise ValueError("this fake only has the bytes of a photo or a file")
        return self._media_data.get(key, b"bytes of %s %d" % (key[0].encode(), key[1]))

    async def download_media(self, media, file=None, thumb=None):
        if thumb is not None:  # the thumbnail of a file: None when Telegram has none
            doc = getattr(media, "document", None)
            if not isinstance(doc, types.Document) or not doc.thumbs:
                return None
            data = self._media_data.get(("thumb", doc.id), b"THUMB OF %d" % doc.id)
            if file is None or file is bytes:
                return data
            path = str(file) + f"photo_{doc.id}.jpg" if str(file).endswith(os.sep) else str(file)
            with open(path, "wb") as f:
                f.write(data)
            return path
        if isinstance(media, types.Photo) and media.id in self._photo_data:
            return self._photo_data[media.id]
        data = self.media_bytes(media)
        if file is None or file is bytes:
            return data
        if isinstance(media, types.MessageMediaPhoto):
            name = f"photo_{media.photo.id}.jpg"
        else:  # a file keeps the name it was sent with
            name = next(
                (a.file_name for a in media.document.attributes if isinstance(a, types.DocumentAttributeFilename)),
                f"file_{media.document.id}",
            )
        path = str(file) + name if str(file).endswith(os.sep) else str(file)
        with open(path, "wb") as f:
            f.write(data)
        return path

    async def upload_file(self, data, file_name=None):
        self._counter[0] += 1
        self._uploads[self._counter[0]] = bytes(data)
        return types.InputFile(id=self._counter[0], parts=1, name=file_name or "file", md5_checksum="")

    def _target(self, req):
        """The channel a request is about (a forward is carried out by its destination)."""
        peer = getattr(req, "to_peer", None) or getattr(req, "peer", None) or getattr(req, "channel", None)
        cid = getattr(peer, "channel_id", None)
        return self.registry.get(cid, self) if cid is not None else self

    def _remember_media(self, m):
        md = getattr(m, "media", None)
        if isinstance(md, types.MessageMediaPhoto) and md.photo:
            self._media[("photo", md.photo.id)] = md
        elif isinstance(md, types.MessageMediaDocument) and md.document:
            self._media[("doc", md.document.id)] = md

    def _media_for(self, inp):
        if isinstance(inp, types.InputMediaPhoto):
            key = ("photo", inp.id.id)
        elif isinstance(inp, types.InputMediaDocument):
            key = ("doc", inp.id.id)
        else:
            raise rpc("MediaInvalidError")
        if key not in self._media:
            raise rpc("MediaEmptyError")
        return self._media[key]

    def _effective(self, m):
        hidden = getattr(m, "_hidden", None)
        return hidden if hidden is not None else m.reply_markup

    def _set_markup(self, m, markup, *, hide=False):
        if markup is not None and (hide or self.drop_all_markup):
            m._hidden, m.reply_markup = markup, None
        else:
            m._hidden, m.reply_markup = None, markup

    def _check_blocks(self, inp):
        """What Telegram refuses in the blocks of an InputRichMessage (RICH_MESSAGE_BLOCK_INVALID), as far as it is known: a
        map that holds a stored point (the blocks it hands out hold those; it wants an input point back), a block that its
        layer doesn't know, a thinking block (the Bot API: drafts only) and a text mention of somebody the message carries
        no access data for.
        `refuse_blocks`: further block classes (names) a test wants refused."""
        users = {getattr(u, "user_id", None) for u in inp.users or []}
        refused = (types.PageBlockMap, types.PageBlockUnsupported, types.PageBlockThinking)
        extra = set(getattr(self, "refuse_blocks", ()))

        def walk(o):
            if (
                isinstance(o, refused)
                or type(o).__name__ in extra
                or (isinstance(o, types.TextMentionName) and o.user_id not in users)
            ):
                raise errors.BadRequestError(
                    functions.messages.SendMessageRequest(peer=types.InputPeerSelf(), message=""),
                    "RICH_MESSAGE_BLOCK_INVALID",
                    400,
                )
            if isinstance(o, (list, tuple)):
                for x in o:
                    walk(x)
            elif hasattr(o, "to_dict") and not isinstance(o, (str, bytes)):
                for v in vars(o).values():
                    walk(v)

        walk(list(inp.blocks))

    def _rich(self, inp):
        """What Telegram makes of an InputRichMessage*: for HTML / Markdown the "parse" is one paragraph holding the source
        (the same source always gives the same blocks); a message sent as blocks keeps them, with its pictures and files."""
        if isinstance(inp, (types.InputRichMessageHTML, types.InputRichMessageMarkdown)):
            src = inp.html if isinstance(inp, types.InputRichMessageHTML) else inp.markdown
            if not src.strip():
                raise rpc("MessageEmptyError")
            kind = "html" if isinstance(inp, types.InputRichMessageHTML) else "markdown"
            return types.RichMessage(
                blocks=[types.PageBlockParagraph(types.TextPlain(f"[{kind}]{src}"))], photos=[], documents=[], rtl=inp.rtl
            )
        if isinstance(inp, types.InputRichMessage):
            self._check_blocks(inp)
            photos = [
                types.Photo(p.id, p.access_hash, p.file_reference, NOW, [], 1) for p in inp.photos or []
            ]
            docs = [
                types.Document(d.id, d.access_hash, d.file_reference, NOW, "video/mp4", 1, 1, []) for d in inp.documents or []
            ]
            return types.RichMessage(blocks=list(inp.blocks), photos=photos, documents=docs, rtl=inp.rtl)
        raise rpc("MediaInvalidError")

    def _new(self, text, entities, markup, media=None, gid=None, invert=None, hide=False, reply_to=None, rich=None):
        self._top += 1
        self._pts += 1
        m = Message(
            id=self._top,
            peer_id=types.PeerChannel(self.cid),
            date=NOW,
            message=text,
            out=True,
            entities=list(entities) if entities else None,
            media=media,
            grouped_id=gid,
            invert_media=invert,
            reply_to=types.MessageReplyHeader(reply_to_msg_id=reply_to) if reply_to else None,
            rich_message=rich,
        )
        if self.hide_keyboard_on_create > 0 and markup is not None:
            hide = True
            self.hide_keyboard_on_create -= 1
        self._set_markup(m, markup, hide=hide)
        if self.restrict_new is not None:
            why = self.restrict_new(m)
            if why:
                m.restriction_reason = [types.RestrictionReason(platform="all", reason="copyright", text=why)]
        self.msgs[m.id] = m
        self._remember_media(m)
        return m

    def _updates(self, msgs):
        return types.Updates(
            updates=[types.UpdateNewChannelMessage(message=m, pts=m.id, pts_count=1) for m in msgs],
            users=[],
            chats=[],
            date=NOW,
            seq=0,
        )

    def _channel_obj(self):
        extra = {}
        if self.restriction:
            extra = dict(
                restricted=True,
                restriction_reason=[types.RestrictionReason(platform="all", reason="copyright", text=self.restriction)],
            )
        return types.Channel(
            id=self.cid, title=self.title, photo=types.ChatPhotoEmpty(), date=NOW, access_hash=USER_HASH,
            username=self.username, broadcast=True, **extra,
        )

    # --------------------------------------------------------------------------------- dispatcher
    async def __call__(self, req):
        target = self._target(req)
        if target is not self:
            return await target(req)
        bytes(req)  # every request the bot builds must serialise
        self.requests.append(req)
        name = type(req).__name__
        err = self.fail.get(name)
        if err is not None:
            raise err
        handler = getattr(self, "_do_" + name, None)
        if handler is None:
            raise NotImplementedError(name)
        return handler(req)

    async def get_messages(self, peer, ids=None):
        chan = self.registry.get(getattr(peer, "channel_id", None), self)
        err = chan.fail.get("GetMessagesRequest")  # reading a channel the bot may not read (a real client sends this request)
        if err is not None:
            raise err
        return [chan.msgs.get(i) for i in ids]

    async def get_input_entity(self, peer):
        if isinstance(peer, int) and peer in self.known_users:
            u = self.known_users[peer]
            return types.InputPeerUser(u.id, u.access_hash)
        raise ValueError(f"Could not find the input entity for {peer!r}")

    # ------------------------------------------------------ the bot and its chat with the userbot account
    async def get_me(self):
        return self._dm.bot_user

    def _is_dm(self, peer):
        """Is `peer` the userbot account (the only user this simulation sends files to)?"""
        return isinstance(peer, types.InputPeerUser) and peer.user_id == self._dm.userbot_id

    async def send_file(self, peer, file, **kw):
        """The bot sends a photo / file to the userbot account: Telegram reuses the file, the account gets its own copy."""
        dm = self._dm
        if not self._is_dm(peer):
            raise NotImplementedError("this fake only sends files to the userbot account")
        if dm.error is not None:
            raise dm.error
        if peer.access_hash != dm.userbot_user.access_hash:  # the bot has to know the account (its access hash)
            raise rpc("PeerIdInvalidError")
        key = media_key(file)
        if key is None:
            raise rpc("MediaInvalidError")
        mine = as_the_userbot_holds_it(file)
        inner = mine.photo if key[0] == "photo" else mine.document
        dm.known[key] = (inner.access_hash, inner.file_reference)
        self._media[key] = file  # the file as the bot holds it: what the channel shows after an edit by reference
        msg = Message(id=next(dm._ids), peer_id=types.PeerUser(dm.bot_user.id), date=NOW, message="", out=False, media=mine)
        dm.msgs.append(msg)
        bot_id = next(dm._bot_ids)
        dm.bot_ids[bot_id] = msg
        dm.sent.append((bot_id, kw))
        return SimpleNamespace(id=bot_id)

    async def delete_messages(self, peer, ids, **kw):
        dm = self._dm
        if not self._is_dm(peer):
            raise NotImplementedError("this fake only deletes in the chat with the userbot account")
        for i in ids:
            dm.deleted.append(i)
            msg = dm.bot_ids.pop(i, None)
            dm.msgs = [m for m in dm.msgs if m is not msg]  # (identity: Telethon's Message has no usable ==)
        return [SimpleNamespace(pts_count=len(ids))]

    # ------------------------------------------------------------------------- message requests
    def _do_GetMessagesRequest(self, req):
        """channels.getMessages as Telegram answers it: the posts as they are now, MessageEmpty for missing ones.
        (Used when a real Telethon client sits in front of this simulation; get_messages() above serves the plain fakes.)"""
        out = []
        for ref in req.id:
            m = self.msgs.get(ref.id)
            if m is None:
                out.append(types.MessageEmpty(id=ref.id, peer_id=types.PeerChannel(self.cid)))
            elif isinstance(m, (types.MessageService, types.MessageEmpty)):
                out.append(m)
            else:
                out.append(
                    types.Message(
                        id=m.id, peer_id=m.peer_id, date=m.date, message=m.message, out=m.out, entities=m.entities,
                        reply_markup=m.reply_markup, media=m.media, grouped_id=m.grouped_id,
                        invert_media=getattr(m, "invert_media", None), reply_to=m.reply_to, pinned=m.pinned,
                        restriction_reason=getattr(m, "restriction_reason", None),
                    )
                )
        return types.messages.ChannelMessages(
            pts=self._top, count=len(out), messages=out, topics=[], chats=[self._channel_obj()], users=[]
        )

    def _do_ForwardMessagesRequest(self, req):
        origin = self.registry.get(getattr(req.from_peer, "channel_id", None), self)
        if origin.restrict_forwards:  # "Restrict saving content" is a setting of the channel the posts come from
            raise rpc("ChatForwardsRestrictedError")
        out, gids = [], {}
        for i in req.id:
            src = origin.msgs.get(i)
            if src is None:
                continue
            gid = gids.setdefault(src.grouped_id, self._next_gid()) if src.grouped_id else None
            out.append(
                self._new(
                    src.message, src.entities, src.reply_markup, src.media, gid, getattr(src, "invert_media", None),
                    hide=self.forward_hides_keyboard, rich=getattr(src, "rich_message", None),
                )
            )
        if not out:
            raise rpc("MessageIdInvalidError")
        return self._updates(out)

    def _next_gid(self):
        self._gid += 1
        return self._gid

    @staticmethod
    def _reply_id(req):
        r = getattr(req, "reply_to", None)
        return r.reply_to_msg_id if r is not None else None

    def _check_reply(self, req):
        rid = self._reply_id(req)
        if rid is not None and rid not in self.msgs:
            raise rpc("MessageIdInvalidError")  # REPLY_MESSAGE_ID_INVALID
        return rid

    def _do_SendMessageRequest(self, req):
        rid = self._check_reply(req)
        if req.rich_message is not None:
            rich = self._rich(req.rich_message)
            m = self._new("", None, req.reply_markup, None, None, None, reply_to=rid, rich=rich)
            m.rich_input = req.rich_message  # what the bot asked for (the tests look at it)
            return self._updates([m])
        return self._updates([self._new(req.message, req.entities, req.reply_markup, None, None, req.invert_media, reply_to=rid)])

    def _do_SendMediaRequest(self, req):
        media = self._media_for(req.media)
        rid = self._check_reply(req)
        return self._updates([self._new(req.message, req.entities, req.reply_markup, media, None, req.invert_media, reply_to=rid)])

    def _do_SendMultiMediaRequest(self, req):
        gid = self._next_gid()
        rid = self._check_reply(req)
        out = [
            self._new(s.message, s.entities, None, self._media_for(s.media), gid, req.invert_media, reply_to=rid)
            for s in req.multi_media
        ]
        return self._updates(out)

    def _do_UpdatePinnedMessageRequest(self, req):
        """Pinning in a channel needs the "Edit messages of others" right; Telegram adds a "pinned" notice to the channel."""
        if not self.rights.get("edit"):
            raise rpc("ChatAdminRequiredError")
        m = self.msgs.get(req.id)
        if m is None or isinstance(m, types.MessageService):
            raise rpc("MessageIdInvalidError")
        if req.unpin:
            m.pinned = None
            return self._updates([])
        if m.pinned:
            raise rpc("MessageNotModifiedError")
        m.pinned = True
        self.pinned_order.append(req.id)
        self._top += 1
        self._pts += 2
        note = types.MessageService(
            id=self._top, peer_id=types.PeerChannel(self.cid), date=NOW, action=types.MessageActionPinMessage(),
            reply_to=types.MessageReplyHeader(reply_to_msg_id=req.id),
        )
        self.msgs[note.id] = note
        return self._updates([note])

    def _do_EditMessageRequest(self, req):
        m = self.msgs.get(req.id)
        if m is None:
            raise rpc("MessageIdInvalidError")
        if not getattr(m, "out", False) and not self.rights.get("edit"):  # somebody else's post needs that right
            raise rpc("MessageAuthorRequiredError")
        if getattr(m, "via_other_bot", False):  # buttons or a post made through another bot: only that bot may edit
            raise rpc("MessageIdInvalidError")
        if req.rich_message is not None:  # the post becomes this rich message; no keyboard sent = keyboard removed
            rich = self._rich(req.rich_message)
            if (
                m.rich_message is not None
                and bytes(m.rich_message) == bytes(rich)
                and _bytes(req.reply_markup) == _bytes(self._effective(m))
            ):
                raise rpc("MessageNotModifiedError")
            m.message, m.entities, m.rich_message, m.rich_input = "", None, rich, req.rich_message
            self._set_markup(m, req.reply_markup)
            self.edits.append(req)
            self._pts += 1
            return self._updates([])
        if req.message is None:  # markup-only edit
            text, ents = m.message, list(m.entities or [])
            markup = req.reply_markup if req.reply_markup is not None else self._effective(m)
        else:
            text, ents, markup = req.message, list(req.entities or []), req.reply_markup  # None = keyboard removed
        if (
            text == m.message
            and [_bytes(e) for e in ents] == [_bytes(e) for e in (m.entities or [])]
            and _bytes(markup) == _bytes(self._effective(m))
        ):
            raise rpc("MessageNotModifiedError")
        m.message, m.entities = text, (ents or None)
        if req.message is not None:
            m.rich_message = None
        self._set_markup(m, markup)
        self.edits.append(req)
        self._pts += 1
        return self._updates([])

    def _do_GetRichMessageRequest(self, req):
        """messages.getRichMessage: the whole rich message of a post (the tests may put a longer one in `full_rich`)."""
        full = getattr(self, "full_rich", {}).get(req.id)
        m = self.msgs.get(req.id)
        if full is None and (m is None or m.rich_message is None):
            raise rpc("MessageIdInvalidError")
        return full if full is not None else m.rich_message

    def _do_DeleteMessagesRequest(self, req):
        ids = list(req.id)
        self.delete_requests.append(ids)
        if self.delete_error is not None:
            raise self.delete_error
        n = 0
        for i in ids:
            if i in self.undeletable or i in self.cant_delete_at_all or i not in self.msgs:
                continue
            del self.msgs[i]
            n += 1
        self._pts += n
        return types.messages.AffectedMessages(pts=self._pts, pts_count=n)

    # ------------------------------------------------------------------------ channel / rights
    def _do_GetFullChannelRequest(self, req):
        inv = types.ChatInviteExported(link=self.primary_invite, admin_id=1, date=NOW) if self.primary_invite else None
        return SimpleNamespace(
            full_chat=SimpleNamespace(
                pts=max(self._pts, self._top), exported_invite=inv, about=self.about, chat_photo=self._chat_photo()
            ),
            chats=[self._channel_obj()],
        )

    def _notice(self, action):
        """The service message Telegram puts into a channel when its name or photo changes."""
        self._top += 1
        self._pts += 1
        note = types.MessageService(id=self._top, peer_id=types.PeerChannel(self.cid), date=NOW, action=action)
        self.msgs[note.id] = note
        return note

    def _need_info_right(self):
        if not self.rights.get("change_info", True):
            raise rpc("ChatAdminRequiredError")

    def _do_EditTitleRequest(self, req):
        self._need_info_right()
        if req.title == self.title:
            raise rpc("ChatNotModifiedError")
        self.title = req.title
        return self._updates([self._notice(types.MessageActionChatEditTitle(req.title))])

    def _do_EditChatAboutRequest(self, req):
        self._need_info_right()
        if req.about == self.about:
            raise rpc("ChatAboutNotModifiedError")
        self.about = req.about
        return True

    def _do_EditPhotoRequest(self, req):
        self._need_info_right()
        data = self._uploads[req.photo.file.id]
        self.set_photo(data)
        return self._updates([self._notice(types.MessageActionChatEditPhoto(self._chat_photo()))])

    def _do_GetParticipantRequest(self, req):
        r = self.rights
        if r is None:
            raise rpc("UserNotParticipantError")
        if not r.get("admin", True):
            return SimpleNamespace(participant=types.ChannelParticipant(user_id=1, date=NOW))
        rights = types.ChatAdminRights(
            post_messages=r.get("post"), edit_messages=r.get("edit"), delete_messages=r.get("delete"),
            invite_users=r.get("invite"), add_admins=r.get("add_admins"), change_info=r.get("change_info", True),
        )
        return SimpleNamespace(participant=types.ChannelParticipantAdmin(user_id=1, promoted_by=1, date=NOW, admin_rights=rights))

    def _do_ExportChatInviteRequest(self, req):
        if not self.rights.get("invite"):
            raise rpc("ChatAdminRequiredError")
        self._n_invites += 1
        h = f"AbCdEfGh{self._n_invites:04d}"
        self.invites[h] = dict(limit=req.usage_limit, uses=0, revoked=False, title=req.title, expire=req.expire_date)
        return types.ChatInviteExported(
            link=f"https://t.me/+{h}", admin_id=1, date=NOW, usage_limit=req.usage_limit, expire_date=req.expire_date, title=req.title
        )

    def _do_EditExportedChatInviteRequest(self, req):
        h = req.link.rsplit("+", 1)[-1]
        if h in self.invites and req.revoked:
            self.invites[h]["revoked"] = True
        return SimpleNamespace(invite=None)

    def _do_EditAdminRequest(self, req):
        if not self.rights.get("add_admins"):
            raise rpc("ChatAdminRequiredError")
        uid = req.user_id.user_id
        known = self.users.get(uid)
        if known is None or req.user_id.access_hash != known.access_hash:
            raise rpc("UserIdInvalidError")
        if _empty_rights(req.admin_rights):
            self.admins.pop(uid, None)
        else:
            self.admins[uid] = req.admin_rights
        return self._updates([])

    def _do_GetParticipantsRequest(self, req):
        if not self.participants_visible:
            raise rpc("ChatAdminRequiredError")
        users = [self.users[u] for u in sorted(self.members) if u in self.users]
        if isinstance(req.filter, types.ChannelParticipantsSearch):
            q = req.filter.q.lower()
            users = [u for u in users if q in (u.username or "").lower() or q in (u.first_name or "").lower()]
        return types.channels.ChannelParticipants(count=len(users), participants=[], chats=[], users=users)


class FakeUserClient:
    """The userbot account's own Telegram connection, sharing one FakeChannelClient with the bot."""

    def __init__(self, chan, *, user_id=777, authorized=True, bot=False, username="helper_acct", first_name="Helper"):
        self.chan = chan
        self.me = types.User(id=user_id, access_hash=555, first_name=first_name, username=username, bot=bot or None)
        chan.users[user_id] = self.me
        chan._dm.userbot_id, chan._dm.userbot_user = user_id, self.me
        self.authorized = authorized
        self.connected = False
        self.known_channel = False
        self.requests = []
        self.deleted = []
        self.uploads = []  # (path, file name, size) of everything uploaded

    async def connect(self):
        self.connected = True

    async def disconnect(self):
        self.connected = False

    async def is_user_authorized(self):
        return self.authorized

    async def get_me(self):
        return self.me

    async def get_input_entity(self, peer):
        if isinstance(peer, types.PeerChannel) and self.known_channel:
            return types.InputPeerChannel(peer.channel_id, USER_HASH)
        if isinstance(peer, str) and self.chan.username and peer.lstrip("@").lower() == self.chan.username.lower():
            return types.InputPeerChannel(1, USER_HASH)
        bot = self.chan._dm.bot_user
        if isinstance(peer, str) and peer.lstrip("@").lower() == bot.username.lower():
            return types.InputPeerUser(bot.id, bot.access_hash)
        raise ValueError(f"Could not find the input entity for {peer!r}")

    async def get_messages(self, peer, limit=None, **kw):
        """The newest messages of this account's chat with the bot (newest first, like Telethon)."""
        dm, bot = self.chan._dm, self.chan._dm.bot_user
        err = self.chan.fail.get("user:GetHistoryRequest")
        if err is not None:
            raise err
        if not (isinstance(peer, types.InputPeerUser) and peer.user_id == bot.id and peer.access_hash == bot.access_hash):
            raise ValueError(f"Could not find the input entity for {peer!r}")
        if dm.read_error is not None:
            raise dm.read_error
        if dm.late > 0:
            dm.late -= 1
            return []
        return list(reversed(dm.msgs))[: limit or 1]

    async def iter_dialogs(self):
        if self.known_channel and self.me.id in self.chan.members:
            yield SimpleNamespace(entity=self.chan._channel_obj())

    def _join(self):
        self.chan.members.add(self.me.id)
        self.known_channel = True

    async def upload_file(self, path, file_name=None):
        with open(path, "rb") as f:
            data = f.read()
        chan = self.chan
        chan._counter[0] += 1
        chan._uploads[chan._counter[0]] = data
        self.uploads.append((path, file_name, len(data)))
        return types.InputFile(id=chan._counter[0], parts=1, name=file_name or os.path.basename(path), md5_checksum="")

    def _uploaded_media(self, inp):
        """The photo / file Telegram makes out of an upload: new ids, the bytes that were sent."""
        chan = self.chan
        data = chan._uploads[inp.file.id]
        chan._counter[0] += 1
        n = chan._counter[0]
        if isinstance(inp, types.InputMediaUploadedPhoto):
            chan._media_data[("photo", n)] = data
            photo = types.Photo(n, n * 7, b"up%d" % n, NOW, [types.PhotoSize("x", 100, 100, len(data))], 2)
            return types.MessageMediaPhoto(photo=photo)
        if isinstance(inp, types.InputMediaUploadedDocument):
            chan._media_data[("doc", n)] = data
            thumbs = None
            if inp.thumb is not None:  # an uploaded picture becomes the file's thumbnail
                chan._media_data[("thumb", n)] = chan._uploads[inp.thumb.id]
                thumbs = [types.PhotoSize("m", 320, 180, len(chan._uploads[inp.thumb.id]))]
            doc = types.Document(n, n * 7, b"up%d" % n, NOW, inp.mime_type, len(data), 2, list(inp.attributes), thumbs=thumbs)
            return types.MessageMediaDocument(document=doc)
        raise rpc("MediaInvalidError")

    def _by_reference(self, inp):
        """An InputMediaPhoto / InputMediaDocument: only a photo / file this account holds, with ITS OWN access data."""
        chan = self.chan
        if chan._dm.reference_error is not None:
            raise chan._dm.reference_error
        key = ("photo", inp.id.id) if isinstance(inp, types.InputMediaPhoto) else ("doc", inp.id.id)
        mine = chan._dm.known.get(key)
        if mine is None:
            raise rpc("MediaEmptyError")  # this account has never seen that file
        if (inp.id.access_hash, inp.id.file_reference) != mine:
            raise rpc("FileReferenceExpiredError")  # somebody else's access data
        return chan._media[key]  # the file as the channel shows it (as the bot sees it)

    def _edit(self, req):
        """messages.editMessage by a user account: it needs the "Edit messages of others" right and can't set a keyboard."""
        chan, uid = self.chan, self.me.id
        if uid not in chan.members:
            raise rpc("ChannelPrivateError")
        admin = chan.admins.get(uid)
        if admin is None or not admin.edit_messages:
            raise rpc("ChatAdminRequiredError")
        m = chan.msgs.get(req.id)
        if m is None:
            raise rpc("MessageIdInvalidError")
        if req.media is None:
            media = None
        elif isinstance(req.media, (types.InputMediaPhoto, types.InputMediaDocument)):
            media = self._by_reference(req.media)
        else:
            media = self._uploaded_media(req.media)
        if req.rich_message is not None:
            rich = chan._rich(req.rich_message)
            if m.rich_message is not None and bytes(m.rich_message) == bytes(rich):
                raise rpc("MessageNotModifiedError")
            m.message, m.entities, m.rich_message, m.rich_input = "", None, rich, req.rich_message
            if not chan.user_edit_keeps_markup:
                chan._set_markup(m, None)
            if chan.user_edit_frees_message:
                m.via_other_bot = False
            chan.user_edits.append(req)
            chan._pts += 1
            return chan._updates([])
        text = m.message if req.message is None else req.message
        ents = list(m.entities or []) if req.message is None else list(req.entities or [])
        if media is None and text == m.message and [_bytes(e) for e in ents] == [_bytes(e) for e in (m.entities or [])]:
            raise rpc("MessageNotModifiedError")
        m.message, m.entities = text, (ents or None)
        if media is not None:
            m.media = media
            chan._remember_media(m)
        if not chan.user_edit_keeps_markup:
            chan._set_markup(m, None)  # no keyboard sent: Telegram drops it
        if chan.user_edit_frees_message:
            m.via_other_bot = False
        chan.user_edits.append(req)
        chan._pts += 1
        return chan._updates([])

    async def __call__(self, req):
        bytes(req)
        self.requests.append(req)
        name = type(req).__name__
        chan = self.chan
        err = chan.fail.get("user:" + name)
        if err is not None:
            raise err
        uid = self.me.id
        if name == "ImportChatInviteRequest":
            inv = chan.invites.get(req.hash)
            if inv is None:
                raise rpc("InviteHashInvalidError")
            if inv["revoked"] or (inv["limit"] and inv["uses"] >= inv["limit"]):
                raise rpc("InviteHashExpiredError")
            if uid in chan.members:
                raise rpc("UserAlreadyParticipantError")
            if chan.require_approval:
                raise rpc("InviteRequestSentError")
            inv["uses"] += 1
            self._join()
            return types.Updates(updates=[], users=[], chats=[chan._channel_obj()], date=NOW, seq=0)
        if name == "CheckChatInviteRequest":
            if uid in chan.members:
                return types.ChatInviteAlready(chat=chan._channel_obj())
            return types.ChatInvite(title="Test", photo=types.PhotoEmpty(0), participants_count=1, color=0)
        if name == "JoinChannelRequest":
            if uid in chan.members:
                raise rpc("UserAlreadyParticipantError")
            self._join()
            return types.Updates(updates=[], users=[], chats=[chan._channel_obj()], date=NOW, seq=0)
        if name == "GetParticipantRequest":
            if uid not in chan.members:
                raise rpc("UserNotParticipantError")
            if uid in chan.admins:
                return SimpleNamespace(
                    participant=types.ChannelParticipantAdmin(user_id=uid, promoted_by=1, date=NOW, admin_rights=chan.admins[uid])
                )
            return SimpleNamespace(participant=types.ChannelParticipant(user_id=uid, date=NOW))
        if name == "EditMessageRequest":
            return self._edit(req)
        if name == "DeleteMessagesRequest":
            ids = list(req.id)
            self.deleted.append(ids)
            adm = chan.admins.get(uid)
            if adm is None or not adm.delete_messages:
                raise rpc("ChatAdminRequiredError")
            n = 0
            for i in ids:
                if i in chan.msgs and i not in chan.cant_delete_at_all:
                    del chan.msgs[i]
                    n += 1
            return types.messages.AffectedMessages(pts=chan._top, pts_count=n)
        raise NotImplementedError(name)