youngkingdom/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-05-05 18:08:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into feat/support-agent-sandbox

Browse Source 
# Conflicts:
#	api/core/app/apps/workflow/app_runner.py
This commit is contained in:
yyh
2026-01-28 11:41:58 +08:00
parent 0d9de79fae 6fff46bc29
commit a38b8987b4
78 changed files with 4395 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
api/.importlinter
View File

@ -104,9 +104,7 @@ forbidden_modules =
ignore_imports =
core.workflow.nodes.loop.loop_node -> core.app.workflow.node_factory
core.workflow.graph_engine.command_channels.redis_channel -> extensions.ext_redis
core.workflow.graph_engine.layers.observability -> configs
core.workflow.graph_engine.layers.observability -> extensions.otel.runtime
core.workflow.graph_engine.layers.persistence -> core.ops.ops_trace_manager
core.workflow.workflow_entry -> core.app.workflow.layers.observability
core.workflow.graph_engine.worker_management.worker_pool -> configs
core.workflow.nodes.agent.agent_node -> core.model_manager
core.workflow.nodes.agent.agent_node -> core.provider_manager
@ -147,7 +145,6 @@ ignore_imports =
core.workflow.workflow_entry -> models.workflow
core.workflow.nodes.agent.agent_node -> core.agent.entities
core.workflow.nodes.agent.agent_node -> core.agent.plugin_entities
core.workflow.graph_engine.layers.persistence -> core.app.entities.app_invoke_entities
core.workflow.nodes.base.node -> core.app.entities.app_invoke_entities
core.workflow.nodes.knowledge_index.knowledge_index_node -> core.app.entities.app_invoke_entities
core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.app_config.entities
@ -217,7 +214,6 @@ ignore_imports =
core.workflow.nodes.llm.node -> core.llm_generator.output_parser.errors
core.workflow.nodes.llm.node -> core.llm_generator.output_parser.structured_output
core.workflow.nodes.llm.node -> core.model_manager
core.workflow.graph_engine.layers.persistence -> core.ops.entities.trace_entity
core.workflow.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.prompt.simple_prompt_transform
core.workflow.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities

2
api/core/app/apps/advanced_chat/app_runner.py
View File

@ -21,6 +21,7 @@ from core.app.entities.queue_entities import (
)
from core.app.features.annotation_reply.annotation_reply import AnnotationReplyFeature
from core.app.layers.conversation_variable_persist_layer import ConversationVariablePersistenceLayer
from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
from core.db.session_factory import session_factory
from core.moderation.base import ModerationError
from core.moderation.input_moderation import InputModeration
@ -29,7 +30,6 @@ from core.variables.variables import Variable
from core.workflow.enums import WorkflowType
from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
from core.workflow.graph_engine.layers.base import GraphEngineLayer
from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
from core.workflow.runtime import GraphRuntimeState, VariablePool

2
api/core/app/apps/pipeline/pipeline_runner.py
View File

@ -9,12 +9,12 @@ from core.app.entities.app_invoke_entities import (
InvokeFrom,
RagPipelineGenerateEntity,
)
from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
from core.app.workflow.node_factory import DifyNodeFactory
from core.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
from core.workflow.entities.graph_init_params import GraphInitParams
from core.workflow.enums import WorkflowType
from core.workflow.graph import Graph
from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
from core.workflow.graph_events import GraphEngineEvent, GraphRunFailedEvent
from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository

2
api/core/app/apps/workflow/app_runner.py
View File

@ -8,10 +8,10 @@ from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
from core.sandbox import Sandbox
from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
from core.workflow.enums import WorkflowType
from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
from core.workflow.graph_engine.layers.base import GraphEngineLayer
from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
from core.workflow.runtime import GraphRuntimeState, VariablePool

6
api/core/app/apps/workflow_app_runner.py
View File

