| 1 | """Test WhatsApp bridge connection health.""" |
| 2 | |
| 3 | from helpers.api import ApiHandler, Request |
| 4 | from helpers.errors import format_error |
| 5 | from helpers import plugins |
| 6 | |
| 7 | |
| 8 | PLUGIN_NAME = "_whatsapp_integration" |
| 9 | |
| 10 | |
| 11 | class TestConnection(ApiHandler): |
| 12 | |
| 13 | async def process(self, input: dict, request: Request) -> dict: |
| 14 | config = input.get("config") or plugins.get_plugin_config(PLUGIN_NAME) or {} |
| 15 | port = int(config.get("bridge_port", 3100)) |
| 16 | results: list[dict] = [] |
| 17 | |
| 18 | await self._test_bridge(port, results) |
| 19 | |
| 20 | ok = all(r["ok"] for r in results) |
| 21 | return {"success": ok, "results": results} |
| 22 | |
| 23 | async def _test_bridge(self, port: int, results: list[dict]) -> None: |
| 24 | from plugins._whatsapp_integration.helpers.wa_client import get_health |
| 25 | from plugins._whatsapp_integration.helpers.bridge_manager import get_bridge_url |
| 26 | |
| 27 | try: |
| 28 | health = await get_health(get_bridge_url(port)) |
| 29 | status = health.get("status", "unknown") |
| 30 | queue = health.get("queueLength", 0) |
| 31 | uptime = health.get("uptime", 0) |
| 32 | |
| 33 | if status == "connected": |
| 34 | results.append({ |
| 35 | "test": "WhatsApp bridge", |
| 36 | "ok": True, |
| 37 | "message": f"Connected and ready (uptime: {uptime:.0f}s, queue: {queue})", |
| 38 | }) |
| 39 | else: |
| 40 | results.append({ |
| 41 | "test": "WhatsApp bridge", |
| 42 | "ok": False, |
| 43 | "message": f"The bridge is running, but WhatsApp is not fully connected yet (status: {status}).", |
| 44 | }) |
| 45 | except Exception as e: |
| 46 | results.append({ |
| 47 | "test": "WhatsApp bridge", |
| 48 | "ok": False, |
| 49 | "message": f"Could not reach the local WhatsApp bridge: {format_error(e)}", |
| 50 | }) |