Spaces:
Running
Running
| """/mediainfo in the chat: reply to a file, get a Telegraph link.""" | |
| import io | |
| import shutil | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| import pytest | |
| from bot import app as appmod | |
| from bot.telegraph import Telegraph, TelegraphError | |
| from .appkit import OWNER, STRANGER, say, send_doc | |
| from .conftest import needs_ffmpeg | |
| from .faketelegraph import serve_fake_telegraph | |
| from .tgfakes import FakeMessage, document_message | |
| URL = "https://telegra.ph/Show-05-10-21" | |
| HINT = "Reply to a file or video with /mediainfo" | |
| class StubTelegraph: | |
| """What the bot talks to instead of Telegraph: remembers the pages it was asked for.""" | |
| def __init__(self, url=URL, error=None): | |
| self.url, self.error, self.calls = url, error, [] | |
| async def create_page(self, title, nodes): | |
| self.calls.append((title, nodes)) | |
| if self.error is not None: | |
| raise self.error | |
| return self.url | |
| def blocks(nodes): | |
| """{heading: text of its block} for a page's nodes.""" | |
| return {n["children"][0]: nodes[i + 1]["children"][0] for i, n in enumerate(nodes) if n["tag"] == "h4"} | |
| def notes(nodes): | |
| return [n["children"][0]["children"][0] for n in nodes if n["tag"] == "p"] | |
| def serve(monkeypatch, media_dir, name="sample.mkv"): | |
| """Make transfer.download copy a real sample file; returns the list of calls.""" | |
| calls = [] | |
| async def fake_download(client, message, dest, **kw): | |
| calls.append(kw) | |
| shutil.copy(media_dir / name, dest) | |
| monkeypatch.setattr(appmod.transfer, "download", fake_download) | |
| return calls | |
| def serve_junk(monkeypatch, content=b"\x00" * 4096): | |
| calls = [] | |
| async def fake_download(client, message, dest, **kw): | |
| calls.append(kw) | |
| Path(dest).write_bytes(content) | |
| monkeypatch.setattr(appmod.transfer, "download", fake_download) | |
| return calls | |
| def leftovers(ctx): | |
| root = Path(ctx.cfg.temp_dir) | |
| return list(root.glob("minfo*")) if root.exists() else [] | |
| def use_stub(ctx, **kw): | |
| ctx.app.telegraph = StubTelegraph(**kw) | |
| return ctx.app.telegraph | |
| def real_size(media_dir, name="sample.mkv"): | |
| return (media_dir / name).stat().st_size | |
| # --------------------------------------------------------------------------- | |
| # the flow | |
| # --------------------------------------------------------------------------- | |
| async def test_reply_to_a_video_and_the_telegraph_link_comes_back(ctx, media_dir, monkeypatch): | |
| calls = serve(monkeypatch, media_dir) | |
| tg = use_stub(ctx) | |
| video = document_message("Show - 05.mkv", real_size(media_dir), mid=70, caption="Dual Audio") | |
| video.date = datetime(2026, 10, 19, 21, 30, 12, tzinfo=timezone.utc) | |
| video.document.dc_id = 4 | |
| ev = await say(ctx, OWNER, "/mediainfo", reply=video) | |
| assert ev.texts == ["🔎 Reading the file…"] | |
| entity, text, kw = ctx.client.edited[-1] | |
| assert entity.id == 1001 # the "Reading…" message became the answer | |
| assert "🔎 <b>MediaInfo</b> · Show - 05.mkv" in text and URL in text and "Matroska" in text and "1 video · 3 audio · 2 subtitle" in text | |
| assert kw["parse_mode"] == "html" and kw["link_preview"] is False | |
| (button,), = kw["buttons"] | |
| assert button.type.url == URL and button.text.startswith("📄") | |
| (title, nodes), = tg.calls | |
| assert title == "MediaInfo – Show - 05.mkv" | |
| page = blocks(nodes) | |
| assert list(page) == ["General", "Video", "Audio #1", "Audio #2", "Audio #3", "Text #1", "Text #2", "Menu", "Telegram"] | |
| assert "Show - 05.mkv" in page["General"] and "probe.mkv" not in page["General"] # the file's own name, not the temp copy | |
| telegram = page["Telegram"] | |
| for line in ("File name : Show - 05.mkv", "MIME type : video/x-matroska", "Sent as : Video", | |
| "Sent on : 2026-10-19 21:30:12 UTC", "Data center : 4", "Caption : Dual Audio"): | |
| assert line in telegram, line | |
| assert len(calls) == 1 and calls[0]["max_bytes"] == appmod.TRACKS_PROBE_BYTES # the start of the file was enough | |
| assert leftovers(ctx) == [] | |
| async def test_without_a_file_there_is_a_hint_and_nothing_is_downloaded(ctx, monkeypatch): | |
| async def no_download(*a, **kw): | |
| raise AssertionError("there is nothing to read") | |
| monkeypatch.setattr(appmod.transfer, "download", no_download) | |
| tg = use_stub(ctx) | |
| for reply in (None, FakeMessage("just some text")): | |
| ev = await say(ctx, OWNER, "/mediainfo", reply=reply) | |
| assert HINT in ev.texts[0] and "Telegraph" in ev.texts[0] | |
| assert tg.calls == [] and ctx.client.edited == [] | |
| async def test_a_file_sent_with_the_command_as_its_caption_is_read_not_renamed(ctx, media_dir, monkeypatch): | |
| serve(monkeypatch, media_dir) | |
| tg = use_stub(ctx) | |
| for mid, caption in ((71, "/mediainfo"), (72, "/mediainfo@my_rename_bot")): | |
| ev = await send_doc(ctx, OWNER, "Show.mkv", size=real_size(media_dir), caption=caption, mid=mid) | |
| assert ev.texts == ["🔎 Reading the file…"] # no "Queued" message from the auto-renamer | |
| assert ctx.app.queue.waiting == 0 and len(tg.calls) == 2 | |
| assert "Caption" not in blocks(tg.calls[0][1])["Telegram"] # the caption was only the command | |
| async def test_other_captions_still_rename_the_file(ctx): | |
| ev = await send_doc(ctx, OWNER, "Show.mkv", size=5 * 1024 * 1024, caption="/mediainfoX", mid=73) | |
| assert ev.texts[0].startswith("📥 <b>Queued</b>") and ctx.app.queue.waiting == 1 | |
| async def test_a_video_whose_index_is_at_the_end_is_read_from_the_whole_file(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((media_dir / "sample.mp4").read_bytes()[:3000]) # an mp4 cut before its index | |
| else: | |
| shutil.copy(media_dir / "sample.mp4", dest) | |
| monkeypatch.setattr(appmod.transfer, "download", fake_download) | |
| tg = use_stub(ctx) | |
| await say(ctx, OWNER, "/mediainfo", reply=document_message("Movie.mp4", 40 * 1024 * 1024, mid=74, mime="video/mp4")) | |
| assert len(calls) == 2 and "max_bytes" in calls[0] and "max_bytes" not in calls[1] | |
| assert any("needs the whole file" in text for _e, text, _k in ctx.client.edited) | |
| (_title, nodes), = tg.calls | |
| assert list(blocks(nodes))[:2] == ["General", "Video"] and "No tracks" not in " ".join(notes(nodes)) | |
| assert URL in ctx.client.edited[-1][1] and leftovers(ctx) == [] | |
| async def test_a_video_with_no_readable_tracks_still_gets_a_page_that_says_so(ctx, monkeypatch, size, why): | |
| calls = serve_junk(monkeypatch) | |
| tg = use_stub(ctx) | |
| await say(ctx, OWNER, "/mediainfo", reply=document_message("Broken.mkv", size, mid=75)) | |
| assert len(calls) == 1 # too small to need more, or too big to fetch whole | |
| (_title, nodes), = tg.calls | |
| assert any(why in note for note in notes(nodes)) | |
| assert list(blocks(nodes)) == ["General", "Telegram"] and URL in ctx.client.edited[-1][1] | |
| assert leftovers(ctx) == [] | |
| async def test_a_file_that_is_not_media_is_read_from_its_start_only(ctx, monkeypatch): | |
| calls = serve_junk(monkeypatch, b"PK\x03\x04" + b"\x00" * 4000) | |
| tg = use_stub(ctx) | |
| await say(ctx, OWNER, "/mediainfo", reply=document_message("archive.zip", 50 * 1024 * 1024, mid=76, video=False, mime="application/zip")) | |
| assert len(calls) == 1 and "max_bytes" in calls[0] | |
| (_title, nodes), = tg.calls | |
| page = blocks(nodes) | |
| assert list(page) == ["General", "Telegram"] and "archive.zip" in page["General"] and "50.0 MiB" in page["General"] | |
| assert "Sent as : Document" in page["Telegram"] and "No tracks" not in " ".join(notes(nodes)) | |
| assert any("Only the start of the file was read" in note for note in notes(nodes)) | |
| async def test_a_download_that_fails_is_reported_and_cleaned_up(ctx, monkeypatch): | |
| async def fail(client, message, dest, **kw): | |
| Path(dest).write_bytes(b"partial") | |
| raise ConnectionError("connection reset") | |
| monkeypatch.setattr(appmod.transfer, "download", fail) | |
| tg = use_stub(ctx) | |
| await say(ctx, OWNER, "/mediainfo", reply=document_message("Show.mkv", 5 * 1024 * 1024, mid=77)) | |
| assert ctx.client.edited[-1][1] == "❌ I couldn't read that file." and tg.calls == [] and leftovers(ctx) == [] | |
| # --------------------------------------------------------------------------- | |
| # when Telegraph does not work | |
| # --------------------------------------------------------------------------- | |
| async def test_when_telegraph_fails_the_report_comes_as_a_text_file(ctx, media_dir, monkeypatch): | |
| serve(monkeypatch, media_dir) | |
| use_stub(ctx, error=TelegraphError("Telegraph asks to wait 5 seconds", "FLOOD_WAIT_5")) | |
| await say(ctx, OWNER, "/mediainfo", reply=document_message("Show: 05?.mkv", real_size(media_dir), mid=78)) | |
| (entity, file, kw), = [f for f in ctx.client.files if isinstance(f[1], io.BytesIO)] | |
| assert entity == OWNER and file.name == "Show_ 05_ - MediaInfo.txt" and kw["force_document"] is True | |
| assert "Telegraph asks to wait 5 seconds, so here it is as a text file." in kw["caption"] and kw["parse_mode"] == "html" | |
| text = file.getvalue().decode() | |
| assert text.startswith("General\n") and "\nVideo\n" in text and "\nTelegram\nFile name" in text and "Show: 05?.mkv" in text | |
| assert (OWNER, [1001]) in ctx.client.deleted # the "Reading…" message goes | |
| assert leftovers(ctx) == [] | |
| async def test_another_telegraph_failure_does_not_show_its_details(ctx, monkeypatch): | |
| serve_junk(monkeypatch) | |
| use_stub(ctx, error=RuntimeError("secret token abc123 rejected")) | |
| await say(ctx, OWNER, "/mediainfo", reply=document_message("Show.mkv", 2 * 1024 * 1024, mid=79)) | |
| (_e, _file, kw), = [f for f in ctx.client.files if isinstance(f[1], io.BytesIO)] | |
| assert "Telegraph is not available, so here it is as a text file." in kw["caption"] and "abc123" not in kw["caption"] | |
| async def test_an_unexpected_error_does_not_leave_the_reading_message_hanging(ctx, monkeypatch): | |
| def broken(message, caption=True): | |
| raise RuntimeError("boom") | |
| monkeypatch.setattr(appmod, "telegram_facts", broken) | |
| ev = await say(ctx, OWNER, "/mediainfo", reply=document_message("Show.mkv", 2 * 1024 * 1024, mid=80)) | |
| assert ev.texts == ["🔎 Reading the file…"] | |
| assert ctx.client.edited[-1][1] == "⚠️ Something went wrong while reading that file." | |
| # --------------------------------------------------------------------------- | |
| # access and plumbing | |
| # --------------------------------------------------------------------------- | |
| async def test_only_allowed_users_can_use_it(ctx, monkeypatch): | |
| tg = use_stub(ctx) | |
| ev = await say(ctx, STRANGER, "/mediainfo", reply=document_message("Show.mkv", 1024, mid=81)) | |
| assert ev.out == [] and tg.calls == [] | |
| async def test_the_command_is_in_the_menu(): | |
| commands = dict(appmod.COMMANDS) | |
| assert "mediainfo" in commands and 3 <= len(commands["mediainfo"]) <= 256 | |
| async def test_the_telegraph_token_lives_in_the_database(ctx): | |
| assert await ctx.app._load_telegraph_token() is None | |
| await ctx.app._save_telegraph_token("abc") | |
| assert await ctx.app._load_telegraph_token() == "abc" and await ctx.db.kv_get("telegraph_token") == "abc" | |
| await ctx.app._save_telegraph_token("") # what the client does when Telegraph rejects the token | |
| assert await ctx.db.kv_get("telegraph_token") is None | |
| async def test_the_whole_chain_with_the_real_client_and_a_fake_telegraph(ctx, media_dir, monkeypatch): | |
| serve(monkeypatch, media_dir, "sample.mp4") | |
| fake = await serve_fake_telegraph() | |
| try: | |
| ctx.app.telegraph = Telegraph(ctx.app._load_telegraph_token, ctx.app._save_telegraph_token, api=fake.url) | |
| ctx.app.telegraph.set_author("Rename Bot", "https://t.me/rename_bot") | |
| await say(ctx, OWNER, "/mediainfo", reply=document_message("Movie.mp4", real_size(media_dir, "sample.mp4"), mid=82, mime="video/mp4")) | |
| await say(ctx, OWNER, "/mediainfo", reply=document_message("Movie 2.mp4", real_size(media_dir, "sample.mp4"), mid=83, mime="video/mp4")) | |
| assert fake.accounts == 1 and len(fake.pages) == 2 # one account, kept for the second page | |
| assert await ctx.db.kv_get("telegraph_token") == "token1" | |
| page = fake.pages[0] | |
| assert page["title"] == "MediaInfo – Movie.mp4" and page["author_name"] == "Rename Bot" and page["author_url"] == "https://t.me/rename_bot" | |
| assert "Audio #2" in page["content"] and "Movie.mp4" in page["content"] | |
| assert "https://telegra.ph/Page-1" in ctx.client.edited[-2][1] and "https://telegra.ph/Page-2" in ctx.client.edited[-1][1] | |
| finally: | |
| await fake.server.close() | |