Blame for tests/test_dbflow.py

"""Uploading a manually renamed episode to the Database Channel and having the channel post bot publish it:
the two chats in the settings, the Channel option of a manual format, the offer, the upload, the notice in the
communication chat and the answers of the channel post bot."""
import asyncio
import re
import time
from types import SimpleNamespace

import pytest
from telethon import errors

from bot import panels
from bot.app import App
from bot.config import ConfigError, load_config
from bot.db import Database, Format
from bot.dbflow import classify, episode_of, line_of, notice_text, reply_target
from bot.formats import default_auto_format
from bot.pipeline import Renamer
from bot.queue import Job
from bot.settings_ops import parse_format_channel

from .appkit import ADMIN_CFG, OWNER, STRANGER, click, data_of, env_for, say
from .tgfakes import FakeClient, FakeEvent, dispatch, document_message, flat_buttons

DB = -1002000000001  # the Database Channel
COMM = -1003000000002  # the communication chat
POST_BOT = 8950016484
SHOW = -1004000000003  # the channel a format publishes in
NAME = "[S01E05] Tokyo Revengers [1080p].mkv"


# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------


@pytest.fixture
async def up(tmp_path):
    """Like ``ctx``, with uploading to the Database Channel switched on (and short waits)."""
    cfg = load_config(env_for(tmp_path, DB_Channel=str(DB), Communication_chat=str(COMM)))
    db = Database(f"sqlite+aiosqlite:///{tmp_path}/app.db")
    await db.init()
    client = FakeClient()
    app = App(client, cfg, db)
    app.dbflow.ack_wait, app.dbflow.result_wait = 2.0, 4.0
    app._register()
    yield SimpleNamespace(app=app, client=client, db=db, cfg=cfg, tmp=tmp_path, flow=app.dbflow)
    await app.shutdown()
    await db.close()


def notices(up):
    """What was written in the communication chat: ``[(message id, text, kwargs)]``."""
    return [(2000 + n, text, kw) for n, (entity, text, kw) in enumerate(up.client.sent, 1) if entity == COMM]


async def wait_for(predicate, timeout=3.0):
    end = time.monotonic() + timeout
    while not predicate():
        assert time.monotonic() < end, "timed out"
        await asyncio.sleep(0.005)


async def the_notice(up, nth=1):
    await wait_for(lambda: len(notices(up)) >= nth)
    return notices(up)[nth - 1]


def ref_of(text):
    return re.search(r"(?m)^Ref: (\w+)$", text).group(1)


def last_edit(up):
    return up.client.edited[-1][1]


async def post_bot_says(up, text, *, reply_to=None, sender=POST_BOT, chat=COMM):
    ev = FakeEvent(sender, text, chat=chat, private=False, reply_to_id=reply_to)
    await dispatch(up.client, ev)
    return ev


async def start_upload(up, *, channel=SHOW, name=NAME, mid=555, offer_id=77, caption=""):
    """The renamed episode is in the user's chat, and the offer is answered with Yes."""
    up.client.messages[mid] = document_message(name, 10 * 1024 * 1024, mid=mid, caption=caption)
    return await click(up, OWNER, f"dbu:y:{OWNER}:{mid}:{channel or 0}", msg_id=offer_id)


def manual_job(name=NAME, channel=SHOW, kind="manual", chat=OWNER):
    fmt = {"id": 5, "name": "One Piece", "kind": kind, "channel": channel}
    return Job(
        id=1, user_id=OWNER, chat_id=chat, msg_id=40, file_name="original.mkv", size=1, new_name=name, fmt=fmt,
        status_msg=SimpleNamespace(id=99),
    )


# ---------------------------------------------------------------------------
# settings
# ---------------------------------------------------------------------------


def test_both_chats_switch_uploading_on(tmp_path):
    cfg = load_config(env_for(tmp_path, DB_CHANNEL=str(DB), COMMUNICATION_CHAT=str(COMM)))
    assert (cfg.db_channel, cfg.comm_chat, cfg.upload_enabled, cfg.post_bot_id) == (DB, COMM, True, 8950016484)


@pytest.mark.parametrize("db_name, comm_name", [("DB_CHANNEL", "COMMUNICATION_CHAT"), ("db_channel", "communication_chat"), ("DB_Channel", "Communication_chat")])
def test_the_names_may_be_written_in_any_case(tmp_path, db_name, comm_name):
    cfg = load_config(env_for(tmp_path, **{db_name: str(DB), comm_name: str(COMM)}))
    assert (cfg.db_channel, cfg.comm_chat, cfg.upload_enabled) == (DB, COMM, True)


