Blame for tests/test_held_back.py

"""A post Telegram holds back - a copyright strike shows "This message couldn't be displayed on your device due to
copyright infringement" instead of the post: /shift builds the copy from what My posts saved of it, /repost leaves it
alone. The notice itself is never copied."""
import asyncio

from telethon import functions, types

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

from .dbutil import db_url_for, fresh_db
from .fakes import FakeChannelClient, make_msg, markup, photo_media, rpc, service_msg, url_btn
from .harness import norm
from .test_repost_flow import fill, make_app
from .test_restricted import CHANNEL_NOTICE, NOTICE, strike
from .test_shift import SRC, old_posts, shift_app
from .test_sync import saved


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


def originals():
    """The posts of the source channel as they were before the strike."""
    return {
        1: make_msg(1, "Welcome", [types.MessageEntityBold(0, 7)]),
        2: make_msg(
            2, "Episode 2 is out", [types.MessageEntityBold(0, 9)],
            markup=markup([url_btn("Download", "https://t.me/goku?start=AAA")]),
        ),
        3: make_msg(3, "Caption with link", [types.MessageEntityTextUrl(13, 4, "https://t.me/srcch/5")], media=photo_media(31)),
        4: service_msg(4),
        5: make_msg(5, "Fifth", reply_to=2),
        6: make_msg(6, "Sixth"),
    }


def held_version(m, flagged=True):
    """What Telegram shows of post `m` after the strike: the notice in place of the post (and, if `flagged`, its own
    restriction mark on the message). Id, album and reply are still there."""
    h = make_msg(
        m.id, NOTICE, out=m.out, grouped_id=m.grouped_id, reply_to=m.reply_to.reply_to_msg_id if m.reply_to else None
    )
    return strike(h) if flagged else h


def live_posts(orig, held, flagged=True):
    return {i: (held_version(m, flagged) if i in held else m) for i, m in orig.items()}


async def save_rows(db, orig, skip=()):
    for i, m in orig.items():
        if i not in skip and not isinstance(m, types.MessageService):
            await saved(db, m)


async def held_setup(tmp_path, *, held=(2, 3, 5), no_row=(), flagged=True, file_available=True, orig=None, last=None):
    """The source (id 1) after the strike, the destination (id 2) and a shift s1 over all of it; My posts has what the
    source's posts looked like before."""
    orig = originals() if orig is None else orig
    src = FakeChannelClient(live_posts(orig, held, flagged), username="srcch", title="Anime Source")
    dst = src.add_channel(FakeChannelClient(old_posts(), username="dstch", cid=2, title="Anime Backup"))
    if file_available:  # Telegram still knows the files: they were not taken down
        for m in orig.values():
            src._remember_media(m)
    db = await fresh_db(db_url_for(tmp_path))
    await db.save_channel(1, 5, "Anime Source", "srcch", 1)
    await db.save_channel(2, 6, "Anime Backup", "dstch", 1)
    await save_rows(db, orig, no_row)
    await db.create_shift("s1", src=SRC, dst_channel_id=2, first_id=1, last_id=last or max(orig), user_id=1)
    return src, dst, db, await db.get_channel(2), await db.get_shift("s1")


def forwarded_ids(chan):
    return [i for r in chan.requests if isinstance(r, functions.messages.ForwardMessagesRequest) for i in r.id]


def sent(chan, kind):
    return [r for r in chan.requests if isinstance(r, kind)]


def shown(chan):
    return {i: m.message for i, m in chan.msgs.items() if i > 10}


