| 1 | """Test inbound and outbound email connectivity for a handler config.""" |
| 2 | |
| 3 | import asyncio |
| 4 | |
| 5 | from helpers.api import ApiHandler, Request |
| 6 | from helpers.errors import format_error |
| 7 | |
| 8 | from plugins._email_integration.helpers.imap_client import ( |
| 9 | connect_exchange, |
| 10 | connect_imap, |
| 11 | disconnect_imap, |
| 12 | get_highest_uid, |
| 13 | ) |
| 14 | from plugins._email_integration.helpers.smtp_client import SmtpConfig, test_smtp, send_reply |
| 15 | |
| 16 | |
| 17 | class TestConnection(ApiHandler): |
| 18 | |
| 19 | async def process(self, input: dict, request: Request) -> dict: |
| 20 | handler = input.get("handler", {}) |
| 21 | results: list[dict] = [] |
| 22 | account_type = handler.get("account_type", "imap") |
| 23 | |
| 24 | if account_type == "exchange": |
| 25 | await self._test_exchange(handler, results) |
| 26 | else: |
| 27 | await self._test_imap(handler, results) |
| 28 | await self._test_smtp(handler, results) |
| 29 | if all(r["ok"] for r in results): |
| 30 | await self._test_send(handler, results) |
| 31 | |
| 32 | ok = all(r["ok"] for r in results) |
| 33 | return {"success": ok, "results": results} |
| 34 | |
| 35 | async def _test_imap(self, handler: dict, results: list[dict]): |
| 36 | try: |
| 37 | client = await connect_imap( |
| 38 | server=handler.get("imap_server", ""), |
| 39 | port=int(handler.get("imap_port", 993)), |
| 40 | username=handler.get("username", ""), |
| 41 | password=handler.get("password", ""), |
| 42 | ) |
| 43 | await get_highest_uid(client) |
| 44 | await disconnect_imap(client) |
| 45 | results.append({ |
| 46 | "test": "Incoming", |
| 47 | "ok": True, |
| 48 | "message": "Incoming mail looks good.", |
| 49 | }) |
| 50 | except Exception as e: |
| 51 | results.append({ |
| 52 | "test": "Incoming", |
| 53 | "ok": False, |
| 54 | "message": f"Could not reach the inbox: {format_error(e)}", |
| 55 | }) |
| 56 | |
| 57 | async def _test_exchange(self, handler: dict, results: list[dict]): |
| 58 | try: |
| 59 | account = await connect_exchange( |
| 60 | server=handler.get("imap_server", ""), |
| 61 | username=handler.get("username", ""), |
| 62 | password=handler.get("password", ""), |
| 63 | ) |
| 64 | loop = asyncio.get_event_loop() |
| 65 | await loop.run_in_executor(None, lambda: account.inbox.total_count) |
| 66 | results.append({ |
| 67 | "test": "Incoming", |
| 68 | "ok": True, |
| 69 | "message": "Exchange inbox looks good.", |
| 70 | }) |
| 71 | except Exception as e: |
| 72 | results.append({ |
| 73 | "test": "Incoming", |
| 74 | "ok": False, |
| 75 | "message": f"Could not reach the Exchange inbox: {format_error(e)}", |
| 76 | }) |
| 77 | |
| 78 | def _smtp_config(self, handler: dict) -> SmtpConfig: |
| 79 | smtp_server = handler.get("smtp_server") or handler.get("imap_server", "") |
| 80 | return SmtpConfig( |
| 81 | server=smtp_server, |
| 82 | port=int(handler.get("smtp_port", 587)), |
| 83 | username=handler.get("username", ""), |
| 84 | password=handler.get("password", ""), |
| 85 | ) |
| 86 | |
| 87 | async def _test_smtp(self, handler: dict, results: list[dict]): |
| 88 | try: |
| 89 | error = await test_smtp(self._smtp_config(handler)) |
| 90 | if error: |
| 91 | results.append({ |
| 92 | "test": "Outgoing", |
| 93 | "ok": False, |
| 94 | "message": f"Could not sign in for sending mail: {error}", |
| 95 | }) |
| 96 | else: |
| 97 | results.append({ |
| 98 | "test": "Outgoing", |
| 99 | "ok": True, |
| 100 | "message": "Outgoing mail looks good.", |
| 101 | }) |
| 102 | except Exception as e: |
| 103 | results.append({ |
| 104 | "test": "Outgoing", |
| 105 | "ok": False, |
| 106 | "message": f"Could not sign in for sending mail: {format_error(e)}", |
| 107 | }) |
| 108 | |
| 109 | async def _test_send(self, handler: dict, results: list[dict]): |
| 110 | try: |
| 111 | error = await send_reply( |
| 112 | config=self._smtp_config(handler), |
| 113 | to=handler.get("username", ""), |
| 114 | subject="Agent Zero - Connection Test", |
| 115 | body="This is a test email from Agent Zero email integration.", |
| 116 | ) |
| 117 | if error: |
| 118 | results.append({ |
| 119 | "test": "Send test email", |
| 120 | "ok": False, |
| 121 | "message": f"Could not send the test email: {error}", |
| 122 | }) |
| 123 | else: |
| 124 | results.append({ |
| 125 | "test": "Send test email", |
| 126 | "ok": True, |
| 127 | "message": "Test email sent to this inbox.", |
| 128 | }) |
| 129 | except Exception as e: |
| 130 | results.append({ |
| 131 | "test": "Send test email", |
| 132 | "ok": False, |
| 133 | "message": f"Could not send the test email: {format_error(e)}", |
| 134 | }) |