Spaces:
Paused
Paused
| """/broadcast (and its spelling /boardcast): one message to every channel, optionally deleted again after a time.""" | |
| import asyncio | |
| from datetime import timedelta | |
| from types import SimpleNamespace | |
| import pytest | |
| from telethon import errors, types | |
| from app.db import Channel, as_utc, utcnow | |
| from app.expiry import MAX_ATTEMPTS, Expirer, human, parse_duration | |
| from app.tgutil import button_url | |
| from .fakes import FakeChannelClient, photo_media | |
| from .harness import App, norm | |
| def run(coro): | |
| return asyncio.run(coro) | |
| def posted(app): | |
| """Messages the bot sent to channels (not to people).""" | |
| return [e for e in app.tg.sent if isinstance(e[1], types.InputPeerChannel)] | |
| async def make_app(tmp_path, channels=2, **cfg): | |
| """Channel 1 "Anime Channel" is the app's own fake; channel 2 "Movie Channel" sits next to it. Posts the bot sends | |
| to a channel really appear in it, so deleting them really deletes.""" | |
| 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) | |
| if channels > 1: | |
| app.tg.add_channel(FakeChannelClient({}, username="moviech", cid=2, title="Movie Channel")) | |
| await app.db.save_channel(2, 6, "Movie Channel", "moviech", 1) | |
| async def send_message(peer, message="", **kw): | |
| if not isinstance(peer, types.InputPeerChannel): # a person (the preview) | |
| app.tg.sent.append(("text", peer, message, kw)) | |
| return SimpleNamespace(id=next(app.tg.ids)) | |
| app.tg.sent.append(("text", peer, message, kw)) | |
| return app.tg.registry[peer.channel_id]._new(message, kw.get("formatting_entities"), kw.get("buttons")) | |
| async def send_file(peer, file, **kw): | |
| app.tg.sent.append(("file", peer, file, kw)) | |
| if not isinstance(peer, types.InputPeerChannel): | |
| return SimpleNamespace(id=next(app.tg.ids)) | |
| return app.tg.registry[peer.channel_id]._new(kw.get("caption"), kw.get("formatting_entities"), kw.get("buttons"), media=file) | |
| app.tg.send_message, app.tg.send_file = send_message, send_file | |
| app.ctx.expirer = Expirer(app.ctx) | |
| return app | |
| async def broadcast(app, command="/broadcast 1h", text="Hello everyone", buttons=None, message=None): | |
| await app.text(command) | |
| await app.text(text, message=message) | |
| if buttons is None: | |
| await app.press(app.out.callback_data("No buttons")) | |
| else: | |
| await app.text(buttons) | |
| await app.press(app.out.callback_data("Send to all channels")) | |
| # ----------------------------------------------------------------------------------- the time | |
| def test_times_are_minutes_hours_days(): | |
| assert parse_duration("50m") == 3000 | |
| assert parse_duration("1h") == 3600 | |
| assert parse_duration("5d") == 5 * 86400 | |
| assert parse_duration("1d12h") == 129600 | |
| assert parse_duration("90 min") == 5400 | |
| assert parse_duration("2 Hours") == 7200 | |
| assert parse_duration("1w") == 604800 | |
| for bad in ("", "soon", "10", "h", "0m", "30s", "5x", "400d", "1h 5"): | |
| with pytest.raises(ValueError): | |
| parse_duration(bad) | |
| def test_times_are_spelled_out(): | |
| assert human(3000) == "50 minutes" | |
| assert human(3600) == "1 hour" | |
| assert human(5400) == "1 hour 30 minutes" | |
| assert human(129600) == "1 day 12 hours" | |
| assert human(5 * 86400) == "5 days" | |
| assert human(90061) == "1 day 1 hour" | |
| assert human(20) == "under a minute" | |
| # ----------------------------------------------------------------------------------- sending | |
| def test_a_timed_broadcast_goes_to_every_channel_and_is_remembered(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| before = utcnow() | |
| await app.text("/broadcast 1h") | |
| assert "2 channel(s)" in norm(app.out.last_text) and "1 hour" in norm(app.out.last_text) | |
| await app.text("Hello everyone") | |
| assert "Buttons?" in norm(app.out.last_text) | |
| await app.press(app.out.callback_data("No buttons")) | |
| shown = norm(app.out.last_text) | |
| assert "Send it to 2 channel(s)" in shown and "Anime Channel" in shown and "Movie Channel" in shown | |
| assert "Deleted again 1 hour after sending" in shown | |
| # the preview went to the person, nothing to the channels yet | |
| assert [e[2] for e in app.tg.sent if not isinstance(e[1], types.InputPeerChannel)] == ["Hello everyone"] | |
| assert posted(app) == [] | |
| await app.press(app.out.callback_data("Send to all channels")) | |
| report = norm(app.out.last_text) | |
| assert "Broadcast sent to 2 of 2" in report and "✅ Anime Channel" in report and "✅ Movie Channel" in report | |
| assert "deleted from these channels and from My posts in 1 hour" in report and "UTC" in report | |
| assert [e[2] for e in posted(app)] == ["Hello everyone", "Hello everyone"] | |
| assert app.tg.registry[1].msgs[1].message == "Hello everyone" and app.tg.registry[2].msgs[1].message == "Hello everyone" | |
| for cid in (1, 2): # in My posts, as published posts of the bot | |
| (p,) = await app.db.sent_posts(cid) | |
| assert (p.message_id, p.text, p.source, p.created_by) == (1, "Hello everyone", "bot", 1) | |
| rows = await app.db.pending_deletes() | |
| assert sorted((r.channel_id, r.message_id) for r in rows) == [(1, 1), (2, 1)] | |
| for r in rows: | |
| assert as_utc(r.delete_at) - before >= timedelta(hours=1) and as_utc(r.delete_at) - utcnow() <= timedelta(hours=1) | |
| assert len({r.group_id for r in rows}) == 1 | |
| assert app.ctx.expirer._next is not None | |
| await app.db.close() | |
| run(go()) | |
| def test_without_a_time_the_message_stays_and_the_other_spelling_works(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await broadcast(app, "/boardcast") | |
| report = norm(app.out.last_text) | |
| assert "Broadcast sent to 2 of 2" in report and "stays in the channels" in report | |
| assert await app.db.pending_deletes() == [] | |
| assert app.ctx.expirer._next is None | |
| assert len(posted(app)) == 2 | |
| await app.db.close() | |
| run(go()) | |
| def test_buttons_and_formatting(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| bold = types.MessageEntityBold(0, 5) | |
| msg = SimpleNamespace(raw_text="Hello all", entities=[bold], media=None, grouped_id=None, forward=None) | |
| await app.text("/broadcast 2d") | |
| await app.text("Hello all", message=msg) | |
| await app.text("Join - https://t.me/animech | Site - https://example.com/x\nMore - t.me/moviech") | |
| assert "Send it to 2 channel(s)" in norm(app.out.last_text) and "2 days" in norm(app.out.last_text) | |
| await app.press(app.out.callback_data("Send to all channels")) | |
| first = posted(app)[0] | |
| assert first[2] == "Hello all" | |
| assert [type(e).__name__ for e in first[3]["formatting_entities"]] == ["MessageEntityBold"] | |
| rows = first[3]["buttons"].rows | |
| assert [[b.text for b in r.buttons] for r in rows] == [["Join", "Site"], ["More"]] | |
| assert button_url(rows[1].buttons[0]) == "https://t.me/moviech" | |
| (p,) = await app.db.sent_posts(2) | |
| assert p.buttons == [[{"t": "Join", "u": "https://t.me/animech"}, {"t": "Site", "u": "https://example.com/x"}], [{"t": "More", "u": "https://t.me/moviech"}]] | |
| assert p.entities == [{"k": "bold", "o": 0, "l": 5}] | |
| await app.db.close() | |
| run(go()) | |
| def test_a_photo_with_a_caption(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| msg = SimpleNamespace(raw_text="New poster", entities=[], media=photo_media(5), grouped_id=None, forward=None) | |
| await broadcast(app, "/broadcast 30m", "New poster", message=msg) | |
| files = [e for e in posted(app) if e[0] == "file"] | |
| assert len(files) == 2 and files[0][3]["caption"] == "New poster" | |
| got = app.tg.registry[2].msgs[1] | |
| assert got.message == "New poster" and isinstance(got.media, types.MessageMediaPhoto) | |
| (p,) = await app.db.sent_posts(2) | |
| assert p.media_kind == "photo" and p.media_file_id.startswith("tl:") and p.text == "New poster" | |
| await app.db.close() | |
| run(go()) | |
| def test_a_channel_that_refuses_does_not_stop_the_others(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| real = app.tg.send_message | |
| async def picky(peer, message="", **kw): | |
| if getattr(peer, "channel_id", None) == 2: | |
| raise errors.ChatWriteForbiddenError(request=None) | |
| return await real(peer, message, **kw) | |
| app.tg.send_message = picky | |
| await broadcast(app, "/broadcast 1h") | |
| report = norm(app.out.last_text) | |
| assert "sent to 1 of 2" in report and "✅ Anime Channel" in report | |
| assert "❌ Movie Channel" in report and "ChatWriteForbiddenError" in report | |
| assert [r.channel_id for r in await app.db.pending_deletes()] == [1] # only what was sent expires | |
| assert await app.db.sent_posts(2) == [] | |
| await app.db.close() | |
| run(go()) | |
| def test_only_the_owner_and_only_good_input(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, admins=frozenset({7})) | |
| app.uid = 7 | |
| await app.text("/broadcast 1h") | |
| assert "private" in app.out.last_text and 7 not in app.ctx.state | |
| app.uid = 1 | |
| await app.text("/broadcast soon") | |
| assert "can't read" in norm(app.out.last_text) and 1 not in app.ctx.state | |
| await app.text("/broadcast 20s") | |
| assert "don't know the unit" in norm(app.out.last_text) | |
| await app.text("/broadcast 400d") | |
| assert "1 year" in app.out.last_text | |
| # an admin cannot press the owner's buttons either | |
| await app.text("/broadcast 1h") | |
| await app.text("Hi") | |
| await app.press(app.out.callback_data("No buttons")) | |
| confirm = app.out.callback_data("Send to all channels") | |
| app.uid = 7 | |
| await app.press(confirm) | |
| assert "Only the bot owner" in app.out.last_text and posted(app) == [] | |
| await app.db.close() | |
| run(go()) | |
| def test_nothing_is_sent_without_a_yes_and_a_button_works_once(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/broadcast") | |
| await app.text("Hi") | |
| await app.press(app.out.callback_data("No buttons")) | |
| await app.press(app.out.callback_data("Cancel")) | |
| assert "Cancelled" in app.out.last_text and posted(app) == [] | |
| await app.text("/broadcast") | |
| await app.text("Hi") | |
| await app.press(app.out.callback_data("No buttons")) | |
| send = app.out.callback_data("Send to all channels") | |
| await app.press(send) | |
| await app.press(send) | |
| assert "expired" in app.out.last_text | |
| assert len(posted(app)) == 2 | |
| await app.db.close() | |
| run(go()) | |
| def test_wrong_buttons_and_a_busy_bot_are_explained(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| await app.text("/broadcast 1h") | |
| await app.text("Hi") | |
| await app.text("no link here") | |
| assert "Try again" in norm(app.out.last_text) | |
| await app.text("Go - https://t.me/x") | |
| send = app.out.callback_data("Send to all channels") | |
| await app.ctx.lock.acquire() | |
| await app.press(send) | |
| assert "still running" in app.out.last_text and posted(app) == [] | |
| app.ctx.lock.release() | |
| await app.press(send) | |
| assert len(posted(app)) == 2 | |
| await app.db.close() | |
| run(go()) | |
| def test_no_channels_no_broadcast(tmp_path): | |
| async def go(): | |
| app = App(tmp_path, edit_delay=0) | |
| await app.start() | |
| await app.text("/broadcast 1h") | |
| assert "No channels yet" in app.out.last_text | |
| await app.db.close() | |
| run(go()) | |
| # ------------------------------------------------------------------------------------ deleting | |
| async def timed_broadcast(app, command="/broadcast 1h"): | |
| await broadcast(app, command) | |
| return utcnow() | |
| def test_when_the_time_is_up_the_message_goes_from_the_channels_and_my_posts(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| now = await timed_broadcast(app) | |
| ex = app.ctx.expirer | |
| early = await ex.sweep(now=now + timedelta(minutes=59)) | |
| assert early["deleted"] == 0 and 1 in app.tg.registry[1].msgs | |
| assert len(await app.db.sent_posts(1)) == 1 | |
| done = await ex.sweep(now=now + timedelta(hours=1, minutes=1)) | |
| assert done["deleted"] == 2 and done["retry"] == 0 and done["failed"] == 0 | |
| assert app.tg.registry[1].msgs == {} and app.tg.registry[2].msgs == {} | |
| assert await app.db.sent_posts(1) == [] and await app.db.sent_posts(2) == [] | |
| assert await app.db.pending_deletes() == [] | |
| assert ex._next is None # nothing left: the loop sleeps until something new is scheduled | |
| await app.db.close() | |
| run(go()) | |
| def test_other_posts_are_not_touched_and_a_message_deleted_by_hand_is_fine(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| keep = await app.db.create_post(channel_id=1, message_id=50, status="sent", source="bot", text="keep me", entities=[], buttons=[]) | |
| now = await timed_broadcast(app) | |
| app.tg.registry[2].msgs.pop(1) # an admin deleted the broadcast in channel 2 already | |
| done = await app.ctx.expirer.sweep(now=now + timedelta(hours=2)) | |
| assert done["deleted"] == 2 | |
| assert (await app.db.get_post(keep.id)) is not None | |
| assert [p.message_id for p in await app.db.sent_posts(1)] == [50] | |
| assert await app.db.sent_posts(2) == [] | |
| await app.db.close() | |
| run(go()) | |
| def test_a_restart_loses_nothing(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| now = await timed_broadcast(app, "/broadcast 5d") | |
| due = now + timedelta(days=5) | |
| fresh = Expirer(app.ctx) # a new process: it knows nothing but the database | |
| assert fresh._next is None | |
| await fresh.load() | |
| assert abs((fresh._next - due).total_seconds()) < 30 | |
| # it was off for a week: everything that came due is deleted when it is back | |
| await fresh.sweep(now=now + timedelta(days=7)) | |
| assert app.tg.registry[1].msgs == {} and app.tg.registry[2].msgs == {} | |
| assert await app.db.sent_posts(1) == [] | |
| await app.db.close() | |
| run(go()) | |
| def test_the_background_loop_wakes_when_something_is_scheduled_and_when_it_is_due(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| ex = app.ctx.expirer | |
| ex.start() | |
| await asyncio.sleep(0.05) # idle: nothing to do, nothing polled | |
| msg = app.tg._new("soon gone", None, None) | |
| when = utcnow() + timedelta(seconds=0.4) | |
| await app.db.schedule_deletes([dict(channel_id=1, message_id=msg.id, delete_at=when, group_id="g", created_by=1)]) | |
| ex.notify(when) | |
| for _ in range(60): | |
| if msg.id not in app.tg.msgs: | |
| break | |
| await asyncio.sleep(0.05) | |
| assert msg.id not in app.tg.msgs | |
| await ex.stop() | |
| await app.db.close() | |
| run(go()) | |
| def test_something_already_overdue_is_deleted_at_start(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| msg = app.tg._new("overdue", None, None) | |
| await app.db.schedule_deletes([dict(channel_id=1, message_id=msg.id, delete_at=utcnow() - timedelta(hours=3), group_id="g", created_by=1)]) | |
| ex = Expirer(app.ctx) | |
| ex.start() | |
| for _ in range(60): | |
| if msg.id not in app.tg.msgs: | |
| break | |
| await asyncio.sleep(0.05) | |
| assert msg.id not in app.tg.msgs | |
| await ex.stop() | |
| await app.db.close() | |
| run(go()) | |
| def test_a_running_job_makes_it_wait(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path) | |
| now = await timed_broadcast(app) | |
| ex = app.ctx.expirer | |
| await app.ctx.lock.acquire() | |
| later = now + timedelta(hours=2) | |
| out = await ex.sweep(now=later) | |
| assert out["waiting"] and out["deleted"] == 0 and 1 in app.tg.registry[1].msgs | |
| assert 50 <= (ex._next - later).total_seconds() <= 70 | |
| app.ctx.lock.release() | |
| out = await ex.sweep(now=later) | |
| assert out["deleted"] == 2 | |
| await app.db.close() | |
| run(go()) | |
| def test_what_the_bot_cannot_delete_is_retried_then_reported(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, channels=1) | |
| now = await timed_broadcast(app) | |
| app.tg.undeletable = {1} # Telegram refuses (an old post) | |
| ex = app.ctx.expirer | |
| t = now + timedelta(hours=2) | |
| for attempt in range(1, MAX_ATTEMPTS): | |
| out = await ex.sweep(now=t) | |
| assert out["retry"] == 1 and out["failed"] == 0, attempt | |
| (row,) = await app.db.pending_deletes() | |
| assert row.attempts == attempt and as_utc(row.delete_at) > t | |
| t = as_utc(row.delete_at) + timedelta(seconds=1) | |
| dms_before = len([e for e in app.tg.sent if not isinstance(e[1], types.InputPeerChannel)]) | |
| out = await ex.sweep(now=t) | |
| assert out["failed"] == 1 | |
| assert await app.db.pending_deletes() == [] | |
| (row,) = await app.db.pending_deletes(status="failed") | |
| assert row.attempts == MAX_ATTEMPTS and row.last_error | |
| # the person who sent it is told; the post stays in My posts because it is still in the channel | |
| told = [e for e in app.tg.sent if not isinstance(e[1], types.InputPeerChannel)][dms_before:] | |
| assert told and "could not delete the expired broadcast" in told[-1][2] | |
| assert len(await app.db.sent_posts(1)) == 1 | |
| assert ex._next is None | |
| await app.db.close() | |
| run(go()) | |
| def test_the_userbot_deletes_what_the_bot_may_not(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, channels=1) | |
| now = await timed_broadcast(app) | |
| app.tg.undeletable = {1} | |
| calls, closed = [], [] | |
| class StubUserbot: | |
| enabled = True | |
| async def fallback_for(self, ch): | |
| async def fn(ids): | |
| calls.append(list(ids)) | |
| for i in ids: | |
| app.tg.msgs.pop(i, None) | |
| async def close(): | |
| closed.append(1) | |
| return fn, SimpleNamespace(close=close) | |
| app.ctx.userbot = StubUserbot() | |
| out = await app.ctx.expirer.sweep(now=now + timedelta(hours=2)) | |
| assert out["deleted"] == 1 and calls == [[1]] and closed == [1] | |
| assert app.tg.msgs == {} and await app.db.sent_posts(1) == [] | |
| await app.db.close() | |
| run(go()) | |
| def test_a_channel_taken_out_of_the_bot_is_reported_not_retried_forever(tmp_path): | |
| async def go(): | |
| app = await make_app(tmp_path, channels=1) | |
| now = await timed_broadcast(app) | |
| for post in await app.db.sent_posts(1): # a real database refuses to drop a channel its posts still point at | |
| await app.db.delete_post(post.id) | |
| async with app.db.Session() as s: | |
| await s.delete(await s.get(Channel, 1)) | |
| await s.commit() | |
| out = await app.ctx.expirer.sweep(now=now + timedelta(hours=2)) | |
| assert out["failed"] == 1 and out["deleted"] == 0 | |
| assert await app.db.pending_deletes() == [] | |
| await app.db.close() | |
| run(go()) | |