Blame for tests/test_syncer.py

"""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