Blame for tests/test_broadcast.py

"""/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())