@pytest.mark.parametrize("extra", [{"DB_CHANNEL": str(DB)}, {"COMMUNICATION_CHAT": str(COMM)}, {}])
def test_one_chat_alone_does_not_switch_it_on(tmp_path, extra):
    assert load_config(env_for(tmp_path, **extra)).upload_enabled is False


def test_a_bare_channel_id_gets_its_minus_100(tmp_path):
    cfg = load_config(env_for(tmp_path, DB_CHANNEL="2000000001", COMMUNICATION_CHAT="3000000002"))
    assert (cfg.db_channel, cfg.comm_chat) == (-1002000000001, -1003000000002)


@pytest.mark.parametrize("name", ["DB_CHANNEL", "COMMUNICATION_CHAT", "CHANNEL_POST_BOT_ID"])
def test_an_id_that_is_not_a_number_is_a_configuration_error(tmp_path, name):
    with pytest.raises(ConfigError, match=name):
        load_config(env_for(tmp_path, **{name: "@somewhere"}))


@pytest.mark.parametrize("value", ["", "0", "off", "OFF", "none"])
def test_empty_or_off_means_not_set(tmp_path, value):
    cfg = load_config(env_for(tmp_path, DB_CHANNEL=value, COMMUNICATION_CHAT=value))
    assert cfg.db_channel is None and cfg.comm_chat is None and cfg.upload_enabled is False


def test_the_post_bot_id_can_be_changed(tmp_path):
    assert load_config(env_for(tmp_path, CHANNEL_POST_BOT_ID="12345")).post_bot_id == 12345


async def test_without_both_chats_nothing_is_offered_and_the_buttons_say_so(ctx):
    assert ctx.app.dbflow is None and ctx.app.renamer.on_delivered is None
    cb = await click(ctx, OWNER, f"dbu:y:{OWNER}:555:{SHOW}")
    assert cb.alerts and "switched off" in cb.alerts[0] and not ctx.client.forwarded and not ctx.client.sent


async def test_with_both_chats_the_offer_is_hooked_into_the_pipeline(up):
    assert up.app.renamer.on_delivered == up.flow.offer


# ---------------------------------------------------------------------------
# the Channel option of a manual format
# ---------------------------------------------------------------------------


async def test_a_manual_format_keeps_its_channel(ctx):
    fmt = await ctx.db.create_format(OWNER, "One Piece")
    assert fmt["channel"] is None
    assert (await ctx.db.update_format(OWNER, fmt["id"], channel=SHOW))["channel"] == SHOW
    assert (await ctx.db.get_format(OWNER, fmt["id"]))["channel"] == SHOW
    assert [f["channel"] for f in await ctx.db.list_formats(OWNER, "manual")] == [SHOW]
    assert (await ctx.db.update_format(OWNER, fmt["id"], channel=None))["channel"] is None


async def test_an_auto_format_has_no_channel(ctx):
    fmt = await ctx.db.create_format(OWNER, "Auto One", kind="auto")
    assert (await ctx.db.update_format(OWNER, fmt["id"], channel=SHOW))["channel"] is None


async def test_formats_saved_before_the_option_have_no_channel(ctx):
    async with ctx.db._sessions() as session:
        session.add(Format(user_id=OWNER, name="Old", data='{"template": "{title}", "kind": "manual"}'))
        session.add(Format(user_id=OWNER, name="Odd", data='{"kind": "manual", "channel": "-100123"}'))  # not a number: ignored
        await session.commit()
    assert [f["channel"] for f in await ctx.db.list_formats(OWNER)] == [None, None]


async def test_a_copy_keeps_the_channel_only_for_the_owners_own_manual_copy(ctx):
    fmt = await ctx.db.create_format(OWNER, "One Piece")
    await ctx.db.update_format(OWNER, fmt["id"], channel=SHOW)
    auto = await ctx.db.copy_format(OWNER, OWNER, fmt["id"], kind="auto")
    assert auto["kind"] == "auto" and auto["channel"] is None
    again = await ctx.db.copy_format(OWNER, OWNER, fmt["id"])
    assert again["kind"] == "manual" and again["channel"] == SHOW
    theirs = await ctx.db.copy_format(ADMIN_CFG, OWNER, fmt["id"])  # somebody else adds it from a share card
    assert theirs["kind"] == "manual" and theirs["channel"] is None
    back = await ctx.db.copy_format(OWNER, OWNER, auto["id"], kind="manual")
    assert back["channel"] is None
    assert (await ctx.db.get_format(OWNER, fmt["id"]))["channel"] == SHOW  # the original is untouched


