Automated sync from GitHub

app/common.py

@@ -18,7 +18,7 @@ from .config import Config
 from .db import Database
 from .replace_engine import fetch_messages, is_skippable
 from .tgutil import build_markup, de_entities, edit_raw, esc, explain_rpc, fetch_media, media_info, peer_of, with_media
-from .userbot import UserbotError, edit_problem_text
+from .userbot import UserbotError, UserbotRefused, edit_problem_text
 
 log = logging.getLogger(__name__)
 
@@ -248,30 +248,53 @@ async def live_media(client, ch, message_id: int) -> Optional[tuple]:
 async def edit_by_userbot(ctx: Ctx, ch, p, *, replace_media: bool, bot_error: str, notify=None) -> Optional[EditReport]:
     """The userbot makes the live post look like `p`. None when there is no userbot to ask; UserbotError when it fails.
 
-    The new photo / file is downloaded by the bot (the userbot can't see the owner's chat with the bot) and uploaded again."""
+    A new photo / file goes through the userbot's own chat with the bot: the bot sends it there and the userbot edits the
+    post with that copy, so nothing is downloaded or uploaded - a big file is as quick as a small one and keeps its
+    thumbnail. Only when that does not work does the bot download the file (with its thumbnail) and the userbot upload
+    it again, which is slow for a big file. `notify(text)` tells the owner what is going on."""
     ub = ctx.userbot
     editor = await ub.editor_for(ch) if ub is not None else None
     if editor is None:
         return None
     report = EditReport(by_userbot=True, bot_error=bot_error)
     folder = None
-    try:
+
+    async def tell(text: str) -> None:
         if notify is not None:
-            await notify()
-        media = None
-        if replace_media and p.media_file_id:
-            folder = tempfile.mkdtemp(prefix="cpb-media-")
-            try:
-                media = await fetch_media(ctx.client, p.media_file_id, p.media_kind, folder)
-            except (errors.RPCError, OSError, ValueError) as e:
-                raise UserbotError(f"I could not download the new media for the userbot ({type(e).__name__}).")
-        await editor.edit(
-            p.message_id,
-            text=p.text or "",
-            entities=de_entities(p.entities),
-            media=media,
-            preview=None if p.media_kind else bool(p.link_preview),
-        )
+            await notify(text)
+
+    fields = dict(
+        text=p.text or "", entities=de_entities(p.entities), preview=None if p.media_kind else bool(p.link_preview)
+    )
+    try:
+        await tell("⏳ Telegram refused the bot, so the userbot is making the edit.")
+        if not (replace_media and p.media_file_id):
+            await editor.edit(p.message_id, **fields)
+        else:
+            done = False
+            media = await editor.media_by_reference(p.media_file_id)
+            if media is not None:
+                try:
+                    await editor.edit(p.message_id, media=media, **fields)
+                    done = True
+                    editor.notes.append("the new file went through the userbot's chat with the bot, so nothing was uploaded")
+                except UserbotRefused as e:
+                    if not e.about_media:
+                        raise
+                    editor.dm_problem = f"Telegram would not take the userbot's copy of the file ({e.name})"
+            if not done:
+                why = editor.dm_problem or "that did not work"
+                editor.notes.append(f"the file was downloaded and uploaded again, because {why}")
+                await tell(
+                    "⏳ The file can't go through the userbot's chat with the bot, so it is downloaded and uploaded "
+                    f"again - a big file takes a while. (Reason: {esc(why)}.)"
+                )
+                folder = tempfile.mkdtemp(prefix="cpb-media-")
+                try:
+                    media = await fetch_media(ctx.client, p.media_file_id, p.media_kind, folder)
+                except (errors.RPCError, OSError, ValueError) as e:
+                    raise UserbotError(f"I could not download the new media for the userbot ({type(e).__name__}).")
+                await editor.edit(p.message_id, media=media, **fields)
     finally:
         await editor.close()
         report.notes = list(editor.notes)

app/handlers/basic.py

@@ -126,13 +126,19 @@ def register(ctx: Ctx) -> None:
                 "if it has the “Add new admins” right, and gives the rights back afterwards.\n\n"
                 "To set it up: on your own computer run <code>python make_userbot_session.py</code> with a "
                 "<b>separate</b> Telegram account, and put the printed text into the bot's secrets as "
-                "<code>USERBOT_SESSION</code>. See the README.",
+                "<code>USERBOT_SESSION</code>. Then send /start to this bot from that account (new media reaches the "
+                "userbot through its chat with the bot - that is what makes a replace quick). See the README.",
             )
             return
         if not ub.ready:
             await ub.connect()
-        kept = "stays an admin" if ub.keep_admin else "is made admin only while deleting"
-        await say(event, f"🤖 <b>Userbot:</b> {esc(ub.describe())}\nIt {kept} (USERBOT_KEEP_ADMIN).")
+        kept = "stays an admin" if ub.keep_admin else "is made admin only while it deletes or edits"
+        await say(
+            event,
+            f"🤖 <b>Userbot:</b> {esc(ub.describe())}\nIt {kept} (USERBOT_KEEP_ADMIN).\n"
+            "It must have started this bot (/start from that account): new media for an edit goes through its chat "
+            "with the bot, which is what makes a big file as quick as a small one.",
+        )
 
     @client.on(
         events.NewMessage(incoming=True, func=lambda e: e.is_private and not (e.raw_text or "").startswith("/"))

app/handlers/posts.py

@@ -160,8 +160,8 @@ def register(ctx: Ctx) -> None:
         post, ch = await load_post(ctx, st["pid"])
         check_lengths(post.text or "", True)
 
-        async def notify() -> None:
-            await say(event, "⏳ Telegram refused the bot, so the userbot is making the edit - a big file takes a while.")
+        async def notify(text: str) -> None:
+            await say(event, text)
 
         post, report = await apply_edit_reported(
             ctx, post, ch, replace_media=True, notify=notify, media_kind=kind, media_file_id=pack_file_id(m)

app/handlers/replace.py

@@ -268,8 +268,8 @@ def register(ctx: Ctx) -> None:
             caption, entities, how = message_text(reply, cfg.markup_input)  # Telegram's formatting, or HTML / Markdown
             check_lengths(caption, True)
 
-            async def notify() -> None:
-                await say(event, "⏳ Telegram refused the bot, so the userbot is making the edit - a big file takes a while.")
+            async def notify(text: str) -> None:
+                await say(event, text)
 
             _, report = await apply_edit_reported(
                 ctx, post, channel, replace_media=True, notify=notify, media_kind=kind, media_file_id=pack_file_id(reply),

app/tgutil.py

@@ -296,6 +296,15 @@ def media_info(m) -> tuple:
     return kind, (media_ref(m) if kind else None)
 
 
+def media_key(media) -> Optional[tuple]:
+    """("photo" | "doc", id) of the photo / file in a message's media; the id is the same on every account."""
+    if isinstance(media, types.MessageMediaPhoto) and getattr(media, "photo", None):
+        return "photo", media.photo.id
+    if isinstance(media, types.MessageMediaDocument) and getattr(media, "document", None):
+        return "doc", media.document.id
+    return None
+
+
 @dataclass
 class MediaFile:
     """A photo or file downloaded from Telegram, with what is needed to upload it again as the same kind of post."""
