工作流测试与模拟系统(中文译文)
原始 DeepWiki 页面:https://deepwiki.com/langgenius/dify/5.7-workflow-testing-and-mock-system
翻译时间:2026-05-27T08:44:26.628Z
翻译模型:deepseek-chat
原文字符数:12914
项目:Dify (dify)
---
工作流测试与模拟系统
相关源文件
以下文件被用作生成此维基页面的上下文:
api/.env.exampleapi/.importlinterapi/app.pyapi/app_factory.pyapi/configs/feature/__init__.pyapi/configs/middleware/__init__.pyapi/configs/observability/__init__.pyapi/configs/observability/otel/otel_config.pyapi/configs/packaging/__init__.pyapi/controllers/console/datasets/datasets.pyapi/core/app/file_access/__init__.pyapi/core/app/file_access/controller.pyapi/core/app/file_access/scope.pyapi/core/rag/datasource/vdb/vector_factory.pyapi/core/rag/datasource/vdb/vector_type.pyapi/core/workflow/node_factory.pyapi/core/workflow/node_runtime.pyapi/extensions/ext_compress.pyapi/extensions/ext_otel.pyapi/extensions/otel/instrumentation.pyapi/pyproject.tomlapi/tests/integration_tests/workflow/nodes/test_code.pyapi/tests/integration_tests/workflow/nodes/test_http.pyapi/tests/integration_tests/workflow/nodes/test_llm.pyapi/tests/integration_tests/workflow/nodes/test_parameter_extractor.pyapi/tests/integration_tests/workflow/nodes/test_template_transform.pyapi/tests/integration_tests/workflow/nodes/test_tool.pyapi/tests/unit_tests/configs/test_dify_config.pyapi/tests/unit_tests/core/rag/datasource/test_retrieval_attachment_access.pyapi/tests/unit_tests/core/workflow/graph_engine/test_mock_factory.pyapi/tests/unit_tests/core/workflow/graph_engine/test_mock_nodes.pyapi/tests/unit_tests/core/workflow/graph_engine/test_table_runner.pyapi/tests/unit_tests/core/workflow/graph_engine/test_tool_in_chatflow.pyapi/tests/unit_tests/core/workflow/nodes/answer/test_answer.pyapi/tests/unit_tests/core/workflow/nodes/code/__init__.pyapi/tests/unit_tests/core/workflow/nodes/code/code_node_spec.pyapi/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_executor.pyapi/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_node.pyapi/tests/unit_tests/core/workflow/nodes/template_transform/__init__.pyapi/tests/unit_tests/core/workflow/nodes/template_transform/template_transform_node_spec.pyapi/tests/unit_tests/core/workflow/nodes/test_if_else.pyapi/tests/unit_tests/core/workflow/nodes/test_list_operator.pyapi/tests/unit_tests/core/workflow/nodes/tool/test_tool_node.pyapi/tests/unit_tests/core/workflow/test_node_factory.pyapi/tests/unit_tests/core/workflow/test_node_runtime.pyapi/uv.lockdocker/.env.exampledocker/README.mddocker/docker-compose-template.yamldocker/docker-compose.middleware.yamldocker/docker-compose.yamldocker/envs/core-services/shared.env.exampledocker/envs/infrastructure/nginx.env.exampledocker/envs/security.env.exampledocker/nginx/conf.d/default.conf.templateweb/package.json
本文档描述了工作流执行的测试基础设施和模拟系统。它涵盖了 MockNodeFactory、模拟节点实现,以及测试工作流时是否依赖外部服务的策略。
目的与范围
工作流模拟系统可以在不依赖外部服务(大语言模型(LLM)API、HTTP 端点、代码执行沙箱)的情况下测试工作流逻辑。它提供:
- 需要第三方服务的节点的模拟实现。
- 通过
MockConfig和节点特定的处理器实现可配置行为。 - 使用容器化环境进行真实服务验证的集成测试。
- 用于隔离测试单个节点的单步执行。
该系统对于工作流图的单元测试、无速率限制或成本的集成测试,以及在持续集成/持续部署(CI/CD)管线中验证工作流逻辑至关重要。
---
测试方法概述
工作流测试策略映射
graph TB
subgraph "自然语言空间(用户意图)"
Intent1["'测试我的工作流逻辑而不消耗额度'"]
Intent2["'使用真实数据调试特定节点'"]
Intent3["'在多个场景上运行回归测试'"]
end
subgraph "代码实体空间(实现)"
MockFactory["MockNodeFactory"]
MockNodes["MockLLMNode, MockCodeNode 等"]
TestContainers["Testcontainers"]
WorkflowService["WorkflowService.run_draft_workflow_node"]
DifyNodeFactory["DifyNodeFactory(真实)"]
end
Intent1 --> MockFactory
MockFactory --> MockNodes
Intent2 --> WorkflowService
WorkflowService --> DifyNodeFactory
Intent3 --> TestContainers
TestContainers --> DifyNodeFactory
主要测试策略:
| 策略 | 使用的工厂 | 依赖 | 使用场景 |
|---|---|---|---|
| 使用模拟的单元测试 | MockNodeFactory | 无 | 快速测试、持续集成/持续部署(CI/CD)、逻辑验证。 |
| 集成测试 | DifyNodeFactory | testcontainers | 数据库/Redis 集成、真实模式验证。 |
| 单步测试 | DifyNodeFactory | 真实服务 | 草稿工作流调试、节点开发。 |
来源: api/core/workflow/node_factory.py:1-20,api/tests/unit_tests/core/workflow/graph_engine/test_mock_factory.py:1-179,api/pyproject.toml:127-127
---
模拟系统架构
MockNodeFactory 与依赖注入
MockNodeFactory 扩展了 DifyNodeFactory,用于拦截节点创建,并为需要外部服务的节点替换模拟实现。它维护了一个应被模拟的节点类型注册表。
classDiagram
class DifyNodeFactory {
-GraphInitParams graph_init_params
-GraphRuntimeState graph_runtime_state
+create_node(node_id, node_config) Node
}
class MockNodeFactory {
-MockConfig mock_config
-dict _mock_node_types
+create_node(node_id, node_config) Node
+register_mock_node_type(node_type, mock_class)
}
class MockLLMNode {
+_run() Generator
}
DifyNodeFactory <|-- MockNodeFactory
MockNodeFactory ..> MockLLMNode : 创建
来源: api/core/workflow/node_factory.py:1-20,api/tests/unit_tests/core/workflow/graph_engine/test_mock_factory.py:37-179
---
模拟节点类型映射
已注册的模拟节点类型
MockNodeFactory 通过检查节点数据中的 type 字段,将标准节点替换为模拟版本。
| 节点类型 | 模拟类 | 注入的额外依赖 |
|---|---|---|
NodeType.LLM | MockLLMNode | credentials_provider,model_factory,model_instance |
NodeType.CODE | MockCodeNode | code_executor,code_limits |
NodeType.HTTP_REQUEST | MockHttpRequestNode | http_client,tool_file_manager_factory |
来源: api/tests/unit_tests/core/workflow/graph_engine/test_mock_factory.py:62-151
---
集成测试基础设施
TestContainers 配置
对于集成测试,Dify 使用 testcontainers 来启动 PostgreSQL、Redis 和沙箱的隔离环境。这确保了测试针对真实的服务实现运行,而不仅仅是逻辑模拟。
sequenceDiagram
participant Test as Pytest 会话
participant TC as testcontainers
participant PG as PostgresContainer
participant RD as RedisContainer
participant SB as SandboxContainer
Test->>TC: 启动容器
TC->>PG: 启动(镜像:postgres:15-alpine)
PG-->>TC: 主机/端口/凭证
TC->>RD: 启动(镜像:redis:6-alpine)
RD-->>TC: 主机/端口
TC->>SB: 启动(镜像:langgenius/dify-sandbox)
TC->>Test: 设置操作系统环境变量(DB_HOST,REDIS_HOST 等)
关键容器组件:
- PostgreSQL:用于存储用户数据、工作流和执行状态。生产模板中使用镜像
postgres:15-alpinedocker/docker-compose.middleware.yaml:4-4。 - Redis:用于缓存、会话管理和 Celery 任务代理。标准镜像为
redis:6-alpinedocker/docker-compose.middleware.yaml:84-84。 - Dify 沙箱:为测试
CodeNode提供安全环境。镜像langgenius/dify-sandbox:0.2.15docker/docker-compose.middleware.yaml:106-106。
来源: api/pyproject.toml:127-127,docker/docker-compose.middleware.yaml:1-129
---
表驱动节点测试
Dify 广泛使用 pytest.mark.parametrize 进行节点逻辑的表驱动测试。这允许单个测试函数验证多个输入/输出场景。
HTTP 请求执行器测试
HttpRequestNode 的 Executor 针对各种 JSON 请求体结构进行测试,包括数字变量和嵌套对象 api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_executor.py:30-190。
代码节点验证
CodeNode 测试验证输出逻辑,确保返回的数据符合预期的模式类型(例如,确保结果是 string 而不是 int)api/tests/integration_tests/workflow/nodes/test_code.py:136-178。
来源: api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_executor.py:30-190,api/tests/integration_tests/workflow/nodes/test_code.py:136-178
---
关键测试工具
VariablePool 模拟
在测试中手动填充 VariablePool,以模拟工作流在特定节点执行时的状态。
variable_pool = VariablePool(
system_variables=default_system_variables(),
user_inputs={},
)
variable_pool.add(["pre_node_id", "number"], 42)
来源: api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_executor.py:32-36
模型实例模拟
对于 LLMNode 和 ParameterExtractorNode,测试通常会模拟 ModelInstance 以返回确定性的 LLMResult 对象,从而避免真实的 API 成本和延迟 api/tests/integration_tests/workflow/nodes/test_llm.py:123-168。
来源: api/tests/integration_tests/workflow/nodes/test_llm.py:123-168,api/tests/integration_tests/workflow/nodes/test_parameter_extractor.py:111-117