agentic_huge_data_base / wiki
页面 Dify · 5.7 工作流测试与模拟系统·DeepWiki 中文全文译文

5.7 · 工作流测试与模拟系统(Workflow Testing and Mock System)

应用编排与外部知识接入 · 本章是 Dify DeepWiki 中文译文的独立章节页,保留原始链接、源码锚点、模块标签和章节层级。

项目Dify 章节5.7 状态全文译文 模块系统架构、测试、发布与运维、接口与服务契约、模型调用与提供方适配
源码线索
  • api/.env.example
  • api/.importlinter
  • api/app.py
  • api/app_factory.py
  • api/configs/feature/__init__.py
  • api/configs/middleware/__init__.py
  • api/configs/observability/__init__.py
  • api/configs/observability/otel/otel_config.py
  • api/configs/packaging/__init__.py
  • api/controllers/console/datasets/datasets.py
模块标签
  • 系统架构
  • 测试、发布与运维
  • 接口与服务契约
  • 模型调用与提供方适配
  • 配置治理

中文译文

工作流测试与模拟系统(中文译文)

原始 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.example
  • api/.importlinter
  • api/app.py
  • api/app_factory.py
  • api/configs/feature/__init__.py
  • api/configs/middleware/__init__.py
  • api/configs/observability/__init__.py
  • api/configs/observability/otel/otel_config.py
  • api/configs/packaging/__init__.py
  • api/controllers/console/datasets/datasets.py
  • api/core/app/file_access/__init__.py
  • api/core/app/file_access/controller.py
  • api/core/app/file_access/scope.py
  • api/core/rag/datasource/vdb/vector_factory.py
  • api/core/rag/datasource/vdb/vector_type.py
  • api/core/workflow/node_factory.py
  • api/core/workflow/node_runtime.py
  • api/extensions/ext_compress.py
  • api/extensions/ext_otel.py
  • api/extensions/otel/instrumentation.py
  • api/pyproject.toml
  • api/tests/integration_tests/workflow/nodes/test_code.py
  • api/tests/integration_tests/workflow/nodes/test_http.py
  • api/tests/integration_tests/workflow/nodes/test_llm.py
  • api/tests/integration_tests/workflow/nodes/test_parameter_extractor.py
  • api/tests/integration_tests/workflow/nodes/test_template_transform.py
  • api/tests/integration_tests/workflow/nodes/test_tool.py
  • api/tests/unit_tests/configs/test_dify_config.py
  • api/tests/unit_tests/core/rag/datasource/test_retrieval_attachment_access.py
  • api/tests/unit_tests/core/workflow/graph_engine/test_mock_factory.py
  • api/tests/unit_tests/core/workflow/graph_engine/test_mock_nodes.py
  • api/tests/unit_tests/core/workflow/graph_engine/test_table_runner.py
  • api/tests/unit_tests/core/workflow/graph_engine/test_tool_in_chatflow.py
  • api/tests/unit_tests/core/workflow/nodes/answer/test_answer.py
  • api/tests/unit_tests/core/workflow/nodes/code/__init__.py
  • api/tests/unit_tests/core/workflow/nodes/code/code_node_spec.py
  • api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_executor.py
  • api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_node.py
  • api/tests/unit_tests/core/workflow/nodes/template_transform/__init__.py
  • api/tests/unit_tests/core/workflow/nodes/template_transform/template_transform_node_spec.py
  • api/tests/unit_tests/core/workflow/nodes/test_if_else.py
  • api/tests/unit_tests/core/workflow/nodes/test_list_operator.py
  • api/tests/unit_tests/core/workflow/nodes/tool/test_tool_node.py
  • api/tests/unit_tests/core/workflow/test_node_factory.py
  • api/tests/unit_tests/core/workflow/test_node_runtime.py
  • api/uv.lock
  • docker/.env.example
  • docker/README.md
  • docker/docker-compose-template.yaml
  • docker/docker-compose.middleware.yaml
  • docker/docker-compose.yaml
  • docker/envs/core-services/shared.env.example
  • docker/envs/infrastructure/nginx.env.example
  • docker/envs/security.env.example
  • docker/nginx/conf.d/default.conf.template
  • web/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)、逻辑验证。
集成测试DifyNodeFactorytestcontainers数据库/Redis 集成、真实模式验证。
单步测试DifyNodeFactory真实服务草稿工作流调试、节点开发。

来源: api/core/workflow/node_factory.py:1-20api/tests/unit_tests/core/workflow/graph_engine/test_mock_factory.py:1-179api/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-20api/tests/unit_tests/core/workflow/graph_engine/test_mock_factory.py:37-179

---

模拟节点类型映射

已注册的模拟节点类型

MockNodeFactory 通过检查节点数据中的 type 字段,将标准节点替换为模拟版本。

节点类型模拟类注入的额外依赖
NodeType.LLMMockLLMNodecredentials_providermodel_factorymodel_instance
NodeType.CODEMockCodeNodecode_executorcode_limits
NodeType.HTTP_REQUESTMockHttpRequestNodehttp_clienttool_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-alpine docker/docker-compose.middleware.yaml:4-4
  • Redis:用于缓存、会话管理和 Celery 任务代理。标准镜像为 redis:6-alpine docker/docker-compose.middleware.yaml:84-84
  • Dify 沙箱:为测试 CodeNode 提供安全环境。镜像 langgenius/dify-sandbox:0.2.15 docker/docker-compose.middleware.yaml:106-106

来源: api/pyproject.toml:127-127docker/docker-compose.middleware.yaml:1-129

---

表驱动节点测试

Dify 广泛使用 pytest.mark.parametrize 进行节点逻辑的表驱动测试。这允许单个测试函数验证多个输入/输出场景。

HTTP 请求执行器测试

HttpRequestNodeExecutor 针对各种 JSON 请求体结构进行测试,包括数字变量和嵌套对象 api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_executor.py:30-190

代码节点验证

CodeNode 测试验证输出逻辑,确保返回的数据符合预期的模式类型(例如,确保结果是 string 而不是 intapi/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-190api/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

模型实例模拟

对于 LLMNodeParameterExtractorNode,测试通常会模拟 ModelInstance 以返回确定性的 LLMResult 对象,从而避免真实的 API 成本和延迟 api/tests/integration_tests/workflow/nodes/test_llm.py:123-168

来源: api/tests/integration_tests/workflow/nodes/test_llm.py:123-168api/tests/integration_tests/workflow/nodes/test_parameter_extractor.py:111-117