bambuddy/backend/tests/integration/test_telegram_reactions.py

"""Answering the outcome prompt with a Telegram reaction (#3046).

The Bot API is stood in for at the httpx layer (the same ``_http_client``
swap the forum-topic tests use), the database is the real test engine: the
poller opens its own sessions, so the tests hand it the test session
factory instead of the module-level one.
"""

from datetime import timedelta
from unittest.mock import AsyncMock, patch

import httpx
import pytest
from httpx import AsyncClient
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from backend.app.models.archive import PrintArchive
from backend.app.models.notification import TelegramPendingVerdict
from backend.app.models.print_log import PrintLogEntry
from backend.app.services.notification_service import TELEGRAM_REACTION_HINT, NotificationService, notification_service
from backend.app.services.print_confirmation import apply_outcome_verdict, one_tap_url
from backend.app.services.telegram_reactions import (
    CONFLICT_COOLDOWN,
    MAX_BACKOFF,
    TelegramReactionPoller,
    verdict_from_reaction,
)
from backend.app.utils.local_time import utcnow_naive

BOT_TOKEN = "123456:AAbbCC"
CHAT_ID = "-1002520100736"
TELEGRAM_CONFIG = {"bot_token": BOT_TOKEN, "chat_id": CHAT_ID}
THUMBS_UP = "\U0001f44d"
THUMBS_DOWN = "\U0001f44e"


class _FakeBotApi:
    """Stand-in for httpx.AsyncClient with scripted responses, in call order."""

    def __init__(self, responses=None):
        self.is_closed = False
        self.calls: list[dict] = []
        self.responses = list(responses or [])

    async def post(self, url, json=None, data=None, files=None):
        self.calls.append({"url": url, "json": json, "data": data, "files": files})
        if self.responses:
            nxt = self.responses.pop(0)
            if isinstance(nxt, Exception):
                raise nxt
            return nxt
        return httpx.Response(200, json={"ok": True, "result": []})

    async def aclose(self):
        self.is_closed = True

    def urls(self) -> list[str]:
        return [c["url"].rsplit("/", 1)[-1] for c in self.calls]


def _updates(*updates: dict) -> httpx.Response:
    return httpx.Response(200, json={"ok": True, "result": list(updates)})


def _reaction(update_id: int, message_id: int, emoji: str | None, chat_id: str = CHAT_ID, kind: str = "emoji") -> dict:
    new_reaction = [] if emoji is None else [{"type": kind, "emoji": emoji}]
    return {
        "update_id": update_id,
        "message_reaction": {
            "chat": {"id": int(chat_id)},
            "message_id": message_id,
            "user": {"id": 42},
            "date": 1_700_000_000,
            "old_reaction": [],
            "new_reaction": new_reaction,
        },
    }


@pytest.fixture
async def poller(test_engine):
    p = TelegramReactionPoller()
    p._session_factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    p._http_client = _FakeBotApi()
    yield p
    await p.aclose()


def _track(poller: TelegramReactionPoller, *providers, token: str = BOT_TOKEN):
    """What sync() would record for these providers, without starting a task."""
    poller._providers.setdefault(token, set()).update(p.id for p in providers)


@pytest.fixture
def telegram_provider(notification_provider_factory):
    async def _create(**kwargs):
        defaults = {
            "name": "Telegram",
            "provider_type": "telegram",
            "config": TELEGRAM_CONFIG,
            "telegram_verdict_mode": "reactions",
        }
        defaults.update(kwargs)
        return await notification_provider_factory(**defaults)

    return _create