@pytest.mark.parametrize(
    "text, expected",
    [
        ("-1001234567890", -1001234567890),
        ("  -1001234567890 ", -1001234567890),
        ("1234567890", -1001234567890),
        ("https://t.me/c/1234567890/5", -1001234567890),
        ("t.me/c/1234567890", -1001234567890),
        ("off", None), ("None", None), ("-", None), ("clear", None),
    ],
)
def test_a_channel_id_is_read(text, expected):
    assert parse_format_channel(text) == (True, expected, "")


@pytest.mark.parametrize("text", ["", "abc", "@mychannel", "https://t.me/mychannel", "-12345", "123", "-1001234567890x", "12 34567890"])
def test_anything_else_is_refused_with_the_way_to_write_it(text):
    ok, value, message = parse_format_channel(text)
    assert not ok and value is None and "-1001234567890" in message


async def test_without_the_two_chats_the_format_has_no_channel_option(ctx):
    fmt = await ctx.db.create_format(OWNER, "One Piece")
    text, buttons = (await click(ctx, OWNER, f"fm:v:{fmt['id']}")).edits[-1]
    assert "Channel" not in text and not [d for d in data_of(buttons) if "channel" in d]
    cb = await click(ctx, OWNER, f"fm:in:{fmt['id']}:channel")
    assert cb.alerts and OWNER not in ctx.app.pending


async def test_a_manual_format_shows_the_channel_option_and_an_auto_format_does_not(up):
    fmt = await up.db.create_format(OWNER, "One Piece")
    i = fmt["id"]
    text, buttons = (await click(up, OWNER, f"fm:v:{i}")).edits[-1]
    assert "<b>Channel</b> ▸ — (optional)" in text
    assert f"fm:in:{i}:channel" in data_of(buttons) and f"fm:chc:{i}" not in data_of(buttons)
    auto = await up.db.create_format(OWNER, "Auto One", kind="auto")
    text, buttons = (await click(up, OWNER, f"fm:v:{auto['id']}")).edits[-1]
    assert "Channel" not in text and f"fm:in:{auto['id']}:channel" not in data_of(buttons)
    cb = await click(up, OWNER, f"fm:in:{auto['id']}:channel")
    assert cb.alerts and OWNER not in up.app.pending


async def test_the_channel_is_typed_in_checked_and_cleared(up):
    fmt = await up.db.create_format(OWNER, "One Piece")
    i = fmt["id"]
    cb = await click(up, OWNER, f"fm:in:{i}:channel", msg_id=710)
    assert cb.edits[-1][0] == panels.FORMAT_PROMPTS["channel"] and up.app.pending[OWNER].kind == "fmt:channel"
    assert data_of(cb.edits[-1][1]) == [f"fm:cx:{i}"]
    ev = await say(up, OWNER, "@not_an_id")
    assert "❌" in ev.texts[0] and "Try again" in ev.texts[0] and OWNER in up.app.pending
    ev = await say(up, OWNER, "1234567890")  # a bare channel id
    assert ev.deleted and OWNER not in up.app.pending
    assert (await up.db.get_format(OWNER, i))["channel"] == -1001234567890
    chat, text, kw = up.client.edited[-1]
    assert kw["message"] == 710 and "Channel saved" in text and "<code>-1001234567890</code>" in text
    text, buttons = (await click(up, OWNER, f"fm:v:{i}")).edits[-1]
    assert "<b>Channel</b> ▸ <code>-1001234567890</code>" in text and f"fm:chc:{i}" in data_of(buttons)
    cb = await click(up, OWNER, f"fm:chc:{i}")  # the button clears it
    assert (await up.db.get_format(OWNER, i))["channel"] is None and "— (optional)" in cb.edits[-1][0]
    await up.db.update_format(OWNER, i, channel=SHOW)
    await click(up, OWNER, f"fm:in:{i}:channel", msg_id=711)
    await say(up, OWNER, "off")  # and so does typing off
    assert (await up.db.get_format(OWNER, i))["channel"] is None


