""" Tests for NotificationManager. """ import pytest from unittest.mock import patch from datetime import datetime, timezone from local_deep_research.notifications.manager import ( NotificationManager, RateLimiter, ) from local_deep_research.notifications.templates import EventType from local_deep_research.notifications.exceptions import RateLimitError class TestNotificationManagerInit: """Tests for NotificationManager initialization.""" def test_init_with_settings_snapshot(self, mocker): """Test initialization with settings snapshot.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, "notifications.rate_limit_per_hour": 10, "notifications.rate_limit_per_day": 50, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) assert manager._settings_snapshot == snapshot assert manager._user_id == "test_user" def test_shared_rate_limiter_singleton(self, mocker): """Test that rate limiter is shared across instances.""" # Reset shared rate limiter NotificationManager._shared_rate_limiter = None snapshot = { "notifications.rate_limit_per_hour": 10, "notifications.rate_limit_per_day": 50, } manager1 = NotificationManager( settings_snapshot=snapshot, user_id="user1" ) manager2 = NotificationManager( settings_snapshot=snapshot, user_id="user2" ) # Both managers should have the same rate limiter instance assert manager1._rate_limiter is manager2._rate_limiter assert ( manager1._rate_limiter is NotificationManager._shared_rate_limiter ) class TestGetSetting: """Tests for _get_setting method.""" def test_get_setting_from_snapshot(self, mocker): """Test getting setting from snapshot.""" snapshot = { "notifications.service_url": "discord://webhook/token", } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) result = manager._get_setting("notifications.service_url") assert result == "discord://webhook/token" def test_get_setting_returns_default(self, mocker): """Test getting setting returns default when not found.""" snapshot = {} manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) result = manager._get_setting( "notifications.nonexistent", default="default_value" ) assert result == "default_value" class TestSendNotification: """Tests for send_notification method.""" def test_send_notification_success(self, mocker): """Test successful notification sending.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) # Mock service.send_event manager.service.send_event = mocker.MagicMock(return_value=True) # Mock rate limiter manager._rate_limiter.is_allowed = mocker.MagicMock(return_value=True) context = { "query": "Test query", "research_id": "123", } result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, ) assert result is True manager._rate_limiter.is_allowed.assert_called_once_with("test_user") manager.service.send_event.assert_called_once() def test_send_notification_disabled_by_setting(self, mocker): """Test notification not sent when disabled by setting.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": False, # Disabled } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.send_event = mocker.MagicMock() context = {"query": "Test"} result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, ) assert result is False manager.service.send_event.assert_not_called() def test_send_notification_no_service_url(self, mocker): """Test notification not sent when service URL missing.""" snapshot = { "notifications.service_url": "", # Empty "notifications.on_research_completed": True, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.send_event = mocker.MagicMock() context = {"query": "Test"} result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, ) assert result is False manager.service.send_event.assert_not_called() def test_send_notification_respects_rate_limit(self, mocker): """Test notification respects rate limiting.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.send_event = mocker.MagicMock() # Mock rate limiter to deny manager._rate_limiter.is_allowed = mocker.MagicMock(return_value=False) context = {"query": "Test"} with pytest.raises(RateLimitError, match="rate limit exceeded"): manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, ) manager.service.send_event.assert_not_called() def test_send_notification_force_bypasses_rate_limit(self, mocker): """Test force=True bypasses rate limiting.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.send_event = mocker.MagicMock(return_value=True) # Mock rate limiter to deny, but force should bypass manager._rate_limiter.is_allowed = mocker.MagicMock(return_value=False) context = {"query": "Test"} result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, force=True, # Bypass rate limit ) assert result is True # Rate limiter should still be checked but result ignored manager._rate_limiter.is_allowed.assert_called_once() manager.service.send_event.assert_called_once() def test_send_notification_force_bypasses_disabled_setting(self, mocker): """Test force=True bypasses disabled setting.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": False, # Disabled } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.send_event = mocker.MagicMock(return_value=True) manager._rate_limiter.is_allowed = mocker.MagicMock(return_value=True) context = {"query": "Test"} result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, force=True, ) assert result is True manager.service.send_event.assert_called_once() def test_send_notification_handles_service_failure(self, mocker): """Test graceful handling of service failures.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) # Mock service to fail manager.service.send_event = mocker.MagicMock(return_value=False) manager._rate_limiter.is_allowed = mocker.MagicMock(return_value=True) context = {"query": "Test"} result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, ) assert result is False def test_send_notification_handles_exception(self, mocker): """Test graceful handling of exceptions.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) # Mock service to raise exception manager.service.send_event = mocker.MagicMock( side_effect=Exception("Service error") ) manager._rate_limiter.is_allowed = mocker.MagicMock(return_value=True) context = {"query": "Test"} # Should not raise, returns False result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, ) assert result is False class TestShouldNotify: """Tests for _should_notify method.""" def test_should_notify_enabled(self, mocker): """Test notification enabled by default.""" snapshot = { "notifications.on_research_completed": True, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) result = manager._should_notify(EventType.RESEARCH_COMPLETED) assert result is True def test_should_notify_disabled(self, mocker): """Test notification disabled by setting.""" snapshot = { "notifications.on_research_completed": False, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) result = manager._should_notify(EventType.RESEARCH_COMPLETED) assert result is False def test_should_notify_default_false(self, mocker): """Test notification defaults to False when setting missing.""" snapshot = {} manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) result = manager._should_notify(EventType.RESEARCH_COMPLETED) # Default is False for missing settings to avoid infinite loops assert result is False class TestTestService: """Tests for test_service method.""" def test_test_service_success(self, mocker): """Test successful service test.""" snapshot = {} manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.test_service = mocker.MagicMock( return_value={"success": True} ) result = manager.test_service("discord://webhook/token") assert result == {"success": True} manager.service.test_service.assert_called_once_with( "discord://webhook/token" ) def test_test_service_failure(self, mocker): """Test failed service test.""" snapshot = {} manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.test_service = mocker.MagicMock( return_value={"success": False, "error": "Connection failed"} ) result = manager.test_service("invalid://url") assert result["success"] is False assert "error" in result class TestMasterSwitchEnvGate: """Tests for the LDR_NOTIFICATIONS_ALLOW_OUTBOUND env-only master switch. See SECURITY.md "Notification Webhook SSRF" — outbound notifications are off until the operator opts in, because Apprise's DNS-rebinding TOCTOU window cannot be closed in code. """ def test_send_notification_returns_false_when_env_unset( self, monkeypatch, mocker ): """If LDR_NOTIFICATIONS_ALLOW_OUTBOUND is unset, send_notification bails early.""" monkeypatch.delenv("LDR_NOTIFICATIONS_ALLOW_OUTBOUND", raising=False) snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.send_event = mocker.MagicMock(return_value=True) manager._rate_limiter.is_allowed = mocker.MagicMock(return_value=True) result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context={"query": "Test"}, ) assert result is False manager.service.send_event.assert_not_called() def test_send_notification_with_force_does_not_bypass_env_gate( self, monkeypatch, mocker ): """force=True bypasses per-user toggles, never the operator switch.""" monkeypatch.delenv("LDR_NOTIFICATIONS_ALLOW_OUTBOUND", raising=False) snapshot = {"notifications.service_url": "discord://webhook/token"} manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.send_event = mocker.MagicMock(return_value=True) result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context={"query": "Test"}, force=True, ) assert result is False manager.service.send_event.assert_not_called() def test_test_service_returns_disabled_error_when_env_unset( self, monkeypatch ): """test_service should refuse with a clear error when the gate is off.""" monkeypatch.delenv("LDR_NOTIFICATIONS_ALLOW_OUTBOUND", raising=False) manager = NotificationManager(settings_snapshot={}, user_id="test_user") result = manager.test_service("discord://webhook/token") assert result["success"] is False assert "LDR_NOTIFICATIONS_ALLOW_OUTBOUND" in result["error"] class TestRateLimiter: """Tests for RateLimiter class.""" def test_allow_first_notification(self): """Test first notification is always allowed.""" limiter = RateLimiter(max_per_hour=10, max_per_day=50) result = limiter.is_allowed("user1") assert result is True def test_allow_within_hourly_limit(self): """Test notifications allowed within hourly limit.""" limiter = RateLimiter(max_per_hour=3, max_per_day=10) # Send 3 notifications (at limit) assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is True # 4th should be denied assert limiter.is_allowed("user1") is False def test_allow_within_daily_limit(self): """Test notifications allowed within daily limit.""" limiter = RateLimiter(max_per_hour=100, max_per_day=3) # Send 3 notifications (at limit) assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is True # 4th should be denied assert limiter.is_allowed("user1") is False def test_allow_separate_users(self): """Test rate limits are per-user.""" limiter = RateLimiter(max_per_hour=2, max_per_day=10) # User 1 hits limit assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is False # User 2 should still be allowed assert limiter.is_allowed("user2") is True assert limiter.is_allowed("user2") is True def test_reset_single_user(self): """Test resetting rate limit for single user.""" limiter = RateLimiter(max_per_hour=2, max_per_day=10) # Hit limit assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user1") is False # Reset user1 limiter.reset("user1") # Should be allowed again assert limiter.is_allowed("user1") is True def test_reset_all_users(self): """Test resetting rate limit for all users.""" limiter = RateLimiter(max_per_hour=2, max_per_day=10) # Multiple users hit limits limiter.is_allowed("user1") limiter.is_allowed("user1") limiter.is_allowed("user2") limiter.is_allowed("user2") # Reset all limiter.reset() # Both should be allowed again assert limiter.is_allowed("user1") is True assert limiter.is_allowed("user2") is True def test_cleanup_inactive_users(self, mocker): """Test periodic cleanup of inactive users.""" limiter = RateLimiter( max_per_hour=10, max_per_day=50, cleanup_interval_hours=1 ) # Add some activity limiter.is_allowed("user1") # Mock datetime to simulate time passing (8 days) from datetime import timedelta fake_now = datetime.now(timezone.utc) + timedelta(days=8) with patch( "local_deep_research.notifications.manager.datetime" ) as mock_dt: mock_dt.now.return_value = fake_now # Trigger cleanup by calling allow limiter._cleanup_inactive_users_if_needed(fake_now) # User1 should be cleaned up assert "user1" not in limiter._hourly_counts assert "user1" not in limiter._daily_counts def test_thread_safety(self, mocker): """Test rate limiter is thread-safe.""" import threading limiter = RateLimiter(max_per_hour=100, max_per_day=1000) results = [] def send_notifications(): for _ in range(10): results.append(limiter.is_allowed("user1")) # Create 5 threads sending notifications concurrently threads = [ threading.Thread(target=send_notifications) for _ in range(5) ] for thread in threads: thread.start() for thread in threads: thread.join() # All should be allowed (50 total, well under limit) assert all(results) assert len(results) == 50 class TestIntegration: """Integration tests for NotificationManager.""" @pytest.fixture(autouse=True) def reset_rate_limiter(self): """Reset shared rate limiter before each test for isolation.""" NotificationManager._shared_rate_limiter = None yield NotificationManager._shared_rate_limiter = None def test_full_notification_workflow(self, mocker): """Test complete notification workflow with settings snapshot.""" # Simulate background thread scenario snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, "notifications.rate_limit_per_hour": 10, "notifications.rate_limit_per_day": 50, } # Create manager with snapshot (no session - thread-safe) manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) # Mock service manager.service.send_event = mocker.MagicMock(return_value=True) # Send notification context = { "query": "What is quantum computing?", "research_id": "123-abc", "summary": "Quantum computing uses quantum mechanics...", } result = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context=context, ) assert result is True # Verify service was called with correct params call_args = manager.service.send_event.call_args assert call_args[0][0] == EventType.RESEARCH_COMPLETED assert call_args[0][1] == context assert call_args[1]["service_urls"] == "discord://webhook/token" def test_multiple_event_types(self, mocker): """Test sending different event types.""" snapshot = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, "notifications.on_research_failed": True, "notifications.on_subscription_update": False, # Disabled } manager = NotificationManager( settings_snapshot=snapshot, user_id="test_user" ) manager.service.send_event = mocker.MagicMock(return_value=True) # Research completed - should send result1 = manager.send_notification( event_type=EventType.RESEARCH_COMPLETED, context={"query": "Test"}, force=True, ) assert result1 is True # Research failed - should send result2 = manager.send_notification( event_type=EventType.RESEARCH_FAILED, context={"query": "Test", "error": "Failed"}, force=True, ) assert result2 is True # Subscription update - should not send (disabled) result3 = manager.send_notification( event_type=EventType.SUBSCRIPTION_UPDATE, context={"subscription_name": "Test"}, ) assert result3 is False class TestPerUserRateLimiting: """Tests for per-user rate limiting functionality.""" def test_rate_limiter_set_user_limits(self): """Test setting per-user rate limits.""" limiter = RateLimiter(max_per_hour=10, max_per_day=50) # Set custom limits for specific users limiter.set_user_limits("user_a", max_per_hour=5, max_per_day=25) limiter.set_user_limits("user_b", max_per_hour=20, max_per_day=100) # Verify limits are set correctly assert limiter.get_user_limits("user_a") == (5, 25) assert limiter.get_user_limits("user_b") == (20, 100) assert limiter.get_user_limits("user_c") == (10, 50) # Uses defaults def test_rate_limiter_respects_per_user_limits(self): """Test that rate limiter enforces different limits per user.""" limiter = RateLimiter(max_per_hour=10, max_per_day=50) # Set conservative limits for user_a limiter.set_user_limits("user_a", max_per_hour=2, max_per_day=5) # User A should be limited to 2 per hour assert limiter.is_allowed("user_a") is True # 1st assert limiter.is_allowed("user_a") is True # 2nd assert limiter.is_allowed("user_a") is False # 3rd - exceeds limit # User B should use default limits (10 per hour) for _ in range(10): assert limiter.is_allowed("user_b") is True assert limiter.is_allowed("user_b") is False # 11th - exceeds default def test_notification_manager_with_user_id(self, mocker): """Test NotificationManager configures per-user limits.""" # Reset shared rate limiter NotificationManager._shared_rate_limiter = None # User A with conservative limits snapshot_a = { "notifications.rate_limit_per_hour": 3, "notifications.rate_limit_per_day": 10, } manager_a = NotificationManager(snapshot_a, user_id="user_a") # User B with generous limits snapshot_b = { "notifications.rate_limit_per_hour": 15, "notifications.rate_limit_per_day": 50, } manager_b = NotificationManager(snapshot_b, user_id="user_b") # Verify both users have the same rate limiter instance (singleton) assert manager_a._rate_limiter is manager_b._rate_limiter # Verify per-user limits are configured correctly limiter = manager_a._rate_limiter assert limiter.get_user_limits("user_a") == (3, 10) assert limiter.get_user_limits("user_b") == (15, 50) def test_per_user_limits_isolation(self, mocker): """Test that users don't interfere with each other's limits.""" # Reset shared rate limiter NotificationManager._shared_rate_limiter = None # Create managers for two users snapshot_a = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, "notifications.rate_limit_per_hour": 2, "notifications.rate_limit_per_day": 5, } manager_a = NotificationManager(snapshot_a, user_id="user_a") snapshot_b = { "notifications.service_url": "discord://webhook/token", "notifications.on_research_completed": True, "notifications.rate_limit_per_hour": 5, "notifications.rate_limit_per_day": 10, } manager_b = NotificationManager(snapshot_b, user_id="user_b") # Mock service manager_a.service.send_event = mocker.MagicMock(return_value=True) manager_b.service.send_event = mocker.MagicMock(return_value=True) context = {"query": "Test"} # User A sends notifications up to their limit (2) result1 = manager_a.send_notification( EventType.RESEARCH_COMPLETED, context ) assert result1 is True result2 = manager_a.send_notification( EventType.RESEARCH_COMPLETED, context ) assert result2 is True # User A exceeds limit with pytest.raises(RateLimitError): manager_a.send_notification(EventType.RESEARCH_COMPLETED, context) # User B should still be able to send (not affected by User A) for _ in range(5): # User B has 5/hour limit result = manager_b.send_notification( EventType.RESEARCH_COMPLETED, context ) assert result is True # User B exceeds their own limit with pytest.raises(RateLimitError): manager_b.send_notification(EventType.RESEARCH_COMPLETED, context) def test_manager_configures_user_limits(self, mocker): """Test that manager configures per-user limits on initialization.""" # Reset shared rate limiter NotificationManager._shared_rate_limiter = None snapshot = { "notifications.rate_limit_per_hour": 7, "notifications.rate_limit_per_day": 30, } # Create manager with user_id manager = NotificationManager(snapshot, user_id="some_user") # Limiter should be created with defaults from snapshot limiter = manager._rate_limiter assert limiter.max_per_hour == 7 assert limiter.max_per_day == 30 # User-specific limits should be configured assert limiter.get_user_limits("some_user") == (7, 30) def test_updating_user_limits_after_initialization(self): """Test that user limits can be updated after manager creation.""" # Reset shared rate limiter NotificationManager._shared_rate_limiter = None # Create manager with initial limits snapshot = { "notifications.rate_limit_per_hour": 5, "notifications.rate_limit_per_day": 20, } manager = NotificationManager(snapshot, user_id="user_a") # Verify initial limits assert manager._rate_limiter.get_user_limits("user_a") == (5, 20) # Update limits directly on rate limiter manager._rate_limiter.set_user_limits( "user_a", max_per_hour=10, max_per_day=40 ) # Verify limits are updated assert manager._rate_limiter.get_user_limits("user_a") == (10, 40) def test_multiple_managers_same_user_updates_limits(self): """Test that creating multiple managers for same user updates limits.""" # Reset shared rate limiter NotificationManager._shared_rate_limiter = None # First manager for user_a with 5/hour snapshot1 = {"notifications.rate_limit_per_hour": 5} manager1 = NotificationManager(snapshot1, user_id="user_a") assert manager1._rate_limiter.get_user_limits("user_a") == (5, 50) # Second manager for user_a with 10/hour (should update) snapshot2 = {"notifications.rate_limit_per_hour": 10} manager2 = NotificationManager(snapshot2, user_id="user_a") # Both managers should see the updated limits assert manager1._rate_limiter.get_user_limits("user_a") == (10, 50) assert manager2._rate_limiter.get_user_limits("user_a") == (10, 50) class TestEgressPolicyURLFilter: """_filter_urls_by_egress_policy gates notification dispatch by scope.""" def _mgr(self, scope, tool="searxng"): snap = { "policy.egress_scope": {"value": scope}, "search.tool": {"value": tool}, } return NotificationManager(settings_snapshot=snap, user_id="u") def test_private_only_blocks_vendor_schemes_and_public_http(self): mgr = self._mgr("private_only", tool="library") out = mgr._filter_urls_by_egress_policy( "slack://t discord://x http://127.0.0.1/h https://public.example.com/h" ) # Only the LOCAL http webhook survives; vendor schemes + public host dropped. assert out == "http://127.0.0.1/h" def test_retired_both_coerces_to_adaptive(self): # `both` is retired (ADR-0007): it coerces to ADAPTIVE, which with the # default public primary (searxng) resolves PUBLIC_ONLY. Vendor schemes # still pass (only PRIVATE_ONLY refuses them), but the LOCAL http # webhook is now dropped because public_only blocks private hosts. mgr = self._mgr("both") out = mgr._filter_urls_by_egress_policy("slack://t http://127.0.0.1/h") assert "slack://t" in out assert "http://127.0.0.1/h" not in out def test_unprotected_scope_passes_all_urls(self): # The escape hatch that replaces the old permissive `both`: vendor # schemes AND a local http webhook both pass (egress protection off; # the hard SSRF/metadata block still applies to metadata IPs). mgr = self._mgr("unprotected") out = mgr._filter_urls_by_egress_policy("slack://t http://127.0.0.1/h") assert "slack://t" in out assert "http://127.0.0.1/h" in out def test_public_only_blocks_private_http_keeps_public(self): mgr = self._mgr("public_only", tool="searxng") out = mgr._filter_urls_by_egress_policy( "http://127.0.0.1/h https://public.example.com/h" ) assert out == "https://public.example.com/h" def test_no_snapshot_passes_through(self): mgr = NotificationManager.__new__(NotificationManager) mgr._settings_snapshot = None assert mgr._filter_urls_by_egress_policy("slack://t") == "slack://t"