Blame for tests/test_restricted.py

"""What TELEGRAM does to a channel (a copyright strike ...) is not an admin's edit: My posts stay exactly as saved."""
from telethon import types

from app.restrictions import PLACEHOLDER_NOTE, is_placeholder_text, message_restriction, restriction_of
from app.sync_engine import MASS_NOTE, diff_post, split_mass_text, sync_channel, sync_ids

from .fakes import markup, url_btn
from .harness import norm
from .test_sync import foreign, looked, old_msg, opts, run, world
from .test_syncer import dms, make_app, three as three_posts

NOTICE = "This message couldn't be displayed on your device due to copyright infringement."
CHANNEL_NOTICE = "This channel is unavailable due to copyright infringement."  # what Telegram says of a struck channel


def strike(m, text=NOTICE):
    """Telegram holds a message back: its own restriction mark on it."""
    m.restriction_reason = [types.RestrictionReason(platform="all", reason="copyright", text=text)]
    return m


def seven():
    return {i: old_msg(i, f"post number {i}", markup=markup([url_btn("Watch", f"https://t.me/a/{i}")])) for i in range(1, 8)}


# ================================================================================================ recognising it
def test_the_notice_telegram_shows_instead_of_a_post_is_recognised():
    for text in (
        NOTICE,
        "This message couldn’t be displayed on your device due to copyright infringement",
        "  This message  couldn't be displayed\non your device due to copyright infringement.  ",
        "This channel can't be displayed because it violated Telegram's Terms of Service.",
        "This post isn't available in your country due to copyright.",
    ):
        assert is_placeholder_text(text), text
    for text in (
        "",
        None,
        "Episode 12 is out now",
        "This video is not available in HD",  # says "not available" but nothing about a restriction
        "Watch: this message couldn't be displayed on your device",  # not the notice itself
        "This message couldn't be displayed on your device due to copyright infringement. " + "x" * 300,  # a long post
    ):
        assert not is_placeholder_text(text), text


def test_telegrams_own_marks_are_read_from_messages_and_channels():
    plain = old_msg(1, "hello")
    assert restriction_of(plain) is None and message_restriction(plain) is None
    held = strike(old_msg(2, "hello"), "Removed after a copyright complaint")
    assert restriction_of(held) == "Removed after a copyright complaint"
    assert message_restriction(held) == "Removed after a copyright complaint"
    two = old_msg(3, "x")
    two.restriction_reason = [
        types.RestrictionReason("ios", "sensitive", "Sensitive content"),
        types.RestrictionReason("android", "sensitive", "Sensitive content"),  # the same words twice: said once
        types.RestrictionReason("all", "copyright", ""),  # no words: its code is used
    ]
    assert restriction_of(two) == "Sensitive content; copyright"
    nameless = old_msg(4, "x")
    nameless.restriction_reason = [types.RestrictionReason("all", "terms", "")]
    assert restriction_of(nameless) == "terms"
    chan = types.Channel(id=1, title="t", photo=types.ChatPhotoEmpty(), date=None, restricted=True)
    assert restriction_of(chan) == "restricted by Telegram"
    assert restriction_of(types.Channel(id=1, title="t", photo=types.ChatPhotoEmpty(), date=None)) is None


def test_a_text_the_saved_post_already_has_is_the_owners_own_wording():
    m = old_msg(1, NOTICE)
    assert message_restriction(m) == PLACEHOLDER_NOTE  # an unknown message: it is the notice
    assert message_restriction(m, "post number 1") == PLACEHOLDER_NOTE  # the saved post said something else
    assert message_restriction(m, NOTICE) is None  # the saved post says exactly that: nothing changed


def test_many_posts_that_suddenly_show_one_text_are_held_back_but_few_are_not(tmp_path):
    async def go():
        chan, db, ch = await world(tmp_path, {i: old_msg(i, f"post number {i}") for i in range(1, 8)})
        rows = {p.message_id: p for p in await db.sent_posts(1)}

        def edits_to(text, ids):
            return [(rows[i], diff_post(rows[i], old_msg(i, text))) for i in ids]

        keep, held = split_mass_text(edits_to("Gone.", range(1, 6)))
        assert keep == [] and len(held) == 5
        keep, held = split_mass_text(edits_to("Gone.", range(1, 5)))  # four are below the line
        assert len(keep) == 4 and held == []
        keep, held = split_mass_text(edits_to("", range(1, 8)))  # all texts removed: a person can do that
        assert len(keep) == 7 and held == []
        keep, held = split_mass_text(edits_to("Gone.", range(1, 6)) + edits_to("Other", [6, 7]))
        assert [p.message_id for p, _ in keep] == [6, 7] and len(held) == 5
        await db.close()

    run(go())


def test_edits_that_all_have_the_same_text_but_the_same_saved_text_are_a_real_edit(tmp_path):
    async def go():
        msgs = {i: old_msg(i, "Join us") for i in range(1, 8)}  # seven posts that said exactly the same
        chan, db, ch = await world(tmp_path, msgs)
        for m in msgs.values():
            m.message = "Join us now"
        rep = await sync_channel(chan, db, ch, opts())
        assert len(rep.edited) == 7 and rep.restricted == []
        await db.close()

    run(go())


