a-cdm × deer-flow 源码分析

主题

测试策略与质量保障 ​

本章目标:

  1. 理解 DeerFlow 为什么把强制 TDD、分层边界测试与接口一致性测试写进开发纪律,而不仅仅是"跑跑单测"。
  2. 掌握 5 类测试(单元 / 边界 / 一致性 / 回归 / 前端)各自的职责、关键文件与在 CI 中的触发时机。
  3. 学会按现有约定为新功能补测试:命名规范、conftest.py 解循环导入的 sys.modules mock 技巧、最小测试模板。

TL;DR ​

DeerFlow 的质量保障建立在四个支点上:backend/tests/ 下统一的 test_<feature>.py 命名约定与强制 TDD(backend/CLAUDE.md TDD 节);一个用 AST 扫描的导入防火墙 test_harness_boundary.py,保证可发布的 deerflow harness 永不依赖 app 层(backend/tests/test_harness_boundary.py:37-46);TestGatewayConformance 把嵌入式客户端的输出过一遍 Gateway 的 Pydantic 响应模型,防止两条 API 路径漂移(backend/tests/test_client.py:2302-2329);以及 conftest.py 用 sys.modules 预 mock 打破生产代码的循环导入链(backend/tests/conftest.py:26-45)。所有这些由 .github/workflows/ 下五条 CI 工作流在每个 PR 上自动执行。

Overview ​

DeerFlow 是一个分层的 super agent harness:可发布的 deerflow 框架包(packages/harness/deerflow/)在底层,不可发布的应用层(app/ 的 Gateway API、IM 通道)在上层,再加上嵌入式 Python 客户端 DeerFlowClient 作为不走 HTTP 的并行入口。这种结构决定了"测试通过"不等于"质量合格"——还需要回答三个架构问题:

  • 分层会不会被悄悄破坏? 如果 harness 里有人 from app.gateway...,框架就无法独立发布,但普通单测发现不了这种依赖方向错误。需要一个专门的边界测试。
  • 两条 API 路径会不会漂移? Gateway HTTP 路由与 DeerFlowClient 返回相同语义的数据。Gateway 用 Pydantic 模型校验响应,客户端却是手搓 dict。一旦 Gateway 加了必填字段而客户端没跟进,消费方代码会在运行时炸。需要一致性测试当编译期检查。
  • 被修过的坑会不会复发? Docker 沙箱模式探测、provisioner kubeconfig 处理等都是历史 bug 高发区,需要回归测试钉死行为。

因此 DeerFlow 把 TDD 写成硬性政策("Every new feature or bug fix MUST be accompanied by unit tests. No exceptions."),并把边界、一致性、回归测试都接入 CI,让架构约束成为可执行的代码而非口头约定。

Architecture ​

测试体系按职责分为五层,加上五条 CI 工作流。后端测试目录 backend/tests/ 当前有 173 个 test_*.py 文件,全部走统一的 test_<feature>.py 命名约定,平铺在一个目录下;make test 通过 PYTHONPATH=. uv run pytest tests/ -v 一次性收集运行(backend/Makefile:10-11)。

Sources:

Components / Subsystems ​

1. 边界测试 test_harness_boundary.py ​

职责:用 ast 解析 packages/harness/deerflow/ 下每个 .py 文件,遍历所有 ast.Import / ast.ImportFrom 节点,一旦发现模块名等于或以 app. 开头就累积成 violation 列表,最后断言列表为空。这把"harness 不得依赖 app 层"的架构规则变成一个会让 CI 红灯的测试。

关键文件:

用 AST 而非文本 grep 的好处是:注释里、字符串里出现的 import app.x 不会误报,只有真正的导入语句才算违规。下图是边界测试的判定状态机:

2. 一致性测试 TestGatewayConformance ​

职责:验证 DeerFlowClient 每个返回 dict 的方法都符合对应的 Gateway Pydantic 响应模型。测试直接 from app.gateway.routers.* 导入真实响应模型(ModelsListResponse、McpConfigResponse、SkillsListResponse、UploadResponse、MemoryConfigResponse 等),调用客户端方法后用 Model(**result) 解析。如果 Gateway 新增必填字段而客户端没提供,Pydantic 抛 ValidationError,CI 立刻捕获这种漂移。

关键文件:

3. conftest.py —— sys.modules mock 解循环导入 ​

职责:让轻量的 config/registry 单测能在隔离环境下被导入。生产代码存在一条循环依赖链:deerflow.subagents.__init__ → .executor → deerflow.agents.thread_state → deerflow.agents.__init__ → lead_agent.agent → subagent_limit_middleware → deerflow.subagents.executor。conftest.py 在任何测试触发导入前,先往 sys.modules 里塞一个 MagicMock 顶替 deerflow.subagents.executor,使 __init__.py 的 from .executor import ... 立即成功而不执行真实模块。它还负责把 app/deerflow 注入 sys.path,并提供 provisioner_module fixture、自动用户上下文 fixture、blocking-IO 探针等。

