tests/test_telegram_reply_transport.py

import os
import unittest
from unittest.mock import patch

import httpx

from api.telegram_webhook import _handle_inline, _tg_answer_callback, _tg_reply, _tg_send


class _Response:
    def __init__(self, status_code=200, payload=None, text=""):
        self.status_code = status_code
        self._payload = payload if payload is not None else {"ok": True}
        self.text = text

    def json(self):
        return self._payload


class _Client:
    instances = []
    response = _Response()
    exception = None
    last_post = None

    def __init__(self, **kwargs):
        self.kwargs = kwargs
        type(self).instances.append(self)

    async def __aenter__(self):
        return self

    async def __aexit__(self, *_args):
        return False

    async def post(self, *args, **kwargs):
        type(self).last_post = (args, kwargs)
        if type(self).exception is not None:
            raise type(self).exception
        return type(self).response


class TelegramReplyTransportTests(unittest.IsolatedAsyncioTestCase):
    def setUp(self):
        _Client.instances = []
        _Client.response = _Response()
        _Client.exception = None
        _Client.last_post = None

    async def test_reply_bypasses_environment_proxy_and_accepts_success(self):
        with patch("httpx.AsyncClient", _Client):
            await _tg_reply(123, "<b>hello</b>", token="test-token")

        self.assertEqual(len(_Client.instances), 1)
        self.assertFalse(_Client.instances[0].kwargs["trust_env"])
        self.assertIsInstance(_Client.instances[0].kwargs["timeout"], httpx.Timeout)

    async def test_reply_uses_authenticated_pages_gateway_when_configured(self):
        with patch.dict(
            os.environ,
            {
                "TELEGRAM_REPLY_PROXY_URL": "https://tma-agente.pages.dev/api/telegram/send",
                "TELEGRAM_REPLY_PROXY_SECRET": "gateway-secret",
            },
            clear=False,
        ), patch("httpx.AsyncClient", _Client):
            await _tg_reply(123, "hello", token="test-token")

        args, kwargs = _Client.last_post
        self.assertEqual(args[0], "https://tma-agente.pages.dev/api/telegram/send")
        self.assertEqual(kwargs["headers"], {"Authorization": "Bearer gateway-secret"})

    async def test_stream_message_uses_gateway_and_returns_message_id(self):
        _Client.response = _Response(payload={"ok": True, "result": {"message_id": 77}})
        with patch.dict(
            os.environ,
            {
                "TELEGRAM_REPLY_PROXY_URL": "https://tma-agente.pages.dev/api/telegram/send",
                "TELEGRAM_REPLY_PROXY_SECRET": "gateway-secret",
            },
            clear=False,
        ), patch("httpx.AsyncClient", _Client):
            message_id = await _tg_send(123, "stream", token="test-token")

        args, kwargs = _Client.last_post
        self.assertEqual(message_id, "77")
        self.assertEqual(args[0], "https://tma-agente.pages.dev/api/telegram/send")
        self.assertEqual(kwargs["json"]["method"], "sendMessage")

    async def test_callback_and_inline_answer_use_gateway(self):
        with patch.dict(
            os.environ,
            {
                "TELEGRAM_REPLY_PROXY_URL": "https://tma-agente.pages.dev/api/telegram/send",
                "TELEGRAM_REPLY_PROXY_SECRET": "gateway-secret",
            },
            clear=False,
        ), patch("httpx.AsyncClient", _Client):
            await _tg_answer_callback("callback-id", token="test-token")
            args, kwargs = _Client.last_post
            self.assertEqual(args[0], "https://tma-agente.pages.dev/api/telegram/send")
            self.assertEqual(kwargs["json"]["method"], "answerCallbackQuery")

            await _handle_inline({"id": "inline-id", "query": "ciao"}, "test-token")

        _args, kwargs = _Client.last_post
        self.assertEqual(kwargs["json"]["method"], "answerInlineQuery")

    async def test_reply_logs_rejected_telegram_response(self):
        _Client.response = _Response(
            status_code=429,
            payload={"ok": False, "description": "Too Many Requests"},
        )
        with patch("httpx.AsyncClient", _Client), self.assertLogs(
            "api.telegram_webhook", level="WARNING"
        ) as logs:
            await _tg_reply(123, "hello", token="test-token")

        self.assertIn("status=429 detail=Too Many Requests", "\n".join(logs.output))

    async def test_reply_logs_timeout_type_when_transport_fails(self):
        _Client.exception = httpx.ReadTimeout("")
        with patch("httpx.AsyncClient", _Client), self.assertLogs(
            "api.telegram_webhook", level="WARNING"
        ) as logs:
            await _tg_reply(123, "hello", token="test-token")

        self.assertIn("ReadTimeout", "\n".join(logs.output))


if __name__ == "__main__":
    unittest.main()