"""Telegram: a fake Bot API stands in for api.telegram.org."""

import asyncio
import json
import logging
from typing import Any

import httpx
import pytest

import fluksio.main  # noqa: F401 - installs the log filter
from fluksio.flow import telegram
from fluksio.flow.alerts import Alert, AlertManager, Channel
from fluksio.flow.events import EventBus

TOKEN = "123456:SECRET-token"


class FakeBot:
    """Answers the Bot API, and remembers what it was asked."""

    def __init__(self, monkeypatch: pytest.MonkeyPatch) -> None:
        self.calls: list[tuple[str, dict[str, Any]]] = []
        #: Per method: a list of ``(status, body)`` answers, used in order.
        self.answers: dict[str, list[tuple[int, dict[str, Any]]]] = {}
        transport = httpx.MockTransport(self._handle)
        monkeypatch.setattr(
            telegram,
            "_client",
            lambda timeout: httpx.AsyncClient(transport=transport, timeout=timeout),
        )

    def _handle(self, request: httpx.Request) -> httpx.Response:
        assert request.url.path.startswith(f"/bot{TOKEN}/")
        method = request.url.path.rsplit("/", 1)[1]
        self.calls.append((method, json.loads(request.content or b"{}")))
        queued = self.answers.get(method)
        if queued:
            status, body = queued.pop(0) if len(queued) > 1 else queued[0]
            return httpx.Response(status, json=body)
        return httpx.Response(200, json={"ok": True, "result": True})

    def failing(self, method: str, status: int, description: str, **extra: Any) -> None:
        body = {"ok": False, "error_code": status, "description": description}
        if extra:
            body["parameters"] = extra
        self.answers[method] = [(status, body)]

    def sent(self) -> list[dict[str, Any]]:
        return [payload for method, payload in self.calls if method == "sendMessage"]


@pytest.fixture
def bot(monkeypatch: pytest.MonkeyPatch) -> FakeBot:
    return FakeBot(monkeypatch)


def test_an_alert_is_sent_escaped_and_quiet_when_it_is_a_warning(bot: FakeBot) -> None:
    asyncio.run(
        telegram.send_message(TOKEN, "42", "<pump> failed", "a & b <c>", quiet=True)
    )

    (message,) = bot.sent()
    assert message["chat_id"] == "42"
    assert message["parse_mode"] == "HTML"
    assert message["text"] == "<b>&lt;pump&gt; failed</b>\na &amp; b &lt;c&gt;"
    assert message["disable_notification"] is True


def test_a_long_body_is_cut_to_fit_a_message(bot: FakeBot) -> None:
    asyncio.run(telegram.send_message(TOKEN, "42", "t", "<" * 10_000))

    (message,) = bot.sent()
    assert len(message["text"]) < 4096
    assert message["text"].endswith("…")


@pytest.mark.parametrize(
    ("status", "description", "expected"),
    [
        (403, "Forbidden: bot was blocked by the user", "blocked or removed"),
        (400, "Bad Request: chat not found", "does not know this chat"),
        (401, "Unauthorized", "rejected the bot token"),
    ],
)
def test_a_refusal_says_what_to_do_and_never_quotes_the_url(
    bot: FakeBot, status: int, description: str, expected: str
) -> None:
    bot.failing("sendMessage", status, description)

    with pytest.raises(telegram.TelegramError) as raised:
        asyncio.run(telegram.send_message(TOKEN, "42", "t", "b"))

    assert expected in str(raised.value)
    assert "SECRET" not in str(raised.value)


def test_a_short_rate_limit_is_waited_out_once(bot: FakeBot) -> None:
    bot.answers["sendMessage"] = [
        (429, {"ok": False, "error_code": 429, "parameters": {"retry_after": 0.01}}),
        (200, {"ok": True, "result": True}),
    ]

    asyncio.run(telegram.send_message(TOKEN, "42", "t", "b"))

    assert len(bot.sent()) == 2


def test_a_long_rate_limit_is_reported_instead(bot: FakeBot) -> None:
    bot.failing("sendMessage", 429, "Too Many Requests", retry_after=120)

    with pytest.raises(telegram.TelegramError, match="rate limiting"):
        asyncio.run(telegram.send_message(TOKEN, "42", "t", "b"))

    assert len(bot.sent()) == 1


