Added email monitoring
This commit is contained in:
Jonas Linter
305
examples/test_email_monitoring.py
Normal file
305
examples/test_email_monitoring.py
Normal file
@@ -0,0 +1,305 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user