dify/api/tests/unit_tests/controllers/trigger/test_webhook.py

import types
from unittest.mock import patch

import pytest
from werkzeug.exceptions import NotFound, RequestEntityTooLarge

import controllers.trigger.webhook as module
from models.trigger import WorkflowWebhookTrigger
from models.workflow import Workflow
from services.errors.app import QuotaExceededError


@pytest.fixture(autouse=True)
def mock_request():
    module.request = types.SimpleNamespace(
        method="POST",
        headers={"x-test": "1"},
        args={"a": "b"},
    )


@pytest.fixture(autouse=True)
def mock_jsonify():
    module.jsonify = lambda payload: payload


def _webhook_trigger() -> WorkflowWebhookTrigger:
    return WorkflowWebhookTrigger(
        webhook_id="wh-1",
        tenant_id="tenant-1",
        app_id="app-1",
        node_id="node-1",
        created_by="account-1",
    )


def _workflow() -> Workflow:
    return Workflow(id="workflow-1")


class TestPrepareWebhookExecution:
    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data")
    def test_prepare_success(self, mock_extract, mock_get):
        webhook_trigger = _webhook_trigger()
        workflow = _workflow()
        mock_get.return_value = (webhook_trigger, workflow, "node_config")
        mock_extract.return_value = {"data": "ok"}

        result = module._prepare_webhook_execution("wh-1")

        assert result == (webhook_trigger, workflow, "node_config", {"data": "ok"}, None)

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data", side_effect=ValueError("bad"))
    def test_prepare_validation_error(self, mock_extract, mock_get):
        webhook_trigger = _webhook_trigger()
        workflow = _workflow()
        mock_get.return_value = (webhook_trigger, workflow, "node_config")

        trigger, workflow, node_config, webhook_data, error = module._prepare_webhook_execution("wh-1")

        assert error == "bad"
        assert webhook_data["method"] == "POST"


class TestHandleWebhook:
    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data")
    @patch.object(module.WebhookService, "trigger_workflow_execution")
    @patch.object(module.WebhookService, "generate_webhook_response")
    def test_success(
        self,
        mock_generate,
        mock_trigger,
        mock_extract,
        mock_get,
    ):
        mock_get.return_value = (_webhook_trigger(), _workflow(), "node_config")
        mock_extract.return_value = {"input": "x"}
        mock_generate.return_value = ({"ok": True}, 200)

        response, status = module.handle_webhook("wh-1")

        assert status == 200
        assert response["ok"] is True
        mock_trigger.assert_called_once()

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data", side_effect=ValueError("bad"))
    def test_bad_request(self, mock_extract, mock_get):
        mock_get.return_value = (_webhook_trigger(), _workflow(), "node_config")

        response, status = module.handle_webhook("wh-1")

        assert status == 400
        assert response["error"] == "Bad Request"

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data")
    @patch.object(
        module.WebhookService,
        "trigger_workflow_execution",
        side_effect=QuotaExceededError(feature="trigger", tenant_id="tenant-1", required=1),
    )
    def test_quota_exceeded(self, mock_trigger, mock_extract, mock_get):
        mock_get.return_value = (_webhook_trigger(), _workflow(), "node_config")
        mock_extract.return_value = {"input": "x"}

        response, status = module.handle_webhook("wh-1")

        assert status == 429
        assert response == {
            "error": "Too Many Requests",
            "message": "Trigger event quota exceeded. Please upgrade your plan.",
        }

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data")
    @patch.object(
        module.WebhookService,
        "trigger_workflow_execution",
        side_effect=QuotaExceededError(feature="workflow", tenant_id="tenant-1", required=1),
    )
    def test_workflow_quota_exceeded(self, mock_trigger, mock_extract, mock_get):
        mock_get.return_value = (_webhook_trigger(), _workflow(), "node_config")
        mock_extract.return_value = {"input": "x"}

        response, status = module.handle_webhook("wh-1")

        assert status == 429
        assert response == {
            "error": "Too Many Requests",
            "message": "Workflow execution quota exceeded. Please upgrade your plan.",
        }

    @patch.object(
        module.WebhookService,
        "get_webhook_trigger_and_workflow",
        side_effect=QuotaExceededError(feature="trigger", tenant_id="tenant-1", required=1),
    )
    def test_rate_limited(self, mock_get):
        response, status = module.handle_webhook("wh-1")

        assert status == 429
        assert response == {
            "error": "Too Many Requests",
            "message": "Trigger event quota exceeded. Please upgrade your plan.",
        }

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow", side_effect=ValueError("missing"))
    def test_value_error_not_found(self, mock_get):
        with pytest.raises(NotFound):
            module.handle_webhook("wh-1")

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow", side_effect=RequestEntityTooLarge())
    def test_request_entity_too_large(self, mock_get):
        with pytest.raises(RequestEntityTooLarge):
            module.handle_webhook("wh-1")

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow", side_effect=Exception("boom"))
    def test_internal_error(self, mock_get):
        response, status = module.handle_webhook("wh-1")

        assert status == 500
        assert response["error"] == "Internal server error"


