import asyncio from app.tgutil import button_url from .harness import App, norm SEEDED = [[{"t": "Download Episodes 01 to 20", "u": "https://t.me/oldbot?start=aaa"}]] async def make_app(tmp_path): app = App(tmp_path) await app.start() await app.db.save_channel(1, 5, "Anime Channel", "animech", 1) return app def sent_markup(app, i=-1): return app.tg.sent[i][3]["buttons"] def test_suggestion_flow_end_to_end(tmp_path): async def run(): app = await make_app(tmp_path) await app.db.create_post(channel_id=1, message_id=10, status="sent", source="bot", text="Click the button below to download 👇", entities=[], buttons=SEEDED) await app.text("/new") assert "Suggestion" in app.out.last_text and "Download Episodes 21 to 40" in app.out.last_text await app.press(app.out.callback_data("Use suggestion")) assert "Send the link" in app.out.last_text await app.text("https://t.me/newbot?start=zzz") assert "Ready" in app.out.last_text and "Publish" in str(app.out.last_buttons()) pid = int(app.out.callback_data("Publish").split(":")[1]) await app.press(f"pb:{pid}") assert "Publish this post" in app.out.last_text await app.press(f"pby:{pid}") assert "Published" in app.out.last_text kind, peer, msg, kw = app.tg.sent[-1] assert msg == "Click the button below to download 👇" btn = sent_markup(app).rows[0].buttons[0] assert btn.text == "Download Episodes 21 to 40" and button_url(btn) == "https://t.me/newbot?start=zzz" p = await app.db.get_post(pid) assert p.status == "sent" and p.message_id == 100 # the next suggestion moves on again await app.text("/new") assert "Download Episodes 41 to 60" in app.out.last_text await app.db.close() asyncio.run(run()) def test_customize_new_post_with_buttons_and_edit_live(tmp_path): async def run(): app = await make_app(tmp_path) await app.text("/new") # nothing to suggest yet -> asks for the post assert "Send the post now" in app.out.last_text await app.text("Hello *world*") # typed as Markdown: read as formatting (MARKUP_INPUT=false would keep it literal) assert "buttons" in app.out.last_text.lower() await app.text("Get it - https://t.me/bot?start=1 | Join - t.me/animech\nMirror - https://example.com/x") pid = int(app.out.callback_data("Publish").split(":")[1]) post = await app.db.get_post(pid) assert post.text == "Hello world" and post.entities == [{"k": "italic", "o": 6, "l": 5}] assert [len(r) for r in post.buttons] == [2, 1] await app.press(f"pb:{pid}") await app.press(f"pby:{pid}") assert app.tg.sent[-1][2] == "Hello world" # edit the live post: change a button's text, then its link await app.press(f"bs:{pid}:0:0") assert "Get it" in app.out.last_text await app.press(f"bt:{pid}:0:0") await app.text("Download now") await app.press(f"bu:{pid}:0:0") await app.text("https://t.me/goku?start=1") last_edit = app.tg.edits[-1][2] assert last_edit.message == "Hello world" b = last_edit.reply_markup.rows[0].buttons[0] assert b.text == "Download now" and button_url(b) == "https://t.me/goku?start=1" # reorder: move the first button right, then down into its own row await app.press(f"bm:{pid}:0:0:r") rows = (await app.db.get_post(pid)).buttons assert [b["t"] for b in rows[0]] == ["Join", "Download now"] await app.press(f"bm:{pid}:0:1:d") rows = (await app.db.get_post(pid)).buttons assert [[b["t"] for b in r] for r in rows] == [["Join"], ["Mirror", "Download now"]] # delete a button await app.press(f"bd:{pid}:0:0") assert [[b["t"] for b in r] for r in (await app.db.get_post(pid)).buttons] == [["Mirror", "Download now"]] live = app.tg.edits[-1][2].reply_markup assert [b.text for b in live.rows[0].buttons] == ["Mirror", "Download now"] # edit text of the live post keeps the buttons await app.press(f"et:{pid}") await app.text("New text") req = app.tg.edits[-1][2] assert req.message == "New text" and len(req.reply_markup.rows[0].buttons) == 2 # delete from the channel await app.press(f"dly:{pid}:a") assert app.tg.deleted == [[100]] and await app.db.get_post(pid) is None await app.db.close() asyncio.run(run()) def test_bad_input_and_permissions(tmp_path): async def run(): app = await make_app(tmp_path) await app.text("/new") await app.text("A post") await app.text("this has no link") assert "No link found" in norm(app.out.last_text) await app.text("/skip") pid = int(app.out.callback_data("Publish").split(":")[1]) await app.press(f"ba:{pid}") await app.text("Only text") # asks for the link next assert "send the link" in app.out.last_text.lower() await app.text("not a link") assert "doesn't look like a link" in norm(app.out.last_text) await app.text("https://t.me/bot?start=q") assert (await app.db.get_post(pid)).buttons[0][0]["u"] == "https://t.me/bot?start=q" # a non-owner is rejected everywhere app.uid = 999 await app.text("/swap @a @b") assert "private" in app.out.last_text.lower() await app.db.close() asyncio.run(run()) def test_drafts_list_and_discard(tmp_path): async def run(): app = await make_app(tmp_path) await app.text("/new") await app.text("draft one") await app.text("/skip") await app.text("/posts") await app.press("pc:d:0") assert "Drafts" in app.out.last_text and "draft one" in str(app.out.last_buttons()) pid = int(app.out.callback_data("draft one").split(":")[1]) await app.press(f"dly:{pid}:b") assert await app.db.get_post(pid) is None await app.db.close() asyncio.run(run())