"""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"