async def test_the_channel_of_a_deleted_format_is_handled(up):
    fmt = await up.db.create_format(OWNER, "One Piece")
    await click(up, OWNER, f"fm:in:{fmt['id']}:channel")
    await up.db.delete_format(OWNER, fmt["id"])
    ev = await say(up, OWNER, "-1001234567890")
    assert "no longer exists" in ev.texts[0] and OWNER not in up.app.pending


async def test_the_format_screen_with_a_channel_still_fits_a_caption(up):
    fmt = await up.db.create_format(OWNER, "N" * 30, template="T" * 200)
    await up.db.update_format(OWNER, fmt["id"], channel=-1009999999999, caption="C" * 700)
    text, _rows = panels.format_view(await up.db.get_format(OWNER, fmt["id"]), upload_on=True)
    assert len(re.sub(r"<[^>]+>", "", text)) < 1024


def test_a_shared_format_card_does_not_show_the_channel():
    fmt = {**default_auto_format(), "kind": "manual", "name": "One Piece", "id": 5, "channel": SHOW}
    text, _rows = panels.shared_format(fmt, "token")
    assert str(SHOW) not in text and "Channel" not in text


# ---------------------------------------------------------------------------
# the messages
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "name, episode",
    [
        (NAME, "05"), ("One Piece - E1147.mkv", "1147"), ("Black Clover - S2 E05.mkv", "05"),
        ("Tokyo Revengers Movie.mkv", ""), ("", ""),
    ],
)
def test_the_episode_number_is_read_from_the_renamed_file(name, episode):
    assert episode_of(name) == episode


def test_the_notice_has_what_the_post_bot_needs():
    text = notice_text(episode="05", channel=SHOW, file=NAME, post_id=4001, ref="ab12cd")
    assert text.splitlines() == [
        "File uploaded to the Database Channel", "Episode: 05", f"Channel: {SHOW}", f"File: {NAME}", "Post: 4001", "Ref: ab12cd"
    ]
    bare = notice_text(episode="", channel=SHOW, file=NAME, post_id=None, ref="ab12cd")
    assert "Episode:" not in bare and "Post:" not in bare and f"Channel: {SHOW}" in bare


def test_the_label_lines_of_a_message_are_found():
    text = "✅ Post published successfully\nChannel: Black Clover English\nPost: https://t.me/c/1/2\nRef: ab12cd"
    assert line_of("channel", text) == "Black Clover English" and line_of("post", text) == "https://t.me/c/1/2"
    assert line_of("ref", text) == "ab12cd" and line_of("episode", text) == "" and line_of("ref", None) == ""


@pytest.mark.parametrize(
    "text, expected",
    [
        ("📥 Message received - checking the Database Channel…\nRef: ab12", ("ack", "")),
        ("message received", ("ack", "")),
        ("✅ Post published successfully\nChannel: X", ("ok", "")),
        ("POST PUBLISHED SUCCESSFULLY", ("ok", "")),
        ("❌ Publication failed: no button found\nFile: a.mkv\nRef: ab12", ("fail", "no button found")),
        ("publication failed:   many   spaces  here", ("fail", "many spaces here")),
        ("Publication failed", ("fail", "")),
        ("❌ Publication failed: no post published successfully", ("fail", "no post published successfully")),  # a failure wins
        ("hello", ("other", "")), ("", ("other", "")), (None, ("other", "")),
    ],
)
def test_what_the_post_bot_says_is_told_apart(text, expected):
    assert classify(text) == expected


def test_the_reply_target_is_read_from_the_message():
    assert reply_target(FakeEvent(POST_BOT, "x", reply_to_id=7)) == 7
    assert reply_target(FakeEvent(POST_BOT, "x")) is None


# ---------------------------------------------------------------------------
# the offer after /rename
# ---------------------------------------------------------------------------


async def test_the_offer_asks_yes_or_no_after_the_episode(up):
    await up.app.dbflow.offer(manual_job(), OWNER, SimpleNamespace(id=555))
    (entity, text, kw), = up.client.sent
    assert entity == OWNER and kw["reply_to"] == 555 and kw["parse_mode"] == "html"
    assert "Upload this episode to the Database Channel?" in text and NAME in text
    assert "Episode: <b>05</b>" in text and f"<code>{SHOW}</code>" in text
    assert flat_buttons(kw["buttons"]) == [("✅ Yes", f"dbu:y:{OWNER}:555:{SHOW}"), ("❌ No", "dbu:n")]


