""" Comprehensive tests for notification-related modules. Covers: - NotificationURLValidator (SSRF prevention, scheme blocking, private IP detection) - NotificationService (send logic, event sending, service testing, URL validation) - NotificationManager (rate limiting, settings snapshot, event routing) - RateLimiter (per-user limits, hourly/daily windows, cleanup) - NotificationTemplate (Jinja2 rendering, custom templates, fallback) - build_notification_url (URL building from settings) - Queue helpers (send_queue_notification, send_queue_failed_notification) - mask_sensitive_url (credential masking) - Exception hierarchy """ import threading from collections import deque from datetime import datetime, timedelta, timezone from unittest.mock import patch import pytest from local_deep_research.notifications.exceptions import ( NotificationError, RateLimitError, SendError, ServiceError, ) from local_deep_research.notifications.manager import ( NotificationManager, RateLimiter, ) from local_deep_research.notifications.service import NotificationService from local_deep_research.notifications.templates import ( EventType, NotificationTemplate, ) from local_deep_research.notifications.url_builder import build_notification_url from local_deep_research.notifications.queue_helpers import ( send_queue_notification, send_queue_failed_notification, ) from local_deep_research.security.notification_validator import ( NotificationURLValidationError, NotificationURLValidator, ) from local_deep_research.security.url_builder import mask_sensitive_url from local_deep_research.security.url_validator import ( URLValidationError, URLValidator, ) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _reset_shared_rate_limiter(): """Reset the singleton rate limiter between tests to avoid cross-contamination.""" NotificationManager._shared_rate_limiter = None yield NotificationManager._shared_rate_limiter = None @pytest.fixture(autouse=True) def _enable_outbound_notifications(monkeypatch): """Module-level autouse: open the operator-level env gate so tests in this file that exercise the inner notification logic don't bail at NotificationManager.__init__ / NotificationService.send. Tests of the gate itself (none in this file) should override with monkeypatch.delenv. See SECURITY.md "Notification Webhook SSRF". """ monkeypatch.setenv("LDR_NOTIFICATIONS_ALLOW_OUTBOUND", "true") @pytest.fixture() def default_settings(): """Settings snapshot that enables notifications with a valid URL.""" return { "notifications.service_url": "discord://webhook_id/token", "notifications.on_research_completed": True, "notifications.on_research_failed": True, "notifications.on_research_queued": True, "notifications.rate_limit_per_hour": 10, "notifications.rate_limit_per_day": 50, "notifications.allow_private_ips": False, } @pytest.fixture() def disabled_settings(): """Settings snapshot with notifications disabled.""" return { "notifications.service_url": "", "notifications.on_research_completed": False, "notifications.on_research_failed": False, "notifications.rate_limit_per_hour": 10, "notifications.rate_limit_per_day": 50, } # =================================================================== # NotificationURLValidator # =================================================================== class TestNotificationURLValidator: """Tests for SSRF prevention and scheme validation.""" # -- Allowed schemes -------------------------------------------------- @pytest.mark.parametrize( "url", [ "discord://webhook_id/token", "slack://token_a/token_b/token_c", "mailto://user:pass@smtp.example.com", "https://hooks.example.com/webhook", "http://hooks.example.com/webhook", "telegram://bot_token/chat_id", "gotify://gotify.example.com/token", "pushover://user_key/token", "ntfy://ntfy.sh/topic", "ntfys://ntfy.sh/topic", "json://hooks.example.com/endpoint", "xml://hooks.example.com/endpoint", "form://hooks.example.com/endpoint", "matrix://user:pass@matrix.org/room", "mattermost://hooks.example.com/token", "rocketchat://hooks.example.com/token", "teams://token_a/token_b/token_c", ], ) def test_allowed_schemes_pass(self, url): is_valid, err = NotificationURLValidator.validate_service_url(url) assert is_valid is True, f"Expected valid for {url}, got error: {err}" assert err is None # -- Blocked schemes --------------------------------------------------- @pytest.mark.parametrize( "url", [ "file:///etc/passwd", "ftp://evil.com/payload", "ftps://evil.com/payload", "data:text/html,", "javascript:alert(1)", "vbscript:msgbox", "about:blank", "blob:http://evil.com/obj", ], ) def test_blocked_schemes_rejected(self, url): is_valid, err = NotificationURLValidator.validate_service_url(url) assert is_valid is False assert err is not None # The error should mention 'protocol' or 'Blocked' assert "protocol" in err.lower() or "blocked" in err.lower() # -- Unknown scheme ---------------------------------------------------- def test_unknown_scheme_rejected(self): is_valid, err = NotificationURLValidator.validate_service_url( "gopher://evil.com" ) assert is_valid is False assert "unsupported" in err.lower() or "protocol" in err.lower() # -- Empty / invalid inputs -------------------------------------------- @pytest.mark.parametrize("url", ["", None, 123, " "]) def test_empty_or_invalid_input(self, url): is_valid, err = NotificationURLValidator.validate_service_url(url) assert is_valid is False def test_url_without_scheme(self): is_valid, err = NotificationURLValidator.validate_service_url( "example.com/webhook" ) assert is_valid is False # -- Private IP blocking ----------------------------------------------- @pytest.mark.parametrize( "url", [ "http://localhost:5000/webhook", "http://127.0.0.1:8080/hook", "https://0.0.0.0/hook", "http://[::1]/hook", ], ) def test_private_ip_blocked_by_default(self, url): is_valid, err = NotificationURLValidator.validate_service_url(url) assert is_valid is False assert "private" in err.lower() or "blocked" in err.lower() @pytest.mark.parametrize( "url", [ "http://localhost:5000/webhook", "http://127.0.0.1:8080/hook", ], ) def test_private_ip_allowed_when_flag_set(self, url): is_valid, err = NotificationURLValidator.validate_service_url( url, allow_private_ips=True ) assert is_valid is True # -- Private IP detection helper ---------------------------------------- @pytest.mark.parametrize( "hostname,expected", [ ("localhost", True), ("127.0.0.1", True), ("::1", True), ("0.0.0.0", True), ("::", True), ("10.0.0.1", True), ("172.16.0.1", True), ("192.168.1.1", True), ("8.8.8.8", False), ("hooks.slack.com", False), ], ) def test_is_private_ip(self, hostname, expected): assert NotificationURLValidator._is_private_ip(hostname) is expected # -- validate_multiple_urls --------------------------------------------- def test_validate_multiple_urls_all_valid(self): urls = "discord://id/token, slack://a/b/c" is_valid, err = NotificationURLValidator.validate_multiple_urls(urls) assert is_valid is True assert err is None def test_validate_multiple_urls_one_invalid(self): urls = "discord://id/token, file:///etc/passwd" is_valid, err = NotificationURLValidator.validate_multiple_urls(urls) assert is_valid is False assert err is not None def test_validate_multiple_urls_empty(self): is_valid, err = NotificationURLValidator.validate_multiple_urls("") assert is_valid is False def test_validate_multiple_urls_only_whitespace(self): is_valid, err = NotificationURLValidator.validate_multiple_urls(" , ") assert is_valid is False # -- validate_service_url_strict ---------------------------------------- def test_strict_validation_raises_on_invalid(self): with pytest.raises(NotificationURLValidationError): NotificationURLValidator.validate_service_url_strict( "file:///etc/passwd" ) def test_strict_validation_returns_true_on_valid(self): assert ( NotificationURLValidator.validate_service_url_strict( "discord://id/token" ) is True ) # =================================================================== # NotificationService # =================================================================== class TestNotificationService: """Tests for the low-level notification service wrapping Apprise.""" def test_init_defaults(self): svc = NotificationService(outbound_allowed=True) assert svc.allow_private_ips is False def test_init_allow_private_ips(self): svc = NotificationService(allow_private_ips=True, outbound_allowed=True) assert svc.allow_private_ips is True # -- _validate_url (static) ------------------------------------------- def test_validate_url_rejects_empty(self): with pytest.raises(ServiceError, match="non-empty"): NotificationService._validate_url("") def test_validate_url_rejects_none(self): with pytest.raises(ServiceError, match="non-empty"): NotificationService._validate_url(None) def test_validate_url_rejects_no_scheme(self): with pytest.raises(ServiceError, match="Invalid URL"): NotificationService._validate_url("example.com/webhook") def test_validate_url_accepts_valid(self): # Should not raise NotificationService._validate_url("discord://id/token") # -- get_service_type -------------------------------------------------- def test_get_service_type_known(self): svc = NotificationService(outbound_allowed=True) assert svc.get_service_type("mailto://user@example.com") == "email" assert svc.get_service_type("discord://id/token") == "discord" assert svc.get_service_type("slack://a/b/c") == "slack" assert svc.get_service_type("tgram://bot_token/chat_id") == "telegram" assert svc.get_service_type("smtp://mail.example.com") == "smtp" assert svc.get_service_type("smtps://mail.example.com") == "smtp" def test_get_service_type_unknown(self): svc = NotificationService(outbound_allowed=True) assert svc.get_service_type("https://hooks.example.com") == "unknown" # -- send (mocked Apprise) --------------------------------------------- @patch("local_deep_research.notifications.service.NotificationURLValidator") @patch("local_deep_research.notifications.service.apprise.Apprise") def test_send_with_service_urls_success(self, MockApprise, MockValidator): MockValidator.validate_multiple_urls.return_value = (True, None) mock_instance = MockApprise.return_value mock_instance.add.return_value = True mock_instance.notify.return_value = True svc = NotificationService(outbound_allowed=True) result = svc.send( title="Test", body="Body", service_urls="discord://id/token", ) assert result is True mock_instance.notify.assert_called_once() @patch("local_deep_research.notifications.service.NotificationURLValidator") @patch("local_deep_research.notifications.service.apprise.Apprise") def test_send_with_invalid_url_raises_service_error( self, MockApprise, MockValidator ): MockValidator.validate_multiple_urls.return_value = ( False, "bad url", ) svc = NotificationService(outbound_allowed=True) with pytest.raises(ServiceError): svc.send( title="Test", body="Body", service_urls="file:///etc/passwd", ) @patch("local_deep_research.notifications.service.NotificationURLValidator") @patch("local_deep_research.notifications.service.apprise.Apprise") def test_send_apprise_add_failure_returns_false( self, MockApprise, MockValidator ): MockValidator.validate_multiple_urls.return_value = (True, None) mock_instance = MockApprise.return_value mock_instance.add.return_value = False svc = NotificationService(outbound_allowed=True) # When add() returns False the original apprise instance is used, # but the temp one is used here so it returns False immediately. result = svc.send( title="Test", body="Body", service_urls="discord://id/token" ) assert result is False def test_send_no_urls_no_configured_services_returns_false(self): svc = NotificationService(outbound_allowed=True) result = svc.send(title="Test", body="Body") assert result is False # -- send_event -------------------------------------------------------- @patch.object(NotificationService, "send", return_value=True) def test_send_event_uses_template(self, mock_send): svc = NotificationService(outbound_allowed=True) result = svc.send_event( event_type=EventType.RESEARCH_COMPLETED, context={"query": "test query", "research_id": "abc123"}, service_urls="discord://id/token", ) assert result is True mock_send.assert_called_once() call_kwargs = mock_send.call_args # The title and body should come from template formatting assert call_kwargs[1]["service_urls"] == "discord://id/token" # -- test_service (mocked Apprise) ------------------------------------ @patch("local_deep_research.notifications.service.NotificationURLValidator") @patch("local_deep_research.notifications.service.apprise.Apprise") def test_test_service_success(self, MockApprise, MockValidator): MockValidator.validate_service_url.return_value = (True, None) mock_instance = MockApprise.return_value mock_instance.add.return_value = True mock_instance.notify.return_value = True svc = NotificationService(outbound_allowed=True) result = svc.test_service("discord://id/token") assert result["success"] is True @patch("local_deep_research.notifications.service.NotificationURLValidator") def test_test_service_invalid_url(self, MockValidator): MockValidator.validate_service_url.return_value = (False, "blocked") svc = NotificationService(outbound_allowed=True) result = svc.test_service("file:///etc/passwd") assert result["success"] is False assert "error" in result @patch("local_deep_research.notifications.service.NotificationURLValidator") def test_test_service_invalid_url_surfaces_validator_reason( self, MockValidator ): # Validator's specific reason (e.g. "Blocked unsafe protocol: file") # must reach the user so they can fix it instead of seeing a # generic "Invalid notification service URL." Without this, an # IPv6-only operator hitting the NAT64 block has no signal at # all that there's a server-side opt-in to flip. MockValidator.validate_service_url.return_value = ( False, "Blocked unsafe protocol: file", ) svc = NotificationService(outbound_allowed=True) result = svc.test_service("file:///etc/passwd") assert result["success"] is False assert "Blocked unsafe protocol: file" in result["error"] # ------------------------------------------------------------------ # IP rejection matrix — single source of truth for which IPs trigger # which user-visible behavior. Add new IP categories here rather # than scattering fixtures across multiple tests. End-to-end against # the REAL NotificationURLValidator (no mock) so validator wording # drift fails here too — see NotificationURLValidator.validate_service_url # docstring for the contract. # ------------------------------------------------------------------ @pytest.mark.parametrize( "url,expect_hint", [ # --- Always-blocked: admin hint MUST be suppressed --- # (none of these are unblockable via LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS # or LDR_SECURITY_ALLOW_NAT64, so naming those env vars # would mislead the user.) # Cloud-metadata IPv4 (AWS/Azure/OCI/DO IMDS, ECS, Tencent, Alibaba) pytest.param( "http://169.254.169.254/latest/meta-data/", False, id="aws-imds" ), pytest.param("http://169.254.170.2/", False, id="aws-ecs-v3"), pytest.param("http://169.254.170.23/", False, id="aws-ecs-v4"), pytest.param("http://169.254.0.23/", False, id="tencent-metadata"), pytest.param( "http://100.100.100.200/", False, id="alibaba-metadata" ), # NAT64-wrapped metadata pytest.param( "http://[64:ff9b::a9fe:a9fe]/", False, id="nat64-rfc6052-wrap-of-imds", ), pytest.param( "http://[64:ff9b:1::a9fe:a9fe]/", False, id="nat64-rfc8215-wrap-of-imds", ), # IPv6 transition prefixes (non-NAT64, always blocked) pytest.param( "http://[2002:7f00:1::]/", False, id="6to4-wrap-of-loopback", ), pytest.param("http://[2001::]/", False, id="teredo-range"), pytest.param("http://[100::]/", False, id="discard-prefix"), pytest.param( "http://[::ffff:169.254.169.254]/", False, id="ipv4-mapped-ipv6-of-imds", ), # Plugin-scheme metadata (different validator prefix) pytest.param( "signal://169.254.169.254:8080/path", False, id="plugin-scheme-signal-to-imds", ), pytest.param( "discord://169.254.169.254/path", False, id="plugin-scheme-discord-to-imds", ), # --- Recoverable: admin hint MUST appear --- # (LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true would actually # unblock these, so the hint is actionable.) pytest.param("http://127.0.0.1/", True, id="ipv4-loopback"), pytest.param("http://10.0.0.1/", True, id="rfc1918-10"), pytest.param("http://172.16.0.1/", True, id="rfc1918-172-16"), pytest.param("http://192.168.1.1/", True, id="rfc1918-192-168"), pytest.param("http://100.64.0.1/", True, id="cgnat-100-64"), pytest.param("http://[::1]/", True, id="ipv6-loopback"), pytest.param("http://[fc00::1]/", True, id="ula-fc00"), pytest.param("http://[fe80::1]/", True, id="link-local-fe80"), pytest.param("http://localhost/", True, id="localhost-dns"), # NAT64-wrapped NON-metadata IPv4 — recoverable ONLY via # LDR_SECURITY_ALLOW_NAT64=true (the private-IPs flag cannot # unblock the NAT64 prefixes). 64:ff9b::5db8:d822 and the # RFC8215 64:ff9b:1:: variant both wrap 93.184.216.34, a public # IPv4. Guards against the NAT64 hint silently going dead — the # exact IPv6-only/NAT64 deployment the hint exists to help. pytest.param( "http://[64:ff9b::5db8:d822]/", True, id="nat64-rfc6052-wrap-of-public", ), pytest.param( "http://[64:ff9b:1::5db8:d822]/", True, id="nat64-rfc8215-wrap-of-public", ), ], ) def test_test_service_ip_rejection_matrix(self, url, expect_hint): """End-to-end against the REAL validator: pin the user-visible behavior for every IP category. If ``expect_hint`` is True, the URL targets a recoverable destination that LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true (private IPs) or LDR_SECURITY_ALLOW_NAT64=true (NAT64-wrapped non-metadata IPv4) would unblock, so the admin hint MUST appear. If False, the URL targets an always-blocked category (cloud-metadata, 6to4, Teredo, discard, IPv4-mapped IPv6 of metadata, NAT64-wrapped metadata, plugin-scheme metadata) and the hint MUST be suppressed — the env var cannot help. Single source of truth: add new IP categories to this matrix rather than scattering fixtures across multiple tests. See NotificationURLValidator.validate_service_url docstring. """ svc = NotificationService(outbound_allowed=True) result = svc.test_service(url) assert result["success"] is False, f"Expected block for {url}" assert result["error"], f"Empty error for {url}" if expect_hint: assert "LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS" in result["error"], ( f"Expected admin hint for {url}, got: {result['error']!r}" ) assert "LDR_SECURITY_ALLOW_NAT64" in result["error"], ( f"Expected NAT64 hint for {url}, got: {result['error']!r}" ) else: assert ( "LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS" not in result["error"] ), f"Hint should be suppressed for {url}, got: {result['error']!r}" assert "LDR_SECURITY_ALLOW_NAT64" not in result["error"], ( f"NAT64 hint should be suppressed for {url}, " f"got: {result['error']!r}" ) @patch("local_deep_research.notifications.service.NotificationURLValidator") @patch("local_deep_research.notifications.service.apprise.Apprise") def test_test_service_add_failure(self, MockApprise, MockValidator): MockValidator.validate_service_url.return_value = (True, None) mock_instance = MockApprise.return_value mock_instance.add.return_value = False svc = NotificationService(outbound_allowed=True) result = svc.test_service("discord://id/token") assert result["success"] is False assert "Failed to add" in result["error"] @patch("local_deep_research.notifications.service.NotificationURLValidator") @patch("local_deep_research.notifications.service.apprise.Apprise") def test_test_service_notify_failure(self, MockApprise, MockValidator): MockValidator.validate_service_url.return_value = (True, None) mock_instance = MockApprise.return_value mock_instance.add.return_value = True mock_instance.notify.return_value = False svc = NotificationService(outbound_allowed=True) result = svc.test_service("discord://id/token") assert result["success"] is False assert "Failed to send" in result["error"] @patch("local_deep_research.notifications.service.NotificationURLValidator") def test_test_service_exception(self, MockValidator): MockValidator.validate_service_url.side_effect = RuntimeError("boom") svc = NotificationService(outbound_allowed=True) result = svc.test_service("discord://id/token") assert result["success"] is False @patch("local_deep_research.notifications.service.NotificationURLValidator") def test_test_service_empty_error_msg_uses_fallback(self, MockValidator): # Defensive fallback: if the validator ever returned (False, "") # (today it never does — every False branch sets a reason), the # user still sees an actionable message rather than an empty # string. Pins the `error_msg or "Invalid..."` guard. MockValidator.validate_service_url.return_value = (False, "") svc = NotificationService(outbound_allowed=True) result = svc.test_service("https://example.com/hook") assert result["success"] is False assert result["error"] == "Invalid notification service URL." def test_test_service_master_switch_off_short_circuits(self): # When LDR_NOTIFICATIONS_ALLOW_OUTBOUND=False (the default), # test_service must short-circuit BEFORE the URL is inspected by # the validator. Pins the operator master-switch gate so a # future refactor cannot accidentally route around it. svc = NotificationService() # outbound_allowed defaults to False result = svc.test_service("http://10.0.0.1/hook") assert result["success"] is False assert "LDR_NOTIFICATIONS_ALLOW_OUTBOUND" in result["error"] # The URL itself was never inspected, so its details must not # leak into the user-facing message: assert "10.0.0.1" not in result["error"] # -- SERVICE_PATTERNS -------------------------------------------------- def test_service_patterns_are_valid_regexes(self): import re for name, pattern in NotificationService.SERVICE_PATTERNS.items(): re.compile(pattern) # Should not raise # =================================================================== # RateLimiter # =================================================================== class TestRateLimiter: """Tests for the in-memory per-user rate limiter.""" def test_basic_allow(self): rl = RateLimiter(max_per_hour=5, max_per_day=20) assert rl.is_allowed("user1") is True def test_hourly_limit_exceeded(self): rl = RateLimiter(max_per_hour=3, max_per_day=100) for _ in range(3): assert rl.is_allowed("user1") is True assert rl.is_allowed("user1") is False def test_daily_limit_exceeded(self): rl = RateLimiter(max_per_hour=100, max_per_day=3) for _ in range(3): assert rl.is_allowed("user1") is True assert rl.is_allowed("user1") is False def test_per_user_isolation(self): rl = RateLimiter(max_per_hour=2, max_per_day=100) # Exhaust user1's hourly limit rl.is_allowed("user1") rl.is_allowed("user1") assert rl.is_allowed("user1") is False # user2 should still be allowed assert rl.is_allowed("user2") is True def test_set_user_limits(self): rl = RateLimiter(max_per_hour=10, max_per_day=50) rl.set_user_limits("user1", max_per_hour=2, max_per_day=5) limits = rl.get_user_limits("user1") assert limits == (2, 5) # User with custom limits rl.is_allowed("user1") rl.is_allowed("user1") assert rl.is_allowed("user1") is False # Default user still has 10/hour for _ in range(10): assert rl.is_allowed("default_user") is True assert rl.is_allowed("default_user") is False def test_get_user_limits_defaults(self): rl = RateLimiter(max_per_hour=10, max_per_day=50) assert rl.get_user_limits("unknown") == (10, 50) def test_reset_single_user(self): rl = RateLimiter(max_per_hour=2, max_per_day=100) rl.is_allowed("user1") rl.is_allowed("user1") assert rl.is_allowed("user1") is False rl.reset("user1") assert rl.is_allowed("user1") is True def test_reset_all_users(self): rl = RateLimiter(max_per_hour=1, max_per_day=100) rl.is_allowed("user1") rl.is_allowed("user2") assert rl.is_allowed("user1") is False assert rl.is_allowed("user2") is False rl.reset() assert rl.is_allowed("user1") is True assert rl.is_allowed("user2") is True def test_old_entries_cleaned(self): rl = RateLimiter(max_per_hour=2, max_per_day=100) now = datetime.now(timezone.utc) old = now - timedelta(hours=2) # Manually inject old entries rl._hourly_counts["user1"] = deque([old, old]) rl._daily_counts["user1"] = deque([old, old]) # After cleaning old entries, the user should be allowed again assert rl.is_allowed("user1") is True def test_cleanup_inactive_users(self): rl = RateLimiter( max_per_hour=10, max_per_day=50, cleanup_interval_hours=0 ) now = datetime.now(timezone.utc) old = now - timedelta(days=8) # Force a cleanup by setting last cleanup to long ago rl._last_cleanup = now - timedelta(hours=25) rl._hourly_counts["stale_user"] = deque([old]) rl._daily_counts["stale_user"] = deque([old]) # Trigger cleanup via is_allowed rl.is_allowed("active_user") assert "stale_user" not in rl._hourly_counts assert "stale_user" not in rl._daily_counts def test_thread_safety(self): """Multiple threads hitting the rate limiter concurrently.""" rl = RateLimiter(max_per_hour=100, max_per_day=1000) results = [] errors = [] def hit(): try: for _ in range(50): rl.is_allowed("concurrent_user") results.append(True) except Exception as e: errors.append(e) threads = [threading.Thread(target=hit) for _ in range(4)] for t in threads: t.start() for t in threads: t.join(timeout=10) assert len(errors) == 0 assert len(results) == 4 # =================================================================== # NotificationManager # =================================================================== class TestNotificationManager: """Tests for the high-level notification manager.""" # Module-level autouse fixture _enable_outbound_notifications opens # the operator-level env gate for all tests in this file. def _make_manager(self, settings, user_id="test_user"): return NotificationManager(settings_snapshot=settings, user_id=user_id) # -- Initialization --------------------------------------------------- def test_init_creates_shared_rate_limiter(self, default_settings): mgr = self._make_manager(default_settings) assert NotificationManager._shared_rate_limiter is not None assert mgr._rate_limiter is NotificationManager._shared_rate_limiter def test_init_reuses_shared_rate_limiter(self, default_settings): mgr1 = self._make_manager(default_settings, "user_a") mgr2 = self._make_manager(default_settings, "user_b") assert mgr1._rate_limiter is mgr2._rate_limiter def test_init_configures_per_user_limits(self, default_settings): mgr = self._make_manager(default_settings, "user_a") limits = mgr._rate_limiter.get_user_limits("user_a") assert limits == (10, 50) # -- _get_setting ------------------------------------------------------ def test_get_setting_returns_value(self, default_settings): mgr = self._make_manager(default_settings) assert mgr._get_setting("notifications.rate_limit_per_hour") == 10 def test_get_setting_returns_default(self, default_settings): mgr = self._make_manager(default_settings) assert ( mgr._get_setting("nonexistent.key", default="fallback") == "fallback" ) # -- _should_notify ---------------------------------------------------- def test_should_notify_enabled(self, default_settings): mgr = self._make_manager(default_settings) assert mgr._should_notify(EventType.RESEARCH_COMPLETED) is True def test_should_notify_disabled(self, disabled_settings): mgr = self._make_manager(disabled_settings) assert mgr._should_notify(EventType.RESEARCH_COMPLETED) is False def test_should_notify_missing_key(self, default_settings): mgr = self._make_manager(default_settings) # EventType.TEST is not in settings, should default to False assert mgr._should_notify(EventType.TEST) is False # -- send_notification ------------------------------------------------- @patch.object(NotificationService, "send_event", return_value=True) def test_send_notification_success(self, mock_send_event, default_settings): mgr = self._make_manager(default_settings) result = mgr.send_notification( EventType.RESEARCH_COMPLETED, {"query": "test", "research_id": "123"}, ) assert result is True mock_send_event.assert_called_once() def test_send_notification_disabled_returns_false(self, disabled_settings): mgr = self._make_manager(disabled_settings) result = mgr.send_notification( EventType.RESEARCH_COMPLETED, {"query": "test"}, ) assert result is False def test_send_notification_no_service_url_returns_false(self): settings = { "notifications.service_url": "", "notifications.on_research_completed": True, "notifications.rate_limit_per_hour": 10, "notifications.rate_limit_per_day": 50, } mgr = self._make_manager(settings) result = mgr.send_notification( EventType.RESEARCH_COMPLETED, {"query": "test"}, ) assert result is False def test_send_notification_rate_limited(self): settings = { "notifications.service_url": "discord://id/token", "notifications.on_research_completed": True, "notifications.rate_limit_per_hour": 1, "notifications.rate_limit_per_day": 50, } mgr = self._make_manager(settings, "rate_test_user") with patch.object(NotificationService, "send_event", return_value=True): mgr.send_notification( EventType.RESEARCH_COMPLETED, {"query": "first"}, ) with pytest.raises(RateLimitError): mgr.send_notification( EventType.RESEARCH_COMPLETED, {"query": "second"}, ) @patch.object(NotificationService, "send_event", return_value=True) def test_send_notification_force_bypasses_rate_limit(self, mock_send): settings = { "notifications.service_url": "discord://id/token", "notifications.on_research_completed": True, "notifications.rate_limit_per_hour": 1, "notifications.rate_limit_per_day": 50, } mgr = self._make_manager(settings, "force_user") mgr.send_notification(EventType.RESEARCH_COMPLETED, {"query": "first"}) # Would normally be rate-limited, but force=True result = mgr.send_notification( EventType.RESEARCH_COMPLETED, {"query": "second"}, force=True ) assert result is True @patch.object(NotificationService, "send_event", return_value=True) def test_send_notification_force_bypasses_disabled(self, mock_send): settings = { "notifications.service_url": "discord://id/token", "notifications.on_research_completed": False, "notifications.rate_limit_per_hour": 10, "notifications.rate_limit_per_day": 50, } mgr = self._make_manager(settings, "force_user2") result = mgr.send_notification( EventType.RESEARCH_COMPLETED, {"query": "forced"}, force=True ) assert result is True @patch.object( NotificationService, "send_event", side_effect=Exception("boom") ) def test_send_notification_exception_returns_false( self, mock_send, default_settings ): mgr = self._make_manager(default_settings) result = mgr.send_notification( EventType.RESEARCH_COMPLETED, {"query": "test"}, ) assert result is False # -- test_service via manager ------------------------------------------ @patch.object( NotificationService, "test_service", return_value={"success": True} ) def test_manager_test_service_delegates(self, mock_test, default_settings): mgr = self._make_manager(default_settings) result = mgr.test_service("discord://id/token") assert result["success"] is True mock_test.assert_called_once_with("discord://id/token") # -- _log_notification ------------------------------------------------- def test_log_notification_with_query(self, default_settings): mgr = self._make_manager(default_settings) # Should not raise mgr._log_notification( EventType.RESEARCH_COMPLETED, {"query": "test query"}, ) def test_log_notification_with_subscription(self, default_settings): mgr = self._make_manager(default_settings) mgr._log_notification( EventType.SUBSCRIPTION_UPDATE, {"subscription_name": "my sub"}, ) def test_log_notification_empty_context(self, default_settings): mgr = self._make_manager(default_settings) mgr._log_notification(EventType.TEST, {}) # =================================================================== # NotificationTemplate # =================================================================== class TestNotificationTemplate: """Tests for Jinja2-based notification templates.""" def test_format_returns_title_and_body(self): result = NotificationTemplate.format( EventType.RESEARCH_COMPLETED, { "query": "quantum computing", "research_id": "abc", "summary": "results", "url": "http://example.com", }, ) assert "title" in result assert "body" in result assert isinstance(result["title"], str) assert isinstance(result["body"], str) def test_format_with_custom_template(self): custom = { "title": "Custom: {query}", "body": "Research {research_id} is done.", } result = NotificationTemplate.format( EventType.RESEARCH_COMPLETED, {"query": "test", "research_id": "123"}, custom_template=custom, ) assert result["title"] == "Custom: test" assert result["body"] == "Research 123 is done." def test_format_custom_template_missing_var(self): custom = { "title": "{missing_var}", "body": "Body", } result = NotificationTemplate.format( EventType.RESEARCH_COMPLETED, {}, custom_template=custom, ) assert ( "Missing variable" in result["body"] or "error" in result["body"].lower() ) def test_format_custom_template_sanitizes_context(self): """Ensure format string attacks are prevented.""" custom = { "title": "{query}", "body": "Body", } result = NotificationTemplate.format( EventType.RESEARCH_COMPLETED, {"query": "{__class__.__mro__}"}, custom_template=custom, ) # The attack payload should be treated as a plain string assert "__class__" in result["title"] def test_fallback_template(self): result = NotificationTemplate._get_fallback_template( EventType.RESEARCH_COMPLETED, {"query": "test"}, ) assert "title" in result assert "body" in result assert "Research Completed" in result["title"] def test_event_type_values(self): """All EventType values are strings.""" for et in EventType: assert isinstance(et.value, str) def test_get_required_context_returns_list(self): result = NotificationTemplate.get_required_context( EventType.RESEARCH_COMPLETED ) assert isinstance(result, list) def test_format_unknown_event_type(self): """Event type without a template file should produce a generic message.""" # RATE_LIMIT_WARNING has no template file in TEMPLATE_FILES result = NotificationTemplate.format( EventType.RATE_LIMIT_WARNING, {"detail": "too many requests"}, ) assert "title" in result assert "body" in result # =================================================================== # build_notification_url # =================================================================== class TestBuildNotificationUrl: """Tests for notification callback URL building.""" def test_with_external_url(self): snapshot = {"app.external_url": "https://myapp.example.com"} url = build_notification_url( "/research/123", settings_snapshot=snapshot, ) assert url == "https://myapp.example.com/research/123" def test_with_host_and_port(self): snapshot = { "app.external_url": "", "app.host": "0.0.0.0", "app.port": 5000, } url = build_notification_url( "/research/456", settings_snapshot=snapshot, ) assert url == "http://localhost:5000/research/456" def test_fallback_url(self): url = build_notification_url( "/research/789", settings_snapshot={}, fallback_base="http://localhost:5000", ) assert url == "http://localhost:5000/research/789" def test_validation_disabled(self): url = build_notification_url( "/research/abc", settings_snapshot={}, validate=False, ) assert "/research/abc" in url def test_validation_catches_bad_url(self): with pytest.raises(URLValidationError): build_notification_url( "/research/123", settings_snapshot={"app.external_url": "ftp://bad.example.com"}, validate=True, ) # =================================================================== # Queue helpers # =================================================================== class TestQueueHelpers: """Tests for queue notification helper functions.""" @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_notification_success(self, MockManager): mock_instance = MockManager.return_value mock_instance.send_notification.return_value = True settings = { "notifications.rate_limit_per_hour": 10, "notifications.rate_limit_per_day": 50, } result = send_queue_notification( username="alice", research_id="r1", query="test query", settings_snapshot=settings, position=3, ) assert result is True mock_instance.send_notification.assert_called_once() @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_notification_rate_limited(self, MockManager): mock_instance = MockManager.return_value mock_instance.send_notification.side_effect = RateLimitError("limited") result = send_queue_notification( username="alice", research_id="r1", query="test", settings_snapshot={}, ) assert result is False @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_notification_exception(self, MockManager): MockManager.side_effect = Exception("boom") result = send_queue_notification( username="alice", research_id="r1", query="test", settings_snapshot={}, ) assert result is False def test_send_queue_failed_no_snapshot(self): result = send_queue_failed_notification( username="alice", research_id="r1", query="test", settings_snapshot=None, ) assert result is False @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_failed_success(self, MockManager): mock_instance = MockManager.return_value mock_instance.send_notification.return_value = True result = send_queue_failed_notification( username="bob", research_id="r2", query="failed query", error_message="timeout", settings_snapshot={"notifications.rate_limit_per_hour": 10}, ) assert result is True @patch( "local_deep_research.notifications.queue_helpers.NotificationManager" ) def test_send_queue_failed_rate_limited(self, MockManager): mock_instance = MockManager.return_value mock_instance.send_notification.side_effect = RateLimitError("limited") result = send_queue_failed_notification( username="bob", research_id="r2", query="test", settings_snapshot={"x": 1}, ) assert result is False # =================================================================== # mask_sensitive_url # =================================================================== class TestMaskSensitiveUrl: """Tests for URL credential masking.""" def test_masks_password(self): url = "https://user:secretpassword@example.com/path" masked = mask_sensitive_url(url) assert "secretpassword" not in masked assert "***" in masked def test_masks_long_tokens_in_path(self): url = "discord://12345678901234567890123456/abcdefghijklmnopqrstuvwxyz" masked = mask_sensitive_url(url) # Long alphanumeric tokens should be masked assert "***" in masked def test_masks_query_string(self): url = "https://hooks.example.com/webhook?token=secret123" masked = mask_sensitive_url(url) assert "secret123" not in masked def test_preserves_scheme(self): url = "https://example.com/path" masked = mask_sensitive_url(url) assert masked.startswith("https://") def test_handles_invalid_url(self): # Should not raise, returns something with *** result = mask_sensitive_url("not a url at all") assert isinstance(result, str) def test_simple_url_no_secrets(self): url = "https://example.com/short" masked = mask_sensitive_url(url) assert "example.com" in masked # =================================================================== # Exception hierarchy # =================================================================== class TestExceptionHierarchy: """Verify exception inheritance is correct.""" def test_service_error_is_notification_error(self): assert issubclass(ServiceError, NotificationError) def test_send_error_is_notification_error(self): assert issubclass(SendError, NotificationError) def test_rate_limit_error_is_notification_error(self): assert issubclass(RateLimitError, NotificationError) def test_notification_error_is_exception(self): assert issubclass(NotificationError, Exception) def test_notification_url_validation_error_is_value_error(self): assert issubclass(NotificationURLValidationError, ValueError) def test_url_validation_error_is_value_error(self): assert issubclass(URLValidationError, ValueError) def test_exceptions_carry_message(self): e = SendError("test message") assert str(e) == "test message" # =================================================================== # URLValidator (callback URL validation) # =================================================================== class TestURLValidatorHTTP: """Tests for validate_http_url used in notification callback URLs.""" def test_valid_https(self): assert ( URLValidator.validate_http_url("https://example.com/path") is True ) def test_valid_http(self): assert URLValidator.validate_http_url("http://example.com/path") is True def test_rejects_ftp(self): with pytest.raises(URLValidationError): URLValidator.validate_http_url("ftp://example.com") def test_rejects_empty(self): with pytest.raises(URLValidationError): URLValidator.validate_http_url("") def test_rejects_none(self): with pytest.raises(URLValidationError): URLValidator.validate_http_url(None) def test_rejects_no_scheme(self): with pytest.raises(URLValidationError): URLValidator.validate_http_url("example.com/path") def test_rejects_javascript(self): with pytest.raises(URLValidationError): URLValidator.validate_http_url("javascript:alert(1)")