@ -157,7 +157,7 @@ class WorkflowBasedAppRunner:
# Create initial runtime state with variable pool containing environment variables
graph_runtime_state = GraphRuntimeState(
variable_pool=VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
environment_variables=workflow.environment_variables,
),
@ -272,7 +272,9 @@ class WorkflowBasedAppRunner:
)
# init graph
graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=node_id)
graph = Graph.init(
graph_config=graph_config, node_factory=node_factory, root_node_id=node_id, skip_validation=True
)
if not graph:
raise ValueError("graph not found in workflow")

10
api/core/app/workflow/layers/__init__.py Normal file
View File

@ -0,0 +1,10 @@
"""Workflow-level GraphEngine layers that depend on outer infrastructure."""
from .observability import ObservabilityLayer
from .persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
__all__ = [
"ObservabilityLayer",
"PersistenceWorkflowInfo",
"WorkflowPersistenceLayer",
]

0
api/core/workflow/graph_engine/layers/observability.py → api/core/app/workflow/layers/observability.py
View File

4
api/core/workflow/graph_engine/layers/persistence.py → api/core/app/workflow/layers/persistence.py
View File

@ -45,7 +45,6 @@ from core.workflow.graph_events import (
from core.workflow.node_events import NodeRunResult
from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
from core.workflow.workflow_entry import WorkflowEntry
from libs.datetime_utils import naive_utc_now
@ -319,6 +318,9 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
# workflow inputs stay reusable without binding future runs to this conversation.
continue
inputs[f"sys.{field_name}"] = value
# Local import to avoid circular dependency during app bootstrapping.
from core.workflow.workflow_entry import WorkflowEntry
handled = WorkflowEntry.handle_special_values(inputs)
return handled or {}

6
api/core/workflow/graph/graph.py
View File

@ -288,6 +288,7 @@ class Graph:
graph_config: Mapping[str, object],
node_factory: NodeFactory,
root_node_id: str | None = None,
skip_validation: bool = False,
) -> Graph:
"""
Initialize graph
@ -346,8 +347,9 @@ class Graph:
root_node=root_node,
)
# Validate the graph structure using built-in validators
get_graph_validator().validate(graph)
if not skip_validation:
# Validate the graph structure using built-in validators
get_graph_validator().validate(graph)
return graph

2
api/core/workflow/graph_engine/layers/__init__.py
View File

@ -8,11 +8,9 @@ with middleware-like components that can observe events and interact with execut
from .base import GraphEngineLayer
from .debug_logging import DebugLoggingLayer
from .execution_limits import ExecutionLimitsLayer
from .observability import ObservabilityLayer
__all__ = [
"DebugLoggingLayer",
"ExecutionLimitsLayer",
"GraphEngineLayer",
"ObservabilityLayer",
]

4
api/core/workflow/runtime/variable_pool.py
View File

@ -44,7 +44,7 @@ class VariablePool(BaseModel):
)
system_variables: SystemVariable = Field(
description="System variables",
default_factory=SystemVariable.empty,
default_factory=SystemVariable.default,
)
environment_variables: Sequence[Variable] = Field(
description="Environment variables.",
@ -309,4 +309,4 @@ class VariablePool(BaseModel):
@classmethod
def empty(cls) -> VariablePool:
"""Create an empty variable pool."""
return cls(system_variables=SystemVariable.empty())
return cls(system_variables=SystemVariable.default())

5
api/core/workflow/system_variable.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import Any
from uuid import uuid4
from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
@ -72,8 +73,8 @@ class SystemVariable(BaseModel):
return data
@classmethod
def empty(cls) -> SystemVariable:
return cls()
def default(cls) -> SystemVariable:
return cls(workflow_execution_id=str(uuid4()))
def to_dict(self) -> dict[SystemVariableKey, Any]:
# NOTE: This method is provided for compatibility with legacy code.

5
api/core/workflow/workflow_entry.py
View File

@ -7,6 +7,7 @@ from typing import Any
from configs import dify_config
from core.app.apps.exc import GenerateTaskStoppedError
from core.app.entities.app_invoke_entities import InvokeFrom
from core.app.workflow.layers.observability import ObservabilityLayer
from core.app.workflow.node_factory import DifyNodeFactory
from core.file.models import File
from core.sandbox import Sandbox
@ -16,7 +17,7 @@ from core.workflow.errors import WorkflowNodeRunFailedError
from core.workflow.graph import Graph
from core.workflow.graph_engine import GraphEngine
from core.workflow.graph_engine.command_channels import InMemoryChannel
from core.workflow.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLayer, ObservabilityLayer
from core.workflow.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLayer
from core.workflow.graph_engine.protocols.command_channel import CommandChannel
from core.workflow.graph_events import GraphEngineEvent, GraphNodeEventBase, GraphRunFailedEvent
from core.workflow.nodes import NodeType
@ -281,7 +282,7 @@ class WorkflowEntry:
# init variable pool
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
environment_variables=[],
)

2
api/extensions/ext_fastopenapi.py
View File

@ -36,7 +36,7 @@ def init_app(app: DifyApp) -> None:
router.include_router(console_router, prefix="/console/api")
CORS(
app,
resources={r"/console/api/*": {"origins": dify_config.CONSOLE_CORS_ALLOW_ORIGINS}},
resources={r"/console/api/.*": {"origins": dify_config.CONSOLE_CORS_ALLOW_ORIGINS}},
supports_credentials=True,
allow_headers=list(AUTHENTICATED_HEADERS),
methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],

2
api/services/rag_pipeline/rag_pipeline.py
View File

@ -436,7 +436,7 @@ class RagPipelineService:
user_inputs=user_inputs,
user_id=account.id,
variable_pool=VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs=user_inputs,
environment_variables=[],
conversation_variables=[],

4
api/services/workflow_service.py
View File

@ -752,7 +752,7 @@ class WorkflowService:
else:
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs=user_inputs,
environment_variables=draft_workflow.environment_variables,
conversation_variables=[],
@ -1160,7 +1160,7 @@ def _setup_variable_pool(
system_variable.conversation_id = conversation_id
system_variable.dialogue_count = 1
else:
system_variable = SystemVariable.empty()
system_variable = SystemVariable.default()
# init variable pool
variable_pool = VariablePool(

107
api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py Normal file
View File

@ -0,0 +1,107 @@
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from core.app.apps.base_app_queue_manager import AppQueueManager
from core.app.apps.workflow.app_runner import WorkflowAppRunner
from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
from core.workflow.runtime import GraphRuntimeState, VariablePool
from core.workflow.system_variable import SystemVariable
from models.workflow import Workflow
def _make_graph_state():
variable_pool = VariablePool(
system_variables=SystemVariable.default(),
user_inputs={},
environment_variables=[],
conversation_variables=[],
)
return MagicMock(), variable_pool, GraphRuntimeState(variable_pool=variable_pool, start_at=0.0)
@pytest.mark.parametrize(
("single_iteration_run", "single_loop_run"),
[
(WorkflowAppGenerateEntity.SingleIterationRunEntity(node_id="iter", inputs={}), None),
(None, WorkflowAppGenerateEntity.SingleLoopRunEntity(node_id="loop", inputs={})),
],
)
def test_run_uses_single_node_execution_branch(
single_iteration_run: Any,
single_loop_run: Any,
) -> None:
app_config = MagicMock()
app_config.app_id = "app"
app_config.tenant_id = "tenant"
app_config.workflow_id = "workflow"
app_generate_entity = MagicMock(spec=WorkflowAppGenerateEntity)
app_generate_entity.app_config = app_config
app_generate_entity.inputs = {}
app_generate_entity.files = []
app_generate_entity.user_id = "user"
app_generate_entity.invoke_from = InvokeFrom.SERVICE_API
app_generate_entity.workflow_execution_id = "execution-id"
app_generate_entity.task_id = "task-id"
app_generate_entity.call_depth = 0
app_generate_entity.trace_manager = None
app_generate_entity.single_iteration_run = single_iteration_run
app_generate_entity.single_loop_run = single_loop_run
workflow = MagicMock(spec=Workflow)
workflow.tenant_id = "tenant"
workflow.app_id = "app"
workflow.id = "workflow"
workflow.type = "workflow"
workflow.version = "v1"
workflow.graph_dict = {"nodes": [], "edges": []}
workflow.environment_variables = []
runner = WorkflowAppRunner(
application_generate_entity=app_generate_entity,
queue_manager=MagicMock(spec=AppQueueManager),
variable_loader=MagicMock(),
workflow=workflow,
system_user_id="system-user",
workflow_execution_repository=MagicMock(),
workflow_node_execution_repository=MagicMock(),
)
graph, variable_pool, graph_runtime_state = _make_graph_state()
mock_workflow_entry = MagicMock()
mock_workflow_entry.graph_engine = MagicMock()
mock_workflow_entry.graph_engine.layer = MagicMock()
mock_workflow_entry.run.return_value = iter([])
with (
patch("core.app.apps.workflow.app_runner.RedisChannel"),
patch("core.app.apps.workflow.app_runner.redis_client"),
patch("core.app.apps.workflow.app_runner.WorkflowEntry", return_value=mock_workflow_entry) as entry_class,
patch.object(
runner,
"_prepare_single_node_execution",
return_value=(
graph,
variable_pool,
graph_runtime_state,
),
) as prepare_single,
patch.object(runner, "_init_graph") as init_graph,
):
runner.run()
prepare_single.assert_called_once_with(
workflow=workflow,
single_iteration_run=single_iteration_run,
single_loop_run=single_loop_run,
)
init_graph.assert_not_called()
entry_kwargs = entry_class.call_args.kwargs
assert entry_kwargs["invoke_from"] == InvokeFrom.DEBUGGER
assert entry_kwargs["variable_pool"] is variable_pool
assert entry_kwargs["graph_runtime_state"] is graph_runtime_state

120
api/tests/unit_tests/core/workflow/graph/test_graph_skip_validation.py Normal file
View File

@ -0,0 +1,120 @@
from __future__ import annotations
from typing import Any
import pytest
from core.app.entities.app_invoke_entities import InvokeFrom
from core.app.workflow.node_factory import DifyNodeFactory
from core.workflow.entities import GraphInitParams
from core.workflow.graph import Graph
from core.workflow.graph.validation import GraphValidationError
from core.workflow.nodes import NodeType
from core.workflow.runtime import GraphRuntimeState, VariablePool
from core.workflow.system_variable import SystemVariable
from models.enums import UserFrom
def _build_iteration_graph(node_id: str) -> dict[str, Any]:
return {
"nodes": [
{
"id": node_id,
"data": {
"type": "iteration",
"title": "Iteration",
"iterator_selector": ["start", "items"],
"output_selector": [node_id, "output"],
},
}
],
"edges": [],
}
def _build_loop_graph(node_id: str) -> dict[str, Any]:
return {
"nodes": [
{
"id": node_id,
"data": {
"type": "loop",
"title": "Loop",
"loop_count": 1,
"break_conditions": [],
"logical_operator": "and",
"loop_variables": [],
"outputs": {},
},
}
],
"edges": [],
}
def _make_factory(graph_config: dict[str, Any]) -> DifyNodeFactory:
graph_init_params = GraphInitParams(
tenant_id="tenant",
app_id="app",
workflow_id="workflow",
graph_config=graph_config,
user_id="user",
user_from=UserFrom.ACCOUNT,
invoke_from=InvokeFrom.DEBUGGER,
call_depth=0,
)
graph_runtime_state = GraphRuntimeState(
variable_pool=VariablePool(
system_variables=SystemVariable.default(),
user_inputs={},
environment_variables=[],
),
start_at=0.0,
)
return DifyNodeFactory(graph_init_params=graph_init_params, graph_runtime_state=graph_runtime_state)
def test_iteration_root_requires_skip_validation():
node_id = "iteration-node"
graph_config = _build_iteration_graph(node_id)
node_factory = _make_factory(graph_config)
with pytest.raises(GraphValidationError):
Graph.init(
graph_config=graph_config,
node_factory=node_factory,
root_node_id=node_id,
)
graph = Graph.init(
graph_config=graph_config,
node_factory=node_factory,
root_node_id=node_id,
skip_validation=True,
)
assert graph.root_node.id == node_id
assert graph.root_node.node_type == NodeType.ITERATION
def test_loop_root_requires_skip_validation():
node_id = "loop-node"
graph_config = _build_loop_graph(node_id)
node_factory = _make_factory(graph_config)
with pytest.raises(GraphValidationError):
Graph.init(
graph_config=graph_config,
node_factory=node_factory,
root_node_id=node_id,
)
graph = Graph.init(
graph_config=graph_config,
node_factory=node_factory,
root_node_id=node_id,
skip_validation=True,
)
assert graph.root_node.id == node_id
assert graph.root_node.node_type == NodeType.LOOP

4
api/tests/unit_tests/core/workflow/graph_engine/layers/conftest.py
View File

@ -90,14 +90,14 @@ def mock_tool_node():
@pytest.fixture
def mock_is_instrument_flag_enabled_false():
"""Mock is_instrument_flag_enabled to return False."""
with patch("core.workflow.graph_engine.layers.observability.is_instrument_flag_enabled", return_value=False):
with patch("core.app.workflow.layers.observability.is_instrument_flag_enabled", return_value=False):
yield
@pytest.fixture
def mock_is_instrument_flag_enabled_true():
"""Mock is_instrument_flag_enabled to return True."""
with patch("core.workflow.graph_engine.layers.observability.is_instrument_flag_enabled", return_value=True):
with patch("core.app.workflow.layers.observability.is_instrument_flag_enabled", return_value=True):
yield

32
api/tests/unit_tests/core/workflow/graph_engine/layers/test_observability.py
View File

@ -15,14 +15,14 @@ from unittest.mock import patch
import pytest
from opentelemetry.trace import StatusCode
from core.app.workflow.layers.observability import ObservabilityLayer
from core.workflow.enums import NodeType
from core.workflow.graph_engine.layers.observability import ObservabilityLayer
class TestObservabilityLayerInitialization:
"""Test ObservabilityLayer initialization logic."""
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_initialization_when_otel_enabled(self, tracer_provider_with_memory_exporter):
"""Test that layer initializes correctly when OTel is enabled."""
@ -32,7 +32,7 @@ class TestObservabilityLayerInitialization:
assert NodeType.TOOL in layer._parsers
assert layer._default_parser is not None
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", False)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", False)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_true")
def test_initialization_when_instrument_flag_enabled(self, tracer_provider_with_memory_exporter):
"""Test that layer enables when instrument flag is enabled."""
@ -46,7 +46,7 @@ class TestObservabilityLayerInitialization:
class TestObservabilityLayerNodeSpanLifecycle:
"""Test node span creation and lifecycle management."""
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_node_span_created_and_ended(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_llm_node
@ -63,7 +63,7 @@ class TestObservabilityLayerNodeSpanLifecycle:
assert spans[0].name == mock_llm_node.title
assert spans[0].status.status_code == StatusCode.OK
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_node_error_recorded_in_span(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_llm_node
@ -82,7 +82,7 @@ class TestObservabilityLayerNodeSpanLifecycle:
assert len(spans[0].events) > 0
assert any("exception" in event.name.lower() for event in spans[0].events)
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_node_end_without_start_handled_gracefully(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_llm_node
@ -100,7 +100,7 @@ class TestObservabilityLayerNodeSpanLifecycle:
class TestObservabilityLayerParserIntegration:
"""Test parser integration for different node types."""
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_default_parser_used_for_regular_node(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_start_node
@ -119,7 +119,7 @@ class TestObservabilityLayerParserIntegration:
assert attrs["node.execution_id"] == mock_start_node.execution_id
assert attrs["node.type"] == mock_start_node.node_type.value
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_tool_parser_used_for_tool_node(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_tool_node
@ -138,7 +138,7 @@ class TestObservabilityLayerParserIntegration:
assert attrs["gen_ai.tool.name"] == mock_tool_node.title
assert attrs["gen_ai.tool.type"] == mock_tool_node._node_data.provider_type.value
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_llm_parser_used_for_llm_node(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_llm_node, mock_result_event
@ -176,7 +176,7 @@ class TestObservabilityLayerParserIntegration:
assert attrs["gen_ai.completion"] == "test completion"
assert attrs["gen_ai.response.finish_reason"] == "stop"
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_retrieval_parser_used_for_retrieval_node(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_retrieval_node, mock_result_event
@ -204,7 +204,7 @@ class TestObservabilityLayerParserIntegration:
assert attrs["retrieval.query"] == "test query"
assert "retrieval.document" in attrs
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_result_event_extracts_inputs_and_outputs(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_start_node, mock_result_event
@ -235,7 +235,7 @@ class TestObservabilityLayerParserIntegration:
class TestObservabilityLayerGraphLifecycle:
"""Test graph lifecycle management."""
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_on_graph_start_clears_contexts(self, tracer_provider_with_memory_exporter, mock_llm_node):
"""Test that on_graph_start clears node contexts."""
@ -248,7 +248,7 @@ class TestObservabilityLayerGraphLifecycle:
layer.on_graph_start()
assert len(layer._node_contexts) == 0
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_on_graph_end_with_no_unfinished_spans(
self, tracer_provider_with_memory_exporter, memory_span_exporter, mock_llm_node
@ -264,7 +264,7 @@ class TestObservabilityLayerGraphLifecycle:
spans = memory_span_exporter.get_finished_spans()
assert len(spans) == 1
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", True)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", True)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_on_graph_end_with_unfinished_spans_logs_warning(
self, tracer_provider_with_memory_exporter, mock_llm_node, caplog
@ -285,7 +285,7 @@ class TestObservabilityLayerGraphLifecycle:
class TestObservabilityLayerDisabledMode:
"""Test behavior when layer is disabled."""
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", False)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", False)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_disabled_mode_skips_node_start(self, memory_span_exporter, mock_start_node):
"""Test that disabled layer doesn't create spans on node start."""
@ -299,7 +299,7 @@ class TestObservabilityLayerDisabledMode:
spans = memory_span_exporter.get_finished_spans()
assert len(spans) == 0
@patch("core.workflow.graph_engine.layers.observability.dify_config.ENABLE_OTEL", False)
@patch("core.app.workflow.layers.observability.dify_config.ENABLE_OTEL", False)
@pytest.mark.usefixtures("mock_is_instrument_flag_enabled_false")
def test_disabled_mode_skips_node_end(self, memory_span_exporter, mock_llm_node):
"""Test that disabled layer doesn't process node end."""

