Spaces:
Paused
Paused
| """A rich message that was made in Telegram and sent to the bot has to be sent again as blocks - and Telegram refuses blocks it | |
| can't take back (RICH_MESSAGE_BLOCK_INVALID). What the bot converts, how it finds the blocks that are refused, and what the | |
| owner is told.""" | |
| import asyncio | |
| from telethon import errors, functions, types | |
| from app.richmsg import ( | |
| Rich, | |
| block_count, | |
| block_labels, | |
| block_name, | |
| digest, | |
| fingerprint, | |
| input_of, | |
| inputify, | |
| is_block_invalid, | |
| pack_blocks, | |
| plain_of, | |
| refused_blocks, | |
| rich_from_received, | |
| rich_of, | |
| to_input, | |
| unpack_blocks, | |
| without_blocks, | |
| ) | |
| from app.tgutil import _b64, explain_rpc | |
| from .fakes import rpc | |
| from .harness import norm | |
| from .test_broadcast import make_app as broadcast_app | |
| from .test_flows import make_app | |
| from .test_markup_flows import new_post | |
| from .test_rich import blocks, rich_msg, same | |
| from .test_rich_flows import native, rich_sends | |
| T = types.TextPlain | |
| PEER = types.InputPeerUser(1, 0) | |
| def block_invalid(): | |
| return errors.BadRequestError( | |
| functions.messages.SendMessageRequest(peer=types.InputPeerSelf(), message=""), "RICH_MESSAGE_BLOCK_INVALID", 400 | |
| ) | |
| def a_map(access_hash=99): | |
| """A map block the way Telegram hands it out: it holds a stored point.""" | |
| return types.PageBlockMap( | |
| types.GeoPoint(long=12.5, lat=41.9, access_hash=access_hash, accuracy_radius=7), 14, 600, 300, | |
| types.PageCaption(T("Rome"), T("")), | |
| ) | |
| def with_bad(n=5, bad=(1, 3)): | |
| """`n` blocks, the ones numbered `bad` are blocks Telegram will not take back.""" | |
| return [types.PageBlockUnsupported() if i in bad else types.PageBlockParagraph(T(f"block {i}")) for i in range(n)] | |
| class Gate: | |
| """A Telegram that refuses a rich message of more than `limit` blocks (and takes anything smaller).""" | |
| def __init__(self, limit=2, error=None): | |
| self.limit, self.error, self.sent, self.deleted, self.ids = limit, error, [], [], iter(range(500, 900)) | |
| async def __call__(self, req): | |
| if isinstance(req, functions.messages.SendMessageRequest): | |
| self.sent.append(req) | |
| if self.error is not None: | |
| raise self.error | |
| if len(req.rich_message.blocks) > self.limit: | |
| raise block_invalid() | |
| return types.UpdateShortSentMessage(id=next(self.ids), pts=0, pts_count=0, date=None, out=True) | |
| if isinstance(req, functions.messages.DeleteMessagesRequest): | |
| self.deleted += list(req.id) | |
| return None | |
| raise NotImplementedError(type(req)) | |
| # ================================================================================================= what is converted | |
| def test_a_stored_map_is_turned_into_an_input_map_wherever_it_is(): | |
| cap = types.PageCaption(T("c"), types.TextEmpty()) | |
| detail = types.PageBlockDetails([types.PageBlockParagraph(T("x")), a_map()], T("More")) | |
| collage = types.PageBlockCollage([a_map()], cap) | |
| original = [bytes(b) for b in (detail, collage)] | |
| out = inputify([detail, collage, types.PageBlockParagraph(T("p"))]) | |
| assert [type(b).__name__ for b in out] == ["PageBlockDetails", "PageBlockCollage", "PageBlockParagraph"] | |
| inner = out[0].blocks[1] | |
| assert isinstance(inner, types.InputPageBlockMap) and isinstance(out[1].items[0], types.InputPageBlockMap) | |
| assert (inner.geo.lat, inner.geo.long, inner.geo.accuracy_radius) == (41.9, 12.5, 7) | |
| assert (inner.zoom, inner.w, inner.h, bytes(inner.caption)) == (14, 600, 300, bytes(a_map().caption)) | |
| assert isinstance(inner.geo, types.InputGeoPoint) | |
| assert [bytes(b) for b in (detail, collage)] == original # what was handed to us is left as it was | |
| assert isinstance(detail.blocks[1], types.PageBlockMap) | |
| bytes(types.InputRichMessage(blocks=out)) # and it can be sent | |
| def test_blocks_without_a_map_come_back_the_same(): | |
| bl = blocks() + [types.PageBlockDetails(blocks("x"), T("t")), types.PageBlockDivider()] | |
| assert same(inputify(bl), bl) | |
| def test_a_received_message_with_a_map_is_sent_with_an_input_map_and_keeps_its_fingerprint(): | |
| rm = rich_msg([types.PageBlockHeading1(T("Trip")), a_map(), types.PageBlockParagraph(T("end"))]) | |
| r = rich_from_received(rm) | |
| sent = to_input(r) | |
| assert [type(b).__name__ for b in sent.blocks] == ["PageBlockHeading1", "InputPageBlockMap", "PageBlockParagraph"] | |
| assert [type(b).__name__ for b in input_of(rm).blocks] == [type(b).__name__ for b in sent.blocks] | |
| assert r.fp == fingerprint(rm) | |
| assert fingerprint(rich_msg([types.PageBlockHeading1(T("Trip")), a_map(access_hash=12345), types.PageBlockParagraph(T("end"))])) == r.fp | |
| def test_blocks_saved_in_telegrams_own_form_by_an_older_version_are_converted_when_sent(): | |
| old = types.InputRichMessage(blocks=[a_map(), types.PageBlockParagraph(T("end"))]) | |
| sent = to_input(Rich("blocks", _b64(old), False, False, None)) | |
| assert [type(b).__name__ for b in sent.blocks] == ["InputPageBlockMap", "PageBlockParagraph"] | |
| assert isinstance(old.blocks[0], types.PageBlockMap) # the saved copy itself is not touched | |
| def test_a_thinking_block_goes_out_as_an_expandable_section(): | |
| thought = types.PageBlockThinking(types.TextBold(T("let me see"))) | |
| nested = types.PageBlockDetails([thought], T("Outer")) | |
| out = inputify([thought, nested]) | |
| for sec in (out[0], out[1].blocks[0]): | |
| assert isinstance(sec, types.PageBlockDetails) and bytes(sec.title) == bytes(T("Thinking")) | |
| (para,) = sec.blocks | |
| assert isinstance(para, types.PageBlockParagraph) and bytes(para.text) == bytes(types.TextBold(T("let me see"))) | |
| assert isinstance(nested.blocks[0], types.PageBlockThinking) # what was handed to us is left as it was | |
| # what the sync compares is the same whichever of the two forms Telegram shows | |
| assert fingerprint(rich_msg([thought])) == fingerprint(rich_msg([out[0]])) | |
| def test_a_mention_of_somebody_the_message_has_no_access_data_for_stays_the_plain_name(): | |
| mention = types.PageBlockParagraph(types.TextConcat([T("hi "), types.TextMentionName(T("Mia"), 5)])) | |
| rm = rich_msg([mention]) | |
| plain = to_input(rich_from_received(rm, [])) | |
| assert plain.blocks[0].text.texts[1].__class__ is types.TextPlain and plain.blocks[0].text.texts[1].text == "Mia" | |
| known = to_input(rich_from_received(rm, [types.InputUser(5, 77)])) | |
| assert isinstance(known.blocks[0].text.texts[1], types.TextMentionName) and known.blocks[0].text.texts[1].user_id == 5 | |
| assert [u.user_id for u in known.users] == [5] | |
| # ==================================================================================== how the owner is told which blocks | |
| def test_blocks_have_names_and_are_listed_from_one(): | |
| r = rich_from_received(rich_msg([types.PageBlockHeading2(T("a")), a_map(), types.PageBlockTable(T(""), [])])) | |
| assert [block_name(b) for b in to_input(r).blocks] == ["heading", "map", "table"] | |
| assert block_name(types.PageBlockUnsupported()) == "a block this version does not know" | |
| assert block_name(types.InputPageBlockMap(types.InputGeoPointEmpty(), 1, 1, 1, types.PageCaption(T(""), T(""))))== "map" | |
| assert block_labels(r, [2, 1, 1, 9]) == "2 (map), 3 (table)" | |
| assert block_count(r) == 3 and block_count(Rich("markdown", "# x")) == 0 | |
| def test_block_numbers_fit_in_a_button(): | |
| for idx in ([], [0], [1, 3], [0, 59], list(range(60))): | |
| word = pack_blocks(idx) | |
| assert unpack_blocks(word) == sorted(idx) and len(f"rx:123456789:abcdef:{word}") <= 64 | |
| assert pack_blocks([]) == "0" | |
| def test_leaving_blocks_out_keeps_the_rest_and_forgets_the_fingerprint(): | |
| rm = rich_msg(with_bad(4, bad=(2,)), photos=[], documents=[], rtl=True) | |
| r = rich_from_received(rm) | |
| kept = without_blocks(r, [2]) | |
| inp = to_input(kept) | |
| assert [type(b).__name__ for b in inp.blocks] == ["PageBlockParagraph"] * 3 and inp.rtl is True | |
| assert kept.fp is None and kept.fmt == "blocks" and kept.rtl is True | |
| assert digest(kept) != digest(r) and block_count(r) == 4 and block_count(kept) == 3 | |
| def test_explain_rpc_knows_the_code_that_telegram_gave_for_a_rich_message(): | |
| text = explain_rpc(block_invalid()) | |
| assert text.startswith("Telegram refused (BadRequestError).") and "would not take one of the blocks" in text | |
| assert "(RICH_MESSAGE_BLOCK_INVALID)" in text | |
| assert "admin rights" in explain_rpc(rpc("ChatAdminRequiredError")) # the others are as they were | |
| other = errors.BadRequestError(None, "SOMETHING_ELSE", 400) | |
| assert "SOMETHING_ELSE" in explain_rpc(other) and not is_block_invalid(other) and is_block_invalid(block_invalid()) | |
| # ================================================================================================ finding the blocks | |
| def refusing_the_unsupported(gate): | |
| """`gate`, but a message with a block of the unsupported kind is refused (like a block Telegram can't take back).""" | |
| async def call(req): | |
| if isinstance(req, functions.messages.SendMessageRequest) and any( | |
| isinstance(b, types.PageBlockUnsupported) for b in req.rich_message.blocks | |
| ): | |
| gate.sent.append(req) | |
| raise block_invalid() | |
| return await gate(req) | |
| return call | |
| def test_the_refused_blocks_are_found_and_the_test_messages_are_deleted(): | |
| async def run(): | |
| gate = Gate(limit=10) | |
| r = rich_from_received(rich_msg(with_bad(8, bad=(2, 5)))) | |
| assert await refused_blocks(refusing_the_unsupported(gate), PEER, r, pause=0) == [2, 5] | |
| assert all(q.silent for q in gate.sent) and len(gate.sent) <= 14 # (the whole message is known to be refused: not asked again) | |
| refused = [q for q in gate.sent if any(isinstance(b, types.PageBlockUnsupported) for b in q.rich_message.blocks)] | |
| taken = len(gate.sent) - len(refused) | |
| assert taken > 0 and gate.deleted == list(range(500, 500 + taken)) # every test message that was made is deleted again | |
| asyncio.run(run()) | |
| def test_nothing_is_found_when_every_part_is_taken_or_it_cannot_be_looked_into(): | |
| async def run(): | |
| r = rich_from_received(rich_msg(with_bad(4, bad=()))) | |
| assert await refused_blocks(Gate(limit=2), PEER, r, pause=0) is None # too long together, fine in parts | |
| assert await refused_blocks(Gate(limit=2), PEER, Rich("markdown", "# x"), pause=0) is None # Telegram reads that itself | |
| assert await refused_blocks(Gate(limit=9), PEER, rich_from_received(rich_msg(with_bad(61, bad=()))), pause=0) is None | |
| broken = Gate(limit=0, error=rpc("ChatWriteForbiddenError")) # not a refused block: something else is wrong | |
| assert await refused_blocks(broken, PEER, rich_from_received(rich_msg(with_bad(3, bad=(1,)))), pause=0) is None | |
| asyncio.run(run()) | |
| def test_a_test_message_that_cannot_be_deleted_does_not_stop_the_search(): | |
| async def run(): | |
| gate = Gate(limit=10) | |
| inner = refusing_the_unsupported(gate) | |
| async def call(req): | |
| if isinstance(req, functions.messages.DeleteMessagesRequest): | |
| raise rpc("MessageIdInvalidError") | |
| return await inner(req) | |
| r = rich_from_received(rich_msg(with_bad(4, bad=(0, 2)))) | |
| assert await refused_blocks(call, PEER, r, pause=0) == [0, 2] | |
| asyncio.run(run()) | |
| # ================================================================================ a forwarded rich message in the bot | |
| MAPPED = [types.PageBlockHeading1(T("Trip")), a_map(), types.PageBlockParagraph(T("end"))] | |
| def test_a_forwarded_message_with_a_map_is_previewed_and_published(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| rm = rich_msg(MAPPED) | |
| pid, prompt = await new_post(app, "", message=native(rm), publish=False) | |
| assert "Rich message kept as it is" in prompt | |
| await app.press(f"pv:{pid}") | |
| kind, peer, inp, kw = rich_sends(app)[-1] | |
| assert isinstance(peer, types.InputPeerUser) | |
| assert [type(b).__name__ for b in inp.blocks] == ["PageBlockHeading1", "InputPageBlockMap", "PageBlockParagraph"] | |
| assert "refused" not in norm(app.out.last_text or "") | |
| await app.press(f"pb:{pid}") | |
| await app.press(f"pby:{pid}") | |
| post = await app.db.get_post(pid) | |
| assert post.status == "sent" | |
| live = app.tg.msgs[post.message_id].rich_message | |
| assert [type(b).__name__ for b in live.blocks][1] == "InputPageBlockMap" | |
| assert rich_of(post).fp == fingerprint(rm) # the sync still knows it | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_forwarded_message_with_a_thinking_block_is_previewed_and_published(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| rm = rich_msg([types.PageBlockThinking(T("pondering")), a_map(), types.PageBlockParagraph(T("answer"))]) | |
| pid, _ = await new_post(app, "", message=native(rm), publish=False) | |
| await app.press(f"pv:{pid}") | |
| assert "refused" not in norm(app.out.last_text or "") | |
| assert [type(b).__name__ for b in rich_sends(app)[-1][2].blocks] == [ | |
| "PageBlockDetails", "InputPageBlockMap", "PageBlockParagraph" | |
| ] | |
| await app.press(f"pb:{pid}") | |
| await app.press(f"pby:{pid}") | |
| post = await app.db.get_post(pid) | |
| live = app.tg.msgs[post.message_id].rich_message | |
| assert post.status == "sent" and bytes(live.blocks[0].title) == bytes(T("Thinking")) | |
| assert rich_of(post).fp == fingerprint(live) # the sync sees what is in the channel as the saved post | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def forwarded_with_bad_blocks(app): | |
| return new_post(app, "", message=native(rich_msg(with_bad(5, bad=(1, 3)))), publish=False) | |
| def test_the_preview_says_which_blocks_telegram_refuses_and_a_draft_can_go_on_without_them(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await forwarded_with_bad_blocks(app) | |
| await app.press(f"pv:{pid}") | |
| text = norm(app.out.last_text) | |
| assert "Telegram refused (BadRequestError)" in text and "would not take one of the blocks" in text | |
| assert "It won't take these blocks (of 5): 2 (a block this version does not know), 4 (a block this version does not know). The others are fine." in text | |
| assert "see /rich for the syntax" not in text # that is for typed messages | |
| assert app.out.last_buttons() == [["✂️ Leave those out"], ["↩️ Back"]] | |
| probes = [e for e in rich_sends(app) if isinstance(e[1], types.InputPeerUser)] | |
| assert all(e[3]["silent"] for e in probes[1:]) and sorted(app.tg.dm_rich_ids) == sorted( | |
| i for ids in app.tg.delete_requests for i in ids | |
| ) # nothing of the check stays in the chat | |
| await app.press(app.out.callback_data("Leave those out")) | |
| assert "Left out: 2 (a block this version does not know), 4 (a block this version does not know)." in norm(app.out.last_text) | |
| post = await app.db.get_post(pid) | |
| kept = to_input(rich_of(post)) | |
| assert [b.text.text for b in kept.blocks] == ["block 0", "block 2", "block 4"] | |
| assert post.text == plain_of(rich_of(post)) == "block 0\nblock 2\nblock 4" and post.status == "draft" | |
| assert len(post.entities) == 1 and post.entities[0]["k"] == "rich" and "p" not in post.entities[0] | |
| before = len(rich_sends(app)) | |
| await app.press(f"pv:{pid}") # now Telegram takes it | |
| assert len(rich_sends(app)) == before + 1 and len(rich_sends(app)[-1][2].blocks) == 3 | |
| await app.press(f"pb:{pid}") | |
| await app.press(f"pby:{pid}") | |
| post = await app.db.get_post(pid) | |
| assert post.status == "sent" and len(app.tg.msgs[post.message_id].rich_message.blocks) == 3 | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_publishing_a_message_telegram_refuses_names_the_blocks_and_publishes_nothing(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await forwarded_with_bad_blocks(app) | |
| await app.press(f"pb:{pid}") | |
| await app.press(f"pby:{pid}") | |
| text = norm(app.out.last_text) | |
| assert "It won't take these blocks (of 5): 2 (" in text and "The post was NOT published." in text | |
| assert "Leave those out" in str(app.out.last_buttons()) | |
| assert (await app.db.get_post(pid)).status == "draft" and app.tg.msgs == {} | |
| await app.press(app.out.callback_data("Leave those out")) | |
| await app.press(f"pb:{pid}") | |
| await app.press(f"pby:{pid}") | |
| assert (await app.db.get_post(pid)).status == "sent" and len(app.tg.msgs) == 1 | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_the_leave_out_button_is_only_for_the_draft_that_was_checked(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await forwarded_with_bad_blocks(app) | |
| await app.press(f"pv:{pid}") | |
| data = app.out.callback_data("Leave those out") | |
| # the post is changed meanwhile: the old button must not cut blocks out of the new message | |
| await app.press(f"et:{pid}") | |
| await app.text("", message=native(rich_msg(with_bad(6, bad=(0,))))) | |
| changed = (await app.db.get_post(pid)).entities | |
| await app.press(data) | |
| assert "not the one that was checked any more" in norm(app.out.last_text) | |
| assert (await app.db.get_post(pid)).entities == changed | |
| # a post that is published is not edited this way | |
| await app.press(f"pv:{pid}") | |
| data = app.out.callback_data("Leave those out") | |
| await app.db.update_post(pid, status="sent") | |
| await app.press(data) | |
| assert "not the one that was checked any more" in norm(app.out.last_text) | |
| assert (await app.db.get_post(pid)).entities == changed | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_when_every_block_is_refused_nothing_is_offered_to_leave_out(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await new_post(app, "", message=native(rich_msg(with_bad(3, bad=(0, 1, 2)))), publish=False) | |
| await app.press(f"pv:{pid}") | |
| text = norm(app.out.last_text) | |
| assert "It won't take any block of that message as it is" in text and "/rich" in text | |
| assert "Leave those out" not in text and "(of 3)" not in text | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_refusal_that_is_not_about_a_block_is_not_looked_into(tmp_path): | |
| async def run(): | |
| app = await make_app(tmp_path) | |
| pid, _ = await forwarded_with_bad_blocks(app) | |
| app.tg.fail["SendMessageRequest"] = rpc("ChatWriteForbiddenError") | |
| await app.press(f"pv:{pid}") | |
| text = norm(app.out.last_text) | |
| assert "ChatWriteForbiddenError" in text and "Telegram could not make that rich message." in text | |
| assert "see /rich for the syntax" not in text and "blocks (of" not in text | |
| assert len(rich_sends(app)) == 1 # no test messages were sent | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_typed_rich_message_that_is_refused_still_points_to_the_syntax(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"] = block_invalid() | |
| await app.press(f"pv:{pid}") | |
| text = norm(app.out.last_text) | |
| assert "would not take one of the blocks" in text and "see /rich for the syntax" in text | |
| assert len(rich_sends(app)) == 1 # (Telegram reads the typed text itself: there are no blocks to try) | |
| await app.db.close() | |
| asyncio.run(run()) | |
| def test_a_broadcast_of_a_message_telegram_refuses_says_why(tmp_path): | |
| async def run(): | |
| app = await broadcast_app(tmp_path) | |
| await app.text("/broadcast 1h") | |
| await app.text("", message=native(rich_msg(with_bad(3, bad=(1,))))) | |
| await app.press(app.out.callback_data("No buttons")) | |
| text = norm(app.out.last_text) | |
| assert "would not take one of the blocks" in text and "can't be sent like this" in text | |
| assert "It won't take these blocks (of 3): 2 (a block this version does not know)." in text | |
| assert all(not chan.msgs for chan in app.tg.registry.values()) | |
| await app.db.close() | |
| asyncio.run(run()) | |