@pytest.fixture
def pending_factory(db_session):
    async def _create(provider_id: int, archive_id: int, message_id: int = 777, **kwargs):
        # A prompt only goes out with a live confirm token (the dispatch mints
        # one), and the send records it on the row: do the same here.
        if "confirm_token" not in kwargs:
            archive = await db_session.get(PrintArchive, archive_id)
            if archive.confirm_token is None:
                archive.confirm_token = f"tok-{archive_id}-{message_id}"
            kwargs["confirm_token"] = archive.confirm_token
        defaults = {
            "provider_id": provider_id,
            "chat_id": CHAT_ID,
            "message_id": message_id,
            "archive_id": archive_id,
            "has_caption": False,
            "message_text": "*Outcome?*\nTest_Print on X1C",
        }
        defaults.update(kwargs)
        row = TelegramPendingVerdict(**defaults)
        db_session.add(row)
        await db_session.commit()
        await db_session.refresh(row)
        return row

    return _create


async def _pending_count(db_session) -> int:
    return len((await db_session.execute(select(TelegramPendingVerdict))).scalars().all())


# ---------------------------------------------------------------------------
# Shared verdict helper
# ---------------------------------------------------------------------------


class TestApplyOutcomeVerdict:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_first_verdict_wins_and_mirrors(self, db_session, printer_factory, archive_factory):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True, confirm_token="tok")

        assert await apply_outcome_verdict(db_session, archive, "reject", source="reaction") is True
        await db_session.commit()
        assert archive.user_verdict == "reject"
        assert archive.user_verdict_source == "reaction"
        assert archive.user_verdict_at is not None
        # The capability is spent, not destroyed: the one-tap route can still
        # tell whoever taps an old button that this print was answered.
        assert archive.confirm_token == "tok"
        assert archive.confirm_token_used_at is not None
        entry = await db_session.scalar(
            select(PrintLogEntry).where(PrintLogEntry.archive_id == archive.id).order_by(PrintLogEntry.id.desc())
        )
        assert entry.user_verdict == "reject"

        # A later verdict from any path is a no-op.
        assert await apply_outcome_verdict(db_session, archive, "good", source="reaction") is False
        assert archive.user_verdict == "reject"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_rejects_unknown_verdict(self, db_session, printer_factory, archive_factory):
        printer = await printer_factory()
        archive = await archive_factory(printer.id)
        with pytest.raises(ValueError):
            await apply_outcome_verdict(db_session, archive, "meh", source="reaction")
        with pytest.raises(ValueError):
            await apply_outcome_verdict(db_session, archive, "good", source="carrier-pigeon")


# ---------------------------------------------------------------------------
# Sending the prompt
# ---------------------------------------------------------------------------


class _SendCapture:
    def __init__(self, message_id=555):
        self.is_closed = False
        self.calls: list[dict] = []
        self.message_id = message_id

    async def post(self, url, data=None, files=None, json=None):
        self.calls.append({"url": url, "data": data, "files": files, "json": json})
        result = {"message_id": self.message_id, "chat": {"id": int(CHAT_ID)}}
        return httpx.Response(200, json={"ok": True, "result": result})


CONFIRM_VARIABLES = {"good_url": "http://bambuddy.local/good", "reject_url": "http://bambuddy.local/reject"}