30
api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_executor.py
View File

@ -16,7 +16,7 @@ from core.workflow.system_variable import SystemVariable
def test_executor_with_json_body_and_number_variable():
# Prepare the variable pool
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
variable_pool.add(["pre_node_id", "number"], 42)
@ -69,7 +69,7 @@ def test_executor_with_json_body_and_number_variable():
def test_executor_with_json_body_and_object_variable():
# Prepare the variable pool
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
variable_pool.add(["pre_node_id", "object"], {"name": "John Doe", "age": 30, "email": "john@example.com"})
@ -124,7 +124,7 @@ def test_executor_with_json_body_and_object_variable():
def test_executor_with_json_body_and_nested_object_variable():
# Prepare the variable pool
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
variable_pool.add(["pre_node_id", "object"], {"name": "John Doe", "age": 30, "email": "john@example.com"})
@ -178,7 +178,7 @@ def test_executor_with_json_body_and_nested_object_variable():
def test_extract_selectors_from_template_with_newline():
variable_pool = VariablePool(system_variables=SystemVariable.empty())
variable_pool = VariablePool(system_variables=SystemVariable.default())
variable_pool.add(("node_id", "custom_query"), "line1\nline2")
node_data = HttpRequestNodeData(
title="Test JSON Body with Nested Object Variable",
@ -205,7 +205,7 @@ def test_extract_selectors_from_template_with_newline():
def test_executor_with_form_data():
# Prepare the variable pool
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
variable_pool.add(["pre_node_id", "text_field"], "Hello, World!")
@ -290,7 +290,7 @@ def test_init_headers():
return Executor(
node_data=node_data,
timeout=timeout,
variable_pool=VariablePool(system_variables=SystemVariable.empty()),
variable_pool=VariablePool(system_variables=SystemVariable.default()),
)
executor = create_executor("aa\n cc:")
@ -324,7 +324,7 @@ def test_init_params():
return Executor(
node_data=node_data,
timeout=timeout,
variable_pool=VariablePool(system_variables=SystemVariable.empty()),
variable_pool=VariablePool(system_variables=SystemVariable.default()),
)
# Test basic key-value pairs
@ -355,7 +355,7 @@ def test_init_params():
def test_empty_api_key_raises_error_bearer():
"""Test that empty API key raises AuthorizationConfigError for bearer auth."""
variable_pool = VariablePool(system_variables=SystemVariable.empty())
variable_pool = VariablePool(system_variables=SystemVariable.default())
node_data = HttpRequestNodeData(
title="test",
method="get",
@ -379,7 +379,7 @@ def test_empty_api_key_raises_error_bearer():
def test_empty_api_key_raises_error_basic():
"""Test that empty API key raises AuthorizationConfigError for basic auth."""
variable_pool = VariablePool(system_variables=SystemVariable.empty())
variable_pool = VariablePool(system_variables=SystemVariable.default())
node_data = HttpRequestNodeData(
title="test",
method="get",
@ -403,7 +403,7 @@ def test_empty_api_key_raises_error_basic():
def test_empty_api_key_raises_error_custom():
"""Test that empty API key raises AuthorizationConfigError for custom auth."""
variable_pool = VariablePool(system_variables=SystemVariable.empty())
variable_pool = VariablePool(system_variables=SystemVariable.default())
node_data = HttpRequestNodeData(
title="test",
method="get",
@ -427,7 +427,7 @@ def test_empty_api_key_raises_error_custom():
def test_whitespace_only_api_key_raises_error():
"""Test that whitespace-only API key raises AuthorizationConfigError."""
variable_pool = VariablePool(system_variables=SystemVariable.empty())
variable_pool = VariablePool(system_variables=SystemVariable.default())
node_data = HttpRequestNodeData(
title="test",
method="get",
@ -451,7 +451,7 @@ def test_whitespace_only_api_key_raises_error():
def test_valid_api_key_works():
"""Test that valid API key works correctly for bearer auth."""
variable_pool = VariablePool(system_variables=SystemVariable.empty())
variable_pool = VariablePool(system_variables=SystemVariable.default())
node_data = HttpRequestNodeData(
title="test",
method="get",
@ -487,7 +487,7 @@ def test_executor_with_json_body_and_unquoted_uuid_variable():
test_uuid = "57eeeeb1-450b-482c-81b9-4be77e95dee2"
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
variable_pool.add(["pre_node_id", "uuid"], test_uuid)
@ -531,7 +531,7 @@ def test_executor_with_json_body_and_unquoted_uuid_with_newlines():
test_uuid = "57eeeeb1-450b-482c-81b9-4be77e95dee2"
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
variable_pool.add(["pre_node_id", "uuid"], test_uuid)
@ -569,7 +569,7 @@ def test_executor_with_json_body_and_unquoted_uuid_with_newlines():
def test_executor_with_json_body_preserves_numbers_and_strings():
"""Test that numbers are preserved and string values are properly quoted."""
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
variable_pool.add(["node", "count"], 42)

2
api/tests/unit_tests/core/workflow/nodes/llm/test_node.py
View File

@ -86,7 +86,7 @@ def graph_init_params() -> GraphInitParams:
@pytest.fixture
def graph_runtime_state() -> GraphRuntimeState:
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
return GraphRuntimeState(

14
api/tests/unit_tests/core/workflow/nodes/webhook/test_webhook_file_conversion.py
View File

@ -111,7 +111,7 @@ def test_webhook_node_file_conversion_to_file_variable():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -184,7 +184,7 @@ def test_webhook_node_file_conversion_with_missing_files():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -219,7 +219,7 @@ def test_webhook_node_file_conversion_with_none_file():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -256,7 +256,7 @@ def test_webhook_node_file_conversion_with_non_dict_file():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -300,7 +300,7 @@ def test_webhook_node_file_conversion_mixed_parameters():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -370,7 +370,7 @@ def test_webhook_node_different_file_types():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -430,7 +430,7 @@ def test_webhook_node_file_conversion_with_non_dict_wrapper():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},

20
api/tests/unit_tests/core/workflow/nodes/webhook/test_webhook_node.py
View File

@ -75,7 +75,7 @@ def test_webhook_node_basic_initialization():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
@ -118,7 +118,7 @@ def test_webhook_node_run_with_headers():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {
@ -154,7 +154,7 @@ def test_webhook_node_run_with_query_params():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -190,7 +190,7 @@ def test_webhook_node_run_with_body_params():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -249,7 +249,7 @@ def test_webhook_node_run_with_file_params():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},
@ -302,7 +302,7 @@ def test_webhook_node_run_mixed_parameters():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {"Authorization": "Bearer token"},
@ -342,7 +342,7 @@ def test_webhook_node_run_empty_webhook_data():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={}, # No webhook_data
)
@ -368,7 +368,7 @@ def test_webhook_node_run_case_insensitive_headers():
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {
@ -398,7 +398,7 @@ def test_webhook_node_variable_pool_user_inputs():
# Add some additional variables to the pool
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {"headers": {}, "query_params": {}, "body": {}, "files": {}},
"other_var": "should_be_included",
@ -429,7 +429,7 @@ def test_webhook_node_different_methods(method):
)
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={
"webhook_data": {
"headers": {},

18
api/tests/unit_tests/core/workflow/test_workflow_entry.py
View File

@ -127,7 +127,7 @@ class TestWorkflowEntry:
return node_config
workflow = StubWorkflow()
variable_pool = VariablePool(system_variables=SystemVariable.empty(), user_inputs={})
variable_pool = VariablePool(system_variables=SystemVariable.default(), user_inputs={})
expected_limits = CodeNodeLimits(
max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
max_number=dify_config.CODE_MAX_NUMBER,
@ -157,7 +157,7 @@ class TestWorkflowEntry:
# Initialize variable pool with environment variables
env_var = StringVariable(name="API_KEY", value="existing_key")
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
environment_variables=[env_var],
user_inputs={},
)
@ -198,7 +198,7 @@ class TestWorkflowEntry:
# Initialize variable pool with conversation variables
conv_var = StringVariable(name="last_message", value="Hello")
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
conversation_variables=[conv_var],
user_inputs={},
)
@ -239,7 +239,7 @@ class TestWorkflowEntry:
"""Test mapping regular node variables from user inputs to variable pool."""
# Initialize empty variable pool
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
@ -281,7 +281,7 @@ class TestWorkflowEntry:
def test_mapping_user_inputs_with_file_handling(self):
"""Test mapping file inputs from user inputs to variable pool."""
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
@ -340,7 +340,7 @@ class TestWorkflowEntry:
def test_mapping_user_inputs_missing_variable_error(self):
"""Test that mapping raises error when required variable is missing."""
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
@ -366,7 +366,7 @@ class TestWorkflowEntry:
def test_mapping_user_inputs_with_alternative_key_format(self):
"""Test mapping with alternative key format (without node prefix)."""
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
@ -396,7 +396,7 @@ class TestWorkflowEntry:
def test_mapping_user_inputs_with_complex_selectors(self):
"""Test mapping with complex node variable keys."""
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)
@ -432,7 +432,7 @@ class TestWorkflowEntry:
def test_mapping_user_inputs_invalid_node_variable(self):
"""Test that mapping handles invalid node variable format."""
variable_pool = VariablePool(
system_variables=SystemVariable.empty(),
system_variables=SystemVariable.default(),
user_inputs={},
)