Blame for tests/test_restricted_source.py

"""A source channel that TELEGRAM restricts (a copyright strike: "This channel is unavailable due to copyright
infringement"): /shift copies it from what My posts saved - every post that has a saved copy, whatever Telegram hands out
of it now - and never copies Telegram's notice. Every copy is read back, so the report can say when Telegram itself hides
the NEW posts (a blocked file, a restricted destination) instead of the bot being blamed for copying a notice."""
import asyncio

from telethon import functions

from app.repost_engine import CHECK_FIRST, RepostOptions, plan_repost, saved_lookup
from app.restrictions import message_restriction
from app.shift_engine import run_shift
from app.tgutil import button_url

from .fakes import make_msg, rpc
from .harness import norm
from .test_held_back import (
    buttons,
    held_app,
    held_setup,
    held_version,
    forwarded_ids,
    originals,
    shown,
    unreadable,
)
from .test_restricted import CHANNEL_NOTICE, NOTICE
from .test_shift import SRC


def run(coro):
    return asyncio.run(coro)


COPIES = {11: "Welcome", 12: "Episode 2 is out", 13: "Caption with link", 14: "Fifth", 15: "Sixth"}  # as My posts has them


def differ(src, ids):
    """What Telegram hands out of these posts now is not what My posts saved (so the source of a copy can be told)."""
    for i in ids:
        old = src.msgs[i]
        src.msgs[i] = make_msg(i, f"LIVE {i}", reply_to=old.reply_to.reply_to_msg_id if old.reply_to else None)


async def struck(tmp_path, *, no_row=(), live_differs=True, **kw):
    """The source is restricted as a whole; its posts are handed out intact (and, by default, differently from what My
    posts saved) and carry no mark of their own - the case that went wrong."""
    src, dst, db, dst_row, shift = await held_setup(tmp_path, held=kw.pop("held", ()), no_row=no_row, **kw)
    src.restriction = CHANNEL_NOTICE
    if live_differs:
        differ(src, [i for i in (1, 2, 3, 5, 6) if i in src.msgs and not message_restriction(src.msgs[i])])
    return src, dst, db, dst_row, shift


# ======================================================================================== the copy loop
def test_a_struck_channel_is_copied_from_my_posts_even_when_telegram_still_hands_the_posts_out(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path)
        dst.fail["ForwardMessagesRequest"] = rpc("MessageIdInvalidError")  # what the real run met: nothing can be forwarded
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert (res.failed, res.aborted, res.held_back) == ([], None, [])
        assert res.from_saved == 5 and res.copied_units == 5 and res.live_restricted == 0 and res.held_why == CHANNEL_NOTICE
        assert shown(dst) == COPIES  # My posts' texts - not "LIVE n"
        assert forwarded_ids(dst) == []  # Telegram is not even asked to copy anything out of the struck channel
        # text, formatting, buttons, media and links come from My posts
        assert [(e.offset, e.length) for e in dst.msgs[12].entities] == [(0, 9)]
        btn = dst.msgs[12].reply_markup.rows[0].buttons[0]
        assert (btn.text, button_url(btn)) == ("Download", "https://t.me/goku?start=AAA")
        assert dst.msgs[13].media.photo.id == 31
        assert dst.msgs[13].entities[0].url == "https://t.me/dstch/14"  # the link to post 5 follows its copy
        assert dst.msgs[14].reply_to.reply_to_msg_id == 12  # an answer answers the copy
        # My posts knows the copies as they are, and the source was only read
        assert (await db.find_by_message(2, 12)).text == "Episode 2 is out"
        assert not [r for r in src.requests if type(r).__name__.startswith(("Edit", "Delete", "Send", "Forward"))]
        await db.close()

    run(go())


def test_a_struck_channel_is_copied_from_my_posts_when_telegram_hands_out_only_the_notice(tmp_path):
    """The same channel when the apps show the notice: the posts carry Telegram's mark, or only its words as the text."""

    async def go():
        for flagged in (True, False):
            src, dst, db, dst_row, shift = await struck(tmp_path / str(flagged), held=(1, 2, 3, 5, 6), flagged=flagged)
            res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
            assert (res.failed, res.aborted, res.held_back) == ([], None, [])
            assert res.from_saved == 5 and shown(dst) == COPIES
            assert not any(NOTICE in (m.message or "") for m in dst.msgs.values())
            await db.close()

    run(go())


