286 lines
11 KiB
Python
286 lines
11 KiB
Python
|
"""
|
||
|
|
Unit tests for AlertManager.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from unittest.mock import Mock, AsyncMock, patch
|
||
|
|
|
||
|
|
from src.monitoring.alerts import AlertManager
|
||
|
|
from config.settings import Settings
|
||
|
|
|
||
|
|
|
||
|
|
class TestAlertManager:
|
||
|
|
"""Test cases for AlertManager."""
|
||
|
|
|
||
|
|
def setup_method(self):
|
||
|
|
"""Set up test fixtures."""
|
||
|
|
self.settings = Settings()
|
||
|
|
self.alert_manager = AlertManager(self.settings)
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_alert_success(self):
|
||
|
|
"""Test sending alert successfully."""
|
||
|
|
# Arrange
|
||
|
|
with patch.object(self.alert_manager, '_send_email_alert', AsyncMock(return_value=True)) as mock_email,\
|
||
|
|
patch.object(self.alert_manager, '_send_sms_alert', AsyncMock(return_value=True)) as mock_sms,\
|
||
|
|
patch.object(self.alert_manager, '_send_webhook_alert', AsyncMock(return_value=True)) as mock_webhook,\
|
||
|
|
patch.object(self.alert_manager, '_send_scada_alert', AsyncMock(return_value=True)) as mock_scada:
|
||
|
|
|
||
|
|
# Act
|
||
|
|
result = await self.alert_manager.send_alert(
|
||
|
|
alert_type='SAFETY_VIOLATION',
|
||
|
|
severity='ERROR',
|
||
|
|
message='Test safety violation',
|
||
|
|
context={'violation_type': 'OVERSPEED'},
|
||
|
|
station_id='STATION_001',
|
||
|
|
pump_id='PUMP_001'
|
||
|
|
)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert result is True
|
||
|
|
assert mock_email.called
|
||
|
|
assert mock_sms.called
|
||
|
|
assert mock_webhook.called
|
||
|
|
assert mock_scada.called
|
||
|
|
|
||
|
|
# Check alert history
|
||
|
|
history = self.alert_manager.get_alert_history()
|
||
|
|
assert len(history) == 1
|
||
|
|
assert history[0]['alert_type'] == 'SAFETY_VIOLATION'
|
||
|
|
assert history[0]['severity'] == 'ERROR'
|
||
|
|
assert history[0]['station_id'] == 'STATION_001'
|
||
|
|
assert history[0]['pump_id'] == 'PUMP_001'
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_alert_partial_failure(self):
|
||
|
|
"""Test sending alert with partial channel failures."""
|
||
|
|
# Arrange
|
||
|
|
with patch.object(self.alert_manager, '_send_email_alert', AsyncMock(return_value=True)) as mock_email,\
|
||
|
|
patch.object(self.alert_manager, '_send_sms_alert', AsyncMock(return_value=False)) as mock_sms,\
|
||
|
|
patch.object(self.alert_manager, '_send_webhook_alert', AsyncMock(return_value=False)) as mock_webhook,\
|
||
|
|
patch.object(self.alert_manager, '_send_scada_alert', AsyncMock(return_value=True)) as mock_scada:
|
||
|
|
|
||
|
|
# Act
|
||
|
|
result = await self.alert_manager.send_alert(
|
||
|
|
alert_type='FAILSAFE_ACTIVATED',
|
||
|
|
severity='CRITICAL',
|
||
|
|
message='Test failsafe activation'
|
||
|
|
)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert result is True # Should still return True if at least one channel succeeded
|
||
|
|
assert mock_email.called
|
||
|
|
assert mock_sms.called
|
||
|
|
assert mock_webhook.called
|
||
|
|
assert mock_scada.called
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_alert_all_failures(self):
|
||
|
|
"""Test sending alert when all channels fail."""
|
||
|
|
# Arrange
|
||
|
|
with patch.object(self.alert_manager, '_send_email_alert', AsyncMock(return_value=False)) as mock_email,\
|
||
|
|
patch.object(self.alert_manager, '_send_sms_alert', AsyncMock(return_value=False)) as mock_sms,\
|
||
|
|
patch.object(self.alert_manager, '_send_webhook_alert', AsyncMock(return_value=False)) as mock_webhook,\
|
||
|
|
patch.object(self.alert_manager, '_send_scada_alert', AsyncMock(return_value=False)) as mock_scada:
|
||
|
|
|
||
|
|
# Act
|
||
|
|
result = await self.alert_manager.send_alert(
|
||
|
|
alert_type='SYSTEM_ERROR',
|
||
|
|
severity='ERROR',
|
||
|
|
message='Test system error'
|
||
|
|
)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert result is False # Should return False if all channels failed
|
||
|
|
assert mock_email.called
|
||
|
|
assert mock_sms.called
|
||
|
|
assert mock_webhook.called
|
||
|
|
assert mock_scada.called
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_email_alert_success(self):
|
||
|
|
"""Test sending email alert successfully."""
|
||
|
|
# Arrange
|
||
|
|
alert_data = {
|
||
|
|
'alert_type': 'TEST_ALERT',
|
||
|
|
'severity': 'INFO',
|
||
|
|
'message': 'Test message',
|
||
|
|
'context': {},
|
||
|
|
'app_name': 'Test App',
|
||
|
|
'app_version': '1.0.0',
|
||
|
|
'timestamp': 1234567890.0
|
||
|
|
}
|
||
|
|
|
||
|
|
with patch('smtplib.SMTP') as mock_smtp:
|
||
|
|
mock_server = Mock()
|
||
|
|
mock_smtp.return_value.__enter__.return_value = mock_server
|
||
|
|
|
||
|
|
# Act
|
||
|
|
result = await self.alert_manager._send_email_alert(alert_data)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert result is True
|
||
|
|
assert mock_smtp.called
|
||
|
|
assert mock_server.send_message.called
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_email_alert_failure(self):
|
||
|
|
"""Test sending email alert with failure."""
|
||
|
|
# Arrange
|
||
|
|
alert_data = {
|
||
|
|
'alert_type': 'TEST_ALERT',
|
||
|
|
'severity': 'INFO',
|
||
|
|
'message': 'Test message',
|
||
|
|
'context': {},
|
||
|
|
'app_name': 'Test App',
|
||
|
|
'app_version': '1.0.0'
|
||
|
|
}
|
||
|
|
|
||
|
|
with patch('smtplib.SMTP', side_effect=Exception("SMTP error")):
|
||
|
|
# Act
|
||
|
|
result = await self.alert_manager._send_email_alert(alert_data)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert result is False
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_sms_alert_critical_only(self):
|
||
|
|
"""Test that SMS alerts are only sent for critical events."""
|
||
|
|
# Arrange
|
||
|
|
alert_data_critical = {
|
||
|
|
'alert_type': 'CRITICAL_ALERT',
|
||
|
|
'severity': 'CRITICAL',
|
||
|
|
'message': 'Critical message'
|
||
|
|
}
|
||
|
|
|
||
|
|
alert_data_info = {
|
||
|
|
'alert_type': 'INFO_ALERT',
|
||
|
|
'severity': 'INFO',
|
||
|
|
'message': 'Info message'
|
||
|
|
}
|
||
|
|
|
||
|
|
# Act - Critical alert
|
||
|
|
result_critical = await self.alert_manager._send_sms_alert(alert_data_critical)
|
||
|
|
|
||
|
|
# Act - Info alert
|
||
|
|
result_info = await self.alert_manager._send_sms_alert(alert_data_info)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert result_critical is True # Should attempt to send critical alerts
|
||
|
|
assert result_info is False # Should not send non-critical alerts
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_webhook_alert_success(self):
|
||
|
|
"""Test sending webhook alert successfully."""
|
||
|
|
# Arrange
|
||
|
|
alert_data = {
|
||
|
|
'alert_type': 'TEST_ALERT',
|
||
|
|
'severity': 'INFO',
|
||
|
|
'message': 'Test message'
|
||
|
|
}
|
||
|
|
|
||
|
|
with patch('aiohttp.ClientSession.post') as mock_post:
|
||
|
|
mock_response = AsyncMock()
|
||
|
|
mock_response.status = 200
|
||
|
|
mock_post.return_value.__aenter__.return_value = mock_response
|
||
|
|
|
||
|
|
# Act
|
||
|
|
result = await self.alert_manager._send_webhook_alert(alert_data)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert result is True
|
||
|
|
assert mock_post.called
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_webhook_alert_failure(self):
|
||
|
|
"""Test sending webhook alert with failure."""
|
||
|
|
# Arrange
|
||
|
|
alert_data = {
|
||
|
|
'alert_type': 'TEST_ALERT',
|
||
|
|
'severity': 'INFO',
|
||
|
|
'message': 'Test message'
|
||
|
|
}
|
||
|
|
|
||
|
|
with patch('aiohttp.ClientSession.post') as mock_post:
|
||
|
|
mock_response = AsyncMock()
|
||
|
|
mock_response.status = 500
|
||
|
|
mock_post.return_value.__aenter__.return_value = mock_response
|
||
|
|
|
||
|
|
# Act
|
||
|
|
result = await self.alert_manager._send_webhook_alert(alert_data)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert result is False
|
||
|
|
assert mock_post.called
|
||
|
|
|
||
|
|
def test_format_email_body(self):
|
||
|
|
"""Test formatting email body."""
|
||
|
|
# Arrange
|
||
|
|
alert_data = {
|
||
|
|
'alert_type': 'SAFETY_VIOLATION',
|
||
|
|
'severity': 'ERROR',
|
||
|
|
'message': 'Speed limit exceeded',
|
||
|
|
'context': {'requested_speed': 55.0, 'max_speed': 50.0},
|
||
|
|
'station_id': 'STATION_001',
|
||
|
|
'pump_id': 'PUMP_001',
|
||
|
|
'timestamp': 1234567890.0,
|
||
|
|
'app_name': 'Test App',
|
||
|
|
'app_version': '1.0.0'
|
||
|
|
}
|
||
|
|
|
||
|
|
# Act
|
||
|
|
body = self.alert_manager._format_email_body(alert_data)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert 'SAFETY_VIOLATION' in body
|
||
|
|
assert 'ERROR' in body
|
||
|
|
assert 'Speed limit exceeded' in body
|
||
|
|
assert 'STATION_001' in body
|
||
|
|
assert 'PUMP_001' in body
|
||
|
|
assert 'requested_speed' in body
|
||
|
|
assert 'Test App v1.0.0' in body
|
||
|
|
|
||
|
|
def test_alert_history_management(self):
|
||
|
|
"""Test alert history management with size limits."""
|
||
|
|
# Arrange - Fill history beyond limit
|
||
|
|
for i in range(1500): # More than max_history_size (1000)
|
||
|
|
self.alert_manager._store_alert_history({
|
||
|
|
'alert_type': f'TEST_{i}',
|
||
|
|
'severity': 'INFO',
|
||
|
|
'message': f'Test message {i}'
|
||
|
|
})
|
||
|
|
|
||
|
|
# Act - Get all history (no limit)
|
||
|
|
history = self.alert_manager.get_alert_history(limit=2000)
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert len(history) == 1000 # Should be limited to max_history_size
|
||
|
|
assert history[0]['alert_type'] == 'TEST_500' # Should keep most recent
|
||
|
|
assert history[-1]['alert_type'] == 'TEST_1499' # Most recent at end
|
||
|
|
|
||
|
|
def test_get_alert_stats(self):
|
||
|
|
"""Test getting alert statistics."""
|
||
|
|
# Arrange
|
||
|
|
alerts = [
|
||
|
|
{'alert_type': 'SAFETY_VIOLATION', 'severity': 'ERROR'},
|
||
|
|
{'alert_type': 'SAFETY_VIOLATION', 'severity': 'ERROR'},
|
||
|
|
{'alert_type': 'FAILSAFE_ACTIVATED', 'severity': 'CRITICAL'},
|
||
|
|
{'alert_type': 'SYSTEM_ERROR', 'severity': 'ERROR'},
|
||
|
|
{'alert_type': 'INFO_ALERT', 'severity': 'INFO'}
|
||
|
|
]
|
||
|
|
|
||
|
|
for alert in alerts:
|
||
|
|
self.alert_manager._store_alert_history(alert)
|
||
|
|
|
||
|
|
# Act
|
||
|
|
stats = self.alert_manager.get_alert_stats()
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert stats['total_alerts'] == 5
|
||
|
|
assert stats['severity_counts']['ERROR'] == 3
|
||
|
|
assert stats['severity_counts']['CRITICAL'] == 1
|
||
|
|
assert stats['severity_counts']['INFO'] == 1
|
||
|
|
assert stats['type_counts']['SAFETY_VIOLATION'] == 2
|
||
|
|
assert stats['type_counts']['FAILSAFE_ACTIVATED'] == 1
|
||
|
|
assert stats['type_counts']['SYSTEM_ERROR'] == 1
|
||
|
|
assert stats['type_counts']['INFO_ALERT'] == 1
|