122 lines
3.7 KiB
Python
122 lines
3.7 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.chat import MessageRef, TelegramAdapter, TelegramError, mask_secret
|
|
|
|
TOKEN = "123456:SECRET-token_abc"
|
|
BASE = "https://api.telegram.org"
|
|
|
|
|
|
def make(handler, **kwargs):
|
|
return TelegramAdapter(TOKEN, "42", transport=httpx.MockTransport(handler), **kwargs)
|
|
|
|
|
|
def ok(result):
|
|
return httpx.Response(200, json={"ok": True, "result": result})
|
|
|
|
|
|
async def test_send_payload_and_ref():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["url"] = str(request.url)
|
|
seen["body"] = json.loads(request.content)
|
|
return ok({"message_id": 7})
|
|
|
|
ref = await make(handler).send("42", "<b>x</b> <")
|
|
assert seen["url"] == f"{BASE}/bot{TOKEN}/sendMessage"
|
|
assert seen["body"]["parse_mode"] == "HTML"
|
|
assert seen["body"]["chat_id"] == "42" and seen["body"]["text"] == "<b>x</b> <"
|
|
assert ref == MessageRef("telegram", "42", "7")
|
|
|
|
|
|
async def test_edit_payload():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["url"] = str(request.url)
|
|
seen["body"] = json.loads(request.content)
|
|
return ok({})
|
|
|
|
await make(handler).edit(MessageRef("telegram", "42", "7"), "neu")
|
|
assert seen["url"].endswith("/editMessageText")
|
|
assert seen["body"]["message_id"] == 7 and seen["body"]["parse_mode"] == "HTML"
|
|
|
|
|
|
async def test_429_retries_once_with_retry_after():
|
|
calls, slept = [], []
|
|
|
|
def handler(request):
|
|
calls.append(1)
|
|
if len(calls) == 1:
|
|
return httpx.Response(429, json={"ok": False, "description": "Too Many Requests",
|
|
"parameters": {"retry_after": 3}})
|
|
return ok({"message_id": 1})
|
|
|
|
async def fake_sleep(seconds):
|
|
slept.append(seconds)
|
|
|
|
await make(handler, sleep=fake_sleep).send("42", "x")
|
|
assert len(calls) == 2 and slept == [3.0]
|
|
|
|
|
|
async def test_429_twice_raises():
|
|
async def fake_sleep(seconds):
|
|
pass
|
|
|
|
def handler(request):
|
|
return httpx.Response(429, json={"ok": False, "description": "Too Many Requests"})
|
|
|
|
with pytest.raises(TelegramError):
|
|
await make(handler, sleep=fake_sleep).send("42", "x")
|
|
|
|
|
|
async def test_not_modified_is_ignored():
|
|
def handler(request):
|
|
return httpx.Response(400, json={"ok": False, "description": "Bad Request: message is not modified"})
|
|
|
|
await make(handler).edit(MessageRef("telegram", "42", "7"), "gleich")
|
|
|
|
|
|
async def test_other_edit_errors_raise():
|
|
def handler(request):
|
|
return httpx.Response(400, json={"ok": False, "description": "Bad Request: chat not found"})
|
|
|
|
with pytest.raises(TelegramError):
|
|
await make(handler).edit(MessageRef("telegram", "42", "7"), "x")
|
|
|
|
|
|
async def test_get_me_returns_username():
|
|
def handler(request):
|
|
return ok({"username": "checkbot_bot"})
|
|
|
|
assert await make(handler).check() == "Bot @checkbot_bot"
|
|
|
|
|
|
async def test_token_never_in_connection_error():
|
|
def handler(request):
|
|
raise httpx.ConnectError(f"cannot connect to {request.url}")
|
|
|
|
adapter = make(handler)
|
|
for call in (adapter.send("42", "x"), adapter.check(), adapter.edit(MessageRef("telegram", "42", "1"), "x")):
|
|
with pytest.raises(TelegramError) as info:
|
|
await call
|
|
assert TOKEN not in str(info.value)
|
|
assert info.value.__cause__ is None and info.value.__suppress_context__
|
|
|
|
|
|
async def test_token_never_in_http_error_description():
|
|
def handler(request):
|
|
return httpx.Response(401, json={"ok": False, "description": f"Unauthorized {TOKEN}"})
|
|
|
|
with pytest.raises(TelegramError) as info:
|
|
await make(handler).check()
|
|
assert TOKEN not in str(info.value)
|
|
|
|
|
|
def test_mask_secret_handles_url_encoded_variant():
|
|
assert TOKEN not in mask_secret(f"x {TOKEN} y", TOKEN)
|
|
assert mask_secret("a b", "") == "a b"
|