| 1 | from helpers.api import ApiHandler, Request |
| 2 | from helpers.errors import format_error |
| 3 | from plugins._telegram_integration.helpers.dependencies import ensure_dependencies |
| 4 | |
| 5 | |
| 6 | class TestConnection(ApiHandler): |
| 7 | |
| 8 | async def process(self, input: dict, request: Request) -> dict: |
| 9 | bot_cfg = input.get("bot", {}) |
| 10 | token = bot_cfg.get("token", "") |
| 11 | results: list[dict] = [] |
| 12 | |
| 13 | if not token: |
| 14 | results.append({ |
| 15 | "test": "Bot token", |
| 16 | "ok": False, |
| 17 | "message": "Add your bot token first.", |
| 18 | }) |
| 19 | return {"success": False, "results": results} |
| 20 | |
| 21 | try: |
| 22 | ensure_dependencies() |
| 23 | from plugins._telegram_integration.helpers.bot_manager import test_token |
| 24 | ok, message = await test_token(token) |
| 25 | results.append({ |
| 26 | "test": "Telegram bot", |
| 27 | "ok": ok, |
| 28 | "message": "Telegram accepted the bot token." if ok else message, |
| 29 | }) |
| 30 | except Exception as e: |
| 31 | results.append({ |
| 32 | "test": "Telegram bot", |
| 33 | "ok": False, |
| 34 | "message": f"Could not validate the bot token: {format_error(e)}", |
| 35 | }) |
| 36 | |
| 37 | return {"success": all(r["ok"] for r in results), "results": results} |