# ============================================================================================ /shift, engine
def test_a_held_back_post_is_copied_from_my_posts_and_never_as_the_notice(tmp_path):
    async def go():
        for flagged in (True, False):  # Telegram's own mark on the message, or only the notice as its text
            src, dst, db, dst_row, shift = await held_setup(tmp_path / str(flagged), 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, res.media_lost) == ([], None, [], [])
            assert res.from_saved == 3 and res.copied_units == 5 and res.used_saved
            # copies: 1->11 2->12 3->13 5->14 6->15 (4 is a service message)
            assert shown(dst) == {
                11: "Welcome", 12: "Episode 2 is out", 13: "Caption with link", 14: "Fifth", 15: "Sixth"
            }
            assert not any(NOTICE in (m.message or "") for m in dst.msgs.values())
            # only the posts Telegram shows as they are were copied by Telegram itself
            assert forwarded_ids(dst) == [1, 6]
            # text, formatting and buttons 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")
            # the photo of the saved post is there, and the link to the post that was copied later points at its copy
            assert dst.msgs[13].media.photo.id == 31
            assert dst.msgs[13].entities[0].url == "https://t.me/dstch/14"
            # a post that answered another answers the copy of it
            assert dst.msgs[14].reply_to.reply_to_msg_id == 12
            # My posts knows the copies as they are
            row = await db.find_by_message(2, 13)
            assert (row.text, row.media_kind) == ("Caption with link", "photo")
            assert (await db.find_by_message(2, 12)).buttons == [[{"t": "Download", "u": "https://t.me/goku?start=AAA"}]]
            # the source is only read
            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_post_that_telegram_shows_as_it_is_still_comes_from_the_channel(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(2,))
        # My posts has an older text for post 1 (somebody edited it and the sync did not run yet): the channel wins
        row = await db.find_by_message(1, 1)
        await db.update_post(row.id, text="an older text")
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.from_saved == 1 and res.held_back == []
        assert dst.msgs[11].message == "Welcome" and dst.msgs[12].message == "Episode 2 is out"
        assert forwarded_ids(dst) == [1, 3, 6]  # post 5 answers a post, so it is posted again; 2 comes from My posts
        await db.close()

    run(go())


def test_a_post_with_telegrams_mark_comes_from_my_posts_even_when_telegram_still_hands_it_out(tmp_path):
    """The rule of a struck channel: what My posts saved wins over what Telegram hands out now - however intact that
    looks. Only a post without a saved copy is copied as Telegram hands it out (it is the post itself, not a notice)."""

    async def go():
        orig = originals()
        live = dict(orig)
        live[2] = strike(make_msg(2, "Episode 2 is out", [types.MessageEntityBold(0, 9)], markup=orig[2].reply_markup))
        live[3] = strike(make_msg(3, "Caption as Telegram hands it out", media=photo_media(31)))  # differs from My posts
        src = FakeChannelClient(live, username="srcch", title="Anime Source")
        dst = src.add_channel(FakeChannelClient(old_posts(), username="dstch", cid=2, title="Anime Backup"))
        for m in orig.values():
            src._remember_media(m)
        db = await fresh_db(db_url_for(tmp_path))
        await db.save_channel(1, 5, "Anime Source", "srcch", 1)
        await db.save_channel(2, 6, "Anime Backup", "dstch", 1)
        await save_rows(db, orig, skip=(2,))  # no saved copy of post 2; post 3 has one
        await db.create_shift("s1", src=SRC, dst_channel_id=2, first_id=1, last_id=6, user_id=1)
        res = await run_shift(src, db, await db.get_shift("s1"), await db.get_channel(2), 99, delay=0, settle_pause=0)
        assert res.held_back == [] and res.failed == [] and res.copied_units == 5
        assert res.from_saved == 1  # post 3
        assert res.live_restricted == 1  # post 2: marked, but nothing saved to use instead and its content is intact
        assert shown(dst) == {11: "Welcome", 12: "Episode 2 is out", 13: "Caption with link", 14: "Fifth", 15: "Sixth"}
        assert forwarded_ids(dst) == [1, 6]  # a restricted post is never copied by Telegram itself
        await db.close()

    run(go())


def test_a_post_that_telegram_empties_out_is_held_back_too(tmp_path):
    async def go():
        orig = originals()
        live = dict(orig)
        live[2] = strike(make_msg(2, ""), "Removed after a copyright complaint")  # nothing of the post is handed out
        src = FakeChannelClient(live, username="srcch", title="Anime Source")
        dst = src.add_channel(FakeChannelClient(old_posts(), username="dstch", cid=2, title="Anime Backup"))
        db = await fresh_db(db_url_for(tmp_path))
        await db.save_channel(1, 5, "Anime Source", "srcch", 1)
        await db.save_channel(2, 6, "Anime Backup", "dstch", 1)
        await save_rows(db, orig)
        await db.create_shift("s1", src=SRC, dst_channel_id=2, first_id=1, last_id=6, user_id=1)
        res = await run_shift(src, db, await db.get_shift("s1"), await db.get_channel(2), 99, delay=0, settle_pause=0)
        assert res.from_saved == 1 and res.held_back == [] and res.failed == []
        assert dst.msgs[12].message == "Episode 2 is out" and res.held_why == "Removed after a copyright complaint"
        await db.close()

    run(go())