class TestSendingThePrompt:
    async def _send(self, provider, db_session, archive_id, image_data=None):
        service = NotificationService()
        client = _SendCapture()
        service._http_client = client
        ok, _ = await service._send_to_provider(
            provider,
            "Outcome?",
            "How did it go",
            db_session,
            image_data=image_data,
            event_type="print_confirm_request",
            variables={**CONFIRM_VARIABLES, "archive_id": archive_id},
        )
        assert ok
        return client

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_reactions_mode_drops_buttons_and_records_message(
        self, db_session, printer_factory, archive_factory, telegram_provider
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True, confirm_token="live-token")
        provider = await telegram_provider(telegram_verdict_mode="reactions")

        client = await self._send(provider, db_session, archive.id)

        body = client.calls[0]["json"]
        assert "reply_markup" not in body
        assert TELEGRAM_REACTION_HINT in body["text"]

        row = await db_session.scalar(select(TelegramPendingVerdict))
        assert row is not None
        assert row.provider_id == provider.id
        assert row.archive_id == archive.id
        assert row.message_id == 555
        assert row.chat_id == CHAT_ID
        assert row.confirm_token == "live-token", "the token the prompt's links carry"
        assert row.has_caption is False
        assert row.message_text and TELEGRAM_REACTION_HINT in row.message_text

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_both_mode_keeps_buttons_and_records_message(
        self, db_session, printer_factory, archive_factory, telegram_provider
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True, confirm_token="live-token")
        provider = await telegram_provider(telegram_verdict_mode="both")

        client = await self._send(provider, db_session, archive.id, image_data=b"\x89PNG")

        call = client.calls[0]
        assert call["url"].endswith("/sendPhoto")
        assert "inline_keyboard" in call["data"]["reply_markup"]
        row = await db_session.scalar(select(TelegramPendingVerdict))
        assert row is not None and row.has_caption is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_buttons_mode_is_unchanged(self, db_session, printer_factory, archive_factory, telegram_provider):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider(telegram_verdict_mode="buttons")

        client = await self._send(provider, db_session, archive.id)

        body = client.calls[0]["json"]
        assert body["reply_markup"] == {
            "inline_keyboard": [
                [
                    {"text": f"{THUMBS_UP} Good", "url": one_tap_url(CONFIRM_VARIABLES["good_url"])},
                    {"text": f"{THUMBS_DOWN} Reject", "url": one_tap_url(CONFIRM_VARIABLES["reject_url"])},
                ]
            ]
        }
        assert TELEGRAM_REACTION_HINT not in body["text"]
        assert await _pending_count(db_session) == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_prompt_without_a_live_token_records_nothing(
        self, db_session, printer_factory, archive_factory, telegram_provider
    ):
        """The prompt's links are dead without a live token, and a reaction
        must not outlive them: nothing to remember."""
        printer = await printer_factory()
        archive = await archive_factory(
            printer.id, confirm_requested=True, confirm_token="spent", confirm_token_used_at=utcnow_naive()
        )
        provider = await telegram_provider(telegram_verdict_mode="reactions")

        await self._send(provider, db_session, archive.id)

        assert await _pending_count(db_session) == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_missing_message_id_records_nothing(
        self, db_session, printer_factory, archive_factory, telegram_provider
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider(telegram_verdict_mode="reactions")

        service = NotificationService()
        client = _SendCapture(message_id=0)
        service._http_client = client
        ok, _ = await service._send_to_provider(
            provider,
            "Outcome?",
            "How did it go",
            db_session,
            event_type="print_confirm_request",
            variables={**CONFIRM_VARIABLES, "archive_id": archive.id},
        )
        assert ok
        assert await _pending_count(db_session) == 0


# ---------------------------------------------------------------------------
# Poller
# ---------------------------------------------------------------------------


class TestVerdictFromReaction:
    def test_thumbs_map_and_others_ignored(self):
        assert verdict_from_reaction([{"type": "emoji", "emoji": THUMBS_UP}]) == "good"
        assert verdict_from_reaction([{"type": "emoji", "emoji": THUMBS_DOWN}]) == "reject"
        assert verdict_from_reaction([{"type": "emoji", "emoji": "❤"}]) is None
        assert verdict_from_reaction([{"type": "custom_emoji", "custom_emoji_id": "1"}]) is None
        assert verdict_from_reaction([]) is None
        assert verdict_from_reaction([{"type": "emoji", "emoji": "❤"}, {"type": "emoji", "emoji": THUMBS_UP}]) == "good"


class TestPoller:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_thumbs_up_marks_good_deletes_row_and_edits_message(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True, confirm_token="tok")
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=777)
        poller._http_client = _FakeBotApi([_updates(_reaction(10, 777, THUMBS_UP))])

        assert await poller.poll_once(BOT_TOKEN) == "ok"

        await db_session.refresh(archive)
        assert archive.user_verdict == "good"
        assert archive.user_verdict_source == "reaction"
        assert archive.confirm_token_used_at is not None
        entry = await db_session.scalar(
            select(PrintLogEntry).where(PrintLogEntry.archive_id == archive.id).order_by(PrintLogEntry.id.desc())
        )
        assert entry.user_verdict == "good"
        assert await _pending_count(db_session) == 0

        assert poller._http_client.urls() == ["getUpdates", "editMessageText"]
        edit = poller._http_client.calls[1]["json"]
        assert edit["chat_id"] == CHAT_ID and edit["message_id"] == 777
        assert edit["text"].endswith("✅ marked as good")
        assert "Test\\_Print" in edit["text"], "stored body is re-escaped for Markdown"
        assert "reply_markup" not in edit
        assert edit["disable_web_page_preview"] is True, "the edit keeps the preview off, like the send"
        # The getUpdates that follows confirms the update we handled.
        assert poller._offsets[BOT_TOKEN] == 10

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_getupdates_request_shape(self, poller, telegram_provider):
        provider = await telegram_provider()
        _track(poller, provider)
        poller._offsets[BOT_TOKEN] = 41
        await poller.poll_once(BOT_TOKEN)
        call = poller._http_client.calls[0]
        assert call["url"] == f"https://api.telegram.org/bot{BOT_TOKEN}/getUpdates"
        assert call["json"] == {"timeout": 50, "allowed_updates": ["message_reaction"], "offset": 42}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_offset_belongs_to_the_bot_not_the_provider(self, poller, telegram_provider):
        """update_ids are a per-bot sequence: after a provider is re-pointed at
        another bot, the first getUpdates for it must not carry the old bot's
        offset, or the new bot's lower ids are confirmed away unseen."""
        provider = await telegram_provider()
        _track(poller, provider)
        poller._offsets[BOT_TOKEN] = 900_000_123

        _track(poller, provider, token="999:newtoken")
        await poller.poll_once("999:newtoken")

        call = poller._http_client.calls[0]
        assert call["url"].startswith("https://api.telegram.org/bot999:newtoken/")
        assert "offset" not in call["json"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_thumbs_down_rejects(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=778)
        poller._http_client = _FakeBotApi([_updates(_reaction(11, 778, THUMBS_DOWN))])

        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict == "reject"
        assert poller._http_client.calls[1]["json"]["text"].endswith("❌ marked as reject")

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_caption_message_is_edited_with_editMessageCaption(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=779, has_caption=True)
        poller._http_client = _FakeBotApi([_updates(_reaction(12, 779, THUMBS_UP))])

        await poller.poll_once(BOT_TOKEN)

        assert poller._http_client.urls() == ["getUpdates", "editMessageCaption"]
        assert poller._http_client.calls[1]["json"]["caption"].endswith("✅ marked as good")

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unknown_message_is_ignored(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=780)
        poller._http_client = _FakeBotApi(
            [_updates(_reaction(13, 999, THUMBS_UP), _reaction(14, 780, THUMBS_UP, chat_id="-100999"))]
        )

        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict is None
        assert await _pending_count(db_session) == 1
        assert poller._http_client.urls() == ["getUpdates"]
        assert poller._offsets[BOT_TOKEN] == 14

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_other_emoji_and_removed_reaction_are_ignored(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=781)
        poller._http_client = _FakeBotApi([_updates(_reaction(15, 781, "❤"), _reaction(16, 781, None))])

        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict is None
        assert await _pending_count(db_session) == 1, "the prompt stays open for a later thumbs"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_reaction_after_a_verdict_is_a_noop(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        """Someone already answered in the web UI: the late reaction neither
        flips the verdict nor edits the message, but the row is retired."""
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True, user_verdict="reject")
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=782)
        poller._http_client = _FakeBotApi([_updates(_reaction(17, 782, THUMBS_UP))])

        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict == "reject"
        assert await _pending_count(db_session) == 0
        assert poller._http_client.urls() == ["getUpdates"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_an_earlier_runs_prompt_cannot_answer_a_reprint(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, monkeypatch
    ):
        """Printing an archive again resets its verdict and confirm token, so
        run 1's links stop working. Run 1's unanswered message has to stop with
        them: a thumbs on it while run 2 prints records nothing, and run 2
        still asks its own question, which its own message then answers."""
        from backend.app.main import dispatch_outcome_confirmation

        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        _track(poller, provider)
        pending_columns = select(TelegramPendingVerdict.message_id, TelegramPendingVerdict.confirm_token)

        # Run 1 completes and asks; nobody answers.
        monkeypatch.setattr(notification_service, "_http_client", _SendCapture(message_id=901))
        assert await dispatch_outcome_confirmation(db_session, printer.id, printer.name, {}, archive.id, None)
        run1_token = archive.confirm_token
        assert (await db_session.execute(pending_columns)).one() == (901, run1_token)

        # Run 2 starts: the reprint reset in main.py (the expected_archive_id path).
        archive.status = "printing"
        archive.user_verdict = None
        archive.user_verdict_source = None
        archive.user_verdict_at = None
        archive.confirm_token = None
        archive.confirm_token_used_at = None
        await db_session.commit()

        # A thumbs-up on run 1's message while run 2 is printing.
        poller._http_client = _FakeBotApi([_updates(_reaction(30, 901, THUMBS_UP))])
        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict is None
        assert poller._http_client.urls() == ["getUpdates"], "the stale message is not edited"
        assert await _pending_count(db_session) == 0

        # Run 2 completes and asks with a token of its own.
        archive.status = "completed"
        await db_session.commit()
        monkeypatch.setattr(notification_service, "_http_client", _SendCapture(message_id=902))
        assert await dispatch_outcome_confirmation(db_session, printer.id, printer.name, {}, archive.id, None)
        assert archive.confirm_token not in (None, run1_token)
        assert (await db_session.execute(pending_columns)).one() == (902, archive.confirm_token)

        # Its own message answers it.
        poller._http_client = _FakeBotApi([_updates(_reaction(31, 902, THUMBS_DOWN))])
        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict == "reject"
        assert archive.user_verdict_source == "reaction"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_prompt_whose_link_was_used_is_retired(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        """A reaction stops working exactly when a link would: once the token
        is spent, or replaced by a newer prompt's, the message is done."""
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True, confirm_token="old")
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=785)
        archive.confirm_token = "newer"
        await db_session.commit()
        poller._http_client = _FakeBotApi([_updates(_reaction(32, 785, THUMBS_UP))])

        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict is None
        assert await _pending_count(db_session) == 0
        assert poller._http_client.urls() == ["getUpdates"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_second_reaction_on_same_message_is_a_noop(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=783)
        poller._http_client = _FakeBotApi(
            [_updates(_reaction(18, 783, THUMBS_UP)), _updates(_reaction(19, 783, THUMBS_DOWN))]
        )

        await poller.poll_once(BOT_TOKEN)
        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict == "good"
        assert poller._http_client.urls() == ["getUpdates", "editMessageText", "getUpdates"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_failed_edit_does_not_lose_the_verdict(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=784)
        poller._http_client = _FakeBotApi(
            [_updates(_reaction(20, 784, THUMBS_UP)), httpx.ConnectError("telegram gone")]
        )

        assert await poller.poll_once(BOT_TOKEN) == "ok"

        await db_session.refresh(archive)
        assert archive.user_verdict == "good"
        assert await _pending_count(db_session) == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_conflict_is_surfaced_and_cooled_down(self, poller, db_session, telegram_provider):
        """409 = webhook set or a second poller: report it on the provider and
        wait the cooldown instead of looping on the next getUpdates."""
        provider = await telegram_provider()
        _track(poller, provider)
        conflict = httpx.Response(
            409,
            json={"ok": False, "error_code": 409, "description": "Conflict: terminated by other getUpdates request"},
        )
        poller._http_client = _FakeBotApi([conflict, conflict])

        assert await poller.poll_once(BOT_TOKEN) == "conflict"

        await db_session.refresh(provider)
        assert "Conflict: terminated by other getUpdates request" in provider.last_error
        assert "webhook" in provider.last_error
        assert provider.last_error_at is not None

        sleeps: list[float] = []

        async def _sleep(seconds):
            sleeps.append(seconds)
            raise CancelledForTest

        poller._sleep = _sleep
        with pytest.raises(CancelledForTest):
            await poller._run(BOT_TOKEN)
        assert sleeps == [CONFLICT_COOLDOWN]
        assert len(poller._http_client.calls) == 2, "one getUpdates per cooldown, not a tight loop"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_bad_bot_token_is_surfaced_and_cooled_down(self, poller, db_session, telegram_provider):
        """401/404 = wrong or deleted bot token: it never fixes itself, so it
        lands on the provider card and waits the cooldown instead of a warning
        every minute forever."""
        provider = await telegram_provider()
        _track(poller, provider)
        unauthorized = httpx.Response(401, json={"ok": False, "error_code": 401, "description": "Unauthorized"})
        poller._http_client = _FakeBotApi([unauthorized, unauthorized])

        assert await poller.poll_once(BOT_TOKEN) == "rejected"

        await db_session.refresh(provider)
        assert "HTTP 401" in provider.last_error
        assert "Unauthorized" in provider.last_error
        assert "bot token" in provider.last_error
        assert provider.last_error_at is not None

        sleeps: list[float] = []

        async def _sleep(seconds):
            sleeps.append(seconds)
            raise CancelledForTest

        poller._sleep = _sleep
        with pytest.raises(CancelledForTest):
            await poller._run(BOT_TOKEN)
        assert sleeps == [CONFLICT_COOLDOWN]
        assert len(poller._http_client.calls) == 2

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_conflict_is_recorded_on_every_provider_of_the_bot(self, poller, db_session, telegram_provider):
        first = await telegram_provider(name="X1C")
        second = await telegram_provider(name="P1S")
        _track(poller, first, second)
        conflict = httpx.Response(409, json={"ok": False, "error_code": 409, "description": "Conflict"})
        poller._http_client = _FakeBotApi([conflict])

        assert await poller.poll_once(BOT_TOKEN) == "conflict"

        for provider in (first, second):
            await db_session.refresh(provider)
            assert provider.last_error and "Conflict" in provider.last_error

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_network_errors_back_off_exponentially(self, poller, telegram_provider):
        provider = await telegram_provider()
        _track(poller, provider)
        poller._http_client = _FakeBotApi([httpx.ConnectError("down")] * 10)
        sleeps: list[float] = []

        async def _sleep(seconds):
            sleeps.append(seconds)
            if len(sleeps) == 8:
                raise CancelledForTest

        poller._sleep = _sleep
        with pytest.raises(CancelledForTest):
            await poller._run(BOT_TOKEN)
        assert sleeps == [1, 2, 4, 8, 16, 32, 60, 60]
        assert max(sleeps) == MAX_BACKOFF

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_prune_drops_rows_older_than_a_week(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        _track(poller, provider)
        await pending_factory(provider.id, archive.id, message_id=1, created_at=utcnow_naive() - timedelta(days=8))
        fresh = await pending_factory(provider.id, archive.id, message_id=2)

        await poller.prune_stale()

        rows = (await db_session.execute(select(TelegramPendingVerdict))).scalars().all()
        assert [r.id for r in rows] == [fresh.id]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_sync_follows_provider_mode_enabled_and_token(self, poller, db_session, telegram_provider):
        never = _NeverAnswers()
        poller._http_client = never
        reactions = await telegram_provider(name="R", telegram_verdict_mode="reactions")
        buttons = await telegram_provider(name="B", telegram_verdict_mode="buttons")
        disabled = await telegram_provider(name="D", telegram_verdict_mode="both", enabled=False)

        await poller.sync()
        assert poller.is_polling(reactions.id)
        assert not poller.is_polling(buttons.id)
        assert not poller.is_polling(disabled.id)
        assert set(poller._tasks) == {BOT_TOKEN}

        # The bot keeps its single poll while any provider wants it; the
        # provider set behind it follows the edits.
        first_task = poller._tasks[BOT_TOKEN]
        reactions.telegram_verdict_mode = "buttons"
        buttons.telegram_verdict_mode = "both"
        await db_session.commit()
        await poller.sync()
        assert not poller.is_polling(reactions.id)
        assert poller.is_polling(buttons.id)
        assert poller._tasks[BOT_TOKEN] is first_task
        assert poller._providers[BOT_TOKEN] == {buttons.id}

        # A token change moves the provider to a new bot: the old bot's poll
        # stops because nobody uses it any more, a fresh one starts.
        buttons.config = '{"bot_token": "999:newtoken", "chat_id": "1"}'
        await db_session.commit()
        await poller.sync()
        assert first_task.cancelled() or first_task.cancelling()
        assert set(poller._tasks) == {"999:newtoken"}
        assert poller.is_polling(buttons.id)

        await poller.aclose()
        assert poller._tasks == {}
        assert never.is_closed

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_providers_sharing_a_bot_share_one_poll(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        """The usual farm layout: one bot, one provider per printer. Telegram
        serves a single getUpdates consumer per bot, so both providers ride
        one task and a reaction to either prompt lands on its archive."""
        poller._http_client = _NeverAnswers()
        x1c = await printer_factory(name="X1C")
        p1s = await printer_factory(name="P1S")
        first = await telegram_provider(name="X1C", printer_id=x1c.id, config={"bot_token": BOT_TOKEN, "chat_id": "1"})
        second = await telegram_provider(name="P1S", printer_id=p1s.id, config={"bot_token": BOT_TOKEN, "chat_id": "2"})

        await poller.sync()
        assert len(poller._tasks) == 1
        assert poller.is_polling(first.id) and poller.is_polling(second.id)
        poller.stop()
        _track(poller, first, second)

        archive_a = await archive_factory(x1c.id, confirm_requested=True)
        archive_b = await archive_factory(p1s.id, confirm_requested=True)
        await pending_factory(first.id, archive_a.id, message_id=100, chat_id="1")
        await pending_factory(second.id, archive_b.id, message_id=100, chat_id="2")
        poller._http_client = _FakeBotApi(
            [_updates(_reaction(30, 100, THUMBS_DOWN, chat_id="2"), _reaction(31, 100, THUMBS_UP, chat_id="1"))]
        )

        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive_a)
        await db_session.refresh(archive_b)
        assert archive_a.user_verdict == "good"
        assert archive_b.user_verdict == "reject"
        assert await _pending_count(db_session) == 0
        assert poller._http_client.urls() == ["getUpdates", "editMessageText", "editMessageText"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_reaction_for_another_bots_provider_is_not_matched(
        self, poller, db_session, printer_factory, archive_factory, telegram_provider, pending_factory
    ):
        """Two bots in private chat with the same person number their
        messages independently, so a match is scoped to the bot's providers."""
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        other = await telegram_provider(name="other bot", config={"bot_token": "999:other", "chat_id": CHAT_ID})
        mine = await telegram_provider(name="mine")
        _track(poller, mine)
        _track(poller, other, token="999:other")
        await pending_factory(other.id, archive.id, message_id=5)
        poller._http_client = _FakeBotApi([_updates(_reaction(40, 5, THUMBS_UP))])

        await poller.poll_once(BOT_TOKEN)

        await db_session.refresh(archive)
        assert archive.user_verdict is None
        assert await _pending_count(db_session) == 1


class CancelledForTest(BaseException):
    """Escape hatch out of the endless poll loop without tripping its handlers."""


class _NeverAnswers:
    """getUpdates that stays open forever, like a quiet chat would."""

    def __init__(self):
        self.is_closed = False

    async def post(self, url, json=None, data=None, files=None):
        import asyncio

        await asyncio.Event().wait()

    async def aclose(self):
        self.is_closed = True


# ---------------------------------------------------------------------------
# Provider API round trip
# ---------------------------------------------------------------------------


class TestProviderApi:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_mode_round_trips_and_defaults_to_buttons(self, async_client: AsyncClient):
        with patch("backend.app.api.routes.notifications.telegram_reaction_poller.sync", new=AsyncMock()) as sync:
            created = await async_client.post(
                "/api/v1/notifications/",
                json={
                    "name": "TG",
                    "provider_type": "telegram",
                    "config": TELEGRAM_CONFIG,
                    "telegram_verdict_mode": "reactions",
                },
            )
            assert created.status_code == 200, created.text
            assert created.json()["telegram_verdict_mode"] == "reactions"
            provider_id = created.json()["id"]
            assert sync.await_count == 1

            listed = await async_client.get("/api/v1/notifications/")
            assert [p["telegram_verdict_mode"] for p in listed.json() if p["id"] == provider_id] == ["reactions"]

            patched = await async_client.patch(
                f"/api/v1/notifications/{provider_id}", json={"telegram_verdict_mode": "both"}
            )
            assert patched.status_code == 200
            assert patched.json()["telegram_verdict_mode"] == "both"
            assert sync.await_count == 2

            invalid = await async_client.patch(
                f"/api/v1/notifications/{provider_id}", json={"telegram_verdict_mode": "webhook"}
            )
            assert invalid.status_code == 422

            other = await async_client.post(
                "/api/v1/notifications/",
                json={"name": "ntfy", "provider_type": "ntfy", "config": {"topic": "t"}},
            )
            assert other.json()["telegram_verdict_mode"] == "buttons"

            deleted = await async_client.delete(f"/api/v1/notifications/{provider_id}")
            assert deleted.status_code == 200
            assert sync.await_count == 4

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_legacy_null_mode_reads_as_buttons(
        self, async_client: AsyncClient, notification_provider_factory, db_session
    ):
        provider = await notification_provider_factory(name="Legacy")
        # The ORM default fills a None at INSERT; a pre-#3046 row holds a real NULL.
        await db_session.execute(
            text("UPDATE notification_providers SET telegram_verdict_mode = NULL WHERE id = :id"), {"id": provider.id}
        )
        await db_session.commit()
        stored = await db_session.scalar(
            text("SELECT telegram_verdict_mode FROM notification_providers WHERE id = :id"), {"id": provider.id}
        )
        assert stored is None, "row under test must actually hold NULL"

        response = await async_client.get(f"/api/v1/notifications/{provider.id}")
        assert response.status_code == 200
        assert response.json()["telegram_verdict_mode"] == "buttons"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleting_a_provider_cascades_its_pending_rows(
        self,
        async_client: AsyncClient,
        db_session,
        printer_factory,
        archive_factory,
        telegram_provider,
        pending_factory,
    ):
        printer = await printer_factory()
        archive = await archive_factory(printer.id, confirm_requested=True)
        provider = await telegram_provider()
        await pending_factory(provider.id, archive.id)

        with patch("backend.app.api.routes.notifications.telegram_reaction_poller.sync", new=AsyncMock()):
            assert (await async_client.delete(f"/api/v1/notifications/{provider.id}")).status_code == 200

        assert await _pending_count(db_session) == 0