"""The sync engine: My posts follows what channel admins do outside the bot (deleted, edited and new posts, service messages).""" import asyncio import datetime from telethon import errors, types from app.db import as_utc, utcnow from app.servicemsg import remove from app.sync_engine import ( SyncOptions, diff_post, find_top, live_fields, message_supported, sync_channel, sync_ids, ) from app.tgutil import media_info from .dbutil import db_url_for, fresh_db from .fakes import FakeChannelClient, make_msg, markup, photo_media, service_msg, url_btn, video_media OLD = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=1) FAST = dict(pause=0) def old_msg(i, text="", **kw): m = make_msg(i, text, **kw) m.date = OLD return m def foreign(chan, text="", *, age=86400, **kw): """A post somebody else made in the channel (not out).""" m = chan._new(text, kw.pop("entities", None), kw.pop("markup", None), kw.pop("media", None)) m.out = False m.date = utcnow() - datetime.timedelta(seconds=age) return m async def saved(db, m, **extra): """The bot's saved copy of live message `m`, exactly as the bot would have saved it a day ago.""" fields = dict( channel_id=1, message_id=m.id, status="sent", source="bot", created_by=99, link_preview=False, created_at=OLD, updated_at=OLD, sent_at=OLD, **live_fields(m), ) fields.update(extra) return await db.create_post(**fields) async def world(tmp_path, msgs, *, save=True): chan = FakeChannelClient(msgs, username="chan") db = await fresh_db(db_url_for(tmp_path)) await db.save_channel(1, 5, "Chan", "chan", 99) ch = await db.get_channel(1) if save: for m in msgs.values(): await saved(db, m) return chan, db, ch def run(coro): return asyncio.run(coro) def opts(**kw): return SyncOptions(**{**FAST, "grace": 0, **kw}) def three(): return { 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"), } # ============================================================================================ comparing one post def test_a_post_that_is_as_saved_has_no_difference(): m = old_msg(1, "hello", markup=markup([url_btn("Go", "https://x.org")])) p = SaveLike(m) d = diff_post(p, m) assert d.fields == {} and not d.buttons_missing and not d.quiet class SaveLike: """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 test_text_formatting_and_buttons_that_changed_are_found(): m = old_msg(1, "hello world", entities=[types.MessageEntityBold(0, 5)], markup=markup([url_btn("Go", "https://x.org")])) p = SaveLike(m, text="hello", entities=[], buttons=[[{"t": "Go", "u": "https://old.org"}]]) d = diff_post(p, m) assert d.fields["text"] == "hello world" assert d.fields["entities"] == [{"k": "bold", "o": 0, "l": 5}] assert d.fields["buttons"] == [[{"t": "Go", "u": "https://x.org"}]] assert not d.quiet and not d.buttons_missing def test_a_keyboard_that_is_not_shown_never_wipes_the_saved_buttons(): m = old_msg(1, "hello") # the channel shows no keyboard p = SaveLike(m, buttons=[[{"t": "Go", "u": "https://x.org"}]]) d = diff_post(p, m) assert d.buttons_missing and "buttons" not in d.fields def test_only_spaces_around_the_text_is_a_quiet_correction(): m = old_msg(1, "hello") d = diff_post(SaveLike(m, text="hello \n"), m) assert d.fields == {"text": "hello"} and d.quiet def test_media_is_compared_by_its_id_not_by_its_file_reference(): m = old_msg(1, "pic", media=photo_media(11)) p = SaveLike(m) other_fetch = old_msg(1, "pic", media=photo_media(11)) other_fetch.media.photo.file_reference = b"a-fresh-reference" assert diff_post(p, other_fetch).fields == {} # same photo, new file reference: nothing changed replaced = old_msg(1, "pic", media=photo_media(12)) d = diff_post(p, replaced) assert d.fields["media_kind"] == "photo" and d.fields["media_file_id"] == media_info(replaced)[1] def test_media_that_seems_to_have_vanished_is_not_forgotten(): with_media = old_msg(1, "pic", media=photo_media(11)) p = SaveLike(with_media) stripped = old_msg(1, "pic") # Telegram showed it without the photo this time assert diff_post(p, stripped).fields == {} def test_what_the_bot_can_hold(): assert message_supported(old_msg(1, "text")) assert message_supported(old_msg(1, "", media=photo_media())) assert message_supported(old_msg(1, "", media=video_media())) assert not message_supported(old_msg(1, "")) # a poll, a sticker ... assert not message_supported(old_msg(1, " \n ")) # ================================================================================================= the first look def test_the_first_look_notes_where_the_channel_stands_and_reads_in_the_older_posts(tmp_path): async def go(): msgs = three() chan, db, ch = await world(tmp_path, msgs, save=False) # nothing saved: the channel has history only for i in (4, 5): chan.msgs[i] = old_msg(i, f"older post {i}") chan._top = chan._pts = 5 rep = await sync_channel(chan, db, ch, opts()) assert rep.first_look and rep.error is None assert rep.adopted == [] and rep.deleted == [] and rep.edited == [] # these are not "new posts" ... assert rep.imported == 5 and rep.history_done # ... they are the posts from before the bot was added posts, total = await db.list_posts(1, None, 0, 50) assert total == 5 and {p.source for p in posts} == {"adopted"} and {p.status for p in posts} == {"sent"} state = await db.get_sync(1) assert (state.last_top, state.last_pts, state.deferred) == (5, 5, 0) hist = await db.get_history(1) assert (hist.top, hist.next_id, hist.imported, hist.done) == (5, 6, 5, True) again = await sync_channel(chan, db, ch, opts()) # nothing is read twice assert again.imported == 0 and again.history_top == 0 and again.changed is False await db.close() run(go()) def test_a_channel_without_any_message_is_fine(tmp_path): async def go(): chan, db, ch = await world(tmp_path, {}, save=False) rep = await sync_channel(chan, db, ch, opts()) assert rep.first_look and rep.error is None and rep.adopted == [] again = await sync_channel(chan, db, ch, opts()) assert again.error is None and again.changed is False await db.close() run(go()) def test_the_first_look_finds_the_newest_message_below_a_long_run_of_deleted_ids(tmp_path): async def go(): chan, db, ch = await world(tmp_path, {1: old_msg(1, "a"), 2: old_msg(2, "b")}, save=False) chan._top, chan._pts = 2, 1000 # a thousand edits and deletes since top = await find_top(chan, types.InputPeerChannel(1, 5), 1000, 0, pause=0) assert top == 2 # the newest saved post is a floor: looking stops there assert await find_top(chan, types.InputPeerChannel(1, 5), 1000, 2, pause=0) == 2 await db.close() run(go()) # ============================================================================================= saved posts vs channel def test_a_post_deleted_in_the_channel_leaves_my_posts(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) await sync_channel(chan, db, ch, opts()) # the first look del chan.msgs[2] chan._pts += 1 rep = await sync_channel(chan, db, ch, opts()) assert rep.deleted == [2] and rep.checked == 3 assert sorted(p.message_id for p in await db.sent_posts(1)) == [1, 3] await db.close() run(go()) def test_a_post_edited_in_the_channel_is_updated_here(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) m2 = chan.msgs[2] m2.message = "second, edited by an admin" m2.entities = [types.MessageEntityItalic(0, 6)] m2.reply_markup = markup([url_btn("Watch now", "https://t.me/new/9")]) rep = await sync_channel(chan, db, ch, opts()) assert rep.edited == [2] and rep.deleted == [] and rep.adopted == [] p = await db.find_by_message(1, 2) assert p.text == "second, edited by an admin" assert p.entities == [{"k": "italic", "o": 0, "l": 6}] assert p.buttons == [[{"t": "Watch now", "u": "https://t.me/new/9"}]] again = await sync_channel(chan, db, ch, opts()) assert again.edited == [] and again.changed is False # nothing left to record await db.close() run(go()) def test_a_missing_keyboard_is_reported_and_the_saved_buttons_stay(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) chan.msgs[2].reply_markup = None # the channel shows no keyboard any more rep = await sync_channel(chan, db, ch, opts()) assert rep.buttons_missing == [2] and rep.edited == [] assert (await db.find_by_message(1, 2)).buttons == [[{"t": "Watch", "u": "https://t.me/a/1"}]] await db.close() run(go()) def test_posts_the_bot_touched_a_moment_ago_are_left_for_later(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) await db.update_post((await db.find_by_message(1, 1)).id, text="just edited through the bot") # updated_at = now chan.msgs[1].message = "an admin's text" rep = await sync_channel(chan, db, ch, opts(grace=120)) assert rep.young == [1] and rep.edited == [] and rep.checked == 2 assert (await db.find_by_message(1, 1)).text == "just edited through the bot" await db.close() run(go()) def test_if_every_saved_post_looks_deleted_nothing_is_removed(tmp_path): async def go(): msgs = {i: old_msg(i, f"post {i}") for i in range(1, 8)} chan, db, ch = await world(tmp_path, msgs) chan.msgs.clear() # e.g. the bot lost access and Telegram shows nothing rep = await sync_channel(chan, db, ch, opts()) assert rep.deleted == [] and rep.anomaly and "All 7" in rep.anomaly assert len(await db.sent_posts(1)) == 7 await db.close() run(go()) def test_one_or_two_posts_can_be_deleted_even_when_that_is_all_there_is(tmp_path): async def go(): chan, db, ch = await world(tmp_path, {1: old_msg(1, "a"), 2: old_msg(2, "b")}) chan.msgs.clear() rep = await sync_channel(chan, db, ch, opts()) assert sorted(rep.deleted) == [1, 2] and rep.anomaly is None await db.close() run(go()) def test_a_change_made_through_the_bot_while_the_channel_was_read_wins(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) chan.msgs[3].message = "an admin's text" pid = (await db.find_by_message(1, 3)).id real_get = chan.get_messages async def get_while_the_owner_edits(peer, ids=None): res = await real_get(peer, ids=ids) await db.update_post(pid, text="the owner's newer text") # happens between reading and saving return res chan.get_messages = get_while_the_owner_edits rep = await sync_channel(chan, db, ch, opts()) assert rep.edited == [] and rep.changed_meanwhile == 1 assert (await db.find_by_message(1, 3)).text == "the owner's newer text" await db.close() run(go()) # ================================================================================================= new posts async def looked(tmp_path, msgs=None): chan, db, ch = await world(tmp_path, msgs or three()) first = await sync_channel(chan, db, ch, opts()) # the first look sets the mark assert first.first_look return chan, db, ch def test_a_new_post_made_outside_the_bot_is_added(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) m = foreign(chan, "posted from the Telegram app", entities=[types.MessageEntityBold(0, 6)], markup=markup([url_btn("Open", "https://example.org/x")])) rep = await sync_channel(chan, db, ch, opts()) assert rep.adopted == [m.id] and rep.first_look is False p = await db.find_by_message(1, m.id) assert (p.source, p.status, p.created_by) == ("adopted", "sent", 0) assert p.text == "posted from the Telegram app" assert p.entities == [{"k": "bold", "o": 0, "l": 6}] assert p.buttons == [[{"t": "Open", "u": "https://example.org/x"}]] assert p.media_kind is None and p.link_preview is False assert as_utc(p.sent_at) == as_utc(m.date) state = await db.get_sync(1) assert state.last_top == m.id again = await sync_channel(chan, db, ch, opts()) assert again.adopted == [] # not saved twice assert len(await db.sent_posts(1)) == 4 await db.close() run(go()) def test_new_media_posts_keep_their_media(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) m = foreign(chan, "caption", media=photo_media(31)) await sync_channel(chan, db, ch, opts()) p = await db.find_by_message(1, m.id) assert p.media_kind == "photo" and p.media_file_id.startswith("tl:") and p.text == "caption" await db.close() run(go()) def test_the_bots_own_posts_that_are_saved_are_not_picked_up_again(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) m = chan._new("made with the bot", None, None) # out=True, like the bot's own post m.date = utcnow() - datetime.timedelta(minutes=10) mine = await db.create_post(channel_id=1, message_id=m.id, status="sent", source="bot", created_by=99, sent_at=OLD, created_at=OLD, updated_at=OLD, **live_fields(m)) rep = await sync_channel(chan, db, ch, opts()) assert rep.adopted == [] and rep.deleted == [] assert (await db.find_by_message(1, m.id)).id == mine.id and mine.source == "bot" await db.close() run(go()) def test_a_message_that_appeared_a_moment_ago_waits_for_the_next_look(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) m = foreign(chan, "brand new", age=1) rep = await sync_channel(chan, db, ch, opts(grace=120)) assert rep.adopted == [] and rep.young_new == [m.id] state = await db.get_sync(1) assert state.last_top == 3 and state.deferred == 1 # the mark stays below it m.date = utcnow() - datetime.timedelta(minutes=10) # time passes rep = await sync_channel(chan, db, ch, opts(grace=120)) assert rep.adopted == [m.id] and rep.young_new == [] state = await db.get_sync(1) assert state.last_top == m.id and state.deferred == 0 await db.close() run(go()) def test_a_quiet_channel_costs_no_reading_at_all(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) before = len(chan.requests) reads = [] real_get = chan.get_messages async def counting(peer, ids=None): reads.append(list(ids)) return await real_get(peer, ids=ids) chan.get_messages = counting rep = await sync_channel(chan, db, ch, opts()) assert rep.adopted == [] and rep.scanned_to == 3 # only the saved posts were compared (ids 1-3) assert reads == [[1, 2, 3]] # nothing was read beyond them: not a single event since the last look assert len(chan.requests) - before == 1 # just the channel counter await db.close() run(go()) def test_messages_of_a_kind_the_bot_cannot_hold_are_left_alone(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) foreign(chan, "") # a poll or a sticker: no text, no supported media rep = await sync_channel(chan, db, ch, opts()) assert rep.adopted == [] and rep.unsupported == 1 await db.close() run(go()) def test_a_post_the_owner_made_the_bot_forget_is_not_brought_back(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) m = foreign(chan, "a post that cannot be deleted") rep = await sync_channel(chan, db, ch, opts(ignore={(1, m.id)})) assert rep.adopted == [] await db.close() run(go()) def test_a_dead_tail_longer_than_the_quick_look_needs_a_deep_look(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) chan._top, chan._pts = 3, 900 # lots of events happened: the counter is far above the newest message first = await sync_channel(chan, db, ch, opts()) assert (await db.get_sync(1)).last_top == 3 and first.first_look chan._top = 600 # 597 ids went to messages that were deleted, then a new post appears chan._pts = 901 m = foreign(chan, "after a very long run of deleted ids") assert m.id == 601 quick = await sync_channel(chan, db, ch, opts()) assert quick.adopted == [] # the quick look reads 300 ids beyond the likely end and stops deep = await sync_channel(chan, db, ch, opts(deep=True)) assert deep.adopted == [601] assert (await db.get_sync(1)).last_top == 601 await db.close() run(go()) def test_the_automatic_looks_go_deep_once_a_day(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) chan._top, chan._pts = 3, 900 await sync_channel(chan, db, ch, opts()) chan._top, chan._pts = 600, 901 m = foreign(chan, "hidden behind a long dead tail") assert (await sync_channel(chan, db, ch, opts())).adopted == [] # a day later the same automatic look reads every possible id old = utcnow() - datetime.timedelta(hours=25) state = await db.get_sync(1) await db.save_sync(1, last_top=state.last_top, last_pts=state.last_pts, deep=False) async with db.Session() as s: row = await s.get(type(state), 1) row.last_deep = old await s.commit() assert (await sync_channel(chan, db, ch, opts())).adopted == [m.id] await db.close() run(go()) # ======================================================================================== service messages def test_service_messages_that_appeared_are_deleted_by_the_look(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) chan.msgs[4] = service_msg(4) chan.msgs[5] = types.MessageService(id=5, peer_id=types.PeerChannel(1), date=OLD, action=types.MessageActionChatEditTitle("New")) chan._top, chan._pts = 5, 5 rep = await sync_channel(chan, db, ch, opts()) assert rep.services_deleted == 2 and rep.services_failed == 0 assert 4 not in chan.msgs and 5 not in chan.msgs and sorted(chan.msgs) == [1, 2, 3] assert rep.adopted == [] await db.close() run(go()) def test_service_messages_stay_when_the_switch_is_off(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) chan.msgs[4] = service_msg(4) chan._top, chan._pts = 4, 4 rep = await sync_channel(chan, db, ch, opts(delete_services=False)) assert rep.services_deleted == 0 and 4 in chan.msgs await db.close() run(go()) def test_a_service_message_the_bot_may_not_delete_is_counted(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) chan.msgs[4] = service_msg(4) chan._top, chan._pts = 4, 4 chan.undeletable.add(4) rep = await sync_channel(chan, db, ch, opts()) assert (rep.services_deleted, rep.services_failed) == (0, 1) and rep.services_error await db.close() run(go()) # ============================================================================================= errors and busy def test_a_channel_the_bot_cannot_read_is_reported_not_raised(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) chan.fail["GetFullChannelRequest"] = errors.ChannelPrivateError(request=None) rep = await sync_channel(chan, db, ch, opts()) assert rep.error and "ChannelPrivateError" in rep.error assert len(await db.sent_posts(1)) == 3 chan.fail.clear() chan.fail["GetMessagesRequest"] = errors.ChannelPrivateError(request=None) rep = await sync_channel(chan, db, ch, opts()) assert rep.error and len(await db.sent_posts(1)) == 3 await db.close() run(go()) def test_a_long_job_that_starts_cuts_the_look_short(tmp_path): async def go(): chan, db, ch = await looked(tmp_path) m = foreign(chan, "new while a repost runs") del chan.msgs[2] chan._pts += 1 rep = await sync_channel(chan, db, ch, opts(), busy=lambda: True) assert rep.busy and rep.deleted == [] and rep.adopted == [] assert len(await db.sent_posts(1)) == 3 rep = await sync_channel(chan, db, ch, opts()) # the job is over assert rep.deleted == [2] and rep.adopted == [m.id] await db.close() run(go()) # =================================================================================== looking at given ids only def test_sync_ids_handles_edits_deletes_new_posts_and_notices(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) chan.msgs[1].message = "edited" del chan.msgs[3] new = foreign(chan, "added") chan.msgs[9] = service_msg(9) chan._top = 9 ids = [1, 3, new.id, 9, 2] rep = await sync_ids(chan, db, ch, ids, opts(), adopt_ids={new.id, 9}) assert rep.edited == [1] and rep.deleted == [3] and rep.adopted == [new.id] assert rep.services_deleted == 1 and 9 not in chan.msgs assert (await db.find_by_message(1, 2)).text == "second" await db.close() run(go()) def test_an_edit_of_a_post_the_bot_never_saved_is_history_and_ignored(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) old = old_msg(7, "an old post, edited today") chan.msgs[7] = old chan._top = 7 rep = await sync_ids(chan, db, ch, [7], opts(), adopt_ids=set()) assert rep.adopted == [] and await db.find_by_message(1, 7) is None await db.close() run(go()) def test_sync_ids_reports_what_is_too_fresh(tmp_path): async def go(): chan, db, ch = await world(tmp_path, three()) new = foreign(chan, "right now", age=0) await db.update_post((await db.find_by_message(1, 1)).id, text="saved a second ago") chan.msgs[1].message = "different" rep = await sync_ids(chan, db, ch, [1, new.id], opts(grace=60), adopt_ids={new.id}) assert rep.young == [1] and rep.young_new == [new.id] and rep.adopted == [] and rep.edited == [] await db.close() run(go()) # ================================================================================================ the database def test_adopting_a_message_twice_saves_it_once(tmp_path): async def go(): db = await fresh_db(db_url_for(tmp_path)) await db.save_channel(1, 5, "Chan", "chan", 99) p, created = await db.adopt_message(1, 50, {"text": "x"}, sent_at=OLD) assert created and p.source == "adopted" again, created = await db.adopt_message(1, 50, {"text": "y"}) assert again is None and created is False assert (await db.find_by_message(1, 50)).text == "x" await db.close() run(go()) def test_a_reposted_copy_takes_the_place_of_a_post_the_sync_saved_first(tmp_path): async def go(): db = await fresh_db(db_url_for(tmp_path)) await db.save_channel(1, 5, "Chan", "chan", 99) await db.adopt_message(1, 50, {"text": "picked up by the sync"}) await db.create_migration("m1", 1, old=None, new=None, include_typed=False, include_posts=False, first_id=1, last_id=9, partial=False, user_id=99) await db.record_repost("m1", 1, [{"old_id": 1, "new_id": 50, "post": {"text": "the copy"}}], 99) p = await db.find_by_message(1, 50) assert p.text == "the copy" and p.source == "bot" await db.close() run(go()) def test_publishing_takes_the_place_of_a_post_the_sync_saved_first(tmp_path): async def go(): db = await fresh_db(db_url_for(tmp_path)) await db.save_channel(1, 5, "Chan", "chan", 99) await db.adopt_message(1, 60, {"text": "picked up first"}) draft = await db.create_post(channel_id=1, status="draft", text="my draft", created_by=99) assert await db.release_slot(1, 60, keep_pid=draft.id) == 1 sent = await db.update_post(draft.id, status="sent", message_id=60) assert sent.message_id == 60 and (await db.find_by_message(1, 60)).text == "my draft" await db.close() run(go()) def test_the_sync_update_only_applies_to_an_unchanged_post(tmp_path): async def go(): db = await fresh_db(db_url_for(tmp_path)) await db.save_channel(1, 5, "Chan", "chan", 99) post = await db.create_post(channel_id=1, message_id=5, status="sent", text="v1", created_by=99) seen = (await db.get_post(post.id)).updated_at assert await db.sync_update_post(post.id, seen, {"text": "v2"}) is True assert await db.sync_update_post(post.id, seen, {"text": "v3"}) is False # it changed since `seen` assert (await db.get_post(post.id)).text == "v2" await db.close() run(go()) def test_servicemsg_remove_reports_refusals_instead_of_raising(tmp_path): async def go(): chan = FakeChannelClient({1: service_msg(1), 2: service_msg(2)}) peer = types.InputPeerChannel(1, 5) ok = await remove(chan, peer, [1, 2]) assert (ok.deleted, ok.failed, ok.error) == (2, 0, None) chan.msgs[3] = service_msg(3) chan.fail["DeleteMessagesRequest"] = errors.ChatAdminRequiredError(request=None) bad = await remove(chan, peer, [3]) assert bad.failed == 1 and bad.error_name == "ChatAdminRequiredError" run(go())