关键文件:

4. Docker / provisioner 回归测试 ​

职责:钉死历史易错行为,防止复发。test_docker_sandbox_mode_detection.py 用临时 config.yaml 驱动 scripts/docker.sh 里的 detect_sandbox_mode shell 函数,断言不同 sandbox.use 配置(local / aio / provisioner / 注释掉的 url / 未知 provider)被正确归类。test_provisioner_kubeconfig.py 通过共用的 provisioner_module fixture 验证 kubeconfig 是目录时快速失败、是文件时正常加载、缺失时回退 in-cluster。

关键文件:

5. 前端测试 vitest / playwright ​

职责:vitest 跑 tests/unit/**/*.test.ts 单元测试,镜像 src/ 布局并通过 @ 别名导入源模块;playwright 跑 tests/e2e/ 下的 Chromium E2E,用 page.route() 拦截所有后端 API,测试真实页面交互。

关键文件:

Data Flow ​

下图展示一次 PR 提交后,CI 如何并行触发后端单测(含边界 + 一致性 + 回归)、lint、前端单测与 E2E。

Implementation Details ​

边界测试的核心是用 AST 而非正则去识别真实导入语句,关键片段:

python
def _collect_imports(filepath: Path) -> list[tuple[int, str]]:
    source = filepath.read_text(encoding="utf-8")
    tree = ast.parse(source, filename=str(filepath))
    results: list[tuple[int, str]] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                results.append((node.lineno, alias.name))
        elif isinstance(node, ast.ImportFrom):
            if node.module:
                results.append((node.lineno, node.module))
    return results

ast.walk 深度遍历整棵语法树,只对 ast.Import(import app.x)和 ast.ImportFrom(from app.x import y)两类节点取模块名与行号;字符串、注释里出现的 app. 字样不会被解析成这两类节点,因此天然免疫误报。判定逻辑 module == "app" or module.startswith("app.") 同时拦截裸 import app 与子模块导入(backend/tests/test_harness_boundary.py:18-46)。

conftest.py 解循环导入的关键只有一行——在收集任何测试之前替换模块:

python
_executor_mock = MagicMock()
_executor_mock.SubagentExecutor = MagicMock
_executor_mock.MAX_CONCURRENT_SUBAGENTS = 3
sys.modules["deerflow.subagents.executor"] = _executor_mock

因为 conftest.py 在 pytest 收集阶段最早被执行,这行 sys.modules[...] = mock 比任何测试模块的 import 都早。当后续代码执行 deerflow.subagents.__init__ 里的 from .executor import SubagentExecutor 时,Python 直接命中 sys.modules 缓存里的 mock,不会再去加载真实 executor.py,循环链就此切断(backend/tests/conftest.py:26-45)。

速查表 ​