def test_telegrams_notice_is_never_posted_even_when_my_posts_holds_it_as_the_text(tmp_path):
    """A saved copy made from a restricted post can hold Telegram's notice as its text (an older version did that), and a
    post may really say it: either way it is indistinguishable from the notice, so it is not copied - and reported."""

    async def go():
        orig = originals()
        orig[6] = make_msg(6, NOTICE)
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(2, 3, 5), orig=orig, flagged=False)
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.from_saved == 3 and res.held_back == [6] and res.failed == []
        assert shown(dst) == {11: "Welcome", 12: "Episode 2 is out", 13: "Caption with link", 14: "Fifth"}
        assert not any(NOTICE in (m.message or "") for m in dst.msgs.values())
        assert 6 not in forwarded_ids(dst) and res.held_why == PLACEHOLDER_NOTE
        await db.close()

    run(go())


def test_a_saved_copy_that_holds_the_notice_is_not_used_for_a_restricted_post(tmp_path):
    async def go():
        orig = originals()
        orig[5] = make_msg(5, "Fifth")  # answers nothing
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(2, 3), no_row=(2,), orig=orig)
        await db.create_post(  # what an older version saved when it synced the struck 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,
        )
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        # post 2: Telegram shows the notice and the saved copy is the notice too: nothing to copy; post 3 is fine
        assert res.held_back == [2] and res.from_saved == 1 and res.failed == []
        assert shown(dst) == {11: "Welcome", 12: "Caption with link", 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_an_answer_to_a_post_that_could_not_be_copied_follows_when_that_post_has_its_copy(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(2,), no_row=(2,))  # post 5 answers post 2
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.held_back == [2] and res.failed == [(5, "ReplyParentMissing")]  # it would answer nothing
        assert shown(dst) == {11: "Welcome", 12: "Caption with link", 13: "Sixth"}
        await save_rows(db, {2: originals()[2]})  # the saved copy arrives
        again = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert again.held_back == [] and again.failed == [] and again.copied_units == 2
        assert dst.msgs[14].message == "Episode 2 is out" and dst.msgs[15].message == "Fifth"
        assert dst.msgs[15].reply_to.reply_to_msg_id == 14  # the answer answers the copy
        await db.close()

    run(go())


def test_a_held_back_post_without_a_saved_copy_is_skipped_and_reported(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, no_row=(3,))
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.held_back == [3] and res.from_saved == 2 and res.failed == [] and res.aborted is None
        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())
        assert res.held_why == NOTICE
        # once My posts has it (a backup put back with /import) the rest can be done
        await save_rows(db, {3: originals()[3]}, skip=())
        again = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert again.held_back == [] and again.from_saved == 1 and again.copied_units == 1 and again.skipped_done == 4
        assert dst.msgs[15].message == "Caption with link" and dst.msgs[15].media.photo.id == 31
        await db.close()

    run(go())


def test_a_channel_that_is_not_in_my_posts_at_all_has_nothing_to_copy_held_back_posts_from(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(1, 2, 3, 5, 6), no_row=(1, 2, 3, 5, 6))
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.held_back == [1, 2, 3, 5, 6] and res.copied_units == 0 and res.failed == []
        assert shown(dst) == {}
        await db.close()

    run(go())


def test_a_post_with_nothing_saved_in_it_is_not_posted_empty(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(6,), no_row=(6,))
        await db.create_post(
            channel_id=1, message_id=6, status="sent", source="adopted", created_by=1, link_preview=False,
            text="  ", entities=[], buttons=[], media_kind=None, media_file_id=None,
        )
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.held_back == [6] and 15 not in dst.msgs
        await db.close()

    run(go())


# ------------------------------------------------------------------------------------------------ the media
def test_a_file_reference_that_has_expired_is_tried_again_without_it(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path)
        original = dst._do_SendMediaRequest
        tries = []

        def picky(req):
            tries.append(req.media.id.file_reference)
            if req.media.id.file_reference:  # the saved reference is old
                raise rpc("FileReferenceExpiredError")
            return original(req)

        dst._do_SendMediaRequest = picky
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert tries == [b"ref31", b""] and res.media_lost == [] and res.failed == []
        assert dst.msgs[13].media.photo.id == 31
        await db.close()

    run(go())


