mirror of
https://github.com/langgenius/dify.git
synced 2026-11-12 14:21:29 +08:00
362 lines
14 KiB
Python
362 lines
14 KiB
Python
from dataclasses import dataclass
|
|
from io import BytesIO
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from flask import Flask
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
from werkzeug.exceptions import NotFound, RequestEntityTooLarge
|
|
|
|
import controllers.trigger.webhook as module
|
|
from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
|
|
from core.app.file_access import DatabaseFileAccessController, FileAccessScope, bind_file_access_scope
|
|
from factories import file_factory
|
|
from models.enums import DEFAULT_END_USER_SESSION_ID, EndUserType
|
|
from models.model import EndUser
|
|
from models.trigger import WorkflowWebhookTrigger
|
|
from models.workflow import Workflow
|
|
from repositories.trigger.app_repository import SQLAlchemyAppTriggerRepository
|
|
from services.errors.app import QuotaExceededError
|
|
from services.trigger.app_trigger_service import AppTriggerService
|
|
from tests.helpers.unit.model_factories import make_end_user, make_workflow
|
|
|
|
|
|
class _RequestStub:
|
|
method = "POST"
|
|
headers = {"x-test": "1"}
|
|
args = {"a": "b"}
|
|
|
|
|
|
@dataclass(eq=False, repr=False)
|
|
class _AppScopedEndUserServicesStub:
|
|
commands: object
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _AppServicesStub:
|
|
trigger_status: AppTriggerService
|
|
|
|
|
|
@dataclass(eq=False, repr=False)
|
|
class _WorkspaceServicesStub:
|
|
workflow_dispatch: object
|
|
|
|
|
|
class _ApplicationServicesStub:
|
|
def __init__(self, commands: object, trigger_status: AppTriggerService) -> None:
|
|
self.app_scoped_end_users = _AppScopedEndUserServicesStub(commands)
|
|
self.apps = _AppServicesStub(trigger_status)
|
|
self.workspaces = _WorkspaceServicesStub(object())
|
|
|
|
|
|
class _EndUserCommandsStub:
|
|
def __init__(self, end_user: EndUser) -> None:
|
|
self._end_user = end_user
|
|
self.calls: list[tuple[EndUserType, str, str, str | None]] = []
|
|
|
|
def get_or_create_end_user_by_type(
|
|
self,
|
|
type: EndUserType,
|
|
tenant_id: str,
|
|
app_id: str,
|
|
user_id: str | None = None,
|
|
) -> EndUser:
|
|
self.calls.append((type, tenant_id, app_id, user_id))
|
|
return self._end_user
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_request():
|
|
module.request = _RequestStub()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_jsonify():
|
|
module.jsonify = lambda payload: payload
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def end_user_commands(monkeypatch: pytest.MonkeyPatch, sqlite_session_factory: sessionmaker[Session]) -> object:
|
|
commands = object()
|
|
triggers = AppTriggerService(triggers=SQLAlchemyAppTriggerRepository(session_factory=sqlite_session_factory))
|
|
monkeypatch.setattr(module, "application_services", lambda: _ApplicationServicesStub(commands, triggers))
|
|
return commands
|
|
|
|
|
|
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 make_workflow(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"
|
|
|
|
@patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
|
|
def test_prepare_stores_uploads_for_the_trigger_end_user_run(
|
|
self,
|
|
mock_get: MagicMock,
|
|
app: Flask,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
sqlite_session_factory: sessionmaker[Session],
|
|
) -> None:
|
|
node_config = {
|
|
"data": {
|
|
"method": "post",
|
|
"content_type": "multipart/form-data",
|
|
"body": [{"name": "chat_history_file", "type": "file", "required": True}],
|
|
}
|
|
}
|
|
mock_get.return_value = (_webhook_trigger(), _workflow(), node_config)
|
|
trigger_end_user = make_end_user(
|
|
end_user_id="trigger-end-user-1",
|
|
tenant_id="tenant-1",
|
|
app_id="app-1",
|
|
end_user_type=EndUserType.TRIGGER,
|
|
session_id=DEFAULT_END_USER_SESSION_ID,
|
|
)
|
|
end_users = _EndUserCommandsStub(trigger_end_user)
|
|
triggers = AppTriggerService(triggers=SQLAlchemyAppTriggerRepository(session_factory=sqlite_session_factory))
|
|
monkeypatch.setattr(module, "application_services", lambda: _ApplicationServicesStub(end_users, triggers))
|
|
monkeypatch.setattr("core.tools.tool_file_manager.storage.save", lambda *_args: None)
|
|
|
|
with app.test_request_context(
|
|
"/triggers/webhook/wh-1",
|
|
method="POST",
|
|
data={"chat_history_file": (BytesIO(b"# chat"), "chat.md", "text/markdown")},
|
|
content_type="multipart/form-data",
|
|
):
|
|
_, _, _, webhook_data, error = module._prepare_webhook_execution("wh-1")
|
|
|
|
assert error is None
|
|
tool_file_id = webhook_data["files"]["chat_history_file"]["related_id"]
|
|
# The async run executes as the trigger end user, whose scope only loads tool files it owns.
|
|
run_scope = FileAccessScope(
|
|
tenant_id="tenant-1",
|
|
user_id=trigger_end_user.id,
|
|
user_from=UserFrom.END_USER,
|
|
invoke_from=InvokeFrom.SERVICE_API,
|
|
)
|
|
with bind_file_access_scope(run_scope):
|
|
file = file_factory.build_from_mapping(
|
|
mapping={"transfer_method": "tool_file", "tool_file_id": tool_file_id},
|
|
tenant_id="tenant-1",
|
|
access_controller=DatabaseFileAccessController(),
|
|
)
|
|
assert file.related_id == tool_file_id
|
|
assert end_users.calls == [(EndUserType.TRIGGER, "tenant-1", "app-1", None)]
|
|
|
|
|
|
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,
|
|
end_user_commands,
|
|
):
|
|
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()
|
|
assert mock_trigger.call_args.kwargs["end_users"] is end_user_commands
|
|
|
|
@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):
|
|
response = module.handle_webhook("wh-1")
|
|
assert response.status_code == 404
|
|
assert response.data == NotFound().get_body().encode()
|
|
|
|
@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):
|
|
response = module.handle_webhook_debug("wh-1")
|
|
assert response.status_code == 404
|
|
assert response.data == NotFound().get_body().encode()
|
|
|
|
@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"
|