306 lines
9.0 KiB
Python
306 lines
9.0 KiB
Python
"""Example script to test email monitoring functionality.
|
|
|
|
This script demonstrates how to:
|
|
1. Configure the email service
|
|
2. Send test emails
|
|
3. Trigger error alerts
|
|
4. Test daily report generation
|
|
|
|
Usage:
|
|
uv run python examples/test_email_monitoring.py
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from alpine_bits_python.config_loader import load_config
|
|
from alpine_bits_python.email_monitoring import (
|
|
DailyReportScheduler,
|
|
EmailAlertHandler,
|
|
)
|
|
from alpine_bits_python.email_service import create_email_service
|
|
from alpine_bits_python.logging_config import get_logger, setup_logging
|
|
|
|
_LOGGER = get_logger(__name__)
|
|
|
|
|
|
async def test_basic_email():
|
|
"""Test 1: Send a basic test email."""
|
|
print("\n" + "=" * 60)
|
|
print("Test 1: Basic Email Sending")
|
|
print("=" * 60)
|
|
|
|
config = load_config()
|
|
email_service = create_email_service(config)
|
|
|
|
if not email_service:
|
|
print("❌ Email service not configured. Check your config.yaml")
|
|
return False
|
|
|
|
print("✓ Email service initialized")
|
|
|
|
# Get the first recipient from error_alerts config
|
|
email_config = config.get("email", {})
|
|
monitoring_config = email_config.get("monitoring", {})
|
|
error_alerts_config = monitoring_config.get("error_alerts", {})
|
|
recipients = error_alerts_config.get("recipients", [])
|
|
|
|
if not recipients:
|
|
print("❌ No recipients configured in error_alerts")
|
|
return False
|
|
|
|
print(f"✓ Sending test email to: {recipients[0]}")
|
|
|
|
success = await email_service.send_email(
|
|
recipients=[recipients[0]],
|
|
subject="AlpineBits Email Test - Basic",
|
|
body=f"""This is a test email from the AlpineBits server.
|
|
|
|
Timestamp: {datetime.now().isoformat()}
|
|
Test: Basic email sending
|
|
|
|
If you received this email, your SMTP configuration is working correctly!
|
|
|
|
---
|
|
AlpineBits Python Server
|
|
Email Monitoring System
|
|
""",
|
|
)
|
|
|
|
if success:
|
|
print("✅ Test email sent successfully!")
|
|
return True
|
|
else:
|
|
print("❌ Failed to send test email. Check logs for details.")
|
|
return False
|
|
|
|
|
|
async def test_error_alert_threshold():
|
|
"""Test 2: Trigger immediate error alert by exceeding threshold."""
|
|
print("\n" + "=" * 60)
|
|
print("Test 2: Error Alert - Threshold Trigger")
|
|
print("=" * 60)
|
|
|
|
config = load_config()
|
|
email_service = create_email_service(config)
|
|
|
|
if not email_service:
|
|
print("❌ Email service not configured")
|
|
return False
|
|
|
|
# Setup logging with email monitoring
|
|
loop = asyncio.get_running_loop()
|
|
email_handler, _ = setup_logging(config, email_service, loop)
|
|
|
|
if not email_handler:
|
|
print("❌ Error alert handler not configured")
|
|
return False
|
|
|
|
print(f"✓ Error alert handler configured (threshold: {email_handler.error_threshold})")
|
|
print(f" Recipients: {email_handler.recipients}")
|
|
|
|
# Generate errors to exceed threshold
|
|
threshold = email_handler.error_threshold
|
|
print(f"\n📨 Generating {threshold} errors to trigger immediate alert...")
|
|
|
|
logger = logging.getLogger("test.error.threshold")
|
|
for i in range(threshold):
|
|
logger.error(f"Test error #{i + 1} - Threshold test at {datetime.now().isoformat()}")
|
|
print(f" → Error {i + 1}/{threshold} logged")
|
|
await asyncio.sleep(0.1) # Small delay between errors
|
|
|
|
# Wait a bit for email to be sent
|
|
print("\n⏳ Waiting for alert email to be sent...")
|
|
await asyncio.sleep(3)
|
|
|
|
print("✅ Threshold test complete! Check your email for the alert.")
|
|
return True
|
|
|
|
|
|
async def test_error_alert_buffer():
|
|
"""Test 3: Trigger buffered error alert by waiting for buffer time."""
|
|
print("\n" + "=" * 60)
|
|
print("Test 3: Error Alert - Buffer Time Trigger")
|
|
print("=" * 60)
|
|
|
|
config = load_config()
|
|
email_service = create_email_service(config)
|
|
|
|
if not email_service:
|
|
print("❌ Email service not configured")
|
|
return False
|
|
|
|
# Setup logging with email monitoring
|
|
loop = asyncio.get_running_loop()
|
|
email_handler, _ = setup_logging(config, email_service, loop)
|
|
|
|
if not email_handler:
|
|
print("❌ Error alert handler not configured")
|
|
return False
|
|
|
|
print(f"✓ Error alert handler configured (buffer: {email_handler.buffer_minutes} minutes)")
|
|
|
|
# Generate fewer errors than threshold
|
|
num_errors = max(1, email_handler.error_threshold - 2)
|
|
print(f"\n📨 Generating {num_errors} errors (below threshold)...")
|
|
|
|
logger = logging.getLogger("test.error.buffer")
|
|
for i in range(num_errors):
|
|
logger.error(f"Test error #{i + 1} - Buffer test at {datetime.now().isoformat()}")
|
|
print(f" → Error {i + 1}/{num_errors} logged")
|
|
|
|
buffer_seconds = email_handler.buffer_minutes * 60
|
|
print(f"\n⏳ Waiting {email_handler.buffer_minutes} minute(s) for buffer to flush...")
|
|
print(" (This will send an email with all buffered errors)")
|
|
|
|
# Wait for buffer time + a bit extra
|
|
await asyncio.sleep(buffer_seconds + 2)
|
|
|
|
print("✅ Buffer test complete! Check your email for the alert.")
|
|
return True
|
|
|
|
|
|
async def test_daily_report():
|
|
"""Test 4: Generate and send a test daily report."""
|
|
print("\n" + "=" * 60)
|
|
print("Test 4: Daily Report")
|
|
print("=" * 60)
|
|
|
|
config = load_config()
|
|
email_service = create_email_service(config)
|
|
|
|
if not email_service:
|
|
print("❌ Email service not configured")
|
|
return False
|
|
|
|
# Create a daily report scheduler
|
|
daily_report_config = (
|
|
config.get("email", {})
|
|
.get("monitoring", {})
|
|
.get("daily_report", {})
|
|
)
|
|
|
|
if not daily_report_config.get("enabled"):
|
|
print("⚠️ Daily reports not enabled in config")
|
|
print(" Set email.monitoring.daily_report.enabled = true")
|
|
return False
|
|
|
|
scheduler = DailyReportScheduler(email_service, daily_report_config)
|
|
print(f"✓ Daily report scheduler configured")
|
|
print(f" Recipients: {scheduler.recipients}")
|
|
print(f" Send time: {scheduler.send_time}")
|
|
|
|
# Add some test statistics
|
|
test_stats = {
|
|
"total_reservations": 42,
|
|
"new_customers": 15,
|
|
"active_hotels": 4,
|
|
"api_requests_today": 1234,
|
|
"average_response_time_ms": 45,
|
|
"success_rate": "99.2%",
|
|
}
|
|
|
|
# Add some test errors
|
|
test_errors = [
|
|
{
|
|
"timestamp": "2025-10-15 08:15:23",
|
|
"level": "ERROR",
|
|
"message": "Connection timeout to external API",
|
|
},
|
|
{
|
|
"timestamp": "2025-10-15 12:45:10",
|
|
"level": "ERROR",
|
|
"message": "Invalid form data submitted",
|
|
},
|
|
{
|
|
"timestamp": "2025-10-15 18:30:00",
|
|
"level": "CRITICAL",
|
|
"message": "Database connection pool exhausted",
|
|
},
|
|
]
|
|
|
|
print("\n📊 Sending test daily report...")
|
|
print(f" Stats: {len(test_stats)} metrics")
|
|
print(f" Errors: {len(test_errors)} entries")
|
|
|
|
success = await email_service.send_daily_report(
|
|
recipients=scheduler.recipients,
|
|
stats=test_stats,
|
|
errors=test_errors,
|
|
)
|
|
|
|
if success:
|
|
print("✅ Daily report sent successfully!")
|
|
return True
|
|
else:
|
|
print("❌ Failed to send daily report. Check logs for details.")
|
|
return False
|
|
|
|
|
|
async def run_all_tests():
|
|
"""Run all email monitoring tests."""
|
|
print("\n" + "=" * 60)
|
|
print("AlpineBits Email Monitoring Test Suite")
|
|
print("=" * 60)
|
|
|
|
tests = [
|
|
("Basic Email", test_basic_email),
|
|
("Error Alert (Threshold)", test_error_alert_threshold),
|
|
("Error Alert (Buffer)", test_error_alert_buffer),
|
|
("Daily Report", test_daily_report),
|
|
]
|
|
|
|
results = []
|
|
|
|
for test_name, test_func in tests:
|
|
try:
|
|
result = await test_func()
|
|
results.append((test_name, result))
|
|
except Exception as e:
|
|
print(f"\n❌ Test '{test_name}' failed with exception: {e}")
|
|
results.append((test_name, False))
|
|
|
|
# Wait between tests to avoid rate limiting
|
|
await asyncio.sleep(2)
|
|
|
|
# Print summary
|
|
print("\n" + "=" * 60)
|
|
print("Test Summary")
|
|
print("=" * 60)
|
|
|
|
passed = sum(1 for _, result in results if result)
|
|
total = len(results)
|
|
|
|
for test_name, result in results:
|
|
status = "✅ PASS" if result else "❌ FAIL"
|
|
print(f"{status}: {test_name}")
|
|
|
|
print(f"\nTotal: {passed}/{total} tests passed")
|
|
|
|
if passed == total:
|
|
print("\n🎉 All tests passed!")
|
|
else:
|
|
print(f"\n⚠️ {total - passed} test(s) failed")
|
|
|
|
|
|
def main():
|
|
"""Main entry point."""
|
|
print("Starting email monitoring tests...")
|
|
print("Make sure you have configured email settings in config.yaml")
|
|
print("and set EMAIL_USERNAME and EMAIL_PASSWORD environment variables.")
|
|
|
|
# Run the tests
|
|
try:
|
|
asyncio.run(run_all_tests())
|
|
except KeyboardInterrupt:
|
|
print("\n\n⚠️ Tests interrupted by user")
|
|
except Exception as e:
|
|
print(f"\n\n❌ Fatal error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|