Spaces:
Paused
Paused
| """Rich messages through the bot: /new, the edits in My posts, the preview, /broadcast, the userbot, /swap, the sync and /repost.""" | |
| import asyncio | |
| from types import SimpleNamespace | |
| from telethon import functions, types | |
| from app.replace_engine import ScanOptions, apply_channel, scan_channel, undo_batch | |
| from app.repost_engine import RepostOptions, final_state, run_repost, stand_in | |
| from app.richmsg import Rich, entry, fingerprint, rich_of | |
| from app.suggest import build_suggestion | |
| from app.sync_engine import diff_post, live_fields, message_supported, sync_channel | |
| from app.tgutil import button_url, pack_file_id | |
| from .fakes import FakeChannelClient, make_msg, markup, photo_media, rpc, url_btn, video_media | |
| from .harness import App, norm | |
| from .test_broadcast import broadcast, make_app as broadcast_app | |
| from .test_engine import setup as engine_setup | |
| from .test_flows import make_app | |
| from .test_markup_flows import msg, new_post | |
| from .test_replace_userbot import BUTTONS, CID, MID, make_app as userbot_app, replacement, user_edits | |
| from .test_repost import NEW, OLD, edit_requests, forwarded_ids, setup as repost_setup | |
| from .test_rich import blocks, rich_msg, same | |
| from .test_sync import old_msg, opts, world | |
| HEADING = "# Release notes\n\n- [x] fixed\n- [ ] todo\n\n| a | b |\n|---|---|\n| 1 | 2 |" | |
| TABLE = '<h2>Schedule</h2><table bordered><tr><th>Day</th><th>Show</th></tr><tr><td>Mon</td><td>Naruto</td></tr></table>' | |
| def native(rm=None, mid=77): | |
| """A rich message sent to the bot (or forwarded to it).""" | |
| return SimpleNamespace( | |
| raw_text="", entities=[], media=None, grouped_id=None, forward=None, rich_message=rm or rich_msg(), id=mid | |
| ) | |
| def raw_edits(app): | |
| """The messages.editMessage requests the bot sent, each once.""" | |
| out = [] | |
| for e in app.tg.edits: | |
| r = e[2] if isinstance(e, tuple) and e[0] == "raw" else e | |
| if isinstance(r, functions.messages.EditMessageRequest) and not any(r is o for o in out): | |
| out.append(r) | |
| return out | |
| def rich_sends(app): | |
| return [e for e in app.tg.sent if e[0] == "rich"] | |
| def md_post(source): | |
| return dict(text="x", entities=[entry(Rich("markdown", source))]) | |
| # ============================================================================================== typed text: /new | |
| def test_markdown_with_a_heading_is_published_as_a_rich_message(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, prompt = await new_post(app, HEADING) | |
| assert "Rich message read from Markdown" in prompt and "/rich" in prompt and "buttons" in prompt.lower() | |
| post = await app.db.get_post(pid) | |
| assert post.status == "sent" and post.entities == [{"k": "rich", "f": "markdown", "s": HEADING}] | |
| assert post.text.startswith("Release notes\n") and "☑ fixed" in post.text and "☐ todo" in post.text | |
| kind, peer, inp, kw = app.tg.sent[-1] | |
| assert kind == "rich" and isinstance(peer, types.InputPeerChannel) and peer.channel_id == 1 | |
| assert isinstance(inp, types.InputRichMessageMarkdown) and inp.markdown == HEADING and not inp.rtl | |
| assert kw["reply_markup"] is None | |
| live = app.tg.msgs[post.message_id] # Telegram made a rich message of it | |
| assert live.rich_message is not None and live.message == "" | |
| assert [e for e in app.tg.sent if e[0] in ("text", "file")] == [] # nothing went out as an ordinary message | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_html_with_a_table_is_published_as_a_rich_message(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, prompt = await new_post(app, TABLE) | |
| assert "Rich message read from HTML" in prompt | |
| post = await app.db.get_post(pid) | |
| assert post.entities == [{"k": "rich", "f": "html", "s": TABLE}] | |
| assert post.text == "Schedule\nDay | Show\nMon | Naruto" | |
| kind, peer, inp, kw = app.tg.sent[-1] | |
| assert kind == "rich" and isinstance(inp, types.InputRichMessageHTML) and inp.html == TABLE | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_the_rich_marker_turns_any_text_into_a_rich_message_with_its_options(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, prompt = await new_post(app, "<rich rtl noautolink>\nhello **x** #tag @name") | |
| assert "Rich message read from Markdown" in prompt | |
| post = await app.db.get_post(pid) | |
| assert post.text == "hello x #tag @name" | |
| assert post.entities == [{"k": "rich", "f": "markdown", "s": "hello **x** #tag @name", "r": 1, "n": 1}] | |
| kind, peer, inp, kw = app.tg.sent[-1] | |
| assert isinstance(inp, types.InputRichMessageMarkdown) and inp.markdown == "hello **x** #tag @name" | |
| assert inp.rtl is True and inp.noautolink is True | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_text_that_only_looks_a_little_like_markup_stays_an_ordinary_post(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| for text in ("#anime #naruto new episode", "Price: $5 and $6 - a == b - I <3 it\n- one\n- two\n---", "**bold** [x](https://t.me/x)"): | |
| pid, prompt = await new_post(app, text) | |
| assert "Rich message" not in prompt | |
| assert rich_of(await app.db.get_post(pid)) is None | |
| assert app.tg.sent[-1][0] == "text" | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_markup_input_off_keeps_typed_blocks_literal(tmp_path): | |
| async def run(): | |
| app = App(tmp_path, markup_input=False) | |
| await app.start() | |
| await app.db.save_channel(1, 5, "Anime Channel", "animech", 1) | |
| pid, prompt = await new_post(app, HEADING) | |
| assert "Rich message" not in prompt | |
| kind, peer, text, kw = app.tg.sent[-1] | |
| assert kind == "text" and text == HEADING and kw["formatting_entities"] == [] | |
| # a rich message made in Telegram is not typed text: it is still kept | |
| pid, prompt = await new_post(app, "", message=native()) | |
| assert "Rich message kept as it is" in prompt and app.tg.sent[-1][0] == "rich" | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_typed_rich_message_that_is_too_long_is_refused(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| await app.text("/new") | |
| await app.text("# Title\n" + "x" * 33000, message=msg("# Title\n" + "x" * 33000)) | |
| assert "32768" in norm(app.out.last_text) | |
| assert (await app.db.list_posts(1, None, 0, 10))[1] == 0 # nothing was saved | |
| await app.db.close() | |
| asyncio.run(run()) | |
| # =================================================================================== a rich message sent to the bot | |
| def test_a_rich_message_sent_to_the_bot_is_published_block_for_block(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| rm = rich_msg(photos=[photo_media(5).photo], documents=[video_media(81).document], rtl=True) | |
| pid, prompt = await new_post(app, "", message=native(rm)) | |
| assert "Rich message kept as it is" in prompt | |
| post = await app.db.get_post(pid) | |
| (e,) = post.entities | |
| assert (e["k"], e["f"], e["r"], e["p"]) == ("rich", "blocks", 1, fingerprint(rm)) and post.text == "Title\nHello" | |
| kind, peer, inp, kw = app.tg.sent[-1] | |
| assert kind == "rich" and isinstance(inp, types.InputRichMessage) and inp.rtl | |
| assert same(inp.blocks, rm.blocks) | |
| assert [(p.id, p.access_hash, p.file_reference) for p in inp.photos] == [(5, 35, b"ref5")] | |
| assert [d.id for d in inp.documents] == [81] | |
| live = app.tg.msgs[post.message_id].rich_message | |
| assert same(live.blocks, rm.blocks) and [p.id for p in live.photos] == [5] | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_rich_message_that_arrives_in_parts_is_asked_for_whole(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| part = rich_msg([types.PageBlockParagraph(types.TextPlain("first"))], part=True) | |
| app.tg.full_rich = {77: rich_msg(blocks("whole"))} | |
| pid, prompt = await new_post(app, "", message=native(part)) | |
| assert "Rich message kept as it is" in prompt | |
| assert (await app.db.get_post(pid)).text == "Title\nwhole" | |
| assert any(isinstance(r, functions.messages.GetRichMessageRequest) and r.id == 77 for r in app.tg.requests) | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_rich_message_that_cannot_be_had_whole_makes_no_post(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| part = rich_msg([types.PageBlockParagraph(types.TextPlain("first"))], part=True) | |
| app.tg.full_rich = {77: rich_msg(blocks(), part=True)} | |
| await app.text("/new") | |
| await app.text("", message=native(part)) | |
| assert "part of that rich message" in norm(app.out.last_text) | |
| assert (await app.db.list_posts(1, None, 0, 10))[1] == 0 | |
| assert app.ctx.state[app.uid]["mode"] == "new_content" # still waiting for the message | |
| await app.text("hello", message=msg("hello")) # and a better one is taken | |
| assert (await app.db.list_posts(1, None, 0, 10))[1] == 1 | |
| await app.db.close() | |
| asyncio.run(run()) | |
| # ================================================================================ buttons, preview, publishing | |
| def test_the_link_buttons_of_a_post_go_with_its_rich_message(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| await app.text("/new") | |
| await app.text(HEADING, message=msg(HEADING)) | |
| await app.text("Download - https://t.me/x?start=a | Other - https://example.com", message=msg("x")) | |
| pid = int(app.out.callback_data("Publish").split(":")[1]) | |
| await app.press(f"pb:{pid}") | |
| await app.press(f"pby:{pid}") | |
| kind, peer, inp, kw = app.tg.sent[-1] | |
| rows = kw["reply_markup"].rows | |
| assert kind == "rich" and [b.text for b in rows[0].buttons] == ["Download", "Other"] | |
| assert [button_url(b) for b in rows[0].buttons] == ["https://t.me/x?start=a", "https://example.com"] | |
| assert app.tg.msgs[(await app.db.get_post(pid)).message_id].reply_markup is not None | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_the_preview_of_a_rich_draft_goes_to_the_owner_and_not_to_the_channel(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, HEADING, publish=False) | |
| await app.press(f"pv:{pid}") | |
| kind, peer, inp, kw = app.tg.sent[-1] | |
| assert kind == "rich" and isinstance(peer, types.InputPeerUser) and inp.markdown == HEADING | |
| assert app.tg.msgs == {} and (await app.db.get_post(pid)).status == "draft" | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_telegram_not_being_able_to_make_the_message_is_explained_in_the_preview_and_when_publishing(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, "<rich>\n<tg-map lat=\"x\"/>", publish=False) | |
| app.tg.fail["SendMessageRequest"] = rpc("MessageEmptyError") | |
| await app.press(f"pv:{pid}") | |
| assert "see /rich for the syntax" in norm(app.out.last_text) | |
| await app.press(f"pb:{pid}") | |
| await app.press(f"pby:{pid}") | |
| assert "NOT published" in norm(app.out.last_text) | |
| assert (await app.db.get_post(pid)).status == "draft" and app.tg.msgs == {} | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_the_panel_of_a_rich_post_has_no_media_or_link_preview(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, HEADING, publish=False) | |
| await app.press(f"po:{pid}") | |
| text = norm(app.out.last_text) | |
| assert "Type: 🧩 rich message (typed as Markdown)" in text and "Release notes" in text | |
| labels = [b for row in app.out.last_buttons() for b in row] | |
| assert "✏️ Text" in labels and "👁 Preview" in labels and "➕ Button" in labels | |
| assert not any("Media" in b or "Link preview" in b for b in labels if b) | |
| native_pid, _ = await new_post(app, "", message=native(), publish=False) | |
| assert "made in Telegram" in norm((await _panel(app, native_pid))) | |
| await app.db.close() | |
| asyncio.run(run()) | |
| async def _panel(app, pid): | |
| await app.press(f"po:{pid}") | |
| return app.out.last_text | |
| # ====================================================================================== editing in My posts | |
| def test_the_text_of_a_published_rich_post_is_changed_with_a_new_rich_message(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, "# One\n\ntext") | |
| before = app.tg.msgs[(await app.db.get_post(pid)).message_id].rich_message | |
| await app.press(f"et:{pid}") | |
| assert "/rich" in norm(app.out.last_text) | |
| new = "# Two\n\n- [ ] a" | |
| await app.text(new, message=msg(new)) | |
| (req,) = raw_edits(app) | |
| assert isinstance(req.rich_message, types.InputRichMessageMarkdown) and req.rich_message.markdown == new | |
| assert req.message is None and not req.entities | |
| post = await app.db.get_post(pid) | |
| assert post.entities == [{"k": "rich", "f": "markdown", "s": new}] and post.text == "Two\n\n☐ a" | |
| assert bytes(app.tg.msgs[post.message_id].rich_message) != bytes(before) # the channel shows the new one | |
| assert "Rich message read from Markdown" in norm(app.out.last_text) | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_native_rich_message_can_replace_the_text_of_a_post(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, "plain words") | |
| await app.press(f"et:{pid}") | |
| await app.text("", message=native(rich_msg(blocks("brand new")))) | |
| (req,) = raw_edits(app) | |
| assert isinstance(req.rich_message, types.InputRichMessage) and req.message is None | |
| post = await app.db.get_post(pid) | |
| assert rich_of(post).fmt == "blocks" and post.text == "Title\nbrand new" | |
| assert "Rich message kept as it is" in norm(app.out.last_text) | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_plain_post_becomes_a_rich_one_and_a_rich_one_a_plain_one_again(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, "hello") | |
| await app.press(f"et:{pid}") | |
| await app.text(HEADING, message=msg(HEADING)) | |
| assert isinstance(raw_edits(app)[-1].rich_message, types.InputRichMessageMarkdown) | |
| assert rich_of(await app.db.get_post(pid)) is not None | |
| await app.press(f"et:{pid}") | |
| await app.text("plain again", message=msg("plain again")) | |
| last = raw_edits(app)[-1] | |
| assert last.message == "plain again" and last.rich_message is None | |
| post = await app.db.get_post(pid) | |
| assert rich_of(post) is None and post.text == "plain again" and post.entities == [] | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_draft_can_be_changed_into_a_rich_message_before_it_is_published(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, "first", publish=False) | |
| await app.press(f"et:{pid}") | |
| await app.text(TABLE, message=msg(TABLE)) | |
| post = await app.db.get_post(pid) | |
| assert post.status == "draft" and rich_of(post).fmt == "html" and raw_edits(app) == [] | |
| await app.press(f"pb:{pid}") | |
| await app.press(f"pby:{pid}") | |
| assert app.tg.sent[-1][0] == "rich" and app.tg.sent[-1][2].html == TABLE | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_the_buttons_of_a_published_rich_post_are_changed_without_losing_the_message(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, HEADING) | |
| await app.press(f"ba:{pid}") | |
| await app.text("Watch", message=msg("Watch")) | |
| await app.text("https://example.com/w", message=msg("https://example.com/w")) | |
| req = raw_edits(app)[-1] | |
| assert isinstance(req.rich_message, types.InputRichMessageMarkdown) and req.rich_message.markdown == HEADING | |
| assert [button_url(b) for b in req.reply_markup.rows[0].buttons] == ["https://example.com/w"] | |
| post = await app.db.get_post(pid) | |
| assert rich_of(post) is not None and post.buttons == [[{"t": "Watch", "u": "https://example.com/w"}]] | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_rich_post_has_no_media_slot(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, HEADING) | |
| await app.press(f"em:{pid}") | |
| assert "has no media slot" in norm(app.out.last_text) | |
| assert app.ctx.state.get(app.uid) is None or app.ctx.state[app.uid]["mode"] != "edit_media" | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_caption_is_never_a_rich_message(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| post = await app.db.create_post( | |
| channel_id=1, message_id=7, status="sent", source="bot", text="old", entities=[], buttons=[], | |
| media_kind="photo", media_file_id=pack_file_id(make_msg(1, "", media=photo_media(3))), | |
| ) | |
| await app.press(f"et:{post.id}") | |
| assert "Rich messages" not in norm(app.out.last_text) | |
| await app.text("", message=native()) # a rich message as a caption | |
| assert "caption can't be a rich message" in norm(app.out.last_text) | |
| await app.text("# Heading\n\n- [ ] task", message=msg("# Heading\n\n- [ ] task")) # typed blocks stay words | |
| saved_post = await app.db.get_post(post.id) | |
| assert rich_of(saved_post) is None and saved_post.text.startswith("# Heading") and saved_post.media_kind == "photo" | |
| assert all(r.rich_message is None for r in raw_edits(app)) | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_rich_message_can_not_be_made_a_media_post_by_the_media_edit(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| await app.text("/new") | |
| await app.text("", message=SimpleNamespace(raw_text="", entities=[], media=photo_media(5), grouped_id=None, forward=None, rich_message=rich_msg())) | |
| post = (await app.db.list_posts(1, None, 0, 10))[0][0] | |
| assert post.media_kind == "photo" and rich_of(post) is None # a photo wins: the message is not a rich post | |
| await app.db.close() | |
| asyncio.run(run()) | |
| # =================================================================================================== /broadcast | |
| def test_a_broadcast_can_be_a_rich_message(tmp_path): | |
| async def run(): | |
| app = await broadcast_app(tmp_path) | |
| await app.text("/broadcast 1h") | |
| assert "/rich" in norm(app.out.last_text) | |
| text = "# Maintenance\n\n- tonight 22:00\n- [ ] backups" | |
| await app.text(text, message=msg(text)) | |
| assert "Rich message read from Markdown" in norm(app.out.last_text) | |
| await app.press(app.out.callback_data("No buttons")) | |
| assert rich_sends(app)[-1][1].__class__ is types.InputPeerUser # the preview was made as a rich message too | |
| await app.press(app.out.callback_data("Send to all channels")) | |
| for cid in (1, 2): | |
| chan = app.tg.registry[cid] | |
| (live,) = [m for m in chan.msgs.values() if m.rich_message is not None] | |
| assert isinstance(live.rich_input, types.InputRichMessageMarkdown) and live.rich_input.markdown == text | |
| (p,) = await app.db.sent_posts(cid) | |
| assert p.message_id == live.id and p.entities == [{"k": "rich", "f": "markdown", "s": text}] | |
| assert p.text.startswith("Maintenance") | |
| assert len(await app.db.pending_deletes()) == 2 # it is still deleted when its time is up | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_broadcast_with_a_native_rich_message_and_buttons(tmp_path): | |
| async def run(): | |
| app = await broadcast_app(tmp_path) | |
| await broadcast(app, "/broadcast 1h", "", buttons="Open - https://t.me/x", message=native()) | |
| for cid in (1, 2): | |
| (live,) = [m for m in app.tg.registry[cid].msgs.values() if m.rich_message is not None] | |
| assert same(live.rich_message.blocks, blocks()) | |
| assert [button_url(b) for b in live.reply_markup.rows[0].buttons] == ["https://t.me/x"] | |
| await app.db.close() | |
| asyncio.run(run()) | |
| # ======================================================================================================== /replace | |
| def test_replace_leaves_a_rich_post_alone(tmp_path): | |
| async def run(): | |
| app, chan, user = await userbot_app(tmp_path) | |
| live = chan.msgs[MID] | |
| live.media, live.message, live.rich_message = None, "", rich_msg() # the channel's post is a rich message | |
| await app.text(f"/replace https://t.me/c/{CID}/{MID}", reply=replacement()) | |
| assert "is a rich message" in norm(app.out.last_text) | |
| assert user_edits(user) == [] and not any(isinstance(r, functions.messages.EditMessageRequest) for r in chan.requests) | |
| assert chan.msgs[MID].rich_message is not None | |
| await app.db.close() | |
| asyncio.run(run()) | |
| # ================================================================================================ the userbot | |
| def test_a_rich_edit_that_telegram_refuses_the_bot_is_made_by_the_userbot(tmp_path): | |
| async def run(): | |
| app, chan, user = await userbot_app(tmp_path) # the post belongs to another bot: the bot may not edit it | |
| live = chan.msgs[MID] | |
| live.media, live.message, live.rich_message = None, "", chan._rich(types.InputRichMessageMarkdown(markdown="# old")) | |
| post = await app.db.create_post( | |
| channel_id=CID, message_id=MID, status="sent", source="adopted", text="old", | |
| entities=[entry(Rich("markdown", "# old"))], buttons=BUTTONS, | |
| ) | |
| await app.press(f"et:{post.id}") | |
| new = "# New\n\n- [x] a" | |
| await app.text(new, message=msg(new)) | |
| (req,) = user_edits(user) | |
| assert isinstance(req.rich_message, types.InputRichMessageMarkdown) and req.rich_message.markdown == new | |
| assert req.message is None and req.reply_markup is None # a user account sends no keyboard | |
| assert isinstance(chan.msgs[MID].rich_message, types.RichMessage) and chan.msgs[MID].rich_input.markdown == new | |
| assert "the userbot made the edit" in norm(app.out.last_text) and "Rich message read from Markdown" in norm(app.out.last_text) | |
| saved_post = await app.db.get_post(post.id) | |
| assert saved_post.entities == [{"k": "rich", "f": "markdown", "s": new}] and saved_post.buttons == BUTTONS | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_the_userbot_editor_can_set_a_rich_message(): | |
| from .test_userbot import channel | |
| from .test_userbot_edit import ready_editor | |
| async def go(): | |
| bot = channel() | |
| ub, made, e = await ready_editor(bot) | |
| await e.edit(5, text="", entities=[], rich=Rich("html", "<h1>New</h1>")) | |
| await e.close() | |
| req = bot.user_edits[-1] | |
| assert isinstance(req.rich_message, types.InputRichMessageHTML) and req.rich_message.html == "<h1>New</h1>" | |
| assert req.message is None and bot.msgs[5].rich_message is not None and bot.msgs[5].message == "" | |
| asyncio.run(go()) | |
| # ============================================================================================ /swap and /handleswap | |
| OLD_LINK = "https://t.me/bharath?start=AAA" | |
| def live_rich_post(i=1, *, out=True): | |
| m = make_msg(i, "", markup=markup([url_btn("Get", OLD_LINK)]), out=out) | |
| m.rich_message = rich_msg( | |
| [types.PageBlockParagraph(types.TextUrl(types.TextPlain("t.me/bharath"), "https://t.me/bharath?start=ZZZ", 0))] | |
| ) | |
| return m | |
| def test_swap_changes_only_the_buttons_of_a_rich_post_and_undo_puts_them_back(tmp_path): | |
| async def run(): | |
| live = live_rich_post() | |
| db, ch, client = await engine_setup(tmp_path, {1: live}) | |
| typed = Rich("markdown", "# Title\n\n@bharath t.me/bharath") | |
| await db.create_post( | |
| channel_id=1, message_id=1, status="sent", source="bot", text="Title", entities=[entry(typed)], | |
| buttons=[[{"t": "Get", "u": OLD_LINK}]], | |
| ) | |
| o = ScanOptions("bharath", "goku") | |
| scan = await scan_channel(client, ch, o, can_edit_others=True) | |
| assert set(scan.infos) == {1} | |
| await db.create_batch("b1", "bharath", "goku", 99) | |
| res = await apply_channel(client, db, scan, o, "b1", 99, edit_delay=0) | |
| assert (res.edited, res.failed) == (1, []) | |
| req = client.edits[-1] | |
| assert req.message is None and req.rich_message is None # only the keyboard was sent | |
| assert button_url(live.reply_markup.rows[0].buttons[0]) == "https://t.me/goku?start=AAA" | |
| assert live.message == "" and live.rich_message is not None | |
| p = await db.find_by_message(1, 1) | |
| assert rich_of(p) == typed and p.text == "Title" and p.buttons == [[{"t": "Get", "u": "https://t.me/goku?start=AAA"}]] | |
| ures = await undo_batch(client, db, "b1", edit_delay=0) | |
| assert ures.restored == 1 and not ures.failed | |
| assert client.edits[-1].message is None and client.edits[-1].rich_message is None | |
| assert button_url(live.reply_markup.rows[0].buttons[0]) == OLD_LINK | |
| p = await db.find_by_message(1, 1) | |
| assert rich_of(p) == typed and p.text == "Title" and p.buttons == [[{"t": "Get", "u": OLD_LINK}]] | |
| await db.close() | |
| asyncio.run(run()) | |
| def test_a_rich_post_nobody_saved_is_adopted_by_a_swap_as_its_blocks(tmp_path): | |
| async def run(): | |
| live = live_rich_post() | |
| db, ch, client = await engine_setup(tmp_path, {1: live}) | |
| o = ScanOptions("bharath", "goku") | |
| scan = await scan_channel(client, ch, o, can_edit_others=True) | |
| await db.create_batch("b1", "bharath", "goku", 99) | |
| await apply_channel(client, db, scan, o, "b1", 99, edit_delay=0) | |
| p = await db.find_by_message(1, 1) | |
| r = rich_of(p) | |
| assert p.source == "adopted" and r.fmt == "blocks" and r.fp == fingerprint(live.rich_message) | |
| assert p.text == "t.me/bharath" and p.buttons == [[{"t": "Get", "u": "https://t.me/goku?start=AAA"}]] | |
| await db.close() | |
| asyncio.run(run()) | |
| def test_swaps_leave_the_text_of_a_rich_post_alone(tmp_path): | |
| async def run(): | |
| for mode in ("links", "handles"): | |
| live = make_msg(1, "") | |
| live.rich_message = rich_msg([types.PageBlockParagraph(types.TextPlain("@bharath https://t.me/bharath?start=A"))]) | |
| db, ch, client = await engine_setup(tmp_path, {1: live}) | |
| scan = await scan_channel(client, ch, ScanOptions("bharath", "goku", mode=mode), can_edit_others=True) | |
| assert scan.infos == {} # no button to change: nothing to do | |
| await db.close() | |
| asyncio.run(run()) | |
| def test_a_handleswap_renames_the_buttons_of_a_rich_post_only(tmp_path): | |
| async def run(): | |
| live = make_msg(1, "", markup=markup([url_btn("Join @bharath", "https://example.com")]), out=True) | |
| live.rich_message = rich_msg() | |
| db, ch, client = await engine_setup(tmp_path, {1: live}) | |
| o = ScanOptions("bharath", "goku", mode="handles") | |
| scan = await scan_channel(client, ch, o, can_edit_others=True) | |
| await db.create_batch("b1", "bharath", "goku", 99) | |
| await apply_channel(client, db, scan, o, "b1", 99, edit_delay=0) | |
| assert live.reply_markup.rows[0].buttons[0].text == "Join @goku" and client.edits[-1].message is None | |
| assert live.rich_message is not None | |
| await db.close() | |
| asyncio.run(run()) | |
| # ===================================================================================================== the sync | |
| class SavedLike: | |
| """Anything with the Post fields.""" | |
| def __init__(self, m, **kw): | |
| for k, v in live_fields(m).items(): | |
| setattr(self, k, v) | |
| self.link_preview = False | |
| for k, v in kw.items(): | |
| setattr(self, k, v) | |
| def rich_live(i, word="Hello", **kw): | |
| m = old_msg(i, "", **kw) | |
| m.rich_message = rich_msg(blocks(word)) | |
| return m | |
| def test_a_saved_rich_post_that_is_unchanged_shows_no_difference(): | |
| m = rich_live(1) | |
| p = SavedLike(m) | |
| d = diff_post(p, m) | |
| assert d.fields == {} and not d.quiet and not d.buttons_missing | |
| assert message_supported(m) and not message_supported(old_msg(2, "")) | |
| def test_a_rich_post_edited_in_the_channel_is_taken_over_as_its_new_blocks(): | |
| m = rich_live(1, "Hello") | |
| p = SavedLike(m) | |
| edited = rich_live(1, "Changed") | |
| d = diff_post(p, edited) | |
| assert set(d.fields) == {"text", "entities"} and d.fields["text"] == "Title\nChanged" | |
| (e,) = d.fields["entities"] | |
| assert e["f"] == "blocks" and e["p"] == fingerprint(edited.rich_message) and not d.quiet | |
| def test_the_bots_own_typed_rich_post_is_only_noted_the_first_time_the_sync_sees_it(): | |
| m = rich_live(1) | |
| typed = Rich("markdown", HEADING) | |
| p = SavedLike(m, text="Release notes", entities=[entry(typed)]) | |
| d = diff_post(p, m) | |
| assert d.quiet and set(d.fields) == {"entities"} | |
| (e,) = d.fields["entities"] | |
| assert (e["f"], e["s"], e["p"]) == ("markdown", HEADING, fingerprint(m.rich_message)) # its source is kept, the fingerprint added | |
| again = SavedLike(m, text="Release notes", entities=d.fields["entities"]) | |
| assert diff_post(again, m).fields == {} | |
| changed = rich_live(1, "Edited by an admin") | |
| assert set(diff_post(again, changed).fields) == {"text", "entities"} | |
| def test_a_rich_post_that_became_an_ordinary_one_or_the_other_way_round_is_followed(): | |
| rich = rich_live(1) | |
| plain = old_msg(1, "now just words") | |
| d = diff_post(SavedLike(rich), plain) | |
| assert d.fields == {"text": "now just words", "entities": []} | |
| d = diff_post(SavedLike(plain), rich) | |
| assert d.fields["text"] == "Title\nHello" and d.fields["entities"][0]["f"] == "blocks" | |
| empty = old_msg(1, "") # Telegram shows nothing this time: that is not an edit | |
| assert diff_post(SavedLike(rich), empty).fields == {} | |
| def test_the_sync_saves_a_rich_post_it_finds_and_leaves_it_alone_afterwards(tmp_path): | |
| async def run(): | |
| chan, db, ch = await world(tmp_path, {1: old_msg(1, "plain"), 2: rich_live(2)}, save=False) | |
| rep = await sync_channel(chan, db, ch, opts()) | |
| p = await db.find_by_message(1, 2) | |
| assert p is not None and rich_of(p).fmt == "blocks" and p.text == "Title\nHello" and p.source == "adopted" | |
| again = await sync_channel(chan, db, ch, opts()) | |
| assert again.edited == [] and again.adopted == [] and rep.error is None | |
| chan.msgs[2].rich_message = rich_msg(blocks("Later")) # somebody edits it in the channel | |
| third = await sync_channel(chan, db, ch, opts()) | |
| assert third.edited == [2] and (await db.find_by_message(1, 2)).text == "Title\nLater" | |
| await db.close() | |
| asyncio.run(run()) | |
| # ===================================================================================================== /repost | |
| def rich_channel(): | |
| welcome = make_msg(1, "Welcome") | |
| same_post = make_msg(2, "") | |
| same_post.rich_message = rich_msg(blocks("one")) | |
| with_button = make_msg(3, "", markup=markup([url_btn("Get", f"https://t.me/{OLD}?start=AAA"), url_btn("Other", "https://example.com/x")])) | |
| with_button.rich_message = rich_msg(blocks("two"), photos=[photo_media(5).photo]) | |
| return {1: welcome, 2: same_post, 3: with_button} | |
| def test_a_repost_copies_rich_posts_and_rebuilds_the_one_with_buttons(tmp_path): | |
| async def run(): | |
| db, ch, chan, mig = await repost_setup(tmp_path, FakeChannelClient(rich_channel()), include_typed=True) | |
| res = await run_repost(chan, db, ch, mig, 99, delay=0, settle_pause=0) | |
| assert (res.copied_units, res.failed, res.aborted) == (3, [], None) | |
| assert forwarded_ids(chan) == [[1], [2]] and edit_requests(chan) == [] | |
| new = {i: m for i, m in chan.msgs.items() if i > 3} | |
| assert list(new) == [4, 5, 6] | |
| assert new[4].message == "Welcome" | |
| assert same(new[5].rich_message.blocks, blocks("one")) # a copy by Telegram: still the same rich message | |
| m = new[6] # the post with buttons is created again, with the link swapped | |
| assert isinstance(m.rich_input, types.InputRichMessage) and same(m.rich_input.blocks, blocks("two")) | |
| assert [(p.id, p.file_reference) for p in m.rich_input.photos] == [(5, b"ref5")] | |
| rows = m.reply_markup.rows | |
| assert [button_url(b) for b in rows[0].buttons] == [f"https://t.me/{NEW}?start=AAA", "https://example.com/x"] | |
| saved_row = await db.find_by_message(1, 6) | |
| assert saved_row.text == "Title\ntwo" and rich_of(saved_row).fmt == "blocks" | |
| assert saved_row.buttons[0][0]["u"] == f"https://t.me/{NEW}?start=AAA" | |
| assert rich_of(await db.find_by_message(1, 5)) is not None | |
| await db.close() | |
| asyncio.run(run()) | |
| def test_a_channel_that_forbids_forwarding_gets_its_rich_posts_rebuilt(tmp_path): | |
| async def run(): | |
| chan = FakeChannelClient(rich_channel()) | |
| chan.restrict_forwards = True | |
| db, ch, chan, mig = await repost_setup(tmp_path, chan) | |
| res = await run_repost(chan, db, ch, mig, 99, delay=0, settle_pause=0) | |
| assert (res.copied_units, res.failed) == (3, []) and forwarded_ids(chan) == [[1]] # only the first, refused, try | |
| new = {i: m for i, m in chan.msgs.items() if i > 3} | |
| assert same(new[5].rich_input.blocks, blocks("one")) and same(new[6].rich_input.blocks, blocks("two")) | |
| assert new[5].reply_markup is None and new[6].reply_markup is not None | |
| await db.close() | |
| asyncio.run(run()) | |
| def test_a_post_held_back_by_telegram_is_rebuilt_from_its_saved_rich_message(): | |
| row = SimpleNamespace( | |
| text="Release notes", entities=[entry(Rich("markdown", HEADING))], buttons=[], media_file_id=None, link_preview=False | |
| ) | |
| live = make_msg(7, "This message couldn't be displayed on your device due to copyright infringement.") | |
| final = final_state(stand_in(live, row), RepostOptions(OLD, NEW)) | |
| assert final.rich == Rich("markdown", HEADING) and final.text == "" and final.entities == [] | |
| assert final_state(make_msg(8, "plain"), RepostOptions(OLD, NEW)).rich is None | |
| # ============================================================================================== everything else | |
| def test_a_rich_post_is_kept_by_an_export_and_an_import(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, HEADING) | |
| data = await app.db.export_all() | |
| (p,) = [x for x in data["posts"] if x["id"] == pid] | |
| assert p["entities"] == [{"k": "rich", "f": "markdown", "s": HEADING}] | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_the_next_post_suggestion_leaves_rich_posts_out(): | |
| post = SimpleNamespace( | |
| text="x", entities=[entry(Rich("markdown", HEADING))], buttons=[[{"t": "Download Episodes 1 to 20", "u": "https://t.me/x?start=a"}]], | |
| ) | |
| assert build_suggestion(post) is None | |
| def test_the_rich_command_and_the_help_pages(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| await app.text("/rich") | |
| said = [norm(t) for kind, t, _ in app.out.log if kind == "say"] | |
| assert len(said) == 2 and "Rich messages" in said[0] and "<tg-map" in said[1] and "# Heading" in said[0] | |
| await app.text("/help") | |
| assert "/rich" in norm(app.out.last_text) | |
| await app.text("/formatting") | |
| assert "/rich" in norm(app.out.last_text) | |
| await app.db.close() | |
| asyncio.run(run()) | |