def test_a_post_of_a_struck_channel_without_a_saved_copy_is_copied_as_telegram_hands_it_out(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path, no_row=(3, 5))
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert (res.failed, res.aborted, res.held_back) == ([], None, [])
        assert res.from_saved == 3 and res.live_restricted == 2  # 1 2 6 from My posts; 3 and 5 as they are now
        assert shown(dst) == {11: "Welcome", 12: "Episode 2 is out", 13: "LIVE 3", 14: "LIVE 5", 15: "Sixth"}
        assert forwarded_ids(dst) == []  # not even those: a restricted post is never copied by Telegram itself
        await db.close()

    run(go())


def test_a_post_of_a_struck_channel_with_only_the_notice_and_no_saved_copy_is_skipped(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path, held=(3,), no_row=(3,), flagged=False)
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.held_back == [3] and res.from_saved == 4 and res.failed == []
        assert shown(dst) == {11: "Welcome", 12: "Episode 2 is out", 13: "Fifth", 14: "Sixth"}
        assert not any(NOTICE in (m.message or "") for m in dst.msgs.values())
        await db.close()

    run(go())


def test_a_saved_copy_that_is_telegrams_notice_is_never_posted(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path, no_row=(2, 3))
        for mid in (2, 3):  # what an older version saved when it synced the struck channel
            await db.create_post(
                channel_id=1, message_id=mid, status="sent", source="adopted", created_by=1, link_preview=False,
                text=NOTICE, entities=[], buttons=[], media_kind=None, media_file_id=None,
            )
        src.msgs[3] = held_version(originals()[3], flagged=False)  # post 3: the notice from Telegram as well
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        # post 2: the saved copy is useless but Telegram hands the post out; post 3: nothing but notices
        assert res.held_back == [3] and res.live_restricted == 1 and res.from_saved == 3 and res.failed == []
        assert shown(dst) == {11: "Welcome", 12: "LIVE 2", 13: "Fifth", 14: "Sixth"}
        assert not any(NOTICE in (m.message or "") for m in dst.msgs.values())
        await db.close()

    run(go())


def test_a_source_that_cannot_be_read_never_gets_a_saved_notice_posted(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(), no_row=(2,))
        await db.create_post(
            channel_id=1, message_id=2, status="sent", source="adopted", created_by=1, link_preview=False,
            text=NOTICE, entities=[], buttons=[], media_kind=None, media_file_id=None,
        )
        unreadable(src)
        await db.set_mark("s1", "fromdb")
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.copied_units == 4 and res.failed == []
        assert shown(dst) == {11: "Welcome", 12: "Caption with link", 13: "Fifth", 14: "Sixth"}
        await db.close()

    run(go())


def test_a_channel_nothing_restricts_is_copied_as_before(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=())
        differ(src, (1, 2, 3, 5, 6))
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.from_saved == 0 and res.live_restricted == 0 and res.held_why is None and res.held_back == []
        assert shown(dst) == {11: "LIVE 1", 12: "LIVE 2", 13: "LIVE 3", 14: "LIVE 5", 15: "LIVE 6"}  # what the channel says
        await db.close()

    run(go())


def test_lifting_the_strike_makes_the_next_run_an_ordinary_one(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path, last=3)
        stopper = {"n": 0}

        async def prog(r):
            stopper["n"] = r.copied_units

        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0, progress=prog, should_stop=lambda: stopper["n"] >= 1)
        assert res.stopped and shown(dst) == {11: "Welcome"}  # post 1, from My posts
        src.restriction = None  # the strike is lifted: the channel says what it says now
        again = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert again.held_why is None and again.from_saved == 0
        assert shown(dst) == {11: "Welcome", 12: "LIVE 2", 13: "LIVE 3"}
        await db.close()

    run(go())


# ================================================================================================ the plan
def test_the_plan_counts_a_struck_channel_as_copied_from_my_posts(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path, no_row=(5,))
        lookup = saved_lookup(db, 1)
        plan = await plan_repost(src, SRC, RepostOptions(), saved=lookup, use_saved=True)
        assert plan.channel_restricted == CHANNEL_NOTICE and plan.held_why == CHANNEL_NOTICE
        # 1 2 3 6 have a saved copy; 5 has none but Telegram still hands it out
        assert (plan.units, plan.held_saved, plan.live_restricted, plan.held_skipped, plan.service) == (5, 4, 1, 0, 1)
        assert plan.media == 1  # post 3 is counted from its saved copy, not from what Telegram hands out now
        # /repost copies the posts of the channel itself, post by post: a struck channel changes nothing about that
        plain = await plan_repost(src, SRC, RepostOptions(), saved=lookup)
        assert (plain.units, plain.held_saved, plain.live_restricted, plain.held_skipped) == (5, 0, 0, 0)
        assert plain.channel_restricted == CHANNEL_NOTICE
        await db.close()

    run(go())


