374 lines
12 KiB
Python
374 lines
12 KiB
Python
"""Tests for email service and monitoring functionality."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from alpine_bits_python.email_monitoring import (
|
|
DailyReportScheduler,
|
|
EmailAlertHandler,
|
|
ErrorRecord,
|
|
)
|
|
from alpine_bits_python.email_service import EmailConfig, EmailService
|
|
|
|
|
|
class TestEmailConfig:
|
|
"""Tests for EmailConfig class."""
|
|
|
|
def test_email_config_initialization(self):
|
|
"""Test basic email configuration initialization."""
|
|
config = {
|
|
"smtp": {
|
|
"host": "smtp.example.com",
|
|
"port": 587,
|
|
"username": "test@example.com",
|
|
"password": "password123",
|
|
"use_tls": True,
|
|
"use_ssl": False,
|
|
},
|
|
"from_address": "sender@example.com",
|
|
"from_name": "Test Sender",
|
|
}
|
|
|
|
email_config = EmailConfig(config)
|
|
|
|
assert email_config.smtp_host == "smtp.example.com"
|
|
assert email_config.smtp_port == 587
|
|
assert email_config.smtp_username == "test@example.com"
|
|
assert email_config.smtp_password == "password123"
|
|
assert email_config.use_tls is True
|
|
assert email_config.use_ssl is False
|
|
assert email_config.from_address == "sender@example.com"
|
|
assert email_config.from_name == "Test Sender"
|
|
|
|
def test_email_config_defaults(self):
|
|
"""Test email configuration with default values."""
|
|
config = {}
|
|
|
|
email_config = EmailConfig(config)
|
|
|
|
assert email_config.smtp_host == "localhost"
|
|
assert email_config.smtp_port == 587
|
|
assert email_config.use_tls is True
|
|
assert email_config.use_ssl is False
|
|
assert email_config.from_address == "noreply@example.com"
|
|
|
|
def test_email_config_tls_ssl_conflict(self):
|
|
"""Test that TLS and SSL cannot both be enabled."""
|
|
config = {
|
|
"smtp": {
|
|
"use_tls": True,
|
|
"use_ssl": True,
|
|
}
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="Cannot use both TLS and SSL"):
|
|
EmailConfig(config)
|
|
|
|
|
|
class TestEmailService:
|
|
"""Tests for EmailService class."""
|
|
|
|
@pytest.fixture
|
|
def email_config(self):
|
|
"""Provide a test email configuration."""
|
|
return EmailConfig(
|
|
{
|
|
"smtp": {
|
|
"host": "smtp.example.com",
|
|
"port": 587,
|
|
"username": "test@example.com",
|
|
"password": "password123",
|
|
"use_tls": True,
|
|
},
|
|
"from_address": "sender@example.com",
|
|
"from_name": "Test Sender",
|
|
}
|
|
)
|
|
|
|
@pytest.fixture
|
|
def email_service(self, email_config):
|
|
"""Provide an EmailService instance."""
|
|
return EmailService(email_config)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_success(self, email_service):
|
|
"""Test successful email sending."""
|
|
with patch.object(email_service, "_send_smtp") as mock_smtp:
|
|
result = await email_service.send_email(
|
|
recipients=["test@example.com"],
|
|
subject="Test Subject",
|
|
body="Test body",
|
|
)
|
|
|
|
assert result is True
|
|
assert mock_smtp.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_no_recipients(self, email_service):
|
|
"""Test email sending with no recipients."""
|
|
result = await email_service.send_email(
|
|
recipients=[],
|
|
subject="Test Subject",
|
|
body="Test body",
|
|
)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_with_html(self, email_service):
|
|
"""Test email sending with HTML body."""
|
|
with patch.object(email_service, "_send_smtp") as mock_smtp:
|
|
result = await email_service.send_email(
|
|
recipients=["test@example.com"],
|
|
subject="Test Subject",
|
|
body="Plain text body",
|
|
html_body="<html><body>HTML body</body></html>",
|
|
)
|
|
|
|
assert result is True
|
|
assert mock_smtp.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_alert(self, email_service):
|
|
"""Test send_alert convenience method."""
|
|
with patch.object(email_service, "send_email", new_callable=AsyncMock) as mock_send:
|
|
mock_send.return_value = True
|
|
|
|
result = await email_service.send_alert(
|
|
recipients=["test@example.com"],
|
|
subject="Alert Subject",
|
|
body="Alert body",
|
|
)
|
|
|
|
assert result is True
|
|
mock_send.assert_called_once_with(
|
|
["test@example.com"], "Alert Subject", "Alert body"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_daily_report(self, email_service):
|
|
"""Test daily report email generation and sending."""
|
|
with patch.object(email_service, "send_email", new_callable=AsyncMock) as mock_send:
|
|
mock_send.return_value = True
|
|
|
|
stats = {
|
|
"total_reservations": 42,
|
|
"new_customers": 15,
|
|
}
|
|
|
|
errors = [
|
|
{
|
|
"timestamp": "2025-10-15 10:30:00",
|
|
"level": "ERROR",
|
|
"message": "Test error message",
|
|
}
|
|
]
|
|
|
|
result = await email_service.send_daily_report(
|
|
recipients=["admin@example.com"],
|
|
stats=stats,
|
|
errors=errors,
|
|
)
|
|
|
|
assert result is True
|
|
assert mock_send.called
|
|
call_args = mock_send.call_args
|
|
assert "admin@example.com" in call_args[0][0]
|
|
assert "Daily Report" in call_args[0][1]
|
|
|
|
|
|
class TestErrorRecord:
|
|
"""Tests for ErrorRecord class."""
|
|
|
|
def test_error_record_creation(self):
|
|
"""Test creating an ErrorRecord from a logging record."""
|
|
log_record = logging.LogRecord(
|
|
name="test.logger",
|
|
level=logging.ERROR,
|
|
pathname="/path/to/file.py",
|
|
lineno=42,
|
|
msg="Test error message",
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
|
|
error_record = ErrorRecord(log_record)
|
|
|
|
assert error_record.level == "ERROR"
|
|
assert error_record.logger_name == "test.logger"
|
|
assert error_record.message == "Test error message"
|
|
assert error_record.module == "file"
|
|
assert error_record.line_no == 42
|
|
|
|
def test_error_record_to_dict(self):
|
|
"""Test converting ErrorRecord to dictionary."""
|
|
log_record = logging.LogRecord(
|
|
name="test.logger",
|
|
level=logging.ERROR,
|
|
pathname="/path/to/file.py",
|
|
lineno=42,
|
|
msg="Test error",
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
|
|
error_record = ErrorRecord(log_record)
|
|
error_dict = error_record.to_dict()
|
|
|
|
assert error_dict["level"] == "ERROR"
|
|
assert error_dict["message"] == "Test error"
|
|
assert error_dict["line_no"] == 42
|
|
assert "timestamp" in error_dict
|
|
|
|
def test_error_record_format_plain_text(self):
|
|
"""Test formatting ErrorRecord as plain text."""
|
|
log_record = logging.LogRecord(
|
|
name="test.logger",
|
|
level=logging.ERROR,
|
|
pathname="/path/to/file.py",
|
|
lineno=42,
|
|
msg="Test error",
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
|
|
error_record = ErrorRecord(log_record)
|
|
formatted = error_record.format_plain_text()
|
|
|
|
assert "ERROR" in formatted
|
|
assert "Test error" in formatted
|
|
assert "file:42" in formatted
|
|
|
|
|
|
class TestEmailAlertHandler:
|
|
"""Tests for EmailAlertHandler class."""
|
|
|
|
@pytest.fixture
|
|
def mock_email_service(self):
|
|
"""Provide a mock email service."""
|
|
service = MagicMock(spec=EmailService)
|
|
service.send_alert = AsyncMock(return_value=True)
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def handler_config(self):
|
|
"""Provide handler configuration."""
|
|
return {
|
|
"recipients": ["alert@example.com"],
|
|
"error_threshold": 3,
|
|
"buffer_minutes": 1, # Short for testing
|
|
"cooldown_minutes": 5,
|
|
"log_levels": ["ERROR", "CRITICAL"],
|
|
}
|
|
|
|
@pytest.fixture
|
|
def alert_handler(self, mock_email_service, handler_config):
|
|
"""Provide an EmailAlertHandler instance."""
|
|
loop = asyncio.new_event_loop()
|
|
handler = EmailAlertHandler(mock_email_service, handler_config, loop)
|
|
yield handler
|
|
loop.close()
|
|
|
|
def test_handler_initialization(self, alert_handler, handler_config):
|
|
"""Test handler initialization."""
|
|
assert alert_handler.error_threshold == 3
|
|
assert alert_handler.buffer_minutes == 1
|
|
assert alert_handler.cooldown_minutes == 5
|
|
assert alert_handler.recipients == ["alert@example.com"]
|
|
|
|
def test_handler_emit_below_threshold(self, alert_handler):
|
|
"""Test emitting errors below threshold."""
|
|
log_record = logging.LogRecord(
|
|
name="test",
|
|
level=logging.ERROR,
|
|
pathname="/test.py",
|
|
lineno=1,
|
|
msg="Error 1",
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
|
|
# Emit 2 errors (below threshold of 3)
|
|
alert_handler.emit(log_record)
|
|
alert_handler.emit(log_record)
|
|
|
|
# Should buffer, not send immediately
|
|
assert len(alert_handler.error_buffer) == 2
|
|
|
|
def test_handler_ignores_non_error_levels(self, alert_handler):
|
|
"""Test that handler ignores INFO and WARNING levels."""
|
|
log_record = logging.LogRecord(
|
|
name="test",
|
|
level=logging.INFO,
|
|
pathname="/test.py",
|
|
lineno=1,
|
|
msg="Info message",
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
|
|
alert_handler.emit(log_record)
|
|
|
|
# Should not buffer INFO messages
|
|
assert len(alert_handler.error_buffer) == 0
|
|
|
|
|
|
class TestDailyReportScheduler:
|
|
"""Tests for DailyReportScheduler class."""
|
|
|
|
@pytest.fixture
|
|
def mock_email_service(self):
|
|
"""Provide a mock email service."""
|
|
service = MagicMock(spec=EmailService)
|
|
service.send_daily_report = AsyncMock(return_value=True)
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def scheduler_config(self):
|
|
"""Provide scheduler configuration."""
|
|
return {
|
|
"recipients": ["report@example.com"],
|
|
"send_time": "08:00",
|
|
"include_stats": True,
|
|
"include_errors": True,
|
|
}
|
|
|
|
@pytest.fixture
|
|
def scheduler(self, mock_email_service, scheduler_config):
|
|
"""Provide a DailyReportScheduler instance."""
|
|
return DailyReportScheduler(mock_email_service, scheduler_config)
|
|
|
|
def test_scheduler_initialization(self, scheduler, scheduler_config):
|
|
"""Test scheduler initialization."""
|
|
assert scheduler.send_time == "08:00"
|
|
assert scheduler.recipients == ["report@example.com"]
|
|
assert scheduler.include_stats is True
|
|
assert scheduler.include_errors is True
|
|
|
|
def test_scheduler_log_error(self, scheduler):
|
|
"""Test logging errors for daily report."""
|
|
error = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"level": "ERROR",
|
|
"message": "Test error",
|
|
}
|
|
|
|
scheduler.log_error(error)
|
|
|
|
assert len(scheduler._error_log) == 1
|
|
assert scheduler._error_log[0]["message"] == "Test error"
|
|
|
|
def test_scheduler_set_stats_collector(self, scheduler):
|
|
"""Test setting stats collector function."""
|
|
|
|
async def mock_collector():
|
|
return {"test": "stats"}
|
|
|
|
scheduler.set_stats_collector(mock_collector)
|
|
|
|
assert scheduler._stats_collector is mock_collector
|