def test_a_file_telegram_refuses_goes_out_without_it_and_is_reported(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, file_available=False)  # taken down
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.media_lost == [3] and res.failed == [] and res.from_saved == 3 and res.aborted is None
        assert dst.msgs[13].message == "Caption with link" and dst.msgs[13].media is None
        assert len(sent(dst, functions.messages.SendMediaRequest)) == 2  # as saved, then without the file reference
        # the entry in My posts says what the post really is
        row = await db.find_by_message(2, 13)
        assert (row.text, row.media_kind, row.media_file_id) == ("Caption with link", None, None)
        await db.close()

    run(go())


def test_a_post_that_is_only_a_refused_file_fails_and_the_others_are_copied(tmp_path):
    async def go():
        orig = originals()
        orig[7] = make_msg(7, "", media=photo_media(71))
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(7,), orig=orig, file_available=False, last=7)
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.failed == [(7, "MediaEmptyError")] and res.media_lost == []
        assert res.copied_units == 5 and 16 not in dst.msgs  # 1 2 3 5 6 were copied; 7 has no text to go out with
        await db.close()

    run(go())


def test_a_held_back_album_is_rebuilt_as_an_album(tmp_path):
    async def go():
        orig = originals()
        orig[8] = make_msg(8, "", media=photo_media(81), grouped_id=80)
        orig[9] = make_msg(9, "album caption", [types.MessageEntityBold(0, 5)], media=photo_media(82), grouped_id=80)
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(8, 9), orig=orig, last=9)
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.from_saved == 1 and res.held_back == [] and res.failed == []  # an album is one post
        album = [dst.msgs[i] for i in (16, 17)]
        assert album[0].grouped_id and album[0].grouped_id == album[1].grouped_id
        assert [m.media.photo.id for m in album] == [81, 82]
        assert album[1].message == "album caption" and [(e.offset, e.length) for e in album[1].entities] == [(0, 5)]
        await db.close()

    run(go())


def test_an_album_of_which_one_picture_has_no_saved_copy_is_skipped_whole(tmp_path):
    async def go():
        orig = originals()
        orig[8] = make_msg(8, "", media=photo_media(81), grouped_id=80)
        orig[9] = make_msg(9, "album caption", media=photo_media(82), grouped_id=80)
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(8, 9), no_row=(9,), orig=orig, last=9)
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.held_back == [8] and 16 not in dst.msgs
        await db.close()

    run(go())


# ------------------------------------------------------------------------------------------------ the plan
def test_the_plan_counts_what_is_copied_from_my_posts_and_what_is_skipped(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, no_row=(5,))
        lookup = saved_lookup(db, 1)
        plan = await plan_repost(src, SRC, RepostOptions(), saved=lookup, use_saved=True)
        # posts 1 2 3 6 are copied (2 and 3 from My posts); 5 is held back without a saved copy; 4 is a service message
        assert (plan.units, plan.held_saved, plan.held_skipped, plan.service) == (4, 2, 1, 1)
        assert plan.held_why == NOTICE and plan.channel_restricted is None
        assert plan.media == 1 and plan.n_buttons == 0 and plan.error is None  # counted from the saved posts
        # as /repost sees it: nothing held back is copied
        plan = await plan_repost(src, SRC, RepostOptions(), saved=lookup)
        assert (plan.units, plan.held_saved, plan.held_skipped) == (2, 0, 3)
        await db.close()

    run(go())


def test_the_plan_says_so_when_the_whole_channel_is_restricted_and_nothing_can_be_copied(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=(1, 2, 3, 5, 6), no_row=(1, 2, 3, 5, 6))
        src.restriction = NOTICE
        plan = await plan_repost(src, SRC, RepostOptions(), saved=saved_lookup(db, 1), use_saved=True)
        assert plan.channel_restricted == NOTICE and plan.units == 0 and plan.held_skipped == 5
        assert "holds back every post" in plan.error and "no saved copy" in plan.error
        await db.close()

    run(go())


def test_a_normal_channel_is_planned_as_before(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=())
        plan = await plan_repost(src, SRC, RepostOptions(), saved=saved_lookup(db, 1), use_saved=True)
        assert (plan.units, plan.held_saved, plan.held_skipped, plan.held_why, plan.channel_restricted) == (5, 0, 0, None, None)
        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert res.from_saved == 0 and res.held_back == [] and res.copied_units == 5
        await db.close()

    run(go())


# ============================================================================================ the owner's view
async def held_app(tmp_path, *, held=(2, 3, 5), no_row=(), file_available=True):
    app, dst = await shift_app(tmp_path)
    orig = originals()
    app.tg._media.clear()  # the account knows no file yet (shift_app had filled the channel with other posts)
    fill(app.tg, live_posts(orig, held))
    if file_available:
        for m in orig.values():
            app.tg._remember_media(m)
    await save_rows(app.db, orig, no_row)
    return app, dst