class TestHandleWebhookDebug:
    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data")
    @patch.object(module.WebhookService, "build_workflow_inputs", return_value={"x": 1})
    @patch.object(module.TriggerDebugEventBus, "dispatch", return_value=0)
    def test_debug_requires_active_listener(
        self,
        mock_dispatch,
        mock_build_inputs,
        mock_extract,
        mock_get,
    ):
        webhook_trigger = _webhook_trigger()
        mock_get.return_value = (webhook_trigger, None, "node_config")
        mock_extract.return_value = {"method": "POST"}

        response, status = module.handle_webhook_debug("wh-1")

        assert status == 409
        assert response["error"] == "No active debug listener"
        assert response["message"] == (
            "The webhook debug URL only works while the Variable Inspector is listening. "
            "Use the published webhook URL to execute the workflow in Celery."
        )
        assert response["execution_url"] == webhook_trigger.webhook_url
        mock_dispatch.assert_called_once()

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data")
    @patch.object(module.WebhookService, "build_workflow_inputs", return_value={"x": 1})
    @patch.object(module.TriggerDebugEventBus, "dispatch", return_value=1)
    @patch.object(module.WebhookService, "generate_webhook_response")
    def test_debug_success(
        self,
        mock_generate,
        mock_dispatch,
        mock_build_inputs,
        mock_extract,
        mock_get,
    ):
        mock_get.return_value = (_webhook_trigger(), None, "node_config")
        mock_extract.return_value = {"method": "POST"}
        mock_generate.return_value = ({"ok": True}, 200)

        response, status = module.handle_webhook_debug("wh-1")

        assert status == 200
        assert response["ok"] is True
        mock_dispatch.assert_called_once()

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
    @patch.object(module.WebhookService, "extract_and_validate_webhook_data", side_effect=ValueError("bad"))
    def test_debug_bad_request(self, mock_extract, mock_get):
        mock_get.return_value = (_webhook_trigger(), None, "node_config")

        response, status = module.handle_webhook_debug("wh-1")

        assert status == 400
        assert response["error"] == "Bad Request"

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow", side_effect=ValueError("missing"))
    def test_debug_not_found(self, mock_get):
        with pytest.raises(NotFound):
            module.handle_webhook_debug("wh-1")

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow", side_effect=RequestEntityTooLarge())
    def test_debug_request_entity_too_large(self, mock_get):
        with pytest.raises(RequestEntityTooLarge):
            module.handle_webhook_debug("wh-1")

    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow", side_effect=Exception("boom"))
    def test_debug_internal_error(self, mock_get):
        response, status = module.handle_webhook_debug("wh-1")

        assert status == 500
        assert response["error"] == "Internal server error"