learningcircuit--local-deep-research/tests/web/services/test_notification_coverage.py

"""
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,<script>alert(1)</script>",
            "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)")