@@ -305,10 +314,13 @@ class MediaFile:
     name: str
     mime_type: Optional[str] = None
     attributes: list = field(default_factory=list)  # the file's name, video size and length, GIF mark ... as it had them
+    thumb_path: Optional[str] = None  # the picture Telegram shows for the file (JPEG); None: it has none
 
 
 async def fetch_media(client, ref: str, kind: str, folder: str) -> MediaFile:
-    """Download the photo / file behind the stored media reference `ref` (as `client` sees it) into `folder`."""
+    """Download the photo / file behind the stored media reference `ref` (as `client` sees it) into `folder`.
+
+    A file's thumbnail comes along: without it the file would be uploaded again with a plain file icon."""
     path = await with_media(ref, lambda media: client.download_media(media, file=folder + os.sep))
     if not path:
         raise ValueError("Telegram gave me nothing to download for that media")
@@ -319,9 +331,24 @@ async def fetch_media(client, ref: str, kind: str, folder: str) -> MediaFile:
         name=os.path.basename(str(path)),
         mime_type=getattr(doc, "mime_type", None),
         attributes=list(getattr(doc, "attributes", None) or []),
+        thumb_path=await fetch_thumb(client, ref, folder) if doc is not None else None,
     )
 
 
+async def fetch_thumb(client, ref: str, folder: str) -> Optional[str]:
+    """The thumbnail of the file behind `ref` (the largest one Telegram has) as a JPEG in `folder`, or None.
+
+    Best effort: a file that goes out without its thumbnail is better than an edit that fails."""
+    where = os.path.join(folder, "thumb")
+    try:
+        os.makedirs(where, exist_ok=True)
+        path = await with_media(ref, lambda media: client.download_media(media, file=where + os.sep, thumb=-1))
+    except (errors.RPCError, OSError, ValueError, TypeError) as e:
+        log.info("could not download the thumbnail: %s", type(e).__name__)
+        return None
+    return str(path) if path and os.path.isfile(str(path)) else None
+
+
 def utf16_len(s: str) -> int:
     return len(s.encode("utf-16-le")) // 2
 

app/userbot.py

@@ -6,6 +6,10 @@ Editing: Telegram refuses the bot an edit of a post that belongs to another bot
 or it was posted through it). The bot always tries first; when Telegram refuses, the userbot makes the edit (text,
 caption or a new photo / file). A user account can't set inline buttons, so Telegram may drop them with that edit -
 the bot puts them back where it can.
