stroblme
5d0ea480bf
The channel sends through the Bot API with httpx only: HTML with escaped text, cut to fit a message, silent for a warning, a short 429 waited out once, and errors that say what to do about a blocked bot, a lost chat or a bad token and never quote the request URL. The bot token is a stored secret: saving a channel refuses a literal one, the card's "Store token" writes the secret and a reference, and no endpoint returns it. A chat id is never typed. Connect starts a ten minute code, and the chat that sends `/start <code>` back through the bot is the one connected, so it is a chat somebody opened a link for. The engine fetches updates only while a code is waiting (no standing poll, no webhook, no inbound address) and ignores everything else without a reply. Connect needs a saved form, and the form takes the connected chat in without touching what was typed since. The token is part of every Bot API URL, so the httpx logger gets the same filter the request lines have, which now also masks an httpx.URL argument.
277 lines
9.4 KiB
Python
277 lines
9.4 KiB
Python
"""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><pump> failed</b>\na & b <c>"
|
|
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
|