# ========================================================================================= the read-back
def test_every_copy_of_a_struck_channel_is_read_back_and_what_telegram_hides_is_counted(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path)
        dst.restrict_new = lambda m: CHANNEL_NOTICE if m.media is not None else None  # the file of post 3 is blocked
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.copied_units == 5 and res.failed == [] and res.held_back == []
        assert res.checked == 5  # every copy of a restricted source
        assert res.hidden == [(3, True)] and res.hidden_why == CHANNEL_NOTICE and res.dst_restricted is None
        # the bot posted the saved text; the notice is Telegram's mark on the new post
        assert dst.msgs[13].message == "Caption with link" and message_restriction(dst.msgs[13]) == CHANNEL_NOTICE
        assert not any(NOTICE in (m.message or "") for m in dst.msgs.values())
        await db.close()

    run(go())


def test_a_destination_that_is_restricted_itself_is_named(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path)
        dst.restriction = CHANNEL_NOTICE
        dst.restrict_new = lambda m: CHANNEL_NOTICE
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert [i for i, _ in res.hidden] == [1, 2, 3] and res.dst_restricted == CHANNEL_NOTICE  # it stopped after 3
        assert res.hidden_stop
        await db.close()

    run(go())


def test_a_normal_channel_is_only_sampled_when_the_copies_are_read_back(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=())
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.copied_units == 5 and res.checked == CHECK_FIRST and res.hidden == [] and res.dst_restricted is None
        assert not [r for r in dst.requests if isinstance(r, functions.channels.GetFullChannelRequest)]  # nothing to ask
        await db.close()

    run(go())


def test_a_read_back_that_fails_does_not_fail_the_copy(tmp_path):
    async def go():
        for error in (rpc("ChannelPrivateError"), ConnectionError("the network dropped")):  # Telegram's refusal, or not
            src, dst, db, dst_row, shift = await struck(tmp_path / type(error).__name__, last=1)  # post 1: no buttons to read back
            dst.fail["GetMessagesRequest"] = error
            res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
            assert res.copied_units == 1 and res.failed == [] and res.checked == 0 and res.hidden == []
            assert shown(dst) == {11: "Welcome"}
            await db.close()

    run(go())


def test_a_source_that_cannot_be_asked_about_its_restriction_is_copied_as_an_unrestricted_one(tmp_path):
    async def go():
        for error in (rpc("ChannelPrivateError"), ConnectionError("the network dropped")):
            src, dst, db, dst_row, shift = await held_setup(tmp_path / type(error).__name__, held=())
            src.fail["GetFullChannelRequest"] = error
            res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
            assert res.copied_units == 5 and res.failed == [] and res.held_why is None and res.from_saved == 0
            await db.close()

    run(go())


# ================================================================================================ the owner's view
def test_the_plan_and_the_result_for_a_struck_channel_read_as_they_should(tmp_path):
    async def go():
        app, dst = await held_app(tmp_path, held=())
        app.tg.restriction = CHANNEL_NOTICE
        dst.fail["ForwardMessagesRequest"] = rpc("MessageIdInvalidError")
        await app.text("/shift @srcch @dstch")
        plan = norm(app.out.last_text)
        assert "Telegram restricts Anime Source itself" in plan
        assert "5 post(s) of Anime Source are restricted by Telegram" in plan and "copied from what My posts saved" in plan
        assert "5 posts" in plan and "The new posts are read back" in plan

        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "5 post(s) are restricted by Telegram" in done and "copied from what My posts saved of them" in done
        assert "5 message(s) copied" in done and "hides" not in done
        assert [m.message for i, m in sorted(dst.msgs.items()) if i > 10] == list(COPIES.values())
        await app.db.close()

    run(go())


def test_the_report_says_when_telegram_itself_hides_the_new_posts(tmp_path):
    async def go():
        app, dst = await held_app(tmp_path, held=())
        app.tg.restriction = CHANNEL_NOTICE
        dst.restrict_new = lambda m: CHANNEL_NOTICE if m.media is not None else None
        await app.text("/shift @srcch @dstch")
        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "Telegram itself hides 1 of the 5 new post(s) I read back" in done
        assert "carry its restriction mark" in done and "never posts one" in done
        assert "1 post(s) of these carry a file" in done and "--no-media" in done
        assert "The saved posts in My posts are intact" in done
        assert "5 message(s) copied" in done  # the copies exist: Undo removes them
        assert any("Undo" in t for row in app.out.last_buttons() for t in row)
        await app.db.close()

    run(go())