def buttons(app):
    return [t for row in app.out.last_buttons() for t in row]


def test_the_shift_plan_and_result_explain_what_comes_from_my_posts(tmp_path):
    async def go():
        app, dst = await held_app(tmp_path, no_row=(5,))
        app.tg.restriction = CHANNEL_NOTICE  # the whole channel is struck: My posts is used for every post it has
        await app.text("/shift @srcch @dstch")
        plan = norm(app.out.last_text)
        assert "Telegram restricts Anime Source itself" in plan and "unavailable due to copyright infringement" in plan
        assert "4 post(s) of Anime Source are restricted by Telegram" in plan
        assert "copied from what My posts saved of them" in plan and "not from what Telegram shows now" in plan
        assert "Telegram shows only its notice for 1 more post(s)" in plan and "no saved copy" in plan and "/import" in plan
        assert "4 posts" in plan  # 1, 2, 3 and 6 are copied
        assert "The new posts are read back" in plan
        assert not any(isinstance(r, functions.messages.SendMessageRequest) for r in dst.requests)  # a plan posts nothing

        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "4 post(s) are restricted by Telegram" in done and "copied from what My posts saved of them" in done
        assert "1 post(s): Telegram shows only its notice and My posts has no saved copy of them" in done and "#5" in done
        assert "not everything was copied" in done  # so the rest can be tried again later
        assert any("Try the rest again" in t for t in buttons(app))
        assert "hides" not in done  # nothing the new posts show is restricted
        assert not any(NOTICE in (m.message or "") for m in dst.msgs.values())
        assert forwarded_ids(dst) == []  # a struck channel: nothing is copied by Telegram itself

        # the missing copy arrives (a backup put back with /import): the rest can be copied
        await save_rows(app.db, {5: originals()[5]})
        await app.press(app.out.callback_data("Try the rest again"))
        done = norm(app.out.last_text)
        assert "message(s) copied" in done and "Undo" in " ".join(buttons(app))
        assert dst.msgs[15].message == "Fifth"
        await app.db.close()

    run(go())


def test_a_refused_file_is_named_in_the_result(tmp_path):
    async def go():
        app, dst = await held_app(tmp_path, file_available=False)
        await app.text("/shift @srcch @dstch")
        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "1 of them went out without their media" in done and "#3" in done
        await app.db.close()

    run(go())


def test_the_usage_tells_about_posts_telegram_holds_back(tmp_path):
    async def go():
        app, _ = await shift_app(tmp_path)
        await app.text("/shift")
        text = norm(app.out.last_text)
        assert "copyright infringement" in text and "copied from what My posts saved of it" in text
        await app.db.close()

    run(go())


# ============================================================================================ /repost
def repost_posts():
    return {
        1: make_msg(1, "First post"),
        2: make_msg(2, "Second post", markup=markup([url_btn("Watch", "https://t.me/goku?start=BBB")])),
        3: make_msg(3, "Third post"),
    }


def test_a_repost_never_copies_the_notice_and_never_deletes_what_it_could_not_copy(tmp_path):
    async def go():
        orig = repost_posts()
        app = await make_app(tmp_path, msgs=live_posts(orig, (2,)))
        await save_rows(app.db, orig)
        await app.text("/repost --no-typed")
        await app.press(app.out.callback_data("Test"))
        plan = norm(app.out.last_text)
        assert "Telegram holds back 1 post(s) of Test" in plan and "Copying the notice would be wrong" in plan
        assert "2 posts" in plan  # posts 1 and 3
        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "1 post(s) are held back by Telegram" in done and "#2" in done and "the old posts are not deleted" in done
        assert "not everything was copied" in done
        assert not any("Delete the old posts" in t for t in buttons(app))  # nothing may be deleted
        # the channel: the three posts, the held back one stays as it is, two copies at the end - and no notice copy
        texts = [m.message for _, m in sorted(app.tg.msgs.items())]
        assert texts == ["First post", NOTICE, "Third post", "First post", "Third post"]
        await app.db.close()

    run(go())


def test_a_repost_of_a_channel_without_held_back_posts_is_unchanged(tmp_path):
    async def go():
        app = await make_app(tmp_path, msgs=repost_posts())
        await app.text("/repost --no-typed")
        await app.press(app.out.callback_data("Test"))
        assert "holds back" not in norm(app.out.last_text)
        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "all posts copied in order" in done and "held back" not in done
        assert any("Delete the old posts" in t for t in buttons(app))
        await app.db.close()

    run(go())