def test_the_buttons_fit_telegrams_limit_even_for_the_biggest_ids():
    data = f"dbu:y:{-1009999999999}:{2147483647}:{-1009999999999}"
    assert len(data.encode()) <= 64


async def test_the_offer_for_a_format_without_a_channel_says_it_is_only_uploaded(up):
    await up.app.dbflow.offer(manual_job(channel=None), OWNER, SimpleNamespace(id=555))
    (_entity, text, kw), = up.client.sent
    assert "has no Channel" in text and "no post is made" in text and "Episode:" not in text
    assert data_of(kw["buttons"])[0] == f"dbu:y:{OWNER}:555:0"


async def test_the_offer_warns_when_the_file_name_has_no_episode_number(up):
    await up.app.dbflow.offer(manual_job(name="Tokyo Revengers Movie.mkv"), OWNER, SimpleNamespace(id=555))
    assert "No episode number found" in up.client.sent[0][1]


async def test_a_file_sent_to_a_channel_is_still_offered_in_the_chat_of_the_user(up):
    await up.app.dbflow.offer(manual_job(), SHOW, SimpleNamespace(id=555))
    (entity, _text, kw), = up.client.sent
    assert entity == OWNER and kw["reply_to"] == 40  # the file the user sent, not the one in the channel
    assert f"dbu:y:{SHOW}:555:{SHOW}" in data_of(kw["buttons"])


async def test_nothing_is_offered_without_a_message_to_point_at(up):
    await up.app.dbflow.offer(manual_job(), OWNER, None)
    assert not up.client.sent


class Reporter:
    def __init__(self, log):
        self.log = log

    async def finish(self, text):
        self.log.append("finished")


async def deliver(up, job, hook, target_channel=None, log=None):
    """Run the delivery step of a rename; ``log`` says in what order the file, the Done message and the hook came."""
    log = [] if log is None else log
    renamer = Renamer(up.client, up.cfg, up.db)
    renamer.on_delivered = hook
    if target_channel:
        await up.db.update_settings(OWNER, target_channel=target_channel)
    original_send = up.client.send_file

    async def send_file(*args, **kw):
        log.append("sent")
        return await original_send(*args, **kw)

    up.client.send_file = send_file
    await renamer._deliver(
        job, await up.db.get_settings(OWNER), document_message(NAME, 10), "input", caption=None, attrs=[], as_document=True,
        thumb=None, final_size=10, warnings=[], rates="", started=time.monotonic(), reporter=Reporter(log),
    )
    return log


async def test_a_manual_rename_is_offered_after_the_file_and_the_done_message(up):
    log, calls = [], []

    async def hook(job, target, sent):
        log.append("offer")
        calls.append((job.id, target, sent.id))

    job = manual_job()
    await deliver(up, job, hook, log=log)
    assert job.state == "done" and calls == [(1, OWNER, 3001)] and log == ["sent", "finished", "offer"]


@pytest.mark.parametrize("kind", ["auto", None])
async def test_auto_renames_are_not_offered(up, kind):
    called = []

    async def hook(*args):
        called.append(args)

    job = manual_job(kind=kind or "auto")
    if kind is None:
        job.fmt = None  # an auto rename of someone who has no auto format
    await deliver(up, job, hook)
    assert called == [] and job.state == "done"


async def test_a_failing_offer_does_not_spoil_the_rename(up):
    async def hook(*args):
        raise RuntimeError("telegram said no")

    job = manual_job()
    log = await deliver(up, job, hook)
    assert job.state == "done" and log == ["sent", "finished"]


async def test_without_the_hook_nothing_happens(up):
    job = manual_job()
    assert await deliver(up, job, None) == ["sent", "finished"] and job.state == "done"


async def test_a_file_sent_to_the_target_channel_is_offered_with_that_channel_as_its_chat(up):
    seen = []

    async def hook(job, target, sent):
        seen.append(target)

    await deliver(up, manual_job(), hook, target_channel=SHOW)
    assert seen == [SHOW]


# ---------------------------------------------------------------------------
# Yes: upload, tell the communication chat, read the answers
# ---------------------------------------------------------------------------


