mirror of
https://github.com/langgenius/dify.git
synced 2026-08-29 03:45:08 +08:00
test: migrate trigger and webhook sessions and ORM models to SQLite (#40594)
This commit is contained in:
Asuka Minato
@@ -5,6 +5,8 @@ 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
|
||||
|
||||
|
||||
@@ -22,29 +24,39 @@ def mock_jsonify():
|
||||
module.jsonify = lambda payload: payload
|
||||
|
||||
|
||||
class DummyWebhookTrigger:
|
||||
webhook_id = "wh-1"
|
||||
webhook_url = "http://localhost:5001/triggers/webhook/wh-1"
|
||||
tenant_id = "tenant-1"
|
||||
app_id = "app-1"
|
||||
node_id = "node-1"
|
||||
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):
|
||||
mock_get.return_value = ("trigger", "workflow", "node_config")
|
||||
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 == ("trigger", "workflow", "node_config", {"data": "ok"}, None)
|
||||
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):
|
||||
mock_get.return_value = ("trigger", "workflow", "node_config")
|
||||
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")
|
||||
|
||||
@@ -64,7 +76,7 @@ class TestHandleWebhook:
|
||||
mock_extract,
|
||||
mock_get,
|
||||
):
|
||||
mock_get.return_value = (DummyWebhookTrigger(), "workflow", "node_config")
|
||||
mock_get.return_value = (_webhook_trigger(), _workflow(), "node_config")
|
||||
mock_extract.return_value = {"input": "x"}
|
||||
mock_generate.return_value = ({"ok": True}, 200)
|
||||
|
||||
@@ -77,7 +89,7 @@ class TestHandleWebhook:
|
||||
@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 = (DummyWebhookTrigger(), "workflow", "node_config")
|
||||
mock_get.return_value = (_webhook_trigger(), _workflow(), "node_config")
|
||||
|
||||
response, status = module.handle_webhook("wh-1")
|
||||
|
||||
@@ -92,7 +104,7 @@ class TestHandleWebhook:
|
||||
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 = (DummyWebhookTrigger(), "workflow", "node_config")
|
||||
mock_get.return_value = (_webhook_trigger(), _workflow(), "node_config")
|
||||
mock_extract.return_value = {"input": "x"}
|
||||
|
||||
response, status = module.handle_webhook("wh-1")
|
||||
@@ -111,7 +123,7 @@ class TestHandleWebhook:
|
||||
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 = (DummyWebhookTrigger(), "workflow", "node_config")
|
||||
mock_get.return_value = (_webhook_trigger(), _workflow(), "node_config")
|
||||
mock_extract.return_value = {"input": "x"}
|
||||
|
||||
response, status = module.handle_webhook("wh-1")
|
||||
@@ -166,7 +178,8 @@ class TestHandleWebhookDebug:
|
||||
mock_extract,
|
||||
mock_get,
|
||||
):
|
||||
mock_get.return_value = (DummyWebhookTrigger(), None, "node_config")
|
||||
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")
|
||||
@@ -177,7 +190,7 @@ class TestHandleWebhookDebug:
|
||||
"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"] == DummyWebhookTrigger.webhook_url
|
||||
assert response["execution_url"] == webhook_trigger.webhook_url
|
||||
mock_dispatch.assert_called_once()
|
||||
|
||||
@patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
|
||||
@@ -193,7 +206,7 @@ class TestHandleWebhookDebug:
|
||||
mock_extract,
|
||||
mock_get,
|
||||
):
|
||||
mock_get.return_value = (DummyWebhookTrigger(), None, "node_config")
|
||||
mock_get.return_value = (_webhook_trigger(), None, "node_config")
|
||||
mock_extract.return_value = {"method": "POST"}
|
||||
mock_generate.return_value = ({"ok": True}, 200)
|
||||
|
||||
@@ -206,7 +219,7 @@ class TestHandleWebhookDebug:
|
||||
@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 = (DummyWebhookTrigger(), None, "node_config")
|
||||
mock_get.return_value = (_webhook_trigger(), None, "node_config")
|
||||
|
||||
response, status = module.handle_webhook_debug("wh-1")
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ and select_trigger_debug_events orchestrator.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
@@ -27,7 +28,9 @@ from core.trigger.debug.event_selectors import (
|
||||
select_trigger_debug_events,
|
||||
)
|
||||
from core.trigger.debug.events import PluginTriggerDebugEvent, WebhookDebugEvent
|
||||
from graphon.enums import BuiltinNodeTypes, NodeType
|
||||
from graphon.enums import BuiltinNodeTypes, NodeType, WorkflowType
|
||||
from models.model import App, AppMode
|
||||
from models.workflow import Workflow
|
||||
from tests.unit_tests.core.trigger.conftest import VALID_PROVIDER_ID
|
||||
|
||||
|
||||
@@ -55,6 +58,35 @@ def _plugin_node_config(provider_id: str = VALID_PROVIDER_ID) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _workflow_with_node(node_type: NodeType | None) -> Workflow:
|
||||
nodes = [] if node_type is None else [{"id": "n1", "data": {"type": node_type, "title": "Trigger"}}]
|
||||
return Workflow.new(
|
||||
tenant_id="t1",
|
||||
app_id="a1",
|
||||
type=WorkflowType.WORKFLOW.value,
|
||||
version=Workflow.VERSION_DRAFT,
|
||||
graph=json.dumps({"nodes": nodes, "edges": []}),
|
||||
features="{}",
|
||||
created_by="u1",
|
||||
environment_variables=[],
|
||||
conversation_variables=[],
|
||||
rag_pipeline_variables=[],
|
||||
)
|
||||
|
||||
|
||||
def _app() -> App:
|
||||
return App(
|
||||
id="a1",
|
||||
tenant_id="t1",
|
||||
name="Trigger App",
|
||||
description="",
|
||||
mode=AppMode.WORKFLOW,
|
||||
enable_site=True,
|
||||
enable_api=True,
|
||||
max_active_requests=0,
|
||||
)
|
||||
|
||||
|
||||
class TestPluginTriggerDebugEventPoller:
|
||||
@patch("core.trigger.debug.event_selectors.TriggerDebugEventBus")
|
||||
def test_returns_workflow_args_on_success(self, mock_bus):
|
||||
@@ -214,51 +246,41 @@ class TestScheduleTriggerDebugEventPoller:
|
||||
|
||||
|
||||
class TestCreateEventPoller:
|
||||
def _workflow_with_node(self, node_type: NodeType):
|
||||
wf = MagicMock()
|
||||
wf.get_node_config_by_id.return_value = {"data": {}}
|
||||
wf.get_node_type_from_node_config.return_value = node_type
|
||||
return wf
|
||||
|
||||
def test_creates_plugin_poller(self):
|
||||
wf = self._workflow_with_node(TRIGGER_PLUGIN_NODE_TYPE)
|
||||
wf = _workflow_with_node(TRIGGER_PLUGIN_NODE_TYPE)
|
||||
poller = create_event_poller(wf, "t1", "u1", "a1", "n1")
|
||||
assert isinstance(poller, PluginTriggerDebugEventPoller)
|
||||
|
||||
def test_creates_webhook_poller(self):
|
||||
wf = self._workflow_with_node(TRIGGER_WEBHOOK_NODE_TYPE)
|
||||
wf = _workflow_with_node(TRIGGER_WEBHOOK_NODE_TYPE)
|
||||
poller = create_event_poller(wf, "t1", "u1", "a1", "n1")
|
||||
assert isinstance(poller, WebhookTriggerDebugEventPoller)
|
||||
|
||||
def test_creates_schedule_poller(self):
|
||||
wf = self._workflow_with_node(TRIGGER_SCHEDULE_NODE_TYPE)
|
||||
wf = _workflow_with_node(TRIGGER_SCHEDULE_NODE_TYPE)
|
||||
poller = create_event_poller(wf, "t1", "u1", "a1", "n1")
|
||||
assert isinstance(poller, ScheduleTriggerDebugEventPoller)
|
||||
|
||||
def test_raises_for_unknown_type(self):
|
||||
wf = MagicMock()
|
||||
wf.get_node_config_by_id.return_value = {"data": {}}
|
||||
wf.get_node_type_from_node_config.return_value = BuiltinNodeTypes.START
|
||||
wf = _workflow_with_node(BuiltinNodeTypes.START)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
create_event_poller(wf, "t1", "u1", "a1", "n1")
|
||||
|
||||
def test_raises_when_node_config_missing(self):
|
||||
wf = MagicMock()
|
||||
wf.get_node_config_by_id.return_value = None
|
||||
wf = _workflow_with_node(None)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
with (
|
||||
patch.object(Workflow, "get_node_config_by_id", return_value=None),
|
||||
pytest.raises(ValueError),
|
||||
):
|
||||
create_event_poller(wf, "t1", "u1", "a1", "n1")
|
||||
|
||||
|
||||
class TestSelectTriggerDebugEvents:
|
||||
def test_returns_first_non_none_event(self):
|
||||
wf = MagicMock()
|
||||
wf.get_node_config_by_id.return_value = {"data": {}}
|
||||
wf.get_node_type_from_node_config.return_value = TRIGGER_WEBHOOK_NODE_TYPE
|
||||
app_model = MagicMock()
|
||||
app_model.tenant_id = "t1"
|
||||
app_model.id = "a1"
|
||||
wf = _workflow_with_node(TRIGGER_WEBHOOK_NODE_TYPE)
|
||||
app_model = _app()
|
||||
|
||||
with patch.object(WebhookTriggerDebugEventPoller, "poll") as mock_poll:
|
||||
expected = MagicMock()
|
||||
@@ -269,12 +291,8 @@ class TestSelectTriggerDebugEvents:
|
||||
assert result is expected
|
||||
|
||||
def test_returns_none_when_no_events(self):
|
||||
wf = MagicMock()
|
||||
wf.get_node_config_by_id.return_value = {"data": {}}
|
||||
wf.get_node_type_from_node_config.return_value = TRIGGER_WEBHOOK_NODE_TYPE
|
||||
app_model = MagicMock()
|
||||
app_model.tenant_id = "t1"
|
||||
app_model.id = "a1"
|
||||
wf = _workflow_with_node(TRIGGER_WEBHOOK_NODE_TYPE)
|
||||
app_model = _app()
|
||||
|
||||
with patch.object(WebhookTriggerDebugEventPoller, "poll", return_value=None):
|
||||
result = select_trigger_debug_events(wf, app_model, "u1", ["n1"])
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import logging
|
||||
from datetime import UTC, datetime
|
||||
from io import BytesIO
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
@@ -9,11 +10,216 @@ from sqlalchemy.engine import Engine
|
||||
from sqlalchemy.orm import Session
|
||||
from werkzeug.datastructures import FileStorage
|
||||
|
||||
from graphon.enums import WorkflowType
|
||||
from models.enums import AppTriggerStatus, AppTriggerType, EndUserType
|
||||
from models.model import App, AppMode, EndUser
|
||||
from models.tools import ToolFile
|
||||
from models.trigger import AppTrigger, WorkflowWebhookTrigger
|
||||
from models.workflow import Workflow
|
||||
from services.errors.app import QuotaExceededError
|
||||
from services.trigger import webhook_service as webhook_service_module
|
||||
from services.trigger.webhook_service import WebhookService
|
||||
|
||||
|
||||
def _webhook_trigger(
|
||||
*,
|
||||
webhook_id: str = "webhook-123",
|
||||
tenant_id: str = "tenant-123",
|
||||
app_id: str = "app-123",
|
||||
node_id: str = "node-123",
|
||||
created_by: str = "account-123",
|
||||
) -> WorkflowWebhookTrigger:
|
||||
return WorkflowWebhookTrigger(
|
||||
webhook_id=webhook_id,
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
node_id=node_id,
|
||||
created_by=created_by,
|
||||
)
|
||||
|
||||
|
||||
def _tool_file() -> ToolFile:
|
||||
tool_file = ToolFile(
|
||||
user_id="user-123",
|
||||
tenant_id="test_tenant",
|
||||
conversation_id=None,
|
||||
file_key="webhook/test.txt",
|
||||
mimetype="text/plain",
|
||||
name="test.txt",
|
||||
size=7,
|
||||
)
|
||||
tool_file.id = "test_file_id"
|
||||
return tool_file
|
||||
|
||||
|
||||
def _workflow(
|
||||
*,
|
||||
workflow_id: str = "workflow-123",
|
||||
tenant_id: str = "tenant-123",
|
||||
app_id: str = "app-123",
|
||||
version: str = Workflow.VERSION_DRAFT,
|
||||
) -> Workflow:
|
||||
workflow = Workflow.new(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
type=WorkflowType.WORKFLOW.value,
|
||||
version=version,
|
||||
graph='{"nodes": [], "edges": []}',
|
||||
features="{}",
|
||||
created_by="account-123",
|
||||
environment_variables=[],
|
||||
conversation_variables=[],
|
||||
rag_pipeline_variables=[],
|
||||
)
|
||||
workflow.id = workflow_id
|
||||
return workflow
|
||||
|
||||
|
||||
def _app(*, tenant_id: str = "tenant-123", app_id: str = "app-123") -> App:
|
||||
return App(
|
||||
id=app_id,
|
||||
tenant_id=tenant_id,
|
||||
name="Webhook App",
|
||||
description="",
|
||||
mode=AppMode.WORKFLOW,
|
||||
enable_site=True,
|
||||
enable_api=True,
|
||||
max_active_requests=0,
|
||||
)
|
||||
|
||||
|
||||
def _app_trigger(*, status: AppTriggerStatus = AppTriggerStatus.ENABLED) -> AppTrigger:
|
||||
return AppTrigger(
|
||||
tenant_id="tenant-123",
|
||||
app_id="app-123",
|
||||
node_id="node-123",
|
||||
trigger_type=AppTriggerType.TRIGGER_WEBHOOK,
|
||||
title="Webhook",
|
||||
status=status,
|
||||
)
|
||||
|
||||
|
||||
def _end_user() -> EndUser:
|
||||
return EndUser(
|
||||
id="end-user-123",
|
||||
tenant_id="tenant-123",
|
||||
app_id="app-123",
|
||||
type=EndUserType.TRIGGER,
|
||||
session_id="webhook-session",
|
||||
)
|
||||
|
||||
|
||||
class TestWebhookServiceLookup:
|
||||
def test_debug_lookup_scopes_draft_workflow_to_trigger_owner(
|
||||
self,
|
||||
sqlite_session: Session,
|
||||
sqlite_engine: Engine,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
target_workflow = _workflow()
|
||||
target_workflow.created_at = datetime(2025, 1, 1, tzinfo=UTC).replace(tzinfo=None)
|
||||
tenant_decoy = _workflow(
|
||||
workflow_id="workflow-decoy",
|
||||
tenant_id="tenant-other",
|
||||
app_id="app-123",
|
||||
)
|
||||
tenant_decoy.created_at = datetime(2026, 1, 1, tzinfo=UTC).replace(tzinfo=None)
|
||||
sqlite_session.add_all([_webhook_trigger(), target_workflow, tenant_decoy])
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine))
|
||||
node_config = {"id": "node-123", "data": {}}
|
||||
|
||||
with patch.object(Workflow, "get_node_config_by_id", autospec=True, return_value=node_config) as get_node:
|
||||
webhook_trigger, workflow, result_node_config = WebhookService.get_webhook_trigger_and_workflow(
|
||||
"webhook-123", is_debug=True
|
||||
)
|
||||
|
||||
assert webhook_trigger.tenant_id == "tenant-123"
|
||||
assert workflow.id == target_workflow.id
|
||||
assert result_node_config is node_config
|
||||
get_node.assert_called_once_with(workflow, "node-123")
|
||||
|
||||
def test_published_lookup_uses_persisted_owner_chain(
|
||||
self,
|
||||
sqlite_session: Session,
|
||||
sqlite_engine: Engine,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
sqlite_session.add_all([_webhook_trigger(), _app_trigger(), _app()])
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine))
|
||||
published_workflow = _workflow(version="published")
|
||||
node_config = {"id": "node-123", "data": {}}
|
||||
|
||||
with (
|
||||
patch.object(webhook_service_module, "WorkflowService") as workflow_service_class,
|
||||
patch.object(Workflow, "get_node_config_by_id", autospec=True, return_value=node_config),
|
||||
):
|
||||
get_published_workflow = workflow_service_class.return_value.get_published_workflow
|
||||
get_published_workflow.return_value = published_workflow
|
||||
webhook_trigger, workflow, result_node_config = WebhookService.get_webhook_trigger_and_workflow(
|
||||
"webhook-123"
|
||||
)
|
||||
|
||||
assert webhook_trigger.app_id == "app-123"
|
||||
assert workflow is published_workflow
|
||||
assert result_node_config is node_config
|
||||
persisted_app = get_published_workflow.call_args.args[0]
|
||||
assert isinstance(persisted_app, App)
|
||||
assert persisted_app.id == "app-123"
|
||||
assert persisted_app.tenant_id == "tenant-123"
|
||||
|
||||
def test_published_lookup_rejects_cross_tenant_app_decoy(
|
||||
self,
|
||||
sqlite_session: Session,
|
||||
sqlite_engine: Engine,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
sqlite_session.add_all(
|
||||
[
|
||||
_webhook_trigger(),
|
||||
_app_trigger(),
|
||||
_app(tenant_id="tenant-other"),
|
||||
]
|
||||
)
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine))
|
||||
|
||||
with pytest.raises(ValueError, match="App not found"):
|
||||
WebhookService.get_webhook_trigger_and_workflow("webhook-123")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"status",
|
||||
[AppTriggerStatus.DISABLED, AppTriggerStatus.UNAUTHORIZED],
|
||||
)
|
||||
def test_published_lookup_rejects_inactive_trigger(
|
||||
self,
|
||||
status: AppTriggerStatus,
|
||||
sqlite_session: Session,
|
||||
sqlite_engine: Engine,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
sqlite_session.add_all([_webhook_trigger(), _app_trigger(status=status)])
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine))
|
||||
|
||||
with pytest.raises(ValueError, match="disabled"):
|
||||
WebhookService.get_webhook_trigger_and_workflow("webhook-123")
|
||||
|
||||
def test_published_lookup_reports_rate_limited_trigger(
|
||||
self,
|
||||
sqlite_session: Session,
|
||||
sqlite_engine: Engine,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
sqlite_session.add_all([_webhook_trigger(), _app_trigger(status=AppTriggerStatus.RATE_LIMITED)])
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine))
|
||||
|
||||
with pytest.raises(QuotaExceededError):
|
||||
WebhookService.get_webhook_trigger_and_workflow("webhook-123")
|
||||
|
||||
|
||||
class TestWebhookServiceUnit:
|
||||
"""Webhook business-logic tests with isolated sessions where the service owns their lifecycle."""
|
||||
|
||||
@@ -25,13 +231,8 @@ class TestWebhookServiceUnit:
|
||||
) -> None:
|
||||
"""Quota failures refund the charge and close the real service-owned session."""
|
||||
monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine))
|
||||
webhook_trigger = MagicMock(
|
||||
webhook_id="webhook-123",
|
||||
tenant_id="tenant-123",
|
||||
app_id="app-123",
|
||||
node_id="node-123",
|
||||
)
|
||||
workflow = MagicMock(id="workflow-123")
|
||||
webhook_trigger = _webhook_trigger()
|
||||
workflow = _workflow()
|
||||
quota_charge = MagicMock()
|
||||
quota_error = QuotaExceededError(feature="workflow", tenant_id="tenant-123", required=1)
|
||||
|
||||
@@ -39,7 +240,7 @@ class TestWebhookServiceUnit:
|
||||
with (
|
||||
patch(
|
||||
"services.trigger.webhook_service.EndUserService.get_or_create_end_user_by_type",
|
||||
return_value=MagicMock(id="end-user-123"),
|
||||
return_value=_end_user(),
|
||||
),
|
||||
patch("services.trigger.webhook_service.QuotaService.reserve", return_value=quota_charge),
|
||||
patch(
|
||||
@@ -71,14 +272,9 @@ class TestWebhookServiceUnit:
|
||||
sqlite_engine: Engine,
|
||||
) -> None:
|
||||
monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine))
|
||||
webhook_trigger = MagicMock(
|
||||
webhook_id="webhook-123",
|
||||
tenant_id="tenant-123",
|
||||
app_id="app-123",
|
||||
node_id="node-123",
|
||||
)
|
||||
workflow = MagicMock(id="workflow-123")
|
||||
end_user = MagicMock(id="end-user-123")
|
||||
webhook_trigger = _webhook_trigger()
|
||||
workflow = _workflow()
|
||||
end_user = _end_user()
|
||||
quota_charge = MagicMock()
|
||||
webhook_data = {
|
||||
"method": "POST",
|
||||
@@ -108,13 +304,8 @@ class TestWebhookServiceUnit:
|
||||
quota_charge.refund.assert_not_called()
|
||||
|
||||
def test_trigger_workflow_execution_end_user_service_failure(self) -> None:
|
||||
webhook_trigger = MagicMock(
|
||||
webhook_id="webhook-123",
|
||||
tenant_id="tenant-123",
|
||||
app_id="app-123",
|
||||
node_id="node-123",
|
||||
)
|
||||
workflow = MagicMock(id="workflow-123")
|
||||
webhook_trigger = _webhook_trigger()
|
||||
workflow = _workflow()
|
||||
webhook_data = {"method": "POST", "headers": {}, "query_params": {}, "body": {}, "files": {}}
|
||||
|
||||
with patch.object(
|
||||
@@ -136,7 +327,7 @@ class TestWebhookServiceUnit:
|
||||
query_string="version=1&format=json",
|
||||
json={"message": "hello", "count": 42},
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
webhook_data = WebhookService.extract_webhook_data(webhook_trigger)
|
||||
|
||||
assert webhook_data["method"] == "POST"
|
||||
@@ -158,7 +349,7 @@ class TestWebhookServiceUnit:
|
||||
headers={"Content-Type": "application/json"},
|
||||
query_string="count=42&threshold=3.14&enabled=true¬e=text",
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
webhook_data = WebhookService.extract_webhook_data(webhook_trigger)
|
||||
|
||||
# After refactoring, raw extraction keeps query params as strings
|
||||
@@ -177,7 +368,7 @@ class TestWebhookServiceUnit:
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
data={"username": "test", "password": "secret"},
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
webhook_data = WebhookService.extract_webhook_data(webhook_trigger)
|
||||
|
||||
assert webhook_data["method"] == "POST"
|
||||
@@ -198,7 +389,7 @@ class TestWebhookServiceUnit:
|
||||
headers={"Content-Type": "multipart/form-data"},
|
||||
data={"message": "test", "file": file_storage},
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
webhook_trigger.tenant_id = "test_tenant"
|
||||
|
||||
with patch.object(WebhookService, "_process_file_uploads", autospec=True) as mock_process_files:
|
||||
@@ -218,7 +409,7 @@ class TestWebhookServiceUnit:
|
||||
with app.test_request_context(
|
||||
"/webhook", method="POST", headers={"Content-Type": "text/plain"}, data="raw text content"
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
webhook_data = WebhookService.extract_webhook_data(webhook_trigger)
|
||||
|
||||
assert webhook_data["method"] == "POST"
|
||||
@@ -232,7 +423,7 @@ class TestWebhookServiceUnit:
|
||||
with app.test_request_context(
|
||||
"/webhook", method="POST", headers={"Content-Type": "application/octet-stream"}, data=binary_content
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
mock_file = MagicMock()
|
||||
mock_file.to_dict.return_value = {"file": "data"}
|
||||
|
||||
@@ -299,7 +490,7 @@ class TestWebhookServiceUnit:
|
||||
with app.test_request_context(
|
||||
"/webhook", method="POST", headers={"Content-Type": "application/json"}, data="invalid json"
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
with pytest.raises(ValueError, match="Invalid JSON body"):
|
||||
WebhookService.extract_webhook_data(webhook_trigger)
|
||||
|
||||
@@ -369,9 +560,7 @@ class TestWebhookServiceUnit:
|
||||
"""Test successful file upload processing."""
|
||||
# Mock ToolFileManager
|
||||
mock_tool_file_instance = mock_tool_file_manager.return_value # Mock file creation
|
||||
mock_tool_file = MagicMock()
|
||||
mock_tool_file.id = "test_file_id"
|
||||
mock_tool_file_instance.create_file_by_raw.return_value = mock_tool_file
|
||||
mock_tool_file_instance.create_file_by_raw.return_value = _tool_file()
|
||||
|
||||
# Mock file factory
|
||||
mock_file_obj = MagicMock()
|
||||
@@ -387,7 +576,7 @@ class TestWebhookServiceUnit:
|
||||
files["file1"].stream.read.return_value = b"content1"
|
||||
files["file2"].stream.read.return_value = b"content2"
|
||||
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
webhook_trigger.tenant_id = "test_tenant"
|
||||
|
||||
result = WebhookService._process_file_uploads(files, webhook_trigger)
|
||||
@@ -406,9 +595,7 @@ class TestWebhookServiceUnit:
|
||||
"""Test file upload processing with errors."""
|
||||
# Mock ToolFileManager
|
||||
mock_tool_file_instance = mock_tool_file_manager.return_value # Mock file creation
|
||||
mock_tool_file = MagicMock()
|
||||
mock_tool_file.id = "test_file_id"
|
||||
mock_tool_file_instance.create_file_by_raw.return_value = mock_tool_file
|
||||
mock_tool_file_instance.create_file_by_raw.return_value = _tool_file()
|
||||
|
||||
# Mock file factory
|
||||
mock_file_obj = MagicMock()
|
||||
@@ -423,7 +610,7 @@ class TestWebhookServiceUnit:
|
||||
files["good_file"].stream.read.return_value = b"content"
|
||||
files["bad_file"].stream.read.side_effect = Exception("Read error")
|
||||
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
webhook_trigger.tenant_id = "test_tenant"
|
||||
|
||||
result = WebhookService._process_file_uploads(files, webhook_trigger)
|
||||
@@ -440,7 +627,7 @@ class TestWebhookServiceUnit:
|
||||
"none_filename": MagicMock(filename=None, content_type="text/plain"),
|
||||
}
|
||||
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
webhook_trigger.tenant_id = "test_tenant"
|
||||
|
||||
result = WebhookService._process_file_uploads(files, webhook_trigger)
|
||||
@@ -585,7 +772,7 @@ class TestWebhookServiceUnit:
|
||||
query_string="count=42&enabled=true",
|
||||
json={"message": "hello", "age": 25},
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
node_config = {
|
||||
"data": {
|
||||
"method": "post",
|
||||
@@ -619,7 +806,7 @@ class TestWebhookServiceUnit:
|
||||
headers={"Content-Type": "application/json"},
|
||||
data='{"invalid": }',
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
node_config = {
|
||||
"data": {
|
||||
"method": "post",
|
||||
@@ -639,7 +826,7 @@ class TestWebhookServiceUnit:
|
||||
method="GET", # Wrong method
|
||||
headers={"Content-Type": "application/json"},
|
||||
):
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
node_config = {
|
||||
"data": {
|
||||
"method": "post", # Expects POST
|
||||
@@ -666,7 +853,7 @@ class TestWebhookServiceUnit:
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match="Required header missing: Authorization"):
|
||||
WebhookService.extract_and_validate_webhook_data(MagicMock(), node_config)
|
||||
WebhookService.extract_and_validate_webhook_data(_webhook_trigger(), node_config)
|
||||
|
||||
def test_extract_and_validate_webhook_request_case_insensitive_headers(self) -> None:
|
||||
app = Flask(__name__)
|
||||
@@ -685,7 +872,7 @@ class TestWebhookServiceUnit:
|
||||
}
|
||||
}
|
||||
|
||||
result = WebhookService.extract_and_validate_webhook_data(MagicMock(), node_config)
|
||||
result = WebhookService.extract_and_validate_webhook_data(_webhook_trigger(), node_config)
|
||||
|
||||
assert result["headers"].get("Authorization") == "Bearer token"
|
||||
|
||||
@@ -707,7 +894,7 @@ class TestWebhookServiceUnit:
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match="Required parameter missing: version"):
|
||||
WebhookService.extract_and_validate_webhook_data(MagicMock(), node_config)
|
||||
WebhookService.extract_and_validate_webhook_data(_webhook_trigger(), node_config)
|
||||
|
||||
def test_extract_and_validate_webhook_request_missing_required_body_param(self) -> None:
|
||||
app = Flask(__name__)
|
||||
@@ -726,7 +913,7 @@ class TestWebhookServiceUnit:
|
||||
}
|
||||
|
||||
with pytest.raises(ValueError, match="Required body parameter missing: message"):
|
||||
WebhookService.extract_and_validate_webhook_data(MagicMock(), node_config)
|
||||
WebhookService.extract_and_validate_webhook_data(_webhook_trigger(), node_config)
|
||||
|
||||
def test_extract_and_validate_webhook_request_missing_required_file(self) -> None:
|
||||
app = Flask(__name__)
|
||||
@@ -744,7 +931,7 @@ class TestWebhookServiceUnit:
|
||||
}
|
||||
}
|
||||
|
||||
result = WebhookService.extract_and_validate_webhook_data(MagicMock(), node_config)
|
||||
result = WebhookService.extract_and_validate_webhook_data(_webhook_trigger(), node_config)
|
||||
|
||||
assert result["files"] == {}
|
||||
|
||||
@@ -757,19 +944,19 @@ class TestWebhookServiceUnit:
|
||||
patch.object(WebhookService, "get_webhook_trigger_and_workflow", autospec=True) as mock_get_trigger,
|
||||
patch.object(WebhookService, "extract_and_validate_webhook_data", autospec=True) as mock_extract,
|
||||
):
|
||||
mock_trigger = MagicMock()
|
||||
mock_workflow = MagicMock()
|
||||
webhook_trigger = _webhook_trigger()
|
||||
workflow = _workflow()
|
||||
mock_config = {"data": {"test": "config"}}
|
||||
mock_data = {"test": "data"}
|
||||
|
||||
mock_get_trigger.return_value = (mock_trigger, mock_workflow, mock_config)
|
||||
mock_get_trigger.return_value = (webhook_trigger, workflow, mock_config)
|
||||
mock_extract.return_value = mock_data
|
||||
|
||||
result = _prepare_webhook_execution("test_webhook", is_debug=False)
|
||||
assert result == (mock_trigger, mock_workflow, mock_config, mock_data, None)
|
||||
assert result == (webhook_trigger, workflow, mock_config, mock_data, None)
|
||||
|
||||
# Reset mock
|
||||
mock_get_trigger.reset_mock()
|
||||
|
||||
result = _prepare_webhook_execution("test_webhook", is_debug=True)
|
||||
assert result == (mock_trigger, mock_workflow, mock_config, mock_data, None)
|
||||
assert result == (webhook_trigger, workflow, mock_config, mock_data, None)
|
||||
|
||||
@@ -14,6 +14,7 @@ from core.workflow.nodes.trigger_webhook.entities import (
|
||||
WebhookParameter,
|
||||
)
|
||||
from graphon.variables.types import SegmentType
|
||||
from models.trigger import WorkflowWebhookTrigger
|
||||
from services.trigger import webhook_service as service_module
|
||||
from services.trigger.webhook_service import WebhookService
|
||||
|
||||
@@ -23,8 +24,16 @@ def flask_app() -> Flask:
|
||||
return Flask(__name__)
|
||||
|
||||
|
||||
def _workflow_trigger(**kwargs: Any) -> Any:
|
||||
return SimpleNamespace(**kwargs)
|
||||
def _workflow_trigger(**kwargs: Any) -> WorkflowWebhookTrigger:
|
||||
values = {
|
||||
"webhook_id": "webhook-123",
|
||||
"tenant_id": "tenant-1",
|
||||
"app_id": "app-1",
|
||||
"node_id": "node-1",
|
||||
"created_by": "user-1",
|
||||
}
|
||||
values.update(kwargs)
|
||||
return WorkflowWebhookTrigger(**values)
|
||||
|
||||
|
||||
class TestWebhookServiceExtractionFallbacks:
|
||||
@@ -34,7 +43,7 @@ class TestWebhookServiceExtractionFallbacks:
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _workflow_trigger()
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="services.trigger.webhook_service"):
|
||||
with flask_app.test_request_context(
|
||||
@@ -57,10 +66,10 @@ class TestWebhookServiceExtractionFallbacks:
|
||||
|
||||
with flask_app.test_request_context("/webhook", method="POST", data="ab"):
|
||||
with pytest.raises(RequestEntityTooLarge):
|
||||
WebhookService.extract_webhook_data(MagicMock())
|
||||
WebhookService.extract_webhook_data(_workflow_trigger())
|
||||
|
||||
def test_extract_octet_stream_body_should_return_none_when_empty_payload(self, flask_app: Flask) -> None:
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _workflow_trigger()
|
||||
|
||||
with flask_app.test_request_context("/webhook", method="POST", data=b""):
|
||||
body, files = WebhookService._extract_octet_stream_body(webhook_trigger)
|
||||
@@ -73,7 +82,7 @@ class TestWebhookServiceExtractionFallbacks:
|
||||
flask_app: Flask,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
webhook_trigger = MagicMock()
|
||||
webhook_trigger = _workflow_trigger()
|
||||
monkeypatch.setattr(
|
||||
WebhookService, "_detect_binary_mimetype", MagicMock(return_value="application/octet-stream")
|
||||
)
|
||||
|
||||
@@ -1,12 +1,56 @@
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
import tasks.trigger_processing_tasks as trigger_processing_tasks_module
|
||||
from core.plugin.entities.plugin_daemon import CredentialType
|
||||
from graphon.enums import WorkflowType
|
||||
from models.enums import EndUserType
|
||||
from models.model import EndUser
|
||||
from models.trigger import TriggerSubscription, WorkflowPluginTrigger
|
||||
from models.workflow import Workflow
|
||||
from services.errors.app import QuotaExceededError
|
||||
from tasks.trigger_processing_tasks import dispatch_triggered_workflow
|
||||
|
||||
|
||||
def _workflow(*, app_id: str = "app-123") -> Workflow:
|
||||
workflow = Workflow.new(
|
||||
tenant_id="tenant-123",
|
||||
app_id=app_id,
|
||||
type=WorkflowType.WORKFLOW.value,
|
||||
version="published",
|
||||
graph=json.dumps(
|
||||
{
|
||||
"nodes": [
|
||||
{
|
||||
"id": "node-123",
|
||||
"data": {"type": trigger_processing_tasks_module.TRIGGER_PLUGIN_NODE_TYPE},
|
||||
}
|
||||
],
|
||||
"edges": [],
|
||||
}
|
||||
),
|
||||
features="{}",
|
||||
created_by="user-123",
|
||||
environment_variables=[],
|
||||
conversation_variables=[],
|
||||
rag_pipeline_variables=[],
|
||||
)
|
||||
workflow.id = "workflow-123"
|
||||
return workflow
|
||||
|
||||
|
||||
def _end_user() -> EndUser:
|
||||
return EndUser(
|
||||
id="end-user-123",
|
||||
tenant_id="tenant-123",
|
||||
app_id="app-123",
|
||||
type=EndUserType.TRIGGER,
|
||||
session_id="trigger-session",
|
||||
)
|
||||
|
||||
|
||||
class TestDispatchTriggeredWorkflow:
|
||||
"""Unit tests covering branch behaviours of ``dispatch_triggered_workflow``.
|
||||
|
||||
@@ -20,21 +64,31 @@ class TestDispatchTriggeredWorkflow:
|
||||
|
||||
@pytest.fixture
|
||||
def subscription(self):
|
||||
sub = MagicMock()
|
||||
sub.id = "subscription-123"
|
||||
sub.tenant_id = "tenant-123"
|
||||
sub.provider_id = "langgenius/test_plugin/test_plugin"
|
||||
sub.endpoint_id = "endpoint-123"
|
||||
sub.credentials = {}
|
||||
sub.credential_type = "api_key"
|
||||
return sub
|
||||
subscription = TriggerSubscription(
|
||||
tenant_id="tenant-123",
|
||||
user_id="user-123",
|
||||
name="Test Subscription",
|
||||
endpoint_id="endpoint-123",
|
||||
provider_id="langgenius/test_plugin/test_plugin",
|
||||
parameters={},
|
||||
properties={},
|
||||
credentials={},
|
||||
credential_type=CredentialType.API_KEY,
|
||||
)
|
||||
subscription.id = "subscription-123"
|
||||
return subscription
|
||||
|
||||
@pytest.fixture
|
||||
def plugin_trigger(self):
|
||||
trigger = MagicMock()
|
||||
trigger = WorkflowPluginTrigger(
|
||||
app_id="app-123",
|
||||
node_id="node-123",
|
||||
tenant_id="tenant-123",
|
||||
provider_id="langgenius/test_plugin/test_plugin",
|
||||
event_name="test_event",
|
||||
subscription_id="subscription-123",
|
||||
)
|
||||
trigger.id = "plugin-trigger-123"
|
||||
trigger.app_id = "app-123"
|
||||
trigger.node_id = "node-123"
|
||||
return trigger
|
||||
|
||||
@pytest.fixture
|
||||
@@ -144,11 +198,8 @@ class TestDispatchTriggeredWorkflow:
|
||||
|
||||
def test_dispatch_marks_rate_limited_when_quota_exceeded(self, subscription, plugin_trigger, dispatch_mocks):
|
||||
"""Covers QuotaExceededError → mark rate-limited + early return."""
|
||||
workflow_mock = MagicMock()
|
||||
workflow_mock.walk_nodes.return_value = iter(
|
||||
[(plugin_trigger.node_id, {"type": trigger_processing_tasks_module.TRIGGER_PLUGIN_NODE_TYPE})]
|
||||
)
|
||||
dispatch_mocks["get_workflows"].return_value = {plugin_trigger.app_id: workflow_mock}
|
||||
workflow = _workflow()
|
||||
dispatch_mocks["get_workflows"].return_value = {plugin_trigger.app_id: workflow}
|
||||
dispatch_mocks["reserve"].side_effect = QuotaExceededError(
|
||||
feature="trigger", tenant_id=subscription.tenant_id, required=1
|
||||
)
|
||||
@@ -169,15 +220,11 @@ class TestDispatchTriggeredWorkflow:
|
||||
self, subscription, plugin_trigger, dispatch_mocks
|
||||
):
|
||||
"""Happy path: end user exists and async trigger succeeds."""
|
||||
workflow_mock = MagicMock()
|
||||
workflow_mock.id = "workflow-123"
|
||||
workflow_mock.walk_nodes.return_value = iter(
|
||||
[(plugin_trigger.node_id, {"type": trigger_processing_tasks_module.TRIGGER_PLUGIN_NODE_TYPE})]
|
||||
)
|
||||
dispatch_mocks["get_workflows"].return_value = {plugin_trigger.app_id: workflow_mock}
|
||||
workflow = _workflow()
|
||||
dispatch_mocks["get_workflows"].return_value = {plugin_trigger.app_id: workflow}
|
||||
|
||||
end_user_mock = MagicMock()
|
||||
dispatch_mocks["create_end_user_batch"].return_value = {plugin_trigger.app_id: end_user_mock}
|
||||
end_user = _end_user()
|
||||
dispatch_mocks["create_end_user_batch"].return_value = {plugin_trigger.app_id: end_user}
|
||||
|
||||
dispatched = dispatch_triggered_workflow(
|
||||
user_id="user-123",
|
||||
@@ -189,7 +236,7 @@ class TestDispatchTriggeredWorkflow:
|
||||
assert dispatched == 1
|
||||
dispatch_mocks["trigger_workflow_async"].assert_called_once()
|
||||
_, kwargs = dispatch_mocks["trigger_workflow_async"].call_args
|
||||
assert kwargs["user"] is end_user_mock
|
||||
assert kwargs["user"] is end_user
|
||||
dispatch_mocks["quota_charge"].commit.assert_called_once()
|
||||
dispatch_mocks["quota_charge"].refund.assert_not_called()
|
||||
dispatch_mocks["mark_rate_limited"].assert_not_called()
|
||||
|
||||
Reference in New Issue
Block a user