Spaces:
Paused
Paused
| """/autopost (a series of episode posts, then the real links) and "Mass replace links" in a post's panel.""" | |
| import asyncio | |
| import pytest | |
| from telethon import errors | |
| from app.autopost import ( | |
| PLACEHOLDER_LINK, | |
| button_label, | |
| parse_autopost_args, | |
| plan_ranges, | |
| post_text, | |
| ranges_summary, | |
| ) | |
| from app.tgutil import button_url | |
| from .harness import App, norm | |
| def run(coro): | |
| return asyncio.run(coro) | |
| async def make_app(tmp_path, **cfg): | |
| app = App(tmp_path, edit_delay=0, **cfg) | |
| await app.start() | |
| await app.db.save_channel(1, 5, "Anime Channel", "animech", 1) | |
| return app | |
| def sent_markup(entry): | |
| return entry[3]["buttons"] | |
| def link_edits(app): | |
| """(message id, new link) for every keyboard-only edit the bot made.""" | |
| out = [] | |
| for kind, mid, req in app.tg.edits: | |
| if kind == "raw" and req.message is None and req.reply_markup is not None: | |
| out.append((mid, button_url(req.reply_markup.rows[0].buttons[0]))) | |
| return out | |
| # ------------------------------------------------------------------------------- the plan | |
| def test_the_ranges_follow_the_interval(): | |
| assert plan_ranges(114, 20) == [(1, 20), (21, 40), (41, 60), (61, 80), (81, 100), (101, 114)] | |
| assert plan_ranges(100, 20) == [(1, 20), (21, 40), (41, 60), (61, 80), (81, 100)] | |
| assert plan_ranges(20, 20) == [(1, 20)] | |
| def test_a_short_end_is_added_to_the_post_before_it(): | |
| assert plan_ranges(102, 20)[-2:] == [(61, 80), (81, 102)] # 2 left over: less than half of 20 | |
| assert plan_ranges(110, 20)[-1] == (81, 110) # exactly half of 20: still added | |
| assert plan_ranges(111, 20)[-2:] == [(81, 100), (101, 111)] # 11 is more than half: its own post | |
| assert plan_ranges(114, 20)[-1] == (101, 114) # 14 is more than half | |
| assert plan_ranges(22, 15) == [(1, 22)] # half of 15 is 7.5: 7 left over is added ... | |
| assert plan_ranges(23, 15) == [(1, 15), (16, 23)] # ... 8 is not | |
| def test_a_total_below_the_interval_is_one_post_and_small_intervals_work(): | |
| assert plan_ranges(14, 20) == [(1, 14)] | |
| assert plan_ranges(3, 20) == [(1, 3)] | |
| assert plan_ranges(3, 1) == [(1, 1), (2, 2), (3, 3)] | |
| assert plan_ranges(5, 2) == [(1, 2), (3, 5)] # one left over = half of 2: added to the post before it | |
| with pytest.raises(ValueError): | |
| plan_ranges(0, 20) | |
| def test_every_episode_is_in_exactly_one_post(): | |
| for total in range(1, 130): | |
| for interval in (1, 2, 3, 7, 10, 15, 20, 25): | |
| ranges = plan_ranges(total, interval) | |
| flat = [n for a, b in ranges for n in range(a, b + 1)] | |
| assert flat == list(range(1, total + 1)), (total, interval) | |
| assert all(b - a + 1 <= interval + interval // 2 for a, b in ranges) | |
| def test_texts_are_the_ones_asked_for(): | |
| assert post_text(1, 20) == "Episodes 01 to 20\n\nClick the button below to download 👇" | |
| assert post_text(101, 114) == "Episodes 101 to 114\n\nClick the button below to download 👇" | |
| assert button_label(21, 40) == "Download Episodes 21 to 40" | |
| assert button_label(5, 5) == "Download Episode 05" | |
| assert ranges_summary(plan_ranges(114, 20)) == "01–20 · 21–40 · 41–60 · 61–80 · 81–100 · 101–114" | |
| assert ranges_summary([(i, i) for i in range(1, 40)]).endswith("… · 38 · 39") | |
| def test_the_arguments(): | |
| assert parse_autopost_args("114 20") == (114, 20) | |
| assert parse_autopost_args(" 12, 5 ") == (12, 5) | |
| for bad in ("", "114", "114 20 3", "a b", "0 20", "20 0", "-5 3", "1.5 2", "999999999 20"): | |
| with pytest.raises(ValueError): | |
| parse_autopost_args(bad) | |
| # ------------------------------------------------------------------------------- the command | |
| def test_the_whole_autopost_flow(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/autopost 114 20") | |
| shown = norm(app.out.last_text) | |
| assert "6 post(s)" in shown and "01–20 · 21–40 · 41–60 · 61–80 · 81–100 · 101–114" in shown | |
| assert "Which channel" in shown and "1. Anime Channel" in shown and app.out.last_buttons()[0] == ["1"] | |
| await app.press(app.out.callback_data("Anime Channel")) | |
| posted = [e for e in app.tg.sent if e[0] == "text" and e[1].channel_id == 1] | |
| assert len(posted) == 6 | |
| assert [e[2] for e in posted] == [post_text(a, b) for a, b in plan_ranges(114, 20)] | |
| for (a, b), entry in zip(plan_ranges(114, 20), posted): | |
| row = sent_markup(entry).rows[0].buttons | |
| assert len(row) == 1 and row[0].text == f"Download Episodes {a:02d} to {b:02d}" | |
| assert button_url(row[0]) == PLACEHOLDER_LINK == "https://t.me/English" | |
| assert entry[3]["link_preview"] is False | |
| # all six are in My posts, as published posts of this channel | |
| posts = await app.db.sent_posts(1) | |
| assert [p.message_id for p in posts] == [100, 101, 102, 103, 104, 105] | |
| assert posts[0].source == "bot" and posts[0].buttons == [[{"t": "Download Episodes 01 to 20", "u": PLACEHOLDER_LINK}]] | |
| # the first link is asked for at once, with a cancel button | |
| ask = norm(app.out.last_text) | |
| assert "1 of 6" in ask and "Download Episodes 01 to 20" in ask and PLACEHOLDER_LINK in ask | |
| assert "Cancel" in str(app.out.last_buttons()) and "Skip" in str(app.out.last_buttons()) | |
| links = [f"https://t.me/mybot?start=part{i}" for i in range(1, 7)] | |
| for i, url in enumerate(links): | |
| await app.text(url) | |
| if i < 5: | |
| nxt = norm(app.out.last_text) | |
| assert f"{i + 2} of 6" in nxt and "now opens" in nxt and url in nxt | |
| a, b = plan_ranges(114, 20)[i + 1] | |
| assert f"Download Episodes {a:02d} to {b:02d}" in nxt | |
| assert "All done" in app.out.last_text and "6 link(s) replaced" in app.out.last_text | |
| # each answer changed only the keyboard of its own message, at once | |
| assert link_edits(app) == [(100 + i, links[i]) for i in range(6)] | |
| assert [button_url(app.tg.edits[i][2].reply_markup.rows[0].buttons[0]) for i in range(6)] == links | |
| posts = await app.db.sent_posts(1) | |
| assert [p.buttons[0][0]["u"] for p in posts] == links | |
| assert [p.buttons[0][0]["t"] for p in posts][-1] == "Download Episodes 101 to 114" | |
| assert 1 not in app.ctx.state # nothing left to answer | |
| await app.db.close() | |
| run(go()) | |
| def test_the_short_end_gets_its_own_post_or_joins_the_last(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/autopost 102 20") | |
| await app.press(app.out.callback_data("Anime Channel")) | |
| texts = [e[2].splitlines()[0] for e in app.tg.sent if e[0] == "text" and e[1].channel_id == 1] | |
| assert texts == ["Episodes 01 to 20", "Episodes 21 to 40", "Episodes 41 to 60", "Episodes 61 to 80", "Episodes 81 to 102"] | |
| await app.db.close() | |
| run(go()) | |
| def test_admins_may_use_it_and_strangers_may_not(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, admins=frozenset({7})) | |
| app.uid = 7 | |
| await app.text("/autopost 40 20") | |
| assert "Which channel" in norm(app.out.last_text) | |
| await app.press(app.out.callback_data("Anime Channel")) | |
| assert len([e for e in app.tg.sent if e[0] == "text" and e[1].channel_id == 1]) == 2 | |
| app.uid = 99 | |
| before = len(app.tg.sent) | |
| await app.text("/autopost 40 20") | |
| assert "private" in app.out.last_text and len(app.tg.sent) == before | |
| await app.db.close() | |
| run(go()) | |
| def test_wrong_numbers_and_no_channels_are_explained(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/autopost") | |
| assert "Usage" in norm(app.out.last_text) and "/autopost 114 20" in norm(app.out.last_text) | |
| for bad in ("114", "x y", "0 5", "114 20 5"): | |
| await app.text(f"/autopost {bad}") | |
| assert "Usage: /autopost <total episodes> <interval>" in norm(app.out.last_text), bad | |
| await app.text("/autopost 5000 1") | |
| assert "at most 200" in app.out.last_text | |
| assert not [e for e in app.tg.sent if e[0] == "text"] # nothing was posted by any of those | |
| (tmp_path / "empty").mkdir() | |
| empty = App(tmp_path / "empty", edit_delay=0) | |
| await empty.start() | |
| await empty.text("/autopost 114 20") | |
| assert "No channels yet" in empty.out.last_text | |
| await empty.db.close() | |
| await app.db.close() | |
| run(go()) | |
| def test_cancelling_the_channel_choice_posts_nothing(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/autopost 114 20") | |
| await app.press(app.out.callback_data("Cancel")) | |
| assert "Cancelled" in app.out.last_text | |
| assert not [e for e in app.tg.sent if e[0] == "text"] | |
| # an old button does nothing any more | |
| await app.press("apc:" + "0" * 8 + ":1") | |
| assert "expired" in app.out.last_text | |
| await app.db.close() | |
| run(go()) | |
| def test_a_double_press_posts_once(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/autopost 40 20") | |
| data = app.out.callback_data("Anime Channel") | |
| await app.press(data) | |
| await app.press(data) | |
| assert "expired" in app.out.last_text | |
| assert len([e for e in app.tg.sent if e[0] == "text" and e[1].channel_id == 1]) == 2 | |
| await app.db.close() | |
| run(go()) | |
| def test_the_bot_needs_the_post_right_and_a_free_hand(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| app.tg.rights = dict(admin=True, post=False, edit=True, delete=True, invite=False, add_admins=False) | |
| await app.text("/autopost 40 20") | |
| pick = app.out.callback_data("Anime Channel") | |
| await app.press(pick) | |
| assert "Post messages" in app.out.last_text | |
| assert not [e for e in app.tg.sent if e[0] == "text"] | |
| # the plan is still there: with the right given, the same button works - but not while a long job runs | |
| app.tg.rights["post"] = True | |
| await app.ctx.lock.acquire() | |
| await app.press(pick) | |
| assert "still running" in app.out.last_text | |
| app.ctx.lock.release() | |
| await app.press(pick) | |
| assert len([e for e in app.tg.sent if e[0] == "text" and e[1].channel_id == 1]) == 2 | |
| await app.db.close() | |
| run(go()) | |
| def test_when_telegram_stops_midway_the_posts_made_still_get_their_links(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| real = app.tg.send_message | |
| calls = [] | |
| async def flaky(peer, message="", **kw): | |
| calls.append(message) | |
| if len(calls) == 3: | |
| raise errors.ChatWriteForbiddenError(request=None) | |
| return await real(peer, message, **kw) | |
| app.tg.send_message = flaky | |
| await app.text("/autopost 114 20") | |
| await app.press(app.out.callback_data("Anime Channel")) | |
| # the summary (an edit of the picker) tells where it stopped; the next message asks for the first link | |
| summary = [norm(t) for k, t, _ in app.out.log if k == "edit" and t and "Stopped early" in t] | |
| assert summary and "Posted 2 of 6" in summary[0] and "Episodes 41 to 60" in summary[0] | |
| assert "1 of 2" in norm(app.out.last_text) | |
| assert len(await app.db.sent_posts(1)) == 2 | |
| await app.text("https://t.me/mybot?start=1") | |
| await app.text("https://t.me/mybot?start=2") | |
| assert "All done" in app.out.last_text and "2 link(s) replaced" in app.out.last_text | |
| await app.db.close() | |
| run(go()) | |
| def test_nothing_posted_means_a_plain_error(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| async def refuse(peer, message="", **kw): | |
| raise errors.ChatWriteForbiddenError(request=None) | |
| app.tg.send_message = refuse | |
| await app.text("/autopost 40 20") | |
| await app.press(app.out.callback_data("Anime Channel")) | |
| assert "Nothing was posted" in app.out.last_text and "Episodes 01 to 20" in app.out.last_text | |
| assert await app.db.sent_posts(1) == [] | |
| await app.db.close() | |
| run(go()) | |
| def test_skipping_and_stopping_while_the_links_are_asked(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/autopost 80 20") | |
| await app.press(app.out.callback_data("Anime Channel")) | |
| await app.text("https://t.me/mybot?start=1") | |
| assert "2 of 4" in norm(app.out.last_text) | |
| await app.press(app.out.callback_data("Skip")) # keeps the placeholder | |
| assert "Skipped" in app.out.last_text and "3 of 4" in norm(app.out.last_text) | |
| cancel = app.out.callback_data("Cancel") | |
| # the Skip button of the earlier message is stale now | |
| await app.press("lqs:1") | |
| assert "earlier step" in app.out.last_text | |
| await app.press(cancel) | |
| shown = norm(app.out.last_text) | |
| assert "Stopped" in shown and "1 link(s) replaced" in shown and "1 skipped" in shown and "2 not asked" in shown | |
| assert 1 not in app.ctx.state | |
| posts = await app.db.sent_posts(1) | |
| assert [p.buttons[0][0]["u"] for p in posts] == ["https://t.me/mybot?start=1", PLACEHOLDER_LINK, PLACEHOLDER_LINK, PLACEHOLDER_LINK] | |
| # nothing more is expected from the user | |
| await app.text("https://t.me/whatever") | |
| assert "Use /new" in app.out.last_text | |
| await app.db.close() | |
| run(go()) | |
| def test_something_that_is_not_a_link_is_asked_again_and_a_refusal_keeps_the_step(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/autopost 40 20") | |
| await app.press(app.out.callback_data("Anime Channel")) | |
| await app.text("hello there") | |
| assert "doesn't look like a link" in norm(app.out.last_text) | |
| # Telegram refuses the keyboard edit: the saved link is unchanged and the same post is asked again | |
| app.tg.fail["EditMessageRequest"] = errors.ChatAdminRequiredError(request=None) | |
| await app.text("t.me/mybot?start=1") | |
| assert "NOT changed" in app.out.last_text | |
| assert (await app.db.sent_posts(1))[0].buttons[0][0]["u"] == PLACEHOLDER_LINK | |
| app.tg.fail.clear() | |
| await app.text("t.me/mybot?start=1") | |
| assert "now opens" in app.out.last_text and "https://t.me/mybot?start=1" in app.out.last_text | |
| assert (await app.db.sent_posts(1))[0].buttons[0][0]["u"] == "https://t.me/mybot?start=1" | |
| await app.db.close() | |
| run(go()) | |
| # ------------------------------------------------------------------- mass replace links | |
| def sent(app, pid_text, message_id, buttons, channel_id=1, text=None): | |
| return app.db.create_post( | |
| channel_id=channel_id, message_id=message_id, status="sent", source="bot", text=text or pid_text, | |
| entities=[], buttons=buttons, link_preview=False, | |
| ) | |
| def one(label, url): | |
| return [[{"t": label, "u": url}]] | |
| async def channel_with_posts(app): | |
| """#1 text, #2 button, #3 text, #4 only another bot's button, #5 button, #6 two buttons, #7 button in ANOTHER channel.""" | |
| await app.db.save_channel(2, 6, "Other", "other", 1) | |
| posts = {} | |
| posts[1] = await sent(app, "just text", 10, []) | |
| posts[2] = await sent(app, "Episodes 01 to 20", 11, one("Download Episodes 01 to 20", "https://t.me/old?start=a")) | |
| posts[3] = await sent(app, "plain again", 12, []) | |
| posts[4] = await sent(app, "reactions only", 13, [[{"t": "👍 3", "raw": "AAAA"}]]) | |
| posts[5] = await sent(app, "Episodes 21 to 40", 14, one("Download Episodes 21 to 40", "https://t.me/old?start=b")) | |
| posts[6] = await sent( | |
| app, "Choose", 15, [[{"t": "480p", "u": "https://t.me/old?start=c"}, {"t": "720p", "u": "https://t.me/old?start=d"}]], | |
| ) | |
| posts[7] = await sent(app, "elsewhere", 5, one("Other channel", "https://t.me/old?start=z"), channel_id=2) | |
| return posts | |
| def test_only_published_posts_with_a_link_button_get_the_option(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| posts = await channel_with_posts(app) | |
| draft = await app.db.create_post(channel_id=1, status="draft", source="bot", text="d", entities=[], buttons=one("Get", "https://t.me/x")) | |
| for key, expected in ((1, False), (2, True), (4, False), (5, True), (6, True)): | |
| await app.press(f"po:{posts[key].id}") | |
| assert ("Mass replace links" in str(app.out.last_buttons())) is expected, key | |
| await app.press(f"po:{draft.id}") | |
| assert "Mass replace links" not in str(app.out.last_buttons()) | |
| await app.press(f"mlp:{draft.id}") | |
| assert "published posts only" in app.out.last_text | |
| await app.press(f"mlp:{posts[1].id}") | |
| assert "no button with a link" in app.out.last_text | |
| await app.db.close() | |
| run(go()) | |
| def test_mass_replace_walks_through_the_button_posts_of_the_channel(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| posts = await channel_with_posts(app) | |
| await app.press(f"po:{posts[2].id}") | |
| await app.press(app.out.callback_data("Mass replace links")) | |
| ask = norm(app.out.last_text) | |
| assert "Mass replace links" in ask and "1 of 4" in ask and "Download Episodes 01 to 20" in ask | |
| assert "https://t.me/old?start=a" in ask and f"Post #{posts[2].id}" in ask | |
| assert "Cancel" in str(app.out.last_buttons()) | |
| await app.text("https://t.me/new?start=1") | |
| ask = norm(app.out.last_text) # the text posts (#3) and the reactions-only post (#4) are skipped | |
| assert "2 of 4" in ask and "Download Episodes 21 to 40" in ask and f"Post #{posts[5].id}" in ask | |
| assert "now opens https://t.me/new?start=1" in ask | |
| await app.text("https://t.me/new?start=2") | |
| ask = norm(app.out.last_text) | |
| assert "3 of 4" in ask and "480p" in ask and "https://t.me/old?start=c" in ask | |
| await app.text("https://t.me/new?start=3") | |
| assert "4 of 4" in norm(app.out.last_text) and "720p" in norm(app.out.last_text) | |
| await app.text("https://t.me/new?start=4") | |
| assert "All done" in app.out.last_text and "4 link(s) replaced" in app.out.last_text | |
| # keyboard-only edits: the text of the posts was never sent again | |
| edits = [(mid, req) for kind, mid, req in app.tg.edits if kind == "raw"] | |
| assert [mid for mid, _ in edits] == [11, 14, 15, 15] | |
| assert all(req.message is None for _, req in edits) | |
| last = edits[-1][1].reply_markup.rows[0].buttons | |
| assert [button_url(b) for b in last] == ["https://t.me/new?start=3", "https://t.me/new?start=4"] | |
| assert [b.text for b in last] == ["480p", "720p"] | |
| fresh = {k: await app.db.get_post(p.id) for k, p in posts.items()} | |
| assert fresh[2].buttons == one("Download Episodes 01 to 20", "https://t.me/new?start=1") | |
| assert fresh[5].buttons[0][0]["u"] == "https://t.me/new?start=2" | |
| assert [b["u"] for b in fresh[6].buttons[0]] == ["https://t.me/new?start=3", "https://t.me/new?start=4"] | |
| assert fresh[4].buttons == [[{"t": "👍 3", "raw": "AAAA"}]] # another bot's button: untouched | |
| assert fresh[7].buttons[0][0]["u"] == "https://t.me/old?start=z" # another channel: untouched | |
| assert fresh[2].text == "Episodes 01 to 20" | |
| await app.db.close() | |
| run(go()) | |
| def test_it_starts_at_the_chosen_post_and_goes_towards_the_newest(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| posts = await channel_with_posts(app) | |
| await app.press(f"mlp:{posts[5].id}") # #2 is older: not asked | |
| assert "1 of 3" in norm(app.out.last_text) and "Download Episodes 21 to 40" in norm(app.out.last_text) | |
| await app.press(app.out.callback_data("Skip")) | |
| assert "2 of 3" in norm(app.out.last_text) and "480p" in norm(app.out.last_text) | |
| await app.press(app.out.callback_data("Cancel")) | |
| shown = norm(app.out.last_text) | |
| assert "Stopped" in shown and "1 skipped" in shown and "2 not asked" in shown | |
| assert (await app.db.get_post(posts[5].id)).buttons[0][0]["u"] == "https://t.me/old?start=b" | |
| assert not [1 for kind, mid, req in app.tg.edits if kind == "raw"] | |
| await app.db.close() | |
| run(go()) | |
| def test_a_post_that_disappears_meanwhile_is_passed_over(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| posts = await channel_with_posts(app) | |
| await app.press(f"mlp:{posts[2].id}") | |
| await app.db.delete_post(posts[5].id) # forgotten in the bot while the list is open | |
| await app.text("https://t.me/new?start=1") | |
| assert "3 of 4" in norm(app.out.last_text) and "480p" in norm(app.out.last_text) # #5 was passed over | |
| await app.text("https://t.me/new?start=2") | |
| await app.text("https://t.me/new?start=3") | |
| assert "All done" in app.out.last_text | |
| assert "3 link(s) replaced" in app.out.last_text and "1 button(s) no longer there" in app.out.last_text | |
| await app.db.close() | |
| run(go()) | |
| def test_telegram_refusing_a_link_keeps_the_same_step(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| posts = await channel_with_posts(app) | |
| await app.press(f"mlp:{posts[2].id}") | |
| skip = app.out.callback_data("Skip") | |
| app.tg.fail["EditMessageRequest"] = errors.MessageIdInvalidError(request=None) | |
| await app.text("https://t.me/new?start=1") | |
| assert "NOT changed" in app.out.last_text | |
| assert (await app.db.get_post(posts[2].id)).buttons[0][0]["u"] == "https://t.me/old?start=a" | |
| app.tg.fail.clear() | |
| await app.press(skip) | |
| assert "2 of 4" in norm(app.out.last_text) | |
| await app.db.close() | |
| run(go()) | |
| def test_the_last_button_post_ends_the_list(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| p = await sent(app, "only one", 3, one("Get", "https://t.me/old")) | |
| await app.press(f"mlp:{p.id}") | |
| assert "1 of 1" in norm(app.out.last_text) | |
| await app.text("https://t.me/new") | |
| assert "All done" in app.out.last_text and "1 link(s) replaced" in app.out.last_text | |
| await app.db.close() | |
| run(go()) | |