async def test_yes_copies_the_episode_and_tells_the_communication_chat(up):
    cb = await start_upload(up, caption="<b>x</b>")
    assert cb.edits[-1][0].startswith("⏳ <b>Processing…</b>") and cb.edits[-1][1] is None and cb.answers
    nid, text, kw = await the_notice(up)
    # copied inside Telegram: nothing was downloaded, nothing was uploaded again
    assert up.client.forwarded == [(DB, 555, OWNER, {"drop_author": True})]
    assert up.client.downloads == [] and up.client.files == []
    ref = ref_of(text)
    assert text.splitlines() == [
        "File uploaded to the Database Channel", "Episode: 05", f"Channel: {SHOW}", f"File: {NAME}", "Post: 4001", f"Ref: {ref}"
    ]
    assert kw["parse_mode"] is None  # plain text: nothing in a file name can be taken for formatting


async def test_the_whole_story_from_yes_to_published(up):
    await start_upload(up)
    nid, text, _ = await the_notice(up)
    ref = ref_of(text)
    await post_bot_says(up, f"📥 Message received - checking the Database Channel…\nRef: {ref}", reply_to=nid)
    await wait_for(lambda: len(up.client.edited) >= 2)
    assert "channel post bot got it" in last_edit(up) and "Waiting" not in last_edit(up)
    done = (
        "✅ Post published successfully\nChannel: Black Clover English\nPost: https://t.me/c/1/2\n"
        f"Button: Download Episode 05 → https://t.me/x\nEpisode: 05\nFile: {NAME}\nRef: {ref}"
    )
    await post_bot_says(up, done, reply_to=nid)
    await up.flow.idle()
    texts = [t for _e, t, _k in up.client.edited]
    assert len(texts) == 3
    assert "Waiting for the channel post bot" in texts[0] and "Uploaded to the Database Channel" in texts[0]
    assert "Post published successfully" in texts[2] and "Black Clover English" in texts[2] and "https://t.me/c/1/2" in texts[2]
    assert NAME in texts[2]
    assert {kw["message"] for _e, _t, kw in up.client.edited} == {77} and {e for e, _t, _k in up.client.edited} == {OWNER}
    assert all(kw["buttons"] is None for _e, _t, kw in up.client.edited)
    assert up.flow.flows == {} and up.flow._busy == set()


async def test_a_failure_of_the_post_bot_is_shown_with_its_reason(up):
    await start_upload(up)
    nid, text, _ = await the_notice(up)
    await post_bot_says(up, f"📥 Message received - checking the Database Channel…\nRef: {ref_of(text)}", reply_to=nid)
    reason = "The latest post of the Database Channel is the file “Other.mkv”, not the one that was reported - nothing was published."
    await post_bot_says(up, f"❌ Publication failed: {reason}\nFile: {NAME}\nRef: {ref_of(text)}", reply_to=nid)
    await up.flow.idle()
    final = last_edit(up)
    assert final.startswith("❌ <b>Publication failed</b>") and reason in final and NAME in final


async def test_names_and_reasons_are_escaped_in_the_edits_but_not_in_the_notice(up):
    await start_upload(up, name="[S01E05] Tom & Jerry <HD>.mkv")
    nid, text, _ = await the_notice(up)
    assert "File: [S01E05] Tom & Jerry <HD>.mkv" in text
    await post_bot_says(up, "❌ Publication failed: bad <b>thing</b> & more\nRef: " + ref_of(text), reply_to=nid)
    await up.flow.idle()
    final = last_edit(up)
    assert "Tom &amp; Jerry &lt;HD&gt;" in final and "bad &lt;b&gt;thing&lt;/b&gt; &amp; more" in final


async def test_no_answer_at_all_means_the_channel_post_bot_is_down(up):
    up.flow.ack_wait = 0.1
    await start_upload(up)
    await up.flow.idle()
    final = last_edit(up)
    assert "down or not responding" in final and "The episode is in the Database Channel" in final and "nothing was published" in final
    assert up.flow.flows == {} and up.flow._busy == set()


async def test_an_acknowledged_notice_that_is_never_answered_is_not_called_down(up):
    up.flow.ack_wait, up.flow.result_wait = 2.0, 0.15
    await start_upload(up)
    nid, text, _ = await the_notice(up)
    await post_bot_says(up, f"📥 Message received - checking the Database Channel…\nRef: {ref_of(text)}", reply_to=nid)
    await up.flow.idle()
    final = last_edit(up)
    assert "No word from the channel post bot" in final and "down" not in final


async def test_an_answer_without_the_acknowledgement_still_counts(up):
    await start_upload(up)
    nid, text, _ = await the_notice(up)
    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(text)}", reply_to=nid)
    await up.flow.idle()
    assert "Post published successfully" in last_edit(up)


