"""High-value tests for the notifications module. Covers gaps not addressed by existing tests: - RateLimiter per-user limits, cleanup, and reset - Queue helper functions with mocked dependencies - NotificationManager rate limit integration - Exception hierarchy """ from datetime import datetime, timezone, timedelta from unittest.mock import patch, MagicMock from local_deep_research.notifications.exceptions import ( NotificationError, ServiceError, SendError, RateLimitError, ) from local_deep_research.notifications.manager import ( RateLimiter, ) from local_deep_research.notifications.templates import EventType from local_deep_research.notifications.queue_helpers import ( send_queue_notification, send_queue_failed_notification, ) class TestExceptionHierarchy: """Test notification exception class hierarchy.""" def test_service_error_is_notification_error(self): """ServiceError inherits from NotificationError.""" assert issubclass(ServiceError, NotificationError) def test_send_error_is_notification_error(self): """SendError inherits from NotificationError.""" assert issubclass(SendError, NotificationError) def test_rate_limit_error_is_notification_error(self): """RateLimitError inherits from NotificationError.""" assert issubclass(RateLimitError, NotificationError) def test_notification_error_is_exception(self): """NotificationError inherits from Exception.""" assert issubclass(NotificationError, Exception) def test_exception_message_preserved(self): """Exception message is accessible via str().""" err = ServiceError("test message") assert str(err) == "test message" class TestRateLimiterPerUserLimits: """Test RateLimiter per-user rate limit configuration.""" def test_default_limits_used_when_no_user_config(self): """Users without explicit config get default limits.""" limiter = RateLimiter(max_per_hour=5, max_per_day=20) h, d = limiter.get_user_limits("unknown_user") assert h == 5 assert d == 20 def test_set_user_limits_applies_to_specific_user(self): """Per-user limits override defaults for that user.""" limiter = RateLimiter(max_per_hour=10, max_per_day=50) limiter.set_user_limits("user_a", max_per_hour=3, max_per_day=15) h, d = limiter.get_user_limits("user_a") assert h == 3 assert d == 15 # Other user still gets defaults h2, d2 = limiter.get_user_limits("user_b") assert h2 == 10 assert d2 == 50 def test_user_limits_enforced_per_user(self): """One user hitting limit doesn't affect another.""" limiter = RateLimiter(max_per_hour=100, max_per_day=100) limiter.set_user_limits("limited_user", max_per_hour=2, max_per_day=100) assert limiter.is_allowed("limited_user") is True assert limiter.is_allowed("limited_user") is True assert limiter.is_allowed("limited_user") is False # Hit hourly limit # Other user is unaffected assert limiter.is_allowed("other_user") is True def test_daily_limit_enforced(self): """Daily limit blocks after enough notifications.""" limiter = RateLimiter(max_per_hour=100, max_per_day=3) assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is False # Hit daily limit class TestRateLimiterReset: """Test RateLimiter reset functionality.""" def test_reset_specific_user(self): """Resetting a user clears their counts.""" limiter = RateLimiter(max_per_hour=2, max_per_day=10) limiter.is_allowed("user1") limiter.is_allowed("user1") assert limiter.is_allowed("user1") is False limiter.reset("user1") assert limiter.is_allowed("user1") is True def test_reset_all_users(self): """Resetting with None clears all users.""" limiter = RateLimiter(max_per_hour=1, max_per_day=10) limiter.is_allowed("user1") limiter.is_allowed("user2") limiter.reset() assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user2") is True def test_reset_nonexistent_user_does_not_raise(self): """Resetting a user with no history doesn't raise.""" limiter = RateLimiter() limiter.reset("nonexistent") # Should not raise class TestRateLimiterCleanup: """Test RateLimiter inactive user cleanup.""" def test_cleanup_removes_inactive_users(self): """Users with no activity for >7 days are cleaned up.""" limiter = RateLimiter( max_per_hour=100, max_per_day=100, cleanup_interval_hours=0 ) # Force _last_cleanup to be old so cleanup triggers limiter._last_cleanup = datetime.now(timezone.utc) - timedelta(hours=25) # Add old entries directly old_time = datetime.now(timezone.utc) - timedelta(days=8) limiter._hourly_counts["old_user"] = MagicMock() limiter._hourly_counts["old_user"].__iter__ = MagicMock( return_value=iter([old_time]) ) limiter._hourly_counts["old_user"].__add__ = MagicMock( return_value=[old_time] ) limiter._daily_counts["old_user"] = MagicMock() limiter._daily_counts["old_user"].__iter__ = MagicMock( return_value=iter([old_time]) ) limiter._daily_counts["old_user"].__add__ = MagicMock( return_value=[old_time] ) # Trigger cleanup via is_allowed for a different user limiter.is_allowed("active_user") # old_user should be cleaned up assert "old_user" not in limiter._hourly_counts class TestQueueHelperFunctions: """Test queue notification helper functions.""" @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_notification_calls_manager(self, mock_manager_cls): """send_queue_notification creates manager and calls send_notification.""" mock_manager = MagicMock() mock_manager.send_notification.return_value = True mock_manager_cls.return_value = mock_manager result = send_queue_notification( username="user1", research_id="r1", query="test query", settings_snapshot={"key": "val"}, position=3, ) assert result is True mock_manager.send_notification.assert_called_once() call_kwargs = mock_manager.send_notification.call_args assert call_kwargs[1]["event_type"] == EventType.RESEARCH_QUEUED context = call_kwargs[1]["context"] assert context["position"] == 3 @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_notification_without_position(self, mock_manager_cls): """send_queue_notification omits position when not provided.""" mock_manager = MagicMock() mock_manager.send_notification.return_value = True mock_manager_cls.return_value = mock_manager send_queue_notification( username="user1", research_id="r1", query="test", settings_snapshot={}, ) context = mock_manager.send_notification.call_args[1]["context"] assert "position" not in context @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_notification_handles_exception(self, mock_manager_cls): """send_queue_notification returns False on exception.""" mock_manager_cls.side_effect = Exception("init failed") result = send_queue_notification( username="user1", research_id="r1", query="test", settings_snapshot={}, ) assert result is False def test_send_queue_failed_notification_no_snapshot_returns_false(self): """send_queue_failed_notification returns False when no settings_snapshot.""" result = send_queue_failed_notification( username="user1", research_id="r1", query="test", error_message="error", settings_snapshot=None, ) assert result is False @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_failed_includes_error_in_context( self, mock_manager_cls ): """send_queue_failed_notification includes error in context.""" mock_manager = MagicMock() mock_manager.send_notification.return_value = True mock_manager_cls.return_value = mock_manager send_queue_failed_notification( username="user1", research_id="r1", query="test", error_message="Something went wrong", settings_snapshot={"key": "val"}, ) context = mock_manager.send_notification.call_args[1]["context"] assert context["error"] == "Something went wrong" assert ( mock_manager.send_notification.call_args[1]["event_type"] == EventType.RESEARCH_FAILED )