"""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="

Test

", 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