async def test_answers_are_matched_by_the_ref_alone_or_by_the_reply_alone(up):
    await start_upload(up, offer_id=71, mid=601)
    nid, text, _ = await the_notice(up)
    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(text)}")  # not a reply: the Ref says whose it is
    await up.flow.idle()
    assert "Post published successfully" in last_edit(up)

    await start_upload(up, offer_id=72, mid=602)
    nid, text, _ = await the_notice(up, 2)
    await post_bot_says(up, "❌ Publication failed: nope", reply_to=nid)  # no Ref: the reply says whose it is
    await up.flow.idle()
    assert "Publication failed" in last_edit(up) and "nope" in last_edit(up)


async def test_with_two_uploads_waiting_the_ref_tells_whose_answer_it_is(up):
    up.client.messages[601] = document_message("[S01E05] A [1080p].mkv", 10, mid=601)
    up.client.messages[602] = document_message("[S01E06] A [1080p].mkv", 10, mid=602)
    await click(up, OWNER, f"dbu:y:{OWNER}:601:{SHOW}", msg_id=71)
    await click(up, OWNER, f"dbu:y:{OWNER}:602:{SHOW}", msg_id=72)
    first, second = await the_notice(up, 1), await the_notice(up, 2)
    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(second[1])}")  # not a reply, and two are waiting
    await wait_for(lambda: any(kw["message"] == 72 and "published successfully" in t for _e, t, kw in up.client.edited))
    assert not any(kw["message"] == 71 and "published successfully" in t for _e, t, kw in up.client.edited)
    assert len(up.flow.flows) == 1 and ref_of(first[1]) in up.flow.flows  # the first is still waiting
    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(first[1])}")
    await up.flow.idle()


async def test_a_plain_message_counts_when_only_one_upload_is_waiting(up):
    await start_upload(up)
    await the_notice(up)
    await post_bot_says(up, "✅ Post published successfully")
    await up.flow.idle()
    assert "Post published successfully" in last_edit(up)


async def test_other_senders_other_chats_other_replies_and_other_words_are_ignored(up):
    await start_upload(up)
    nid, text, _ = await the_notice(up)
    ref = ref_of(text)
    ok = f"✅ Post published successfully\nRef: {ref}"
    await post_bot_says(up, ok, reply_to=nid, sender=STRANGER)  # somebody else
    await post_bot_says(up, ok, reply_to=nid, chat=-1005000000005)  # another chat
    await post_bot_says(up, f"hello there\nRef: {ref}", reply_to=nid)  # not an answer
    await post_bot_says(up, "✅ Post published successfully", reply_to=1)  # a reply to something else
    await asyncio.sleep(0.05)
    assert up.flow.flows[ref].inbox.empty() and len(up.client.edited) == 1
    await post_bot_says(up, ok, reply_to=nid)
    await up.flow.idle()
    assert "Post published successfully" in last_edit(up)


async def test_two_uploads_at_once_each_get_their_own_answer(up):
    up.client.messages[601] = document_message("[S01E05] A [1080p].mkv", 10, mid=601)
    up.client.messages[602] = document_message("[S01E06] A [1080p].mkv", 10, mid=602)
    await click(up, OWNER, f"dbu:y:{OWNER}:601:{SHOW}", msg_id=71)
    await click(up, OWNER, f"dbu:y:{OWNER}:602:{SHOW}", msg_id=72)
    first, second = await the_notice(up, 1), await the_notice(up, 2)
    assert "Episode: 05" in first[1] and "Episode: 06" in second[1] and ref_of(first[1]) != ref_of(second[1])
    await post_bot_says(up, f"❌ Publication failed: nope\nRef: {ref_of(second[1])}", reply_to=second[0])  # out of order
    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(first[1])}", reply_to=first[0])
    await up.flow.idle()
    last = {kw["message"]: text for _e, text, kw in up.client.edited}
    assert "Post published successfully" in last[71] and "Publication failed" in last[72]


async def test_a_format_without_a_channel_is_only_uploaded(up):
    await start_upload(up, channel=None)
    await up.flow.idle()
    assert len(up.client.forwarded) == 1 and not notices(up)
    final = last_edit(up)
    assert final.startswith("✅ <b>Uploaded to the Database Channel</b>") and "has no Channel" in final and NAME in final
    assert up.flow.flows == {}