def test_a_network_error_does_not_carry_the_token(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    def refuse(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError(f"cannot reach {request.url}")

    transport = httpx.MockTransport(refuse)
    monkeypatch.setattr(
        telegram,
        "_client",
        lambda timeout: httpx.AsyncClient(transport=transport, timeout=timeout),
    )

    with pytest.raises(telegram.TelegramError) as raised:
        asyncio.run(telegram.call(TOKEN, "getMe"))

    assert "SECRET" not in str(raised.value)
    assert "api.telegram.org/bot***" in str(raised.value)


def test_the_token_stays_out_of_httpx_log_lines(
    bot: FakeBot, caplog: pytest.LogCaptureFixture
) -> None:
    with caplog.at_level(logging.INFO, logger="httpx"):
        asyncio.run(telegram.call(TOKEN, "getMe"))

    assert "api.telegram.org/bot***/getMe" in caplog.text
    assert "SECRET" not in caplog.text


def test_an_alert_channel_needs_a_token_and_a_chat() -> None:
    alerts = AlertManager(EventBus())
    alert = Alert(title="t", body="b")

    async def send(config: dict[str, Any]) -> None:
        await alerts.send(
            Channel(name="phone", kind="telegram", config=config),
            alert,
            raise_on_error=True,
        )

    with pytest.raises(ValueError, match="bot token"):
        asyncio.run(send({}))
    with pytest.raises(ValueError, match="not connected"):
        asyncio.run(send({"token": TOKEN}))


# -----------------------------------------------------------------------------
# Connecting a chat
# -----------------------------------------------------------------------------


def a_start(update_id: int, text: str, **chat: Any) -> dict[str, Any]:
    return {"update_id": update_id, "message": {"text": text, "chat": chat}}


def link(
    bot: FakeBot, updates: list[dict[str, Any]]
) -> tuple[telegram.Linker, list[tuple[str, str, str]], str]:
    """Open a code, and have the bot's first poll answer with *updates*.

    ``{code}`` in an update's text is replaced by the code that was issued.
    """
    linker = telegram.Linker()
    linked: list[tuple[str, str, str]] = []

    async def on_linked(channel: str, chat_id: str, label: str) -> None:
        linked.append((channel, chat_id, label))

    code = linker.start("garage", TOKEN, on_linked)
    filled = json.loads(json.dumps(updates).replace("{code}", code))
    bot.answers["getUpdates"] = [
        (200, {"ok": True, "result": filled}),
        (200, {"ok": True, "result": []}),
    ]
    return linker, linked, code


def test_the_chat_that_sends_the_code_is_connected_and_nobody_else(
    bot: FakeBot,
) -> None:
    async def scenario() -> tuple[telegram.Linker, list[tuple[str, str, str]]]:
        linker, linked, _ = link(
            bot,
            [
                a_start(1, "hello", id=7, first_name="Stranger"),
                a_start(2, "/start not-a-live-code", id=8, first_name="Guesser"),
                a_start(3, "/start {code}", id=42, first_name="Anna", last_name="B"),
            ],
        )
        await asyncio.wait_for(asyncio.gather(*linker._tasks.values()), 5)
        return linker, linked

    linker, linked = asyncio.run(scenario())

    assert linked == [("garage", "42", "Anna B")]
    # Only the connected chat is answered; the others get no reply at all.
    assert [m["chat_id"] for m in bot.sent()] == ["42"]
    assert linker.status("garage") == (False, 0, "")


def test_a_group_connects_through_start_at_bot(bot: FakeBot) -> None:
    async def scenario() -> list[tuple[str, str, str]]:
        linker, linked, _ = link(
            bot, [a_start(1, "/start@fluksio_bot {code}", id=-100, title="Garage")]
        )
        await asyncio.wait_for(asyncio.gather(*linker._tasks.values()), 5)
        return linked

    assert asyncio.run(scenario()) == [("garage", "-100", "Garage")]


def test_a_code_works_once(bot: FakeBot) -> None:
    async def scenario() -> list[tuple[str, str, str]]:
        linker, linked, _ = link(
            bot,
            [
                a_start(1, "/start {code}", id=42, first_name="Anna"),
                a_start(2, "/start {code}", id=43, first_name="Ben"),
            ],
        )
        await asyncio.wait_for(asyncio.gather(*linker._tasks.values()), 5)
        return linked

    assert [chat for _, chat, _ in asyncio.run(scenario())] == ["42"]


def test_an_expired_code_is_not_listened_for(bot: FakeBot) -> None:
    async def scenario() -> bool:
        linker = telegram.Linker()

        async def on_linked(channel: str, chat_id: str, label: str) -> None:
            raise AssertionError("expired")

        linker.start("garage", TOKEN, on_linked, ttl=0.0)
        await asyncio.gather(*linker._tasks.values())
        return linker.status("garage")[0]

    assert asyncio.run(scenario()) is False
    assert not [call for call in bot.calls if call[0] == "getUpdates"]


def test_a_webhook_or_second_poller_is_reported_not_retried(bot: FakeBot) -> None:
    bot.failing("getUpdates", 409, "Conflict: terminated by other getUpdates request")

    async def scenario() -> tuple[bool, int, str]:
        linker = telegram.Linker()

        async def on_linked(channel: str, chat_id: str, label: str) -> None:
            pass

        linker.start("garage", TOKEN, on_linked)
        await asyncio.wait_for(asyncio.gather(*linker._tasks.values()), 5)
        return linker.status("garage")

    waiting, _, error = asyncio.run(scenario())

    assert waiting is False
    assert "bot of its own" in error