Spaces:
Paused
Paused
| """The live side of the sync (what Telegram's updates say), the timed check, service-message deletion and /sync.""" | |
| import asyncio | |
| import datetime | |
| from telethon import errors, types | |
| from app.db import utcnow | |
| from app.syncer import Syncer | |
| from .fakes import FakeChannelClient, markup, photo_media, url_btn | |
| from .harness import App, norm | |
| from .test_sync import OLD, foreign, old_msg, saved | |
| NOW = datetime.datetime.now(datetime.timezone.utc) | |
| def run(coro): | |
| return asyncio.run(coro) | |
| # ------------------------------------------------------------------------------------------------ set-up | |
| async def make_app(tmp_path, *, posts=(), **cfg): | |
| """Channel 1 "Anime Channel" (the app's own fake) with `posts` saved the way the bot saved them a day ago.""" | |
| app = App(tmp_path, edit_delay=0, **cfg) | |
| await app.start() | |
| app.tg.username, app.tg.title = "animech", "Anime Channel" | |
| await app.db.save_channel(1, 5, "Anime Channel", "animech", 1) | |
| for m in posts: | |
| app.tg.msgs[m.id] = m | |
| app.tg._top = app.tg._pts = max(app.tg._top, m.id) | |
| await saved(app.db, m) | |
| sy = app.ctx.syncer | |
| sy.debounce, sy.max_debounce, sy.lock_wait, sy.rt_grace, sy.pause, sy.start_delay = 0.01, 0.2, 0.02, 0, 0, 0.01 | |
| return app | |
| def three(): | |
| return [ | |
| old_msg(1, "first"), | |
| old_msg(2, "second", markup=markup([url_btn("Watch", "https://t.me/a/1")])), | |
| old_msg(3, "third"), | |
| ] | |
| def new_update(m): | |
| return types.UpdateNewChannelMessage(message=m, pts=m.id, pts_count=1) | |
| def edit_update(m): | |
| return types.UpdateEditChannelMessage(message=m, pts=m.id, pts_count=1) | |
| def delete_update(cid, ids): | |
| return types.UpdateDeleteChannelMessages(channel_id=cid, messages=list(ids), pts=0, pts_count=len(ids)) | |
| def notice(i, action=None, cid=1): | |
| return types.MessageService( | |
| id=i, peer_id=types.PeerChannel(cid), date=NOW, action=action or types.MessageActionPinMessage() | |
| ) | |
| def dms(app): | |
| """Texts the bot sent to people (not to channels).""" | |
| return [e[2] for e in app.tg.sent if not isinstance(e[1], types.InputPeerChannel)] | |
| # ============================================================================================ service messages | |
| def test_every_kind_of_notice_is_deleted_the_moment_it_appears(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| actions = [ | |
| types.MessageActionChatEditTitle("A new name"), | |
| types.MessageActionChatEditPhoto(photo=types.PhotoEmpty(0)), | |
| types.MessageActionChatDeletePhoto(), | |
| types.MessageActionPinMessage(), | |
| types.MessageActionGroupCall(call=types.InputGroupCall(1, 2)), | |
| types.MessageActionHistoryClear(), | |
| ] | |
| for n, action in enumerate(actions, start=10): | |
| app.tg.msgs[n] = notice(n, action) | |
| await app.raw(new_update(app.tg.msgs[n])) | |
| assert [i for i in app.tg.msgs if i >= 10] == [] | |
| assert sorted(app.tg.msgs) == [1, 2, 3] # posts are never touched | |
| assert len(await app.db.sent_posts(1)) == 3 | |
| await app.db.close() | |
| run(go()) | |
| def test_notices_stay_when_the_switch_is_off(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three(), delete_service_messages=False) | |
| app.tg.msgs[10] = notice(10) | |
| await app.raw(new_update(app.tg.msgs[10])) | |
| assert 10 in app.tg.msgs | |
| await app.db.close() | |
| run(go()) | |
| def test_notices_in_channels_that_are_not_connected_are_left_alone(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| other = app.tg.add_channel(FakeChannelClient({5: notice(5, cid=2)}, username="other", cid=2, title="Other")) | |
| before = len(app.tg.requests) + len(other.requests) | |
| await app.raw(new_update(other.msgs[5])) | |
| assert 5 in other.msgs and len(app.tg.requests) + len(other.requests) == before # not even a request | |
| # a channel that was removed from the bot counts as not connected too | |
| await app.db.set_channel_active(1, False) | |
| app.ctx.syncer.forget_cache() | |
| app.tg.msgs[10] = notice(10) | |
| await app.raw(new_update(app.tg.msgs[10])) | |
| assert 10 in app.tg.msgs | |
| await app.db.close() | |
| run(go()) | |
| def test_the_owner_is_told_once_when_the_bot_may_not_delete(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| app.tg.fail["DeleteMessagesRequest"] = errors.ChatAdminRequiredError(request=None) | |
| for i in (10, 11, 12): | |
| app.tg.msgs[i] = notice(i) | |
| await app.raw(new_update(app.tg.msgs[i])) | |
| texts = [norm(t) for t in dms(app)] | |
| assert len(texts) == 1 and "Anime Channel" in texts[0] and "Delete messages" in texts[0] | |
| assert "DELETE_SERVICE_MESSAGES" in texts[0] | |
| assert {10, 11, 12} <= set(app.tg.msgs) | |
| # a day later the same problem is worth saying again | |
| app.ctx.syncer.tell_every = 0 | |
| await app.raw(new_update(app.tg.msgs[10])) | |
| assert len(dms(app)) == 2 | |
| await app.db.close() | |
| run(go()) | |
| def test_other_refusals_are_only_logged(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| app.tg.fail["DeleteMessagesRequest"] = errors.MessageDeleteForbiddenError(request=None) | |
| app.tg.msgs[10] = notice(10) | |
| await app.raw(new_update(app.tg.msgs[10])) | |
| assert dms(app) == [] and 10 in app.tg.msgs | |
| await app.db.close() | |
| run(go()) | |
| def test_notices_are_deleted_even_when_the_automatic_sync_is_switched_off(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three(), sync_interval_minutes=0) | |
| app.tg.msgs[10] = notice(10) | |
| await app.raw(new_update(app.tg.msgs[10])) | |
| assert 10 not in app.tg.msgs | |
| await app.db.close() | |
| run(go()) | |
| def test_notices_are_deleted_while_a_long_job_runs(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| async with app.ctx.lock: | |
| app.tg.msgs[10] = notice(10) | |
| await app.raw(new_update(app.tg.msgs[10])) | |
| assert 10 not in app.tg.msgs | |
| await app.db.close() | |
| run(go()) | |
| # ============================================================================================== live changes | |
| def test_a_new_post_by_someone_else_is_added_a_moment_later(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| m = foreign(app.tg, "from the Telegram app", markup=markup([url_btn("Open", "https://example.org")])) | |
| await app.raw(new_update(m)) | |
| assert await app.db.find_by_message(1, m.id) is None # not yet: the wait lets the bot finish its own business | |
| await sy.flush_events() | |
| p = await app.db.find_by_message(1, m.id) | |
| assert p.source == "adopted" and p.text == "from the Telegram app" and p.buttons == [[{"t": "Open", "u": "https://example.org"}]] | |
| await app.db.close() | |
| run(go()) | |
| def test_the_bots_own_posts_are_not_looked_at(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| mine = app.tg._new("sent by the bot", None, None) # out=True | |
| await app.raw(new_update(mine)) | |
| assert not sy._pending() | |
| await app.db.close() | |
| run(go()) | |
| def test_nothing_is_queued_while_a_long_job_runs_or_when_the_sync_is_off(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| m = foreign(app.tg, "x") | |
| async with app.ctx.lock: | |
| await app.raw(new_update(m)) | |
| await app.raw(edit_update(app.tg.msgs[1])) | |
| assert not sy._pending() | |
| off = await make_app(tmp_path / "off", posts=three(), sync_interval_minutes=0) | |
| m2 = foreign(off.tg, "y") | |
| await off.raw(new_update(m2)) | |
| await off.raw(edit_update(off.tg.msgs[1])) | |
| await off.raw(delete_update(1, [2])) | |
| assert not off.ctx.syncer._pending() and len(await off.db.sent_posts(1)) == 3 | |
| await app.db.close() | |
| await off.db.close() | |
| (tmp_path / "off").mkdir() | |
| run(go()) | |
| def test_an_edit_made_by_an_admin_is_recorded(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| m = app.tg.msgs[3] | |
| m.message, m.entities = "third, with a correction", [types.MessageEntityBold(0, 5)] | |
| await app.raw(edit_update(m)) | |
| await sy.flush_events() | |
| p = await app.db.find_by_message(1, 3) | |
| assert p.text == "third, with a correction" and p.entities == [{"k": "bold", "o": 0, "l": 5}] | |
| assert sy.last_reports[1].edited == [3] | |
| await app.db.close() | |
| run(go()) | |
| def test_the_echo_of_an_edit_made_through_the_bot_changes_nothing(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| sy.rt_grace = 60 # the bot saved the post a moment ago | |
| pid = (await app.db.find_by_message(1, 1)).id | |
| await app.db.update_post(pid, text="the owner's new text") # updated_at = now | |
| app.tg.msgs[1].message = "the owner's new text" | |
| await app.raw(edit_update(app.tg.msgs[1])) | |
| sy.max_tries = 1 | |
| await sy.flush_events() | |
| assert (await app.db.find_by_message(1, 1)).text == "the owner's new text" | |
| assert sy.last_reports[1].edited == [] | |
| await app.db.close() | |
| run(go()) | |
| def test_a_post_deleted_by_an_admin_leaves_my_posts_at_once(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| m = foreign(app.tg, "will be deleted before the wait is over") | |
| await app.raw(new_update(m)) | |
| await app.raw(edit_update(app.tg.msgs[2])) | |
| del app.tg.msgs[2] | |
| await app.raw(delete_update(1, [2, m.id])) | |
| assert await app.db.find_by_message(1, 2) is None # no flush needed | |
| assert sy._news == {1: set()} and sy._edits == {1: set()} # and nothing is left to look at for them | |
| await sy.flush_events() | |
| assert await app.db.find_by_message(1, m.id) is None | |
| assert sorted(p.message_id for p in await app.db.sent_posts(1)) == [1, 3] | |
| await app.db.close() | |
| run(go()) | |
| def test_deleted_posts_of_other_channels_are_ignored(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| await app.raw(delete_update(77, [1, 2, 3])) | |
| assert len(await app.db.sent_posts(1)) == 3 | |
| await app.db.close() | |
| run(go()) | |
| def test_a_message_that_is_still_too_fresh_is_looked_at_again(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| sy.rt_grace, sy.max_tries = 0.15, 500 | |
| m = foreign(app.tg, "just posted", age=0) | |
| await app.raw(new_update(m)) | |
| await asyncio.wait_for(sy.flush_events(), 5) | |
| p = await app.db.find_by_message(1, m.id) | |
| assert p is not None and p.source == "adopted" | |
| assert not sy._pending() and sy._tries == {} | |
| await app.db.close() | |
| run(go()) | |
| def test_a_message_that_never_gets_old_enough_is_given_up(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| sy.rt_grace, sy.max_tries = 3600, 2 | |
| m = foreign(app.tg, "from the future", age=0) | |
| await app.raw(new_update(m)) | |
| await asyncio.wait_for(sy.flush_events(), 5) | |
| assert await app.db.find_by_message(1, m.id) is None and not sy._pending() and sy._tries == {} | |
| await app.db.close() | |
| run(go()) | |
| def test_the_queue_waits_for_a_long_job_to_finish(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| m = foreign(app.tg, "posted before the long job started") | |
| await app.raw(new_update(m)) | |
| async with app.ctx.lock: | |
| task = asyncio.create_task(sy.flush_events()) | |
| await asyncio.sleep(0.15) | |
| assert not task.done() and await app.db.find_by_message(1, m.id) is None | |
| await asyncio.wait_for(task, 5) | |
| assert (await app.db.find_by_message(1, m.id)).source == "adopted" | |
| await app.db.close() | |
| run(go()) | |
| def test_a_burst_of_changes_becomes_one_look_at_the_whole_channel(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| sy.max_ids = 2 | |
| for i in (1, 2, 3): | |
| app.tg.msgs[i].message = f"edited {i}" | |
| await app.raw(edit_update(app.tg.msgs[i])) | |
| await sy.flush_events() | |
| rep = sy.last_reports[1] | |
| assert rep.checked == 3 and sorted(rep.edited) == [1, 2, 3] # compared as a whole, not id by id | |
| await app.db.close() | |
| run(go()) | |
| # ===================================================================================================== the timer | |
| def test_the_timed_check_looks_at_every_channel_and_gives_way_to_long_jobs(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| app.tg.msgs[2].message = "edited meanwhile" | |
| async with app.ctx.lock: | |
| assert await sy.run_all() == [] # a long job is running: nothing is read | |
| assert (await app.db.find_by_message(1, 2)).text == "second" | |
| reports = await sy.run_all() | |
| assert [r.channel.id for r in reports] == [1] and reports[0].edited == [2] | |
| assert sy.last_reports[1] is reports[0] | |
| await app.db.close() | |
| run(go()) | |
| def test_one_channel_that_fails_does_not_stop_the_others(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| broken = app.tg.add_channel(FakeChannelClient({1: old_msg(1, "x")}, username="broken", cid=2, title="Broken")) | |
| await app.db.save_channel(2, 6, "Broken", "broken", 1) | |
| await app.db.create_post(channel_id=2, message_id=1, status="sent", source="bot", created_by=1, text="x", | |
| created_at=OLD, updated_at=OLD, sent_at=OLD) | |
| broken.fail["GetMessagesRequest"] = RuntimeError("the network dropped") | |
| app.tg.msgs[3].message = "edited" | |
| reports = await app.ctx.syncer.run_all() | |
| by_id = {r.channel.id: r for r in reports} | |
| assert by_id[1].edited == [3] and by_id[1].error is None | |
| assert by_id[2].error and "RuntimeError" in by_id[2].error | |
| await app.db.close() | |
| run(go()) | |
| def test_the_timer_starts_and_stops_with_the_bot(tmp_path): | |
| async def go(): | |
| off = await make_app(tmp_path / "off", posts=three(), sync_interval_minutes=0) | |
| off.ctx.syncer.start() | |
| assert off.ctx.syncer._tasks == [] and off.ctx.syncer.enabled is False | |
| app = await make_app(tmp_path, posts=three()) | |
| sy = app.ctx.syncer | |
| app.tg.msgs[1].message = "edited" | |
| sy.start() | |
| assert len(sy._tasks) == 2 | |
| for _ in range(100): | |
| if 1 in sy.last_reports: | |
| break | |
| await asyncio.sleep(0.05) | |
| assert sy.last_reports[1].edited == [1] | |
| await sy.stop() | |
| assert sy._tasks == [] | |
| await app.db.close() | |
| await off.db.close() | |
| (tmp_path / "off").mkdir() | |
| run(go()) | |
| def test_the_syncer_is_part_of_the_context_the_bot_builds(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| assert isinstance(app.ctx.syncer, Syncer) | |
| await app.db.close() | |
| run(go()) | |
| # ============================================================================================ the /sync command | |
| def test_sync_is_for_the_owner_only(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| app.uid = 555 | |
| await app.text("/sync") | |
| assert "private" in norm(app.out.last_text).lower() or "owner" in norm(app.out.last_text).lower() | |
| assert app.ctx.syncer.last_reports == {} | |
| await app.db.close() | |
| run(go()) | |
| def test_sync_needs_a_channel_and_understands_only_its_options(tmp_path): | |
| async def go(): | |
| bare = App(tmp_path, edit_delay=0) | |
| await bare.start() | |
| await bare.text("/sync") | |
| assert "No channels registered" in norm(bare.out.last_text) | |
| await bare.db.close() | |
| app = await make_app(tmp_path / "b", posts=three()) | |
| await app.text("/sync --bogus") | |
| assert "I don't know “--bogus”" in norm(app.out.last_text) and "Usage: /sync" in norm(app.out.last_text) | |
| await app.text("/sync --channel @nowhere") | |
| assert "isn't registered" in norm(app.out.last_text) | |
| await app.db.close() | |
| (tmp_path / "b").mkdir() | |
| run(go()) | |
| def test_sync_reports_what_it_found_in_every_channel(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| await app.text("/sync") # the first look | |
| first = norm(app.out.last_text) | |
| assert "Sync finished" in first and "Anime Channel" in first and "First look" in first | |
| assert "3 saved post(s) compared" in first | |
| app.tg.msgs[2].message = "second, edited" | |
| del app.tg.msgs[3] | |
| app.tg._pts += 2 | |
| new = foreign(app.tg, "from the app") | |
| app.tg.msgs[new.id + 1] = notice(new.id + 1) | |
| app.tg._top = new.id + 1 | |
| app.tg._pts += 1 | |
| app.tg.msgs[2].reply_markup = None # and the channel shows no keyboard for post 2 | |
| await app.text("/sync") | |
| text = norm(app.out.last_text) | |
| assert "1 deleted in the channel" in text and "message 3" in text | |
| assert "1 edited in the channel" in text and "message 2" in text | |
| assert "1 new post(s) made outside the bot" in text and f"message {new.id}" in text | |
| assert "1 service message(s) deleted" in text | |
| assert "show no buttons in the channel (message 2)" in text and "Check buttons" in text | |
| assert not app.ctx.lock.locked() | |
| await app.text("/sync") | |
| assert "Everything matches." in norm(app.out.last_text) or "show no buttons" in norm(app.out.last_text) | |
| await app.db.close() | |
| run(go()) | |
| def test_sync_looks_at_one_channel_only_when_asked(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| app.tg.add_channel(FakeChannelClient({}, username="movies", cid=2, title="Movies")) | |
| await app.db.save_channel(2, 6, "Movies", "movies", 1) | |
| await app.text("/sync --channel @movies") | |
| text = norm(app.out.last_text) | |
| assert "Movies" in text and "Anime Channel" not in text | |
| assert set(app.ctx.syncer.last_reports) == {2} | |
| await app.db.close() | |
| run(go()) | |
| def test_sync_holds_the_long_job_lock_and_refuses_a_second_run(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| gate = asyncio.Event() | |
| real_get = app.tg.get_messages | |
| seen = [] | |
| async def slow(peer, ids=None): | |
| seen.append(app.ctx.lock.locked()) | |
| await gate.wait() | |
| return await real_get(peer, ids=ids) | |
| app.tg.get_messages = slow | |
| first = asyncio.create_task(app.text("/sync")) | |
| await asyncio.sleep(0.1) | |
| await app.text("/sync") | |
| assert "still running" in norm(app.out.last_text) | |
| gate.set() | |
| await asyncio.wait_for(first, 5) | |
| assert seen and all(seen) and not app.ctx.lock.locked() | |
| await app.db.close() | |
| run(go()) | |
| def test_sync_reports_a_channel_it_cannot_read(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| app.tg.fail["GetFullChannelRequest"] = errors.ChannelPrivateError(request=None) | |
| await app.text("/sync") | |
| text = norm(app.out.last_text) | |
| assert "Anime Channel" in text and "ChannelPrivateError" in text | |
| assert len(await app.db.sent_posts(1)) == 3 | |
| await app.db.close() | |
| run(go()) | |
| # ================================================================================== the rest of the bot | |
| def test_forgetting_a_post_in_the_bot_keeps_the_sync_from_bringing_it_back(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| p = await app.db.find_by_message(1, 2) | |
| await app.press(f"dly:{p.id}:b") | |
| assert await app.db.find_by_message(1, 2) is None and (1, 2) in app.ctx.sync_ignore | |
| assert 2 in app.tg.msgs # still in the channel | |
| await app.text("/sync") # the first look | |
| del app.tg.msgs[2] | |
| assert (1, 2) in app.ctx.sync_ignore | |
| await app.db.close() | |
| run(go()) | |
| def test_a_post_the_sync_saved_first_does_not_stop_the_bot_from_publishing(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| draft = await app.db.create_post(channel_id=1, status="draft", text="my new post", created_by=1) | |
| sent = app.tg._new("my new post", None, None) # what Telegram will answer the publish with | |
| await app.db.adopt_message(1, sent.id, {"text": "picked up by the sync a moment earlier"}) | |
| async def send_message(peer, message="", **kw): | |
| return sent | |
| app.tg.send_message = send_message | |
| await app.press(f"pby:{draft.id}") | |
| p = await app.db.find_by_message(1, sent.id) | |
| assert p.id == draft.id and p.source == "bot" and p.text == "my new post" | |
| await app.db.close() | |
| run(go()) | |
| def test_the_panel_says_where_an_adopted_post_came_from(tmp_path): | |
| from app.ui import panel_text | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| p, _ = await app.db.adopt_message(1, 40, {"text": "made in the app"}) | |
| ch = await app.db.get_channel(1) | |
| assert "not made with this bot" in panel_text(p, ch) | |
| await app.db.close() | |
| run(go()) | |
| def test_photos_of_new_posts_survive_the_trip_to_my_posts(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, posts=three()) | |
| m = foreign(app.tg, "a captioned photo", media=photo_media(41)) | |
| await app.raw(new_update(m)) | |
| await app.ctx.syncer.flush_events() | |
| p = await app.db.find_by_message(1, m.id) | |
| assert p.media_kind == "photo" and p.media_file_id.startswith("tl:") | |
| await app.db.close() | |
| run(go()) | |
| def test_utcnow_is_what_the_fresh_checks_compare_with(): | |
| assert (utcnow() - NOW).total_seconds() < 3600 # sanity for the helpers above | |