# =============================================================================================== the whole channel
def test_a_channel_telegram_restricted_is_left_alone(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        before = await db.get_sync(1)
        chan.restriction = NOTICE
        del chan.msgs[2]
        chan.msgs[1].message = "edited by an admin"
        new = foreign(chan, "a new post")
        chan._pts += 2
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.restricted_channel == NOTICE and rep.error is None
        assert (rep.deleted, rep.edited, rep.adopted, rep.restricted) == ([], [], [], [])
        assert sorted(p.message_id for p in await db.sent_posts(1)) == list(range(1, 8))  # nothing removed
        assert (await db.find_by_message(1, 1)).text == "post number 1"  # nothing edited
        assert await db.find_by_message(1, new.id) is None  # nothing added
        after = await db.get_sync(1)
        assert (after.last_top, after.last_pts) == (before.last_top, before.last_pts)  # the mark did not move

        chan.restriction = None  # the strike is lifted: the next look is the normal one again
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.restricted_channel is None
        assert (rep.deleted, rep.edited, rep.adopted) == ([2], [1], [new.id])
        await db.close()

    run(go())


def test_the_first_look_at_a_restricted_channel_changes_nothing_either(tmp_path):
    async def go():
        chan, db, ch = await world(tmp_path, seven())
        chan.restriction = NOTICE
        for m in chan.msgs.values():
            m.message = NOTICE
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.restricted_channel == NOTICE and rep.first_look is False
        assert await db.get_sync(1) is None
        assert [p.text for p in await db.sent_posts(1)] == [f"post number {i}" for i in range(1, 8)]
        await db.close()

    run(go())


# ============================================================================================ a copyright strike
def test_every_post_turning_into_the_notice_changes_nothing_in_my_posts(tmp_path):
    """The case that was asked about: Telegram swaps the text of EVERY post for its notice."""

    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        for m in chan.msgs.values():
            m.message, m.entities, m.reply_markup = NOTICE, None, None  # what the strike leaves in the channel
        chan._pts += 7
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.edited == [] and rep.deleted == [] and rep.buttons_missing == []
        assert sorted(rep.restricted) == list(range(1, 8)) and rep.restricted_why == PLACEHOLDER_NOTE
        for i, p in enumerate(await db.sent_posts(1), 1):
            assert p.text == f"post number {i}" and p.buttons == [[{"t": "Watch", "u": f"https://t.me/a/{i}"}]]
        await db.close()

    run(go())


def test_posts_carrying_telegrams_restriction_mark_are_left_as_saved(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        for i in (2, 3):
            strike(chan.msgs[i])
            chan.msgs[i].message = "(something else)"
        chan.msgs[4].message = "a real edit by an admin"
        chan._pts += 3
        rep = await sync_channel(chan, db, ch, opts())
        assert sorted(rep.restricted) == [2, 3] and rep.restricted_why == NOTICE
        assert rep.edited == [4]  # everything else is looked at as usual
        assert (await db.find_by_message(1, 2)).text == "post number 2"
        assert (await db.find_by_message(1, 4)).text == "a real edit by an admin"
        await db.close()

    run(go())


def test_a_post_that_is_really_gone_is_still_removed_while_others_are_held_back(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        strike(chan.msgs[2])
        chan.msgs[2].message = NOTICE
        del chan.msgs[6]
        chan._pts += 2
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.deleted == [6] and rep.restricted == [2]
        assert (await db.find_by_message(1, 2)).text == "post number 2"
        await db.close()

    run(go())


def test_a_notice_does_not_stop_a_normal_edit_of_another_post(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        chan.msgs[1].message = NOTICE
        chan.msgs[5].message = "the admin fixed a typo"
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.restricted == [1] and rep.edited == [5]
        await db.close()

    run(go())


def test_five_posts_that_all_show_the_same_unknown_text_are_held_back(tmp_path):
    """The notice in another language: it is not in any list, but nobody edits five posts into one text."""

    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        for i in range(1, 6):
            chan.msgs[i].message = "Este mensaje no se puede mostrar en tu dispositivo."
        rep = await sync_channel(chan, db, ch, opts())
        assert sorted(rep.restricted) == [1, 2, 3, 4, 5] and rep.edited == [] and rep.restricted_why == MASS_NOTE
        assert (await db.find_by_message(1, 1)).text == "post number 1"
        await db.close()

    run(go())


def test_four_posts_with_the_same_new_text_are_ordinary_edits(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        for i in range(1, 5):
            chan.msgs[i].message = "Moved to the new channel."
        rep = await sync_channel(chan, db, ch, opts())
        assert sorted(rep.edited) == [1, 2, 3, 4] and rep.restricted == []
        await db.close()

    run(go())


# ================================================================================================= new posts
def test_new_posts_that_telegram_holds_back_are_not_added(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, three_posts_dict())
        a = foreign(chan, "a normal new post")
        b = strike(foreign(chan, "hidden by Telegram"))
        c = foreign(chan, NOTICE)
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.adopted == [a.id] and sorted(rep.restricted) == [b.id, c.id]
        assert await db.find_by_message(1, b.id) is None and await db.find_by_message(1, c.id) is None
        again = await sync_channel(chan, db, ch, opts())  # looked at once, not over and over
        assert again.restricted == [] and again.adopted == []
        await db.close()

    run(go())


def three_posts_dict():
    return {m.id: m for m in three_posts()}


# =============================================================================================== the live look
def test_the_live_look_leaves_a_restricted_post_alone(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        strike(chan.msgs[3])
        chan.msgs[3].message = NOTICE
        chan.msgs[4].message = "a real edit"
        rep = await sync_ids(chan, db, ch, [3, 4], opts())
        assert rep.restricted == [3] and rep.edited == [4]
        assert (await db.find_by_message(1, 3)).text == "post number 3"
        await db.close()

    run(go())


def test_the_live_look_holds_back_a_burst_of_identical_texts(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        for i in range(1, 6):
            chan.msgs[i].message = "Blocked."
        rep = await sync_ids(chan, db, ch, [1, 2, 3, 4, 5], opts())
        assert sorted(rep.restricted) == [1, 2, 3, 4, 5] and rep.edited == []
        await db.close()

    run(go())


def test_the_live_look_does_not_add_a_restricted_new_post(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        m = foreign(chan, NOTICE)
        rep = await sync_ids(chan, db, ch, [m.id], opts(), adopt_ids=[m.id])
        assert rep.adopted == [] and rep.restricted == [m.id]
        await db.close()

    run(go())


# ============================================================================== what "everything looks deleted" means
def test_posts_really_deleted_by_an_admin_all_go_even_when_that_is_everything(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        chan.msgs.clear()
        chan._pts += 7  # seven deletions are seven events
        rep = await sync_channel(chan, db, ch, opts())
        assert sorted(rep.deleted) == list(range(1, 8)) and rep.anomaly is None
        assert await db.sent_posts(1) == []
        await db.close()

    run(go())


def test_posts_that_merely_vanish_from_view_are_not_removed_without_the_events(tmp_path):
    async def go():
        chan, db, ch = await looked(tmp_path, seven())
        chan.msgs.clear()  # nothing happened in the channel: the bot just sees nothing
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.deleted == [] and rep.anomaly and "All 7" in rep.anomaly
        chan._pts += 3  # three events are not enough to explain seven missing posts
        rep = await sync_channel(chan, db, ch, opts())
        assert rep.deleted == [] and rep.anomaly
        assert len(await db.sent_posts(1)) == 7
        await db.close()

    run(go())


# ================================================================================================= the owners
def test_sync_tells_what_it_left_alone_in_a_restricted_channel(tmp_path):
    async def go():
        app = await make_app(tmp_path, posts=three_posts())
        await app.text("/sync")  # the first look
        app.tg.restriction = NOTICE
        app.tg.msgs[2].message = NOTICE
        await app.text("/sync")
        text = norm(app.out.last_text)
        assert "Telegram has restricted this channel" in text and "copyright infringement" in text
        assert "stay exactly as they are" in text and "Everything matches" not in text
        assert (await app.db.find_by_message(1, 2)).text == "second"
        await app.db.close()

    run(go())


def test_sync_tells_which_posts_it_held_back(tmp_path):
    async def go():
        app = await make_app(tmp_path, posts=three_posts())
        await app.text("/sync")
        strike(app.tg.msgs[2])
        app.tg.msgs[2].message = "(hidden)"
        await app.text("/sync")
        text = norm(app.out.last_text)
        assert "1 post(s) are held back by Telegram" in text and "left exactly as saved" in text and "message 2" in text
        assert "Everything matches" not in text
        await app.db.close()

    run(go())


def test_the_owners_are_told_about_a_restriction_once_not_every_check(tmp_path):
    async def go():
        app = await make_app(tmp_path, posts=three_posts())
        sy = app.ctx.syncer
        app.tg.restriction = NOTICE
        await sy.run_all()
        await sy.run_all()
        await sy.run_all()
        told = [t for t in dms(app) if "restricted" in t]
        assert len(told) == 1 and "Anime Channel" in told[0] and "stay exactly as they are" in told[0]
        sy._told.clear()  # a week later
        await sy.run_all()
        assert len([t for t in dms(app) if "restricted" in t]) == 2
        await app.db.close()

    run(go())


def test_the_owners_are_told_about_held_back_posts_too(tmp_path):
    async def go():
        app = await make_app(tmp_path, posts=three_posts())
        sy = app.ctx.syncer
        await sy.run_all()
        strike(app.tg.msgs[3])
        await sy.run_all()
        await sy.run_all()
        told = [t for t in dms(app) if "holds back" in t]
        assert len(told) == 1 and "1 post(s)" in told[0]
        await app.db.close()

    run(go())