Spaces:
Running
Running
| """Handler-level tests: real Telethon event builders, fake events and client.""" | |
| import asyncio | |
| import shutil | |
| import time | |
| from pathlib import Path | |
| import pytest | |
| from telethon import functions | |
| from bot import app as appmod | |
| from bot import errorlog, panels | |
| from bot.app import COMMANDS, OWNER_COMMANDS, PANEL_NAMES, App, Pending, cmd_pattern, is_renameable | |
| from bot.config import load_config | |
| from bot.db import Database | |
| from bot.queue import JobQueue | |
| from .conftest import needs_ffmpeg | |
| from .appkit import ( | |
| ADMIN_CFG, ADMIN_DB, EPISODE, EPISODE_NEW, LOG, OWNER, STRANGER, auto_thumb, autofmt, click, data_of, edit_auto, env_for, jpeg, | |
| prefs, say, send_doc, set_auto_thumb, | |
| ) | |
| from .tgfakes import ( | |
| FakeCallback, FakeClient, FakeEvent, FakeMessage, dispatch, document_message, flat_buttons, make_file, variables_line, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # access control & routing | |
| # --------------------------------------------------------------------------- | |
| async def test_strangers_are_ignored_and_reported_once(ctx): | |
| for text in ("/start", "hello", "/rename x", "/cancel"): | |
| assert (await say(ctx, STRANGER, text)).out == [] | |
| assert (await send_doc(ctx, STRANGER)).out == [] | |
| cb = await click(ctx, STRANGER, "m:home") | |
| assert cb.edits == [] and not ctx.app.queue.jobs | |
| reports = [t for chat, t, _kw in ctx.client.sent if chat == LOG] | |
| assert len(reports) == 1 and str(STRANGER) in reports[0] and f"/addadmin {STRANGER}" in reports[0] | |
| async def test_a_plain_message_from_a_stranger_is_also_reported(ctx): | |
| await say(ctx, STRANGER, "hi there") | |
| assert len([1 for chat, _t, _k in ctx.client.sent if chat == LOG]) == 1 | |
| async def test_owner_and_admins_from_config_and_db_are_allowed(ctx): | |
| await ctx.db.add_admin(ADMIN_DB, OWNER) | |
| ctx.app.admins |= set(await ctx.db.list_admins()) | |
| for uid in (OWNER, ADMIN_CFG, ADMIN_DB): | |
| assert (await say(ctx, uid, "/help")).out, uid | |
| assert ctx.client.sent == [] # nobody was reported | |
| async def test_group_chats_are_ignored(ctx): | |
| assert (await say(ctx, OWNER, "/start", private=False)).out == [] | |
| assert (await say(ctx, STRANGER, "/start", private=False)).out == [] | |
| assert ctx.client.sent == [] | |
| async def test_owner_only_commands_are_refused_for_admins(ctx): | |
| for text in ("/addadmin 123456", "/deladmin 123456", "/admins"): | |
| ev = await say(ctx, ADMIN_CFG, text) | |
| assert ev.texts == ["⛔ Only the owner can do that."], text | |
| assert 123456 not in ctx.app.admins | |
| assert (await click(ctx, ADMIN_CFG, "m:admin")).alerts == ["Owner only."] | |
| assert (await click(ctx, ADMIN_CFG, "adm:stats")).alerts == ["Owner only."] | |
| assert (await click(ctx, ADMIN_CFG, "adm:clear")).alerts == ["Owner only."] | |
| assert ctx.client.sent == [] | |
| def test_cmd_pattern_accepts_args_bot_suffix_and_any_case(): | |
| p = cmd_pattern("rename") | |
| assert p.match("/rename New Name").group(1) == "New Name" | |
| assert p.match("/rename@MyBot New Name").group(1) == "New Name" | |
| assert p.match("/RENAME").group(1) is None | |
| assert p.match("/rename\nline one\nline two").group(1) == "line one\nline two" | |
| assert not p.match("/renamex") and not p.match("rename") and not p.match("/re") | |
| def test_every_menu_command_has_a_handler_and_telegram_compatible_text(ctx): | |
| table = ctx.app._command_table() | |
| for name, description in COMMANDS: | |
| assert name in table | |
| assert name.islower() and 1 <= len(name) <= 32 | |
| assert 3 <= len(description) <= 256 | |
| assert OWNER_COMMANDS <= set(table) | |
| assert {"settitle", "setauthor", "setartist", "setaudio", "setsubtitle", "setvideo", "removewords", "setchannel"} <= set(table) | |
| assert dict(COMMANDS)["status"] == "Show basic bot uptime, ping, and version information." | |
| assert dict(COMMANDS)["stats"] == "Show detailed system stats such as users, CPU, RAM, and disk usage." | |
| async def test_setup_registers_command_menu_loads_admins_and_survives_log_channel_errors(tmp_path): | |
| cfg = load_config(env_for(tmp_path)) | |
| db = Database(f"sqlite+aiosqlite:///{tmp_path}/setup.db") | |
| await db.init() | |
| await db.add_admin(ADMIN_DB, OWNER) | |
| client = FakeClient() | |
| client.bad_entities.add(LOG) | |
| app = App(client, cfg, db) | |
| await app.setup() | |
| try: | |
| assert client.parse_mode == "html" | |
| assert ADMIN_DB in app.admins and ADMIN_CFG in app.admins | |
| (request,) = [r for r in client.requests if isinstance(r, functions.bots.SetBotCommandsRequest)] | |
| assert [c.command for c in request.commands] == [c for c, _ in COMMANDS] | |
| assert len(client.handlers) > 10 | |
| finally: | |
| await app.shutdown() | |
| await db.close() | |
| # --------------------------------------------------------------------------- | |
| # /start, /help and the one-line settings commands | |
| # --------------------------------------------------------------------------- | |
| async def test_start_shows_the_home_panel_and_hides_admin_from_admins(ctx): | |
| ev = await say(ctx, OWNER, "/start") | |
| text, buttons, kw = ev.out[0] | |
| assert "Auto Rename Bot" in text and kw["parse_mode"] == "html" | |
| assert {"m:auto", "m:rename", "m:meta", "m:ftype", "m:formats", "m:help", "m:admin"} <= set(data_of(buttons)) | |
| assert "m:cap" not in data_of(buttons) # the caption and the thumbnail are edited in /formats | |
| assert "m:admin" not in data_of((await say(ctx, ADMIN_CFG, "/start")).out[0][1]) | |
| async def test_start_variants(ctx, text): | |
| assert (await say(ctx, OWNER, text)).out | |
| async def test_start_and_help(ctx): | |
| ctx.app.pending[OWNER] = Pending("remove", OWNER, 1, "rename", False, time.time() + 100) | |
| await say(ctx, OWNER, "/start") | |
| assert OWNER not in ctx.app.pending | |
| ev = await say(ctx, OWNER, "/help") | |
| assert "How to use" in ev.texts[0] and "x:close" in data_of(ev.out[0][1]) | |
| async def test_mode_command(ctx): | |
| ev = await say(ctx, OWNER, "/mode caption") | |
| assert (await prefs(ctx))["mode"] == "caption" and "CAPTION" in ev.texts[0] | |
| for text in ("/mode", "/mode nonsense"): | |
| ev = await say(ctx, OWNER, text) | |
| assert "RENAME MODE" in ev.texts[0] and "set:mode:filename" in data_of(ev.out[0][1]) | |
| assert (await prefs(ctx))["mode"] == "caption" | |
| await say(ctx, OWNER, "/mode FILENAME") | |
| assert (await prefs(ctx))["mode"] == "filename" | |
| async def test_filetype_command_and_aliases(ctx): | |
| for arg, expected in (("video", "video"), ("audio", "audio"), ("document", "document"), ("original", "same"), | |
| ("doc", "document"), ("file", "document"), ("same", "same")): | |
| await say(ctx, OWNER, f"/filetype {arg}") | |
| assert (await prefs(ctx))["file_type"] == expected, arg | |
| ev = await say(ctx, OWNER, "/filetype") | |
| assert "FILE TYPE" in ev.texts[0] | |
| ev = await say(ctx, OWNER, "/filetype bogus") | |
| assert "FILE TYPE" in ev.texts[0] and (await prefs(ctx))["file_type"] == "same" | |
| async def test_metadata_command(ctx): | |
| await say(ctx, OWNER, "/metadata on") | |
| assert (await prefs(ctx))["metadata_on"] is True | |
| await say(ctx, OWNER, "/metadata off") | |
| assert (await prefs(ctx))["metadata_on"] is False | |
| ev = await say(ctx, OWNER, "/metadata") | |
| assert "currently: Off" in ev.texts[0] | |
| async def test_the_template_and_caption_of_the_auto_format_are_edited_in_its_formats_screen(ctx): | |
| auto = await ctx.db.create_format(OWNER, "Mine", kind="auto") | |
| fid = auto["id"] | |
| await click(ctx, OWNER, f"fm:in:{fid}:template", msg_id=500) | |
| await say(ctx, OWNER, "[S{season}E{episode}] {title} [{quality}]", mid=60) | |
| assert "Template saved" in ctx.client.edited[-1][1] | |
| assert (await autofmt(ctx))["template"] == "[S{season}E{episode}] {title} [{quality}]" | |
| assert "template" not in await prefs(ctx) and "caption" not in await prefs(ctx) | |
| assert "[S{season}E{episode}] {title} [{quality}]" in (await click(ctx, OWNER, "m:rename")).edits[-1][0] # shown there, not edited | |
| await click(ctx, OWNER, f"fm:in:{fid}:caption", msg_id=501) | |
| await say(ctx, OWNER, "<b>{filename}</b> {size}", mid=61) | |
| assert (await autofmt(ctx))["caption"] == "<b>{filename}</b> {size}" | |
| assert await ctx.db.count_formats(OWNER) == 1 # both went into the same format | |
| async def test_the_template_caption_and_thumbnail_commands_are_gone(ctx, command): | |
| assert command not in [name for name, _description in COMMANDS] and command not in ctx.app._command_table() | |
| await set_auto_thumb(ctx, b"\xff\xd8thumb") | |
| before = await autofmt(ctx) | |
| ev = await say(ctx, OWNER, f"/{command} something {{title}}") | |
| assert ev.out == [] # not a command any more: nothing answers it | |
| assert await autofmt(ctx) == before and await auto_thumb(ctx) == b"\xff\xd8thumb" | |
| async def test_removewords_command(ctx): | |
| await say(ctx, OWNER, "/removewords @Chan, [Group]") | |
| assert (await prefs(ctx))["remove_words"] == ["@Chan", "[Group]"] | |
| assert "@Chan, [Group]" in (await say(ctx, OWNER, "/removewords")).texts[0] | |
| await say(ctx, OWNER, "/removewords off") | |
| assert (await prefs(ctx))["remove_words"] == [] | |
| async def test_template_and_caption_prompts_list_each_variable_as_its_own_copyable_block(ctx): | |
| fid = (await ctx.db.create_format(OWNER, "Mine", kind="auto"))["id"] | |
| line, blocks = variables_line((await click(ctx, OWNER, f"fm:in:{fid}:template")).edits[-1][0]) | |
| assert blocks == line.split(": ", 1)[1].split() and "{quality}" in blocks and "{3episode}" in blocks and "{4episode}" not in blocks and len(blocks) == 11 | |
| line, blocks = variables_line((await click(ctx, OWNER, f"fm:in:{fid}:caption")).edits[-1][0]) | |
| assert blocks == line.split(": ", 1)[1].split() and "{size}" in blocks and "{3episode}" in blocks and "{4episode}" not in blocks and len(blocks) == 13 | |
| async def test_settings_are_per_user(ctx): | |
| await say(ctx, OWNER, "/mode caption") | |
| await say(ctx, ADMIN_CFG, "/filetype audio") | |
| assert (await prefs(ctx, OWNER))["mode"] == "caption" and (await prefs(ctx, OWNER))["file_type"] == "same" | |
| assert (await prefs(ctx, ADMIN_CFG))["mode"] == "filename" and (await prefs(ctx, ADMIN_CFG))["file_type"] == "audio" | |
| async def test_metadata_field_commands(ctx, command, field): | |
| ev = await say(ctx, OWNER, f"/{command} {{lang}} Value") | |
| assert (await prefs(ctx))["meta"][field] == "{lang} Value" and "✅" in ev.texts[0] | |
| assert "{lang} Value" in (await say(ctx, OWNER, f"/{command}")).texts[0] | |
| await say(ctx, OWNER, f"/{command} off") | |
| assert (await prefs(ctx))["meta"][field] == "" | |
| async def test_keepaudio_by_language_and_by_number(ctx): | |
| ev = await say(ctx, OWNER, "/keepaudio English Japanese") | |
| s = await prefs(ctx) | |
| assert s["keep_audio"] == ["english", "japanese"] and s["metadata_on"] is True and "turned it On" in ev.texts[0] | |
| await say(ctx, OWNER, "/keepaudio 2 1") | |
| assert (await prefs(ctx))["keep_audio"] == ["2", "1"] | |
| assert "2, 1" in (await say(ctx, OWNER, "/keepaudio")).texts[0] | |
| await say(ctx, OWNER, "/keepaudio off") | |
| assert (await prefs(ctx))["keep_audio"] == [] | |
| ev = await say(ctx, OWNER, "/keepaudio 0") | |
| assert ev.texts[0].startswith("❌") | |
| async def test_keepsubtitle_is_independent_of_keepaudio(ctx): | |
| await say(ctx, OWNER, "/keepaudio 2 1") | |
| await say(ctx, OWNER, "/keepsubtitle english") | |
| s = await prefs(ctx) | |
| assert s["keep_audio"] == ["2", "1"] and s["keep_subtitle"] == ["english"] | |
| await say(ctx, OWNER, "/keepsubtitle off") | |
| s = await prefs(ctx) | |
| assert s["keep_audio"] == ["2", "1"] and s["keep_subtitle"] == [] | |
| async def test_setchannel(ctx): | |
| assert "not set" in (await say(ctx, OWNER, "/setchannel")).texts[0] | |
| ev = await say(ctx, OWNER, "/setchannel -1002223334445") | |
| assert (await prefs(ctx))["target_channel"] == -1002223334445 and "✅" in ev.texts[0] | |
| ev = await say(ctx, OWNER, "/setchannel @name") | |
| assert ev.texts[0].startswith("❌") and (await prefs(ctx))["target_channel"] == -1002223334445 | |
| ctx.client.bad_entities.add(-1005556667778) | |
| ev = await say(ctx, OWNER, "/setchannel -1005556667778") | |
| assert "can't access" in ev.texts[0] and (await prefs(ctx))["target_channel"] == -1002223334445 | |
| await say(ctx, OWNER, "/setchannel off") | |
| assert (await prefs(ctx))["target_channel"] is None | |
| # --------------------------------------------------------------------------- | |
| # thumbnails | |
| # --------------------------------------------------------------------------- | |
| async def send_photo(ctx, mid=20, sender=OWNER, data=None, **kw): | |
| ev = FakeEvent(sender, message=FakeMessage("", mid=mid, photo=True, **kw), photo_bytes=jpeg() if data is None else data) | |
| await dispatch(ctx.client, ev) | |
| return ev | |
| async def test_a_photo_sent_or_forwarded_to_the_bot_is_not_made_a_thumbnail(ctx): | |
| ev = await send_photo(ctx) | |
| assert ev.out == [] and ctx.client.files == [] and ctx.client.edited == [] # no answer, nothing saved | |
| assert await ctx.db.count_formats(OWNER) == 0 # not even a format was made for it | |
| auto = await ctx.db.create_format(OWNER, "Auto", kind="auto") | |
| manual = await ctx.db.create_format(OWNER, "Manual", kind="manual") | |
| assert (await send_photo(ctx, 21)).out == [] | |
| assert (await send_photo(ctx, 22, fwd=True)).out == [] # a forwarded one is no different | |
| assert await ctx.db.get_format_thumb(OWNER, auto["id"]) is None and await ctx.db.get_format_thumb(OWNER, manual["id"]) is None | |
| assert not (await autofmt(ctx))["has_thumb"] and ctx.client.files == [] | |
| async def test_only_the_thumbnail_prompt_of_a_format_takes_a_photo(ctx): | |
| auto = await ctx.db.create_format(OWNER, "Auto", kind="auto") | |
| for kind in ("remove", "mt_title", "keep_audio"): # no other prompt takes it | |
| ctx.app.pending[OWNER] = Pending(kind, OWNER, 1, "rename", False, time.time() + 100) | |
| assert (await send_photo(ctx)).out == [] and ctx.app.pending[OWNER].kind == kind | |
| ctx.app.pending[OWNER] = Pending("fmt:template", OWNER, 1, "formats", False, time.time() + 100, fid=auto["id"]) | |
| assert (await send_photo(ctx)).out == [] and ctx.app.pending[OWNER].kind == "fmt:template" | |
| ctx.app.pending[OWNER] = Pending("fmt:thumb", OWNER, 1, "formats", False, time.time() - 1, fid=auto["id"]) # one that ran out | |
| assert (await send_photo(ctx)).out == [] | |
| assert await ctx.db.get_format_thumb(OWNER, auto["id"]) is None | |
| ctx.app.pending[OWNER] = Pending("fmt:thumb", OWNER, 500, "formats", False, time.time() + 100, fid=auto["id"]) | |
| await send_photo(ctx, 30) # the one that is waiting for it does | |
| assert (await ctx.db.get_format_thumb(OWNER, auto["id"]))[:2] == b"\xff\xd8" and OWNER not in ctx.app.pending | |
| async def test_unreadable_photos_are_reported_while_a_format_waits_for_its_thumbnail(ctx): | |
| auto = await ctx.db.create_format(OWNER, "Auto", kind="auto") | |
| for data, said in ((b"not an image", "couldn't read"), (None, "couldn't download")): | |
| ctx.app.pending[OWNER] = Pending("fmt:thumb", OWNER, 500, "formats", False, time.time() + 100, fid=auto["id"]) | |
| ev = FakeEvent(OWNER, message=FakeMessage("", photo=True), photo_bytes=data) | |
| await dispatch(ctx.client, ev) | |
| assert said in ev.texts[0] and OWNER in ctx.app.pending # and the prompt stays open for another try | |
| assert await ctx.db.get_format_thumb(OWNER, auto["id"]) is None | |
| stranger = await send_photo(ctx, sender=STRANGER) | |
| assert stranger.out == [] and await autofmt(ctx, STRANGER) is None | |
| async def test_the_buttons_of_the_old_caption_template_and_thumbnail_screens_do_nothing(ctx): | |
| auto = await set_auto_thumb(ctx, b"\xff\xd8keep") | |
| await edit_auto(ctx, template="{title}", caption="<i>{filename}</i>") | |
| before = await autofmt(ctx) | |
| old = ["m:cap", "in:template", "in:caption", "clr:caption"] + [f"th:{a}{b}" for a in ("view", "del", "rm") for b in ("", f":{auto['id']}")] | |
| for data in old: | |
| cb = await click(ctx, OWNER, data) | |
| assert cb.edits == [] and cb.alerts == [] and not cb.deleted, data | |
| assert not ctx.app.pending and ctx.client.files == [] | |
| assert await autofmt(ctx) == before and await auto_thumb(ctx) == b"\xff\xd8keep" # nothing was changed | |
| # --------------------------------------------------------------------------- | |
| # renaming | |
| # --------------------------------------------------------------------------- | |
| async def test_a_file_is_queued_with_the_template_name_and_a_cancel_button(ctx): | |
| ev = await send_doc(ctx, OWNER, mid=10) | |
| (job,) = ctx.app.queue.jobs.values() | |
| assert job.new_name == EPISODE_NEW and (job.user_id, job.chat_id, job.msg_id) == (OWNER, OWNER, 10) | |
| assert job.file_name == EPISODE and job.size == 50 * 1024 * 1024 and job.fmt is None and job.state == "queued" | |
| text, buttons, kw = ev.out[0] | |
| assert "Queued" in text and "starting soon" in text and EPISODE_NEW in text | |
| assert flat_buttons(buttons) == [("✖ Cancel", f"job:x:{job.id}")] | |
| assert job.status_msg is not None | |
| ev = await send_doc(ctx, OWNER, "Another.File.mp4", mid=11) | |
| assert "1 ahead of it" in ev.texts[0] | |
| async def test_caption_mode_reads_the_caption(ctx): | |
| await say(ctx, OWNER, "/mode caption") | |
| await send_doc(ctx, OWNER, "xyz123.mkv", caption="Tokyo Revengers S02E07 720p", mid=12) | |
| (job,) = ctx.app.queue.jobs.values() | |
| assert job.new_name.endswith(".mkv") and "Tokyo Revengers" in job.new_name and "S02E07" in job.new_name | |
| async def test_remove_words_are_applied_to_the_new_name(ctx): | |
| await say(ctx, OWNER, "/removewords @SomeChannel") | |
| await send_doc(ctx, OWNER, "[@SomeChannel] Tokyo Revengers - 05 (1080p).mkv") | |
| (job,) = ctx.app.queue.jobs.values() | |
| assert "SomeChannel" not in job.new_name and "Tokyo Revengers" in job.new_name | |
| async def test_the_one_piece_style_name_gets_its_episode_number(ctx): | |
| await edit_auto(ctx, template="One Piece - E{episode}") | |
| ev = await send_doc(ctx, OWNER, "1147 - One Piece [Dual] 720p @Anime_X_Universe.mkv", mid=21) | |
| (job,) = ctx.app.queue.jobs.values() | |
| assert job.new_name == "One Piece - E1147.mkv" and job.notes == [] | |
| assert "One Piece - E1147.mkv" in ev.texts[0] and "⚠️" not in ev.texts[0] | |
| async def test_a_template_variable_that_cannot_be_filled_is_flagged(ctx): | |
| await edit_auto(ctx, template="One Piece - E{episode}") | |
| ev = await send_doc(ctx, OWNER, "Plain Movie.mkv") | |
| (job,) = ctx.app.queue.jobs.values() | |
| assert job.new_name == "One Piece - E.mkv" | |
| assert "⚠️" in ev.texts[0] and "{episode}" in ev.texts[0] and "file name" in ev.texts[0] and "change the template" in ev.texts[0] | |
| assert job.notes and "{episode}" in job.notes[0] | |
| await say(ctx, OWNER, "/mode caption") | |
| ev = await send_doc(ctx, OWNER, "Plain Movie.mkv", mid=22) | |
| assert "in the caption" in ev.texts[0] | |
| async def test_the_default_template_never_warns(ctx): | |
| ev = await send_doc(ctx, OWNER, "Plain Movie.mkv") | |
| assert "⚠️" not in ev.texts[0] # bracket groups vanish on purpose | |
| class SlowEvent(FakeEvent): | |
| """A message whose reply takes a while, like a slow round trip to Telegram.""" | |
| def __init__(self, *a, delay=0.0, **kw): | |
| super().__init__(*a, **kw) | |
| self.delay = delay | |
| async def reply(self, text=None, **kw): | |
| await asyncio.sleep(self.delay) | |
| return await super().respond(text, **kw) | |
| async def test_jobs_are_ordered_by_the_message_that_triggered_them(ctx): | |
| await send_doc(ctx, OWNER, "B.mkv", mid=31) | |
| await send_doc(ctx, OWNER, "A.mkv", mid=30) | |
| await send_doc(ctx, OWNER, "old.mkv", 1024, mid=34) | |
| assert sorted(j.order for j in ctx.app.queue.jobs.values()) == [30, 31, 34] | |
| assert [j.file_name for j in ctx.app.queue.for_user(OWNER)] == ["A.mkv", "B.mkv", "old.mkv"] | |
| async def test_a_forwarded_batch_comes_out_in_the_order_it_was_sent(ctx): | |
| """Telethon handles messages that arrive together concurrently; the results must still be in order.""" | |
| ran = [] | |
| async def runner(job): | |
| ran.append(job.file_name) | |
| ctx.app.queue = JobQueue(runner, workers=1, settle=0.3) | |
| ctx.app.queue.start() | |
| await asyncio.sleep(0.02) # the worker is idle, waiting for files | |
| names = { | |
| 1145: "1145 - One Piece [Dual] 720p @Anime_X_Universe.mkv", | |
| 1146: "1146 - One Piece [Dual] 720p @Anime_X_Universe.mkv", | |
| 1147: "1147 - One Piece [Dual] 720p @Anime_X_Universe.mkv", | |
| } | |
| delays = {1145: 0.10, 1146: 0.00, 1147: 0.05} # the replies finish in a different order than the files arrived | |
| events = [ | |
| SlowEvent(OWNER, message=document_message(name, 1024, mid=mid), delay=delays[mid]) for mid, name in names.items() | |
| ] | |
| await asyncio.gather(*(dispatch(ctx.client, ev) for ev in reversed(events))) | |
| await asyncio.sleep(0.7) | |
| assert ran == [names[1145], names[1146], names[1147]] | |
| async def test_oversized_files_are_refused(ctx): | |
| ev = await send_doc(ctx, OWNER, size=101 * 1024 * 1024) | |
| assert "limit is 100.00 MB" in ev.texts[0] and not ctx.app.queue.jobs | |
| ev = await send_doc(ctx, OWNER, size=100 * 1024 * 1024) # exactly at the limit is fine | |
| assert "Queued" in ev.texts[0] | |
| async def test_stickers_voice_notes_and_gifs_are_not_renamed(ctx): | |
| for flags in ({"sticker": True}, {"voice": True}, {"gif": True}, {"video_note": True}): | |
| ev = FakeEvent(OWNER, message=FakeMessage("", file=make_file("x.webm"), **flags)) | |
| await dispatch(ctx.client, ev) | |
| assert ev.out == [], flags | |
| assert not ctx.app.queue.jobs | |
| assert is_renameable(document_message()) and not is_renameable(FakeMessage("text")) | |
| async def test_a_file_without_a_name_gets_a_generated_one(ctx): | |
| ev = FakeEvent(OWNER, message=FakeMessage("", mid=77, file=make_file(None, 1024, "video/mp4", ext=".mp4"), video=True)) | |
| await dispatch(ctx.client, ev) | |
| (job,) = ctx.app.queue.jobs.values() | |
| assert job.file_name == "video_77.mp4" and job.new_name.endswith(".mp4") | |
| # --------------------------------------------------------------------------- | |
| # queue and cancel | |
| # --------------------------------------------------------------------------- | |
| async def test_queue_listing_is_scoped_to_the_user_unless_owner(ctx): | |
| assert "Nothing in the queue" in (await say(ctx, OWNER, "/queue")).texts[0] | |
| await send_doc(ctx, OWNER, "Owner.File.mkv") | |
| await send_doc(ctx, ADMIN_CFG, "Admin.File.mkv") | |
| admin_view = (await say(ctx, ADMIN_CFG, "/queue")).texts[0] | |
| assert "Admin File" in admin_view and "Owner File" not in admin_view | |
| owner_view = (await say(ctx, OWNER, "/queue")).texts[0] | |
| assert "Admin File" in owner_view and "Owner File" in owner_view and str(ADMIN_CFG) in owner_view | |
| async def test_cancel_commands_respect_ownership(ctx): | |
| await send_doc(ctx, OWNER, "Owner.File.mkv") # job 1 | |
| await send_doc(ctx, ADMIN_CFG, "Admin.File.mkv") # job 2 | |
| await send_doc(ctx, ADMIN_CFG, "Admin.Two.mkv") # job 3 | |
| assert "couldn't find" in (await say(ctx, ADMIN_CFG, "/cancel 1")).texts[0] # not theirs | |
| assert ctx.app.queue.jobs[1].state == "queued" | |
| assert "Cancelling 2 job(s)" in (await say(ctx, ADMIN_CFG, "/cancel all")).texts[0] # only their own | |
| assert set(ctx.app.queue.jobs) == {1} | |
| assert "no active jobs" in (await say(ctx, ADMIN_CFG, "/cancel")).texts[0] | |
| await send_doc(ctx, ADMIN_CFG, "Admin.Three.mkv") # job 4 | |
| assert "Cancelling 2 job(s)" in (await say(ctx, OWNER, "/cancel all")).texts[0] # owner: jobs 1 and 4 | |
| assert not ctx.app.queue.jobs | |
| async def test_cancelled_queued_jobs_update_their_status_message(ctx): | |
| await send_doc(ctx, OWNER) | |
| await say(ctx, OWNER, "/cancel 1") | |
| for _ in range(5): | |
| await asyncio.sleep(0) | |
| (entity, text, kw), = ctx.client.edited | |
| assert "Cancelled" in text and kw["buttons"] is None | |
| async def test_cancel_button_on_a_queued_file(ctx): | |
| await send_doc(ctx, OWNER) | |
| await send_doc(ctx, OWNER, "Second.mkv") | |
| cb = await click(ctx, ADMIN_CFG, "job:x:1") # somebody else's job | |
| assert cb.toasts == ["That job already finished."] and 1 in ctx.app.queue.jobs | |
| cb = await click(ctx, OWNER, "job:x:1") | |
| assert cb.toasts == ["Cancelling…"] and 1 not in ctx.app.queue.jobs | |
| assert (await click(ctx, OWNER, "job:x:1")).toasts == ["That job already finished."] | |
| admin_doc = await send_doc(ctx, ADMIN_CFG, "Admin.mkv") | |
| assert admin_doc.out | |
| assert (await click(ctx, OWNER, "job:x:3")).toasts == ["Cancelling…"] # the owner may cancel anything | |
| async def test_cancel_clears_a_pending_prompt_first(ctx): | |
| ctx.app.pending[OWNER] = Pending("template", OWNER, 1, "rename", False, time.time() + 100) | |
| await send_doc(ctx, OWNER) | |
| ev = await say(ctx, OWNER, "/cancel") | |
| assert "nothing was changed" in ev.texts[0] and OWNER not in ctx.app.pending | |
| assert ctx.app.queue.jobs # the job is untouched | |
| assert "Cancelling 1 job(s)" in (await say(ctx, OWNER, "/cancel")).texts[0] | |
| # --------------------------------------------------------------------------- | |
| # admin management, status and stats | |
| # --------------------------------------------------------------------------- | |
| async def test_owner_manages_admins(ctx): | |
| ev = await say(ctx, OWNER, "/addadmin 555555") | |
| assert "can now use the bot" in ev.texts[0] and 555555 in ctx.app.admins | |
| assert 555555 in await ctx.db.list_admins() | |
| assert (await say(ctx, 555555, "/help")).out # works immediately | |
| assert "already have access" in (await say(ctx, OWNER, "/addadmin 555555")).texts[0] | |
| assert "already have access" in (await say(ctx, OWNER, f"/addadmin {OWNER}")).texts[0] | |
| assert "already have access" in (await say(ctx, OWNER, f"/addadmin {ADMIN_CFG}")).texts[0] | |
| for bad in ("/addadmin", "/addadmin abc", "/addadmin 12", "/addadmin @user"): | |
| assert "Usage" in (await say(ctx, OWNER, bad)).texts[0], bad | |
| listing = (await say(ctx, OWNER, "/admins")).texts[0] | |
| assert str(OWNER) in listing and "555555" in listing and str(ADMIN_CFG) in listing | |
| assert "was removed" in (await say(ctx, OWNER, "/deladmin 555555")).texts[0] | |
| assert 555555 not in ctx.app.admins and 555555 not in await ctx.db.list_admins() | |
| assert (await say(ctx, 555555, "/help")).out == [] # access gone | |
| assert "wasn't in the list" in (await say(ctx, OWNER, "/deladmin 555555")).texts[0] | |
| assert "ADMIN_IDS" in (await say(ctx, OWNER, f"/deladmin {ADMIN_CFG}")).texts[0] | |
| assert ADMIN_CFG in ctx.app.admins | |
| assert "Usage" in (await say(ctx, OWNER, "/deladmin")).texts[0] | |
| async def test_status_pings_telegram_and_reports_version_info(ctx): | |
| ev = await say(ctx, OWNER, "/status") | |
| text = ev.texts[0] | |
| assert "Bot status" in text and "Uptime" in text and "Ping" in text and "Version" in text and "Telethon" in text | |
| assert any(isinstance(r, functions.PingRequest) for r in ctx.client.requests) | |
| assert (await say(ctx, ADMIN_CFG, "/status")).out # admins may use it too | |
| async def test_status_shows_what_makes_transfers_fast_and_how_they_went(ctx, monkeypatch): | |
| from bot import fasttransfer | |
| monkeypatch.setattr(fasttransfer, "stats", fasttransfer.TransferStats(fast_down=4, fast_up=3, small_up=1)) | |
| text = (await say(ctx, OWNER, "/status")).texts[0] | |
| assert "<b>Speed</b>" in text and "Encryption: cryptg ✅" in text and "Parallel transfer:" in text | |
| assert "<b>Downloads:</b> 4 parallel · 0 standard" in text and "<b>Uploads:</b> 3 parallel · 1 up to 10 MB · 0 standard" in text | |
| fasttransfer.stats.fell_back("upload", RuntimeError("boom")) | |
| assert "failed 1×" in (await say(ctx, OWNER, "/status")).texts[0] | |
| async def test_status_survives_a_failed_ping(ctx): | |
| async def broken(request): | |
| raise ConnectionError("offline") | |
| ctx.client.__class__ = type("Broken", (FakeClient,), {"__call__": lambda self, request: broken(request)}) | |
| assert "n/a" in (await say(ctx, OWNER, "/status")).texts[0] | |
| async def test_stats_reports_users_files_and_system_usage(ctx): | |
| await say(ctx, ADMIN_CFG, "/help") | |
| await ctx.db.record_job(OWNER, 3 * 1024 ** 3) | |
| text = (await say(ctx, OWNER, "/stats")).texts[0] | |
| for expected in ("System stats", "Users", "Files renamed", "3.00 GB", "Queue", "CPU", "RAM", "Disk"): | |
| assert expected in text, expected | |
| assert (await say(ctx, ADMIN_CFG, "/stats")).out | |
| assert (await say(ctx, STRANGER, "/stats")).out == [] | |
| # --------------------------------------------------------------------------- | |
| # /tracks | |
| # --------------------------------------------------------------------------- | |
| async def test_tracks_needs_a_reply_to_a_file(ctx): | |
| for reply in (None, FakeMessage("text", mid=3)): | |
| assert "Reply to a video or audio file" in (await say(ctx, OWNER, "/tracks", reply=reply)).texts[0] | |
| async def test_tracks_lists_numbered_audio_and_subtitles_and_cleans_up(ctx, media_dir, monkeypatch): | |
| calls = [] | |
| async def fake_download(client, message, dest, **kw): | |
| calls.append(kw) | |
| shutil.copy(media_dir / "sample.mkv", dest) | |
| monkeypatch.setattr(appmod.transfer, "download", fake_download) | |
| reply = document_message("Show.mkv", (media_dir / "sample.mkv").stat().st_size, mid=33) | |
| ev = await say(ctx, OWNER, "/tracks", reply=reply, mid=34) | |
| assert ev.texts == ["🔎 Reading the tracks…"] | |
| (status, text, kw), = ctx.client.edited | |
| assert status.id == 1001 and "Show.mkv" in text | |
| assert "Audio" in text and "Subtitles" in text and "Japanese" in text and "English" in text and "use these numbers" in text | |
| assert len(calls) == 1 and calls[0]["max_bytes"] == appmod.TRACKS_PROBE_BYTES | |
| assert not list(Path(ctx.cfg.temp_dir).glob("tracks*")) | |
| async def test_tracks_downloads_the_whole_file_when_the_start_is_not_enough(ctx, media_dir, monkeypatch): | |
| calls = [] | |
| async def fake_download(client, message, dest, **kw): | |
| calls.append(kw) | |
| if "max_bytes" in kw: | |
| Path(dest).write_bytes(b"\x00" * 4096) # e.g. an mp4 whose index is at the end | |
| else: | |
| shutil.copy(media_dir / "sample.mp4", dest) | |
| monkeypatch.setattr(appmod.transfer, "download", fake_download) | |
| reply = document_message("Show.mp4", 40 * 1024 * 1024, mid=33) | |
| await say(ctx, OWNER, "/tracks", reply=reply, mid=34) | |
| assert len(calls) == 2 and "max_bytes" not in calls[1] | |
| assert "Audio" in ctx.client.edited[-1][1] and "needs the whole file" in ctx.client.edited[0][1] | |
| async def test_tracks_gives_up_gracefully(ctx, monkeypatch): | |
| async def junk_download(client, message, dest, **kw): | |
| Path(dest).write_bytes(b"\x00" * 4096) | |
| monkeypatch.setattr(appmod.transfer, "download", junk_download) | |
| for size in (2 * 1024 * 1024, 900 * 1024 * 1024): # small file / too big for a full download | |
| ctx.client.edited.clear() | |
| reply = document_message("Bad.mkv", size, mid=33) | |
| await say(ctx, OWNER, "/tracks", reply=reply, mid=34) | |
| assert "couldn't read any tracks" in ctx.client.edited[-1][1], size | |
| assert not list(Path(ctx.cfg.temp_dir).glob("tracks*")) | |
| # --------------------------------------------------------------------------- | |
| # menus (callbacks) | |
| # --------------------------------------------------------------------------- | |
| async def test_menu_navigation_edits_the_message_in_place(ctx): | |
| cb = await click(ctx, OWNER, "m:meta") | |
| text, buttons = cb.edits[-1] | |
| assert "metadata is currently" in text and "m:tool" in data_of(buttons) | |
| assert cb.answers == [(None, False)] | |
| cb = await click(ctx, OWNER, "m:tool") | |
| assert "TOOL" in cb.edits[-1][0] and "m:meta" in data_of(cb.edits[-1][1]) | |
| cb = await click(ctx, OWNER, "m:home") | |
| assert "Auto Rename Bot" in cb.edits[-1][0] | |
| cb = await click(ctx, OWNER, "m:admin") | |
| assert "ADMIN" in cb.edits[-1][0] | |
| cb = await click(ctx, OWNER, "m:bogus") | |
| assert cb.edits == [] and cb.answers == [(None, False)] | |
| async def test_toggles_update_settings_and_refresh_the_panel(ctx): | |
| cb = await click(ctx, OWNER, "set:mode:caption") | |
| assert (await prefs(ctx))["mode"] == "caption" and cb.toasts == ["Mode: Caption"] | |
| assert "Caption ✅" in [label for label, _d in flat_buttons(cb.edits[-1][1])] | |
| cb = await click(ctx, OWNER, "set:ftype:video") | |
| assert (await prefs(ctx))["file_type"] == "video" and cb.toasts == ["File type: Video"] | |
| cb = await click(ctx, OWNER, "set:ftype:same") | |
| assert cb.toasts == ["File type: Same as original"] | |
| cb = await click(ctx, OWNER, "set:meta:on") | |
| assert (await prefs(ctx))["metadata_on"] is True and cb.toasts == ["Metadata ON"] | |
| await click(ctx, OWNER, "set:meta:off") | |
| assert (await prefs(ctx))["metadata_on"] is False | |
| before = await prefs(ctx) | |
| for junk in ("set:mode:bogus", "set:nothing:x", "set:ftype:zip", "set:meta:maybe", "set:mode"): | |
| cb = await click(ctx, OWNER, junk) | |
| assert cb.edits == [] and cb.answers == [(None, False)], junk | |
| assert await prefs(ctx) == before | |
| async def test_clear_buttons(ctx): | |
| await say(ctx, OWNER, "/keepaudio 2 1") | |
| await say(ctx, OWNER, "/keepsubtitle 1") | |
| cb = await click(ctx, OWNER, "clr:keep_audio") | |
| s = await prefs(ctx) | |
| assert s["keep_audio"] == [] and s["keep_subtitle"] == ["1"] and "TOOL" in cb.edits[-1][0] | |
| await click(ctx, OWNER, "clr:keep_subtitle") | |
| assert (await prefs(ctx))["keep_subtitle"] == [] | |
| cb = await click(ctx, OWNER, "clr:whatever") | |
| assert cb.edits == [] | |
| async def test_close_deletes_the_message_and_falls_back_to_editing(ctx): | |
| ctx.app.pending[OWNER] = Pending("remove", OWNER, 1, "rename", False, time.time() + 100) | |
| cb = await click(ctx, OWNER, "x:close") | |
| assert cb.deleted and OWNER not in ctx.app.pending | |
| old = FakeCallback(OWNER, "x:close") | |
| old.delete_error = RuntimeError("MESSAGE_DELETE_FORBIDDEN") | |
| await dispatch(ctx.client, old) | |
| assert old.edits == [("Closed.", None)] and not old.deleted | |
| async def test_metadata_help_and_unknown_callbacks(ctx): | |
| cb = await click(ctx, OWNER, "help:meta") | |
| assert "How to set metadata" in cb.edits[-1][0] | |
| for junk in ("", "zzz", "adm", "th:what", "job:x:abc", "in:nope"): | |
| cb = await click(ctx, OWNER, junk) | |
| assert cb.answers and not cb.alerts, junk | |
| async def test_admin_panel_buttons(ctx): | |
| await send_doc(ctx, OWNER) | |
| await send_doc(ctx, OWNER, "Two.mkv") | |
| cb = await click(ctx, OWNER, "adm:stats", chat=OWNER) | |
| (chat, text, kw), = ctx.client.sent | |
| assert chat == OWNER and "System stats" in text and "2 waiting" in text | |
| await click(ctx, OWNER, "adm:status") | |
| assert "Bot status" in ctx.client.sent[-1][1] and "<b>Speed</b>" in ctx.client.sent[-1][1] | |
| cb = await click(ctx, OWNER, "adm:clear") | |
| assert cb.alerts == ["Cancelling 2 job(s)."] and not ctx.app.queue.jobs | |
| async def test_every_button_of_every_panel_is_handled(ctx, panel): | |
| await set_auto_thumb(ctx, b"\xff\xd8thumb") | |
| await ctx.db.create_format(OWNER, "Other auto", kind="auto") | |
| await ctx.db.create_format(OWNER, "Manual one", kind="manual") | |
| _text, rows = await ctx.app.panel(OWNER, panel) | |
| clicked = 0 | |
| for row in rows: | |
| for _label, data in row: | |
| cb = await click(ctx, OWNER, data) | |
| assert cb.answers or cb.deleted, (panel, data) | |
| assert not any("went wrong" in a for a in cb.alerts), (panel, data) | |
| clicked += 1 | |
| assert clicked >= 1 | |
| async def test_every_prompt_opens_and_cancels(ctx, kind): | |
| cb = await click(ctx, OWNER, f"in:{kind}") | |
| p = ctx.app.pending[OWNER] | |
| assert p.kind == kind and p.back in PANEL_NAMES and not p.is_photo | |
| text, buttons = cb.edits[-1] | |
| assert text == panels.PROMPTS[kind] and data_of(buttons) == [f"in:cancel:{p.back}"] | |
| cb = await click(ctx, OWNER, f"in:cancel:{p.back}") | |
| assert OWNER not in ctx.app.pending and cb.toasts == ["Cancelled."] and cb.edits | |
| # --------------------------------------------------------------------------- | |
| # "tap a button, then send the value" | |
| # --------------------------------------------------------------------------- | |
| async def test_remove_words_prompt_flow(ctx): | |
| await click(ctx, OWNER, "in:remove", msg_id=500) | |
| ev = await say(ctx, OWNER, "@Chan, [Group]", mid=60) | |
| assert ev.deleted and OWNER not in ctx.app.pending # the typed value is removed to keep the chat tidy | |
| assert (await prefs(ctx))["remove_words"] == ["@Chan", "[Group]"] | |
| (chat, text, kw), = ctx.client.edited | |
| assert chat == OWNER and kw["message"] == 500 and "I'll strip" in text and "RENAME MODE" in text | |
| assert "set:mode:filename" in data_of(kw["buttons"]) and "in:template" not in data_of(kw["buttons"]) | |
| async def test_keep_audio_prompt_flow_enables_metadata_and_returns_to_the_tool_panel(ctx): | |
| await click(ctx, OWNER, "in:keep_audio") | |
| await say(ctx, OWNER, "English Japanese") | |
| s = await prefs(ctx) | |
| assert s["keep_audio"] == ["english", "japanese"] and s["metadata_on"] is True | |
| (chat, text, kw), = ctx.client.edited | |
| assert "TOOL" in text and "English, Japanese" in text and "in:keep_subtitle" in data_of(kw["buttons"]) | |
| async def test_invalid_input_keeps_the_prompt_open(ctx): | |
| await click(ctx, OWNER, "in:mt_title") | |
| ev = await say(ctx, OWNER, "x" * 300) | |
| assert "too long" in ev.texts[0] and "Try again" in ev.texts[0] and not ev.deleted | |
| assert OWNER in ctx.app.pending and ctx.client.edited == [] | |
| ev = await say(ctx, OWNER, "My Title") | |
| assert OWNER not in ctx.app.pending and (await prefs(ctx))["meta"]["title"] == "My Title" | |
| async def test_expired_prompts_and_other_users_text_are_ignored(ctx): | |
| await click(ctx, OWNER, "in:mt_title") | |
| assert (await say(ctx, ADMIN_CFG, "Title")).out == [] # not their prompt | |
| assert OWNER in ctx.app.pending | |
| ctx.app.pending[OWNER].expires = time.time() - 1 | |
| ev = await say(ctx, OWNER, "Title") | |
| assert ev.out == [] and not ev.deleted and OWNER not in ctx.app.pending | |
| assert (await prefs(ctx))["meta"]["title"] == "" and (await prefs(ctx, ADMIN_CFG))["meta"]["title"] == "" # nothing was saved | |
| async def test_plain_text_without_a_prompt_is_ignored(ctx): | |
| ev = await say(ctx, OWNER, "hello bot") | |
| assert ev.out == [] and not ev.deleted | |
| async def test_prompt_on_a_photo_banner_message_is_flagged_and_long_updates_fall_back(ctx, monkeypatch): | |
| await click(ctx, OWNER, "in:mt_title", photo=True) | |
| assert ctx.app.pending[OWNER].is_photo and ctx.app.pending[OWNER].back == "meta" | |
| monkeypatch.setattr(appmod, "visible_len", lambda _t: 5000) # pretend it would not fit a caption | |
| await say(ctx, OWNER, "My Title") | |
| (_chat, text, _kw), = ctx.client.edited | |
| assert "Open the menu to see the update" in text and "My Title" in text | |
| assert (await prefs(ctx))["meta"]["title"] == "My Title" | |
| async def test_when_the_menu_message_cannot_be_edited_a_new_one_is_sent(ctx): | |
| async def refuse(*a, **k): | |
| raise RuntimeError("MESSAGE_ID_INVALID") | |
| await click(ctx, OWNER, "in:remove") | |
| ctx.client.edit_message = refuse | |
| ev = await say(ctx, OWNER, "@Chan, [Group]") | |
| assert (await prefs(ctx))["remove_words"] == ["@Chan", "[Group]"] | |
| assert "I'll strip" in ev.texts[0] and ev.out[0][1] # sent as a fresh message with buttons | |
| # --------------------------------------------------------------------------- | |
| # robustness | |
| # --------------------------------------------------------------------------- | |
| async def test_handler_failures_are_reported_to_the_user(ctx, monkeypatch): | |
| async def boom(*a, **k): | |
| raise RuntimeError("database is down") | |
| monkeypatch.setattr(ctx.db, "get_settings", boom) | |
| ev = await say(ctx, OWNER, "/start") | |
| assert ev.texts == ["⚠️ Something went wrong. Please try again."] | |
| cb = await click(ctx, OWNER, "m:meta") | |
| assert cb.alerts == ["⚠️ Something went wrong."] | |
| # and the bot keeps working afterwards | |
| monkeypatch.undo() | |
| assert "Auto Rename Bot" in (await say(ctx, OWNER, "/start")).texts[0] | |
| async def test_handler_failures_reach_the_log_channel_with_the_update_and_the_traceback(ctx, monkeypatch): | |
| channel_log = errorlog.install(ctx.app._log_send, errorlog.secrets_from(ctx.cfg)) # the root logger, as in main.py | |
| async def boom(*a, **k): | |
| raise RuntimeError("database is down") | |
| monkeypatch.setattr(ctx.db, "get_settings", boom) | |
| ev = await say(ctx, OWNER, "/start") | |
| await channel_log.aclose() | |
| assert ev.texts == ["⚠️ Something went wrong. Please try again."] | |
| (_chat, text, kw), = [s for s in ctx.client.sent if s[0] == LOG] | |
| assert kw["parse_mode"] == "html" | |
| assert text.startswith( | |
| '<pre><code class="language-python">Error handling update:\n' | |
| f"FakeEvent(sender_id={OWNER}, chat_id={OWNER}, message_id=1, text='/start')\n\nTraceback (most recent call last):" | |
| ) | |
| assert text.endswith("RuntimeError: database is down</code></pre>") | |
| async def test_an_error_inside_a_helper_still_names_the_update_that_started_it(ctx, monkeypatch): | |
| channel_log = errorlog.install(ctx.app._log_send) | |
| async def boom(*a, **k): | |
| raise RuntimeError("telegraph exploded") | |
| monkeypatch.setattr(ctx.app, "_mediainfo", boom) | |
| reply = document_message("Movie.mkv", 5 * 1024 * 1024) | |
| await say(ctx, OWNER, "/mediainfo", reply=reply) | |
| await channel_log.aclose() | |
| (_chat, text, _kw), = [s for s in ctx.client.sent if s[0] == LOG] | |
| assert text.startswith('<pre><code class="language-python">/mediainfo failed\nFakeEvent(sender_id=') | |
| assert "text='/mediainfo'" in text and text.endswith("RuntimeError: telegraph exploded</code></pre>") | |
| async def test_a_file_is_handled_by_exactly_one_handler(ctx): | |
| ctx.app.pending[OWNER] = Pending("template", OWNER, 1, "rename", False, time.time() + 100) | |
| ev = await send_doc(ctx, OWNER) | |
| assert len(ev.out) == 1 and "Queued" in ev.texts[0] | |
| assert OWNER in ctx.app.pending # a file does not feed the text prompt | |