256 lines
9.3 KiB
Python
256 lines
9.3 KiB
Python
"""Webhook-mode transport tests: the bot's HTTP boundary (ticket 01) and the
|
|
webhook registration lifecycle (ticket 02).
|
|
|
|
The webhook server is the single new seam introduced by the
|
|
webhook-report-payments spec, so it is exercised in-process over real HTTP
|
|
against the aiohttp application (TestServer/TestClient). The dispatcher is
|
|
observed through an outer update middleware — forged requests must never
|
|
reach it. The Telegram Bot API side of the lifecycle (set/delete webhook) is
|
|
observed through a recording stand-in bot.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from aiogram import Dispatcher
|
|
from aiogram.types import Update
|
|
from aiohttp.test_utils import TestClient, TestServer
|
|
from pydantic import ValidationError
|
|
|
|
from src.contract_check.bot.config import BotSettings
|
|
from src.contract_check.bot.webhook import (
|
|
build_webhook_app,
|
|
derive_webhook_path,
|
|
webhook_public_url,
|
|
)
|
|
|
|
BOT_TOKEN = "123456:webhook-unit-test-token"
|
|
SECRET = "unit-test-secret_token-01"
|
|
PATH = derive_webhook_path(BOT_TOKEN, SECRET)
|
|
|
|
UPDATE_PAYLOAD: dict[str, Any] = {
|
|
"update_id": 42,
|
|
"message": {
|
|
"message_id": 1,
|
|
"date": 1735689600,
|
|
"text": "привет",
|
|
"chat": {"id": 4242, "type": "private"},
|
|
"from": {"id": 4242, "is_bot": False, "first_name": "Tester"},
|
|
},
|
|
}
|
|
|
|
|
|
class RecordingBot:
|
|
"""Duck-typed Bot: records lifecycle calls, never touches the network."""
|
|
|
|
def __init__(self, token: str = BOT_TOKEN) -> None:
|
|
self.id = int(token.split(":")[0])
|
|
self.calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
async def set_webhook(self, **kwargs: Any) -> None:
|
|
self.calls.append(("set_webhook", kwargs))
|
|
|
|
async def delete_webhook(self, **kwargs: Any) -> None:
|
|
self.calls.append(("delete_webhook", kwargs))
|
|
|
|
|
|
def make_settings(**overrides: Any) -> BotSettings:
|
|
base: dict[str, Any] = {
|
|
"env": "test",
|
|
"bot_token": BOT_TOKEN,
|
|
"bot_service_token": "svc-token",
|
|
"update_mode": "webhook",
|
|
"webhook_public_base_url": "https://edge.example.com",
|
|
"webhook_secret_token": SECRET,
|
|
}
|
|
base.update(overrides)
|
|
return BotSettings(**base) # type: ignore[call-arg]
|
|
|
|
|
|
def make_dispatcher(seen: list[Update]) -> Dispatcher:
|
|
"""Dispatcher that records every fed update and runs no handlers."""
|
|
|
|
async def capture(handler: Any, event: Update, data: dict[str, Any]) -> None:
|
|
seen.append(event)
|
|
return None
|
|
|
|
dp = Dispatcher()
|
|
dp.update.outer_middleware(capture)
|
|
return dp
|
|
|
|
|
|
@pytest.fixture
|
|
async def webhook_client() -> Any:
|
|
"""Started webhook app behind a real in-process HTTP server."""
|
|
|
|
seen: list[Update] = []
|
|
app = build_webhook_app(make_dispatcher(seen), RecordingBot(), make_settings())
|
|
async with TestClient(TestServer(app)) as client:
|
|
yield client, seen
|
|
|
|
|
|
# ── Ticket 01: HTTP boundary ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestWebhookHttpBoundary:
|
|
async def test_valid_update_with_correct_secret_is_accepted(self, webhook_client: Any) -> None:
|
|
client, seen = webhook_client
|
|
resp = await client.post(
|
|
PATH,
|
|
json=UPDATE_PAYLOAD,
|
|
headers={"X-Telegram-Bot-Api-Secret-Token": SECRET},
|
|
)
|
|
assert resp.status == 200
|
|
assert len(seen) == 1
|
|
assert seen[0].update_id == 42
|
|
assert seen[0].message is not None
|
|
assert seen[0].message.text == "привет"
|
|
|
|
async def test_wrong_secret_token_rejected_403(self, webhook_client: Any) -> None:
|
|
client, seen = webhook_client
|
|
resp = await client.post(
|
|
PATH,
|
|
json=UPDATE_PAYLOAD,
|
|
headers={"X-Telegram-Bot-Api-Secret-Token": "wrong-secret"},
|
|
)
|
|
assert resp.status == 403
|
|
assert seen == []
|
|
|
|
async def test_missing_secret_token_rejected_403(self, webhook_client: Any) -> None:
|
|
client, seen = webhook_client
|
|
resp = await client.post(PATH, json=UPDATE_PAYLOAD)
|
|
assert resp.status == 403
|
|
assert seen == []
|
|
|
|
async def test_empty_secret_token_rejected_403(self, webhook_client: Any) -> None:
|
|
client, seen = webhook_client
|
|
resp = await client.post(
|
|
PATH,
|
|
json=UPDATE_PAYLOAD,
|
|
headers={"X-Telegram-Bot-Api-Secret-Token": ""},
|
|
)
|
|
assert resp.status == 403
|
|
assert seen == []
|
|
|
|
async def test_non_ascii_secret_token_rejected_403(self, webhook_client: Any) -> None:
|
|
"""A forged header must get 403, never a 500 from compare_digest."""
|
|
client, seen = webhook_client
|
|
resp = await client.post(
|
|
PATH,
|
|
json=UPDATE_PAYLOAD,
|
|
headers={"X-Telegram-Bot-Api-Secret-Token": "не-секрет"},
|
|
)
|
|
assert resp.status == 403
|
|
assert seen == []
|
|
|
|
async def test_malformed_json_with_valid_secret_rejected_400(self, webhook_client: Any) -> None:
|
|
client, seen = webhook_client
|
|
resp = await client.post(
|
|
PATH,
|
|
data=b"not-json",
|
|
headers={"X-Telegram-Bot-Api-Secret-Token": SECRET},
|
|
)
|
|
assert resp.status == 400
|
|
assert seen == []
|
|
|
|
async def test_healthz_returns_200(self, webhook_client: Any) -> None:
|
|
client, _ = webhook_client
|
|
resp = await client.get("/healthz")
|
|
assert resp.status == 200
|
|
|
|
async def test_unknown_path_returns_404(self, webhook_client: Any) -> None:
|
|
client, seen = webhook_client
|
|
resp = await client.post(
|
|
"/tg-webhook/not-the-derived-path",
|
|
json=UPDATE_PAYLOAD,
|
|
headers={"X-Telegram-Bot-Api-Secret-Token": SECRET},
|
|
)
|
|
assert resp.status == 404
|
|
assert seen == []
|
|
|
|
|
|
# ── Ticket 01: mode-switch configuration ─────────────────────────────────────
|
|
|
|
|
|
class TestUpdateModeConfig:
|
|
def test_default_mode_is_polling(self) -> None:
|
|
settings = BotSettings(env="test", bot_token=BOT_TOKEN, bot_service_token="svc") # type: ignore[call-arg]
|
|
assert settings.update_mode == "polling"
|
|
|
|
def test_invalid_mode_rejected(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
make_settings(update_mode="banana")
|
|
|
|
def test_webhook_mode_requires_public_base_url(self) -> None:
|
|
with pytest.raises(ValidationError, match="webhook_public_base_url"):
|
|
make_settings(webhook_public_base_url="")
|
|
|
|
def test_webhook_base_url_scheme_validated(self) -> None:
|
|
with pytest.raises(ValidationError, match="http"):
|
|
make_settings(webhook_public_base_url="ftp://edge.example.com")
|
|
|
|
def test_webhook_mode_requires_secret_token(self) -> None:
|
|
with pytest.raises(ValidationError, match="webhook_secret_token"):
|
|
make_settings(webhook_secret_token="")
|
|
|
|
def test_webhook_secret_token_charset_validated(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
make_settings(webhook_secret_token="bad secret!")
|
|
|
|
def test_polling_mode_tolerates_empty_webhook_settings(self) -> None:
|
|
settings = make_settings(
|
|
update_mode="polling",
|
|
webhook_public_base_url="",
|
|
webhook_secret_token="",
|
|
)
|
|
assert settings.update_mode == "polling"
|
|
|
|
|
|
# ── Secret-derived path ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestWebhookPathDerivation:
|
|
def test_path_is_deterministic_and_prefixed(self) -> None:
|
|
assert PATH == derive_webhook_path(BOT_TOKEN, SECRET)
|
|
assert PATH.startswith("/tg-webhook/")
|
|
assert len(PATH) > len("/tg-webhook/")
|
|
|
|
def test_path_depends_on_secret_and_bot_token(self) -> None:
|
|
assert PATH != derive_webhook_path(BOT_TOKEN, "another-secret-token")
|
|
assert PATH != derive_webhook_path("999999:other-bot", SECRET)
|
|
|
|
def test_public_url_joins_base_and_path(self) -> None:
|
|
assert webhook_public_url(make_settings()) == f"https://edge.example.com{PATH}"
|
|
|
|
def test_public_url_tolerates_trailing_slash(self) -> None:
|
|
settings = make_settings(webhook_public_base_url="https://edge.example.com/")
|
|
assert webhook_public_url(settings) == f"https://edge.example.com{PATH}"
|
|
|
|
|
|
# ── Ticket 02: registration lifecycle ────────────────────────────────────────
|
|
|
|
|
|
class TestWebhookLifecycle:
|
|
async def test_startup_registers_webhook_with_drop_pending_updates(self) -> None:
|
|
|
|
bot = RecordingBot()
|
|
app = build_webhook_app(make_dispatcher([]), bot, make_settings())
|
|
async with TestServer(app):
|
|
pass
|
|
|
|
assert bot.calls[0][0] == "set_webhook"
|
|
kwargs = bot.calls[0][1]
|
|
assert kwargs["url"] == f"https://edge.example.com{PATH}"
|
|
assert kwargs["secret_token"] == SECRET
|
|
assert kwargs["drop_pending_updates"] is True
|
|
|
|
async def test_shutdown_deregisters_webhook(self) -> None:
|
|
|
|
bot = RecordingBot()
|
|
app = build_webhook_app(make_dispatcher([]), bot, make_settings())
|
|
async with TestServer(app):
|
|
pass
|
|
|
|
assert [name for name, _ in bot.calls][-1] == "delete_webhook"
|