测试类别作用关键文件在 CI 何时跑Source
单元测试验证单个函数/模块行为,test_<feature>.py,173 个文件平铺于 tests/backend/tests/test_*.py每个非 draft PR / push main(make test)backend/Makefile:10-11
边界测试AST 扫描 harness 包,禁止 import app.* 维持分层backend/tests/test_harness_boundary.py随后端单测一起跑backend/tests/test_harness_boundary.py:37-46
一致性测试客户端 dict 过 Gateway Pydantic 模型,防两路 API 漂移backend/tests/test_client.py TestGatewayConformance随后端单测一起跑backend/tests/test_client.py:2302-2329
回归测试钉死 Docker 模式探测 / provisioner kubeconfig 历史行为test_docker_sandbox_mode_detection.py、test_provisioner_kubeconfig.py随后端单测一起跑backend/tests/test_docker_sandbox_mode_detection.py:44-107
解循环导入conftest 用 sys.modules mock 顶替 executor,注入 sys.pathbackend/tests/conftest.py收集阶段自动加载backend/tests/conftest.py:26-45
前端单元测试vitest 跑 tests/unit/**/*.test.tsfrontend/vitest.config.ts每个非 draft PR / push main(pnpm test).github/workflows/frontend-unit-tests.yml:41-44
前端 E2Eplaywright Chromium,page.route() mock 后端frontend/playwright.config.tsPR/push 且改动 frontend/**.github/workflows/e2e-tests.yml:9-53
Lint/格式ruff check + ruff format --check;前端 eslint+prettier+typecheckbackend/ruff.toml、lint-check.yml每次 PR(任意分支)/ push main.github/workflows/lint-check.yml:31-33
pre-commit提交前本地 ruff(后端)/ eslint+prettier(前端).pre-commit-config.yaml本地 git commit(可选启用).pre-commit-config.yaml:1-33
容器构建打 v* tag 时构建并推送后端/前端镜像.github/workflows/container.yaml推送 v* tag 时.github/workflows/container.yaml:3-7

扩展指南 ​

强制约束(来自 TDD 政策) ​

"Every new feature or bug fix MUST be accompanied by unit tests. No exceptions." 落地约定:

  1. 测试放在 backend/tests/,文件名遵循 test_<feature>.py(平铺,不建子目录)。
  2. 改动前后都跑全量:make test,必须全绿才算功能完成。
  3. 轻量 config/util 模块优先写纯单测,不依赖外部服务。
  4. 若新模块在测试时触发循环导入,在 tests/conftest.py 加一条 sys.modules mock(照搬 deerflow.subagents.executor 的写法)。
  5. PR 前还要 make lint 与 make format(backend/CONTRIBUTING.md:240-247)。

最小测试模板(后端) ​

依据 backend/CONTRIBUTING.md:223-238 的范式:

python
# backend/tests/test_my_feature.py
import pytest
from deerflow.models.factory import create_chat_model


def test_create_chat_model_with_valid_name():
    """合法模型名应能创建模型实例。"""
    model = create_chat_model("gpt-4")
    assert model is not None


def test_create_chat_model_with_invalid_name():
    """非法模型名应抛 ValueError。"""
    with pytest.raises(ValueError):
        create_chat_model("nonexistent-model")

运行单个文件:PYTHONPATH=. uv run pytest tests/test_my_feature.py -v(backend/CLAUDE.md TDD 节)。

为客户端新增方法时 ​

只要 DeerFlowClient 新增一个返回 dict 且对应某个 Gateway 路由的方法,就应在 TestGatewayConformance 加一个用例:from app.gateway.routers.x import XResponse,然后 parsed = XResponse(**client.x()),断言关键字段。这样 Gateway 模型一变,客户端漏跟会被 CI 拦下(backend/tests/test_client.py:2302-2329)。

前端新增测试 ​

单测放 tests/unit/ 并镜像 src/ 布局(如 tests/unit/core/api/stream-mode.test.ts 对应 src/core/api/stream-mode.ts),用 @/ 别名导入;E2E 放 tests/e2e/*.spec.ts,用 page.route() mock 所有后端 API(frontend/vitest.config.ts:11-13)。

Common Pitfalls / Tips ​

  • 循环导入要在 conftest 加 sys.modules mock:这是政策明文要求的标准解法,且必须在模块顶层(收集阶段)注入,放进 fixture 就晚了——fixture 执行时真实模块往往已被导入(backend/tests/conftest.py:35-45)。
  • 持久层测试默认有用户上下文:_auto_user_context 是 autouse fixture,会自动注入 test-user-autouse;若你要验证"无用户上下文"行为,必须显式标 @pytest.mark.no_auto_user,否则永远测不到 RuntimeError 分支(backend/tests/conftest.py:179-205)。
  • 边界测试报错要看输出的 路径:行号:违规信息直接给出文件相对路径与导入行号,定位即修,不要去改这个测试本身(backend/tests/test_harness_boundary.py:42-46)。
  • 一致性测试失败 ≠ 测试有问题:ValidationError 通常意味着 Gateway 模型加了客户端没补的字段,应去补客户端而不是放宽断言(backend/tests/test_client.py:2303-2308)。
  • 回归测试依赖 bash / 临时 config:Docker 模式探测测试在无 bash 环境会整体 skip(pytestmark),provisioner 测试依赖 provisioner_module fixture 动态加载 docker/provisioner/app.py,改路径要同步更新 fixture(backend/tests/test_docker_sandbox_mode_detection.py:14-24, backend/tests/conftest.py:48-63)。
  • draft PR 不跑测试:后端/前端/E2E 工作流都带 if: github.event.pull_request.draft == false;想触发 CI 需把 PR 标记为 ready for review(.github/workflows/backend-unit-tests.yml:18)。
  • lint 与 format 是两道关:make lint 同时跑 ruff check . 与 ruff format --check .,后者只检查不改文件;本地用 make format 自动修复后再提交,行宽上限 240(backend/Makefile:13-18, backend/ruff.toml:1-13)。
  • 可选 blocking-IO 探针:pytest --detect-blocking-io-fail 会让事件循环上的阻塞调用直接失败,排查异步代码里误用同步 I/O 时很有用(backend/tests/conftest.py:73-86)。

References ​

章节关系
09-Harness与App分层边界.md本章的边界测试 test_harness_boundary.py 正是用来强制执行第 9 章描述的 harness/app 分层依赖方向
32-嵌入式Python客户端.mdTestGatewayConformance 校验的对象就是第 32 章的 DeerFlowClient,确保它与 Gateway API 响应一致
13-Gateway-API与路由体系.md一致性测试导入第 13 章 Gateway 路由里的 Pydantic 响应模型作为校验基准
08-Docker部署与运维.mdDocker/provisioner 回归测试钉死的是第 8 章描述的沙箱模式探测与部署行为

公司内部参考 · 由 claude-wiki-gen 基于源码自动生成的二次分析

a-cdm（公司内部）· deer-flow © ByteDance