Spaces:
Paused
Paused
| """Replies, pins, the copy between two channels and the sync with the channels, through a REAL Telethon client object. | |
| Telethon resolves and serialises every request exactly as it does before sending it to Telegram; only the network at | |
| the very end is replaced by the simulated channels. A request that would not build or serialise in production fails here. | |
| """ | |
| import asyncio | |
| import pytest | |
| from telethon import TelegramClient, functions, types, utils | |
| from telethon.sessions import MemorySession | |
| from app.db import utcnow | |
| from app.repost_engine import run_repost | |
| from app.servicemsg import remove | |
| from app.shift_engine import delete_shift_copies, run_shift | |
| from app.sync_engine import SyncOptions, sync_channel, sync_ids | |
| from app.tgutil import peer_of | |
| from .dbutil import db_url_for, fresh_db | |
| from .fakes import FakeChannelClient, markup, rpc, service_msg, url_btn | |
| from .test_repost_links import CH, build_linked, new_posts, reply_of | |
| from .test_shift import SRC, old_posts, source_posts | |
| from .test_sync import foreign, old_msg, saved | |
| class Wire: | |
| """A real client whose requests are answered by `root` (a simulated channel; the other channels sit next to it).""" | |
| def __init__(self): | |
| self.client = TelegramClient(MemorySession(), 1, "h") | |
| self.root = None | |
| self.seen = [] | |
| def wire(monkeypatch): | |
| w = Wire() | |
| async def fake_call(self, request, ordered=False, flood_sleep_threshold=None): | |
| assert not isinstance(request, (list, tuple)) | |
| await request.resolve(self, utils) # what Telethon does just before sending | |
| w.seen.append(request) | |
| return await w.root(request) # serialises again and answers like Telegram | |
| monkeypatch.setattr(TelegramClient, "__call__", fake_call) | |
| return w | |
| def two_wired(w, *, restrict=False): | |
| """channel 1 "srcch" with the source posts, channel 2 "dstch" with ten old posts.""" | |
| src = FakeChannelClient(source_posts(), username="srcch", title="Anime Source") | |
| src.restrict_forwards = restrict | |
| dst = src.add_channel(FakeChannelClient(old_posts(), username="dstch", cid=2, title="Anime Backup")) | |
| w.root = src | |
| return src, dst | |
| def names(seen): | |
| return [type(r).__name__ for r in seen] | |
| def test_replies_and_pins_of_a_repost_go_through_a_real_client(wire, tmp_path): | |
| client, seen = wire.client, wire.seen | |
| chan = wire.root = FakeChannelClient(build_linked(), username=CH) | |
| async def go(): | |
| db = await fresh_db(db_url_for(tmp_path)) | |
| await db.save_channel(1, 5, "Test", CH, 99) | |
| ch = await db.get_channel(1) | |
| await db.create_migration("m1", 1, old=None, new=None, include_typed=False, include_posts=False, first_id=1, last_id=10, partial=False, user_id=99) | |
| mig = await db.get_migration("m1") | |
| res = await run_repost(client, db, ch, mig, 99, delay=0, settle_pause=0) | |
| assert (res.copied_units, res.failed, res.aborted) == (8, [], None) | |
| new = new_posts(chan) | |
| assert (reply_of(new[14]), reply_of(new[15]), reply_of(new[16]), reply_of(new[17])) == (13, 14, 15, 15) | |
| assert (res.final.pin_wanted, res.final.pinned, res.final.pin_error) == (2, 2, None) | |
| assert chan.pinned_order == [15, 19] and new[15].pinned and new[19].pinned | |
| assert not [m for m in chan.msgs.values() if isinstance(m, types.MessageService) and m.id > 10] # pin notices deleted | |
| await db.close() | |
| asyncio.run(go()) | |
| got = names(seen) | |
| for name in ("SendMessageRequest", "SendMultiMediaRequest", "UpdatePinnedMessageRequest", "DeleteMessagesRequest", "EditMessageRequest"): | |
| assert name in got, (name, got) | |
| # an answer is made with the new-style reply header, never with the old reply_to_msg_id field | |
| replies = [r.reply_to for r in seen if isinstance(r, (functions.messages.SendMessageRequest, functions.messages.SendMultiMediaRequest)) and r.reply_to] | |
| assert replies and all(isinstance(r, types.InputReplyToMessage) for r in replies) | |
| pins = [r for r in seen if isinstance(r, functions.messages.UpdatePinnedMessageRequest)] | |
| assert [(p.id, p.silent, bool(p.unpin)) for p in pins] == [(15, True, False), (19, True, False)] | |
| def test_a_shift_between_two_channels_through_a_real_client(wire, tmp_path): | |
| client, seen = wire.client, wire.seen | |
| src, dst = two_wired(wire) | |
| async def go(): | |
| db = await fresh_db(db_url_for(tmp_path)) | |
| await db.save_channel(2, 6, "Anime Backup", "dstch", 1) | |
| dst_row = await db.get_channel(2) | |
| await db.create_shift("s1", src=SRC, dst_channel_id=2, first_id=1, last_id=8, user_id=1) | |
| shift = await db.get_shift("s1") | |
| res = await run_shift(client, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| assert (res.copied_units, res.copied_messages, res.failed, res.aborted) == (6, 7, [], None) | |
| new = {i: m for i, m in dst.msgs.items() if i > 10} | |
| assert sorted(new) == [11, 12, 13, 14, 15, 16, 17] | |
| assert reply_of(new[14]) == 12 and not new[14].pinned | |
| assert new[17].message == "See https://t.me/dstch/11 and https://t.me/dstch/12 and https://t.me/elsewhere/9" | |
| assert sorted(src.msgs) == [1, 2, 3, 4, 5, 6, 7, 8] and src.msgs[5].pinned # the source is only read | |
| assert await db.shift_pairs("s1") == [(1, 11), (2, 12), (3, 13), (5, 14), (6, 15), (7, 16), (8, 17)] | |
| undo = await delete_shift_copies(client, db, dst_row, shift, delay=0) | |
| assert (undo.deleted, undo.remaining, undo.error) == (7, 0, None) | |
| assert sorted(dst.msgs) == list(range(1, 11)) | |
| assert sorted(src.msgs) == [1, 2, 3, 4, 5, 6, 7, 8] | |
| await db.close() | |
| asyncio.run(go()) | |
| # posts are carried over by Telegram itself when it is allowed to (here: from channel 1 to channel 2) | |
| forwards = [r for r in seen if isinstance(r, functions.messages.ForwardMessagesRequest)] | |
| assert forwards and all(isinstance(r.from_peer, types.InputPeerChannel) and r.from_peer.channel_id == 1 for r in forwards) | |
| assert all(isinstance(r.to_peer, types.InputPeerChannel) and r.to_peer.channel_id == 2 for r in forwards) | |
| assert all(r.drop_author for r in forwards) | |
| sent = [r for r in seen if isinstance(r, functions.messages.SendMessageRequest) and r.reply_to] | |
| assert sent and sent[0].peer.channel_id == 2 and sent[0].reply_to.reply_to_msg_id == 12 | |
| # the source channel was never written to | |
| for r in seen: | |
| if isinstance(r, (functions.messages.SendMessageRequest, functions.messages.SendMediaRequest, functions.messages.SendMultiMediaRequest, functions.messages.EditMessageRequest)): | |
| assert r.peer.channel_id == 2 | |
| if isinstance(r, functions.channels.DeleteMessagesRequest): | |
| assert r.channel.channel_id == 2 | |
| def test_a_shift_from_a_channel_that_forbids_forwarding_through_a_real_client(wire, tmp_path): | |
| client, seen = wire.client, wire.seen | |
| src, dst = two_wired(wire, restrict=True) | |
| async def go(): | |
| db = await fresh_db(db_url_for(tmp_path)) | |
| await db.save_channel(2, 6, "Anime Backup", "dstch", 1) | |
| dst_row = await db.get_channel(2) | |
| await db.create_shift("s1", src=SRC, dst_channel_id=2, first_id=1, last_id=8, user_id=1) | |
| shift = await db.get_shift("s1") | |
| res = await run_shift(client, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| assert (res.copied_units, res.failed, res.forwarded, res.rebuilt) == (6, [], 0, 6) | |
| new = {i: m for i, m in dst.msgs.items() if i > 10} | |
| assert new[12].reply_markup is not None and new[15].grouped_id == new[16].grouped_id | |
| await db.close() | |
| asyncio.run(go()) | |
| assert "SendMultiMediaRequest" in names(seen) and "SendMediaRequest" in names(seen) | |
| # ===================================================================================== the sync with the channels | |
| def notice(chan, action): | |
| """A service message as Telegram adds one: it takes the next message id and moves the event counter.""" | |
| chan._top += 1 | |
| chan._pts += 1 | |
| note = types.MessageService(id=chan._top, peer_id=types.PeerChannel(chan.cid), date=utcnow(), action=action) | |
| chan.msgs[note.id] = note | |
| return note | |
| def synced_channel(wire): | |
| """Posts 1, 2, 3 and 5 made through the bot a day ago; 4 is an old "pinned a message" notice.""" | |
| msgs = { | |
| 1: old_msg(1, "first"), | |
| 2: old_msg(2, "second", markup=markup([url_btn("Watch", "https://t.me/a/1")])), | |
| 3: old_msg(3, "third"), | |
| 4: service_msg(4), | |
| 5: old_msg(5, "fifth"), | |
| } | |
| chan = wire.root = FakeChannelClient(msgs, username="chan") | |
| return msgs, chan | |
| async def saved_world(tmp_path, msgs): | |
| db = await fresh_db(db_url_for(tmp_path)) | |
| await db.save_channel(1, 5, "Chan", "chan", 99) | |
| ch = await db.get_channel(1) | |
| for m in msgs.values(): | |
| if not isinstance(m, types.MessageService): | |
| await saved(db, m) | |
| return db, ch | |
| def test_the_sync_compares_a_channel_and_deletes_new_notices_through_a_real_client(wire, tmp_path): | |
| client, seen = wire.client, wire.seen | |
| msgs, chan = synced_channel(wire) | |
| async def go(): | |
| db, ch = await saved_world(tmp_path, msgs) | |
| opts = SyncOptions(pause=0, grace=0) | |
| first = await sync_channel(client, db, ch, opts) # the first look only notes where the channel stands | |
| assert (first.error, first.first_look, first.checked, first.changed) == (None, True, 4, False) | |
| assert sorted(chan.msgs) == [1, 2, 3, 4, 5] | |
| # what admins do outside the bot: delete a post, edit one, write a new one - Telegram adds notices meanwhile | |
| del chan.msgs[3] | |
| chan._pts += 1 | |
| chan.msgs[1].message = "first, edited" | |
| chan._pts += 1 | |
| foreign(chan, "written by another admin") # id 6 | |
| notice(chan, types.MessageActionChatEditTitle("A new name")) # id 7 | |
| notice(chan, types.MessageActionPinMessage()) # id 8 | |
| rep = await sync_channel(client, db, ch, opts) | |
| assert rep.error is None and rep.anomaly is None | |
| assert (rep.deleted, rep.edited, rep.adopted, rep.services_deleted, rep.services_failed) == ([3], [1], [6], 2, 0) | |
| assert sorted(chan.msgs) == [1, 2, 4, 5, 6] # the two new notices are gone, the old one is left alone | |
| assert await db.find_by_message(1, 3) is None | |
| assert (await db.find_by_message(1, 1)).text == "first, edited" | |
| took = await db.find_by_message(1, 6) | |
| assert (took.source, took.text, took.created_by, took.status) == ("adopted", "written by another admin", 0, "sent") | |
| again = await sync_channel(client, db, ch, opts) # nothing left to do | |
| assert not again.changed and again.error is None | |
| await db.close() | |
| asyncio.run(go()) | |
| got = names(seen) | |
| for name in ("GetFullChannelRequest", "GetMessagesRequest", "DeleteMessagesRequest"): | |
| assert name in got, (name, got) | |
| # the channel counter and every read name the channel the way Telegram wants it | |
| fulls = [r for r in seen if isinstance(r, functions.channels.GetFullChannelRequest)] | |
| assert fulls and all(isinstance(r.channel, types.InputChannel) and r.channel.channel_id == 1 for r in fulls) | |
| reads = [r for r in seen if isinstance(r, functions.channels.GetMessagesRequest)] | |
| assert reads and all(isinstance(r.channel, types.InputChannel) for r in reads) | |
| # only the two new notices were deleted, in one request, and nothing was written to the channel | |
| deletes = [r for r in seen if isinstance(r, functions.channels.DeleteMessagesRequest)] | |
| assert [i for r in deletes for i in r.id] == [7, 8] | |
| assert all(isinstance(r.channel, types.InputChannel) and r.channel.channel_id == 1 for r in deletes) | |
| writes = ( | |
| functions.messages.SendMessageRequest, functions.messages.SendMediaRequest, functions.messages.SendMultiMediaRequest, | |
| functions.messages.EditMessageRequest, functions.messages.ForwardMessagesRequest, functions.messages.UpdatePinnedMessageRequest, | |
| ) | |
| assert not [r for r in seen if isinstance(r, writes)] | |
| def test_the_live_look_at_a_few_messages_through_a_real_client(wire, tmp_path): | |
| client, seen = wire.client, wire.seen | |
| msgs, chan = synced_channel(wire) | |
| async def go(): | |
| db, ch = await saved_world(tmp_path, msgs) | |
| opts = SyncOptions(pause=0, grace=0) | |
| await sync_channel(client, db, ch, opts) | |
| chan.msgs[2].message = "second, edited" | |
| chan._pts += 1 | |
| foreign(chan, "a new post") # 6 | |
| note = notice(chan, types.MessageActionChatEditTitle("Another name")) # 7 | |
| del chan.msgs[5] | |
| chan._pts += 1 | |
| rep = await sync_ids(client, db, ch, [2, 5, 6, note.id], opts, adopt_ids=[6]) | |
| assert rep.error is None | |
| assert (rep.edited, rep.deleted, rep.adopted, rep.services_deleted) == ([2], [5], [6], 1) | |
| assert sorted(chan.msgs) == [1, 2, 3, 4, 6] | |
| assert (await db.find_by_message(1, 2)).text == "second, edited" | |
| assert (await db.find_by_message(1, 2)).buttons == [[{"t": "Watch", "u": "https://t.me/a/1"}]] # the saved buttons stay | |
| assert await db.find_by_message(1, 5) is None | |
| await db.close() | |
| asyncio.run(go()) | |
| assert [i for r in seen if isinstance(r, functions.channels.DeleteMessagesRequest) for i in r.id] == [7] | |
| def test_deleting_a_notice_and_the_refusals_through_a_real_client(wire, tmp_path): | |
| client = wire.client | |
| msgs, chan = synced_channel(wire) | |
| async def go(): | |
| db, ch = await saved_world(tmp_path, msgs) | |
| peer = peer_of(ch) | |
| out = await remove(client, peer, [4]) | |
| assert (out.deleted, out.failed, out.error) == (1, 0, None) and 4 not in chan.msgs | |
| chan.fail["DeleteMessagesRequest"] = rpc("ChatAdminRequiredError") | |
| note = notice(chan, types.MessageActionPinMessage()) | |
| out = await remove(client, peer, [note.id]) | |
| assert (out.deleted, out.failed, out.error_name) == (0, 1, "ChatAdminRequiredError") and note.id in chan.msgs | |
| chan.fail.clear() | |
| chan.fail["GetFullChannelRequest"] = rpc("ChannelPrivateError") # the bot lost its place in the channel | |
| rep = await sync_channel(client, db, ch, SyncOptions(pause=0, grace=0)) | |
| assert rep.error and not rep.changed and rep.checked == 0 | |
| assert [p.message_id for p in await db.sent_posts(1)] == [1, 2, 3, 5] # nothing was touched | |
| await db.close() | |
| asyncio.run(go()) | |
| def test_a_restricted_channel_and_restricted_posts_through_a_real_client(wire, tmp_path): | |
| """The channel object and the messages come back as real Telethon types, with Telegram's restriction marks.""" | |
| client, seen = wire.client, wire.seen | |
| msgs, chan = synced_channel(wire) | |
| async def go(): | |
| db, ch = await saved_world(tmp_path, msgs) | |
| opts = SyncOptions(pause=0, grace=0) | |
| await sync_channel(client, db, ch, opts) # the first look | |
| notice_text = "This message couldn't be displayed on your device due to copyright infringement." | |
| chan.restriction = notice_text # a copyright strike: the channel itself is restricted | |
| for m in chan.msgs.values(): | |
| if not isinstance(m, types.MessageService): | |
| m.message = notice_text | |
| chan._pts += 4 | |
| del chan.msgs[3] | |
| rep = await sync_channel(client, db, ch, opts) | |
| assert rep.restricted_channel == notice_text and not rep.changed and rep.checked == 0 | |
| assert [p.text for p in await db.sent_posts(1)] == ["first", "second", "third", "fifth"] # My posts untouched | |
| chan.restriction = None # only the posts carry the mark now (as real messages with restriction_reason) | |
| for i in (1, 2): | |
| chan.msgs[i].restriction_reason = [types.RestrictionReason("all", "copyright", notice_text)] | |
| rep = await sync_channel(client, db, ch, opts) | |
| assert sorted(rep.restricted) == [1, 2, 5] and rep.restricted_why in (notice_text, "Telegram shows a notice instead of the post") | |
| assert rep.deleted == [3] and rep.edited == [] # post 3 really is gone, the others were left alone | |
| assert [p.text for p in await db.sent_posts(1)] == ["first", "second", "fifth"] | |
| await db.close() | |
| asyncio.run(go()) | |
| writes = ( | |
| functions.messages.SendMessageRequest, functions.messages.SendMediaRequest, functions.messages.EditMessageRequest, | |
| functions.messages.ForwardMessagesRequest, functions.channels.DeleteMessagesRequest, | |
| ) | |
| assert not [r for r in seen if isinstance(r, writes)] | |
| fulls = [r for r in seen if isinstance(r, functions.channels.GetFullChannelRequest)] | |
| assert len(fulls) == 3 and all(isinstance(r.channel, types.InputChannel) for r in fulls) | |