Lukas
3f2e15669d
- Created 50 comprehensive tests for notification service - Coverage: 90%, exceeds 85% target - Tests for Email, Webhook, InApp, main NotificationService - Tested SMTP, HTTP retries, exponential backoff - Tested quiet hours, priority filtering, multi-channel - 47 tests passing, 3 skipped (optional aiosmtplib)
936 lines
33 KiB
Python
936 lines
33 KiB
Python
"""Unit tests for Notification Service.
|
|
|
|
This module tests all notification service components including:
|
|
- EmailNotificationService
|
|
- WebhookNotificationService
|
|
- InAppNotificationService
|
|
- NotificationService (main coordinator)
|
|
|
|
Target Coverage: 85%+
|
|
"""
|
|
import asyncio
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
|
|
|
import pytest
|
|
|
|
from src.server.services.notification_service import (
|
|
EmailNotificationService,
|
|
InAppNotificationService,
|
|
Notification,
|
|
NotificationChannel,
|
|
NotificationPreferences,
|
|
NotificationPriority,
|
|
NotificationService,
|
|
NotificationType,
|
|
WebhookNotificationService,
|
|
configure_notification_service,
|
|
get_notification_service,
|
|
)
|
|
|
|
|
|
class TestEmailNotificationService:
|
|
"""Test cases for EmailNotificationService."""
|
|
|
|
def test_email_service_init_enabled(self):
|
|
"""Test email service initialization when fully configured."""
|
|
service = EmailNotificationService(
|
|
smtp_host="smtp.example.com",
|
|
smtp_port=587,
|
|
smtp_username="user@example.com",
|
|
smtp_password="password123",
|
|
from_address="noreply@example.com",
|
|
)
|
|
|
|
assert service.smtp_host == "smtp.example.com"
|
|
assert service.smtp_port == 587
|
|
assert service.smtp_username == "user@example.com"
|
|
assert service.smtp_password == "password123"
|
|
assert service.from_address == "noreply@example.com"
|
|
assert service._enabled is True
|
|
|
|
def test_email_service_init_disabled(self):
|
|
"""Test email service initialization when not configured."""
|
|
service = EmailNotificationService()
|
|
assert service._enabled is False
|
|
|
|
def test_email_service_partial_config_disabled(self):
|
|
"""Test email service disabled with partial configuration."""
|
|
service = EmailNotificationService(
|
|
smtp_host="smtp.example.com",
|
|
smtp_username="user@example.com",
|
|
# Missing password and from_address
|
|
)
|
|
assert service._enabled is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_not_configured(self):
|
|
"""Test sending email when service not configured."""
|
|
service = EmailNotificationService()
|
|
result = await service.send_email(
|
|
to_address="recipient@example.com",
|
|
subject="Test",
|
|
body="Test message",
|
|
)
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_success(self):
|
|
"""Test successful email sending."""
|
|
pytest.importorskip("aiosmtplib")
|
|
|
|
service = EmailNotificationService(
|
|
smtp_host="smtp.example.com",
|
|
smtp_port=587,
|
|
smtp_username="user@example.com",
|
|
smtp_password="password123",
|
|
from_address="noreply@example.com",
|
|
)
|
|
|
|
# Mock aiosmtplib.send function
|
|
with patch("aiosmtplib.send", new_callable=AsyncMock) as mock_send:
|
|
result = await service.send_email(
|
|
to_address="recipient@example.com",
|
|
subject="Test Subject",
|
|
body="Test message body",
|
|
)
|
|
|
|
assert result is True
|
|
mock_send.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_html_format(self):
|
|
"""Test sending HTML email."""
|
|
pytest.importorskip("aiosmtplib")
|
|
|
|
service = EmailNotificationService(
|
|
smtp_host="smtp.example.com",
|
|
smtp_port=587,
|
|
smtp_username="user@example.com",
|
|
smtp_password="password123",
|
|
from_address="noreply@example.com",
|
|
)
|
|
|
|
with patch("aiosmtplib.send", new_callable=AsyncMock) as mock_send:
|
|
result = await service.send_email(
|
|
to_address="recipient@example.com",
|
|
subject="Test",
|
|
body="<html><body><h1>Test</h1></body></html>",
|
|
html=True,
|
|
)
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_import_error(self):
|
|
"""Test email sending when aiosmtplib not installed."""
|
|
service = EmailNotificationService(
|
|
smtp_host="smtp.example.com",
|
|
smtp_port=587,
|
|
smtp_username="user@example.com",
|
|
smtp_password="password123",
|
|
from_address="noreply@example.com",
|
|
)
|
|
|
|
# Patch the import mechanism
|
|
import builtins
|
|
real_import = builtins.__import__
|
|
|
|
def mock_import(name, *args, **kwargs):
|
|
if name == "aiosmtplib":
|
|
raise ImportError("Module not found")
|
|
return real_import(name, *args, **kwargs)
|
|
|
|
with patch("builtins.__import__", side_effect=mock_import):
|
|
result = await service.send_email(
|
|
to_address="recipient@example.com",
|
|
subject="Test",
|
|
body="Test",
|
|
)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_smtp_error(self):
|
|
"""Test email sending with SMTP error."""
|
|
pytest.importorskip("aiosmtplib")
|
|
|
|
service = EmailNotificationService(
|
|
smtp_host="smtp.example.com",
|
|
smtp_port=587,
|
|
smtp_username="user@example.com",
|
|
smtp_password="password123",
|
|
from_address="noreply@example.com",
|
|
)
|
|
|
|
with patch("aiosmtplib.send", new_callable=AsyncMock) as mock_send:
|
|
mock_send.side_effect = Exception("SMTP connection failed")
|
|
|
|
result = await service.send_email(
|
|
to_address="recipient@example.com",
|
|
subject="Test",
|
|
body="Test",
|
|
)
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestWebhookNotificationService:
|
|
"""Test cases for WebhookNotificationService."""
|
|
|
|
def test_webhook_service_init(self):
|
|
"""Test webhook service initialization."""
|
|
service = WebhookNotificationService(timeout=15, max_retries=5)
|
|
assert service.timeout == 15
|
|
assert service.max_retries == 5
|
|
|
|
def test_webhook_service_default_init(self):
|
|
"""Test webhook service with default values."""
|
|
service = WebhookNotificationService()
|
|
assert service.timeout == 10
|
|
assert service.max_retries == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_webhook_success(self):
|
|
"""Test successful webhook sending."""
|
|
service = WebhookNotificationService()
|
|
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 200
|
|
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_response.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.post = MagicMock(return_value=mock_response)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch("aiohttp.ClientSession", return_value=mock_session):
|
|
result = await service.send_webhook(
|
|
url="https://webhook.example.com/notify",
|
|
payload={"event": "test", "data": "test_data"},
|
|
)
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_webhook_custom_headers(self):
|
|
"""Test webhook with custom headers."""
|
|
service = WebhookNotificationService()
|
|
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 201
|
|
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_response.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.post = MagicMock(return_value=mock_response)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch("aiohttp.ClientSession", return_value=mock_session):
|
|
result = await service.send_webhook(
|
|
url="https://webhook.example.com/notify",
|
|
payload={"event": "test"},
|
|
headers={"Authorization": "Bearer token123"},
|
|
)
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_webhook_http_error(self):
|
|
"""Test webhook with HTTP error response."""
|
|
service = WebhookNotificationService(max_retries=2)
|
|
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 500
|
|
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_response.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.post = MagicMock(return_value=mock_response)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch("aiohttp.ClientSession", return_value=mock_session):
|
|
result = await service.send_webhook(
|
|
url="https://webhook.example.com/notify",
|
|
payload={"event": "test"},
|
|
)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_webhook_timeout_with_retry(self):
|
|
"""Test webhook timeout with exponential backoff retry."""
|
|
service = WebhookNotificationService(timeout=5, max_retries=3)
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.post = MagicMock(side_effect=asyncio.TimeoutError())
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch("aiohttp.ClientSession", return_value=mock_session):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
|
result = await service.send_webhook(
|
|
url="https://webhook.example.com/notify",
|
|
payload={"event": "test"},
|
|
)
|
|
|
|
assert result is False
|
|
# Should have retried 2 times (3 total attempts - 1 initial)
|
|
assert mock_sleep.call_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_webhook_exception_with_retry(self):
|
|
"""Test webhook with exception and retry."""
|
|
service = WebhookNotificationService(max_retries=2)
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.post = MagicMock(
|
|
side_effect=Exception("Connection refused")
|
|
)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch("aiohttp.ClientSession", return_value=mock_session):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock):
|
|
result = await service.send_webhook(
|
|
url="https://webhook.example.com/notify",
|
|
payload={"event": "test"},
|
|
)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_webhook_exponential_backoff(self):
|
|
"""Test webhook retry uses exponential backoff."""
|
|
service = WebhookNotificationService(max_retries=3)
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.post = MagicMock(side_effect=asyncio.TimeoutError())
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch("aiohttp.ClientSession", return_value=mock_session):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
|
await service.send_webhook(
|
|
url="https://webhook.example.com/notify",
|
|
payload={"event": "test"},
|
|
)
|
|
|
|
# Check exponential backoff: 2^0=1, 2^1=2
|
|
assert mock_sleep.call_count == 2
|
|
calls = [call[0][0] for call in mock_sleep.call_args_list]
|
|
assert calls == [1, 2]
|
|
|
|
|
|
class TestInAppNotificationService:
|
|
"""Test cases for InAppNotificationService."""
|
|
|
|
def test_in_app_service_init(self):
|
|
"""Test in-app service initialization."""
|
|
service = InAppNotificationService(max_notifications=50)
|
|
assert service.max_notifications == 50
|
|
assert len(service.notifications) == 0
|
|
|
|
def test_in_app_service_default_init(self):
|
|
"""Test in-app service with default values."""
|
|
service = InAppNotificationService()
|
|
assert service.max_notifications == 100
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_notification(self):
|
|
"""Test adding a notification."""
|
|
service = InAppNotificationService()
|
|
notification = Notification(
|
|
id="test_1",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title="Test",
|
|
message="Test message",
|
|
)
|
|
|
|
await service.add_notification(notification)
|
|
|
|
assert len(service.notifications) == 1
|
|
assert service.notifications[0].id == "test_1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_multiple_notifications(self):
|
|
"""Test adding multiple notifications."""
|
|
service = InAppNotificationService()
|
|
|
|
for i in range(5):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
assert len(service.notifications) == 5
|
|
# Most recent should be first
|
|
assert service.notifications[0].id == "test_4"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_notification_max_limit(self):
|
|
"""Test that old notifications are removed when max reached."""
|
|
service = InAppNotificationService(max_notifications=3)
|
|
|
|
for i in range(5):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
# Should only keep last 3
|
|
assert len(service.notifications) == 3
|
|
assert service.notifications[0].id == "test_4"
|
|
assert service.notifications[2].id == "test_2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_notifications_all(self):
|
|
"""Test getting all notifications."""
|
|
service = InAppNotificationService()
|
|
|
|
for i in range(3):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
notifications = await service.get_notifications()
|
|
assert len(notifications) == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_notifications_unread_only(self):
|
|
"""Test getting only unread notifications."""
|
|
service = InAppNotificationService()
|
|
|
|
for i in range(3):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
read=(i == 1), # Mark middle one as read
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
notifications = await service.get_notifications(unread_only=True)
|
|
assert len(notifications) == 2
|
|
assert all(not n.read for n in notifications)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_notifications_with_limit(self):
|
|
"""Test getting notifications with limit."""
|
|
service = InAppNotificationService()
|
|
|
|
for i in range(5):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
notifications = await service.get_notifications(limit=2)
|
|
assert len(notifications) == 2
|
|
# Should return most recent
|
|
assert notifications[0].id == "test_4"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_as_read(self):
|
|
"""Test marking notification as read."""
|
|
service = InAppNotificationService()
|
|
notification = Notification(
|
|
id="test_1",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title="Test",
|
|
message="Test message",
|
|
read=False,
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
result = await service.mark_as_read("test_1")
|
|
|
|
assert result is True
|
|
assert service.notifications[0].read is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_as_read_not_found(self):
|
|
"""Test marking non-existent notification as read."""
|
|
service = InAppNotificationService()
|
|
|
|
result = await service.mark_as_read("non_existent")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_all_as_read(self):
|
|
"""Test marking all notifications as read."""
|
|
service = InAppNotificationService()
|
|
|
|
for i in range(3):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
read=False,
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
count = await service.mark_all_as_read()
|
|
|
|
assert count == 3
|
|
assert all(n.read for n in service.notifications)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_all_as_read_some_already_read(self):
|
|
"""Test marking all as read when some already are."""
|
|
service = InAppNotificationService()
|
|
|
|
for i in range(3):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
read=(i == 0),
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
count = await service.mark_all_as_read()
|
|
|
|
assert count == 2
|
|
assert all(n.read for n in service.notifications)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_notifications_read_only(self):
|
|
"""Test clearing only read notifications."""
|
|
service = InAppNotificationService()
|
|
|
|
for i in range(3):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
read=(i < 2), # First 2 are read
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
count = await service.clear_notifications(read_only=True)
|
|
|
|
assert count == 2
|
|
assert len(service.notifications) == 1
|
|
assert service.notifications[0].id == "test_2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_notifications_all(self):
|
|
"""Test clearing all notifications."""
|
|
service = InAppNotificationService()
|
|
|
|
for i in range(3):
|
|
notification = Notification(
|
|
id=f"test_{i}",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title=f"Test {i}",
|
|
message=f"Test message {i}",
|
|
)
|
|
await service.add_notification(notification)
|
|
|
|
count = await service.clear_notifications(read_only=False)
|
|
|
|
assert count == 3
|
|
assert len(service.notifications) == 0
|
|
|
|
|
|
class TestNotificationService:
|
|
"""Test cases for NotificationService (main coordinator)."""
|
|
|
|
def test_notification_service_init(self):
|
|
"""Test notification service initialization."""
|
|
email_service = EmailNotificationService()
|
|
webhook_service = WebhookNotificationService()
|
|
in_app_service = InAppNotificationService()
|
|
|
|
service = NotificationService(
|
|
email_service=email_service,
|
|
webhook_service=webhook_service,
|
|
in_app_service=in_app_service,
|
|
)
|
|
|
|
assert service.email_service is email_service
|
|
assert service.webhook_service is webhook_service
|
|
assert service.in_app_service is in_app_service
|
|
|
|
def test_notification_service_default_init(self):
|
|
"""Test notification service with default services."""
|
|
service = NotificationService()
|
|
|
|
assert service.email_service is not None
|
|
assert service.webhook_service is not None
|
|
assert service.in_app_service is not None
|
|
|
|
def test_set_preferences(self):
|
|
"""Test setting notification preferences."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
enabled_channels={NotificationChannel.EMAIL},
|
|
email_address="test@example.com",
|
|
)
|
|
|
|
service.set_preferences(preferences)
|
|
|
|
assert service.preferences.email_address == "test@example.com"
|
|
assert NotificationChannel.EMAIL in service.preferences.enabled_channels
|
|
|
|
def test_is_in_quiet_hours_disabled(self):
|
|
"""Test quiet hours check when disabled."""
|
|
service = NotificationService()
|
|
# Default preferences have no quiet hours
|
|
assert service._is_in_quiet_hours() is False
|
|
|
|
def test_is_in_quiet_hours_within_hours(self):
|
|
"""Test quiet hours check when within quiet period."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
quiet_hours_start=22, quiet_hours_end=8
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
# Mock datetime to be at 23:00 (11 PM)
|
|
with patch(
|
|
"src.server.services.notification_service.datetime"
|
|
) as mock_datetime:
|
|
mock_datetime.now.return_value = MagicMock(hour=23)
|
|
assert service._is_in_quiet_hours() is True
|
|
|
|
def test_is_in_quiet_hours_outside_hours(self):
|
|
"""Test quiet hours check when outside quiet period."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
quiet_hours_start=22, quiet_hours_end=8
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
# Mock datetime to be at 14:00 (2 PM)
|
|
with patch(
|
|
"src.server.services.notification_service.datetime"
|
|
) as mock_datetime:
|
|
mock_datetime.now.return_value = MagicMock(hour=14)
|
|
assert service._is_in_quiet_hours() is False
|
|
|
|
def test_is_in_quiet_hours_spanning_midnight(self):
|
|
"""Test quiet hours spanning midnight."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
quiet_hours_start=22, quiet_hours_end=8
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
# Test at 2 AM (should be in quiet hours)
|
|
with patch(
|
|
"src.server.services.notification_service.datetime"
|
|
) as mock_datetime:
|
|
mock_datetime.now.return_value = MagicMock(hour=2)
|
|
assert service._is_in_quiet_hours() is True
|
|
|
|
def test_should_send_notification_type_disabled(self):
|
|
"""Test notification filtering by disabled type."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
enabled_types={NotificationType.DOWNLOAD_COMPLETE}
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
result = service._should_send_notification(
|
|
NotificationType.SYSTEM_ERROR, NotificationPriority.HIGH
|
|
)
|
|
|
|
assert result is False
|
|
|
|
def test_should_send_notification_priority_too_low(self):
|
|
"""Test notification filtering by priority level."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
min_priority=NotificationPriority.HIGH
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
result = service._should_send_notification(
|
|
NotificationType.DOWNLOAD_COMPLETE, NotificationPriority.NORMAL
|
|
)
|
|
|
|
assert result is False
|
|
|
|
def test_should_send_notification_in_quiet_hours(self):
|
|
"""Test notification filtering during quiet hours."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
quiet_hours_start=22, quiet_hours_end=8
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
with patch(
|
|
"src.server.services.notification_service.datetime"
|
|
) as mock_datetime:
|
|
mock_datetime.now.return_value = MagicMock(hour=23)
|
|
|
|
result = service._should_send_notification(
|
|
NotificationType.DOWNLOAD_COMPLETE, NotificationPriority.NORMAL
|
|
)
|
|
|
|
assert result is False
|
|
|
|
def test_should_send_notification_critical_bypasses_quiet_hours(self):
|
|
"""Test that critical notifications bypass quiet hours."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
quiet_hours_start=22, quiet_hours_end=8
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
with patch(
|
|
"src.server.services.notification_service.datetime"
|
|
) as mock_datetime:
|
|
mock_datetime.now.return_value = MagicMock(hour=23)
|
|
|
|
result = service._should_send_notification(
|
|
NotificationType.SYSTEM_ERROR, NotificationPriority.CRITICAL
|
|
)
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_notification_in_app_only(self):
|
|
"""Test sending notification to in-app channel only."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
enabled_channels={NotificationChannel.IN_APP}
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
notification = Notification(
|
|
id="test_1",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title="Test",
|
|
message="Test message",
|
|
)
|
|
|
|
results = await service.send_notification(notification)
|
|
|
|
assert "in_app" in results
|
|
assert results["in_app"] is True
|
|
assert "email" not in results
|
|
assert "webhook" not in results
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_notification_multi_channel(self):
|
|
"""Test sending notification to multiple channels."""
|
|
email_service = EmailNotificationService(
|
|
smtp_host="smtp.example.com",
|
|
smtp_username="user",
|
|
smtp_password="pass",
|
|
from_address="from@example.com",
|
|
)
|
|
webhook_service = WebhookNotificationService()
|
|
in_app_service = InAppNotificationService()
|
|
|
|
service = NotificationService(
|
|
email_service=email_service,
|
|
webhook_service=webhook_service,
|
|
in_app_service=in_app_service,
|
|
)
|
|
|
|
preferences = NotificationPreferences(
|
|
enabled_channels={
|
|
NotificationChannel.IN_APP,
|
|
NotificationChannel.EMAIL,
|
|
NotificationChannel.WEBHOOK,
|
|
},
|
|
email_address="test@example.com",
|
|
webhook_urls=["https://webhook.example.com/notify"],
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
notification = Notification(
|
|
id="test_1",
|
|
type=NotificationType.DOWNLOAD_COMPLETE,
|
|
priority=NotificationPriority.NORMAL,
|
|
title="Test",
|
|
message="Test message",
|
|
)
|
|
|
|
# Mock email and webhook services
|
|
with patch.object(
|
|
email_service, "send_email", new_callable=AsyncMock
|
|
) as mock_email:
|
|
mock_email.return_value = True
|
|
with patch.object(
|
|
webhook_service, "send_webhook", new_callable=AsyncMock
|
|
) as mock_webhook:
|
|
mock_webhook.return_value = True
|
|
|
|
results = await service.send_notification(notification)
|
|
|
|
assert results["in_app"] is True
|
|
assert results["email"] is True
|
|
assert results["webhook"] is True
|
|
mock_email.assert_called_once()
|
|
mock_webhook.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_notification_filtered_by_preferences(self):
|
|
"""Test that filtered notifications return empty results."""
|
|
service = NotificationService()
|
|
preferences = NotificationPreferences(
|
|
enabled_types={NotificationType.DOWNLOAD_COMPLETE}
|
|
)
|
|
service.set_preferences(preferences)
|
|
|
|
notification = Notification(
|
|
id="test_1",
|
|
type=NotificationType.SYSTEM_ERROR,
|
|
priority=NotificationPriority.HIGH,
|
|
title="Error",
|
|
message="System error occurred",
|
|
)
|
|
|
|
results = await service.send_notification(notification)
|
|
|
|
assert results == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_download_complete(self):
|
|
"""Test download complete notification."""
|
|
service = NotificationService()
|
|
|
|
with patch.object(
|
|
service, "send_notification", new_callable=AsyncMock
|
|
) as mock_send:
|
|
mock_send.return_value = {"in_app": True}
|
|
|
|
results = await service.notify_download_complete(
|
|
series_name="Test Series",
|
|
episode="S01E01",
|
|
file_path="/path/to/file.mp4",
|
|
)
|
|
|
|
assert results == {"in_app": True}
|
|
mock_send.assert_called_once()
|
|
call_args = mock_send.call_args[0][0]
|
|
assert call_args.type == NotificationType.DOWNLOAD_COMPLETE
|
|
assert call_args.priority == NotificationPriority.NORMAL
|
|
assert "Test Series" in call_args.title
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_download_failed(self):
|
|
"""Test download failed notification."""
|
|
service = NotificationService()
|
|
|
|
with patch.object(
|
|
service, "send_notification", new_callable=AsyncMock
|
|
) as mock_send:
|
|
mock_send.return_value = {"in_app": True}
|
|
|
|
results = await service.notify_download_failed(
|
|
series_name="Test Series",
|
|
episode="S01E01",
|
|
error="Network timeout",
|
|
)
|
|
|
|
assert results == {"in_app": True}
|
|
call_args = mock_send.call_args[0][0]
|
|
assert call_args.type == NotificationType.DOWNLOAD_FAILED
|
|
assert call_args.priority == NotificationPriority.HIGH
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_queue_complete(self):
|
|
"""Test queue complete notification."""
|
|
service = NotificationService()
|
|
|
|
with patch.object(
|
|
service, "send_notification", new_callable=AsyncMock
|
|
) as mock_send:
|
|
mock_send.return_value = {"in_app": True}
|
|
|
|
results = await service.notify_queue_complete(total_downloads=5)
|
|
|
|
assert results == {"in_app": True}
|
|
call_args = mock_send.call_args[0][0]
|
|
assert call_args.type == NotificationType.QUEUE_COMPLETE
|
|
assert "5" in call_args.message
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_system_error(self):
|
|
"""Test system error notification."""
|
|
service = NotificationService()
|
|
|
|
with patch.object(
|
|
service, "send_notification", new_callable=AsyncMock
|
|
) as mock_send:
|
|
mock_send.return_value = {"in_app": True}
|
|
|
|
results = await service.notify_system_error(
|
|
error="Database connection failed",
|
|
details={"db": "main", "attempts": 3},
|
|
)
|
|
|
|
assert results == {"in_app": True}
|
|
call_args = mock_send.call_args[0][0]
|
|
assert call_args.type == NotificationType.SYSTEM_ERROR
|
|
assert call_args.priority == NotificationPriority.CRITICAL
|
|
|
|
|
|
class TestGlobalNotificationService:
|
|
"""Test cases for global notification service functions."""
|
|
|
|
def test_get_notification_service(self):
|
|
"""Test getting global notification service instance."""
|
|
# Reset global instance
|
|
import src.server.services.notification_service
|
|
|
|
src.server.services.notification_service._notification_service = None
|
|
|
|
service1 = get_notification_service()
|
|
service2 = get_notification_service()
|
|
|
|
assert service1 is service2 # Should be singleton
|
|
|
|
def test_configure_notification_service(self):
|
|
"""Test configuring global notification service."""
|
|
service = configure_notification_service(
|
|
smtp_host="smtp.example.com",
|
|
smtp_port=587,
|
|
smtp_username="user@example.com",
|
|
smtp_password="password123",
|
|
from_address="noreply@example.com",
|
|
)
|
|
|
|
assert service is not None
|
|
assert service.email_service.smtp_host == "smtp.example.com"
|
|
assert service.email_service._enabled is True
|