learningcircuit--local-deep-research/tests/notifications/test_service_edge_cases.py

"""
Tests for NotificationService edge cases: test_service() error paths
and get_service_type() pattern coverage.
"""

from unittest.mock import patch, MagicMock

from local_deep_research.notifications.service import NotificationService


class TestTestServiceErrorPaths:
    """Tests for test_service() error branches."""

    @patch("local_deep_research.notifications.service.apprise.Apprise")
    @patch(
        "local_deep_research.notifications.service.NotificationURLValidator.validate_service_url",
        return_value=(
            False,
            "Blocked private/internal IP address: 169.254.169.254",
        ),
    )
    def test_ssrf_validation_fails(self, mock_validator, mock_apprise_class):
        # The classic SSRF target — AWS IMDS at 169.254.169.254 — is the
        # canonical fixture for this branch. The hint is intentionally
        # NOT asserted here: this test pins only that the validator's
        # reason reaches the user. Hint suppression for always-blocked
        # categories (metadata, 6to4, Teredo, etc.) is pinned end-to-end
        # by test_test_service_ip_rejection_matrix in
        # test_notification_coverage.py.
        service = NotificationService(outbound_allowed=True)
        result = service.test_service("http://169.254.169.254/metadata")
        assert result["success"] is False
        # The validator's reason is surfaced verbatim — it echoes only
        # the user-supplied hostname, so it is not a server-side leak.
        assert "private/internal IP" in result["error"]
        assert "169.254.169.254" in result["error"]

    @patch("local_deep_research.notifications.service.apprise.Apprise")
    @patch(
        "local_deep_research.notifications.service.NotificationURLValidator.validate_service_url",
        return_value=(True, None),
    )
    def test_apprise_add_fails(self, mock_validator, mock_apprise_class):
        mock_instance = MagicMock()
        mock_instance.add.return_value = False
        mock_apprise_class.return_value = mock_instance

        service = NotificationService(outbound_allowed=True)
        result = service.test_service("discord://webhook/token")
        assert result["success"] is False
        assert "Failed to add" in result["error"]

    @patch("local_deep_research.notifications.service.apprise.Apprise")
    @patch(
        "local_deep_research.notifications.service.NotificationURLValidator.validate_service_url",
        return_value=(True, None),
    )
    def test_apprise_notify_fails(self, mock_validator, mock_apprise_class):
        mock_instance = MagicMock()
        mock_instance.add.return_value = True
        mock_instance.notify.return_value = False
        mock_apprise_class.return_value = mock_instance

        service = NotificationService(outbound_allowed=True)
        result = service.test_service("discord://webhook/token")
        assert result["success"] is False
        assert "Failed to send" in result["error"]

    @patch("local_deep_research.notifications.service.apprise.Apprise")
    @patch(
        "local_deep_research.notifications.service.NotificationURLValidator.validate_service_url",
        return_value=(True, None),
    )
    def test_success_path(self, mock_validator, mock_apprise_class):
        mock_instance = MagicMock()
        mock_instance.add.return_value = True
        mock_instance.notify.return_value = True
        mock_apprise_class.return_value = mock_instance

        service = NotificationService(outbound_allowed=True)
        result = service.test_service("discord://webhook/token")
        assert result["success"] is True
        assert "successfully" in result["message"]

    @patch("local_deep_research.notifications.service.apprise.Apprise")
    @patch(
        "local_deep_research.notifications.service.NotificationURLValidator.validate_service_url",
        return_value=(False, "Blocked unsafe protocol: javascript"),
    )
    def test_validation_reason_surfaced_without_internals(
        self, mock_validator, mock_apprise_class
    ):
        service = NotificationService(outbound_allowed=True)
        result = service.test_service("javascript://evil.com")
        assert result["success"] is False
        # The validator's reason IS surfaced (it echoes only the
        # user-supplied scheme). What must NEVER appear is server-internal
        # detail: stack traces, exception class names, DNS resolver
        # internals. The validator's error_msg never contains these, but
        # pin the contract so a future regression that smuggles them in
        # fails this test.
        assert "Blocked unsafe protocol" in result["error"]
        for forbidden in (
            "Traceback",
            "socket.gaierror",
            "Exception",
            "internal DNS",
            "raise ",
        ):
            assert forbidden not in result["error"]


class TestGetServiceTypePatterns:
    """Tests for get_service_type() across all SERVICE_PATTERNS."""

    def test_email_detection(self):
        service = NotificationService(outbound_allowed=True)
        assert service.get_service_type("mailto://user@example.com") == "email"

    def test_slack_detection(self):
        service = NotificationService(outbound_allowed=True)
        assert (
            service.get_service_type("slack://token_a/token_b/token_c")
            == "slack"
        )

    def test_telegram_detection(self):
        service = NotificationService(outbound_allowed=True)
        assert (
            service.get_service_type("tgram://bottoken/chat_id") == "telegram"
        )

    def test_smtp_detection(self):
        service = NotificationService(outbound_allowed=True)
        assert service.get_service_type("smtp://user:pass@mail.com") == "smtp"

    def test_smtps_detection(self):
        service = NotificationService(outbound_allowed=True)
        assert service.get_service_type("smtps://user:pass@mail.com") == "smtp"