# ============================================================================ a source that can't be read at all
def unreadable(chan):
    """The channel is banned or private for the bot: nothing in it can be read any more."""
    chan.fail["GetMessagesRequest"] = rpc("ChannelPrivateError")
    chan.fail["GetFullChannelRequest"] = rpc("ChannelPrivateError")


def test_a_source_that_cannot_be_read_is_copied_from_my_posts_alone(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=())
        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.from_db and res.used_saved and res.failed == [] and res.aborted is None and res.held_back == []
        assert res.from_saved == 5 and res.copied_units == 5
        # copies: 1->11 2->12 3->13 5->14 6->15 (there is nothing saved of the service message 4)
        assert shown(dst) == {11: "Welcome", 12: "Episode 2 is out", 13: "Caption with link", 14: "Fifth", 15: "Sixth"}
        assert forwarded_ids(dst) == []  # nothing is copied by Telegram: the source can't be read
        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 is None  # My posts does not know what a post answered
        assert (await db.find_by_message(2, 13)).media_kind == "photo"
        await db.close()

    run(go())


def test_the_range_is_respected_and_a_stopped_shift_goes_on_from_my_posts(tmp_path):
    async def go():
        src, dst, db, dst_row, shift = await held_setup(tmp_path, held=())
        unreadable(src)
        await db.set_mark("s1", "fromdb")
        calls = []

        async def prog(r):
            calls.append(r.copied_units)

        res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0, progress=prog, should_stop=lambda: len(calls) >= 2)
        assert res.stopped and res.copied_units == 2 and shown(dst) == {11: "Welcome", 12: "Episode 2 is out"}
        again = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0)
        assert again.from_db and not again.stopped and again.copied_units == 3 and again.skipped_done == 2
        assert shown(dst) == {11: "Welcome", 12: "Episode 2 is out", 13: "Caption with link", 14: "Fifth", 15: "Sixth"}
        await db.close()

    run(go())


def test_a_post_with_nothing_to_post_is_not_taken_from_my_posts(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="", 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_the_shift_command_offers_my_posts_when_the_source_can_not_be_read(tmp_path):
    async def go():
        app, dst = await held_app(tmp_path, held=())
        unreadable(app.tg)
        await app.text("/shift @srcch @dstch")
        plan = norm(app.out.last_text)
        assert "I can't read Anime Source" in plan and "My posts has the saved posts of it" in plan
        assert "5 posts" in plan and "Post ids 1-6" in plan
        assert "replies and albums" in plan and "from what My posts saved of it" in plan
        assert dst.msgs.keys() == set(range(1, 11))  # a plan posts nothing

        await app.press(app.out.callback_data("Start copying"))
        done = norm(app.out.last_text)
        assert "5 message(s) copied" in done and "The source can't be read, so 5 post(s) were copied" in done
        sid = app.out.callback_data("Undo").split(":")[1]
        assert await app.db.marks_of(sid) == {"fromdb"}
        assert dst.msgs[12].message == "Episode 2 is out"

        await app.press(app.out.callback_data("Undo"))
        await app.press(app.out.callback_data("Yes, remove"))
        assert "5 copied message(s) removed" in norm(app.out.last_text)
        await app.db.close()

    run(go())


def test_the_range_of_a_shift_from_my_posts(tmp_path):
    async def go():
        app, dst = await held_app(tmp_path, held=())
        unreadable(app.tg)
        await app.text("/shift @srcch @dstch 2 3")
        plan = norm(app.out.last_text)
        assert "Post ids 2-3" in plan and "2 posts" in plan
        await app.press(app.out.callback_data("Start copying"))
        assert [m.message for i, m in sorted(dst.msgs.items()) if i > 10] == ["Episode 2 is out", "Caption with link"]
        await app.db.close()

    run(go())


def test_an_unreadable_source_that_my_posts_does_not_know_is_still_an_error(tmp_path):
    async def go():
        app, dst = await shift_app(tmp_path)  # My posts has nothing of the source
        unreadable(app.tg)
        await app.text("/shift @srcch @dstch")
        text = norm(app.out.last_text)
        assert "I can't read Anime Source" in text and "no saved posts of it" in text
        assert not any(isinstance(r, functions.messages.SendMessageRequest) for r in dst.requests)
        await app.db.close()

    run(go())