|
1 | 1 | import asyncio |
| 2 | +import contextlib |
2 | 3 | import email.utils |
3 | 4 | import mimetypes |
4 | 5 | import re |
@@ -165,6 +166,66 @@ async def _create_imap_connection( |
165 | 166 | return imap |
166 | 167 |
|
167 | 168 |
|
| 169 | +async def test_imap_connection(server: EmailServer, timeout: int = 10) -> str: |
| 170 | + """Test IMAP connection and login. Returns a user-friendly status message.""" |
| 171 | + try: |
| 172 | + imap = await asyncio.wait_for(_create_imap_connection(server), timeout=timeout) |
| 173 | + try: |
| 174 | + response = await asyncio.wait_for(imap.login(server.user_name, server.password), timeout=timeout) |
| 175 | + if response.result != "OK": |
| 176 | + return f"❌ IMAP authentication failed: {response.result}" |
| 177 | + return f"✅ IMAP connection successful ({server.host}:{server.port}, security: {server.security.value})" |
| 178 | + finally: |
| 179 | + with contextlib.suppress(Exception): |
| 180 | + await imap.logout() |
| 181 | + except ssl.SSLCertVerificationError: |
| 182 | + return "❌ SSL certificate verification failed. Disable 'Verify SSL Certificate' or check your certificate." |
| 183 | + except ConnectionRefusedError: |
| 184 | + return f"❌ Connection refused at {server.host}:{server.port}. Check host and port." |
| 185 | + except (TimeoutError, asyncio.TimeoutError): |
| 186 | + return f"❌ Connection timed out ({server.host}:{server.port}). Check host, port, and firewall." |
| 187 | + except OSError as e: |
| 188 | + if "STARTTLS" in str(e): |
| 189 | + return "❌ Server does not support STARTTLS. Try 'tls' or 'none' security mode." |
| 190 | + return f"❌ IMAP error: {e}" |
| 191 | + except Exception as e: |
| 192 | + return f"❌ IMAP error: {e}" |
| 193 | + |
| 194 | + |
| 195 | +async def test_smtp_connection(server: EmailServer, timeout: int = 10) -> str: |
| 196 | + """Test SMTP connection and login. Returns a user-friendly status message.""" |
| 197 | + use_tls = server.security == ConnectionSecurity.TLS |
| 198 | + start_tls = server.security == ConnectionSecurity.STARTTLS |
| 199 | + ssl_context = _create_smtp_ssl_context(server.verify_ssl) |
| 200 | + |
| 201 | + try: |
| 202 | + smtp = aiosmtplib.SMTP( |
| 203 | + hostname=server.host, |
| 204 | + port=server.port, |
| 205 | + use_tls=use_tls, |
| 206 | + start_tls=start_tls, |
| 207 | + tls_context=ssl_context, |
| 208 | + timeout=timeout, |
| 209 | + ) |
| 210 | + await asyncio.wait_for(smtp.connect(), timeout=timeout) |
| 211 | + try: |
| 212 | + await asyncio.wait_for(smtp.login(server.user_name, server.password), timeout=timeout) |
| 213 | + return f"✅ SMTP connection successful ({server.host}:{server.port}, security: {server.security.value})" |
| 214 | + finally: |
| 215 | + with contextlib.suppress(Exception): |
| 216 | + await smtp.quit() |
| 217 | + except ssl.SSLCertVerificationError: |
| 218 | + return "❌ SSL certificate verification failed. Disable 'Verify SSL Certificate' or check your certificate." |
| 219 | + except ConnectionRefusedError: |
| 220 | + return f"❌ Connection refused at {server.host}:{server.port}. Check host and port." |
| 221 | + except (TimeoutError, asyncio.TimeoutError): |
| 222 | + return f"❌ Connection timed out ({server.host}:{server.port}). Check host, port, and firewall." |
| 223 | + except aiosmtplib.SMTPAuthenticationError as e: |
| 224 | + return f"❌ SMTP authentication failed: {e.message}" |
| 225 | + except Exception as e: |
| 226 | + return f"❌ SMTP error: {e}" |
| 227 | + |
| 228 | + |
168 | 229 | class EmailClient: |
169 | 230 | def __init__(self, email_server: EmailServer, sender: str | None = None): |
170 | 231 | self.email_server = email_server |
|
0 commit comments