async def test_when_forwarding_is_refused_the_file_is_sent_again_by_its_id(up):
    up.client.forward_errors.append(errors.ChatForwardsRestrictedError(None))
    await start_upload(up, caption="Some caption")
    nid, text, _ = await the_notice(up)
    assert up.client.forwarded == []
    (entity, file, kw), = up.client.files
    assert entity == DB and file is up.client.messages[555].media  # the file itself is not sent again, only its id
    assert kw["caption"] == "Some caption" and kw["formatting_entities"] is None and kw["parse_mode"] is None
    assert "Post: 3001" in text and up.client.downloads == []


async def test_a_failed_upload_can_be_tried_again(up):
    up.client.forward_errors.append(RuntimeError("no rights"))
    up.client.fail_send_file = True
    await start_upload(up)
    await up.flow.idle()
    chat, text, kw = up.client.edited[-1]
    assert "The upload to the Database Channel failed" in text and "RuntimeError: cannot send file" in text and "admin" in text
    assert flat_buttons(kw["buttons"]) == [("🔁 Try again", f"dbu:y:{OWNER}:555:{SHOW}"), ("✖ Close", "x:close")]
    assert not notices(up) and up.flow._busy == set()  # nothing was announced
    up.client.fail_send_file = False
    await click(up, OWNER, f"dbu:y:{OWNER}:555:{SHOW}", msg_id=77)
    await the_notice(up)
    assert len(up.client.forwarded) == 1


async def test_an_episode_that_was_deleted_is_reported(up):
    await click(up, OWNER, f"dbu:y:{OWNER}:999:{SHOW}", msg_id=77)
    await up.flow.idle()
    assert "not there anymore" in last_edit(up) and not up.client.forwarded and not notices(up)


async def test_a_second_tap_while_it_is_working_does_nothing(up):
    await start_upload(up)
    again = await click(up, OWNER, f"dbu:y:{OWNER}:555:{SHOW}", msg_id=77)
    assert again.toasts == ["I'm already on it…"] and again.edits == []
    await the_notice(up)
    await asyncio.sleep(0.05)
    assert len(up.client.forwarded) == 1 and len(notices(up)) == 1


async def test_no_does_not_upload(up):
    cb = await click(up, OWNER, "dbu:n", msg_id=77)
    assert "not uploaded" in cb.edits[-1][0] and cb.edits[-1][1] is None and cb.answers
    assert not up.client.forwarded and not up.client.sent


async def test_a_notice_that_cannot_be_sent_is_reported(up):
    up.client.message_errors[COMM] = RuntimeError("chat not found")
    await start_upload(up)
    await up.flow.idle()
    final = last_edit(up)
    assert "could not write in the communication chat" in final and "chat not found" in final and "Uploaded" in final
    assert len(up.client.forwarded) == 1 and up.flow.flows == {} and up.flow._busy == set()


async def test_an_unexpected_error_is_shown_instead_of_leaving_processing_on_the_screen(up):
    async def broken(*args, **kw):
        raise RuntimeError("boom")

    up.client.get_messages = broken
    await start_upload(up)
    await up.flow.idle()
    assert "Something went wrong" in last_edit(up) and "RuntimeError: boom" in last_edit(up) and up.flow._busy == set()


async def test_a_message_that_was_deleted_meanwhile_does_not_break_the_flow(up):
    async def gone(*args, **kw):
        raise errors.MessageIdInvalidError(None)

    up.client.edit_message = gone
    await start_upload(up)
    nid, text, _ = await the_notice(up)
    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(text)}", reply_to=nid)
    await up.flow.idle()
    assert up.flow.flows == {} and up.flow._busy == set()


async def test_shutdown_stops_the_waiting(up):
    await start_upload(up)
    await the_notice(up)
    await up.app.shutdown()
    assert up.flow._tasks == set() and up.flow.flows == {} and up.flow._busy == set()


async def test_strangers_cannot_press_the_buttons(up):
    up.client.messages[555] = document_message(NAME, 10, mid=555)
    await click(up, STRANGER, f"dbu:y:{STRANGER}:555:{SHOW}")
    assert not up.client.forwarded and not up.flow._tasks


@pytest.mark.parametrize("data", ["dbu", "dbu:y", "dbu:y:1:2", "dbu:y:a:b:c", "dbu:z", "dbu:y:1:2:3:4"])
async def test_a_broken_button_does_nothing(up, data):
    cb = await click(up, OWNER, data)
    assert cb.answers and not up.client.forwarded and up.flow._busy == set() and not up.flow._tasks