+A new photo / file goes to the userbot through the userbot's own chat with the bot: the bot sends it there (Telegram
+just points to the file it already has - nothing is uploaded), the userbot finds it in that chat and edits the post
+with that copy, again by reference. So a big file is as quick as a small one and keeps its thumbnail. Only when that
+chat does not work (the userbot account never started the bot) is the file downloaded and uploaded again.
 
 How it gets into the channel:
   1. the bot makes a one-time invite link (or reads the channel's own link) and the userbot joins with it
@@ -27,7 +31,7 @@ from typing import Optional
 from telethon import TelegramClient, errors, functions, types, utils
 from telethon.sessions import StringSession
 
-from .tgutil import MediaFile, edit_raw, flood_retry, get_rights, peer_of
+from .tgutil import MediaFile, edit_raw, flood_retry, get_rights, media_from_ref, media_key, peer_of, with_media
 
 log = logging.getLogger(__name__)
 
@@ -41,6 +45,32 @@ class UserbotError(Exception):
     """Something the owner can fix; the text is meant to be shown."""
 
 
+class UserbotRefused(UserbotError):
+    """Telegram refused what the userbot tried; `name` is the Telegram error (e.g. MediaInvalidError)."""
+
+    def __init__(self, text: str, name: str = ""):
+        super().__init__(text)
+        self.name = name
+
+    @property
+    def about_media(self) -> bool:
+        """The refusal is about the photo / file the userbot used (so trying it another way may help)."""
+        return self.name in _MEDIA_REFUSED
+
+
+# What Telegram answers when the photo / file the userbot points to can't be used for an edit
+_MEDIA_REFUSED = {
+    "MediaInvalidError", "MediaEmptyError", "DocumentInvalidError", "PhotoInvalidError",
+    "FileReferenceExpiredError", "FileReferenceInvalidError", "FileReferenceEmptyError",
+}
+# What Telegram answers when the bot may not write to the userbot account (it never started the bot, or blocked it)
+_NOT_STARTED = {
+    "UserIsBlockedError", "PeerIdInvalidError", "InputUserDeactivatedError", "UserDeactivatedError", "UserIdInvalidError",
+}
+_DM_POLLS = 10  # how often, and how long apart, the userbot looks in its chat with the bot for the file it was sent
+_DM_WAIT = 0.5
+
+
 def parse_invite_hash(link: str) -> Optional[str]:
     m = _INVITE_RX.search(link or "")
     return (m.group(1) or m.group(2)) if m else None
@@ -479,32 +509,127 @@ class UserbotEditor(_ChannelJob):
     WHAT = "edit other people's posts"
     VERB = "edit"
 
-    async def edit(self, message_id: int, *, text: str, entities: list, media: Optional[MediaFile] = None, preview=None) -> None:
-        """Make the post show `text` (with `entities`) and, when given, the photo / file `media` (uploaded again)."""
+    def __init__(self, ub: Userbot, ch):
+        super().__init__(ub, ch)
+        self._handed: list = []  # (the userbot account as the bot addresses it, the bot's id of the message): files put in its chat
+        self.dm_problem: Optional[str] = None  # why a file could not go through that chat (None: it did, or it was not tried)
+
+    async def close(self) -> None:
+        """Clean up the channel (rights, link) and the files the bot put into the userbot's chat with the bot."""
+        handed, self._handed = self._handed, []
+        for peer, message_id in handed:
+            try:
+                await self.ub.bot.delete_messages(peer, [message_id])
+            except Exception as e:  # noqa: BLE001 - at worst a message is left in the helper account's chat
+                log.info("could not delete the helper message in the userbot's chat: %s", type(e).__name__)
+        await super().close()
+
+    async def edit(self, message_id: int, *, text: str, entities: list, media=None, preview=None) -> None:
+        """Make the post show `text` (with `entities`) and, when given, `media`: what media_by_reference() returned (the
+        file does not move at all) or a MediaFile (uploaded again)."""
         await self._begin()
         client = self.ub.client
         try:
-            input_media = await self._input_media(media) if media is not None else None
+            if isinstance(media, MediaFile):
+                media = await self._upload(media)
             await flood_retry(
-                lambda: edit_raw(
-                    client, self.peer, message_id, text=text, entities=entities, preview=preview, media=input_media
-                )
+                lambda: edit_raw(client, self.peer, message_id, text=text, entities=entities, preview=preview, media=media)
             )
         except errors.MessageNotModifiedError:
             return  # it already looks like that
         except errors.RPCError as e:
             name = type(e).__name__
-            raise UserbotError(f"Telegram refused the userbot as well ({name}). {_EDIT_HINTS.get(name) or str(e)[:200]}")
+            raise UserbotRefused(
+                f"Telegram refused the userbot as well ({name}). {_EDIT_HINTS.get(name) or str(e)[:200]}", name
+            )
+
+    # ----------------------------------------------------------- a file by way of the userbot's chat with the bot
+    async def media_by_reference(self, ref: str):
+        """The new photo / file as the userbot's own account holds it, ready to edit the post with - or None.
 
-    async def _input_media(self, f: MediaFile):
+        The bot sends it to the userbot's chat with the bot (Telegram reuses the file it already has, nothing is
+        downloaded or uploaded) and the userbot picks the message up there. The access data of a file belongs to one
+        account, so the bot's own reference would be useless to the userbot: this is how the userbot gets its own.
+        None when it does not work; `dm_problem` says why, and the caller uploads the file instead."""
+        await self._begin()
+        self.dm_problem = None
+        try:
+            return await self._hand_over(ref)
+        except (errors.RPCError, ValueError, TypeError, OSError) as e:
+            self.dm_problem = f"it failed with {type(e).__name__}"
+            log.info("could not hand the file to the userbot: %s", type(e).__name__, exc_info=True)
+            return None
+
+    async def _hand_over(self, ref: str):
+        bot, client = self.ub.bot, self.ub.client
+        want = media_key(media_from_ref(ref))
+        if want is None:
+            raise ValueError("not a photo or a file")
+        try:  # both ends of the chat first: nothing is sent when one of them can't be found
+            user = await self._bot_view_of_userbot()
+        except UserbotError:
+            self.dm_problem = "the bot cannot find the userbot account"
+            return None
+        try:
+            bot_peer = await self._userbot_view_of_bot()
+        except UserbotError as e:
+            self.dm_problem = str(e)
+            return None
+        peer = types.InputPeerUser(user.user_id, user.access_hash)
+        try:
+            sent = await with_media(ref, lambda media: flood_retry(lambda: bot.send_file(peer, media, silent=True)))
+        except errors.RPCError as e:
+            name = type(e).__name__
+            self.dm_problem = (
+                "the userbot account has not started the bot yet (send /start to the bot from that account)"
+                if name in _NOT_STARTED
+                else f"Telegram would not let the bot send the file to the userbot ({name})"
+            )
+            return None
+        self._handed.append((peer, sent.id))  # deleted again in close()
+        for i in range(_DM_POLLS):  # the message has to reach the userbot's side first
+            if i:
+                await asyncio.sleep(_DM_WAIT)
+            try:
+                msgs = await flood_retry(lambda: client.get_messages(bot_peer, limit=10))
+            except errors.RPCError as e:
+                self.dm_problem = f"the userbot could not read its chat with the bot ({type(e).__name__})"
+                return None
+            for m in msgs:  # newest first; the same photo / file has the same id on both accounts
+                if media_key(getattr(m, "media", None)) == want:
+                    return utils.get_input_media(m.media)
+        self.dm_problem = "the file did not show up in the userbot's chat with the bot"
+        return None
+
+    async def _userbot_view_of_bot(self):
+        """The bot as the userbot account addresses it (found by the bot's username)."""
+        me = await self.ub.bot.get_me()
+        name = getattr(me, "username", None)
+        if not name:
+            raise UserbotError("the bot has no username, so the userbot cannot find its chat with it")
+        try:
+            return await self.ub.client.get_input_entity(name)
+        except (errors.RPCError, ValueError):
+            raise UserbotError("the userbot could not find the bot by its username")
+
+    # --------------------------------------------------------------------------------- the slow way
+    async def _upload(self, f: MediaFile):
+        """Upload a downloaded photo / file (with its thumbnail) as the new media of the post."""
         client = self.ub.client
         handle = await flood_retry(lambda: client.upload_file(f.path, file_name=f.name))
         if f.kind == "photo":
             return types.InputMediaUploadedPhoto(file=handle)
+        thumb = None
+        if f.thumb_path:
+            try:
+                thumb = await flood_retry(lambda: client.upload_file(f.thumb_path))
+            except (errors.RPCError, OSError) as e:  # the file goes without it rather than not at all
+                log.info("could not upload the thumbnail: %s", type(e).__name__)
         return types.InputMediaUploadedDocument(
             file=handle,
             mime_type=f.mime_type or "application/octet-stream",
             attributes=list(f.attributes),
+            thumb=thumb,
             force_file=True if f.kind == "document" else None,  # a file sent "as a file" stays one
         )
 

tests/fakes.py

@@ -1,12 +1,13 @@
 """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
+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):
@@ -81,11 +82,52 @@ def photo_media(pid=11):
     return types.MessageMediaPhoto(photo=photo)
 
 
-def video_media(did=21):
-    doc = types.Document(did, did * 7, b"fr%d" % did, NOW, "video/mp4", 1000, 2, [types.DocumentAttributeVideo(5.0, 320, 240)])
+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())
 
@@ -150,6 +192,7 @@ class FakeChannelClient:
         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
 
@@ -164,6 +207,7 @@ class FakeChannelClient:
         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:
@@ -191,7 +235,18 @@ class FakeChannelClient:
             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):
+    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)
@@ -325,6 +380,47 @@ class FakeChannelClient:
             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.
@@ -564,6 +660,7 @@ class FakeUserClient:
         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
@@ -588,8 +685,26 @@ class FakeUserClient:
             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())
@@ -619,10 +734,27 @@ class FakeUserClient:
             return types.MessageMediaPhoto(photo=photo)
         if isinstance(inp, types.InputMediaUploadedDocument):
             chan._media_data[("doc", n)] = data
-            doc = types.Document(n, n * 7, b"up%d" % n, NOW, inp.mime_type, len(data), 2, list(inp.attributes))
+            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
@@ -634,7 +766,12 @@ class FakeUserClient:
         m = chan.msgs.get(req.id)
         if m is None:
             raise rpc("MessageIdInvalidError")
-        media = self._uploaded_media(req.media) if req.media is not None else None
+        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)
         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 [])]:

tests/harness.py

@@ -131,6 +131,8 @@ class FakeTG(FakeChannelClient):
         return m
 
     async def send_file(self, peer, file, **kw):
+        if self._is_dm(peer):  # to the userbot account: the chat simulation
+            return await super().send_file(peer, file, **kw)
         m = SimpleNamespace(id=next(self.ids))
         self.sent.append(("file", peer, file, kw))
         return m
@@ -146,6 +148,8 @@ class FakeTG(FakeChannelClient):
             raise rpc("MessageIdInvalidError")
 
     async def delete_messages(self, peer, ids, **kw):
+        if self._is_dm(peer):
+            return await super().delete_messages(peer, ids, **kw)
         self.deleted.append(list(ids))
         return [SimpleNamespace(pts_count=len(ids))]
 

tests/test_replace_userbot.py

@@ -4,11 +4,12 @@ import asyncio
 import pytest
 from telethon import types
 
+import app.userbot as userbot_module
 from app.common import UserError, apply_edit_reported
 from app.tgutil import media_info, pack_file_id
 from app.userbot import Userbot
 
-from .fakes import FakeChannelClient, FakeUserClient, make_msg, markup, photo_media, rpc, url_btn
+from .fakes import FakeChannelClient, FakeUserClient, make_msg, markup, photo_media, rpc, url_btn, video_media
 from .harness import App, norm
 
 CID, MID = 1723201445, 9084
@@ -39,12 +40,25 @@ def replacement(caption="new caption", entities=None):
     return make_msg(1, caption, entities, media=photo_media(20))
 
 
+def video_replacement(caption="new caption", did=31, thumb=True):
+    return make_msg(1, caption, None, media=video_media(did, thumb=thumb))
+
+
 def said(app):
     return [norm(t) for kind, t, _ in app.out.log if kind == "say"]
 
 
-async def replace(app, **kw):
-    await app.text(f"/replace https://t.me/c/{CID}/{MID}", reply=replacement(**kw))
+def waiting(app):
+    """The "⏳ ..." messages the owner got while the userbot worked."""
+    return [t for t in said(app) if t.startswith("⏳")]
+
+
+def user_edits(user):
+    return [r for r in user.requests if type(r).__name__ == "EditMessageRequest"]
+
+
+async def replace(app, message=None, **kw):
+    await app.text(f"/replace https://t.me/c/{CID}/{MID}", reply=message or replacement(**kw))
 
 
 async def the_post(app):
@@ -57,17 +71,20 @@ def test_when_telegram_refuses_the_bot_the_userbot_replaces_media_and_caption(tm
         app, chan, user = await make_app(tmp_path)
         await replace(app)
         m = chan.msgs[MID]
-        assert m.message == "new caption" and m.media.photo.id != 10 and m.media.photo.id != 20
-        assert chan._media_data[("photo", m.media.photo.id)] == b"NEW PHOTO"  # the bot downloaded it, the userbot uploaded it
-        assert len(user.uploads) == 1 and user.uploads[0][2] == len(b"NEW PHOTO")
+        assert m.message == "new caption" and m.media.photo.id == 20  # the very file the owner sent: nothing was copied
+        # the file went bot -> the userbot's chat with the bot -> the post: nothing downloaded, nothing uploaded
+        assert user.uploads == [] and len(chan._dm.sent) == 1 and len(user_edits(user)) == 1
+        assert chan._dm.msgs == [] and len(chan._dm.deleted) == 1  # the helper message is gone again
         # the post in My posts shows what the channel shows now, and keeps its saved buttons
         post = await the_post(app)
-        assert post.text == "new caption" and post.buttons == BUTTONS
-        assert post.media_file_id == media_info(m)[1] != pack_file_id(replacement())
-        # the owner is told what happened, in order
-        first, last = said(app)[0], said(app)[-1]
-        assert "Telegram refused the bot, so the userbot is making the edit" in first
+        assert post.text == "new caption" and post.buttons == BUTTONS and post.media_kind == "photo"
+        assert post.media_file_id == media_info(m)[1]
+        # the owner is told what happened, in order - one waiting message, no "big file" warning
+        assert len(waiting(app)) == 1 and "Telegram refused the bot, so the userbot is making the edit" in waiting(app)[0]
+        assert "big file" not in waiting(app)[0]
+        last = said(app)[-1]
         assert "Replaced the media in Database Channel" in last and "Telegram refused the bot (MessageIdInvalidError)" in last
+        assert "the new file went through the userbot's chat with the bot, so nothing was uploaded" in last
         # the helper left things as it found them
         assert chan.admins == {} and all(i["revoked"] for i in chan.invites.values()) and not app.ctx.userbot.lock.locked()
         await app.db.close()
@@ -119,6 +136,7 @@ def test_the_userbot_is_not_asked_when_the_bot_can_do_it(tmp_path):
         app, chan, user = await make_app(tmp_path, foreign=False)
         await replace(app)
         assert user.requests == [] and not user.connected  # nobody else was bothered
+        assert chan._dm.sent == [] and waiting(app) == []
         assert app.tg.edits[-1][1] == MID and "Replaced the media" in said(app)[-1]
         assert "userbot" not in said(app)[-1]
         await app.db.close()
@@ -213,7 +231,7 @@ def test_a_caption_typed_in_markdown_is_edited_by_the_userbot_when_the_bot_is_re
         m = chan.msgs[MID]
         assert m.message == "new caption" and [(type(e).__name__, e.offset, e.length) for e in m.entities] == [("MessageEntityBold", 0, 3)]
         assert m.media.photo.id == 10  # only the caption changed, the photo stays
-        assert user.uploads == []
+        assert user.uploads == [] and chan._dm.sent == []  # no file moved, the chat was not used
         note = norm(app.out.last_text)
         assert "Formatting read from Markdown" in note and "the userbot made the edit" in note
         saved = await app.db.get_post(post.id)
@@ -223,7 +241,7 @@ def test_a_caption_typed_in_markdown_is_edited_by_the_userbot_when_the_bot_is_re
     asyncio.run(run())
 
 
-def test_new_media_sent_to_the_post_panel_is_uploaded_by_the_userbot(tmp_path):
+def test_new_media_sent_to_the_post_panel_reaches_the_userbot_through_its_chat(tmp_path):
     async def run():
         app, chan, user = await make_app(tmp_path)
         post = await app.db.create_post(
@@ -233,9 +251,8 @@ def test_new_media_sent_to_the_post_panel_is_uploaded_by_the_userbot(tmp_path):
         await app.press(f"em:{post.id}")
         await app.text("", message=replacement(caption=""))
         m = chan.msgs[MID]
-        assert m.message == "old caption" and m.media.photo.id not in (10, 20)
-        assert chan._media_data[("photo", m.media.photo.id)] == b"NEW PHOTO"
-        assert any("Telegram refused the bot, so the userbot is making the edit" in t for t in said(app))
+        assert m.message == "old caption" and m.media.photo.id == 20 and user.uploads == []
+        assert len(waiting(app)) == 1 and "Telegram refused the bot, so the userbot is making the edit" in waiting(app)[0]
         assert "the userbot made the edit" in norm(app.out.last_text)
         assert (await app.db.get_post(post.id)).media_file_id == media_info(m)[1]
         await app.db.close()
@@ -243,6 +260,179 @@ def test_new_media_sent_to_the_post_panel_is_uploaded_by_the_userbot(tmp_path):
     asyncio.run(run())
 
 
+# ------------------------------------------------------------------------------- the thumbnail stays
+def test_a_video_keeps_its_thumbnail_when_it_goes_through_the_chat(tmp_path):
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        await replace(app, message=video_replacement())
+        m = chan.msgs[MID]
+        assert m.media.document.id == 31 and m.media.document.thumbs  # the same file: its thumbnail comes with it
+        assert user.uploads == [] and (await the_post(app)).media_kind == "video"
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_video_keeps_its_thumbnail_when_it_has_to_be_uploaded_again(tmp_path):
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        chan._dm.error = rpc("UserIsBlockedError")  # the chat does not work: the slow way
+        chan._media_data[("thumb", 31)] = b"THE THUMBNAIL"
+        await replace(app, message=video_replacement())
+        doc = chan.msgs[MID].media.document
+        assert doc.id != 31 and doc.thumbs  # a new upload, but not a bare file icon
+        assert chan._media_data[("thumb", doc.id)] == b"THE THUMBNAIL"
+        assert [u[2] for u in user.uploads] == [len(b"bytes of doc 31"), len(b"THE THUMBNAIL")]  # the file, then its picture
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_file_without_a_thumbnail_is_uploaded_without_one(tmp_path):
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        chan._dm.error = rpc("UserIsBlockedError")
+        await replace(app, message=video_replacement(thumb=False))
+        assert chan.msgs[MID].media.document.thumbs is None and len(user.uploads) == 1
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+# ----------------------------------------------------------------------------- when the chat does not work
+def test_when_the_userbot_never_started_the_bot_the_file_is_downloaded_and_uploaded_again(tmp_path):
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        chan._dm.error = rpc("UserIsBlockedError")
+        await replace(app)
+        m = chan.msgs[MID]
+        assert m.message == "new caption" and m.media.photo.id not in (10, 20)  # a new upload
+        assert chan._media_data[("photo", m.media.photo.id)] == b"NEW PHOTO"  # the bot downloaded it, the userbot uploaded it
+        assert len(user.uploads) == 1 and user.uploads[0][2] == len(b"NEW PHOTO") and chan._dm.sent == []
+        # the owner is told, twice while waiting, why it is the slow way and what fixes that
+        first, second = waiting(app)
+        assert "Telegram refused the bot, so the userbot is making the edit" in first
+        assert "downloaded and uploaded again" in second and "a big file takes a while" in second
+        assert "has not started the bot" in second and "/start" in second
+        last = said(app)[-1]
+        assert "Replaced the media" in last and "the file was downloaded and uploaded again, because the userbot account has not started the bot" in last
+        post = await the_post(app)
+        assert post.media_file_id == media_info(m)[1] and post.text == "new caption"
+        assert chan.admins == {} and not app.ctx.userbot.lock.locked()
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_refusal_about_the_userbots_copy_of_the_file_falls_back_to_the_upload(tmp_path):
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        chan._dm.reference_error = rpc("FileReferenceExpiredError")  # Telegram will not take the userbot's copy
+        await replace(app)
+        m = chan.msgs[MID]
+        assert m.message == "new caption" and m.media.photo.id not in (10, 20) and len(user.uploads) == 1
+        assert len(user_edits(user)) == 2  # the edit with the copy was refused, the one with the upload was not
+        assert chan._dm.msgs == [] and len(chan._dm.deleted) == 1  # the helper message was cleaned up all the same
+        second = waiting(app)[1]
+        assert "Telegram would not take the userbot's copy of the file (FileReferenceExpiredError)" in second
+        assert "Telegram would not take the userbot's copy of the file" in said(app)[-1]
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_refusal_that_is_not_about_the_file_is_not_answered_with_a_long_upload(tmp_path):
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        chan.fail["user:EditMessageRequest"] = rpc("MessageIdInvalidError")  # the post itself can't be edited
+        await replace(app)
+        text = said(app)[-1]
+        assert "The userbot could not do it either" in text and "Telegram refused the userbot as well (MessageIdInvalidError)" in text
+        assert user.uploads == [] and len(user_edits(user)) == 1  # trying again with an upload would fail the same way
+        assert len(waiting(app)) == 1 and "Nothing was changed" in text
+        assert chan._dm.msgs == [] and len(chan._dm.deleted) == 1 and chan.admins == {}
+        assert chan.msgs[MID].message == "old caption" and (await the_post(app)).text == "old caption"
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_when_the_upload_fails_as_well_the_owner_hears_why_and_nothing_changes(tmp_path):
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        chan._dm.error = rpc("UserIsBlockedError")
+        chan.fail["user:EditMessageRequest"] = rpc("MediaInvalidError")
+        await replace(app)
+        text = said(app)[-1]
+        assert "The userbot could not do it either" in text and "MediaInvalidError" in text and "Nothing was changed" in text
+        assert chan.msgs[MID].media.photo.id == 10 and (await the_post(app)).text == "old caption"
+        assert chan.admins == {} and not app.ctx.userbot.lock.locked()
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("edit_error", [None, "MessageIdInvalidError", "FileReferenceExpiredError", "MediaInvalidError"])
+@pytest.mark.parametrize("copy_error", [None, "FileReferenceExpiredError"])
+@pytest.mark.parametrize("send_error", [None, "UserIsBlockedError", "ChatWriteForbiddenError"])
+@pytest.mark.parametrize("reading", ["fine", "late", "unreadable"])
+def test_whatever_goes_wrong_the_channel_is_left_clean_and_the_owner_is_told(
+    tmp_path, monkeypatch, edit_error, copy_error, send_error, reading
+):
+    """Every combination of: the chat refusing the bot, the chat being slow or unreadable, Telegram refusing the userbot's
+    copy, Telegram refusing every edit. Whatever happens: no helper message, admin right or invite link is left behind, the
+    userbot's lock is free, and the post and My posts are either both changed or both untouched - and the owner is told."""
+    monkeypatch.setattr(userbot_module, "_DM_WAIT", 0)
+    monkeypatch.setattr(userbot_module, "_DM_POLLS", 3)
+
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        if edit_error:
+            chan.fail["user:EditMessageRequest"] = rpc(edit_error)
+        if copy_error:
+            chan._dm.reference_error = rpc(copy_error)
+        if send_error:
+            chan._dm.error = rpc(send_error)
+        if reading == "late":
+            chan._dm.late = 99
+        elif reading == "unreadable":
+            chan._dm.read_error = rpc("PeerIdInvalidError")
+        await replace(app)
+        last = said(app)[-1]
+        assert chan.admins == {} and all(i["revoked"] for i in chan.invites.values()) and not app.ctx.userbot.lock.locked()
+        assert chan._dm.msgs == []  # no helper message is left in the userbot's chat
+        post, live = await the_post(app), chan.msgs[MID]
+        if "Replaced the media" in last:  # it worked: the channel and My posts agree
+            assert live.message == "new caption" and live.media.photo.id != 10 and post.text == "new caption"
+            assert post.media_file_id == media_info(live)[1]
+            assert user.uploads or live.media.photo.id == 20  # a copy by reference, or an upload
+        else:  # it did not: nothing was changed anywhere, and the owner is told why
+            assert "Nothing was changed" in last and "The userbot could not do it either" in last
+            assert live.message == "old caption" and live.media.photo.id == 10 and post.text == "old caption"
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+# --------------------------------------------------------------------------------------------- /userbot
+def test_the_userbot_status_says_it_has_to_start_the_bot(tmp_path):
+    async def run():
+        app, chan, user = await make_app(tmp_path)
+        await app.text("/userbot")
+        text = said(app)[-1]
+        assert "connected as @helper_acct" in text and "made admin only while it deletes or edits" in text
+        assert "must have started this bot" in text and "/start" in text
+        (tmp_path / "off").mkdir()
+        off, _, _ = await make_app(tmp_path / "off", userbot=False)
+        await off.text("/userbot")
+        text = said(off)[-1]
+        assert "not set up" in text and "send /start to this bot from that account" in text
+        await app.db.close()
+        await off.db.close()
+
+    asyncio.run(run())
+
+
 # ------------------------------------------------------------- Telegram's own formatting, HTML, Markdown
 def sent_caption(app):
     kw = app.tg.edits[-1][2]

tests/test_userbot_dm.py

@@ -0,0 +1,417 @@
+"""A new photo / file reaches the userbot through its own chat with the bot: no download, no upload, the thumbnail stays.
+
+The bot sends the media to the userbot account (Telegram reuses the file it already has), the userbot finds it there and
+edits the post with ITS copy. When that chat does not work, the file is downloaded and uploaded again - with its thumbnail."""
+import os
+
+import pytest
+from telethon import TelegramClient, errors, types
+from telethon.sessions import MemorySession
+
+import app.userbot as userbot_module
+from app.tgutil import MediaFile, fetch_media, fetch_thumb, media_ref
+from app.userbot import UserbotError, UserbotRefused
+
+from .fakes import NOW, make_msg, photo_media, rpc, video_media
+from .test_userbot import FULL, channel, run
+from .test_userbot_edit import ready_editor
+
+
+@pytest.fixture(autouse=True)
+def quick_polling(monkeypatch):
+    monkeypatch.setattr(userbot_module, "_DM_WAIT", 0)  # the tests that wait for a message must not wait for real
+    monkeypatch.setattr(userbot_module, "_DM_POLLS", 4)
+
+
+def ref_of(media):
+    """The stored reference of a photo / file the owner sent to the bot."""
+    return media_ref(make_msg(1, "", media=media))
+
+
+# ------------------------------------------------------------------------------------------- the quick way
+def test_a_file_goes_through_the_userbots_chat_and_the_post_shows_that_same_file():
+    async def go():
+        bot = channel()
+        bot.msgs[5].media = photo_media(10)
+        new = video_media(31, thumb=True)
+        ub, made, e = await ready_editor(bot)
+        got = await e.media_by_reference(ref_of(new))
+        assert isinstance(got, types.InputMediaDocument) and got.id.id == 31 and e.dm_problem is None
+        # what the userbot presents is ITS copy: the same file, but with the access data of its own account
+        assert (got.id.access_hash, got.id.file_reference) == bot._dm.known[("doc", 31)]
+        assert (got.id.access_hash, got.id.file_reference) != (new.document.access_hash, new.document.file_reference)
+        await e.edit(5, text="cap", entities=[], media=got)
+        doc = bot.msgs[5].media.document
+        assert doc.id == 31 and doc.thumbs  # the very same file, so it keeps its thumbnail
+        assert made["client"].uploads == [] and bot.user_edits[-1].media is got and bot.user_edits[-1].message == "cap"
+        await e.close()
+
+    run(go())
+
+
+def test_a_photo_goes_through_the_chat_too():
+    async def go():
+        bot = channel()
+        ub, made, e = await ready_editor(bot)
+        got = await e.media_by_reference(ref_of(photo_media(20)))
+        assert isinstance(got, types.InputMediaPhoto) and got.id.id == 20
+        await e.edit(5, text="", entities=[], media=got)
+        assert bot.msgs[5].media.photo.id == 20 and made["client"].uploads == []
+        await e.close()
+
+    run(go())
+
+
+def test_the_helper_message_is_sent_quietly_and_deleted_at_the_end():
+    async def go():
+        bot = channel()
+        ub, made, e = await ready_editor(bot)
+        await e.edit(5, text="x", entities=[], media=await e.media_by_reference(ref_of(photo_media(20))))
+        (bot_id, kw), = bot._dm.sent
+        assert kw == {"silent": True}  # the account's phone does not ring for it
+        assert len(bot._dm.msgs) == 1  # still there while the job runs ...
+        await e.close()
+        assert bot._dm.deleted == [bot_id] and bot._dm.msgs == []  # ... and gone afterwards
+        assert bot.admins == {} and not ub.lock.locked()
+
+    run(go())
+
+
+def test_the_userbot_finds_its_file_among_the_other_messages_of_the_chat():
+    async def go():
+        bot = channel()
+        bot._dm.msgs += [make_msg(901, "hello", media=None), make_msg(902, "", media=photo_media(98)), make_msg(903, "", media=video_media(97))]
+        ub, made, e = await ready_editor(bot)
+        got = await e.media_by_reference(ref_of(video_media(31)))
+        assert got.id.id == 31  # picked by the id of the file, not by position
+        await e.close()
+        assert [m.id for m in bot._dm.msgs] == [901, 902, 903]  # the others were not touched
+
+    run(go())
+
+
+def test_a_file_that_reaches_the_userbot_late_is_waited_for():
+    async def go():
+        bot = channel()
+        bot._dm.late = 2  # the first two reads find nothing yet
+        ub, made, e = await ready_editor(bot)
+        got = await e.media_by_reference(ref_of(photo_media(20)))
+        assert got is not None and got.id.id == 20 and e.dm_problem is None
+        await e.close()
+
+    run(go())
+
+
+# ------------------------------------------------------------------------------- when the chat does not work
+def test_a_file_that_never_arrives_is_reported_and_the_helper_message_is_still_cleaned_up():
+    async def go():
+        bot = channel()
+        bot._dm.late = 99
+        ub, made, e = await ready_editor(bot)
+        assert await e.media_by_reference(ref_of(photo_media(20))) is None
+        assert e.dm_problem == "the file did not show up in the userbot's chat with the bot"
+        await e.close()
+        assert len(bot._dm.deleted) == 1 and bot._dm.msgs == []
+
+    run(go())
+
+
+@pytest.mark.parametrize("name", ["UserIsBlockedError", "PeerIdInvalidError", "InputUserDeactivatedError"])
+def test_an_account_that_never_started_the_bot_is_told_what_to_do(name):
+    async def go():
+        bot = channel()
+        bot._dm.error = rpc(name)
+        ub, made, e = await ready_editor(bot)
+        assert await e.media_by_reference(ref_of(photo_media(20))) is None
+        assert "has not started the bot" in e.dm_problem and "/start" in e.dm_problem
+        await e.close()
+        assert bot._dm.deleted == [] and bot._dm.sent == []  # nothing was sent, so nothing is cleaned up
+
+    run(go())
+
+
+def test_any_other_refusal_of_the_send_is_named():
+    async def go():
+        bot = channel()
+        bot._dm.error = rpc("ChatWriteForbiddenError")
+        ub, made, e = await ready_editor(bot)
+        assert await e.media_by_reference(ref_of(photo_media(20))) is None
+        assert e.dm_problem == "Telegram would not let the bot send the file to the userbot (ChatWriteForbiddenError)"
+        await e.close()
+
+    run(go())
+
+
+def test_a_chat_the_userbot_cannot_read_is_named():
+    async def go():
+        bot = channel()
+        bot._dm.read_error = rpc("PeerIdInvalidError")
+        ub, made, e = await ready_editor(bot)
+        assert await e.media_by_reference(ref_of(photo_media(20))) is None
+        assert e.dm_problem == "the userbot could not read its chat with the bot (PeerIdInvalidError)"
+        await e.close()
+        assert bot._dm.msgs == []  # the helper message is deleted all the same
+
+    run(go())
+
+
+def test_a_bot_without_a_username_cannot_be_found_and_nothing_is_sent():
+    async def go():
+        bot = channel()
+        bot._dm.bot_user.username = None
+        ub, made, e = await ready_editor(bot)
+        assert await e.media_by_reference(ref_of(photo_media(20))) is None
+        assert "no username" in e.dm_problem
+        await e.close()
+        assert bot._dm.sent == []
+
+    run(go())
+
+
+def test_a_bot_the_userbot_cannot_resolve_is_named_and_nothing_is_sent():
+    async def go():
+        bot = channel()
+        ub, made, e = await ready_editor(bot)
+
+        async def nothing(peer):
+            raise ValueError("Cannot find any entity")
+
+        made["client"].get_input_entity = nothing
+        assert await e.media_by_reference(ref_of(photo_media(20))) is None
+        assert e.dm_problem == "the userbot could not find the bot by its username"
+        await e.close()
+        assert bot._dm.sent == []
+
+    run(go())
+
+
+def test_a_userbot_account_the_bot_cannot_find_gets_nothing():
+    async def go():
+        bot = channel()
+        bot.participants_visible = False  # the bot can't list the members, and has never met the account
+        ub, made, e = await ready_editor(bot)
+        made["client"]._join()
+        bot.admins[777] = types.ChatAdminRights(edit_messages=True)  # it may edit already: no promotion, no lookup needed so far
+        assert await e.media_by_reference(ref_of(photo_media(20))) is None
+        assert e.dm_problem == "the bot cannot find the userbot account"
+        await e.close()
+        assert bot._dm.sent == []
+
+    run(go())
+
+
+def test_the_channel_comes_first_nothing_is_sent_when_the_userbot_cannot_get_in():
+    async def go():
+        bot = channel(rights=dict(FULL, invite=False))  # no invite link: the userbot can't join
+        ub, made, e = await ready_editor(bot)
+        with pytest.raises(UserbotError) as exc:
+            await e.media_by_reference(ref_of(photo_media(20)))
+        assert "“Edit messages of others”" in str(exc.value)
+        await e.close()
+        assert bot._dm.sent == []
+
+    run(go())
+
+
+# ------------------------------------------------------------------------------- when Telegram refuses the edit
+def test_a_refused_edit_is_named_and_the_helper_message_is_still_deleted():
+    async def go():
+        bot = channel()
+        bot.fail["user:EditMessageRequest"] = rpc("MessageIdInvalidError")
+        ub, made, e = await ready_editor(bot)
+        got = await e.media_by_reference(ref_of(photo_media(20)))
+        with pytest.raises(UserbotRefused) as exc:
+            await e.edit(5, text="x", entities=[], media=got)
+        assert exc.value.name == "MessageIdInvalidError" and not exc.value.about_media
+        assert "Telegram refused the userbot as well (MessageIdInvalidError)" in str(exc.value)
+        await e.close()
+        assert bot._dm.msgs == [] and bot.admins == {} and not ub.lock.locked()
+
+    run(go())
+
+
+@pytest.mark.parametrize(
+    "name,about_media",
+    [
+        ("FileReferenceExpiredError", True), ("FileReferenceInvalidError", True), ("MediaEmptyError", True),
+        ("MediaInvalidError", True), ("DocumentInvalidError", True), ("PhotoInvalidError", True),
+        ("MessageIdInvalidError", False), ("ChatAdminRequiredError", False), ("MessageAuthorRequiredError", False),
+    ],
+)
+def test_a_refusal_about_the_file_is_told_apart_from_the_others(name, about_media):
+    assert UserbotRefused("x", name).about_media is about_media
+
+
+def test_the_userbots_edit_with_somebody_elses_access_data_is_refused_by_the_simulation():
+    """Guards the simulation itself: the bot's own reference must NOT work for the userbot (that is why the chat exists)."""
+
+    async def go():
+        bot = channel()
+        new = photo_media(20)
+        ub, made, e = await ready_editor(bot)
+        await e.media_by_reference(ref_of(new))  # the userbot now holds its own copy ...
+        bots_own = types.InputMediaPhoto(id=types.InputPhoto(20, new.photo.access_hash, new.photo.file_reference))
+        with pytest.raises(UserbotRefused) as exc:  # ... but not the bot's
+            await e.edit(5, text="x", entities=[], media=bots_own)
+        assert exc.value.name == "FileReferenceExpiredError" and exc.value.about_media
+        unknown = types.InputMediaPhoto(id=types.InputPhoto(77, 1, b"x"))
+        with pytest.raises(UserbotRefused) as exc:
+            await e.edit(5, text="x", entities=[], media=unknown)
+        assert exc.value.name == "MediaEmptyError"
+        await e.close()
+
+    run(go())
+
+
+# ----------------------------------------------------------------------------------- the slow way keeps the thumbnail
+def test_an_uploaded_file_takes_its_thumbnail_along(tmp_path):
+    async def go():
+        bot = channel()
+        video, thumb = tmp_path / "Episode 12.mkv", tmp_path / "photo_1.jpg"
+        video.write_bytes(b"VIDEO BYTES")
+        thumb.write_bytes(b"THE THUMBNAIL")
+        attrs = [types.DocumentAttributeFilename("Episode 12.mkv"), types.DocumentAttributeVideo(5.0, 320, 240)]
+        ub, made, e = await ready_editor(bot)
+        await e.edit(5, text="", entities=[], media=MediaFile(str(video), "video", "Episode 12.mkv", "video/x-matroska", attrs, str(thumb)))
+        await e.close()
+        sent = bot.user_edits[-1].media
+        assert isinstance(sent, types.InputMediaUploadedDocument) and sent.thumb is not None
+        assert made["client"].uploads == [(str(video), "Episode 12.mkv", 11), (str(thumb), None, 13)]
+        doc = bot.msgs[5].media.document
+        assert doc.thumbs and bot._media_data[("thumb", doc.id)] == b"THE THUMBNAIL"
+
+    run(go())
+
+
+def test_a_thumbnail_that_cannot_be_uploaded_does_not_stop_the_edit(tmp_path):
+    async def go():
+        bot = channel()
+        video = tmp_path / "x.mkv"
+        video.write_bytes(b"V")
+        ub, made, e = await ready_editor(bot)
+        await e.edit(5, text="", entities=[], media=MediaFile(str(video), "video", "x.mkv", thumb_path=str(tmp_path / "gone.jpg")))
+        await e.close()
+        assert bot.user_edits[-1].media.thumb is None and bot.msgs[5].media.document.thumbs is None  # it went without
+
+    run(go())
+
+
+def test_a_file_without_a_thumbnail_is_uploaded_without_one(tmp_path):
+    async def go():
+        bot = channel()
+        video = tmp_path / "x.mkv"
+        video.write_bytes(b"V")
+        ub, made, e = await ready_editor(bot)
+        await e.edit(5, text="", entities=[], media=MediaFile(str(video), "document", "x.mkv"))
+        await e.close()
+        assert bot.user_edits[-1].media.thumb is None and len(made["client"].uploads) == 1
+
+    run(go())
+
+
+# ---------------------------------------------------------------------------------------- downloading the thumbnail
+def patch_download(monkeypatch, wanted_sizes):
+    """Telethon's real download_media runs; only the transfer of the bytes is replaced (it records what was asked for)."""
+
+    async def fake_download(self, input_location, file, **kw):
+        wanted_sizes.append(getattr(input_location, "thumb_size", None))
+        with open(file, "wb") as f:
+            f.write(b"THUMB" if getattr(input_location, "thumb_size", "") else b"THE BYTES")
+        return file
+
+    monkeypatch.setattr(TelegramClient, "_download_file", fake_download)
+
+
+def doc_ref(thumbs, attrs=None):
+    attrs = attrs or [types.DocumentAttributeFilename("Episode 12.mkv"), types.DocumentAttributeVideo(5.0, 320, 240)]
+    doc = types.Document(21, 7, b"fr", NOW, "video/x-matroska", 1000, 2, attrs, thumbs=thumbs)
+    return ref_of(types.MessageMediaDocument(document=doc))
+
+
+def test_the_largest_thumbnail_is_downloaded_with_the_file(tmp_path, monkeypatch):
+    asked = []
+    patch_download(monkeypatch, asked)
+    ref = doc_ref([types.PhotoStrippedSize("i", b"\x01\x28\x28"), types.PhotoSize("s", 100, 56, 900), types.PhotoSize("m", 320, 180, 4000)])
+
+    async def go():
+        client = TelegramClient(MemorySession(), 1, "h")
+        got = await fetch_media(client, ref, "video", str(tmp_path))
+        assert open(got.path, "rb").read() == b"THE BYTES" and got.name == "Episode 12.mkv"
+        assert got.thumb_path and got.thumb_path.endswith(".jpg") and os.path.dirname(got.thumb_path) != os.path.dirname(got.path)
+        assert open(got.thumb_path, "rb").read() == b"THUMB"
+        assert asked == ["", "m"]  # the file itself, then the biggest picture of the file
+
+    run(go())
+
+
+def test_a_file_without_a_thumbnail_has_none(tmp_path, monkeypatch):
+    asked = []
+    patch_download(monkeypatch, asked)
+
+    async def go():
+        client = TelegramClient(MemorySession(), 1, "h")
+        got = await fetch_media(client, doc_ref(None), "video", str(tmp_path))
+        assert got.thumb_path is None and os.path.isfile(got.path) and asked == [""]
+
+    run(go())
+
+
+def test_a_file_that_only_has_the_tiny_inline_thumbnail_still_gets_one(tmp_path, monkeypatch):
+    asked = []
+    patch_download(monkeypatch, asked)
+    stripped = types.PhotoStrippedSize("i", b"\x01\x28\x28\xa2\x8a\x28\x03")
+
+    async def go():
+        client = TelegramClient(MemorySession(), 1, "h")
+        got = await fetch_media(client, doc_ref([stripped]), "video", str(tmp_path))
+        assert got.thumb_path and os.path.getsize(got.thumb_path) > 0  # Telethon turns the inline bytes into a JPEG itself
+        assert asked == [""]  # no download needed: the picture was inside the message
+
+    run(go())
+
+
+def test_a_thumbnail_that_cannot_be_downloaded_is_left_out_and_the_file_is_kept(tmp_path, monkeypatch):
+    async def fake_download(self, input_location, file, **kw):
+        if getattr(input_location, "thumb_size", ""):
+            raise errors.FileReferenceExpiredError(request=None)
+        with open(file, "wb") as f:
+            f.write(b"THE BYTES")
+        return file
+
+    monkeypatch.setattr(TelegramClient, "_download_file", fake_download)
+    ref = doc_ref([types.PhotoSize("m", 320, 180, 4000)])
+
+    async def go():
+        client = TelegramClient(MemorySession(), 1, "h")
+        got = await fetch_media(client, ref, "video", str(tmp_path))
+        assert got.thumb_path is None and open(got.path, "rb").read() == b"THE BYTES"
+
+    run(go())
+
+
+def test_a_photo_is_fetched_without_a_thumbnail_step(tmp_path, monkeypatch):
+    asked = []
+    patch_download(monkeypatch, asked)
+
+    async def go():
+        client = TelegramClient(MemorySession(), 1, "h")
+        got = await fetch_media(client, ref_of(photo_media(11)), "photo", str(tmp_path))
+        assert got.thumb_path is None and len(asked) == 1  # the photo itself, nothing else
+
+    run(go())
+
+
+def test_a_client_that_cannot_give_thumbnails_is_not_a_problem(tmp_path):
+    class Plain:  # a download_media that knows nothing of thumbnails
+        async def download_media(self, media, file=None):
+            path = str(file) + "x.bin"
+            with open(path, "wb") as f:
+                f.write(b"B")
+            return path
+
+    async def go():
+        got = await fetch_media(Plain(), doc_ref([types.PhotoSize("m", 320, 180, 4000)]), "document", str(tmp_path))
+        assert got.thumb_path is None and os.path.isfile(got.path)
+        assert await fetch_thumb(Plain(), doc_ref(None), str(tmp_path)) is None
+
+    run(go())