agentic-soc-platform/backend/apps/webhook/tests.py

from unittest.mock import patch

from django.test import SimpleTestCase, TestCase
from django.urls import Resolver404, resolve
from rest_framework.test import APIClient

from apps.settings.models import RuntimeConfig
from apps.settings.runtime_config import invalidate
from apps.webhook.schemas import WebhookResult
from apps.webhook.service import (
    WebhookRedisError,
    handle_kibana_webhook,
    handle_splunk_webhook,
)


class FakeRedisClient:
    def __init__(self):
        self.messages = []

    def send_message(self, stream, data, *, maxlen=None):
        self.messages.append({"stream": stream, "data": data, "maxlen": maxlen})
        return f"{len(self.messages)}-0"


class FailingRedisClient:
    def send_message(self, stream, data, *, maxlen=None):
        raise RuntimeError("redis unavailable")


class WebhookServiceTests(TestCase):
    def set_stream_maxlen(self, value):
        config = RuntimeConfig.get_current()
        config.stream_maxlen = value
        config.save(update_fields=["stream_maxlen", "updated_at"])
        invalidate("runtime")

    def test_handle_splunk_webhook_writes_result_to_search_stream(self):
        self.set_stream_maxlen(123)
        redis_client = FakeRedisClient()

        result = handle_splunk_webhook(
            {"search_name": " test-search ", "result": {"host": "server-1"}},
            redis_client=redis_client,
        )

        self.assertEqual(result.stream, "test-search")
        self.assertEqual(result.sent, 1)
        self.assertEqual(result.skipped, 0)
        self.assertEqual(result.message_ids, ["1-0"])
        self.assertEqual(
            redis_client.messages,
            [{"stream": "test-search", "data": {"host": "server-1"}, "maxlen": 123}],
        )

    def test_handle_splunk_webhook_rejects_empty_search_name(self):
        with self.assertRaisesMessage(ValueError, "search_name is required."):
            handle_splunk_webhook({"search_name": " ", "result": {}}, redis_client=FakeRedisClient())

    def test_handle_kibana_webhook_writes_normalized_sources(self):
        self.set_stream_maxlen(456)
        redis_client = FakeRedisClient()

        result = handle_kibana_webhook(
            {
                "rule": {"name": " kibana-rule "},
                "context": {
                    "hits": [
                        {
                            "_source": {
                                "user": {"keyword": "alice"},
                                "host.keyword": "ignored",
                                "tags": [{"keyword": "prod"}],
                            }
                        }
                    ]
                },
            },
            redis_client=redis_client,
        )

        self.assertEqual(result.stream, "kibana-rule")
        self.assertEqual(result.sent, 1)
        self.assertEqual(result.skipped, 0)
        self.assertEqual(result.message_ids, ["1-0"])
        self.assertEqual(
            redis_client.messages,
            [{"stream": "kibana-rule", "data": {"user": "alice", "tags": ["prod"]}, "maxlen": 456}],
        )

    def test_handle_kibana_webhook_skips_empty_sources(self):
        redis_client = FakeRedisClient()

        result = handle_kibana_webhook(
            {"rule": {"name": "rule"}, "context": {"hits": [{"_source": {}}, None]}},
            redis_client=redis_client,
        )

        self.assertEqual(result.sent, 0)
        self.assertEqual(result.skipped, 2)
        self.assertEqual(redis_client.messages, [])

    def test_handle_kibana_webhook_raises_webhook_error_on_redis_failure(self):
        with self.assertRaisesMessage(WebhookRedisError, "Failed to write Redis stream rule: RuntimeError"):
            handle_kibana_webhook(
                {"rule": {"name": "rule"}, "context": {"hits": [{"_source": {"event": "login"}}]}},
                redis_client=FailingRedisClient(),
            )


class WebhookAPITests(SimpleTestCase):
    def setUp(self):
        self.client = APIClient()

    @patch("apps.webhook.views.handle_splunk_webhook")
    def test_splunk_webhook_route_uses_new_api_path(self, handle_splunk_webhook_mock):
        handle_splunk_webhook_mock.return_value = WebhookResult(stream="search", sent=1, message_ids=["1-0"])

        response = self.client.post(
            "/api/webhook/splunk/",
            {"search_name": "search", "result": {"host": "server-1"}},
            format="json",
        )

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["stream"], "search")
        handle_splunk_webhook_mock.assert_called_once()

    @patch("apps.webhook.views.handle_kibana_webhook")
    def test_kibana_webhook_route_uses_new_api_path(self, handle_kibana_webhook_mock):
        handle_kibana_webhook_mock.return_value = WebhookResult(stream="rule", sent=1, message_ids=["1-0"])

        response = self.client.post(
            "/api/webhook/kibana/",
            {"rule": {"name": "rule"}, "context": {"hits": [{"_source": {"host": "server-1"}}]}},
            format="json",
        )

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["stream"], "rule")
        handle_kibana_webhook_mock.assert_called_once()

    @patch("apps.webhook.views.handle_splunk_webhook")
    def test_webhook_redis_error_returns_service_unavailable(self, handle_splunk_webhook_mock):
        handle_splunk_webhook_mock.side_effect = WebhookRedisError("Failed to write Redis stream search: RuntimeError")

        response = self.client.post(
            "/api/webhook/splunk/",
            {"search_name": "search", "result": {"host": "server-1"}},
            format="json",
        )

        self.assertEqual(response.status_code, 503)
        self.assertEqual(response.json()["detail"], "Failed to write Redis stream search: RuntimeError")

    def test_old_agentic_forwarder_route_is_removed(self):
        with self.assertRaises(Resolver404):
            resolve("/api/agentic/forwarder/splunk/")