def test_the_report_names_a_restricted_destination(tmp_path):
    async def go():
        app, dst = await held_app(tmp_path, held=())
        app.tg.restriction = CHANNEL_NOTICE
        dst.restriction = CHANNEL_NOTICE
        dst.restrict_new = lambda m: CHANNEL_NOTICE
        await app.text("/shift @srcch @dstch")
        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "Telegram itself hides 3 of the 3 new post(s) I read back" in done
        assert "Telegram restricts the destination channel itself" in done
        await app.db.close()

    run(go())


def test_a_repost_plan_for_a_struck_channel_points_at_shift(tmp_path):
    from .test_repost_flow import make_app

    async def go():
        app = await make_app(tmp_path, msgs={1: make_msg(1, "First post"), 2: make_msg(2, "Second post")})
        app.tg.restriction = CHANNEL_NOTICE
        await app.text("/repost --no-typed")
        await app.press(app.out.callback_data("Test"))
        plan = norm(app.out.last_text)
        assert "Telegram restricts Test itself" in plan and "use /shift" in plan
        assert "2 posts" in plan and "no saved copy" not in plan  # /repost does not look at My posts
        await app.db.close()

    run(go())


# ====================================================================== stopping when Telegram hides the copies
def test_a_shift_stops_by_itself_when_telegram_hides_every_new_copy(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path)
        dst.restrict_new = lambda m: CHANNEL_NOTICE  # every new post is hidden
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.stopped and res.hidden_stop and res.copied_units == 3  # not 5: it did not go on
        assert [i for i, _ in res.hidden] == [1, 2, 3] and res.checked == 3
        assert shown(dst) == {11: "Welcome", 12: "Episode 2 is out", 13: "Caption with link"}
        assert res.phase == "copy" and res.final is None  # the finishing touches wait for the rest
        await db.close()

    run(go())


def test_the_owner_can_go_on_anyway_and_it_does_not_stop_again(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path)
        dst.restrict_new = lambda m: CHANNEL_NOTICE
        first = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert first.hidden_stop and first.copied_units == 3
        await db.set_mark("s1", "hiddenok")  # what pressing Continue anyway rests on
        again = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert not again.stopped and not again.hidden_stop and again.copied_units == 2 and again.skipped_done == 3
        assert shown(dst) == COPIES
        await db.close()

    run(go())


def test_a_shift_does_not_stop_when_only_some_copies_are_hidden(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path)
        dst.restrict_new = lambda m: CHANNEL_NOTICE if m.media is not None else None  # only the photo
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert not res.stopped and res.copied_units == 5 and res.hidden == [(3, True)]
        await db.close()

    run(go())


def test_a_shift_with_fewer_copies_than_the_threshold_is_not_stopped(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await struck(tmp_path, last=2)
        dst.restrict_new = lambda m: CHANNEL_NOTICE
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert not res.stopped and res.copied_units == 2 and len(res.hidden) == 2  # reported, nothing left to stop
        await db.close()

    run(go())


def test_the_stopped_shift_explains_itself_and_continue_anyway_goes_on(tmp_path):
    async def go():
        app, dst = await held_app(tmp_path, held=())
        app.tg.restriction = CHANNEL_NOTICE
        dst.restrict_new = lambda m: CHANNEL_NOTICE
        await app.text("/shift @srcch @dstch")
        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "stopped" in done and "3 message(s) copied so far" in done
        assert "I stopped after 3 post(s)" in done and "Telegram hides every new copy I read back" in done
        assert "Telegram itself hides 3 of the 3 new post(s) I read back" in done
        assert any("Continue" in t for t in buttons(app)) and any("Undo" in t for t in buttons(app))
        sid = app.out.callback_data("Undo").split(":")[1]
        assert await app.db.marks_of(sid) == {"hiddenok"}
        assert len([i for i in dst.msgs if i > 10]) == 3  # three posts, not the whole channel

        await app.press(app.out.callback_data("Continue"))
        done = norm(app.out.last_text)
        assert "5 message(s) copied" in done and "I stopped" not in done  # it went on, as the owner chose
        assert len([i for i in dst.msgs if i > 10]) == 5
        await app.db.close()

    run(go())